From 45f935b988cb8a0bc3fe4f00237d4b7e32aa14a3 Mon Sep 17 00:00:00 2001 From: Antoine Toussaint <312536060+antoinetoussaint-byte@users.noreply.github.com> Date: Tue, 3 Nov 2026 23:41:12 -0500 Subject: [PATCH] feat(gitops): observe prints the refusals a cell records against a delivered workload MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After a promotion merges and Argo CD syncs, a pod the cell's ValidatingAdmissionPolicy refuses is never created. Argo CD reports the application Progressing or Degraded and says nothing of why; the why is a FailedCreate event on the workload's controller (the Deployment's ReplicaSet, the StatefulSet, the DaemonSet, the Job), visible only in `kubectl describe rs`. `codefly deploy gitops observe` waited its whole budget and ended with "health observation on Argo CD timed out" — or, when the budget ran out during an `argocd app get`, with "signal: killed". `observe` now reads, while an application is still Progressing or Degraded, the events of the delivered workloads' controllers through the cluster the environment declares (its kubeconfig and context, resolved the way every other cluster call of the environment resolves them), and prints every FailedCreate verbatim, once, as it is first seen, naming the workload. An admission denial ends the observation at once, non-zero: no retry of the controller creates that pod. Any other FailedCreate is printed and the wait goes on; the timeout then names the workloads it was still waiting on, with Argo CD's health message, and every refusal seen. A cluster that cannot be read is reported once and the Argo CD health wait continues as before. The budget running out mid-call is now reported as the timeout and what it waited on, not as the killed call. The polling loop moves to awaitApplications, unchanged in what it awaits. Co-Authored-By: Claude Fable 5.1 --- cmd/deploy/gitops.go | 2 + docs/commands.md | 26 +++ pkg/control/gitops.go | 14 ++ pkg/gitops/observe.go | 154 +++++++++---- pkg/gitops/observe_workloads.go | 296 ++++++++++++++++++++++++ pkg/gitops/observe_workloads_test.go | 330 +++++++++++++++++++++++++++ pkg/gitops/types.go | 11 + 7 files changed, 784 insertions(+), 49 deletions(-) create mode 100644 pkg/gitops/observe_workloads.go create mode 100644 pkg/gitops/observe_workloads_test.go diff --git a/cmd/deploy/gitops.go b/cmd/deploy/gitops.go index f3a5a80b..1e1e0a29 100644 --- a/cmd/deploy/gitops.go +++ b/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 diff --git a/docs/commands.md b/docs/commands.md index ce31d68e..2bf44cfa 100644 --- a/docs/commands.md +++ b/docs/commands.md @@ -1215,6 +1215,32 @@ codefly deploy gitops rollback payments --env production \ --to-revision ``` +`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 diff --git a/pkg/control/gitops.go b/pkg/control/gitops.go index 1cbb0c37..cd5fd45f 100644 --- a/pkg/control/gitops.go +++ b/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) } diff --git a/pkg/gitops/observe.go b/pkg/gitops/observe.go index 7a56a751..cd2b33c5 100644 --- a/pkg/gitops/observe.go +++ b/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: + } + } + +} diff --git a/pkg/gitops/observe_workloads.go b/pkg/gitops/observe_workloads.go new file mode 100644 index 00000000..0b1c730b --- /dev/null +++ b/pkg/gitops/observe_workloads.go @@ -0,0 +1,296 @@ +package gitops + +// Argo CD reconciles a publication and reports a workload that never comes +// up as Progressing or Degraded, saying nothing of why. The why is on the +// workload's controller: a pod the cell's admission policy refuses is never +// created, so the ReplicaSet, StatefulSet, DaemonSet or Job records a +// FailedCreate event carrying the refusal, and nothing else does — not the +// Deployment, not the Application. Observation reads those events through +// the cluster the environment declares, prints each distinct refusal +// verbatim the first time it is seen, while it is still waiting, and ends +// the wait on an admission denial: no retry creates that pod. + +import ( + "context" + "encoding/json" + "fmt" + "io" + "sort" + "strings" + "time" +) + +// workloadKinds are the controllers a delivered workload is one of, and so +// the kinds whose events say why its pods are not created. +var workloadKinds = map[string]bool{kindDeployment: true, kindStatefulSet: true, kindDaemonSet: true, kindJob: true} + +// deliveredWorkload is one workload an Argo CD application reconciles. +type deliveredWorkload struct { + Namespace, Kind, Name string +} + +func (w deliveredWorkload) String() string { return w.Namespace + "/" + w.Kind + "/" + w.Name } + +// workloadRefusal is one event a controller recorded while failing to create +// a pod: the workload it concerns, the reason, and the message exactly as the +// cluster wrote it. +type workloadRefusal struct { + Workload deliveredWorkload + Reason string + Message string +} + +func (r *workloadRefusal) String() string { + return fmt.Sprintf("%s %s: %s", r.Workload, r.Reason, r.Message) +} + +// Denied reports whether the refusal is an admission denial — a +// ValidatingAdmissionPolicy or a webhook "denied request" — which no retry of +// the controller resolves. It is the denial's own wording that is matched, +// not the "is forbidden" every 403 carries: a quota exceeded or a service +// account the admission cache has not seen yet is forbidden too, and the +// controller's next retry creates that pod. +func (r *workloadRefusal) Denied() bool { + return strings.Contains(strings.ToLower(r.Message), "denied") +} + +// workloadObserver reads the refusals of an observation's delivered workloads +// through kubectl against the cluster the environment declares, and prints +// each distinct one once. +type workloadObserver struct { + kubeconfig string + kubeContext string + out io.Writer + seen map[string]bool + // refusals is every distinct refusal seen so far, in the order seen. + refusals []workloadRefusal + // readErr is the first failure to read the cluster; after one, the + // observer stops reading rather than failing every sweep the same way, + // and the timeout names it. + readErr error +} + +func newWorkloadObserver(request *ObserveRequest) *workloadObserver { + out := request.Output + if out == nil { + out = io.Discard + } + return &workloadObserver{kubeconfig: request.Kubeconfig, kubeContext: request.KubeContext, out: out, seen: map[string]bool{}} +} + +// denied is every admission denial seen so far. +func (o *workloadObserver) denied() []workloadRefusal { + var denied []workloadRefusal + for i := range o.refusals { + if o.refusals[i].Denied() { + denied = append(denied, o.refusals[i]) + } + } + return denied +} + +// inspect reads the controller events of every workload the applications +// deliver, prints the refusals not seen before, and returns them. A cluster +// that cannot be read is reported once and does not end the observation: +// Argo CD's own health is still awaited as before. +func (o *workloadObserver) inspect(ctx context.Context, apps []*argoApplication) []workloadRefusal { + if o.readErr != nil { + return nil + } + byNamespace := deliveredWorkloads(apps) + namespaces := make([]string, 0, len(byNamespace)) + for namespace := range byNamespace { + namespaces = append(namespaces, namespace) + } + sort.Strings(namespaces) + var fresh []workloadRefusal + for _, namespace := range namespaces { + refusals, err := o.namespaceRefusals(ctx, namespace, byNamespace[namespace]) + if err != nil { + o.readErr = err + fmt.Fprintf(o.out, "cannot read the delivered workloads' events: %v\n", err) + return fresh + } + for _, refusal := range refusals { + key := refusal.Workload.String() + "\x00" + refusal.Message + if o.seen[key] { + continue + } + o.seen[key] = true + o.refusals = append(o.refusals, refusal) + fresh = append(fresh, refusal) + fmt.Fprintf(o.out, "%s\n", &refusal) + } + } + return fresh +} + +// deliveredWorkloads is the workloads the applications reconcile, by +// namespace, as Argo CD lists them, each with the moment its application's +// current sync began: an event recorded before it belongs to a previous +// rollout — a ReplicaSet the Deployment keeps in its history, a Job re-synced +// under the same name — and says nothing of this one. An application that +// reports no start leaves its workloads unbounded. +func deliveredWorkloads(apps []*argoApplication) map[string]map[deliveredWorkload]time.Time { + byNamespace := map[string]map[deliveredWorkload]time.Time{} + for _, app := range apps { + since, _ := time.Parse(time.RFC3339, app.Status.OperationState.StartedAt) + for _, resource := range app.Status.Resources { + if !workloadKinds[resource.Kind] { + continue + } + if byNamespace[resource.Namespace] == nil { + byNamespace[resource.Namespace] = map[deliveredWorkload]time.Time{} + } + byNamespace[resource.Namespace][deliveredWorkload{Namespace: resource.Namespace, Kind: resource.Kind, Name: resource.Name}] = since + } + } + return byNamespace +} + +// kubeEvent is the part of a Kubernetes Event the refusal is read from. +type kubeEvent struct { + Type string `json:"type"` + Reason string `json:"reason"` + Message string `json:"message"` + InvolvedObject struct { + Kind string `json:"kind"` + Name string `json:"name"` + } `json:"involvedObject"` + LastTimestamp string `json:"lastTimestamp"` + EventTime string `json:"eventTime"` + Series struct { + LastObservedTime string `json:"lastObservedTime"` + } `json:"series"` +} + +// lastSeen is the latest moment the event was observed, over the three +// fields an Event carries it in; zero when it carries none that parses. +func (e *kubeEvent) lastSeen() time.Time { + var latest time.Time + for _, value := range []string{e.LastTimestamp, e.EventTime, e.Series.LastObservedTime} { + if at, err := time.Parse(time.RFC3339Nano, value); err == nil && at.After(latest) { + latest = at + } + } + return latest +} + +// kubeOwned is the part of a ReplicaSet that names the Deployment it serves. +type kubeOwned struct { + Metadata struct { + Name string `json:"name"` + OwnerReferences []struct { + Kind string `json:"kind"` + Name string `json:"name"` + } `json:"ownerReferences"` + } `json:"metadata"` +} + +// namespaceRefusals reads one namespace's events and returns those recorded +// against a delivered workload's controller since its sync began: a +// FailedCreate, or a warning that is an admission denial. A Deployment's pods +// are created by its ReplicaSet, so the ReplicaSets are read too and each +// event on one is attributed to the Deployment that owns it. An event whose +// time cannot be read is kept: shown once too often rather than hidden. +func (o *workloadObserver) namespaceRefusals(ctx context.Context, namespace string, workloads map[deliveredWorkload]time.Time) ([]workloadRefusal, error) { + var events struct { + Items []kubeEvent `json:"items"` + } + if err := o.get(ctx, namespace, "events", &events); err != nil { + return nil, err + } + owners := map[string]string{} + for workload := range workloads { + if workload.Kind != kindDeployment { + continue + } + var replicaSets struct { + Items []kubeOwned `json:"items"` + } + if err := o.get(ctx, namespace, "replicasets", &replicaSets); err != nil { + return nil, err + } + for _, replicaSet := range replicaSets.Items { + for _, owner := range replicaSet.Metadata.OwnerReferences { + if owner.Kind == kindDeployment { + owners[replicaSet.Metadata.Name] = owner.Name + } + } + } + break + } + var refusals []workloadRefusal + for i := range events.Items { + event := &events.Items[i] + refusal := workloadRefusal{Reason: event.Reason, Message: event.Message} + if event.Reason != "FailedCreate" && (event.Type != "Warning" || !refusal.Denied()) { + continue + } + kind, name := event.InvolvedObject.Kind, event.InvolvedObject.Name + if kind == kindReplicaSet { + kind, name = kindDeployment, owners[name] + } + refusal.Workload = deliveredWorkload{Namespace: namespace, Kind: kind, Name: name} + since, delivered := workloads[refusal.Workload] + if !delivered { + continue + } + if seen := event.lastSeen(); !since.IsZero() && !seen.IsZero() && seen.Before(since) { + continue + } + refusals = append(refusals, refusal) + } + return refusals, nil +} + +// get lists one resource kind of a namespace as JSON through kubectl, against +// the kubeconfig and context the environment declares; declaring neither +// reads kubectl's current context, as every other cluster call does. +func (o *workloadObserver) get(ctx context.Context, namespace, kind string, into any) error { + var args []string + if o.kubeconfig != "" { + args = append(args, "--kubeconfig", o.kubeconfig) + } + if o.kubeContext != "" { + args = append(args, "--context", o.kubeContext) + } + args = append(args, "get", kind, "-n", namespace, "-o", "json") + output, err := command(ctx, "", "kubectl", args...) + if err != nil { + return err + } + if err := json.Unmarshal([]byte(output), into); err != nil { + return fmt.Errorf("decode %s of namespace %s: %w", kind, namespace, err) + } + return nil +} + +// unhealthyResources names every resource the applications report as other +// than Healthy, with the message Argo CD attached, for the timeout to say +// which workload it waited on. +func unhealthyResources(apps []*argoApplication) []string { + var described []string + for _, app := range apps { + for _, resource := range app.Status.Resources { + if resource.Health.Status == "" || resource.Health.Status == healthyStatus { + continue + } + line := fmt.Sprintf("%s/%s/%s health=%s", resource.Namespace, resource.Kind, resource.Name, resource.Health.Status) + if resource.Health.Message != "" { + line += ": " + resource.Health.Message + } + described = append(described, line) + } + } + return described +} + +// describeRefusals joins refusals for an error message. +func describeRefusals(refusals []workloadRefusal) string { + described := make([]string, 0, len(refusals)) + for i := range refusals { + described = append(described, refusals[i].String()) + } + return strings.Join(described, "; ") +} diff --git a/pkg/gitops/observe_workloads_test.go b/pkg/gitops/observe_workloads_test.go new file mode 100644 index 00000000..65905bab --- /dev/null +++ b/pkg/gitops/observe_workloads_test.go @@ -0,0 +1,330 @@ +package gitops + +import ( + "bytes" + "context" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +const ( + cellDenial = `Error creating: pods "api-5d9f6c7b8-" is forbidden: ValidatingAdmissionPolicy 'cell-admission' with binding 'cell-admission' denied request: no approved execution matches this pod` + quotaFail = `Error creating: pods "api-5d9f6c7b8-" is forbidden: exceeded quota: compute-resources, requested: pods=1, used: pods=4, limited: pods=4` + syncBegan = "2026-11-03T10:00:00Z" +) + +func TestObserveEndsOnAnAdmissionDenialAndPrintsItVerbatim(t *testing.T) { + request := observedPublication(t) + request.Kubeconfig = "/tmp/blue.kubeconfig" + request.KubeContext = "gke_blue" + var out bytes.Buffer + request.Output = &out + installFakeArgoSequence(t, argoProjectJSON(request.Repository), progressingApplication(request)) + calls := installFakeKubectl(t, eventsJSON("ReplicaSet", "api-5d9f6c7b8", "FailedCreate", cellDenial), replicaSetsJSON("api-5d9f6c7b8", "api")) + + _, err := Observe(context.Background(), &request) + if err == nil { + t.Fatal("expected the observation to end on the denial") + } + want := "payments/Deployment/api FailedCreate: " + cellDenial + if !strings.Contains(err.Error(), want) || !strings.Contains(err.Error(), "no retry creates them") { + t.Fatalf("error = %v", err) + } + if got := strings.Count(out.String(), want); got != 1 { + t.Fatalf("refusal printed %d time(s) in %q, want once", got, out.String()) + } + log, readErr := os.ReadFile(calls) + if readErr != nil { + t.Fatal(readErr) + } + first := strings.SplitN(string(log), "\n", 2)[0] + if !strings.HasPrefix(first, "--kubeconfig /tmp/blue.kubeconfig --context gke_blue get events -n payments -o json") { + t.Fatalf("kubectl was called as %q, want the environment's kubeconfig and context", first) + } +} + +func TestObservePrintsAFailedCreateOnceAndKeepsWaitingForHealth(t *testing.T) { + request := observedPublication(t) + request.Timeout = 30 * time.Second + var out bytes.Buffer + request.Output = &out + healthy := argoApplicationJSON("payments-api", request.Repository, observedServicePath(request), request.Revision, "Healthy", "Succeeded") + installFakeArgoSequence(t, argoProjectJSON(request.Repository), progressingApplication(request), progressingApplication(request), progressingApplication(request), healthy) + calls := installFakeKubectl(t, eventsJSON("ReplicaSet", "api-5d9f6c7b8", "FailedCreate", quotaFail), replicaSetsJSON("api-5d9f6c7b8", "api")) + + if _, err := Observe(context.Background(), &request); err != nil { + t.Fatalf("a FailedCreate the controller retries past is not a verdict: %v", err) + } + want := "payments/Deployment/api FailedCreate: " + quotaFail + "\n" + if out.String() != want { + t.Fatalf("output = %q, want exactly one line %q", out.String(), want) + } + log, err := os.ReadFile(calls) + if err != nil { + t.Fatal(err) + } + if sweeps := strings.Count(string(log), "get events"); sweeps < 2 { + t.Fatalf("events read %d time(s) across three pending sweeps, want every pending sweep", sweeps) + } +} + +func TestObserveTimeoutNamesTheWorkloadsItWaitedOnAndTheirRefusals(t *testing.T) { + request := observedPublication(t) + request.Timeout = 2 * time.Second + request.PollInterval = 20 * time.Millisecond + installFakeArgoSequence(t, argoProjectJSON(request.Repository), progressingApplication(request)) + installFakeKubectl(t, eventsJSON("ReplicaSet", "api-5d9f6c7b8", "FailedCreate", quotaFail), replicaSetsJSON("api-5d9f6c7b8", "api")) + + _, err := Observe(context.Background(), &request) + if err == nil { + t.Fatal("expected the observation to time out") + } + for _, want := range []string{ + "timed out", + "still waiting on payments/Deployment/api health=Progressing: Waiting for rollout to finish: 0 of 1 updated replicas are available", + "the cluster refused payments/Deployment/api FailedCreate: " + quotaFail, + } { + if !strings.Contains(err.Error(), want) { + t.Fatalf("error %q lacks %q", err, want) + } + } +} + +func TestObserveAttributesStatefulSetAndJobEventsAndIgnoresOtherWorkloads(t *testing.T) { + request := observedPublication(t) + var out bytes.Buffer + request.Output = &out + application := fmt.Sprintf(`{ + "metadata":{"name":"payments-api"}, + "spec":{"project":"payments","source":{"repoURL":%q,"path":%q,"targetRevision":%q},"destination":{"server":"https://cluster.example.com","namespace":"payments"}}, + "status":{ + "sync":{"status":"Synced","revision":%q}, + "health":{"status":"Progressing"}, + "operationState":{"phase":"Succeeded","syncResult":{"revision":%q}}, + "resources":[ + {"group":"apps","kind":"StatefulSet","namespace":"payments","name":"store","health":{"status":"Progressing"}}, + {"group":"batch","kind":"Job","namespace":"payments","name":"migrate","health":{"status":"Progressing"}} + ] + } +}`, request.Repository, observedServicePath(request), request.Revision, request.Revision, request.Revision) + installFakeArgoSequence(t, projectAllowingWorkloads(request.Repository), application) + events := fmt.Sprintf(`{"items":[%s,%s,%s]}`, + eventJSON("StatefulSet", "store", "FailedCreate", "create Pod store-0 in StatefulSet store failed error: "+cellDenial), + eventJSON("Job", "migrate", "FailedCreate", "Error creating: "+cellDenial), + eventJSON("Job", "someone-elses", "FailedCreate", "Error creating: "+cellDenial), + ) + installFakeKubectl(t, events, `{"items":[]}`) + + _, err := Observe(context.Background(), &request) + if err == nil { + t.Fatal("expected the observation to end on the denials") + } + if !strings.Contains(err.Error(), "the pods of 2 workload(s)") { + t.Fatalf("error = %v", err) + } + for _, want := range []string{"payments/StatefulSet/store FailedCreate: create Pod store-0", "payments/Job/migrate FailedCreate: Error creating"} { + if !strings.Contains(out.String(), want) { + t.Fatalf("output %q lacks %q", out.String(), want) + } + } + if strings.Contains(out.String(), "someone-elses") { + t.Fatalf("output %q names a workload the publication did not deliver", out.String()) + } +} + +func TestObserveIgnoresRefusalsRecordedBeforeTheSyncBegan(t *testing.T) { + request := observedPublication(t) + request.Timeout = 30 * time.Second + var out bytes.Buffer + request.Output = &out + healthy := argoApplicationJSON("payments-api", request.Repository, observedServicePath(request), request.Revision, "Healthy", "Succeeded") + installFakeArgoSequence(t, argoProjectJSON(request.Repository), progressingApplication(request), progressingApplication(request), healthy) + // The previous publication's ReplicaSet, denied by the cell an hour ago and + // kept in the Deployment's history; this sync's ReplicaSet is rolling out. + events := fmt.Sprintf(`{"items":[%s,%s]}`, + eventAtJSON("ReplicaSet", "api-5d9f6c7b8", "FailedCreate", cellDenial, "2026-11-03T09:12:00Z"), + eventAtJSON("ReplicaSet", "api-7c4b9d2f1", "ScalingReplicaSet", "Scaled up replica set api-7c4b9d2f1 to 1", "2026-11-03T10:00:30Z"), + ) + installFakeKubectl(t, events, replicaSetsJSON("api-5d9f6c7b8", "api", "api-7c4b9d2f1", "api")) + + if _, err := Observe(context.Background(), &request); err != nil { + t.Fatalf("a denial of the previous rollout ended this one: %v", err) + } + if out.Len() != 0 { + t.Fatalf("output = %q, want nothing of a previous rollout", out.String()) + } +} + +func TestObserveKeepsARefusalWhoseTimeItCannotRead(t *testing.T) { + request := observedPublication(t) + installFakeArgoSequence(t, argoProjectJSON(request.Repository), progressingApplication(request)) + installFakeKubectl(t, eventsJSON("ReplicaSet", "api-5d9f6c7b8", "FailedCreate", cellDenial), replicaSetsJSON("api-5d9f6c7b8", "api")) + undated := strings.ReplaceAll(eventsJSON("ReplicaSet", "api-5d9f6c7b8", "FailedCreate", cellDenial), `"lastTimestamp":"2026-11-03T10:01:00Z",`, "") + t.Setenv("CODEFLY_TEST_KUBECTL_EVENTS", undated) + + _, err := Observe(context.Background(), &request) + if err == nil || !strings.Contains(err.Error(), cellDenial) { + t.Fatalf("error = %v, want the undated denial shown rather than hidden", err) + } +} + +func TestObserveReadsNoClusterEventsWhileArgoIsHealthy(t *testing.T) { + request := observedPublication(t) + installFakeArgo(t, argoProjectJSON(request.Repository), argoApplicationJSON( + "payments-api", request.Repository, observedServicePath(request), request.Revision, "Healthy", "Succeeded", + )) + calls := installFakeKubectl(t, `{"items":[]}`, `{"items":[]}`) + if _, err := Observe(context.Background(), &request); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(calls); !os.IsNotExist(err) { + t.Fatalf("kubectl was consulted for a healthy application (stat err = %v)", err) + } +} + +func TestObserveReportsAnUnreadableClusterOnceAndStillAwaitsArgo(t *testing.T) { + request := observedPublication(t) + request.Timeout = 2 * time.Second + request.PollInterval = 20 * time.Millisecond + var out bytes.Buffer + request.Output = &out + installFakeArgoSequence(t, argoProjectJSON(request.Repository), progressingApplication(request)) + bin := t.TempDir() + calls := filepath.Join(t.TempDir(), "calls") + script := "#!/bin/sh\necho called >> \"$CODEFLY_TEST_KUBECTL_CALLS\"\necho 'error: context \"gke_blue\" does not exist' >&2\nexit 1\n" + if err := os.WriteFile(filepath.Join(bin, "kubectl"), []byte(script), 0o755); err != nil { + t.Fatal(err) + } + t.Setenv("CODEFLY_TEST_KUBECTL_CALLS", calls) + t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH")) + + _, err := Observe(context.Background(), &request) + if err == nil || !strings.Contains(err.Error(), "timed out") || !strings.Contains(err.Error(), `events could not be read (kubectl`) { + t.Fatalf("error = %v", err) + } + if got := strings.Count(out.String(), "cannot read the delivered workloads' events"); got != 1 { + t.Fatalf("unreadable cluster reported %d time(s) in %q, want once", got, out.String()) + } + log, readErr := os.ReadFile(calls) + if readErr != nil { + t.Fatal(readErr) + } + if strings.Count(string(log), "called") != 1 { + t.Fatalf("kubectl retried after failing: %q", log) + } +} + +// progressingApplication is the payments-api application still rolling out +// its one Deployment, as Argo CD reports it: Progressing, and nothing of why. +func progressingApplication(request ObserveRequest) string { + return fmt.Sprintf(`{ + "metadata":{"name":"payments-api"}, + "spec":{"project":"payments","source":{"repoURL":%q,"path":%q,"targetRevision":%q},"destination":{"server":"https://cluster.example.com","namespace":"payments"}}, + "status":{ + "sync":{"status":"Synced","revision":%q}, + "health":{"status":"Progressing"}, + "operationState":{"phase":"Succeeded","startedAt":%q,"syncResult":{"revision":%q}}, + "resources":[{"group":"apps","kind":"Deployment","namespace":"payments","name":"api","health":{"status":"Progressing","message":"Waiting for rollout to finish: 0 of 1 updated replicas are available..."}}] + } +}`, request.Repository, observedServicePath(request), request.Revision, request.Revision, syncBegan, request.Revision) +} + +func projectAllowingWorkloads(repository string) string { + return fmt.Sprintf(`{ + "metadata":{"name":"payments"}, + "spec":{ + "sourceRepos":[%q], + "destinations":[{"server":"https://cluster.example.com","namespace":"payments"}], + "clusterResourceWhitelist":[], + "namespaceResourceWhitelist":[{"group":"apps","kind":"Deployment"},{"group":"apps","kind":"StatefulSet"},{"group":"batch","kind":"Job"}] + } +}`, repository) +} + +// eventJSON is an event recorded a minute into the sync. +func eventJSON(kind, name, reason, message string) string { + return eventAtJSON(kind, name, reason, message, "2026-11-03T10:01:00Z") +} + +func eventAtJSON(kind, name, reason, message, at string) string { + return fmt.Sprintf(`{"type":"Warning","reason":%q,"message":%q,"lastTimestamp":%q,"involvedObject":{"kind":%q,"name":%q,"namespace":"payments"}}`, reason, message, at, kind, name) +} + +func eventsJSON(kind, name, reason, message string) string { + return `{"items":[` + eventJSON(kind, name, reason, message) + `]}` +} + +func replicaSetsJSON(owned ...string) string { + items := make([]string, 0, len(owned)/2) + for i := 0; i+1 < len(owned); i += 2 { + items = append(items, fmt.Sprintf(`{"metadata":{"name":%q,"ownerReferences":[{"kind":"Deployment","name":%q}]}}`, owned[i], owned[i+1])) + } + return `{"items":[` + strings.Join(items, ",") + `]}` +} + +// installFakeArgoSequence answers each `argocd app get` with the next +// application in the sequence, repeating the last one. +func installFakeArgoSequence(t *testing.T, project string, applications ...string) { + t.Helper() + bin := t.TempDir() + dir := t.TempDir() + for index, application := range applications { + if err := os.WriteFile(filepath.Join(dir, fmt.Sprintf("app-%d.json", index+1)), []byte(application), 0o644); err != nil { + t.Fatal(err) + } + } + script := fmt.Sprintf(`#!/bin/sh +if [ "$1" = "proj" ]; then + printf '%%s\n' "$CODEFLY_TEST_ARGO_PROJECT" + exit 0 +fi +if [ "$1" = "cluster" ]; then + printf '%%s\n' "$CODEFLY_TEST_ARGO_CLUSTER" + exit 0 +fi +count=0 +if [ -f "$CODEFLY_TEST_ARGO_COUNT" ]; then count=$(cat "$CODEFLY_TEST_ARGO_COUNT"); fi +count=$((count + 1)) +printf '%%s' "$count" > "$CODEFLY_TEST_ARGO_COUNT" +if [ "$count" -gt %d ]; then count=%d; fi +cat "$CODEFLY_TEST_ARGO_SEQUENCE/app-$count.json" +`, len(applications), len(applications)) + if err := os.WriteFile(filepath.Join(bin, "argocd"), []byte(script), 0o755); err != nil { + t.Fatal(err) + } + t.Setenv("CODEFLY_TEST_ARGO_PROJECT", project) + t.Setenv("CODEFLY_TEST_ARGO_CLUSTER", `{"server":"https://cluster.example.com","name":"test","config":{"tls":true}}`) + t.Setenv("CODEFLY_TEST_ARGO_COUNT", filepath.Join(dir, "count")) + t.Setenv("CODEFLY_TEST_ARGO_SEQUENCE", dir) + t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH")) +} + +// installFakeKubectl answers `kubectl ... get events` and `get replicasets` +// with the given lists and logs every invocation's arguments, one per line, +// to the file it returns. +func installFakeKubectl(t *testing.T, events, replicaSets string) string { + t.Helper() + bin := t.TempDir() + calls := filepath.Join(t.TempDir(), "calls") + script := `#!/bin/sh +echo "$@" >> "$CODEFLY_TEST_KUBECTL_CALLS" +case " $* " in + *" get events "*) printf '%s\n' "$CODEFLY_TEST_KUBECTL_EVENTS"; exit 0 ;; + *" get replicasets "*) printf '%s\n' "$CODEFLY_TEST_KUBECTL_REPLICASETS"; exit 0 ;; +esac +echo "unexpected kubectl $*" >&2 +exit 2 +` + if err := os.WriteFile(filepath.Join(bin, "kubectl"), []byte(script), 0o755); err != nil { + t.Fatal(err) + } + t.Setenv("CODEFLY_TEST_KUBECTL_CALLS", calls) + t.Setenv("CODEFLY_TEST_KUBECTL_EVENTS", events) + t.Setenv("CODEFLY_TEST_KUBECTL_REPLICASETS", replicaSets) + t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH")) + return calls +} diff --git a/pkg/gitops/types.go b/pkg/gitops/types.go index e4dd7c62..b08bc81f 100644 --- a/pkg/gitops/types.go +++ b/pkg/gitops/types.go @@ -1,6 +1,7 @@ package gitops import ( + "io" "time" "github.com/codefly-dev/cli/pkg/delivery/signing" @@ -611,6 +612,16 @@ type ObserveRequest struct { Local bool Timeout time.Duration PollInterval time.Duration + // Kubeconfig and KubeContext are the cluster the environment declares, + // through which the delivered workloads' controller events are read + // while Argo CD reports an application Progressing or Degraded. Empty, + // kubectl's current context is read. + Kubeconfig string + KubeContext string + // Output is where each refusal the cluster records against a delivered + // workload is printed, verbatim and once, as it is first seen. Nil + // discards them; the error returned still names them. + Output io.Writer } type ReviewEvidence struct {