feat: expose operation graph in progress report

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))
 	}

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)

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"))
 }

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")

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 {