builder: Enable Command Substitution for Email / SMTP Configuration (#746)

Author: yottahmd

cmd/logger.go

@@ -9,6 +9,7 @@ import (
 	"path/filepath"
 	"time"
 
+	"github.com/dagu-org/dagu/internal/cmdutil"
 	"github.com/dagu-org/dagu/internal/config"
 	"github.com/dagu-org/dagu/internal/fileutil"
 	"github.com/dagu-org/dagu/internal/logger"
@@ -52,17 +53,23 @@ type logFileSettings struct {
 // openLogFile creates and opens a log file based on the provided settings.
 // It creates the necessary directory structure and returns the file handle.
 func openLogFile(config logFileSettings) (*os.File, error) {
+	logDir, err := cmdutil.SubstituteCommands(os.ExpandEnv(config.LogDir))
+	if err != nil {
+		return nil, fmt.Errorf("failed to expand log directory: %w", err)
+	}
+	config.LogDir = logDir
+
 	if err := validateSettings(config); err != nil {
 		return nil, fmt.Errorf("invalid log settings: %w", err)
 	}
 
-	logDir, err := setupLogDirectory(config)
+	outputDir, err := setupLogDirectory(config)
 	if err != nil {
 		return nil, fmt.Errorf("failed to setup log directory: %w", err)
 	}
 
 	filename := buildLogFilename(config)
-	return createLogFile(filepath.Join(logDir, filename))
+	return createLogFile(filepath.Join(outputDir, filename))
 }
 
 // validateSettings ensures all required fields are properly set

internal/agent/agent.go

@@ -10,6 +10,7 @@ import (
 	"net/http"
 	"os"
 	"regexp"
+	"runtime/debug"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -140,12 +141,12 @@ func (a *Agent) Run(ctx context.Context) error {
 		return fmt.Errorf("failed to setup unix socket server: %w", err)
 	}
 	listenerErrCh := make(chan error)
-	go func() {
+	go execWithRecovery(ctx, func() {
 		err := a.socketServer.Serve(ctx, listenerErrCh)
 		if err != nil && !errors.Is(err, sock.ErrServerRequestedShutdown) {
 			logger.Error(ctx, "Failed to start socket frontend", "err", err)
 		}
-	}()
+	})
 
 	// Stop the socket server when finishing the DAG execution.
 	defer func() {
@@ -164,7 +165,7 @@ func (a *Agent) Run(ctx context.Context) error {
 	// example, when started, stopped, or cancelled, etc.
 	done := make(chan *scheduler.Node)
 	defer close(done)
-	go func() {
+	go execWithRecovery(ctx, func() {
 		for node := range done {
 			status := a.Status()
 			if err := a.historyStore.Write(ctx, status); err != nil {
@@ -174,19 +175,19 @@ func (a *Agent) Run(ctx context.Context) error {
 				logger.Error(ctx, "Failed to report step", "err", err)
 			}
 		}
-	}()
+	})
 
 	// Write the first status just after the start to store the running status.
 	// If the DAG is already finished, skip it.
-	go func() {
+	go execWithRecovery(ctx, func() {
 		time.Sleep(waitForRunning)
 		if a.finished.Load() {
 			return
 		}
 		if err := a.historyStore.Write(ctx, a.Status()); err != nil {
 			logger.Error(ctx, "Status write failed", "err", err)
 		}
-	}()
+	})
 
 	// Start the DAG execution.
 	logger.Info(ctx, "DAG execution started", "reqId", a.requestID, "name", a.dag.Name, "params", a.dag.Params)
@@ -314,7 +315,7 @@ func (a *Agent) setup(ctx context.Context) error {
 
 	a.scheduler = a.newScheduler()
 	a.reporter = newReporter(
-		mailer.New(&mailer.NewMailerArgs{
+		mailer.New(mailer.Config{
 			Host:     a.dag.SMTP.Host,
 			Port:     a.dag.SMTP.Port,
 			Username: a.dag.SMTP.Username,
@@ -501,6 +502,21 @@ func (a *Agent) checkIsAlreadyRunning(ctx context.Context) error {
 	return nil
 }
 
+func execWithRecovery(ctx context.Context, fn func()) {
+	defer func() {
+		if panicObj := recover(); panicObj != nil {
+			err, ok := panicObj.(error)
+			if !ok {
+				err = fmt.Errorf("panic: %v", panicObj)
+			}
+			st := string(debug.Stack())
+			logger.Error(ctx, "Panic occurred", "err", err, "st", st)
+		}
+	}()
+
+	fn()
+}
+
 type httpError struct {
 	Code    int
 	Message string

internal/agent/reporter.go

@@ -52,7 +52,7 @@ func (r *reporter) reportStep(
 }
 
 // report is a function that reports the status of the scheduler.
-func (r *reporter) getSummary(ctx context.Context, status *model.Status, err error) string {
+func (r *reporter) getSummary(_ context.Context, status *model.Status, err error) string {
 	var buf bytes.Buffer
 	_, _ = buf.Write([]byte("\n"))
 	_, _ = buf.Write([]byte("Summary ->\n"))

internal/cmdutil/cmdutil.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"os"
 	"os/exec"
+	"reflect"
 	"regexp"
 	"strings"
 	"unicode"
@@ -229,3 +230,50 @@ func GetShellCommand(configuredShell string) string {
 
 	return ""
 }
+
+// SubstituteStringFields processes all string fields in a struct by expanding environment
+// variables and substituting command outputs. It takes a struct value and returns a new
+// modified struct value.
+func SubstituteStringFields[T any](obj T) (T, error) {
+	v := reflect.ValueOf(obj)
+	if v.Kind() != reflect.Struct {
+		return obj, fmt.Errorf("input must be a struct, got %T", obj)
+	}
+
+	modified := reflect.New(v.Type()).Elem()
+	modified.Set(v)
+
+	if err := processStructFields(modified); err != nil {
+		return obj, fmt.Errorf("failed to process fields: %w", err)
+	}
+
+	return modified.Interface().(T), nil
+}
+
+func processStructFields(v reflect.Value) error {
+	t := v.Type()
+	for i := 0; i < v.NumField(); i++ {
+		field := v.Field(i)
+		if !field.CanSet() {
+			continue
+		}
+
+		// nolint:exhaustive
+		switch field.Kind() {
+		case reflect.String:
+			value := field.String()
+			value = os.ExpandEnv(value)
+			processed, err := SubstituteCommands(value)
+			if err != nil {
+				return fmt.Errorf("field %q: %w", t.Field(i).Name, err)
+			}
+			field.SetString(processed)
+
+		case reflect.Struct:
+			if err := processStructFields(field); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}

internal/cmdutil/cmdutil_test.go

@@ -5,6 +5,7 @@ package cmdutil
 
 import (
 	"os"
+	"reflect"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -168,3 +169,157 @@ func TestSplitCommandWithParse(t *testing.T) {
 		require.Equal(t, "hello", args[0])
 	})
 }
+
+func TestSubstituteStringFields(t *testing.T) {
+	// Set up test environment variables
+	os.Setenv("TEST_VAR", "test_value")
+	os.Setenv("NESTED_VAR", "nested_value")
+	defer os.Unsetenv("TEST_VAR")
+	defer os.Unsetenv("NESTED_VAR")
+
+	type Nested struct {
+		NestedField   string
+		NestedCommand string
+		unexported    string
+	}
+
+	type TestStruct struct {
+		SimpleField  string
+		EnvField     string
+		CommandField string
+		MultiField   string
+		EmptyField   string
+		unexported   string
+		NestedStruct Nested
+	}
+
+	tests := []struct {
+		name    string
+		input   TestStruct
+		want    TestStruct
+		wantErr bool
+	}{
+		{
+			name: "basic substitution",
+			input: TestStruct{
+				SimpleField:  "hello",
+				EnvField:     "$TEST_VAR",
+				CommandField: "`echo hello`",
+				MultiField:   "$TEST_VAR and `echo command`",
+				EmptyField:   "",
+				NestedStruct: Nested{
+					NestedField:   "$NESTED_VAR",
+					NestedCommand: "`echo nested`",
+					unexported:    "should not change",
+				},
+				unexported: "should not change",
+			},
+			want: TestStruct{
+				SimpleField:  "hello",
+				EnvField:     "test_value",
+				CommandField: "hello",
+				MultiField:   "test_value and command",
+				EmptyField:   "",
+				NestedStruct: Nested{
+					NestedField:   "nested_value",
+					NestedCommand: "nested",
+					unexported:    "should not change",
+				},
+				unexported: "should not change",
+			},
+			wantErr: false,
+		},
+		{
+			name: "invalid command",
+			input: TestStruct{
+				CommandField: "`invalid_command_that_does_not_exist`",
+			},
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := SubstituteStringFields(tt.input)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("SubstituteStringFields() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !tt.wantErr && !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("SubstituteStringFields() = %+v, want %+v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestSubstituteStringFields_NonStruct(t *testing.T) {
+	_, err := SubstituteStringFields("not a struct")
+	if err == nil {
+		t.Error("SubstituteStringFields() should return error for non-struct input")
+	}
+}
+
+func TestSubstituteStringFields_NestedStructs(t *testing.T) {
+	type DeepNested struct {
+		Field string
+	}
+
+	type Nested struct {
+		Field      string
+		DeepNested DeepNested
+	}
+
+	type Root struct {
+		Field  string
+		Nested Nested
+	}
+
+	input := Root{
+		Field: "$TEST_VAR",
+		Nested: Nested{
+			Field: "`echo nested`",
+			DeepNested: DeepNested{
+				Field: "$NESTED_VAR",
+			},
+		},
+	}
+
+	// Set up environment
+	os.Setenv("TEST_VAR", "test_value")
+	os.Setenv("NESTED_VAR", "deep_nested_value")
+	defer os.Unsetenv("TEST_VAR")
+	defer os.Unsetenv("NESTED_VAR")
+
+	want := Root{
+		Field: "test_value",
+		Nested: Nested{
+			Field: "nested",
+			DeepNested: DeepNested{
+				Field: "deep_nested_value",
+			},
+		},
+	}
+
+	got, err := SubstituteStringFields(input)
+	if err != nil {
+		t.Fatalf("SubstituteStringFields() error = %v", err)
+	}
+
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("SubstituteStringFields() = %+v, want %+v", got, want)
+	}
+}
+
+func TestSubstituteStringFields_EmptyStruct(t *testing.T) {
+	type Empty struct{}
+
+	input := Empty{}
+	got, err := SubstituteStringFields(input)
+	if err != nil {
+		t.Fatalf("SubstituteStringFields() error = %v", err)
+	}
+
+	if !reflect.DeepEqual(got, input) {
+		t.Errorf("SubstituteStringFields() = %+v, want %+v", got, input)
+	}
+}