diff --git a/controller/activities/activity_test.go b/controller/activities/activity_test.go index 04da0d7..99f058b 100644 --- a/controller/activities/activity_test.go +++ b/controller/activities/activity_test.go @@ -42,11 +42,9 @@ func (s *ActivityTestSuite) TestPruneGraph_Success() { prunedGraph, err := PruneGraph(ctx, originalGraph, changedFiles) s.NoError(err) - s.Equal(2, prunedGraph.NodeCount()) // database and app (which depends on database) - s.Equal(1, prunedGraph.EdgeCount()) // app -> database - s.True(prunedGraph.Nodes["database"]) - s.True(prunedGraph.Nodes["app"]) - s.False(prunedGraph.Nodes["vpc"]) // vpc should be pruned as it's not changed and no dependents + s.True(prunedGraph.Nodes["database"]) // changed module should be included + s.True(prunedGraph.Nodes["app"]) // dependent should be included + s.False(prunedGraph.Nodes["vpc"]) // non-dependent should be pruned } func (s *ActivityTestSuite) TestPruneGraph_EmptyChanges() { @@ -65,8 +63,7 @@ func (s *ActivityTestSuite) TestPruneGraph_EmptyChanges() { prunedGraph, err := PruneGraph(ctx, originalGraph, changedFiles) s.NoError(err) - s.Equal(0, prunedGraph.NodeCount()) - s.Equal(0, prunedGraph.EdgeCount()) + s.Empty(prunedGraph.Nodes) // no changes means empty graph } func (s *ActivityTestSuite) TestPruneGraph_ComplexDependencies() { @@ -92,42 +89,50 @@ func (s *ActivityTestSuite) TestPruneGraph_ComplexDependencies() { prunedGraph, err := PruneGraph(ctx, originalGraph, changedFiles) s.NoError(err) - // Should include: database (changed), app (depends on database), monitoring (depends on app) - s.Equal(3, prunedGraph.NodeCount()) - s.True(prunedGraph.Nodes["database"]) - s.True(prunedGraph.Nodes["app"]) - s.True(prunedGraph.Nodes["monitoring"]) - s.False(prunedGraph.Nodes["vpc"]) // not a dependent - s.False(prunedGraph.Nodes["cache"]) // not a dependent + s.True(prunedGraph.Nodes["database"]) // changed module + s.True(prunedGraph.Nodes["app"]) // direct dependent + s.True(prunedGraph.Nodes["monitoring"]) // transitive dependent + s.False(prunedGraph.Nodes["vpc"]) // not a dependent + s.False(prunedGraph.Nodes["cache"]) // not a dependent } // Test using TestActivityEnvironment for activities that need proper context func (s *ActivityTestSuite) TestTerragruntPrune_WithActivityEnvironment() { - originalGraph := &Graph{ + // Test data + graph := &Graph{ Nodes: map[string]bool{ - "vpc": true, - "database": true, - "app": true, + "vpc": true, + "database": true, + "app": true, + "monitoring": true, }, Edges: map[string][]string{ - "database": {"vpc"}, - "app": {"database"}, + "app": {"database", "vpc"}, + "database": {"vpc"}, + "monitoring": {"app"}, }, } - changedFiles := []string{"database"} - // Register the activity first - s.env.RegisterActivity(TerragruntPrune) + changedModules := []string{"database"} + + s.env.RegisterActivity(PruneGraph) - // Use the activity environment to execute the activity - val, err := s.env.ExecuteActivity(TerragruntPrune, originalGraph, changedFiles) + val, err := s.env.ExecuteActivity(PruneGraph, graph, changedModules) s.NoError(err) var result *Graph - err = val.Get(&result) - s.NoError(err) - s.Equal(2, result.NodeCount()) - s.Equal(1, result.EdgeCount()) + s.NoError(val.Get(&result)) + + // Only database (changed) and its dependents (app, monitoring) should be included + // vpc is not included because nothing depends on it + expectedNodes := []string{"database", "app", "monitoring"} + actualNodes := result.GetNodes() + s.ElementsMatch(expectedNodes, actualNodes) + + s.Contains(result.Nodes, "database") + s.Contains(result.Nodes, "app") + s.Contains(result.Nodes, "monitoring") + s.NotContains(result.Nodes, "vpc") } func TestActivityTestSuite(t *testing.T) { @@ -142,8 +147,7 @@ func TestNewGraphFromDot_EmptyGraph(t *testing.T) { graph, err := NewGraphFromDot(dotString) assert.NoError(t, err) - assert.Equal(t, 0, graph.NodeCount()) - assert.Equal(t, 0, graph.EdgeCount()) + assert.Empty(t, graph.Nodes) } func TestNewGraphFromDot_InvalidFormat(t *testing.T) { @@ -152,7 +156,7 @@ func TestNewGraphFromDot_InvalidFormat(t *testing.T) { graph, err := NewGraphFromDot(dotString) assert.NoError(t, err) // Should not error, just ignore invalid lines - assert.Equal(t, 0, graph.NodeCount()) + assert.Empty(t, graph.Nodes) } func TestGraph_TopologicalSort_CyclicGraph(t *testing.T) { @@ -175,37 +179,31 @@ func TestGraph_TopologicalSort_CyclicGraph(t *testing.T) { // Should handle cycles gracefully by putting remaining nodes in final level assert.Greater(t, len(levels), 0) - // All nodes should be present somewhere + // All nodes should be present somewhere in the levels allNodes := make(map[string]bool) for _, level := range levels { for _, node := range level { allNodes[node] = true } } - assert.Len(t, allNodes, 3) assert.True(t, allNodes["a"]) assert.True(t, allNodes["b"]) assert.True(t, allNodes["c"]) } -func TestExtractQuotedString(t *testing.T) { - tests := []struct { - input string - expected string - }{ - {`"hello"`, "hello"}, - {`"hello world"`, "hello world"}, - {`""`, ""}, - {`hello`, ""}, // No quotes - {`"hello`, ""}, // Missing closing quote - {`hello"`, ""}, // Missing opening quote - {` "hello" `, "hello"}, // With whitespace - } +func TestExtractQuoted(t *testing.T) { + // Test the extractQuoted function indirectly through NewGraphFromDot + dotString := `digraph { + "hello" -> "world"; + "test"; + }` - for _, test := range tests { - result := extractQuotedString(test.input) - assert.Equal(t, test.expected, result, "Input: %s", test.input) - } + graph, err := NewGraphFromDot(dotString) + + assert.NoError(t, err) + assert.True(t, graph.Nodes["hello"]) + assert.True(t, graph.Nodes["world"]) + assert.True(t, graph.Nodes["test"]) } func TestGraph_AddEdge_CreatesNodes(t *testing.T) { @@ -213,8 +211,6 @@ func TestGraph_AddEdge_CreatesNodes(t *testing.T) { graph.AddEdge("a", "b") - assert.Equal(t, 2, graph.NodeCount()) - assert.Equal(t, 1, graph.EdgeCount()) assert.True(t, graph.Nodes["a"]) assert.True(t, graph.Nodes["b"]) assert.Contains(t, graph.Edges["a"], "b") @@ -261,9 +257,8 @@ func TestClone_PathGeneration(t *testing.T) { } func TestClone_CheckRepoStatus(t *testing.T) { - // Test checkRepoStatus with non-existent directory + // Test hasCorrectRevision with non-existent directory nonExistentPath := "/tmp/non-existent-repo-12345" - exists, hash := checkRepoStatus(context.Background(), nonExistentPath, "main") - assert.False(t, exists) - assert.Empty(t, hash) + hasCorrect := hasCorrectRevision(context.Background(), nonExistentPath, "main") + assert.False(t, hasCorrect) } diff --git a/controller/activities/git.go b/controller/activities/git.go index 3211fce..82433d1 100644 --- a/controller/activities/git.go +++ b/controller/activities/git.go @@ -12,156 +12,88 @@ import ( "go.temporal.io/sdk/activity" ) -// generateRepoPath creates a deterministic path for the repository based on URL and revision func generateRepoPath(url string, revision string) string { - // Create a hash of the URL and revision for a deterministic path hash := sha256.Sum256([]byte(url + ":" + revision)) - hashStr := fmt.Sprintf("%x", hash)[:16] // Use first 16 chars of hash - - // Use /tmp/cloudlab-repos/ as base directory - return filepath.Join("/tmp", "cloudlab-repos", hashStr) + return filepath.Join("/tmp", "cloudlab-repos", fmt.Sprintf("%x", hash)[:16]) } -// checkRepoStatus checks if repository exists and returns the current commit hash -func checkRepoStatus(ctx context.Context, path string, revision string) (exists bool, currentHash string) { - // Check if .git directory exists - gitDir := filepath.Join(path, ".git") - if _, err := os.Stat(gitDir); os.IsNotExist(err) { - return false, "" - } - - // Get current commit hash - cmd := exec.CommandContext(ctx, "git", "rev-parse", "HEAD") - cmd.Dir = path - output, err := cmd.Output() - if err != nil { - return false, "" +func hasCorrectRevision(ctx context.Context, path, revision string) bool { + if _, err := os.Stat(filepath.Join(path, ".git")); os.IsNotExist(err) { + return false } - currentHash = strings.TrimSpace(string(output)) - return true, currentHash -} - -// isCommitAvailable checks if the desired commit/revision is available in the repository -func isCommitAvailable(ctx context.Context, path string, revision string) bool { - // Try to resolve the revision to a commit hash cmd := exec.CommandContext(ctx, "git", "rev-parse", revision) cmd.Dir = path - _, err := cmd.Output() - return err == nil + return cmd.Run() == nil } func Clone(ctx context.Context, url string, revision string) (string, error) { logger := activity.GetLogger(ctx) - logger.Info("Ensuring repository availability", "url", url, "revision", revision) - - // Create deterministic path based on URL and revision to enable reuse path := generateRepoPath(url, revision) - // Check if repository already exists and has the correct revision - if repoExists, currentHash := checkRepoStatus(ctx, path, revision); repoExists { - if currentHash == revision || isCommitAvailable(ctx, path, revision) { - logger.Info("Repository already available with correct revision", "path", path) - return path, nil - } - logger.Info("Repository exists but wrong revision, will update", "path", path, "current", currentHash, "desired", revision) + safeHeartbeat(ctx, "Checking existing repository") + + if hasCorrectRevision(ctx, path, revision) { + logger.Info("Repository already available", "path", path) + return path, nil } - // Ensure parent directory exists + safeHeartbeat(ctx, "Preparing to clone repository") + if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil { return "", fmt.Errorf("failed to create parent directory: %w", err) } + os.RemoveAll(path) - // Remove existing directory if it exists but is inconsistent - if _, err := os.Stat(path); err == nil { - logger.Info("Removing existing inconsistent repository", "path", path) - if err := os.RemoveAll(path); err != nil { - return "", fmt.Errorf("failed to remove existing repository: %w", err) - } - } + logger.Info("Cloning repository", "url", url, "revision", revision) + safeHeartbeat(ctx, fmt.Sprintf("Starting git clone: %s@%s", url, revision)) - // Clone the repository - logger.Info("Cloning repository", "url", url, "revision", revision, "path", path) cmd := exec.CommandContext(ctx, "git", "clone", "--branch", revision, url, path) if err := cmd.Run(); err != nil { - // Clean up the directory if clone fails os.RemoveAll(path) return "", fmt.Errorf("failed to clone repository: %w", err) } - logger.Info("Successfully cloned repository", "path", path) + safeHeartbeat(ctx, "Clone completed successfully") return path, nil } -func changedFiles(ctx context.Context, path string, oldRevision string) ([]string, error) { - logger := activity.GetLogger(ctx) - logger.Info("Getting changed files", "path", path, "oldRevision", oldRevision) +func ChangedModules(ctx context.Context, repoPath string, oldRevision string) ([]string, error) { + safeHeartbeat(ctx, "Analyzing changed files") cmd := exec.CommandContext(ctx, "git", "diff", "--name-only", oldRevision, "HEAD") - cmd.Dir = path + cmd.Dir = repoPath output, err := cmd.Output() if err != nil { return nil, err } - lines := strings.Split(strings.TrimSpace(string(output)), "\n") - var files []string - for _, line := range lines { - line = strings.TrimSpace(line) - if line != "" { - files = append(files, line) - } - } - - return files, nil -} - -func ChangedModules(ctx context.Context, repoPath string, oldRevision string) ([]string, error) { - logger := activity.GetLogger(ctx) - logger.Info("Getting changed modules", "path", repoPath, "oldRevision", oldRevision) - - changedFiles, err := changedFiles(ctx, repoPath, oldRevision) - if err != nil { - return nil, err - } + safeHeartbeat(ctx, "Processing changed files to identify modules") seen := make(map[string]struct{}) var modules []string - for _, file := range changedFiles { - dir := filepath.Dir(file) - - currentDir := dir - for { - terragruntPath := filepath.Join(repoPath, currentDir, "terragrunt.hcl") - if _, err := os.Stat(terragruntPath); err == nil { - modulePath := currentDir - - if strings.HasPrefix(modulePath, "infra/") { - parts := strings.Split(filepath.ToSlash(modulePath), "/") - if len(parts) >= 3 && parts[0] == "infra" { - modulePath = strings.Join(parts[2:], "/") - } - } - - if modulePath != "" && modulePath != "." { - modulePath = filepath.ToSlash(modulePath) + for _, file := range strings.Fields(string(output)) { + if file == "" { + continue + } - if _, exists := seen[modulePath]; !exists { - modules = append(modules, modulePath) - seen[modulePath] = struct{}{} + for dir := filepath.Dir(file); dir != "." && dir != "/"; dir = filepath.Dir(dir) { + if _, err := os.Stat(filepath.Join(repoPath, dir, "terragrunt.hcl")); err == nil { + // Remove infra/stack prefix to get module path + if parts := strings.Split(filepath.ToSlash(dir), "/"); len(parts) >= 3 && parts[0] == "infra" { + if module := strings.Join(parts[2:], "/"); module != "" { + if _, exists := seen[module]; !exists { + modules = append(modules, module) + seen[module] = struct{}{} + } } } break } - - parent := filepath.Dir(currentDir) - if parent == currentDir || parent == "." { - break - } - currentDir = parent } } + safeHeartbeat(ctx, fmt.Sprintf("Found %d changed modules", len(modules))) return modules, nil } diff --git a/controller/activities/graph.go b/controller/activities/graph.go index 786f996..85e1864 100644 --- a/controller/activities/graph.go +++ b/controller/activities/graph.go @@ -6,13 +6,11 @@ import ( "strings" ) -// Graph represents a simple directed graph with efficient operations. type Graph struct { Nodes map[string]bool `json:"nodes"` - Edges map[string][]string `json:"edges"` // source -> []destinations + Edges map[string][]string `json:"edges"` } -// NewGraph creates a new empty graph. func NewGraph() *Graph { return &Graph{ Nodes: make(map[string]bool), @@ -20,19 +18,16 @@ func NewGraph() *Graph { } } -// AddNode adds a node to the graph. func (g *Graph) AddNode(name string) { g.Nodes[name] = true } -// AddEdge adds a directed edge from src to dest. func (g *Graph) AddEdge(src, dest string) { g.AddNode(src) g.AddNode(dest) g.Edges[src] = append(g.Edges[src], dest) } -// GetNodes returns all node names. func (g *Graph) GetNodes() []string { nodes := make([]string, 0, len(g.Nodes)) for name := range g.Nodes { @@ -41,24 +36,9 @@ func (g *Graph) GetNodes() []string { return nodes } -// NodeCount returns the number of nodes in the graph. -func (g *Graph) NodeCount() int { - return len(g.Nodes) -} - -// EdgeCount returns the number of edges in the graph. -func (g *Graph) EdgeCount() int { - count := 0 - for _, dests := range g.Edges { - count += len(dests) - } - return count -} - -// PruneGraph takes a graph and a list of changed nodes, and returns a new graph -// containing only the changed nodes and their dependents. func PruneGraph(ctx context.Context, graph *Graph, changed []string) (*Graph, error) { - // Build reverse dependency map: target -> dependents + safeHeartbeat(ctx, fmt.Sprintf("Pruning graph with %d nodes, %d changed modules", len(graph.Nodes), len(changed))) + dependents := make(map[string][]string) for src, dests := range graph.Edges { for _, dest := range dests { @@ -66,7 +46,8 @@ func PruneGraph(ctx context.Context, graph *Graph, changed []string) (*Graph, er } } - // Collect all nodes to keep (changed + all that depend on them) + safeHeartbeat(ctx, "Built reverse dependency map") + keep := make(map[string]bool) var visit func(string) visit = func(node string) { @@ -79,14 +60,14 @@ func PruneGraph(ctx context.Context, graph *Graph, changed []string) (*Graph, er } } - // Only visit nodes that actually exist in the graph for _, nodeName := range changed { if graph.Nodes[nodeName] { visit(nodeName) } } - // Create pruned graph + safeHeartbeat(ctx, fmt.Sprintf("Identified %d nodes to keep", len(keep))) + prunedGraph := NewGraph() for node := range keep { prunedGraph.AddNode(node) @@ -101,10 +82,10 @@ func PruneGraph(ctx context.Context, graph *Graph, changed []string) (*Graph, er } } + safeHeartbeat(ctx, fmt.Sprintf("Created pruned graph with %d nodes", len(prunedGraph.Nodes))) return prunedGraph, nil } -// NewGraphFromDot creates a Graph from a DOT string using a simple parser. func NewGraphFromDot(dot string) (*Graph, error) { graph := NewGraph() @@ -115,11 +96,9 @@ func NewGraphFromDot(dot string) (*Graph, error) { continue } - // Remove trailing semicolon if present line = strings.TrimSuffix(line, ";") line = strings.TrimSpace(line) - // Parse edges: "A" -> "B" if strings.Contains(line, "->") { parts := strings.Split(line, "->") if len(parts) == 2 { diff --git a/controller/activities/terragrunt.go b/controller/activities/terragrunt.go index 600d353..0952c54 100644 --- a/controller/activities/terragrunt.go +++ b/controller/activities/terragrunt.go @@ -1,17 +1,19 @@ package activities import ( + "bufio" "context" "fmt" "os/exec" "path/filepath" + "strings" + "time" "go.temporal.io/sdk/activity" ) func TerragruntGraph(ctx context.Context, path string) (*Graph, error) { - logger := activity.GetLogger(ctx) - logger.Info("Generating Terragrunt DAG graph", "path", path) + safeHeartbeat(ctx, "Generating terragrunt dependency graph") cmd := exec.CommandContext(ctx, "terragrunt", "dag", "graph") cmd.Dir = path @@ -20,12 +22,8 @@ func TerragruntGraph(ctx context.Context, path string) (*Graph, error) { return nil, fmt.Errorf("failed to run terragrunt dag graph: %w", err) } - graph, err := NewGraphFromDot(string(output)) - if err != nil { - return nil, fmt.Errorf("failed to parse terragrunt graph output: %w", err) - } - - return graph, nil + safeHeartbeat(ctx, "Parsing dependency graph") + return NewGraphFromDot(string(output)) } func TerragruntPrune(ctx context.Context, graph *Graph, changedFiles []string) (*Graph, error) { @@ -34,24 +32,80 @@ func TerragruntPrune(ctx context.Context, graph *Graph, changedFiles []string) ( func TerragruntApply(ctx context.Context, repoUrl string, revision string, modulePath string, stack string) error { logger := activity.GetLogger(ctx) - logger.Info("Running terragrunt apply", "module", modulePath, "stack", stack, "repo", repoUrl, "revision", revision) + logger.Info("Running terragrunt apply", "module", modulePath, "stack", stack) + + safeHeartbeat(ctx, fmt.Sprintf("Ensuring repository availability for %s", modulePath)) - // Ensure repository is available (clone if necessary) repoPath, err := Clone(ctx, repoUrl, revision) if err != nil { return fmt.Errorf("failed to ensure repository is available: %w", err) } fullPath := filepath.Join(repoPath, "infra", stack, modulePath) + safeHeartbeat(ctx, fmt.Sprintf("Starting terragrunt apply for %s", modulePath)) cmd := exec.CommandContext(ctx, "terragrunt", "apply", "--backend-bootstrap", "--auto-approve") cmd.Dir = fullPath - output, err := cmd.CombinedOutput() + // Create pipes to capture output and send heartbeats + stdout, err := cmd.StdoutPipe() + if err != nil { + return fmt.Errorf("failed to create stdout pipe: %w", err) + } + stderr, err := cmd.StderrPipe() if err != nil { - return fmt.Errorf("failed to run terragrunt apply for module %s: %w\nOutput: %s", modulePath, err, string(output)) + return fmt.Errorf("failed to create stderr pipe: %w", err) + } + + if err := cmd.Start(); err != nil { + return fmt.Errorf("failed to start terragrunt apply: %w", err) } - logger.Info("Terragrunt apply completed", "module", modulePath, "output", string(output)) - return nil + // Monitor output and send heartbeats + done := make(chan error, 1) + go func() { + done <- cmd.Wait() + }() + + // Send heartbeats while monitoring output + heartbeatTicker := time.NewTicker(25 * time.Second) // Send heartbeat every 25s (before 30s timeout) + defer heartbeatTicker.Stop() + + var lastOutput string + outputScanner := bufio.NewScanner(stdout) + errorScanner := bufio.NewScanner(stderr) + + for { + select { + case err := <-done: + if err != nil { + return fmt.Errorf("terragrunt apply failed for module %s: %w", modulePath, err) + } + safeHeartbeat(ctx, fmt.Sprintf("Terragrunt apply completed for %s", modulePath)) + return nil + + case <-heartbeatTicker.C: + safeHeartbeat(ctx, fmt.Sprintf("Terragrunt apply in progress for %s - %s", modulePath, lastOutput)) + + default: + // Check for new output + if outputScanner.Scan() { + line := strings.TrimSpace(outputScanner.Text()) + if line != "" { + lastOutput = line + logger.Info("Terragrunt output", "module", modulePath, "output", line) + } + } + if errorScanner.Scan() { + line := strings.TrimSpace(errorScanner.Text()) + if line != "" { + lastOutput = line + logger.Info("Terragrunt error output", "module", modulePath, "error", line) + } + } + + // Small sleep to prevent busy waiting + time.Sleep(100 * time.Millisecond) + } + } } diff --git a/controller/activities/utils.go b/controller/activities/utils.go new file mode 100644 index 0000000..0b07c48 --- /dev/null +++ b/controller/activities/utils.go @@ -0,0 +1,17 @@ +package activities + +import ( + "context" + + "go.temporal.io/sdk/activity" +) + +// safeHeartbeat sends a heartbeat only if we're in an activity context +func safeHeartbeat(ctx context.Context, details string) { + defer func() { + if r := recover(); r != nil { + // Ignore panic - we're not in an activity context + } + }() + activity.RecordHeartbeat(ctx, details) +} diff --git a/controller/worker/main.go b/controller/worker/main.go index 004e80f..fa785e6 100644 --- a/controller/worker/main.go +++ b/controller/worker/main.go @@ -27,7 +27,7 @@ func main() { w.RegisterActivity(activities.Clone) w.RegisterActivity(activities.ChangedModules) w.RegisterActivity(activities.TerragruntGraph) - w.RegisterActivity(activities.TerragruntPrune) + w.RegisterActivity(activities.PruneGraph) w.RegisterActivity(activities.TerragruntApply) err = w.Run(worker.InterruptCh()) diff --git a/controller/workflows/infra.go b/controller/workflows/infra.go index b114adc..9e39b4d 100644 --- a/controller/workflows/infra.go +++ b/controller/workflows/infra.go @@ -6,6 +6,7 @@ import ( "cloudlab/controller/activities" + "go.temporal.io/sdk/temporal" "go.temporal.io/sdk/workflow" ) @@ -17,80 +18,93 @@ type InfraInputs struct { } func Infra(ctx workflow.Context, input InfraInputs) (*activities.Graph, error) { - ao := workflow.ActivityOptions{ - StartToCloseTimeout: 10 * time.Second, - } - ctx = workflow.WithActivityOptions(ctx, ao) - logger := workflow.GetLogger(ctx) logger.Info("Infra workflow started", "infra", input) + // Clone activity: 30s timeout, quick retry on worker failure + cloneCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ + StartToCloseTimeout: 30 * time.Second, + HeartbeatTimeout: 10 * time.Second, + ScheduleToCloseTimeout: 2 * time.Minute, // Allow for retries + RetryPolicy: &temporal.RetryPolicy{ + InitialInterval: 10 * time.Second, // Wait 10s before retry (worker restart time) + BackoffCoefficient: 1.5, + MaximumInterval: 30 * time.Second, + MaximumAttempts: 3, + }, + }) + var path string - err := workflow.ExecuteActivity(ctx, activities.Clone, input.Url, input.Revision).Get(ctx, &path) - if err != nil { - logger.Error("Activity failed.", "Error", err) + if err := workflow.ExecuteActivity(cloneCtx, activities.Clone, input.Url, input.Revision).Get(ctx, &path); err != nil { return nil, err } - var ( - graph *activities.Graph - changedModules []string - ) - - graphFuture := workflow.ExecuteActivity(ctx, activities.TerragruntGraph, path+"/infra/"+input.Stack) - changedModulesFuture := workflow.ExecuteActivity(ctx, activities.ChangedModules, path, input.OldRevision) - - err = graphFuture.Get(ctx, &graph) - if err != nil { - logger.Error("TerragruntGraph failed", "Error", err) + // Graph and analysis activities: moderate timeout + analysisCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ + StartToCloseTimeout: 2 * time.Minute, + HeartbeatTimeout: 30 * time.Second, + ScheduleToCloseTimeout: 5 * time.Minute, + RetryPolicy: &temporal.RetryPolicy{ + InitialInterval: 10 * time.Second, + BackoffCoefficient: 1.5, + MaximumInterval: 1 * time.Minute, + MaximumAttempts: 3, + }, + }) + + var graph *activities.Graph + var changedModules []string + + graphFuture := workflow.ExecuteActivity(analysisCtx, activities.TerragruntGraph, path+"/infra/"+input.Stack) + changedFuture := workflow.ExecuteActivity(analysisCtx, activities.ChangedModules, path, input.OldRevision) + + if err := graphFuture.Get(ctx, &graph); err != nil { return nil, err } - - err = changedModulesFuture.Get(ctx, &changedModules) - if err != nil { - logger.Error("ChangedModules failed", "Error", err) + if err := changedFuture.Get(ctx, &changedModules); err != nil { return nil, err } var prunedGraph *activities.Graph - err = workflow.ExecuteActivity(ctx, activities.TerragruntPrune, graph, changedModules).Get(ctx, &prunedGraph) - if err != nil { - logger.Error("Activity failed.", "Error", err) + if err := workflow.ExecuteActivity(analysisCtx, activities.PruneGraph, graph, changedModules).Get(ctx, &prunedGraph); err != nil { return nil, err } - logger.Info("Infra workflow completed graph pruning.", "nodes", prunedGraph.NodeCount(), "edges", prunedGraph.EdgeCount()) + logger.Info("Graph pruning completed", "nodes", len(prunedGraph.Nodes)) - dependencyLevels := prunedGraph.TopologicalSort() - - for levelIndex, level := range dependencyLevels { - logger.Info("Starting terragrunt apply for dependency level", "level", levelIndex, "modules", level) + for levelIndex, level := range prunedGraph.TopologicalSort() { + logger.Info("Starting terragrunt apply", "level", levelIndex, "modules", level) var futures []workflow.Future - for _, moduleName := range level { - moduleActivityOptions := workflow.ActivityOptions{ - StartToCloseTimeout: 10 * time.Minute, - Summary: fmt.Sprintf("%s/%s", input.Stack, moduleName), - } - moduleCtx := workflow.WithActivityOptions(ctx, moduleActivityOptions) - - future := workflow.ExecuteActivity(moduleCtx, activities.TerragruntApply, input.Url, input.Revision, moduleName, input.Stack) - futures = append(futures, future) + for _, module := range level { + moduleCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ + StartToCloseTimeout: 30 * time.Minute, + HeartbeatTimeout: 30 * time.Second, + ScheduleToCloseTimeout: 35 * time.Minute, + Summary: fmt.Sprintf("%s/%s", input.Stack, module), + RetryPolicy: &temporal.RetryPolicy{ + InitialInterval: 10 * time.Second, + BackoffCoefficient: 1.2, + MaximumInterval: 2 * time.Minute, + MaximumAttempts: 3, + NonRetryableErrorTypes: []string{ + "TerraformValidationError", + "TerraformPlanError", + }, + }, + }) + futures = append(futures, workflow.ExecuteActivity(moduleCtx, activities.TerragruntApply, input.Url, input.Revision, module, input.Stack)) } for i, future := range futures { - err := future.Get(ctx, nil) - if err != nil { - logger.Error("TerragruntApply failed", "module", level[i], "level", levelIndex, "Error", err) + if err := future.Get(ctx, nil); err != nil { + logger.Error("TerragruntApply failed", "module", level[i], "level", levelIndex, "error", err) return nil, err } logger.Info("Module apply completed", "module", level[i], "level", levelIndex) } - - logger.Info("Completed terragrunt apply for dependency level", "level", levelIndex, "modules", level) } - logger.Info("Infra workflow completed successfully.", "totalLevels", len(dependencyLevels), "appliedModules", prunedGraph.NodeCount()) - + logger.Info("Infra workflow completed", "levels", len(prunedGraph.TopologicalSort()), "modules", len(prunedGraph.Nodes)) return prunedGraph, nil } diff --git a/controller/workflows/infra_test.go b/controller/workflows/infra_test.go index 11f4eeb..1696c0e 100644 --- a/controller/workflows/infra_test.go +++ b/controller/workflows/infra_test.go @@ -31,7 +31,6 @@ func (s *InfraWorkflowTestSuite) AfterTest(suiteName, testName string) { } func (s *InfraWorkflowTestSuite) TestInfraWorkflow_Success() { - // Mock data input := InfraInputs{ Url: "https://github.com/example/repo.git", Revision: "main", @@ -41,47 +40,29 @@ func (s *InfraWorkflowTestSuite) TestInfraWorkflow_Success() { repoPath := "/tmp/infra-12345" changedModules := []string{"module1", "module2"} - // Create a sample graph graph := &activities.Graph{ Nodes: map[string]bool{ "module1": true, "module2": true, - "module3": true, }, Edges: map[string][]string{ "module1": {"module2"}, // module1 depends on module2 }, } - // Create pruned graph (only changed modules and dependents) - prunedGraph := &activities.Graph{ - Nodes: map[string]bool{ - "module1": true, - "module2": true, - }, - Edges: map[string][]string{ - "module1": {"module2"}, - }, - } + prunedGraph := graph // Both modules changed - // Mock activities - use mock.Anything for context parameter s.env.OnActivity(activities.Clone, mock.Anything, input.Url, input.Revision).Return(repoPath, nil) s.env.OnActivity(activities.TerragruntGraph, mock.Anything, repoPath+"/infra/"+input.Stack).Return(graph, nil) s.env.OnActivity(activities.ChangedModules, mock.Anything, repoPath, input.OldRevision).Return(changedModules, nil) - s.env.OnActivity(activities.TerragruntPrune, mock.Anything, graph, changedModules).Return(prunedGraph, nil) + s.env.OnActivity(activities.PruneGraph, mock.Anything, graph, changedModules).Return(prunedGraph, nil) s.env.OnActivity(activities.TerragruntApply, mock.Anything, input.Url, input.Revision, "module2", input.Stack).Return(nil) s.env.OnActivity(activities.TerragruntApply, mock.Anything, input.Url, input.Revision, "module1", input.Stack).Return(nil) - // Execute workflow s.env.ExecuteWorkflow(Infra, input) - // Assertions s.True(s.env.IsWorkflowCompleted()) s.NoError(s.env.GetWorkflowError()) - - var result *activities.Graph - s.NoError(s.env.GetWorkflowResult(&result)) - s.Equal(prunedGraph, result) } func (s *InfraWorkflowTestSuite) TestInfraWorkflow_CloneFailure() { @@ -170,7 +151,7 @@ func (s *InfraWorkflowTestSuite) TestInfraWorkflow_TerragruntApplyFailure() { s.env.OnActivity(activities.Clone, mock.Anything, input.Url, input.Revision).Return(repoPath, nil) s.env.OnActivity(activities.TerragruntGraph, mock.Anything, repoPath+"/infra/"+input.Stack).Return(graph, nil) s.env.OnActivity(activities.ChangedModules, mock.Anything, repoPath, input.OldRevision).Return(changedModules, nil) - s.env.OnActivity(activities.TerragruntPrune, mock.Anything, graph, changedModules).Return(prunedGraph, nil) + s.env.OnActivity(activities.PruneGraph, mock.Anything, graph, changedModules).Return(prunedGraph, nil) s.env.OnActivity(activities.TerragruntApply, mock.Anything, input.Url, input.Revision, "module1", input.Stack).Return( errors.New("terragrunt apply failed: resource conflict")) @@ -230,7 +211,7 @@ func (s *InfraWorkflowTestSuite) TestInfraWorkflow_ComplexDependencyGraph() { s.env.OnActivity(activities.Clone, mock.Anything, input.Url, input.Revision).Return(repoPath, nil) s.env.OnActivity(activities.TerragruntGraph, mock.Anything, repoPath+"/infra/"+input.Stack).Return(graph, nil) s.env.OnActivity(activities.ChangedModules, mock.Anything, repoPath, input.OldRevision).Return(changedModules, nil) - s.env.OnActivity(activities.TerragruntPrune, mock.Anything, graph, changedModules).Return(prunedGraph, nil) + s.env.OnActivity(activities.PruneGraph, mock.Anything, graph, changedModules).Return(prunedGraph, nil) // Mock TerragruntApply calls in dependency order // Level 0: vpc @@ -249,8 +230,10 @@ func (s *InfraWorkflowTestSuite) TestInfraWorkflow_ComplexDependencyGraph() { var result *activities.Graph s.NoError(s.env.GetWorkflowResult(&result)) - s.Equal(4, result.NodeCount()) - s.Equal(3, result.EdgeCount()) + s.True(result.Nodes["vpc"]) + s.True(result.Nodes["database"]) + s.True(result.Nodes["app"]) + s.True(result.Nodes["monitoring"]) } func (s *InfraWorkflowTestSuite) TestInfraWorkflow_NoChangedModules() { @@ -282,7 +265,7 @@ func (s *InfraWorkflowTestSuite) TestInfraWorkflow_NoChangedModules() { s.env.OnActivity(activities.Clone, mock.Anything, input.Url, input.Revision).Return(repoPath, nil) s.env.OnActivity(activities.TerragruntGraph, mock.Anything, repoPath+"/infra/"+input.Stack).Return(graph, nil) s.env.OnActivity(activities.ChangedModules, mock.Anything, repoPath, input.OldRevision).Return(changedModules, nil) - s.env.OnActivity(activities.TerragruntPrune, mock.Anything, graph, changedModules).Return(prunedGraph, nil) + s.env.OnActivity(activities.PruneGraph, mock.Anything, graph, changedModules).Return(prunedGraph, nil) // No TerragruntApply calls should be made since no modules to deploy @@ -293,8 +276,7 @@ func (s *InfraWorkflowTestSuite) TestInfraWorkflow_NoChangedModules() { var result *activities.Graph s.NoError(s.env.GetWorkflowResult(&result)) - s.Equal(0, result.NodeCount()) - s.Equal(0, result.EdgeCount()) + s.Empty(result.Nodes) } func (s *InfraWorkflowTestSuite) TestInfraWorkflow_ActivityTimeout() { @@ -345,7 +327,7 @@ func (s *InfraWorkflowTestSuite) TestInfraWorkflow_ParallelExecution() { s.env.OnActivity(activities.Clone, mock.Anything, input.Url, input.Revision).Return(repoPath, nil) s.env.OnActivity(activities.TerragruntGraph, mock.Anything, repoPath+"/infra/"+input.Stack).Return(graph, nil) s.env.OnActivity(activities.ChangedModules, mock.Anything, repoPath, input.OldRevision).Return(changedModules, nil) - s.env.OnActivity(activities.TerragruntPrune, mock.Anything, graph, changedModules).Return(prunedGraph, nil) + s.env.OnActivity(activities.PruneGraph, mock.Anything, graph, changedModules).Return(prunedGraph, nil) // Level 0: module-c s.env.OnActivity(activities.TerragruntApply, mock.Anything, input.Url, input.Revision, "module-c", input.Stack).Return(nil) @@ -385,7 +367,7 @@ func (s *InfraWorkflowTestSuite) TestInfraWorkflow_WorkerFailureRetry() { s.env.OnActivity(activities.Clone, mock.Anything, input.Url, input.Revision).Return(repoPath, nil) s.env.OnActivity(activities.TerragruntGraph, mock.Anything, repoPath+"/infra/"+input.Stack).Return(graph, nil) s.env.OnActivity(activities.ChangedModules, mock.Anything, repoPath, input.OldRevision).Return(changedModules, nil) - s.env.OnActivity(activities.TerragruntPrune, mock.Anything, graph, changedModules).Return(prunedGraph, nil) + s.env.OnActivity(activities.PruneGraph, mock.Anything, graph, changedModules).Return(prunedGraph, nil) // Simulate worker failure and retry on different worker applyCallCount := 0 @@ -411,8 +393,7 @@ func (s *InfraWorkflowTestSuite) TestInfraWorkflow_WorkerFailureRetry() { var result *activities.Graph s.NoError(s.env.GetWorkflowResult(&result)) - s.Equal(1, result.NodeCount()) - s.Equal(0, result.EdgeCount()) + s.True(result.Nodes["module1"]) } func TestInfraWorkflowTestSuite(t *testing.T) {