feat(gitops): observe prints the refusals a cell records against a delivered workload

cmd/deploy/gitops.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"os"
 	"slices"
 	"sort"
 	"strings"
@@ -270,6 +271,7 @@ var gitOpsObserveCmd = &cobra.Command{
 			Commit: publication.Commit, Tree: publication.Tree, RenderDigest: publication.RenderDigest,
 			Repository: publication.Repository, Path: publication.Path,
 			PullRequest: publication.PullRequest, Local: gitOpsLocal, Timeout: gitOpsTimeout,
+			Output: os.Stdout,
 		})
 		if err != nil {
 			return err

docs/commands.md

@@ -1215,6 +1215,32 @@ codefly deploy gitops rollback payments --env production \
   --to-revision <previous-reviewed-commit>
 ```
 
+`observe` waits, bounded by `--timeout`, for Argo CD to report every selected
+application Synced and Healthy at the reviewed revision — which is each
+delivered workload rolled out. While an application is still Progressing or
+Degraded, Argo CD says nothing of why, so `observe` reads the events of the
+delivered workloads' controllers (a Deployment's ReplicaSets, a StatefulSet, a
+DaemonSet, a Job) through the cluster the environment declares (`cluster:
+kubeconfig` and `context`; kubectl's current context when it declares none)
+and prints every `FailedCreate` recorded since the application's sync began
+verbatim, once, as it is first seen, naming the workload — an event older
+than the sync belongs to a previous rollout (a ReplicaSet the Deployment keeps
+in its history) and is not this one's:
+
+```
+payments/Deployment/api FailedCreate: Error creating: pods "api-5d9f6c7b8-" is forbidden: ValidatingAdmissionPolicy 'cell-admission' with binding 'cell-admission' denied request: no approved execution matches this pod
+```
+
+An admission denial — the cell's policy or a webhook, whose message says it
+`denied request` — ends the observation at once with a non-zero exit, because
+no retry of the controller creates that pod. Any other `FailedCreate` (a quota
+exceeded, a service account the admission cache has not seen yet — forbidden
+too, but retried past) is printed and the wait goes on; the timeout then
+names the workloads it was still waiting on, with
+the health message Argo CD attached, and every refusal seen. A cluster that
+cannot be read is reported once and does not end the observation: Argo CD's
+health is awaited as before, and the timeout says the events were unreadable.
+
 `render --all` renders every module the workspace composes, the host first,
 and derives the environment's cell file once at the end from all of them, so
 one command renders one composition and a stale sibling tree is never a reason

pkg/control/gitops.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/codefly-dev/cli/pkg/deployments"
 	"github.com/codefly-dev/cli/pkg/gitops"
 	"github.com/codefly-dev/cli/pkg/internal/mutationauthority"
 	"github.com/codefly-dev/cli/pkg/orchestration"
@@ -73,6 +74,19 @@ func (p *planeImpl) ObserveGitOps(ctx context.Context, request *gitops.ObserveRe
 		return gitops.ObserveResult{}, fmt.Errorf("select environment %q: %w", normalized.Environment, err)
 	}
 	normalized.Local = env.IsK3d()
+	// The refusals a cell records against a delivered workload are read
+	// through the cluster the environment declares, the same resolution
+	// every other cluster call of the environment uses.
+	if normalized.Kubeconfig == "" {
+		kubeconfig, err := deployments.GetK8sConfig(ctx, env)
+		if err != nil {
+			return gitops.ObserveResult{}, fmt.Errorf("resolve the kubeconfig of environment %q: %w", normalized.Environment, err)
+		}
+		normalized.Kubeconfig = kubeconfig
+	}
+	if normalized.KubeContext == "" && env.Cluster != nil {
+		normalized.KubeContext = env.Cluster.Context
+	}
 	return gitops.NewCoordinator().Observe(ctx, &normalized)
 }
 

pkg/gitops/observe.go

@@ -5,6 +5,7 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net/url"
 	"os"
@@ -59,6 +60,7 @@ type argoApplication struct {
 		} `json:"health"`
 		OperationState struct {
 			Phase      string `json:"phase"`
+			StartedAt  string `json:"startedAt"`
 			SyncResult struct {
 				Revision  string   `json:"revision"`
 				Revisions []string `json:"revisions"`
@@ -73,6 +75,10 @@ type argoApplication struct {
 			Kind      string `json:"kind"`
 			Namespace string `json:"namespace"`
 			Name      string `json:"name"`
+			Health    struct {
+				Status  string `json:"status"`
+				Message string `json:"message"`
+			} `json:"health"`
 		} `json:"resources"`
 	} `json:"status"`
 }
@@ -130,58 +136,15 @@ func Observe(ctx context.Context, input *ObserveRequest) (ObserveResult, error)
 	if err != nil {
 		return ObserveResult{}, err
 	}
-	timeout := request.Timeout
-	if timeout <= 0 {
-		timeout = 10 * time.Minute
+	last, err := awaitApplications(ctx, &project, request, servicePaths)
+	if err != nil {
+		return ObserveResult{}, err
 	}
-	interval := request.PollInterval
-	if interval <= 0 {
-		interval = 5 * time.Second
+	names := make([]string, 0, len(last))
+	for name := range last {
+		names = append(names, name)
 	}
-	observeCtx, cancel := context.WithTimeout(ctx, timeout)
-	defer cancel()
-
-	names := append([]string(nil), request.Applications...)
 	sort.Strings(names)
-	last := map[string]ApplicationEvidence{}
-	healthySweeps := 0
-	for {
-		current := map[string]ApplicationEvidence{}
-		allHealthy := true
-		for _, name := range names {
-			_, evidence, done, observeErr := observeApplication(observeCtx, &project, name, request, servicePaths)
-			if observeErr != nil {
-				return ObserveResult{}, observeErr
-			}
-			last[name] = evidence
-			if done {
-				current[name] = evidence
-			} else {
-				allHealthy = false
-			}
-		}
-		if allHealthy {
-			healthySweeps++
-		} else {
-			healthySweeps = 0
-		}
-		if healthySweeps == 2 {
-			last = current
-			break
-		}
-		timer := time.NewTimer(interval)
-		select {
-		case <-observeCtx.Done():
-			timer.Stop()
-			var states []string
-			for _, name := range names {
-				state := last[name]
-				states = append(states, fmt.Sprintf("%s(sync=%s health=%s operation=%s revision=%s)", name, state.Sync, state.Health, state.Operation, state.Revision))
-			}
-			return ObserveResult{}, fmt.Errorf("health observation on Argo CD timed out: %s", strings.Join(states, ", "))
-		case <-timer.C:
-		}
-	}
 
 	evidence := Evidence{
 		SchemaVersion: EvidenceSchemaVersion, Module: request.Module, Environment: request.Environment,
@@ -807,3 +770,96 @@ func observeReview(ctx context.Context, pullRequest, publishedCommit, repository
 	}
 	return evidence, nil
 }
+
+// awaitApplications polls every selected application until each has been
+// Synced, Healthy and Succeeded at the reviewed revision for two consecutive
+// sweeps, within the request's budget, and returns the evidence of the last
+// sweep. While an application is still Progressing or Degraded, the delivered
+// workloads' controllers are read for the refusals Argo CD does not report.
+func awaitApplications(ctx context.Context, project *argoProject, request *ObserveRequest, servicePaths map[string]struct{}) (map[string]ApplicationEvidence, error) {
+	timeout := request.Timeout
+	if timeout <= 0 {
+		timeout = 10 * time.Minute
+	}
+	interval := request.PollInterval
+	if interval <= 0 {
+		interval = 5 * time.Second
+	}
+	observeCtx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+
+	names := append([]string(nil), request.Applications...)
+	sort.Strings(names)
+	last := map[string]ApplicationEvidence{}
+	healthySweeps := 0
+	workloads := newWorkloadObserver(request)
+	var pending []*argoApplication
+	// timedOut says what the observation was still waiting on when the
+	// budget ran out: each application's state, the workloads Argo CD
+	// reports unhealthy, and every refusal the cluster recorded against them.
+	timedOut := func() error {
+		states := make([]string, 0, len(names))
+		for _, name := range names {
+			state := last[name]
+			states = append(states, fmt.Sprintf("%s(sync=%s health=%s operation=%s revision=%s)", name, state.Sync, state.Health, state.Operation, state.Revision))
+		}
+		message := fmt.Sprintf("health observation on Argo CD timed out: %s", strings.Join(states, ", "))
+		if unhealthy := unhealthyResources(pending); len(unhealthy) > 0 {
+			message += "; still waiting on " + strings.Join(unhealthy, "; ")
+		}
+		if len(workloads.refusals) > 0 {
+			message += "; the cluster refused " + describeRefusals(workloads.refusals)
+		}
+		if workloads.readErr != nil {
+			message += fmt.Sprintf("; the delivered workloads' events could not be read (%v)", workloads.readErr)
+		}
+		return errors.New(message)
+	}
+	for {
+		current := map[string]ApplicationEvidence{}
+		allHealthy := true
+		sweep := make([]*argoApplication, 0, len(names))
+		for _, name := range names {
+			app, evidence, done, observeErr := observeApplication(observeCtx, project, name, request, servicePaths)
+			if observeErr != nil {
+				if observeCtx.Err() != nil {
+					// The budget ran out mid-call: the verdict is the
+					// timeout and what it waited on, not the killed call.
+					return nil, timedOut()
+				}
+				return nil, observeErr
+			}
+			last[name] = evidence
+			if done {
+				current[name] = evidence
+			} else {
+				allHealthy = false
+				sweep = append(sweep, &app)
+			}
+		}
+		pending = sweep
+		if allHealthy {
+			healthySweeps++
+		} else {
+			healthySweeps = 0
+			// An application still Progressing or Degraded says nothing of
+			// why; its workloads' controllers do. A denial ends the wait: the
+			// pod it refused is never created, however long Argo CD waits.
+			workloads.inspect(observeCtx, pending)
+			if denied := workloads.denied(); len(denied) > 0 {
+				return nil, fmt.Errorf("the cluster refused the pods of %d workload(s) of this publication, and no retry creates them: %s", len(denied), describeRefusals(denied))
+			}
+		}
+		if healthySweeps == 2 {
+			return current, nil
+		}
+		timer := time.NewTimer(interval)
+		select {
+		case <-observeCtx.Done():
+			timer.Stop()
+			return nil, timedOut()
+		case <-timer.C:
+		}
+	}
+
+}