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 {