From 4d3c2b0205bcdc783f03a73ff875a47088a8c5d1 Mon Sep 17 00:00:00 2001 From: Dmitry Mordvinov Date: Wed, 16 Sep 2026 13:24:54 +0300 Subject: [PATCH] feat: expose operation graph in progress report Operations now carry id, category and dependsOn; waitingFor and stageReports are removed in favor of a single flat operations slice. Meta and release operations are included, so cross-stage dependencies are visible through the stage boundary operations. Operations are ordered by a stable topological sort with id as the tie-break. Plans of one run form a single graph: operations of every plan after the first get an ordinal id prefix and the root operations of a plan depend on the final operations of the plan before it. Operations of a finished plan that never started are reported as Canceled, a new status, instead of staying Pending. Untouched resources are those without plan operations, including resources absent from the cluster whose creation a policy skipped; NoOp/Completed states that nothing was done, not that the resource exists. They are reported for the first plan only, so the NoUntouchedResources option is gone. A plan that is skipped reports its untouched resources alone via UntouchedResourcesOnly. AutoRollback together with LegacyProgressReportCh is rejected, and the reporter plumbing of the rollback path is removed as unreachable. ReleaseInstall and ReleaseUninstall now close LegacyProgressReportCh on every return path, so a consumer can simply range over it. The progrep JSON shape changes: stageReports and waitingFor are gone, operations, id, category, dependsOn and the Canceled status are added. docs/progress-report.md describes the report for library consumers. Signed-off-by: Dmitry Mordvinov --- docs/progress-report.md | 354 +++++ pkg/action/common.go | 7 +- pkg/action/release_install.go | 37 +- pkg/action/release_install_ai_test.go | 62 + pkg/action/release_uninstall.go | 13 +- pkg/legacy/progrep/progress_report.go | 48 +- pkg/plan/helpers_ai_test.go | 125 ++ pkg/plan/legacy_progress_report.go | 86 +- pkg/plan/legacy_progress_reporter.go | 315 ++-- pkg/plan/legacy_progress_reporter_ai_test.go | 1470 ++++++++++++------ pkg/plan/plan_execute.go | 7 +- 11 files changed, 1798 insertions(+), 726 deletions(-) create mode 100644 docs/progress-report.md diff --git a/docs/progress-report.md b/docs/progress-report.md new file mode 100644 index 00000000..7cb1c1bc --- /dev/null +++ b/docs/progress-report.md @@ -0,0 +1,354 @@ +# Progress report for library consumers + + + + +- [Enabling the report](#enabling-the-report) +- [Snapshots](#snapshots) +- [Report shape](#report-shape) +- [Operation types](#operation-types) +- [Statuses](#statuses) +- [Ordering and graph](#ordering-and-graph) +- [Several plans in one run](#several-plans-in-one-run) +- [Untouched resources](#untouched-resources) +- [Counting progress](#counting-progress) +- [Examples](#examples) + - [Successful upgrade](#successful-upgrade) + - [Failed install with a failure plan](#failed-install-with-a-failure-plan) + + + +Nelm can stream the progress of `ReleaseInstall` to an embedding application as a series of snapshots. Each snapshot lists every operation of the deployment plan with its current status and its dependencies, so the consumer can render progress, show what the deployment is waiting for, or draw the plan as a graph. `ReleaseUninstall` accepts the same option and behaves the same way, the text below says `ReleaseInstall` for brevity. + +## Enabling the report + +Pass a buffered channel in `ReleaseInstallOptions.LegacyProgressReportCh` and consume it while `ReleaseInstall` runs. `ReleaseInstall` closes the channel when it returns, so a plain `range` is all a consumer needs: + +```go +reportCh := make(chan progrep.ProgressReport, 1) + +var ( + last progrep.ProgressReport + wg sync.WaitGroup +) + +wg.Add(1) +go func() { + defer wg.Done() + + for report := range reportCh { + last = report + } +}() + +err := action.ReleaseInstall(ctx, releaseName, releaseNamespace, action.ReleaseInstallOptions{ + Chart: chartPath, + LegacyProgressReportCh: reportCh, +}) + +wg.Wait() +``` + +Rules for the channel: + +- It must be buffered with capacity of at least 1, `ReleaseInstall` panics otherwise. +- Consume it concurrently with `ReleaseInstall`. Intermediate snapshots are sent without blocking and are dropped when the channel is full, but the final snapshot is sent with a blocking send, so a consumer that stops reading before `ReleaseInstall` returns deadlocks the deployment. +- Do not close the channel yourself: `ReleaseInstall` closes it when it returns, on every path, including early errors. Snapshots still in the buffer at that moment are delivered before the `range` ends. After a `Timeout` the deployment may still be winding down inside nelm, its snapshots after the return are dropped. +- A snapshot handed to the consumer is never modified afterwards. +- `AutoRollback` cannot be combined with `LegacyProgressReportCh`: `ReleaseInstall` returns an error before doing anything. Rollback reporting needs its own design. + +## Snapshots + +1. The first snapshot arrives when the plan is built, before any operation runs: every operation is listed, all `Pending`, untouched resources already `Completed`. +2. A snapshot arrives on every status change, unless the consumer has not taken the previous one yet, then it is skipped. +3. The final snapshot arrives when the deployment is over, with a blocking send, so it is never skipped. Nothing in it is `Pending`: what did not run is `Canceled`. + +Every snapshot is complete and self-contained: it lists every operation reported so far with its current status, so keeping the latest one is enough, and a skipped snapshot loses nothing but intermediate statuses. The set of operations is fixed from the first snapshot, with one exception: after a failure the operations of the failure plan are appended, so the report grows once, see [Several plans in one run](#several-plans-in-one-run). + +## Report shape + +The types live in `github.com/werf/nelm/pkg/legacy/progrep`. A report has a single field, `Operations`, a list of operations with the following fields: + +| Field | Meaning | +|---|---| +| `ID` | Unique within the report. Opaque key that `DependsOn` refers to, do not parse it: the other fields carry everything it encodes. Operations of the second and every following plan of a run are prefixed with the plan's ordinal number, `2/...`, see [Several plans in one run](#several-plans-in-one-run). | +| `Type` | What the operation does, see [Operation types](#operation-types). | +| `Category` | `meta`, `resource`, `track` or `release`, see [Operation types](#operation-types). | +| `Status` | `Pending`, `Progressing`, `Completed`, `Failed` or `Canceled`, see [Statuses](#statuses). | +| `Iteration` | Distinguishes several operations of the same type on the same resource within one plan, e.g. a resource deployed twice by a chart. Usually 0. | +| `GroupVersionKind` | Group, version and kind of the Kubernetes object. | +| `Name` | Name of the Kubernetes object. | +| `Namespace` | Effective namespace of the Kubernetes object: the release namespace for namespaced objects without an explicit one, empty for cluster-scoped objects. | +| `DependsOn` | IDs of the operations that must finish before this one starts. Always present, empty for operations without predecessors. | + +`GroupVersionKind`, `Name` and `Namespace` describe the Kubernetes object of `resource` and `track` operations. `meta` and `release` operations have no object, these fields are empty for them. + +## Operation types + +| Category | Types | Notes | +|---|---|---| +| `meta` | `StageStart`, `StageEnd` | Boundaries of deployment stages (`init`, `pre-install`, `install`, ...) and of `werf.io/weight` groups inside them. They do nothing themselves, but they carry the structure: operations of a later stage depend on operations of an earlier one through them. | +| `resource` | `Create`, `Update`, `Apply`, `Recreate`, `Delete`, `NoOp` | Mutate a Kubernetes object. `NoOp` is reserved for [untouched resources](#untouched-resources). | +| `track` | `TrackReadiness`, `TrackPresence`, `TrackAbsence` | Wait for an object to become ready, appear or disappear. Never mutate anything. | +| `release` | `CreateRelease`, `UpdateRelease`, `DeleteRelease` | Mutate the Helm release record. `CreateRelease` is among the first operations, `UpdateRelease` marks the release deployed at the very end, or failed in a failure plan. | + +## Statuses + +| Status | Meaning | +|---|---| +| `Pending` | Not started yet. Every operation of a plan starts with it. | +| `Progressing` | Running. | +| `Completed` | Finished successfully. For `NoOp` it means nothing was done, see [Untouched resources](#untouched-resources). | +| `Failed` | Finished with an error. Nelm stops scheduling new operations of the plan after that. | +| `Canceled` | Never started because its plan stopped before reaching it, e.g. after another operation failed. The final snapshot never contains `Pending`: everything that did not run is `Canceled`. | + +## Ordering and graph + +Nelm deploys a release by building a directed acyclic graph of operations and executing it: an operation starts once all of its predecessors are done, operations that do not depend on each other run in parallel. The report exposes that graph as is, `DependsOn` are the edges. + +Operations are listed in execution order, and the order is deterministic: it never changes between snapshots, only statuses do, so consecutive snapshots diff cleanly. + +`DependsOn` reproduces the whole plan graph, including edges to and from `meta` operations. A plan has a single root, the `StageStart` of its first stage, and a single sink, the `StageEnd` of its last stage. Filtering `meta` operations out of the graph breaks connectivity between stages, filter them for display only. + +## Several plans in one run + +After a failure the report grows: the operations of the failure plan are appended after the install plan. The failure plan marks the release failed and deletes the resources that ask for it with `werf.io/delete-policy: failed`. What the consumer sees: + +- New operations with an ordinal prefix in `ID`: `2/...`, and `3/...` for a further plan. The first plan is unprefixed, so a run without failures never shows prefixes. Both `ID` and `DependsOn` use the prefixed form. +- Operations of the install plan that never ran turn `Canceled`. +- The graph stays connected: the root operations of the new plan depend on the sink operations of the plan before it, so the final snapshot reads top to bottom as a chronology of the whole run. + +## Untouched resources + +An untouched resource is a chart resource the plan has no operations for: nothing needs to be done, or a resource policy forbids doing it. They are reported so that progress counts cover the whole chart: ten unchanged resources and two updated ones show as 12 operations, not 2. + +- Type `NoOp`, category `resource`, status `Completed` from the first snapshot. +- `DependsOn` is empty and no operation depends on them: they are not part of the execution flow. +- They come first in the slice, before the plan operations. +- Only the first plan reports them. A failure plan acts upon a few resources of a release the first plan has already described in full. +- `NoOp`/`Completed` states that nothing was done to the resource during the release. The resource may be absent from the cluster (creation skipped by `werf.io/resource-policy`) or differ from the chart (update skipped by the same policy). It is not a statement about the resource being present or in the desired state. + +When the release is skipped because the cluster already matches the chart, the report consists of untouched resources alone. + +## Counting progress + +Count `resource` and `track` operations only. `meta` operations are stage boundaries and `release` operations are bookkeeping of the release record, both inflate the totals without telling the user anything about their resources: + +```go +var completed, remaining int +for _, op := range report.Operations { + if op.Category != progrep.OperationCategoryResource && op.Category != progrep.OperationCategoryTrack { + continue + } + + if op.Status == progrep.OperationStatusCompleted { + completed++ + } else { + remaining++ + } +} +``` + +## Examples + +The examples show final snapshots as serialized, with the JSON keys. + +### Successful upgrade + +A final snapshot of an upgrade where one ConfigMap is unchanged and two others are updated, abridged. The unchanged ConfigMap is an untouched resource and comes first. The two updated ConfigMaps belong to the same stage and have no dependencies on each other, so both `Apply` operations depend on the same `StageStart` and run in parallel, and the `StageEnd` waits for both trackings. + +```yaml +operations: +- id: noop/1/0/my-namespace::ConfigMap:cm-unchanged + type: NoOp + category: resource + status: Completed + Group: "" + Version: v1 + Kind: ConfigMap + name: cm-unchanged + namespace: my-namespace + dependsOn: [] +- id: noop/1/0/stage/init/start + type: StageStart + category: meta + status: Completed + dependsOn: [] +- id: create-release/1/0/my-namespace:my-release:2 + type: CreateRelease + category: release + status: Completed + dependsOn: [noop/1/0/stage/init/start] +- id: noop/1/0/stage/init/end + type: StageEnd + category: meta + status: Completed + dependsOn: [create-release/1/0/my-namespace:my-release:2] +- id: noop/1/0/stage/install/start + type: StageStart + category: meta + status: Completed + dependsOn: [noop/1/0/stage/init/end] +- id: apply/1/0/::ConfigMap:cm-changed + type: Apply + category: resource + status: Completed + Group: "" + Version: v1 + Kind: ConfigMap + name: cm-changed + namespace: my-namespace + dependsOn: [noop/1/0/stage/install/start] +- id: apply/1/0/::ConfigMap:cm-other + type: Apply + category: resource + status: Completed + Group: "" + Version: v1 + Kind: ConfigMap + name: cm-other + namespace: my-namespace + dependsOn: [noop/1/0/stage/install/start] +- id: track-readiness/1/0/::ConfigMap:cm-changed + type: TrackReadiness + category: track + status: Completed + Group: "" + Version: v1 + Kind: ConfigMap + name: cm-changed + namespace: my-namespace + dependsOn: [apply/1/0/::ConfigMap:cm-changed] +- id: track-readiness/1/0/::ConfigMap:cm-other + type: TrackReadiness + category: track + status: Completed + Group: "" + Version: v1 + Kind: ConfigMap + name: cm-other + namespace: my-namespace + dependsOn: [apply/1/0/::ConfigMap:cm-other] +- id: noop/1/0/stage/install/end + type: StageEnd + category: meta + status: Completed + dependsOn: + - track-readiness/1/0/::ConfigMap:cm-changed + - track-readiness/1/0/::ConfigMap:cm-other +- id: noop/1/0/stage/final/start + type: StageStart + category: meta + status: Completed + dependsOn: [noop/1/0/stage/install/end] +- id: update-release/1/0/my-namespace:my-release:2 + type: UpdateRelease + category: release + status: Completed + dependsOn: [noop/1/0/stage/final/start] +- id: noop/1/0/stage/final/end + type: StageEnd + category: meta + status: Completed + dependsOn: [update-release/1/0/my-namespace:my-release:2] +``` + +Counting `resource` and `track` operations gives 5 completed and 0 remaining. + +### Failed install with a failure plan + +A final snapshot of a failed install, abridged. The chart has a ConfigMap in weight -10, a Job in weight 0 that fails and carries `werf.io/delete-policy: failed`, and a ConfigMap in weight 10 that is never created. The failure plan marks the release failed and deletes the Job. + +```yaml +operations: +- id: noop/1/0/stage/init/start + type: StageStart + category: meta + status: Completed + dependsOn: [] +- id: create-release/1/0/my-namespace:my-release:1 + type: CreateRelease + category: release + status: Completed + dependsOn: [noop/1/0/stage/init/start] +- id: noop/1/0/stage/init/end + type: StageEnd + category: meta + status: Completed + dependsOn: [create-release/1/0/my-namespace:my-release:1] +# ... stage install, weight -10: cm-before created and tracked ... +- id: noop/1/0/stage/install/weight:0/start + type: StageStart + category: meta + status: Completed + dependsOn: [noop/1/0/stage/install/weight:-10/end] +- id: create/1/0/:batch:Job:failing-job + type: Create + category: resource + status: Completed + Group: batch + Version: v1 + Kind: Job + name: failing-job + namespace: my-namespace + dependsOn: [noop/1/0/stage/install/weight:0/start] +- id: track-readiness/1/0/:batch:Job:failing-job + type: TrackReadiness + category: track + status: Failed + Group: batch + Version: v1 + Kind: Job + name: failing-job + namespace: my-namespace + dependsOn: [create/1/0/:batch:Job:failing-job] +- id: noop/1/0/stage/install/weight:0/end + type: StageEnd + category: meta + status: Canceled + dependsOn: [track-readiness/1/0/:batch:Job:failing-job] +# ... weight 10: cm-after and its tracking, stage final and update-release, all Canceled ... +- id: noop/1/0/stage/final/end + type: StageEnd + category: meta + status: Canceled + dependsOn: [update-release/1/0/my-namespace:my-release:1] +- id: 2/noop/1/0/stage/init/start + type: StageStart + category: meta + status: Completed + dependsOn: [noop/1/0/stage/final/end] +- id: 2/update-release/1/0/my-namespace:my-release:1 + type: UpdateRelease + category: release + status: Completed + dependsOn: [2/noop/1/0/stage/init/start] +# ... 2/noop/1/0/stage/init/end, 2/noop/1/0/stage/uninstall/start ... +- id: 2/delete/1/0/:batch:Job:failing-job + type: Delete + category: resource + status: Completed + Group: batch + Version: v1 + Kind: Job + name: failing-job + namespace: my-namespace + dependsOn: [2/noop/1/0/stage/uninstall/start] +- id: 2/track-absence/1/0/:batch:Job:failing-job + type: TrackAbsence + category: track + status: Completed + Group: batch + Version: v1 + Kind: Job + name: failing-job + namespace: my-namespace + dependsOn: [2/delete/1/0/:batch:Job:failing-job] +- id: 2/noop/1/0/stage/uninstall/end + type: StageEnd + category: meta + status: Completed + dependsOn: [2/track-absence/1/0/:batch:Job:failing-job] +``` + +Counting `resource` and `track` operations gives 5 completed and 3 remaining: the failed tracking of the Job and the canceled ConfigMap with its tracking. diff --git a/pkg/action/common.go b/pkg/action/common.go index e4057896..857b599c 100644 --- a/pkg/action/common.go +++ b/pkg/action/common.go @@ -200,10 +200,9 @@ func runFailurePlan(ctx context.Context, releaseNamespace string, failedPlan *pl log.Default.Debug(ctx, "Execute failure plan") if err := plan.ExecutePlan(ctx, releaseNamespace, failurePlan, taskStore, logStore, informerFactory, history, clientFactory, plan.ExecutePlanOptions{ - LegacyProgressReporter: opts.LegacyProgressReporter, - TrackingOptions: opts.TrackingOptions, - NetworkParallelism: opts.NetworkParallelism, - NoUntouchedResourcesReport: true, + LegacyProgressReporter: opts.LegacyProgressReporter, + TrackingOptions: opts.TrackingOptions, + NetworkParallelism: opts.NetworkParallelism, }); err != nil { critErrs.Add(fmt.Errorf("execute failure plan: %w", err)) } diff --git a/pkg/action/release_install.go b/pkg/action/release_install.go index dc11e60f..1f75bfe5 100644 --- a/pkg/action/release_install.go +++ b/pkg/action/release_install.go @@ -98,9 +98,10 @@ type ReleaseInstallOptions struct { // LegacyNoReleaseLock, when true, disables acquiring the werf-synchronization release lock in the cluster. LegacyNoReleaseLock bool // LegacyProgressReportCh, when non-nil, receives ProgressReport snapshots during deployment. - // Must be a buffered channel with capacity >= 1. The caller owns the channel and is responsible - // for its lifecycle. Intermediate reports may be dropped if the consumer is slow; the final - // report is guaranteed (blocking send). ReleaseInstall does not close this channel. + // Must be a buffered channel with capacity >= 1. Intermediate reports may be dropped if the + // consumer is slow; the final report is guaranteed (blocking send). ReleaseInstall closes the + // channel when it returns, on every path. Incompatible with AutoRollback. See + // docs/progress-report.md. LegacyProgressReportCh chan<- progrep.ProgressReport // NetworkParallelism limits the number of concurrent network-related operations (API calls, resource fetches). // Defaults to DefaultNetworkParallelism if not set or <= 0. @@ -139,9 +140,8 @@ type runRollbackPlanOptions struct { common.ReleaseInstallRuntimeOptions common.TrackingOptions - LegacyProgressReporter *plan.LegacyProgressReporter - NetworkParallelism int - RollbackGraphPath string + NetworkParallelism int + RollbackGraphPath string } type runRollbackPlanResult struct { @@ -151,6 +151,10 @@ type runRollbackPlanResult struct { } func ReleaseInstall(ctx context.Context, releaseName, releaseNamespace string, opts ReleaseInstallOptions) error { + if opts.LegacyProgressReportCh != nil { + defer close(opts.LegacyProgressReportCh) + } + ctx, ctxCancelFn := context.WithCancelCause(ctx) if opts.Timeout == 0 { @@ -543,9 +547,10 @@ func releaseInstall(ctx context.Context, ctxCancelFn context.CancelCauseFunc, re if opts.LegacyProgressReportCh != nil { reporter := plan.NewLegacyProgressReporter(opts.LegacyProgressReportCh) - defer close(opts.LegacyProgressReportCh) - reporter.StartStage(installPlan, releaseNamespace, instResInfos, clientFactory.Mapper(), plan.StartStageOptions{}) + reporter.StartPlan(installPlan, releaseNamespace, instResInfos, clientFactory.Mapper(), plan.StartPlanOptions{ + UntouchedResourcesOnly: true, + }) reporter.Stop(ctx) } @@ -588,7 +593,6 @@ func releaseInstall(ctx context.Context, ctxCancelFn context.CancelCauseFunc, re var reporter *plan.LegacyProgressReporter if opts.LegacyProgressReportCh != nil { reporter = plan.NewLegacyProgressReporter(opts.LegacyProgressReportCh) - defer close(opts.LegacyProgressReportCh) } log.Default.Debug(ctx, "Execute release install plan") @@ -639,7 +643,6 @@ func releaseInstall(ctx context.Context, ctxCancelFn context.CancelCauseFunc, re runRollbackPlanResult, nonCritErrs, critErrs := runRollbackPlan(ctx, releaseName, releaseNamespace, newRelease, prevDeployedRelease, taskStore, logStore, informerFactory, history, clientFactory, runRollbackPlanOptions{ ReleaseInstallRuntimeOptions: opts.ReleaseInstallRuntimeOptions, TrackingOptions: opts.TrackingOptions, - LegacyProgressReporter: reporter, NetworkParallelism: opts.NetworkParallelism, RollbackGraphPath: opts.RollbackGraphPath, }) @@ -742,6 +745,10 @@ func applyReleaseInstallOptionsDefaults(opts ReleaseInstallOptions, currentDir, opts.LegacyLogRegistryStreamOut = io.Discard } + if opts.AutoRollback && opts.LegacyProgressReportCh != nil { + return ReleaseInstallOptions{}, fmt.Errorf("auto rollback is not supported together with legacy progress reporting") + } + if opts.NetworkParallelism <= 0 { opts.NetworkParallelism = common.DefaultNetworkParallelism } @@ -974,10 +981,6 @@ func runRollbackPlan(ctx context.Context, releaseName, releaseNamespace string, }) if releaseIsUpToDate && planIsUseless { - if opts.LegacyProgressReporter != nil { - opts.LegacyProgressReporter.StartStage(rollbackPlan, releaseNamespace, instResInfos, clientFactory.Mapper(), plan.StartStageOptions{}) - } - log.Default.Info(ctx, color.Style{color.Bold, color.Green}.Render("Skipped rollback release")+" %q (namespace: %q): cluster resources already as desired", releaseName, releaseNamespace) return &runRollbackPlanResult{}, nonCritErrs, critErrs @@ -986,7 +989,6 @@ func runRollbackPlan(ctx context.Context, releaseName, releaseNamespace string, log.Default.Debug(ctx, "Execute rollback plan") executePlanErr := plan.ExecutePlan(ctx, releaseNamespace, rollbackPlan, taskStore, logStore, informerFactory, history, clientFactory, plan.ExecutePlanOptions{ - LegacyProgressReporter: opts.LegacyProgressReporter, TrackingOptions: opts.TrackingOptions, NetworkParallelism: opts.NetworkParallelism, InstallableResourceInfos: instResInfos, @@ -1013,9 +1015,8 @@ func runRollbackPlan(ctx context.Context, releaseName, releaseNamespace string, if executePlanErr != nil { runFailurePlanResult, nonCrErrs, crErrs := runFailurePlan(ctx, releaseNamespace, rollbackPlan, instResInfos, relInfos, taskStore, logStore, informerFactory, history, clientFactory, runFailureInstallPlanOptions{ - LegacyProgressReporter: opts.LegacyProgressReporter, - TrackingOptions: opts.TrackingOptions, - NetworkParallelism: opts.NetworkParallelism, + TrackingOptions: opts.TrackingOptions, + NetworkParallelism: opts.NetworkParallelism, }) critErrs.Add(crErrs) diff --git a/pkg/action/release_install_ai_test.go b/pkg/action/release_install_ai_test.go index cc3df1d3..7ea6c09d 100644 --- a/pkg/action/release_install_ai_test.go +++ b/pkg/action/release_install_ai_test.go @@ -18,7 +18,9 @@ import ( "k8s.io/client-go/dynamic" "k8s.io/client-go/kubernetes" + "github.com/werf/nelm/pkg/common" "github.com/werf/nelm/pkg/kube" + "github.com/werf/nelm/pkg/legacy/progrep" "github.com/werf/nelm/pkg/resource/spec" ) @@ -123,6 +125,28 @@ func (c *createNamespaceKubeClient) ServerVersion(ctx context.Context) (*version panic("not implemented") } +func TestAI_ApplyReleaseInstallOptionsDefaults_AllowsProgressReportWithoutAutoRollback(t *testing.T) { + opts := ReleaseInstallOptions{ + LegacyProgressReportCh: make(chan progrep.ProgressReport, 1), + TempDirPath: t.TempDir(), + } + + _, err := applyReleaseInstallOptionsDefaults(opts, t.TempDir(), t.TempDir()) + require.NoError(t, err) +} + +func TestAI_ApplyReleaseInstallOptionsDefaults_RejectsAutoRollbackWithProgressReport(t *testing.T) { + opts := ReleaseInstallOptions{ + AutoRollback: true, + LegacyProgressReportCh: make(chan progrep.ProgressReport, 1), + TempDirPath: t.TempDir(), + } + + _, err := applyReleaseInstallOptionsDefaults(opts, t.TempDir(), t.TempDir()) + require.Error(t, err) + assert.Contains(t, err.Error(), "auto rollback") +} + func TestAI_CreateReleaseNamespaceBothProbesForbiddenAggregates(t *testing.T) { cmErr := newForbiddenErr("configmaps", "werf-synchronization") nsErr := newForbiddenErr("namespaces", "my-namespace") @@ -242,6 +266,44 @@ func TestAI_CreateReleaseNamespaceRealCreateFailurePropagates(t *testing.T) { assert.Equal(t, createNamespaceCall{dryRun: false, kind: "Namespace"}, kubeClient.calls[2]) } +func TestAI_ReleaseInstall_ClosesProgressReportChannelOnEarlyError(t *testing.T) { + reportCh := make(chan progrep.ProgressReport, 1) + + err := ReleaseInstall(context.Background(), "rel", "ns", ReleaseInstallOptions{ + AutoRollback: true, + LegacyProgressReportCh: reportCh, + TempDirPath: t.TempDir(), + }) + require.Error(t, err) + + select { + case _, ok := <-reportCh: + assert.False(t, ok, "the channel must be closed, not carry a report") + default: + t.Fatal("the channel must be closed when ReleaseInstall returns") + } +} + +func TestAI_ReleaseUninstall_ClosesProgressReportChannelOnEarlyError(t *testing.T) { + reportCh := make(chan progrep.ProgressReport, 1) + + opts := ReleaseUninstallOptions{ + LegacyProgressReportCh: reportCh, + TempDirPath: t.TempDir(), + } + opts.ReleaseStorageDriver = common.ReleaseStorageDriverMemory + + err := ReleaseUninstall(context.Background(), "rel", "ns", opts) + require.Error(t, err) + + select { + case _, ok := <-reportCh: + assert.False(t, ok, "the channel must be closed, not carry a report") + default: + t.Fatal("the channel must be closed when ReleaseUninstall returns") + } +} + func newForbiddenErr(resource, name string) error { return apierrors.NewForbidden(schema.GroupResource{Resource: resource}, name, errors.New("forbidden")) } diff --git a/pkg/action/release_uninstall.go b/pkg/action/release_uninstall.go index c8fe69e3..5659291f 100644 --- a/pkg/action/release_uninstall.go +++ b/pkg/action/release_uninstall.go @@ -45,9 +45,9 @@ type ReleaseUninstallOptions struct { // LegacyNoReleaseLock, when true, disables acquiring the werf-synchronization release lock in the cluster. LegacyNoReleaseLock bool // LegacyProgressReportCh, when non-nil, receives ProgressReport snapshots during deployment. - // Must be a buffered channel with capacity >= 1. The caller owns the channel and is responsible - // for its lifecycle. Intermediate reports may be dropped if the consumer is slow; the final - // report is guaranteed (blocking send). ReleaseUninstall does not close this channel. + // Must be a buffered channel with capacity >= 1. Intermediate reports may be dropped if the + // consumer is slow; the final report is guaranteed (blocking send). ReleaseUninstall closes the + // channel when it returns, on every path. See docs/progress-report.md. LegacyProgressReportCh chan<- progrep.ProgressReport // NetworkParallelism limits the number of concurrent network-related operations (API calls, resource fetches). // Defaults to DefaultNetworkParallelism if not set or <= 0. @@ -82,6 +82,10 @@ type ReleaseUninstallOptions struct { // Uninstall the Helm release along with its resources from the cluster. func ReleaseUninstall(ctx context.Context, releaseName, releaseNamespace string, opts ReleaseUninstallOptions) error { + if opts.LegacyProgressReportCh != nil { + defer close(opts.LegacyProgressReportCh) + } + ctx, ctxCancelFn := context.WithCancelCause(ctx) if opts.Timeout == 0 { @@ -299,9 +303,6 @@ func releaseUninstall(ctx context.Context, ctxCancelFn context.CancelCauseFunc, var reporter *plan.LegacyProgressReporter if opts.LegacyProgressReportCh != nil { reporter = plan.NewLegacyProgressReporter(opts.LegacyProgressReportCh) - defer func() { - close(opts.LegacyProgressReportCh) - }() } log.Default.Debug(ctx, "Execute release delete plan") diff --git a/pkg/legacy/progrep/progress_report.go b/pkg/legacy/progrep/progress_report.go index 27e4b9d3..d8a8c60e 100644 --- a/pkg/legacy/progrep/progress_report.go +++ b/pkg/legacy/progrep/progress_report.go @@ -3,10 +3,16 @@ package progrep import "k8s.io/apimachinery/pkg/runtime/schema" const ( + OperationCategoryMeta OperationCategory = "meta" + OperationCategoryResource OperationCategory = "resource" + OperationCategoryTrack OperationCategory = "track" + OperationCategoryRelease OperationCategory = "release" + OperationStatusPending OperationStatus = "Pending" OperationStatusProgressing OperationStatus = "Progressing" OperationStatusCompleted OperationStatus = "Completed" OperationStatusFailed OperationStatus = "Failed" + OperationStatusCanceled OperationStatus = "Canceled" OperationTypeCreate OperationType = "Create" OperationTypeUpdate OperationType = "Update" @@ -17,33 +23,47 @@ const ( OperationTypeTrackReadiness OperationType = "TrackReadiness" OperationTypeTrackPresence OperationType = "TrackPresence" OperationTypeTrackAbsence OperationType = "TrackAbsence" + OperationTypeStageStart OperationType = "StageStart" + OperationTypeStageEnd OperationType = "StageEnd" + OperationTypeCreateRelease OperationType = "CreateRelease" + OperationTypeUpdateRelease OperationType = "UpdateRelease" + OperationTypeDeleteRelease OperationType = "DeleteRelease" ) +type OperationCategory string + type OperationType string type OperationStatus string -// ProgressReport contains stage reports ordered chronologically; the last element is the -// currently active stage. +// ProgressReport lists ALL operations of every plan executed so far, in execution order: from +// the very first report every operation of the current plan is present (initially as Pending), +// and operations of later plans (e.g. a failure plan) are appended after the earlier ones. The +// plans form a single graph: the root operations of a later plan depend on the final operations +// of the plan before it. Operations of a finished plan that were never started are Canceled. The first plan describes the complete desired state of the release, so +// the resources it leaves untouched are listed too, as NoOp with status Completed, at the +// beginning of the slice. Later plans, e.g. a failure plan, never add untouched resources. +// +// An untouched resource is one the plan has no operations for. NoOp/Completed means only that +// nothing was done to the resource during the release: the resource may be absent from the +// cluster (e.g. creation skipped by a resource policy) or differ from the chart (e.g. update +// skipped by a resource policy). It is not a statement about the resource being present or +// in the desired state. type ProgressReport struct { - StageReports []StageReport `json:"stageReports"` -} - -// StageReport contains ALL operations in the plan -- from the very first report, every -// operation is present (initially as Pending). A stage of a plan that describes a complete -// desired state, such as an install or a rollback plan, additionally lists the resources that -// plan leaves untouched, as NoOp with status Completed; that set is supplied per stage and -// matches the revision the stage deploys. A failure plan acts upon a few resources only, so -// its stage lists just its own operations. -type StageReport struct { Operations []Operation `json:"operations"` } +// Operation ID is unique within a report and is referenced by DependsOn of other operations. +// Operations of the first plan carry the plan's own operation ID; operations of every following +// plan are prefixed with the plan's ordinal number, e.g. "2/apply/1/0/...". Meta and release +// operations have an empty ObjectRef. type Operation struct { OperationRef - Status OperationStatus `json:"status"` - WaitingFor []OperationRef `json:"waitingFor"` + ID string `json:"id"` + Category OperationCategory `json:"category"` + Status OperationStatus `json:"status"` + DependsOn []string `json:"dependsOn"` } type OperationRef struct { diff --git a/pkg/plan/helpers_ai_test.go b/pkg/plan/helpers_ai_test.go index 4eddf290..9ac03852 100644 --- a/pkg/plan/helpers_ai_test.go +++ b/pkg/plan/helpers_ai_test.go @@ -5,9 +5,12 @@ package plan import ( "fmt" "math/rand" + "sort" + "testing" "github.com/dominikbraun/graph" "github.com/samber/lo" + "github.com/stretchr/testify/require" "k8s.io/apimachinery/pkg/api/meta" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime/schema" @@ -62,6 +65,13 @@ func installableInfoToDeleteOnAnyOutcome(name string) *InstallableResourceInfo { return info } +func createConfigMapOp(name string) *Operation { + return &Operation{ + Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, + Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec(name, "", gvkConfigMap)}, + } +} + func installableInfoToDeleteOnFailedInstall(name string) *InstallableResourceInfo { info := installableInfoNamed(name, ResourceInstallTypeApply) info.MustDeleteOnFailedInstall = true @@ -76,6 +86,16 @@ func installableInfoToDeleteOnSuccessfulInstall(name string) *InstallableResourc return info } +func realInstallableInfo(name string, stage common.Stage) *InstallableResourceInfo { + return &InstallableResourceInfo{ + ResourceMeta: makeResourceMeta(name, "default", gvkConfigMap), + LocalResource: &resource.InstallableResource{ResourceSpec: makeResourceSpec(name, "default", gvkConfigMap)}, + MustInstall: ResourceInstallTypeApply, + MustTrackReadiness: true, + Stage: stage, + } +} + func deletableInfoWithUID(uid types.UID) *DeletableResourceInfo { return &DeletableResourceInfo{GetResult: unstructuredWithUID(uid)} } @@ -92,12 +112,34 @@ func installableInfoWithUID(uid types.UID) *InstallableResourceInfo { return &InstallableResourceInfo{GetResult: unstructuredWithUID(uid)} } +func lastReportOperations(t *testing.T, ch <-chan progrep.ProgressReport) []progrep.Operation { + t.Helper() + + reports := drainChannel(ch) + require.NotEmpty(t, reports, "expected at least one report") + + return reports[len(reports)-1].Operations +} + func makeResourceSpec(name, namespace string, gvk schema.GroupVersionKind) *spec.ResourceSpec { return &spec.ResourceSpec{ ResourceMeta: makeResourceMeta(name, namespace, gvk), } } +func makeUntouchedInfo(name, namespace string, gvk schema.GroupVersionKind) *InstallableResourceInfo { + obj := &unstructured.Unstructured{} + obj.SetGroupVersionKind(gvk) + obj.SetName(name) + obj.SetNamespace(namespace) + + return &InstallableResourceInfo{ + ResourceMeta: makeResourceMeta(name, namespace, gvk), + MustInstall: ResourceInstallTypeNone, + GetResult: obj, + } +} + func buildTestPlan(ops []*Operation, deps map[int][]int) *Plan { p := NewPlan() @@ -133,6 +175,43 @@ func drainChannel(ch <-chan progrep.ProgressReport) []progrep.ProgressReport { } } +// graphRootsAndSinks treats the given operations as a self-contained graph: roots have no +// dependencies among them, sinks are not depended upon by any of them. +func graphRootsAndSinks(ops []progrep.Operation) ([]string, []string) { + ids := lo.SliceToMap(ops, func(op progrep.Operation) (string, struct{}) { + return op.ID, struct{}{} + }) + + dependedUpon := make(map[string]struct{}) + + var roots []string + for _, op := range ops { + internalDeps := lo.Filter(op.DependsOn, func(id string, _ int) bool { + return lo.HasKey(ids, id) + }) + + if len(internalDeps) == 0 { + roots = append(roots, op.ID) + } + + for _, id := range internalDeps { + dependedUpon[id] = struct{}{} + } + } + + var sinks []string + for _, op := range ops { + if !lo.HasKey(dependedUpon, op.ID) { + sinks = append(sinks, op.ID) + } + } + + sort.Strings(roots) + sort.Strings(sinks) + + return roots, sinks +} + func makeResourceMeta(name, namespace string, gvk schema.GroupVersionKind) *spec.ResourceMeta { return &spec.ResourceMeta{ Name: name, @@ -141,6 +220,18 @@ func makeResourceMeta(name, namespace string, gvk schema.GroupVersionKind) *spec } } +func operationIDs(ops []progrep.Operation) []string { + return lo.Map(ops, func(op progrep.Operation, _ int) string { + return op.ID + }) +} + +func operationsByID(ops []progrep.Operation) map[string]progrep.Operation { + return lo.SliceToMap(ops, func(op progrep.Operation) (string, progrep.Operation) { + return op.ID, op + }) +} + func randomTrackingGraph(rnd *rand.Rand, opsCount int, edgeChance float64) ([]OperationCategory, map[int][]int) { allCategories := []OperationCategory{OperationCategoryResource, OperationCategoryTrack, OperationCategoryMeta} @@ -161,6 +252,29 @@ func randomTrackingGraph(rnd *rand.Rand, opsCount int, edgeChance float64) ([]Op return categories, deps } +func reachesThroughDependsOn(byID map[string]progrep.Operation, from, to string) bool { + visited := make(map[string]struct{}) + queue := []string{from} + + for len(queue) > 0 { + id := queue[0] + queue = queue[1:] + + if id == to { + return true + } + + if _, ok := visited[id]; ok { + continue + } + + visited[id] = struct{}{} + queue = append(queue, byID[id].DependsOn...) + } + + return false +} + // Pre-optimization implementation of squashFinalTrackingOperations, kept as a reference to assert // the optimized one against. func squashFinalTrackingOperationsReference(p *Plan) { @@ -188,6 +302,17 @@ func squashFinalTrackingOperationsReference(p *Plan) { } } +func stageMetaOp(opID string) *Operation { + return &Operation{ + Type: OperationTypeNoop, Version: OperationVersionNoop, Category: OperationCategoryMeta, + Config: &OperationConfigNoop{OpID: opID}, + } +} + +func startTestPlan(reporter *LegacyProgressReporter, p *Plan, untouched []*InstallableResourceInfo, opts StartPlanOptions) { + reporter.StartPlan(p, "default", untouched, newFakeRESTMapper(), opts) +} + func trackingTestOperations(categories []OperationCategory) []*Operation { return lo.Map(categories, func(category OperationCategory, i int) *Operation { return &Operation{ diff --git a/pkg/plan/legacy_progress_report.go b/pkg/plan/legacy_progress_report.go index b2ff5583..9a9eca3f 100644 --- a/pkg/plan/legacy_progress_report.go +++ b/pkg/plan/legacy_progress_report.go @@ -2,10 +2,12 @@ package plan import ( "fmt" + "strings" "k8s.io/apimachinery/pkg/api/meta" "k8s.io/apimachinery/pkg/runtime/schema" + "github.com/werf/nelm/pkg/common" "github.com/werf/nelm/pkg/legacy/progrep" "github.com/werf/nelm/pkg/resource/spec" ) @@ -42,6 +44,37 @@ func extractObjectRef(op *Operation, resolvedNamespaces map[string]string) progr } } +func mapOperationType(op *Operation) progrep.OperationType { + switch op.Type { + case OperationTypeCreate: + return progrep.OperationTypeCreate + case OperationTypeUpdate: + return progrep.OperationTypeUpdate + case OperationTypeDelete: + return progrep.OperationTypeDelete + case OperationTypeApply: + return progrep.OperationTypeApply + case OperationTypeRecreate: + return progrep.OperationTypeRecreate + case OperationTypeTrackReadiness: + return progrep.OperationTypeTrackReadiness + case OperationTypeTrackPresence: + return progrep.OperationTypeTrackPresence + case OperationTypeTrackAbsence: + return progrep.OperationTypeTrackAbsence + case OperationTypeCreateRelease: + return progrep.OperationTypeCreateRelease + case OperationTypeUpdateRelease: + return progrep.OperationTypeUpdateRelease + case OperationTypeDeleteRelease: + return progrep.OperationTypeDeleteRelease + case OperationTypeNoop: + return mapNoopOperationType(op) + default: + panic(fmt.Sprintf("unexpected operation type %q", op.Type)) + } +} + func reportOperationStatus(op *Operation, status OperationStatus, reporter *LegacyProgressReporter) { op.Status = status @@ -52,6 +85,36 @@ func reportOperationStatus(op *Operation, status OperationStatus, reporter *Lega reporter.ReportStatus(op.ID(), mapOperationStatus(status)) } +// Stage boundaries are plain noop operations in the plan; only the suffix of their config ID +// tells a stage start from a stage end, the same way findMetaOperationPairs does. +func mapNoopOperationType(op *Operation) progrep.OperationType { + configID := op.Config.ID() + + switch { + case strings.HasSuffix(configID, "/"+common.StageStartSuffix): + return progrep.OperationTypeStageStart + case strings.HasSuffix(configID, "/"+common.StageEndSuffix): + return progrep.OperationTypeStageEnd + default: + panic(fmt.Sprintf("unexpected noop operation %q", op.ID())) + } +} + +func mapOperationCategory(c OperationCategory) progrep.OperationCategory { + switch c { + case OperationCategoryMeta: + return progrep.OperationCategoryMeta + case OperationCategoryResource: + return progrep.OperationCategoryResource + case OperationCategoryTrack: + return progrep.OperationCategoryTrack + case OperationCategoryRelease: + return progrep.OperationCategoryRelease + default: + panic(fmt.Sprintf("unexpected operation category %q", c)) + } +} + func mapOperationStatus(s OperationStatus) progrep.OperationStatus { switch s { case OperationStatusUnknown: @@ -67,29 +130,6 @@ func mapOperationStatus(s OperationStatus) progrep.OperationStatus { } } -func mapOperationType(t OperationType) progrep.OperationType { - switch t { - case OperationTypeCreate: - return progrep.OperationTypeCreate - case OperationTypeUpdate: - return progrep.OperationTypeUpdate - case OperationTypeDelete: - return progrep.OperationTypeDelete - case OperationTypeApply: - return progrep.OperationTypeApply - case OperationTypeRecreate: - return progrep.OperationTypeRecreate - case OperationTypeTrackReadiness: - return progrep.OperationTypeTrackReadiness - case OperationTypeTrackPresence: - return progrep.OperationTypeTrackPresence - case OperationTypeTrackAbsence: - return progrep.OperationTypeTrackAbsence - default: - panic(fmt.Sprintf("unexpected operation type %q", t)) - } -} - // operationResourceMeta extracts GVK, name, and namespace from an operation's // config. Returns ok=false for operation configs that don't carry resource metadata // (e.g. Noop, release operations). diff --git a/pkg/plan/legacy_progress_reporter.go b/pkg/plan/legacy_progress_reporter.go index 57b2da50..913205eb 100644 --- a/pkg/plan/legacy_progress_reporter.go +++ b/pkg/plan/legacy_progress_reporter.go @@ -3,7 +3,10 @@ package plan import ( "context" "fmt" + "sort" + "strings" + "github.com/dominikbraun/graph" "github.com/samber/lo" "k8s.io/apimachinery/pkg/api/meta" @@ -36,35 +39,75 @@ func (r *LegacyProgressReporter) ReportStatus(opID string, status progrep.Operat return } - s.ops[idx].status = status + s.ops[idx].Status = status - report := buildProgressReport(s.frozen, s.ops) - sendNonBlocking(r.reportCh, report) + if len(r.reportCh) == cap(r.reportCh) { + return + } + + sendNonBlocking(r.reportCh, buildProgressReport(s.ops)) }) } -func (r *LegacyProgressReporter) StartStage(p *Plan, releaseNamespace string, installableResourceInfos []*InstallableResourceInfo, mapper meta.RESTMapper, opts StartStageOptions) { +// StartPlan appends the operations of the plan that is about to be executed to the report and +// makes them addressable by ReportStatus. Operations of the previously started plans stay in the +// report as they are. The plans are chained: the root operations of the new plan depend on the +// final operations of the previous one, so the whole run reads as a single graph. Operations of +// the previous plan that never started are marked Canceled. Untouched resources are reported for +// the first plan only: later plans act upon a release the first plan has already described in +// full. +func (r *LegacyProgressReporter) StartPlan(p *Plan, releaseNamespace string, installableResourceInfos []*InstallableResourceInfo, mapper meta.RESTMapper, opts StartPlanOptions) { resolvedNamespaces := buildResolvedNamespaces(p, releaseNamespace, mapper) - if opts.NoUntouchedResources { - r.startStage(p, resolvedNamespaces, nil, nil) - - return - } - untouchedResolvedNamespaces := make(map[string]string, len(installableResourceInfos)) for _, info := range installableResourceInfos { untouchedResolvedNamespaces[info.ID()] = resolveNamespace(info.GroupVersionKind, info.Namespace, releaseNamespace, mapper) } - r.startStage(p, resolvedNamespaces, installableResourceInfos, untouchedResolvedNamespaces) + r.state.RWTransaction(func(s *progressReporterState) { + cancelPendingOperations(s.ops) + + s.plansCount++ + idPrefix := planIDPrefix(s.plansCount) + + var planOps []progrep.Operation + if !opts.UntouchedResourcesOnly { + predMap := lo.Must(p.Graph.PredecessorMap()) + + planOps = buildPlanOperations(p, predMap, resolvedNamespaces, idPrefix, s.lastPlanSinkIDs) + s.lastPlanSinkIDs = planSinkOperationIDs(predMap, idPrefix) + } + + seenRefs := make(map[progrep.ObjectRef]struct{}, len(planOps)) + for _, op := range planOps { + if op.Category == progrep.OperationCategoryResource || op.Category == progrep.OperationCategoryTrack { + seenRefs[op.ObjectRef] = struct{}{} + } + } + + if s.plansCount == 1 { + s.ops = append(s.ops, buildUntouchedOperations(installableResourceInfos, untouchedResolvedNamespaces, seenRefs)...) + } + + s.opIndex = make(map[string]int, len(planOps)) + for _, op := range planOps { + s.opIndex[strings.TrimPrefix(op.ID, idPrefix)] = len(s.ops) + s.ops = append(s.ops, op) + } + + sendNonBlocking(r.reportCh, buildProgressReport(s.ops)) + }) } +// Stop sends the final report with a blocking send. Operations that never started are marked +// Canceled first: nothing is going to run them anymore. func (r *LegacyProgressReporter) Stop(ctx context.Context) { var report progrep.ProgressReport r.state.RWTransaction(func(s *progressReporterState) { - report = buildProgressReport(s.frozen, s.ops) + cancelPendingOperations(s.ops) + + report = buildProgressReport(s.ops) }) func() { @@ -77,173 +120,155 @@ func (r *LegacyProgressReporter) Stop(ctx context.Context) { }() } -func (r *LegacyProgressReporter) startStage(p *Plan, resolvedNamespaces map[string]string, untouched []*InstallableResourceInfo, untouchedResolvedNamespaces map[string]string) { - r.state.RWTransaction(func(s *progressReporterState) { - if len(s.ops) > 0 { - s.frozen = append(s.frozen, buildStageReport(s.ops)) - } - - predMap := lo.Must(p.Graph.PredecessorMap()) - ops := p.Operations() - - var entries []opEntry - - entryIndex := make(map[string]int) - seenRefs := make(map[progrep.ObjectRef]struct{}) +type StartPlanOptions struct { + // UntouchedResourcesOnly, when true, omits the plan's own operations and reports the untouched + // resources alone. Set it for a plan that is not going to be executed: its operations would + // otherwise stay Pending forever. + UntouchedResourcesOnly bool +} - for _, op := range ops { - if op.Category != OperationCategoryResource && op.Category != OperationCategoryTrack { - continue - } +type progressReporterState struct { + // lastPlanSinkIDs are the report IDs of the operations without successors in the most recently + // started plan that had operations. The root operations of the next plan depend on them. + lastPlanSinkIDs []string + // opIndex maps the raw operation IDs of the most recently started plan to their positions in + // ops. Operations of earlier plans are no longer addressable: by the time the next plan starts + // they are either done or canceled. + opIndex map[string]int + ops []progrep.Operation + plansCount int +} - ref := extractObjectRef(op, resolvedNamespaces) - typ := mapOperationType(op.Type) - idx := len(entries) - entryIndex[op.ID()] = idx - seenRefs[ref] = struct{}{} - - entries = append(entries, opEntry{ - iteration: int(op.Iteration), - ref: ref, - status: progrep.OperationStatusPending, - typ: typ, - }) - } +func sendNonBlocking(ch chan<- progrep.ProgressReport, report progrep.ProgressReport) { + safeSend(ch, report) +} - for _, info := range untouched { - if info.GetResult == nil { - continue - } +func buildPlanOperations(p *Plan, predMap map[string]map[string]graph.Edge[string], resolvedNamespaces map[string]string, idPrefix string, rootDependsOn []string) []progrep.Operation { + opIDs := lo.Must(graph.StableTopologicalSort(p.Graph, func(a, b string) bool { + return a < b + })) - ref := progrep.ObjectRef{ - GroupVersionKind: info.GroupVersionKind, - Name: info.Name, - Namespace: untouchedResolvedNamespaces[info.ID()], - } + result := make([]progrep.Operation, 0, len(opIDs)) - if _, ok := seenRefs[ref]; ok { - continue - } + for _, opID := range opIDs { + op := lo.Must(p.Operation(opID)) - seenRefs[ref] = struct{}{} + dependsOn := lo.Keys(predMap[opID]) + sort.Strings(dependsOn) - entries = append(entries, opEntry{ - iteration: 0, - ref: ref, - status: progrep.OperationStatusCompleted, - // Untouched resources have no real operation; NoOp is a - // neutral label for an already-present, unchanged resource shown as Completed. - typ: progrep.OperationTypeNoOp, - }) + for i := range dependsOn { + dependsOn[i] = idPrefix + dependsOn[i] } - for _, op := range ops { - idx, ok := entryIndex[op.ID()] - if !ok { - continue - } - - var predIndices []int - for predID := range predMap[op.ID()] { - if predIdx, predOk := entryIndex[predID]; predOk { - predIndices = append(predIndices, predIdx) - } - } + if len(dependsOn) == 0 { + dependsOn = append(dependsOn, rootDependsOn...) + } - entries[idx].predIndices = predIndices + var ref progrep.ObjectRef + if op.Category == OperationCategoryResource || op.Category == OperationCategoryTrack { + ref = extractObjectRef(op, resolvedNamespaces) } - s.ops = entries - s.opIndex = entryIndex + result = append(result, progrep.Operation{ + OperationRef: progrep.OperationRef{ + ObjectRef: ref, + Type: mapOperationType(op), + Iteration: int(op.Iteration), + }, + ID: idPrefix + opID, + Category: mapOperationCategory(op.Category), + Status: progrep.OperationStatusPending, + DependsOn: dependsOn, + }) + } - report := buildProgressReport(s.frozen, s.ops) - sendNonBlocking(r.reportCh, report) - }) + return result } -type StartStageOptions struct { - // NoUntouchedResources, when true, omits untouched resources from the stage report. - // Set it for delta plans, like a failure plan, which only carry operations for the few - // resources they act upon: there the release-wide inventory of untouched resources is - // not part of what the stage does. - NoUntouchedResources bool -} +func buildProgressReport(ops []progrep.Operation) progrep.ProgressReport { + operations := make([]progrep.Operation, len(ops)) + copy(operations, ops) -type progressReporterState struct { - frozen []progrep.StageReport - opIndex map[string]int - ops []opEntry + return progrep.ProgressReport{ + Operations: operations, + } } -type opEntry struct { - iteration int - predIndices []int - ref progrep.ObjectRef - status progrep.OperationStatus - typ progrep.OperationType -} +// Untouched resources have no operation in the plan and thus no position in its graph, so they +// are reported without edges, before the plan operations, ordered by ID. NoOp is a neutral label +// for a resource the plan leaves as is, shown as Completed. +func buildUntouchedOperations(untouched []*InstallableResourceInfo, untouchedResolvedNamespaces map[string]string, seenRefs map[progrep.ObjectRef]struct{}) []progrep.Operation { + var result []progrep.Operation + + for _, info := range untouched { + ref := progrep.ObjectRef{ + GroupVersionKind: info.GroupVersionKind, + Name: info.Name, + Namespace: untouchedResolvedNamespaces[info.ID()], + } -func sendNonBlocking(ch chan<- progrep.ProgressReport, report progrep.ProgressReport) { - safeSend(ch, report) -} + if _, ok := seenRefs[ref]; ok { + continue + } -func buildProgressReport(frozen []progrep.StageReport, ops []opEntry) progrep.ProgressReport { - stageReports := make([]progrep.StageReport, 0, len(frozen)+1) + seenRefs[ref] = struct{}{} - for _, sr := range frozen { - opsCopy := make([]progrep.Operation, len(sr.Operations)) - copy(opsCopy, sr.Operations) - stageReports = append(stageReports, progrep.StageReport{Operations: opsCopy}) + result = append(result, progrep.Operation{ + OperationRef: progrep.OperationRef{ + ObjectRef: ref, + Type: progrep.OperationTypeNoOp, + Iteration: info.Iteration, + }, + ID: OperationID(OperationTypeNoop, OperationVersionNoop, OperationIteration(info.Iteration), info.ID()), + Category: progrep.OperationCategoryResource, + Status: progrep.OperationStatusCompleted, + DependsOn: []string{}, + }) } - operations := make([]progrep.Operation, len(ops)) - for i, e := range ops { - var waitingFor []progrep.OperationRef - - for _, predIdx := range e.predIndices { - if ops[predIdx].status != progrep.OperationStatusCompleted { - waitingFor = append(waitingFor, progrep.OperationRef{ - ObjectRef: ops[predIdx].ref, - Type: ops[predIdx].typ, - Iteration: ops[predIdx].iteration, - }) - } - } + sort.Slice(result, func(i, j int) bool { + return result[i].ID < result[j].ID + }) - operations[i] = progrep.Operation{ - OperationRef: progrep.OperationRef{ - ObjectRef: e.ref, - Type: e.typ, - Iteration: e.iteration, - }, - Status: e.status, - WaitingFor: waitingFor, + return result +} + +// ExecutePlan waits for every started operation before returning, so once a plan is over, what +// is still Pending was never scheduled and never will be. +func cancelPendingOperations(ops []progrep.Operation) { + for i := range ops { + if ops[i].Status == progrep.OperationStatusPending { + ops[i].Status = progrep.OperationStatusCanceled } } +} - stageReports = append(stageReports, progrep.StageReport{Operations: operations}) - - return progrep.ProgressReport{ - StageReports: stageReports, +func planIDPrefix(planNumber int) string { + if planNumber == 1 { + return "" } + + return fmt.Sprintf("%d/", planNumber) } -func buildStageReport(ops []opEntry) progrep.StageReport { - operations := make([]progrep.Operation, len(ops)) - for i, e := range ops { - operations[i] = progrep.Operation{ - OperationRef: progrep.OperationRef{ - ObjectRef: e.ref, - Type: e.typ, - Iteration: e.iteration, - }, - Status: e.status, +func planSinkOperationIDs(predMap map[string]map[string]graph.Edge[string], idPrefix string) []string { + hasSuccessors := make(map[string]struct{}, len(predMap)) + for _, preds := range predMap { + for predID := range preds { + hasSuccessors[predID] = struct{}{} } } - return progrep.StageReport{ - Operations: operations, + var sinkIDs []string + for opID := range predMap { + if _, ok := hasSuccessors[opID]; !ok { + sinkIDs = append(sinkIDs, idPrefix+opID) + } } + + sort.Strings(sinkIDs) + + return sinkIDs } func safeSend(ch chan<- progrep.ProgressReport, report progrep.ProgressReport) (sent bool) { diff --git a/pkg/plan/legacy_progress_reporter_ai_test.go b/pkg/plan/legacy_progress_reporter_ai_test.go index 8c68af49..19abc737 100644 --- a/pkg/plan/legacy_progress_reporter_ai_test.go +++ b/pkg/plan/legacy_progress_reporter_ai_test.go @@ -4,14 +4,20 @@ package plan import ( "context" + "encoding/json" + "fmt" + "strings" + "sync" "testing" "time" + "github.com/samber/lo" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime/schema" + "github.com/werf/nelm/pkg/common" + helmrelease "github.com/werf/nelm/pkg/helm/pkg/release" "github.com/werf/nelm/pkg/legacy/progrep" ) @@ -19,10 +25,7 @@ func TestAI_BuildResolvedNamespaces(t *testing.T) { mapper := newFakeRESTMapper() releaseNS := "release-ns" - opNamespaced := &Operation{ - Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, - Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm1", "", gvkConfigMap)}, - } + opNamespaced := createConfigMapOp("cm1") opNamespacedExplicit := &Operation{ Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm2", "custom-ns", gvkConfigMap)}, @@ -31,10 +34,7 @@ func TestAI_BuildResolvedNamespaces(t *testing.T) { Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("my-ns", "", gvkNamespace)}, } - opMeta := &Operation{ - Type: OperationTypeNoop, Version: OperationVersionNoop, Category: OperationCategoryMeta, - Config: &OperationConfigNoop{OpID: "stage/start"}, - } + opMeta := stageMetaOp("stage/install/start") opTrack := &Operation{ Type: OperationTypeTrackReadiness, Version: OperationVersionTrackReadiness, Category: OperationCategoryTrack, Config: &OperationConfigTrackReadiness{ResourceMeta: makeResourceMeta("dep1", "", gvkDeployment)}, @@ -46,7 +46,7 @@ func TestAI_BuildResolvedNamespaces(t *testing.T) { assert.Equal(t, releaseNS, resolved[opNamespaced.ID()]) assert.Equal(t, "custom-ns", resolved[opNamespacedExplicit.ID()]) - assert.Equal(t, "", resolved[opClusterScoped.ID()]) + assert.Empty(t, resolved[opClusterScoped.ID()]) _, metaPresent := resolved[opMeta.ID()] assert.False(t, metaPresent, "meta operations should not appear in resolved namespaces") @@ -64,11 +64,8 @@ func TestAI_ExtractObjectRef(t *testing.T) { wantGVK schema.GroupVersionKind }{ { - name: "Create", - op: &Operation{ - Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, - Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm1", "", gvkConfigMap)}, - }, + name: "Create", + op: createConfigMapOp("cm1"), wantName: "cm1", wantGVK: gvkConfigMap, }, @@ -151,70 +148,26 @@ func TestAI_ExtractObjectRef(t *testing.T) { } func TestAI_ExtractObjectRef_PanicsOnUnexpectedConfig(t *testing.T) { - op := &Operation{ - Type: OperationTypeNoop, - Version: OperationVersionNoop, - Category: OperationCategoryMeta, - Config: &OperationConfigNoop{OpID: "test"}, - } - assert.Panics(t, func() { - extractObjectRef(op, map[string]string{}) + extractObjectRef(stageMetaOp("stage/install/start"), map[string]string{}) }) } -func TestAI_MapOperationStatus(t *testing.T) { - tests := []struct { - input OperationStatus - expected progrep.OperationStatus - }{ - {input: OperationStatusUnknown, expected: progrep.OperationStatusPending}, - {input: OperationStatusPending, expected: progrep.OperationStatusProgressing}, - {input: OperationStatusCompleted, expected: progrep.OperationStatusCompleted}, - {input: OperationStatusFailed, expected: progrep.OperationStatusFailed}, - } - - for _, tt := range tests { - name := string(tt.input) - if name == "" { - name = "unknown" - } - - t.Run(name, func(t *testing.T) { - assert.Equal(t, tt.expected, mapOperationStatus(tt.input)) - }) - } -} - -func TestAI_MapOperationStatus_DefaultCase(t *testing.T) { - assert.Equal(t, progrep.OperationStatusPending, mapOperationStatus("some-unexpected-status")) +func TestAI_MapOperationCategory_PanicsOnUnknown(t *testing.T) { + assert.Panics(t, func() { + mapOperationCategory("unknown-category") + }) } -func TestAI_MapOperationType(t *testing.T) { - tests := []struct { - input OperationType - expected progrep.OperationType - }{ - {input: OperationTypeCreate, expected: progrep.OperationTypeCreate}, - {input: OperationTypeUpdate, expected: progrep.OperationTypeUpdate}, - {input: OperationTypeDelete, expected: progrep.OperationTypeDelete}, - {input: OperationTypeApply, expected: progrep.OperationTypeApply}, - {input: OperationTypeRecreate, expected: progrep.OperationTypeRecreate}, - {input: OperationTypeTrackReadiness, expected: progrep.OperationTypeTrackReadiness}, - {input: OperationTypeTrackPresence, expected: progrep.OperationTypeTrackPresence}, - {input: OperationTypeTrackAbsence, expected: progrep.OperationTypeTrackAbsence}, - } - - for _, tt := range tests { - t.Run(string(tt.input), func(t *testing.T) { - assert.Equal(t, tt.expected, mapOperationType(tt.input)) - }) - } +func TestAI_MapOperationType_PanicsOnNoopWithoutStageSuffix(t *testing.T) { + assert.Panics(t, func() { + mapOperationType(stageMetaOp("something-else")) + }) } func TestAI_MapOperationType_PanicsOnUnknown(t *testing.T) { assert.Panics(t, func() { - mapOperationType("unknown-type") + mapOperationType(&Operation{Type: "unknown-type"}) }) } @@ -235,181 +188,249 @@ func TestAI_NewLegacyProgressReporter_PanicsOnUnbufferedChannel(t *testing.T) { }) } -func TestAI_ReportOperationStatus_SetsStatusAndReports(t *testing.T) { +func TestAI_ProgressReport_JSONShape(t *testing.T) { ch := make(chan progrep.ProgressReport, 64) reporter := NewLegacyProgressReporter(ch) - op := &Operation{ - Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, - Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm1", "", gvkConfigMap)}, - } - p := buildTestPlan([]*Operation{op}, nil) - reporter.startStage(p, map[string]string{op.ID(): "default"}, nil, nil) - drainChannel(ch) + start := stageMetaOp("stage/install/start") + cm := createConfigMapOp("cm1") + p := buildTestPlan([]*Operation{start, cm}, map[int][]int{1: {0}}) - reportOperationStatus(op, OperationStatusPending, reporter) - - assert.Equal(t, OperationStatusPending, op.Status) + startTestPlan(reporter, p, nil, StartPlanOptions{}) reports := drainChannel(ch) require.NotEmpty(t, reports) - activeOps := reports[len(reports)-1].StageReports[0].Operations - require.Len(t, activeOps, 1) - assert.Equal(t, progrep.OperationStatusProgressing, activeOps[0].Status) + raw, err := json.Marshal(reports[len(reports)-1]) + require.NoError(t, err) + + var decoded map[string]any + require.NoError(t, json.Unmarshal(raw, &decoded)) + + require.Contains(t, decoded, "operations") + assert.NotContains(t, decoded, "stageReports") + + operations, ok := decoded["operations"].([]any) + require.True(t, ok) + require.Len(t, operations, 2) + + startJSON, ok := operations[0].(map[string]any) + require.True(t, ok) + assert.Equal(t, start.ID(), startJSON["id"]) + assert.Equal(t, "meta", startJSON["category"]) + assert.Equal(t, "StageStart", startJSON["type"]) + assert.Equal(t, "Pending", startJSON["status"]) + assert.Equal(t, []any{}, startJSON["dependsOn"], "dependsOn must be an empty array, not null") + assert.NotContains(t, startJSON, "waitingFor") + + cmJSON, ok := operations[1].(map[string]any) + require.True(t, ok) + assert.Equal(t, cm.ID(), cmJSON["id"]) + assert.Equal(t, "resource", cmJSON["category"]) + assert.Equal(t, []any{start.ID()}, cmJSON["dependsOn"]) + assert.Equal(t, "ConfigMap", cmJSON["Kind"]) + assert.Equal(t, "cm1", cmJSON["name"]) + assert.Equal(t, "default", cmJSON["namespace"]) } -func TestAI_ReportOperationStatus_SetsStatusWithoutReporter(t *testing.T) { - op := &Operation{ - Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, - Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm1", "", gvkConfigMap)}, +func TestAI_ReportOperationStatus_SetsStatusAndReports(t *testing.T) { + tests := []struct { + planStatus OperationStatus + reportStatus progrep.OperationStatus + }{ + {planStatus: OperationStatusUnknown, reportStatus: progrep.OperationStatusPending}, + {planStatus: OperationStatusPending, reportStatus: progrep.OperationStatusProgressing}, + {planStatus: OperationStatusCompleted, reportStatus: progrep.OperationStatusCompleted}, + {planStatus: OperationStatusFailed, reportStatus: progrep.OperationStatusFailed}, } + for _, tt := range tests { + t.Run(string(tt.reportStatus), func(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) + + op := createConfigMapOp("cm1") + startTestPlan(reporter, buildTestPlan([]*Operation{op}, nil), nil, StartPlanOptions{}) + drainChannel(ch) + + reportOperationStatus(op, tt.planStatus, reporter) + + assert.Equal(t, tt.planStatus, op.Status) + + ops := lastReportOperations(t, ch) + require.Len(t, ops, 1) + assert.Equal(t, tt.reportStatus, ops[0].Status) + }) + } +} + +func TestAI_ReportOperationStatus_SetsStatusWithoutReporter(t *testing.T) { + op := createConfigMapOp("cm1") + reportOperationStatus(op, OperationStatusCompleted, nil) assert.Equal(t, OperationStatusCompleted, op.Status) } -func TestAI_ReportStatus_DoesNotPanicOnClosedChannel(t *testing.T) { +func TestAI_ReportStatus_ConcurrentCallsAreSafe(t *testing.T) { ch := make(chan progrep.ProgressReport, 1) reporter := NewLegacyProgressReporter(ch) - ops := []*Operation{ - { - Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, - Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm1", "", gvkConfigMap)}, - }, + ops := lo.Times(50, func(i int) *Operation { + return createConfigMapOp(fmt.Sprintf("cm-%02d", i)) + }) + startTestPlan(reporter, buildTestPlan(ops, nil), nil, StartPlanOptions{}) + + var wg sync.WaitGroup + for _, op := range ops { + wg.Add(1) + + go func(op *Operation) { + defer wg.Done() + + reporter.ReportStatus(op.ID(), progrep.OperationStatusProgressing) + reporter.ReportStatus(op.ID(), progrep.OperationStatusCompleted) + }(op) } - p := buildTestPlan(ops, nil) - reporter.startStage(p, map[string]string{ops[0].ID(): "default"}, nil, nil) + + wg.Wait() + drainChannel(ch) + + reporter.Stop(context.Background()) + + final := lastReportOperations(t, ch) + require.Len(t, final, 50) + + for _, op := range final { + assert.Equal(t, progrep.OperationStatusCompleted, op.Status) + } +} + +func TestAI_ReportStatus_DoesNotPanicOnClosedChannel(t *testing.T) { + ch := make(chan progrep.ProgressReport, 1) + reporter := NewLegacyProgressReporter(ch) + + op := createConfigMapOp("cm1") + p := buildTestPlan([]*Operation{op}, nil) + startTestPlan(reporter, p, nil, StartPlanOptions{}) close(ch) assert.NotPanics(t, func() { - reporter.ReportStatus(ops[0].ID(), progrep.OperationStatusCompleted) + reporter.ReportStatus(op.ID(), progrep.OperationStatusCompleted) }) } -func TestAI_ReportStatus_SendsSnapshot(t *testing.T) { +func TestAI_ReportStatus_KeepsOperationOrder(t *testing.T) { ch := make(chan progrep.ProgressReport, 64) reporter := NewLegacyProgressReporter(ch) - ops := []*Operation{ - { - Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, - Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm1", "", gvkConfigMap)}, - }, - { - Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, - Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("svc1", "", gvkService)}, - }, - } - p := buildTestPlan(ops, nil) + a := createConfigMapOp("cm-a") + b := createConfigMapOp("cm-b") + c := createConfigMapOp("cm-c") + untouched := []*InstallableResourceInfo{makeUntouchedInfo("cm-u", "default", gvkConfigMap)} + startTestPlan(reporter, buildTestPlan([]*Operation{a, b, c}, map[int][]int{2: {0, 1}}), untouched, StartPlanOptions{}) - resolvedNS := map[string]string{ - ops[0].ID(): "default", - ops[1].ID(): "default", - } + before := operationIDs(lastReportOperations(t, ch)) - reporter.startStage(p, resolvedNS, nil, nil) - drainChannel(ch) + reporter.ReportStatus(c.ID(), progrep.OperationStatusCompleted) + reporter.ReportStatus(a.ID(), progrep.OperationStatusFailed) - reporter.ReportStatus(ops[0].ID(), progrep.OperationStatusCompleted) + assert.Equal(t, before, operationIDs(lastReportOperations(t, ch))) +} - reports := drainChannel(ch) - require.NotEmpty(t, reports, "expected at least one report after ReportStatus") +func TestAI_ReportStatus_PreviousPlanOperationsNotAddressable(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) - last := reports[len(reports)-1] - require.Len(t, last.StageReports, 1) + op1 := createConfigMapOp("cm1") + startTestPlan(reporter, buildTestPlan([]*Operation{op1}, nil), nil, StartPlanOptions{}) - activeOps := last.StageReports[0].Operations - require.Len(t, activeOps, 2) + op2 := createConfigMapOp("cm2") + startTestPlan(reporter, buildTestPlan([]*Operation{op2}, nil), nil, StartPlanOptions{}) + drainChannel(ch) - opStatuses := map[string]progrep.OperationStatus{} - for _, op := range activeOps { - opStatuses[op.Name] = op.Status - } + reporter.ReportStatus(op1.ID(), progrep.OperationStatusCompleted) + assert.Empty(t, drainChannel(ch), "operations of an earlier plan must not be addressable") + + reporter.ReportStatus(op2.ID(), progrep.OperationStatusCompleted) - assert.Equal(t, progrep.OperationStatusCompleted, opStatuses["cm1"]) - assert.Equal(t, progrep.OperationStatusPending, opStatuses["svc1"]) + ops := lastReportOperations(t, ch) + require.Len(t, ops, 2) + assert.Equal(t, progrep.OperationStatusCanceled, ops[0].Status, "the ignored status change did not leak into the previous plan") + assert.Equal(t, progrep.OperationStatusCompleted, ops[1].Status) } -func TestAI_ReportStatus_UnknownOpIDIsIgnored(t *testing.T) { +func TestAI_ReportStatus_SendsSnapshot(t *testing.T) { ch := make(chan progrep.ProgressReport, 64) reporter := NewLegacyProgressReporter(ch) - ops := []*Operation{ - { - Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, - Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm1", "", gvkConfigMap)}, - }, + cm := createConfigMapOp("cm1") + svc := &Operation{ + Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, + Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("svc1", "", gvkService)}, } - p := buildTestPlan(ops, nil) + p := buildTestPlan([]*Operation{cm, svc}, nil) - reporter.startStage(p, map[string]string{ops[0].ID(): "default"}, nil, nil) + startTestPlan(reporter, p, nil, StartPlanOptions{}) drainChannel(ch) - reporter.ReportStatus("nonexistent/op/id", progrep.OperationStatusCompleted) + reporter.ReportStatus(cm.ID(), progrep.OperationStatusCompleted) - reports := drainChannel(ch) - assert.Empty(t, reports, "expected no report for unknown op ID") + ops := operationsByID(lastReportOperations(t, ch)) + require.Len(t, ops, 2) + + assert.Equal(t, progrep.OperationStatusCompleted, ops[cm.ID()].Status) + assert.Equal(t, progrep.OperationStatusPending, ops[svc.ID()].Status) } -func TestAI_ReportStatus_WaitingForPopulation(t *testing.T) { +func TestAI_ReportStatus_SentReportsAreImmutable(t *testing.T) { ch := make(chan progrep.ProgressReport, 64) reporter := NewLegacyProgressReporter(ch) - opA := &Operation{ - Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, - Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm1", "", gvkConfigMap)}, - } - opB := &Operation{ - Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, - Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("svc1", "", gvkService)}, - } + op := createConfigMapOp("cm1") + startTestPlan(reporter, buildTestPlan([]*Operation{op}, nil), nil, StartPlanOptions{}) - p := buildTestPlan([]*Operation{opA, opB}, map[int][]int{1: {0}}) + initial := lastReportOperations(t, ch) + require.Equal(t, progrep.OperationStatusPending, initial[0].Status) - resolvedNS := map[string]string{ - opA.ID(): "default", - opB.ID(): "default", - } - reporter.startStage(p, resolvedNS, nil, nil) + reporter.ReportStatus(op.ID(), progrep.OperationStatusCompleted) - reports := drainChannel(ch) - require.NotEmpty(t, reports) + assert.Equal(t, progrep.OperationStatusPending, initial[0].Status, "a report handed to the consumer must not change afterwards") + assert.Equal(t, progrep.OperationStatusCompleted, lastReportOperations(t, ch)[0].Status) +} - last := reports[len(reports)-1] - activeOps := last.StageReports[0].Operations +func TestAI_ReportStatus_SkipsSnapshotWhileChannelIsFull(t *testing.T) { + ch := make(chan progrep.ProgressReport, 1) + reporter := NewLegacyProgressReporter(ch) - var opBReport *progrep.Operation - for i := range activeOps { - if activeOps[i].Name == "svc1" { - opBReport = &activeOps[i] - } - } + op := createConfigMapOp("cm1") + startTestPlan(reporter, buildTestPlan([]*Operation{op}, nil), nil, StartPlanOptions{}) + require.Len(t, ch, 1, "the initial report fills the channel") - require.NotNil(t, opBReport, "expected svc1 in operations") - require.Len(t, opBReport.WaitingFor, 1, "svc1 should be waiting for cm1") - assert.Equal(t, "cm1", opBReport.WaitingFor[0].Name) - assert.Equal(t, progrep.OperationTypeCreate, opBReport.WaitingFor[0].Type) - assert.Equal(t, 0, opBReport.WaitingFor[0].Iteration) + reporter.ReportStatus(op.ID(), progrep.OperationStatusCompleted) - reporter.ReportStatus(opA.ID(), progrep.OperationStatusCompleted) - reports = drainChannel(ch) - require.NotEmpty(t, reports) + stale := <-ch + assert.Equal(t, progrep.OperationStatusPending, stale.Operations[0].Status, "the report already in the channel is left as is") + assert.Empty(t, ch, "no snapshot is queued while the consumer is behind") - last = reports[len(reports)-1] - activeOps = last.StageReports[0].Operations + reporter.Stop(context.Background()) - opBReport = nil - for i := range activeOps { - if activeOps[i].Name == "svc1" { - opBReport = &activeOps[i] - } - } + final := <-ch + assert.Equal(t, progrep.OperationStatusCompleted, final.Operations[0].Status, "the status change is kept and reaches the final report") +} - require.NotNil(t, opBReport) - assert.Empty(t, opBReport.WaitingFor, "svc1 should no longer be waiting after cm1 completed") +func TestAI_ReportStatus_UnknownOpIDIsIgnored(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) + + p := buildTestPlan([]*Operation{createConfigMapOp("cm1")}, nil) + + startTestPlan(reporter, p, nil, StartPlanOptions{}) + drainChannel(ch) + + reporter.ReportStatus("nonexistent/op/id", progrep.OperationStatusCompleted) + + assert.Empty(t, drainChannel(ch), "expected no report for unknown op ID") } func TestAI_ResolveNamespace(t *testing.T) { @@ -492,202 +513,603 @@ func TestAI_SendNonBlocking_DropsWhenFull(t *testing.T) { ch <- progrep.ProgressReport{} - sendNonBlocking(ch, progrep.ProgressReport{StageReports: []progrep.StageReport{{}}}) + sendNonBlocking(ch, progrep.ProgressReport{Operations: []progrep.Operation{{}}}) assert.Len(t, ch, 1) msg := <-ch - assert.Empty(t, msg.StageReports, "expected the original empty report, not the dropped one") + assert.Empty(t, msg.Operations, "expected the original empty report, not the dropped one") } -func TestAI_StartStage_FiltersNonResourceOps(t *testing.T) { +func TestAI_StartPlan_CancelsNothingAfterCompletedPlan(t *testing.T) { ch := make(chan progrep.ProgressReport, 64) reporter := NewLegacyProgressReporter(ch) - ops := []*Operation{ - { - Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, - Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm1", "", gvkConfigMap)}, - }, - { - Type: OperationTypeNoop, Version: OperationVersionNoop, Category: OperationCategoryMeta, - Config: &OperationConfigNoop{OpID: "stage/start"}, - }, + start := stageMetaOp("stage/install/start") + a := createConfigMapOp("cm-a") + b := createConfigMapOp("cm-b") + end := stageMetaOp("stage/install/end") + untouched := []*InstallableResourceInfo{makeUntouchedInfo("cm-untouched", "default", gvkConfigMap)} + firstPlanOps := []*Operation{start, a, b, end} + startTestPlan(reporter, buildTestPlan(firstPlanOps, map[int][]int{1: {0}, 2: {0}, 3: {1, 2}}), untouched, StartPlanOptions{}) + + for _, op := range firstPlanOps { + reporter.ReportStatus(op.ID(), progrep.OperationStatusCompleted) } - p := buildTestPlan(ops, nil) - reporter.startStage(p, map[string]string{ops[0].ID(): "default"}, nil, nil) - reports := drainChannel(ch) - require.NotEmpty(t, reports) + drainChannel(ch) + + next := createConfigMapOp("cm-next") + startTestPlan(reporter, buildTestPlan([]*Operation{next}, nil), nil, StartPlanOptions{}) + + ops := lastReportOperations(t, ch) + require.Len(t, ops, len(firstPlanOps)+2) + + for _, op := range ops { + assert.NotEqual(t, progrep.OperationStatusCanceled, op.Status, "a fully completed plan leaves nothing to cancel: %s", op.ID) + } - last := reports[len(reports)-1] - require.Len(t, last.StageReports, 1) + byID := operationsByID(ops) + for _, op := range firstPlanOps { + assert.Equal(t, progrep.OperationStatusCompleted, byID[op.ID()].Status, op.ID()) + } - assert.Len(t, last.StageReports[0].Operations, 1) - assert.Equal(t, "cm1", last.StageReports[0].Operations[0].Name) + assert.Equal(t, progrep.OperationStatusCompleted, byID["noop/1/0/default::ConfigMap:cm-untouched"].Status) + assert.Equal(t, progrep.OperationStatusPending, byID["2/"+next.ID()].Status) } -func TestAI_StartStage_FreezesPreviousStage(t *testing.T) { +func TestAI_StartPlan_CancelsPendingOperationsOfPreviousPlan(t *testing.T) { ch := make(chan progrep.ProgressReport, 64) reporter := NewLegacyProgressReporter(ch) - ops1 := []*Operation{ - { - Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, - Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm1", "", gvkConfigMap)}, - }, - } - p1 := buildTestPlan(ops1, nil) - reporter.startStage(p1, map[string]string{ops1[0].ID(): "default"}, nil, nil) + done := createConfigMapOp("cm-done") + failed := createConfigMapOp("cm-failed") + neverStarted := createConfigMapOp("cm-never") + untouched := []*InstallableResourceInfo{makeUntouchedInfo("cm-untouched", "default", gvkConfigMap)} + startTestPlan(reporter, buildTestPlan([]*Operation{done, failed, neverStarted}, map[int][]int{1: {0}, 2: {1}}), untouched, StartPlanOptions{}) - reporter.ReportStatus(ops1[0].ID(), progrep.OperationStatusCompleted) + reporter.ReportStatus(done.ID(), progrep.OperationStatusCompleted) + reporter.ReportStatus(failed.ID(), progrep.OperationStatusFailed) drainChannel(ch) - ops2 := []*Operation{ - { - Type: OperationTypeDelete, Version: OperationVersionDelete, Category: OperationCategoryResource, - Config: &OperationConfigDelete{ResourceMeta: makeResourceMeta("svc1", "", gvkService)}, - }, - } - p2 := buildTestPlan(ops2, nil) - reporter.startStage(p2, map[string]string{ops2[0].ID(): "default"}, nil, nil) + next := createConfigMapOp("cm-next") + startTestPlan(reporter, buildTestPlan([]*Operation{next}, nil), nil, StartPlanOptions{}) - reports := drainChannel(ch) - require.NotEmpty(t, reports) + ops := operationsByID(lastReportOperations(t, ch)) + assert.Equal(t, progrep.OperationStatusCompleted, ops[done.ID()].Status) + assert.Equal(t, progrep.OperationStatusFailed, ops[failed.ID()].Status) + assert.Equal(t, progrep.OperationStatusCanceled, ops[neverStarted.ID()].Status, "an operation the previous plan never reached is canceled") + assert.Equal(t, progrep.OperationStatusCompleted, ops["noop/1/0/default::ConfigMap:cm-untouched"].Status) + assert.Equal(t, progrep.OperationStatusPending, ops["2/"+next.ID()].Status, "the new plan starts Pending") +} - last := reports[len(reports)-1] - require.Len(t, last.StageReports, 2, "expected frozen stage + active stage") +func TestAI_StartPlan_DependsOnSortedByID(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) + + a := createConfigMapOp("cm-a") + b := createConfigMapOp("cm-b") + c := createConfigMapOp("cm-c") + p := buildTestPlan([]*Operation{a, b, c}, map[int][]int{2: {1, 0}}) - frozenOps := last.StageReports[0].Operations - require.Len(t, frozenOps, 1) - assert.Equal(t, "cm1", frozenOps[0].Name) - assert.Equal(t, progrep.OperationStatusCompleted, frozenOps[0].Status) + startTestPlan(reporter, p, nil, StartPlanOptions{}) - activeOps := last.StageReports[1].Operations - require.Len(t, activeOps, 1) - assert.Equal(t, "svc1", activeOps[0].Name) - assert.Equal(t, progrep.OperationStatusPending, activeOps[0].Status) + ops := operationsByID(lastReportOperations(t, ch)) + assert.Equal(t, []string{a.ID(), b.ID()}, ops[c.ID()].DependsOn) + assert.Empty(t, ops[a.ID()].DependsOn) + assert.Empty(t, ops[b.ID()].DependsOn) } -func TestAI_StartStage_NoUntouchedResourcesOmitsBackfill(t *testing.T) { - mapper := newFakeRESTMapper() - releaseNS := "release-ns" +func TestAI_StartPlan_DependsOnUsesPrefixedIDsInLaterPlans(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) - op := &Operation{ - Type: OperationTypeDelete, Version: OperationVersionDelete, Category: OperationCategoryResource, - Config: &OperationConfigDelete{ResourceMeta: makeResourceMeta("job1", releaseNS, gvkConfigMap)}, + startTestPlan(reporter, buildTestPlan([]*Operation{createConfigMapOp("cm1")}, nil), nil, StartPlanOptions{}) + + a := createConfigMapOp("cm-a") + b := createConfigMapOp("cm-b") + startTestPlan(reporter, buildTestPlan([]*Operation{a, b}, map[int][]int{1: {0}}), nil, StartPlanOptions{}) + + ops := operationsByID(lastReportOperations(t, ch)) + require.Contains(t, ops, "2/"+b.ID()) + assert.Equal(t, []string{"2/" + a.ID()}, ops["2/"+b.ID()].DependsOn) +} + +func TestAI_StartPlan_IncludesMetaAndReleaseOperations(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) + + start := stageMetaOp("stage/install/start") + cm := createConfigMapOp("cm1") + end := stageMetaOp("stage/install/end") + rel := &Operation{ + Type: OperationTypeDeleteRelease, Version: OperationVersionDeleteRelease, Category: OperationCategoryRelease, + Config: &OperationConfigDeleteRelease{ReleaseName: "rel", ReleaseNamespace: "default", ReleaseRevision: 1}, } - p := buildTestPlan([]*Operation{op}, nil) + p := buildTestPlan([]*Operation{start, cm, end, rel}, map[int][]int{1: {0}, 2: {1}, 3: {2}}) - untouched := []*InstallableResourceInfo{ - makeUntouchedInfo("cm1", releaseNS, gvkConfigMap), - makeUntouchedInfo("cm2", releaseNS, gvkConfigMap), + startTestPlan(reporter, p, nil, StartPlanOptions{}) + + ops := lastReportOperations(t, ch) + require.Equal(t, []string{start.ID(), cm.ID(), end.ID(), rel.ID()}, operationIDs(ops)) + + assert.Equal(t, progrep.OperationTypeStageStart, ops[0].Type) + assert.Equal(t, progrep.OperationCategoryMeta, ops[0].Category) + assert.Equal(t, progrep.ObjectRef{}, ops[0].ObjectRef) + assert.Empty(t, ops[0].DependsOn) + + assert.Equal(t, progrep.OperationTypeCreate, ops[1].Type) + assert.Equal(t, progrep.OperationCategoryResource, ops[1].Category) + assert.Equal(t, []string{start.ID()}, ops[1].DependsOn) + + assert.Equal(t, progrep.OperationTypeStageEnd, ops[2].Type) + assert.Equal(t, progrep.OperationCategoryMeta, ops[2].Category) + assert.Equal(t, []string{cm.ID()}, ops[2].DependsOn) + + assert.Equal(t, progrep.OperationTypeDeleteRelease, ops[3].Type) + assert.Equal(t, progrep.OperationCategoryRelease, ops[3].Category) + assert.Equal(t, progrep.ObjectRef{}, ops[3].ObjectRef) + assert.Equal(t, []string{end.ID()}, ops[3].DependsOn) + + for _, op := range ops { + assert.Equal(t, progrep.OperationStatusPending, op.Status) } +} - t.Run("included by default", func(t *testing.T) { - ch := make(chan progrep.ProgressReport, 64) - reporter := NewLegacyProgressReporter(ch) +func TestAI_StartPlan_LaterPlanRootsDependOnPreviousPlanSinks(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) - reporter.StartStage(p, releaseNS, untouched, mapper, StartStageOptions{}) + a := createConfigMapOp("cm-a") + b := createConfigMapOp("cm-b") + c := createConfigMapOp("cm-c") + untouched := []*InstallableResourceInfo{makeUntouchedInfo("cm-untouched", "default", gvkConfigMap)} + startTestPlan(reporter, buildTestPlan([]*Operation{a, b, c}, map[int][]int{1: {0}}), untouched, StartPlanOptions{}) + + x := createConfigMapOp("cm-x") + y := createConfigMapOp("cm-y") + z := createConfigMapOp("cm-z") + startTestPlan(reporter, buildTestPlan([]*Operation{x, y, z}, map[int][]int{1: {0}}), untouched, StartPlanOptions{}) + + ops := operationsByID(lastReportOperations(t, ch)) + + previousSinks := []string{b.ID(), c.ID()} + assert.Equal(t, previousSinks, ops["2/"+x.ID()].DependsOn, "root of the next plan depends on all sinks of the previous plan") + assert.Equal(t, previousSinks, ops["2/"+z.ID()].DependsOn) + assert.Equal(t, []string{"2/" + x.ID()}, ops["2/"+y.ID()].DependsOn, "non-root operations keep their own predecessors only") + assert.Empty(t, ops["noop/1/0/default::ConfigMap:cm-untouched"].DependsOn, "untouched resources stay edgeless") + assert.Empty(t, ops[a.ID()].DependsOn, "the first plan has nothing to depend on") +} - reports := drainChannel(ch) - require.NotEmpty(t, reports) +func TestAI_StartPlan_MetaOperationStatusReported(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) - ops := reports[len(reports)-1].StageReports[0].Operations - assert.Len(t, ops, 3, "plan op plus both untouched resources") - }) + start := stageMetaOp("stage/install/start") + p := buildTestPlan([]*Operation{start}, nil) - t.Run("omitted when NoUntouchedResources", func(t *testing.T) { - ch := make(chan progrep.ProgressReport, 64) - reporter := NewLegacyProgressReporter(ch) + startTestPlan(reporter, p, nil, StartPlanOptions{}) + drainChannel(ch) - reporter.StartStage(p, releaseNS, untouched, mapper, StartStageOptions{ - NoUntouchedResources: true, - }) + reporter.ReportStatus(start.ID(), progrep.OperationStatusCompleted) - reports := drainChannel(ch) - require.NotEmpty(t, reports) + ops := lastReportOperations(t, ch) + require.Len(t, ops, 1) + assert.Equal(t, progrep.OperationStatusCompleted, ops[0].Status) +} - ops := reports[len(reports)-1].StageReports[0].Operations - require.Len(t, ops, 1, "only the plan's own operation") - assert.Equal(t, "job1", ops[0].Name) - assert.Equal(t, progrep.OperationTypeDelete, ops[0].Type) - }) +func TestAI_StartPlan_OperationFields(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) + + op := createConfigMapOp("cm1") + op.Iteration = 1 + p := buildTestPlan([]*Operation{op}, nil) + + startTestPlan(reporter, p, nil, StartPlanOptions{}) + + ops := lastReportOperations(t, ch) + require.Len(t, ops, 1) + + assert.Equal(t, op.ID(), ops[0].ID) + assert.Equal(t, "create/1/1/::ConfigMap:cm1", ops[0].ID) + assert.Equal(t, progrep.OperationCategoryResource, ops[0].Category) + assert.Equal(t, progrep.OperationTypeCreate, ops[0].Type) + assert.Equal(t, 1, ops[0].Iteration) + assert.Equal(t, progrep.OperationStatusPending, ops[0].Status) + assert.Equal(t, gvkConfigMap, ops[0].GroupVersionKind) + assert.Equal(t, "cm1", ops[0].Name) + assert.Equal(t, "default", ops[0].Namespace) + assert.Empty(t, ops[0].DependsOn) } -func TestAI_StartStage_ReportStatusNeverAffectsUntouchedEntry(t *testing.T) { +func TestAI_StartPlan_OperationNamespaceResolution(t *testing.T) { ch := make(chan progrep.ProgressReport, 64) reporter := NewLegacyProgressReporter(ch) - op := &Operation{ + defaulted := createConfigMapOp("cm1") + explicit := &Operation{ Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, - Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm1", "default", gvkConfigMap)}, + Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm2", "custom-ns", gvkConfigMap)}, + } + clusterScoped := &Operation{ + Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, + Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("my-ns", "ignored", gvkNamespace)}, + } + unknownKind := &Operation{ + Type: OperationTypeTrackReadiness, Version: OperationVersionTrackReadiness, Category: OperationCategoryTrack, + Config: &OperationConfigTrackReadiness{ResourceMeta: makeResourceMeta("widget", "", gvkCRD)}, } - p := buildTestPlan([]*Operation{op}, nil) - untouched := makeUntouchedInfo("cm2", "default", gvkConfigMap) + p := buildTestPlan([]*Operation{defaulted, explicit, clusterScoped, unknownKind}, nil) + reporter.StartPlan(p, "release-ns", nil, newFakeRESTMapper(), StartPlanOptions{}) + + ops := operationsByID(lastReportOperations(t, ch)) + assert.Equal(t, "release-ns", ops[defaulted.ID()].Namespace) + assert.Equal(t, "custom-ns", ops[explicit.ID()].Namespace) + assert.Empty(t, ops[clusterScoped.ID()].Namespace) + assert.Equal(t, "release-ns", ops[unknownKind.ID()].Namespace, "unknown kinds are assumed namespaced") +} + +func TestAI_StartPlan_OperationTypesAndCategories(t *testing.T) { + rel := &helmrelease.Release{Name: "rel", Namespace: "default", Version: 1, Info: &helmrelease.Info{}} - reporter.startStage( - p, - map[string]string{op.ID(): "default"}, - []*InstallableResourceInfo{untouched}, - map[string]string{untouched.ID(): "default"}, - ) + tests := []struct { + name string + op *Operation + wantType progrep.OperationType + wantCategory progrep.OperationCategory + }{ + { + name: "Create", + op: createConfigMapOp("cm1"), + wantType: progrep.OperationTypeCreate, + wantCategory: progrep.OperationCategoryResource, + }, + { + name: "Update", + op: &Operation{ + Type: OperationTypeUpdate, Version: OperationVersionUpdate, Category: OperationCategoryResource, + Config: &OperationConfigUpdate{ResourceSpec: makeResourceSpec("cm1", "", gvkConfigMap)}, + }, + wantType: progrep.OperationTypeUpdate, + wantCategory: progrep.OperationCategoryResource, + }, + { + name: "Apply", + op: &Operation{ + Type: OperationTypeApply, Version: OperationVersionApply, Category: OperationCategoryResource, + Config: &OperationConfigApply{ResourceSpec: makeResourceSpec("cm1", "", gvkConfigMap)}, + }, + wantType: progrep.OperationTypeApply, + wantCategory: progrep.OperationCategoryResource, + }, + { + name: "Recreate", + op: &Operation{ + Type: OperationTypeRecreate, Version: OperationVersionRecreate, Category: OperationCategoryResource, + Config: &OperationConfigRecreate{ResourceSpec: makeResourceSpec("cm1", "", gvkConfigMap)}, + }, + wantType: progrep.OperationTypeRecreate, + wantCategory: progrep.OperationCategoryResource, + }, + { + name: "Delete", + op: &Operation{ + Type: OperationTypeDelete, Version: OperationVersionDelete, Category: OperationCategoryResource, + Config: &OperationConfigDelete{ResourceMeta: makeResourceMeta("cm1", "", gvkConfigMap)}, + }, + wantType: progrep.OperationTypeDelete, + wantCategory: progrep.OperationCategoryResource, + }, + { + name: "TrackReadiness", + op: &Operation{ + Type: OperationTypeTrackReadiness, Version: OperationVersionTrackReadiness, Category: OperationCategoryTrack, + Config: &OperationConfigTrackReadiness{ResourceMeta: makeResourceMeta("dep1", "", gvkDeployment)}, + }, + wantType: progrep.OperationTypeTrackReadiness, + wantCategory: progrep.OperationCategoryTrack, + }, + { + name: "TrackPresence", + op: &Operation{ + Type: OperationTypeTrackPresence, Version: OperationVersionTrackPresence, Category: OperationCategoryTrack, + Config: &OperationConfigTrackPresence{ResourceMeta: makeResourceMeta("svc1", "", gvkService)}, + }, + wantType: progrep.OperationTypeTrackPresence, + wantCategory: progrep.OperationCategoryTrack, + }, + { + name: "TrackAbsence", + op: &Operation{ + Type: OperationTypeTrackAbsence, Version: OperationVersionTrackAbsence, Category: OperationCategoryTrack, + Config: &OperationConfigTrackAbsence{ResourceMeta: makeResourceMeta("cm1", "", gvkConfigMap)}, + }, + wantType: progrep.OperationTypeTrackAbsence, + wantCategory: progrep.OperationCategoryTrack, + }, + { + name: "CreateRelease", + op: &Operation{ + Type: OperationTypeCreateRelease, Version: OperationVersionCreateRelease, Category: OperationCategoryRelease, + Config: &OperationConfigCreateRelease{Release: rel}, + }, + wantType: progrep.OperationTypeCreateRelease, + wantCategory: progrep.OperationCategoryRelease, + }, + { + name: "UpdateRelease", + op: &Operation{ + Type: OperationTypeUpdateRelease, Version: OperationVersionUpdateRelease, Category: OperationCategoryRelease, + Config: &OperationConfigUpdateRelease{Release: rel}, + }, + wantType: progrep.OperationTypeUpdateRelease, + wantCategory: progrep.OperationCategoryRelease, + }, + { + name: "DeleteRelease", + op: &Operation{ + Type: OperationTypeDeleteRelease, Version: OperationVersionDeleteRelease, Category: OperationCategoryRelease, + Config: &OperationConfigDeleteRelease{ReleaseName: "rel", ReleaseNamespace: "default", ReleaseRevision: 1}, + }, + wantType: progrep.OperationTypeDeleteRelease, + wantCategory: progrep.OperationCategoryRelease, + }, + { + name: "StageStart", + op: stageMetaOp("stage/install/start"), + wantType: progrep.OperationTypeStageStart, + wantCategory: progrep.OperationCategoryMeta, + }, + { + name: "StageEnd", + op: stageMetaOp("stage/install/end"), + wantType: progrep.OperationTypeStageEnd, + wantCategory: progrep.OperationCategoryMeta, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) + + startTestPlan(reporter, buildTestPlan([]*Operation{tt.op}, nil), nil, StartPlanOptions{}) + + ops := lastReportOperations(t, ch) + require.Len(t, ops, 1) + assert.Equal(t, tt.op.ID(), ops[0].ID) + assert.Equal(t, tt.wantType, ops[0].Type) + assert.Equal(t, tt.wantCategory, ops[0].Category) + }) + } +} + +func TestAI_StartPlan_RealPlansFormSingleChainedGraph(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) + + preInstall := realInstallableInfo("cm-pre", common.StagePreInstall) + install := realInstallableInfo("cm-main", common.StageInstall) + install.MustDeleteOnFailedInstall = true + + untouched := realInstallableInfo("cm-untouched", common.StageInstall) + untouched.MustInstall = ResourceInstallTypeNone + untouched.MustTrackReadiness = false + + infos := []*InstallableResourceInfo{preInstall, install, untouched} + + relInfos := []*ReleaseInfo{{ + Release: &helmrelease.Release{ + Name: "rel", + Namespace: "default", + Version: 1, + Info: &helmrelease.Info{}, + }, + Must: ReleaseTypeInstall, + MustFailOnFailedDeploy: true, + }} + + installPlan, err := BuildPlan(infos, nil, relInfos, BuildPlanOptions{}) + require.NoError(t, err) + + startTestPlan(reporter, installPlan, infos, StartPlanOptions{}) + + ops := lastReportOperations(t, ch) + byID := operationsByID(ops) + + planOps := lo.Filter(ops, func(op progrep.Operation, _ int) bool { + return op.Type != progrep.OperationTypeNoOp + }) + assert.Len(t, planOps, len(installPlan.Operations()), "every plan operation is reported") + assert.Equal(t, "noop/1/0/default::ConfigMap:cm-untouched", ops[0].ID, "untouched resource comes first") + + roots, sinks := graphRootsAndSinks(planOps) + require.Equal(t, []string{stageOperationID(common.StageInit, common.StageStartSuffix)}, roots, "a real plan has a single root") + require.Equal(t, []string{stageOperationID(common.StageFinal, common.StageEndSuffix)}, sinks, "a real plan has a single sink") + + createRelID := OperationID(OperationTypeCreateRelease, OperationVersionCreateRelease, 0, relInfos[0].Release.ID()) + updateRelID := OperationID(OperationTypeUpdateRelease, OperationVersionUpdateRelease, 0, relInfos[0].Release.ID()) + + assert.Equal(t, progrep.OperationTypeCreateRelease, byID[createRelID].Type) + assert.Equal(t, progrep.OperationCategoryRelease, byID[createRelID].Category) + assert.Equal(t, progrep.OperationTypeUpdateRelease, byID[updateRelID].Type) + assert.Equal(t, progrep.OperationCategoryRelease, byID[updateRelID].Category) + + preApplyID := OperationID(OperationTypeApply, OperationVersionApply, 0, preInstall.ID()) + mainApplyID := OperationID(OperationTypeApply, OperationVersionApply, 0, install.ID()) + mainTrackID := OperationID(OperationTypeTrackReadiness, OperationVersionTrackReadiness, 0, install.ID()) + + assert.True(t, reachesThroughDependsOn(byID, mainApplyID, preApplyID), "cross-stage order survives through meta operations") + assert.True(t, reachesThroughDependsOn(byID, updateRelID, mainTrackID), "release update depends on tracking") + assert.False(t, reachesThroughDependsOn(byID, preApplyID, mainApplyID)) + + for _, id := range []string{createRelID, preApplyID, mainApplyID} { + reportOperationStatus(lo.Must(installPlan.Operation(id)), OperationStatusCompleted, reporter) + } + + reportOperationStatus(lo.Must(installPlan.Operation(mainTrackID)), OperationStatusFailed, reporter) drainChannel(ch) - reporter.ReportStatus(untouched.ID(), progrep.OperationStatusFailed) + failurePlan, err := BuildFailurePlan(installPlan, infos, relInfos, BuildFailurePlanOptions{}) + require.NoError(t, err) + require.NotEmpty(t, failurePlan.Operations()) - noReports := drainChannel(ch) - assert.Empty(t, noReports, "untouched entry ID must not be addressable by ReportStatus") + startTestPlan(reporter, failurePlan, infos, StartPlanOptions{}) - reporter.ReportStatus(op.ID(), progrep.OperationStatusCompleted) + ops = lastReportOperations(t, ch) + byID = operationsByID(ops) - reports := drainChannel(ch) - require.NotEmpty(t, reports) + failureOps := lo.Filter(ops, func(op progrep.Operation, _ int) bool { + return strings.HasPrefix(op.ID, "2/") + }) + assert.Len(t, failureOps, len(failurePlan.Operations())) + assert.Len(t, ops, len(planOps)+1+len(failureOps), "install plan, untouched and failure plan are all kept") + + failureRoots, _ := graphRootsAndSinks(failureOps) + require.Len(t, failureRoots, 1, "a real failure plan has a single root") + assert.Equal(t, sinks, byID[failureRoots[0]].DependsOn, "the failure plan continues from the sink of the install plan") + + failureDeleteID := "2/" + OperationID(OperationTypeDelete, OperationVersionDelete, 0, install.ID()) + assert.True(t, reachesThroughDependsOn(byID, failureDeleteID, mainTrackID), "the whole run is one connected graph") + assert.Equal(t, progrep.OperationStatusFailed, byID[mainTrackID].Status) + assert.Equal(t, progrep.OperationStatusCanceled, byID[sinks[0]].Status, "the install plan never reached its end") + assert.Equal(t, progrep.OperationStatusCanceled, byID[updateRelID].Status) + assert.Equal(t, progrep.OperationStatusCompleted, byID[mainApplyID].Status) + assert.Equal(t, progrep.OperationStatusPending, byID[failureDeleteID].Status) +} - activeOps := reports[len(reports)-1].StageReports[0].Operations - require.Len(t, activeOps, 2) +func TestAI_StartPlan_SecondPlanAppendedWithPrefix(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) + + op1 := createConfigMapOp("cm1") + startTestPlan(reporter, buildTestPlan([]*Operation{op1}, nil), nil, StartPlanOptions{}) + reporter.ReportStatus(op1.ID(), progrep.OperationStatusCompleted) + drainChannel(ch) - statuses := map[string]progrep.OperationStatus{} - for _, o := range activeOps { - statuses[o.Name] = o.Status + op2 := &Operation{ + Type: OperationTypeDelete, Version: OperationVersionDelete, Category: OperationCategoryResource, + Config: &OperationConfigDelete{ResourceMeta: makeResourceMeta("svc1", "", gvkService)}, } + startTestPlan(reporter, buildTestPlan([]*Operation{op2}, nil), nil, StartPlanOptions{}) + + ops := lastReportOperations(t, ch) + require.Equal(t, []string{op1.ID(), "2/" + op2.ID()}, operationIDs(ops)) + + assert.Equal(t, progrep.OperationStatusCompleted, ops[0].Status) + assert.Equal(t, "cm1", ops[0].Name) - assert.Equal(t, progrep.OperationStatusCompleted, statuses["cm1"]) - assert.Equal(t, progrep.OperationStatusCompleted, statuses["cm2"], "untouched entry must remain Completed") + assert.Equal(t, progrep.OperationStatusPending, ops[1].Status) + assert.Equal(t, "svc1", ops[1].Name) + assert.Equal(t, progrep.OperationTypeDelete, ops[1].Type) } -func TestAI_StartStage_UntouchedAbsentResourceOmitted(t *testing.T) { +func TestAI_StartPlan_SkippedPlanDoesNotBreakChain(t *testing.T) { ch := make(chan progrep.ProgressReport, 64) reporter := NewLegacyProgressReporter(ch) - op := &Operation{ - Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, - Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm1", "default", gvkConfigMap)}, + op1 := createConfigMapOp("cm1") + startTestPlan(reporter, buildTestPlan([]*Operation{op1}, nil), nil, StartPlanOptions{}) + + skipped := stageMetaOp("stage/install/start") + startTestPlan(reporter, buildTestPlan([]*Operation{skipped}, nil), nil, StartPlanOptions{UntouchedResourcesOnly: true}) + + op3 := createConfigMapOp("cm3") + startTestPlan(reporter, buildTestPlan([]*Operation{op3}, nil), nil, StartPlanOptions{}) + + ops := operationsByID(lastReportOperations(t, ch)) + require.Contains(t, ops, "3/"+op3.ID()) + assert.Equal(t, []string{op1.ID()}, ops["3/"+op3.ID()].DependsOn, "a plan without reported operations is skipped over by the chain") +} + +func TestAI_StartPlan_ThirdPlanGetsOwnPrefix(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) + + op := createConfigMapOp("cm1") + + startTestPlan(reporter, buildTestPlan([]*Operation{op}, nil), nil, StartPlanOptions{}) + startTestPlan(reporter, buildTestPlan([]*Operation{op}, nil), nil, StartPlanOptions{}) + startTestPlan(reporter, buildTestPlan([]*Operation{op}, nil), nil, StartPlanOptions{}) + + ops := lastReportOperations(t, ch) + assert.Equal(t, []string{op.ID(), "2/" + op.ID(), "3/" + op.ID()}, operationIDs(ops)) +} + +func TestAI_StartPlan_TopologicalOrderIsDeterministic(t *testing.T) { + a := createConfigMapOp("cm-a") + b := createConfigMapOp("cm-b") + c := createConfigMapOp("cm-c") + d := createConfigMapOp("cm-d") + + var firstOrder []string + + for i := 0; i < 20; i++ { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) + + p := buildTestPlan([]*Operation{d, c, b, a}, map[int][]int{1: {3, 2}}) + startTestPlan(reporter, p, nil, StartPlanOptions{}) + + order := operationIDs(lastReportOperations(t, ch)) + require.Len(t, order, 4) + + assert.Less(t, lo.IndexOf(order, a.ID()), lo.IndexOf(order, c.ID()), "a precedes its dependent c") + assert.Less(t, lo.IndexOf(order, b.ID()), lo.IndexOf(order, c.ID()), "b precedes its dependent c") + + if firstOrder == nil { + firstOrder = order + + continue + } + + require.Equal(t, firstOrder, order, "order must not depend on map iteration") } +} + +func TestAI_StartPlan_UntouchedAbsentResourceIncluded(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) + + op := createConfigMapOp("cm1") p := buildTestPlan([]*Operation{op}, nil) - untouched := &InstallableResourceInfo{ + absent := &InstallableResourceInfo{ ResourceMeta: makeResourceMeta("cm2", "default", gvkConfigMap), MustInstall: ResourceInstallTypeNone, } - reporter.startStage( - p, - map[string]string{op.ID(): "default"}, - []*InstallableResourceInfo{untouched}, - map[string]string{untouched.ID(): "default"}, - ) + startTestPlan(reporter, p, []*InstallableResourceInfo{absent}, StartPlanOptions{}) - reports := drainChannel(ch) - require.NotEmpty(t, reports) + ops := operationsByID(lastReportOperations(t, ch)) + require.Len(t, ops, 2, "a resource without operations is untouched even if absent from the cluster") + + absentID := OperationID(OperationTypeNoop, OperationVersionNoop, 0, absent.ID()) + require.Contains(t, ops, absentID) + assert.Equal(t, progrep.OperationTypeNoOp, ops[absentID].Type) + assert.Equal(t, progrep.OperationStatusCompleted, ops[absentID].Status) +} - activeOps := reports[len(reports)-1].StageReports[0].Operations - require.Len(t, activeOps, 1, "untouched resource absent from cluster must not be emitted") - assert.Equal(t, "cm1", activeOps[0].Name) +func TestAI_StartPlan_UntouchedDeduplicatedAcrossIterations(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) + + op := createConfigMapOp("cm1") + op.Iteration = 1 + p := buildTestPlan([]*Operation{op}, nil) + + iterationZero := makeUntouchedInfo("cm1", "default", gvkConfigMap) + + startTestPlan(reporter, p, []*InstallableResourceInfo{iterationZero}, StartPlanOptions{}) + + ops := lastReportOperations(t, ch) + require.Len(t, ops, 1, "a resource deployed in a later iteration is not untouched") + assert.Equal(t, op.ID(), ops[0].ID) } -func TestAI_StartStage_UntouchedDeduplicatedAgainstPlanOp(t *testing.T) { +func TestAI_StartPlan_UntouchedDeduplicatedAgainstPlanOp(t *testing.T) { ch := make(chan progrep.ProgressReport, 64) reporter := NewLegacyProgressReporter(ch) @@ -699,84 +1121,139 @@ func TestAI_StartStage_UntouchedDeduplicatedAgainstPlanOp(t *testing.T) { untouched := makeUntouchedInfo("dep1", "default", gvkDeployment) - reporter.startStage( - p, - map[string]string{op.ID(): "default"}, - []*InstallableResourceInfo{untouched}, - map[string]string{untouched.ID(): "default"}, - ) - - reports := drainChannel(ch) - require.NotEmpty(t, reports) + startTestPlan(reporter, p, []*InstallableResourceInfo{untouched}, StartPlanOptions{}) - activeOps := reports[len(reports)-1].StageReports[0].Operations - require.Len(t, activeOps, 1, "force-tracked untouched resource must appear exactly once via its plan op") - assert.Equal(t, "dep1", activeOps[0].Name) - assert.Equal(t, progrep.OperationStatusPending, activeOps[0].Status) + ops := lastReportOperations(t, ch) + require.Len(t, ops, 1, "force-tracked untouched resource must appear exactly once via its plan op") + assert.Equal(t, op.ID(), ops[0].ID) + assert.Equal(t, progrep.OperationStatusPending, ops[0].Status) } -func TestAI_StartStage_UntouchedInventoryDeduplicated(t *testing.T) { +func TestAI_StartPlan_UntouchedDeduplicatedByObjectRefNotByInfoNamespace(t *testing.T) { ch := make(chan progrep.ProgressReport, 64) reporter := NewLegacyProgressReporter(ch) p := buildTestPlan(nil, nil) - untouched1 := makeUntouchedInfo("cm1", "default", gvkConfigMap) - untouched2 := makeUntouchedInfo("cm1", "default", gvkConfigMap) - - reporter.startStage( - p, - map[string]string{}, - []*InstallableResourceInfo{untouched1, untouched2}, - map[string]string{untouched1.ID(): "default"}, - ) + explicitNS := makeUntouchedInfo("cm1", "default", gvkConfigMap) + defaultedNS := makeUntouchedInfo("cm1", "", gvkConfigMap) - reports := drainChannel(ch) - require.NotEmpty(t, reports) + startTestPlan(reporter, p, []*InstallableResourceInfo{explicitNS, defaultedNS}, StartPlanOptions{}) - activeOps := reports[len(reports)-1].StageReports[0].Operations - require.Len(t, activeOps, 1, "duplicate untouched infos must be emitted once") - assert.Equal(t, "cm1", activeOps[0].Name) + ops := lastReportOperations(t, ch) + require.Len(t, ops, 1, "infos resolving to the same object must be emitted once") + assert.Equal(t, "default", ops[0].Namespace) } -func TestAI_StartStage_UntouchedNamespaceResolution(t *testing.T) { +func TestAI_StartPlan_UntouchedFields(t *testing.T) { ch := make(chan progrep.ProgressReport, 64) reporter := NewLegacyProgressReporter(ch) - mapper := newFakeRESTMapper() - releaseNS := "release-ns" + op := createConfigMapOp("cm1") + p := buildTestPlan([]*Operation{op}, nil) - explicit := makeUntouchedInfo("cm1", "custom-ns", gvkConfigMap) - defaulted := makeUntouchedInfo("cm2", "", gvkConfigMap) - clusterScoped := makeUntouchedInfo("my-ns", "", gvkNamespace) + untouched := makeUntouchedInfo("cm2", "default", gvkConfigMap) + + startTestPlan(reporter, p, []*InstallableResourceInfo{untouched}, StartPlanOptions{}) + + ops := lastReportOperations(t, ch) + require.Len(t, ops, 2) + + untouchedOp := ops[0] + assert.Equal(t, "noop/1/0/default::ConfigMap:cm2", untouchedOp.ID) + assert.Equal(t, OperationID(OperationTypeNoop, OperationVersionNoop, 0, untouched.ID()), untouchedOp.ID) + assert.Equal(t, progrep.OperationCategoryResource, untouchedOp.Category) + assert.Equal(t, progrep.OperationTypeNoOp, untouchedOp.Type) + assert.Equal(t, 0, untouchedOp.Iteration) + assert.Equal(t, progrep.OperationStatusCompleted, untouchedOp.Status) + assert.Equal(t, gvkConfigMap, untouchedOp.GroupVersionKind) + assert.Equal(t, "cm2", untouchedOp.Name) + assert.Equal(t, "default", untouchedOp.Namespace) + assert.NotNil(t, untouchedOp.DependsOn) + assert.Empty(t, untouchedOp.DependsOn) +} + +func TestAI_StartPlan_UntouchedFirstSortedByID(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) - infos := []*InstallableResourceInfo{explicit, defaulted, clusterScoped} + op := createConfigMapOp("cm-a") + p := buildTestPlan([]*Operation{op}, nil) - untouchedResolvedNS := map[string]string{} - for _, info := range infos { - untouchedResolvedNS[info.ID()] = resolveNamespace(info.GroupVersionKind, info.Namespace, releaseNS, mapper) + untouched := []*InstallableResourceInfo{ + makeUntouchedInfo("cm-z", "default", gvkConfigMap), + makeUntouchedInfo("cm-m", "default", gvkConfigMap), } + startTestPlan(reporter, p, untouched, StartPlanOptions{}) + + ops := lastReportOperations(t, ch) + assert.Equal(t, []string{ + "noop/1/0/default::ConfigMap:cm-m", + "noop/1/0/default::ConfigMap:cm-z", + op.ID(), + }, operationIDs(ops)) +} + +func TestAI_StartPlan_UntouchedIgnoredInLaterPlans(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) + + untouched := []*InstallableResourceInfo{makeUntouchedInfo("cm1", "default", gvkConfigMap)} + + startTestPlan(reporter, buildTestPlan(nil, nil), untouched, StartPlanOptions{}) + + op := createConfigMapOp("cm2") + startTestPlan(reporter, buildTestPlan([]*Operation{op}, nil), untouched, StartPlanOptions{}) + + ops := lastReportOperations(t, ch) + assert.Equal(t, []string{ + "noop/1/0/default::ConfigMap:cm1", + "2/" + op.ID(), + }, operationIDs(ops), "only the first plan contributes untouched resources") +} + +func TestAI_StartPlan_UntouchedInventoryDeduplicated(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) + p := buildTestPlan(nil, nil) - reporter.startStage(p, map[string]string{}, infos, untouchedResolvedNS) - reports := drainChannel(ch) - require.NotEmpty(t, reports) + untouched1 := makeUntouchedInfo("cm1", "default", gvkConfigMap) + untouched2 := makeUntouchedInfo("cm1", "default", gvkConfigMap) - activeOps := reports[len(reports)-1].StageReports[0].Operations - require.Len(t, activeOps, 3) + startTestPlan(reporter, p, []*InstallableResourceInfo{untouched1, untouched2}, StartPlanOptions{}) - namespaces := map[string]string{} - for _, o := range activeOps { - namespaces[o.Name] = o.Namespace - } + ops := lastReportOperations(t, ch) + require.Len(t, ops, 1, "duplicate untouched infos must be emitted once") + assert.Equal(t, "cm1", ops[0].Name) +} - assert.Equal(t, "custom-ns", namespaces["cm1"]) - assert.Equal(t, releaseNS, namespaces["cm2"]) - assert.Empty(t, namespaces["my-ns"]) +func TestAI_StartPlan_UntouchedIterationsGetDistinctIDs(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) + + gvkWebhookV1 := schema.GroupVersionKind{Group: "admissionregistration.k8s.io", Version: "v1", Kind: "MutatingWebhookConfiguration"} + gvkWebhookV1beta1 := schema.GroupVersionKind{Group: "admissionregistration.k8s.io", Version: "v1beta1", Kind: "MutatingWebhookConfiguration"} + + first := makeUntouchedInfo("hook", "", gvkWebhookV1) + second := makeUntouchedInfo("hook", "", gvkWebhookV1beta1) + second.Iteration = 1 + + startTestPlan(reporter, buildTestPlan(nil, nil), []*InstallableResourceInfo{first, second}, StartPlanOptions{}) + + ops := lastReportOperations(t, ch) + require.Len(t, ops, 2, "same-named resources of different API versions are distinct objects") + + assert.Equal(t, []string{ + "noop/1/0/:admissionregistration.k8s.io:MutatingWebhookConfiguration:hook", + "noop/1/1/:admissionregistration.k8s.io:MutatingWebhookConfiguration:hook", + }, operationIDs(ops)) + assert.Equal(t, 0, ops[0].Iteration) + assert.Equal(t, 1, ops[1].Iteration) } -func TestAI_StartStage_UntouchedReemittedAcrossStages(t *testing.T) { +func TestAI_StartPlan_UntouchedKeptWhenLaterPlanTouchesResource(t *testing.T) { ch := make(chan progrep.ProgressReport, 64) reporter := NewLegacyProgressReporter(ch) @@ -784,176 +1261,134 @@ func TestAI_StartStage_UntouchedReemittedAcrossStages(t *testing.T) { makeUntouchedInfo("cm-untouched", "default", gvkConfigMap), makeUntouchedInfo("svc-shared", "default", gvkService), } - untouchedNamespaces := map[string]string{ - untouched[0].ID(): "default", - untouched[1].ID(): "default", - } - mainOp := &Operation{ - Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, - Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm-main", "default", gvkConfigMap)}, - } - mainPlan := buildTestPlan([]*Operation{mainOp}, nil) - - reporter.startStage( - mainPlan, - map[string]string{mainOp.ID(): "default"}, - untouched, - untouchedNamespaces, - ) + mainOp := createConfigMapOp("cm-main") + startTestPlan(reporter, buildTestPlan([]*Operation{mainOp}, nil), untouched, StartPlanOptions{}) drainChannel(ch) failureOp := &Operation{ Type: OperationTypeDelete, Version: OperationVersionDelete, Category: OperationCategoryResource, Config: &OperationConfigDelete{ResourceMeta: makeResourceMeta("svc-shared", "default", gvkService)}, } - failurePlan := buildTestPlan([]*Operation{failureOp}, nil) + startTestPlan(reporter, buildTestPlan([]*Operation{failureOp}, nil), untouched, StartPlanOptions{}) + + ops := lastReportOperations(t, ch) + require.Equal(t, []string{ + "noop/1/0/default::ConfigMap:cm-untouched", + "noop/1/0/default::Service:svc-shared", + mainOp.ID(), + "2/" + failureOp.ID(), + }, operationIDs(ops)) + + assert.Equal(t, progrep.OperationStatusCompleted, ops[1].Status, "the untouched entry of the first plan is a fact of history and stays") + assert.Equal(t, progrep.OperationTypeNoOp, ops[1].Type) + assert.Equal(t, progrep.OperationStatusPending, ops[3].Status) + assert.Equal(t, progrep.OperationTypeDelete, ops[3].Type) +} - reporter.startStage( - failurePlan, - map[string]string{failureOp.ID(): "default"}, - untouched, - untouchedNamespaces, - ) +func TestAI_StartPlan_UntouchedNamespaceResolution(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) - reports := drainChannel(ch) - require.NotEmpty(t, reports) + explicit := makeUntouchedInfo("cm1", "custom-ns", gvkConfigMap) + defaulted := makeUntouchedInfo("cm2", "", gvkConfigMap) + clusterScoped := makeUntouchedInfo("my-ns", "", gvkNamespace) - last := reports[len(reports)-1] - require.Len(t, last.StageReports, 2) + reporter.StartPlan(buildTestPlan(nil, nil), "release-ns", []*InstallableResourceInfo{explicit, defaulted, clusterScoped}, newFakeRESTMapper(), StartPlanOptions{}) - frozen := map[string]progrep.Operation{} - for _, o := range last.StageReports[0].Operations { - frozen[o.Name] = o - } - assert.Contains(t, frozen, "cm-untouched", "untouched entry must be retained in the frozen prior stage") - assert.Equal(t, progrep.OperationStatusCompleted, frozen["cm-untouched"].Status) - assert.Equal(t, progrep.OperationTypeNoOp, frozen["cm-untouched"].Type) - - active := last.StageReports[1].Operations - activeByName := map[string]progrep.Operation{} - for _, o := range active { - activeByName[o.Name] = o - } - - require.Contains(t, activeByName, "cm-untouched", "untouched entry must be re-emitted into the new active stage") - assert.Equal(t, progrep.OperationStatusCompleted, activeByName["cm-untouched"].Status) - assert.Equal(t, progrep.OperationTypeNoOp, activeByName["cm-untouched"].Type) + ops := lastReportOperations(t, ch) + require.Len(t, ops, 3) - sharedCount := 0 - for _, o := range active { - if o.Name == "svc-shared" { - sharedCount++ - } + namespaces := map[string]string{} + for _, o := range ops { + namespaces[o.Name] = o.Namespace } - assert.Equal(t, 1, sharedCount, "untouched resource matching a plan operation must appear exactly once") - assert.Equal(t, progrep.OperationTypeDelete, activeByName["svc-shared"].Type, "the shared resource must be represented by its plan operation, not the NoOp untouched entry") - assert.Len(t, active, 2, "active stage must contain the plan op plus the non-duplicated untouched entry") + + assert.Equal(t, "custom-ns", namespaces["cm1"]) + assert.Equal(t, "release-ns", namespaces["cm2"]) + assert.Empty(t, namespaces["my-ns"]) } -func TestAI_StartStage_UntouchedResourceCompletedFromFirstSnapshot(t *testing.T) { +func TestAI_StartPlan_UntouchedNotAddressableByReportStatus(t *testing.T) { ch := make(chan progrep.ProgressReport, 64) reporter := NewLegacyProgressReporter(ch) - op := &Operation{ - Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, - Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm1", "default", gvkConfigMap)}, - } + op := createConfigMapOp("cm1") p := buildTestPlan([]*Operation{op}, nil) untouched := makeUntouchedInfo("cm2", "default", gvkConfigMap) - reporter.startStage( - p, - map[string]string{op.ID(): "default"}, - []*InstallableResourceInfo{untouched}, - map[string]string{untouched.ID(): "default"}, - ) - - reports := drainChannel(ch) - require.NotEmpty(t, reports) + startTestPlan(reporter, p, []*InstallableResourceInfo{untouched}, StartPlanOptions{}) + drainChannel(ch) - activeOps := reports[len(reports)-1].StageReports[0].Operations - require.Len(t, activeOps, 2) + reporter.ReportStatus(OperationID(OperationTypeNoop, OperationVersionNoop, 0, untouched.ID()), progrep.OperationStatusFailed) + assert.Empty(t, drainChannel(ch), "untouched entry ID must not be addressable by ReportStatus") - var untouchedOp *progrep.Operation - for i := range activeOps { - if activeOps[i].Name == "cm2" { - untouchedOp = &activeOps[i] - } - } + reporter.ReportStatus(op.ID(), progrep.OperationStatusCompleted) - require.NotNil(t, untouchedOp, "untouched resource must appear in stage report") - assert.Equal(t, progrep.OperationStatusCompleted, untouchedOp.Status) - assert.Equal(t, progrep.OperationTypeNoOp, untouchedOp.Type) - assert.Equal(t, gvkConfigMap, untouchedOp.GroupVersionKind) - assert.Equal(t, "default", untouchedOp.Namespace) - assert.Empty(t, untouchedOp.WaitingFor) + ops := operationsByID(lastReportOperations(t, ch)) + require.Len(t, ops, 2) + assert.Equal(t, progrep.OperationStatusCompleted, ops[op.ID()].Status) + assert.Equal(t, progrep.OperationStatusCompleted, ops["noop/1/0/default::ConfigMap:cm2"].Status) } -func TestAI_StartStage_UntouchedScopedToStageAndFrozen(t *testing.T) { +func TestAI_StartPlan_UntouchedResourcesOnly(t *testing.T) { ch := make(chan progrep.ProgressReport, 64) reporter := NewLegacyProgressReporter(ch) - op1 := &Operation{ - Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, - Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm1", "default", gvkConfigMap)}, + start := stageMetaOp("stage/install/start") + end := stageMetaOp("stage/install/end") + rel := &Operation{ + Type: OperationTypeDeleteRelease, Version: OperationVersionDeleteRelease, Category: OperationCategoryRelease, + Config: &OperationConfigDeleteRelease{ReleaseName: "rel", ReleaseNamespace: "default", ReleaseRevision: 1}, } - p1 := buildTestPlan([]*Operation{op1}, nil) + p := buildTestPlan([]*Operation{start, end, rel}, map[int][]int{1: {0}, 2: {1}}) - untouched := makeUntouchedInfo("cm2", "default", gvkConfigMap) + untouched := []*InstallableResourceInfo{ + makeUntouchedInfo("cm1", "default", gvkConfigMap), + makeUntouchedInfo("cm2", "default", gvkConfigMap), + } - reporter.startStage( - p1, - map[string]string{op1.ID(): "default"}, - []*InstallableResourceInfo{untouched}, - map[string]string{untouched.ID(): "default"}, - ) - drainChannel(ch) + startTestPlan(reporter, p, untouched, StartPlanOptions{UntouchedResourcesOnly: true}) - op2 := &Operation{ - Type: OperationTypeDelete, Version: OperationVersionDelete, Category: OperationCategoryResource, - Config: &OperationConfigDelete{ResourceMeta: makeResourceMeta("svc1", "default", gvkService)}, + ops := lastReportOperations(t, ch) + require.Len(t, ops, 2, "a plan that is not executed contributes no operations of its own") + + for _, op := range ops { + assert.Equal(t, progrep.OperationTypeNoOp, op.Type) + assert.Equal(t, progrep.OperationStatusCompleted, op.Status) } - p2 := buildTestPlan([]*Operation{op2}, nil) - reporter.startStage(p2, map[string]string{op2.ID(): "default"}, nil, nil) + reporter.ReportStatus(start.ID(), progrep.OperationStatusCompleted) + assert.Empty(t, drainChannel(ch), "omitted plan operations must not be addressable") +} - reports := drainChannel(ch) - require.NotEmpty(t, reports) +func TestAI_Stop_CancelsPendingOperations(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) - last := reports[len(reports)-1] - require.Len(t, last.StageReports, 2) + failed := createConfigMapOp("cm-failed") + neverStarted := createConfigMapOp("cm-never") + startTestPlan(reporter, buildTestPlan([]*Operation{failed, neverStarted}, map[int][]int{1: {0}}), nil, StartPlanOptions{}) - frozenNames := map[string]bool{} - for _, o := range last.StageReports[0].Operations { - frozenNames[o.Name] = true - } + reporter.ReportStatus(failed.ID(), progrep.OperationStatusFailed) + drainChannel(ch) - assert.True(t, frozenNames["cm2"], "untouched entry must be retained in the frozen prior stage") + reporter.Stop(context.Background()) - activeNames := map[string]bool{} - for _, o := range last.StageReports[1].Operations { - activeNames[o.Name] = true - } + reports := drainChannel(ch) + require.Len(t, reports, 1) - assert.False(t, activeNames["cm2"], "untouched entry must not leak into the new active stage") - assert.True(t, activeNames["svc1"]) - assert.Len(t, last.StageReports[1].Operations, 1) + ops := operationsByID(reports[0].Operations) + assert.Equal(t, progrep.OperationStatusFailed, ops[failed.ID()].Status) + assert.Equal(t, progrep.OperationStatusCanceled, ops[neverStarted.ID()].Status, "the final report leaves nothing Pending") } func TestAI_Stop_DoesNotPanicOnClosedChannel(t *testing.T) { ch := make(chan progrep.ProgressReport, 1) reporter := NewLegacyProgressReporter(ch) - ops := []*Operation{ - { - Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, - Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm1", "", gvkConfigMap)}, - }, - } - p := buildTestPlan(ops, nil) - reporter.startStage(p, map[string]string{ops[0].ID(): "default"}, nil, nil) + startTestPlan(reporter, buildTestPlan([]*Operation{createConfigMapOp("cm1")}, nil), nil, StartPlanOptions{}) close(ch) @@ -962,29 +1397,64 @@ func TestAI_Stop_DoesNotPanicOnClosedChannel(t *testing.T) { }) } -func TestAI_Stop_SendsFinalReport(t *testing.T) { +func TestAI_Stop_LeavesCompletedOperationsAlone(t *testing.T) { ch := make(chan progrep.ProgressReport, 64) reporter := NewLegacyProgressReporter(ch) - ops := []*Operation{ - { - Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, - Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm1", "", gvkConfigMap)}, - }, + op := createConfigMapOp("cm1") + untouched := []*InstallableResourceInfo{makeUntouchedInfo("cm2", "default", gvkConfigMap)} + startTestPlan(reporter, buildTestPlan([]*Operation{op}, nil), untouched, StartPlanOptions{}) + reporter.ReportStatus(op.ID(), progrep.OperationStatusCompleted) + drainChannel(ch) + + reporter.Stop(context.Background()) + + reports := drainChannel(ch) + require.Len(t, reports, 1) + + for _, o := range reports[0].Operations { + assert.Equal(t, progrep.OperationStatusCompleted, o.Status) } - p := buildTestPlan(ops, nil) - reporter.startStage(p, map[string]string{ops[0].ID(): "default"}, nil, nil) - reporter.ReportStatus(ops[0].ID(), progrep.OperationStatusCompleted) +} + +func TestAI_Stop_ReportsAllPlans(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) + + op1 := createConfigMapOp("cm1") + startTestPlan(reporter, buildTestPlan([]*Operation{op1}, nil), nil, StartPlanOptions{}) + reporter.ReportStatus(op1.ID(), progrep.OperationStatusFailed) + op2 := createConfigMapOp("cm2") + startTestPlan(reporter, buildTestPlan([]*Operation{op2}, nil), nil, StartPlanOptions{}) + reporter.ReportStatus(op2.ID(), progrep.OperationStatusCompleted) drainChannel(ch) - ctx := context.Background() - reporter.Stop(ctx) + reporter.Stop(context.Background()) + + reports := drainChannel(ch) + require.Len(t, reports, 1) + require.Equal(t, []string{op1.ID(), "2/" + op2.ID()}, operationIDs(reports[0].Operations)) + assert.Equal(t, progrep.OperationStatusFailed, reports[0].Operations[0].Status) + assert.Equal(t, progrep.OperationStatusCompleted, reports[0].Operations[1].Status) +} + +func TestAI_Stop_SendsFinalReport(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) + + op := createConfigMapOp("cm1") + startTestPlan(reporter, buildTestPlan([]*Operation{op}, nil), nil, StartPlanOptions{}) + reporter.ReportStatus(op.ID(), progrep.OperationStatusCompleted) + + drainChannel(ch) + + reporter.Stop(context.Background()) reports := drainChannel(ch) require.Len(t, reports, 1, "Stop should send exactly one final report") - finalOps := reports[0].StageReports[0].Operations + finalOps := reports[0].Operations require.Len(t, finalOps, 1) assert.Equal(t, progrep.OperationStatusCompleted, finalOps[0].Status) } @@ -995,14 +1465,7 @@ func TestAI_Stop_SkipsOnCanceledContext(t *testing.T) { reporter := NewLegacyProgressReporter(ch) - ops := []*Operation{ - { - Type: OperationTypeCreate, Version: OperationVersionCreate, Category: OperationCategoryResource, - Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm1", "", gvkConfigMap)}, - }, - } - p := buildTestPlan(ops, nil) - reporter.startStage(p, map[string]string{ops[0].ID(): "default"}, nil, nil) + startTestPlan(reporter, buildTestPlan([]*Operation{createConfigMapOp("cm1")}, nil), nil, StartPlanOptions{}) ctx, cancel := context.WithCancel(context.Background()) cancel() @@ -1011,16 +1474,3 @@ func TestAI_Stop_SkipsOnCanceledContext(t *testing.T) { assert.Len(t, ch, 1) } - -func makeUntouchedInfo(name, namespace string, gvk schema.GroupVersionKind) *InstallableResourceInfo { - obj := &unstructured.Unstructured{} - obj.SetGroupVersionKind(gvk) - obj.SetName(name) - obj.SetNamespace(namespace) - - return &InstallableResourceInfo{ - ResourceMeta: makeResourceMeta(name, namespace, gvk), - MustInstall: ResourceInstallTypeNone, - GetResult: obj, - } -} diff --git a/pkg/plan/plan_execute.go b/pkg/plan/plan_execute.go index e9d598d5..422abbd7 100644 --- a/pkg/plan/plan_execute.go +++ b/pkg/plan/plan_execute.go @@ -29,9 +29,6 @@ type ExecutePlanOptions struct { InstallableResourceInfos []*InstallableResourceInfo LegacyProgressReporter *LegacyProgressReporter NetworkParallelism int - // NoUntouchedResourcesReport, when true, omits untouched resources from the progress - // report for this plan. Set it for delta plans, like a failure plan. - NoUntouchedResourcesReport bool } // Executes the given plan. It doesn't care what kind of plan it is (install, upgrade, failure plan, @@ -45,9 +42,7 @@ func ExecutePlan(parentCtx context.Context, releaseNamespace string, plan *Plan, opts.NetworkParallelism = lo.Max([]int{opts.NetworkParallelism, 1}) if opts.LegacyProgressReporter != nil { - opts.LegacyProgressReporter.StartStage(plan, releaseNamespace, opts.InstallableResourceInfos, clientFactory.Mapper(), StartStageOptions{ - NoUntouchedResources: opts.NoUntouchedResourcesReport, - }) + opts.LegacyProgressReporter.StartPlan(plan, releaseNamespace, opts.InstallableResourceInfos, clientFactory.Mapper(), StartPlanOptions{}) } workerPool := pool.New().WithContext(ctx).WithMaxGoroutines(opts.NetworkParallelism).WithCancelOnError().WithFirstError()