diff --git a/pkg/action/common.go b/pkg/action/common.go index 52e7ff1b..96eab065 100644 --- a/pkg/action/common.go +++ b/pkg/action/common.go @@ -190,9 +190,10 @@ 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, + LegacyProgressReporter: opts.LegacyProgressReporter, + TrackingOptions: opts.TrackingOptions, + NetworkParallelism: opts.NetworkParallelism, + InstallableResourceInfos: installableInfos, }); 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 fc542bb4..a0a07cdf 100644 --- a/pkg/action/release_install.go +++ b/pkg/action/release_install.go @@ -541,6 +541,13 @@ func releaseInstall(ctx context.Context, ctxCancelFn context.CancelCauseFunc, re printNotes(ctx, newRelease.Info.Notes) } + if opts.LegacyProgressReportCh != nil { + reporter := plan.NewLegacyProgressReporter(opts.LegacyProgressReportCh) + reporter.StartStage(installPlan, releaseNamespace, instResInfos, clientFactory.Mapper()) + reporter.Stop(ctx) + close(opts.LegacyProgressReportCh) + } + log.Default.Info(ctx, color.Style{color.Bold, color.Green}.Render(fmt.Sprintf("Skipped release %q (namespace: %q): cluster resources already as desired", releaseName, releaseNamespace))) return nil @@ -588,9 +595,10 @@ func releaseInstall(ctx context.Context, ctxCancelFn context.CancelCauseFunc, re log.Default.Debug(ctx, "Execute release install plan") executePlanErr := plan.ExecutePlan(ctx, releaseNamespace, installPlan, taskStore, logStore, informerFactory, history, clientFactory, plan.ExecutePlanOptions{ - LegacyProgressReporter: reporter, - TrackingOptions: opts.TrackingOptions, - NetworkParallelism: opts.NetworkParallelism, + LegacyProgressReporter: reporter, + TrackingOptions: opts.TrackingOptions, + NetworkParallelism: opts.NetworkParallelism, + InstallableResourceInfos: instResInfos, }) if executePlanErr != nil { criticalErrs.Add(fmt.Errorf("execute release install plan: %w", executePlanErr)) @@ -967,6 +975,10 @@ func runRollbackPlan(ctx context.Context, releaseName, releaseNamespace string, }) if releaseIsUpToDate && planIsUseless { + if opts.LegacyProgressReporter != nil { + opts.LegacyProgressReporter.StartStage(rollbackPlan, releaseNamespace, instResInfos, clientFactory.Mapper()) + } + 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 @@ -975,9 +987,10 @@ 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, + LegacyProgressReporter: opts.LegacyProgressReporter, + TrackingOptions: opts.TrackingOptions, + NetworkParallelism: opts.NetworkParallelism, + InstallableResourceInfos: instResInfos, }) if executePlanErr != nil { critErrs.Add(fmt.Errorf("execute rollback plan: %w", executePlanErr)) diff --git a/pkg/action/release_rollback.go b/pkg/action/release_rollback.go index ee4dd9b5..ec1b4fe0 100644 --- a/pkg/action/release_rollback.go +++ b/pkg/action/release_rollback.go @@ -415,8 +415,9 @@ func releaseRollback(ctx context.Context, ctxCancelFn context.CancelCauseFunc, r log.Default.Debug(ctx, "Execute release install plan") executePlanErr := plan.ExecutePlan(ctx, releaseNamespace, installPlan, taskStore, logStore, informerFactory, history, clientFactory, plan.ExecutePlanOptions{ - TrackingOptions: opts.TrackingOptions, - NetworkParallelism: opts.NetworkParallelism, + TrackingOptions: opts.TrackingOptions, + NetworkParallelism: opts.NetworkParallelism, + InstallableResourceInfos: instResInfos, }) if executePlanErr != nil { criticalErrs.Add(fmt.Errorf("execute release install plan: %w", executePlanErr)) diff --git a/pkg/legacy/progrep/progress_report.go b/pkg/legacy/progrep/progress_report.go index e55477fb..4fdd38ce 100644 --- a/pkg/legacy/progrep/progress_report.go +++ b/pkg/legacy/progrep/progress_report.go @@ -13,6 +13,7 @@ const ( OperationTypeDelete OperationType = "Delete" OperationTypeApply OperationType = "Apply" OperationTypeRecreate OperationType = "Recreate" + OperationTypeNoOp OperationType = "NoOp" OperationTypeTrackReadiness OperationType = "TrackReadiness" OperationTypeTrackPresence OperationType = "TrackPresence" OperationTypeTrackAbsence OperationType = "TrackAbsence" @@ -28,8 +29,9 @@ 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). +// StageReport contains ALL operations in the plan, plus untouched resources reported as +// NoOp with status Completed -- from the very first report, every operation is present +// (initially as Pending). type StageReport struct { Operations []Operation `json:"operations"` } diff --git a/pkg/plan/legacy_progress_reporter.go b/pkg/plan/legacy_progress_reporter.go index 5ab2a878..f1cff30e 100644 --- a/pkg/plan/legacy_progress_reporter.go +++ b/pkg/plan/legacy_progress_reporter.go @@ -5,6 +5,7 @@ import ( "fmt" "github.com/samber/lo" + "k8s.io/apimachinery/pkg/api/meta" kdutil "github.com/werf/kubedog/pkg/trackers/dyntracker/util" "github.com/werf/nelm/pkg/legacy/progrep" @@ -42,6 +43,17 @@ func (r *LegacyProgressReporter) ReportStatus(opID string, status progrep.Operat }) } +func (r *LegacyProgressReporter) StartStage(p *Plan, releaseNamespace string, installableResourceInfos []*InstallableResourceInfo, mapper meta.RESTMapper) { + resolvedNamespaces := buildResolvedNamespaces(p, releaseNamespace, mapper) + + 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) +} + func (r *LegacyProgressReporter) Stop(ctx context.Context) { var report progrep.ProgressReport @@ -59,7 +71,7 @@ func (r *LegacyProgressReporter) Stop(ctx context.Context) { }() } -func (r *LegacyProgressReporter) startStage(p *Plan, resolvedNamespaces map[string]string) { +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)) @@ -71,6 +83,7 @@ func (r *LegacyProgressReporter) startStage(p *Plan, resolvedNamespaces map[stri var entries []opEntry entryIndex := make(map[string]int) + seenRefs := make(map[progrep.ObjectRef]struct{}) for _, op := range ops { if op.Category != OperationCategoryResource && op.Category != OperationCategoryTrack { @@ -81,6 +94,7 @@ func (r *LegacyProgressReporter) startStage(p *Plan, resolvedNamespaces map[stri typ := mapOperationType(op.Type) idx := len(entries) entryIndex[op.ID()] = idx + seenRefs[ref] = struct{}{} entries = append(entries, opEntry{ iteration: int(op.Iteration), @@ -90,6 +104,33 @@ func (r *LegacyProgressReporter) startStage(p *Plan, resolvedNamespaces map[stri }) } + for _, info := range untouched { + if info.GetResult == nil { + continue + } + + ref := progrep.ObjectRef{ + GroupVersionKind: info.GroupVersionKind, + Name: info.Name, + Namespace: untouchedResolvedNamespaces[info.ID()], + } + + if _, ok := seenRefs[ref]; ok { + continue + } + + seenRefs[ref] = struct{}{} + + 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 _, op := range ops { idx, ok := entryIndex[op.ID()] if !ok { diff --git a/pkg/plan/legacy_progress_reporter_ai_test.go b/pkg/plan/legacy_progress_reporter_ai_test.go index 0d8fc2a8..0f0f8956 100644 --- a/pkg/plan/legacy_progress_reporter_ai_test.go +++ b/pkg/plan/legacy_progress_reporter_ai_test.go @@ -9,6 +9,7 @@ import ( "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/legacy/progrep" @@ -243,7 +244,7 @@ func TestAI_ReportOperationStatus_SetsStatusAndReports(t *testing.T) { Config: &OperationConfigCreate{ResourceSpec: makeResourceSpec("cm1", "", gvkConfigMap)}, } p := buildTestPlan([]*Operation{op}, nil) - reporter.startStage(p, map[string]string{op.ID(): "default"}) + reporter.startStage(p, map[string]string{op.ID(): "default"}, nil, nil) drainChannel(ch) reportOperationStatus(op, OperationStatusPending, reporter) @@ -279,7 +280,7 @@ func TestAI_ReportStatus_DoesNotPanicOnClosedChannel(t *testing.T) { }, } p := buildTestPlan(ops, nil) - reporter.startStage(p, map[string]string{ops[0].ID(): "default"}) + reporter.startStage(p, map[string]string{ops[0].ID(): "default"}, nil, nil) close(ch) @@ -309,7 +310,7 @@ func TestAI_ReportStatus_SendsSnapshot(t *testing.T) { ops[1].ID(): "default", } - reporter.startStage(p, resolvedNS) + reporter.startStage(p, resolvedNS, nil, nil) drainChannel(ch) reporter.ReportStatus(ops[0].ID(), progrep.OperationStatusCompleted) @@ -344,7 +345,7 @@ func TestAI_ReportStatus_UnknownOpIDIsIgnored(t *testing.T) { } p := buildTestPlan(ops, nil) - reporter.startStage(p, map[string]string{ops[0].ID(): "default"}) + reporter.startStage(p, map[string]string{ops[0].ID(): "default"}, nil, nil) drainChannel(ch) reporter.ReportStatus("nonexistent/op/id", progrep.OperationStatusCompleted) @@ -372,7 +373,7 @@ func TestAI_ReportStatus_WaitingForPopulation(t *testing.T) { opA.ID(): "default", opB.ID(): "default", } - reporter.startStage(p, resolvedNS) + reporter.startStage(p, resolvedNS, nil, nil) reports := drainChannel(ch) require.NotEmpty(t, reports) @@ -514,7 +515,7 @@ func TestAI_StartStage_FiltersNonResourceOps(t *testing.T) { }, } p := buildTestPlan(ops, nil) - reporter.startStage(p, map[string]string{ops[0].ID(): "default"}) + reporter.startStage(p, map[string]string{ops[0].ID(): "default"}, nil, nil) reports := drainChannel(ch) require.NotEmpty(t, reports) @@ -537,7 +538,7 @@ func TestAI_StartStage_FreezesPreviousStage(t *testing.T) { }, } p1 := buildTestPlan(ops1, nil) - reporter.startStage(p1, map[string]string{ops1[0].ID(): "default"}) + reporter.startStage(p1, map[string]string{ops1[0].ID(): "default"}, nil, nil) reporter.ReportStatus(ops1[0].ID(), progrep.OperationStatusCompleted) drainChannel(ch) @@ -549,7 +550,7 @@ func TestAI_StartStage_FreezesPreviousStage(t *testing.T) { }, } p2 := buildTestPlan(ops2, nil) - reporter.startStage(p2, map[string]string{ops2[0].ID(): "default"}) + reporter.startStage(p2, map[string]string{ops2[0].ID(): "default"}, nil, nil) reports := drainChannel(ch) require.NotEmpty(t, reports) @@ -568,6 +569,333 @@ func TestAI_StartStage_FreezesPreviousStage(t *testing.T) { assert.Equal(t, progrep.OperationStatusPending, activeOps[0].Status) } +func TestAI_StartStage_ReportStatusNeverAffectsUntouchedEntry(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)}, + } + 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"}, + ) + drainChannel(ch) + + reporter.ReportStatus(untouched.ID(), progrep.OperationStatusFailed) + + noReports := drainChannel(ch) + assert.Empty(t, noReports, "untouched entry ID must not be addressable by ReportStatus") + + reporter.ReportStatus(op.ID(), progrep.OperationStatusCompleted) + + reports := drainChannel(ch) + require.NotEmpty(t, reports) + + activeOps := reports[len(reports)-1].StageReports[0].Operations + require.Len(t, activeOps, 2) + + statuses := map[string]progrep.OperationStatus{} + for _, o := range activeOps { + statuses[o.Name] = o.Status + } + + assert.Equal(t, progrep.OperationStatusCompleted, statuses["cm1"]) + assert.Equal(t, progrep.OperationStatusCompleted, statuses["cm2"], "untouched entry must remain Completed") +} + +func TestAI_StartStage_UntouchedAbsentResourceOmitted(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)}, + } + p := buildTestPlan([]*Operation{op}, nil) + + untouched := &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"}, + ) + + reports := drainChannel(ch) + require.NotEmpty(t, reports) + + 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_StartStage_UntouchedDeduplicatedAgainstPlanOp(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) + + op := &Operation{ + Type: OperationTypeTrackReadiness, Version: OperationVersionTrackReadiness, Category: OperationCategoryTrack, + Config: &OperationConfigTrackReadiness{ResourceMeta: makeResourceMeta("dep1", "default", gvkDeployment)}, + } + p := buildTestPlan([]*Operation{op}, nil) + + 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) + + 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) +} + +func TestAI_StartStage_UntouchedInventoryDeduplicated(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"}, + ) + + reports := drainChannel(ch) + require.NotEmpty(t, reports) + + 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) +} + +func TestAI_StartStage_UntouchedNamespaceResolution(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) + + mapper := newFakeRESTMapper() + releaseNS := "release-ns" + + explicit := makeUntouchedInfo("cm1", "custom-ns", gvkConfigMap) + defaulted := makeUntouchedInfo("cm2", "", gvkConfigMap) + clusterScoped := makeUntouchedInfo("my-ns", "", gvkNamespace) + + infos := []*InstallableResourceInfo{explicit, defaulted, clusterScoped} + + untouchedResolvedNS := map[string]string{} + for _, info := range infos { + untouchedResolvedNS[info.ID()] = resolveNamespace(info.GroupVersionKind, info.Namespace, releaseNS, mapper) + } + + p := buildTestPlan(nil, nil) + reporter.startStage(p, map[string]string{}, infos, untouchedResolvedNS) + + reports := drainChannel(ch) + require.NotEmpty(t, reports) + + activeOps := reports[len(reports)-1].StageReports[0].Operations + require.Len(t, activeOps, 3) + + namespaces := map[string]string{} + for _, o := range activeOps { + namespaces[o.Name] = o.Namespace + } + + assert.Equal(t, "custom-ns", namespaces["cm1"]) + assert.Equal(t, releaseNS, namespaces["cm2"]) + assert.Empty(t, namespaces["my-ns"]) +} + +func TestAI_StartStage_UntouchedReemittedAcrossStages(t *testing.T) { + ch := make(chan progrep.ProgressReport, 64) + reporter := NewLegacyProgressReporter(ch) + + untouched := []*InstallableResourceInfo{ + 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, + ) + drainChannel(ch) + + failureOp := &Operation{ + Type: OperationTypeDelete, Version: OperationVersionDelete, Category: OperationCategoryResource, + Config: &OperationConfigDelete{ResourceMeta: makeResourceMeta("svc-shared", "default", gvkService)}, + } + failurePlan := buildTestPlan([]*Operation{failureOp}, nil) + + reporter.startStage( + failurePlan, + map[string]string{failureOp.ID(): "default"}, + untouched, + untouchedNamespaces, + ) + + reports := drainChannel(ch) + require.NotEmpty(t, reports) + + last := reports[len(reports)-1] + require.Len(t, last.StageReports, 2) + + 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) + + sharedCount := 0 + for _, o := range active { + if o.Name == "svc-shared" { + sharedCount++ + } + } + 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") +} + +func TestAI_StartStage_UntouchedResourceCompletedFromFirstSnapshot(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)}, + } + 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) + + activeOps := reports[len(reports)-1].StageReports[0].Operations + require.Len(t, activeOps, 2) + + var untouchedOp *progrep.Operation + for i := range activeOps { + if activeOps[i].Name == "cm2" { + untouchedOp = &activeOps[i] + } + } + + 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) +} + +func TestAI_StartStage_UntouchedScopedToStageAndFrozen(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)}, + } + p1 := buildTestPlan([]*Operation{op1}, nil) + + untouched := makeUntouchedInfo("cm2", "default", gvkConfigMap) + + reporter.startStage( + p1, + map[string]string{op1.ID(): "default"}, + []*InstallableResourceInfo{untouched}, + map[string]string{untouched.ID(): "default"}, + ) + drainChannel(ch) + + op2 := &Operation{ + Type: OperationTypeDelete, Version: OperationVersionDelete, Category: OperationCategoryResource, + Config: &OperationConfigDelete{ResourceMeta: makeResourceMeta("svc1", "default", gvkService)}, + } + p2 := buildTestPlan([]*Operation{op2}, nil) + + reporter.startStage(p2, map[string]string{op2.ID(): "default"}, nil, nil) + + reports := drainChannel(ch) + require.NotEmpty(t, reports) + + last := reports[len(reports)-1] + require.Len(t, last.StageReports, 2) + + frozenNames := map[string]bool{} + for _, o := range last.StageReports[0].Operations { + frozenNames[o.Name] = true + } + + assert.True(t, frozenNames["cm2"], "untouched entry must be retained in the frozen prior stage") + + activeNames := map[string]bool{} + for _, o := range last.StageReports[1].Operations { + activeNames[o.Name] = true + } + + 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) +} + func TestAI_Stop_DoesNotPanicOnClosedChannel(t *testing.T) { ch := make(chan progrep.ProgressReport, 1) reporter := NewLegacyProgressReporter(ch) @@ -579,7 +907,7 @@ func TestAI_Stop_DoesNotPanicOnClosedChannel(t *testing.T) { }, } p := buildTestPlan(ops, nil) - reporter.startStage(p, map[string]string{ops[0].ID(): "default"}) + reporter.startStage(p, map[string]string{ops[0].ID(): "default"}, nil, nil) close(ch) @@ -599,7 +927,7 @@ func TestAI_Stop_SendsFinalReport(t *testing.T) { }, } p := buildTestPlan(ops, nil) - reporter.startStage(p, map[string]string{ops[0].ID(): "default"}) + reporter.startStage(p, map[string]string{ops[0].ID(): "default"}, nil, nil) reporter.ReportStatus(ops[0].ID(), progrep.OperationStatusCompleted) drainChannel(ch) @@ -628,7 +956,7 @@ func TestAI_Stop_SkipsOnCanceledContext(t *testing.T) { }, } p := buildTestPlan(ops, nil) - reporter.startStage(p, map[string]string{ops[0].ID(): "default"}) + reporter.startStage(p, map[string]string{ops[0].ID(): "default"}, nil, nil) ctx, cancel := context.WithCancel(context.Background()) cancel() @@ -637,3 +965,16 @@ 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 7ad340d7..d7d149c0 100644 --- a/pkg/plan/plan_execute.go +++ b/pkg/plan/plan_execute.go @@ -25,8 +25,9 @@ import ( type ExecutePlanOptions struct { common.TrackingOptions - LegacyProgressReporter *LegacyProgressReporter - NetworkParallelism int + InstallableResourceInfos []*InstallableResourceInfo + LegacyProgressReporter *LegacyProgressReporter + NetworkParallelism int } // Executes the given plan. It doesn't care what kind of plan it is (install, upgrade, failure plan, @@ -40,8 +41,7 @@ func ExecutePlan(parentCtx context.Context, releaseNamespace string, plan *Plan, opts.NetworkParallelism = lo.Max([]int{opts.NetworkParallelism, 1}) if opts.LegacyProgressReporter != nil { - resolvedNS := buildResolvedNamespaces(plan, releaseNamespace, clientFactory.Mapper()) - opts.LegacyProgressReporter.startStage(plan, resolvedNS) + opts.LegacyProgressReporter.StartStage(plan, releaseNamespace, opts.InstallableResourceInfos, clientFactory.Mapper()) } workerPool := pool.New().WithContext(ctx).WithMaxGoroutines(opts.NetworkParallelism).WithCancelOnError().WithFirstError()