feat: Add sync phases and waves customizability (#537)

* move `asJSON` to utils and rename to `UnstructuredAsJSON`

* add `SyncWaveAnnotation` for resource ordering

* add `GetSyncWave` function for handling `SyncWaveAnnotation`

* move and integrate the default wave logic into the `GetSyncWave` function for resource ordering

* remove unused `GetLastProgressTimestamp` function and cleanup imports in `common.go`

* remove unused `ServiceID` function and cleanup imports in `common.go`

* add `SyncPhase` constants and `GetSyncPhase` function for handling resource apply phases

* move `const.go` contents to `annotations.go` for better file organization

* integrate wave construction into `Phase` and improve resource synchronization handling

* add skipped resource tracking and integrate skip/delete filters into phase construction

* remove `CacheFilter` from applier initialization in service reconciler

* add `WithFilter` to applier initialization in service reconciler

* commit work in progress

* feat(phase): add support for phased resource application and enhanced health evaluation

* rename `kindToDefaultSyncWave` to `kindSyncPriorities` and replace `lastDefaultWave` with `defaultSyncPriority`

* fix `DeletedCount` to use `deleteWave.Len` instead of the last wave length

* improve phase management by adding `HasWaves` method, ensuring phase initialization, and refactoring `Next` logic for better clarity

* add `sync_phase` tracking in database schema, queries, and data structures

* replace base32 with base64 for hash encoding in `utils` package to optimize output format

* add check to skip insert query execution if no values are provided in `db_store`

* feat(applier): improve logging, enhance resource filtering, and refine phase application logic

* replace `AreResourcesHealthy` with `GetResourceHealth` to improve health status tracking and error handling across store and applier components

* improve logging in `applier` to provide better visibility into phase health and failure handling

* revise `GetResourceHealth` test cases to ensure accurate tracking of pending and failed states

* fix unit tests

* fix unit tests

* add initial version of resource  check function

* add tests for `HasSomeResources` to validate handling of different resource scenarios and edge cases

* feat(annotations): add Helm hook annotations and extend sync phase definitions

* handle Helm hooks

* update `defaultPhase` to accept annotations map directly instead of unstructured type

* implement `HelmFilter` to handle resource application logic based on Helm hook annotations during installs and upgrades

* fix(tests): correct resource references in db_store_test to align with updated test scenarios

* feat(sync-wave): add Helm hook weight support and enhance default sync wave determination based on resource kind

* add service warnings

* add manifest table

* add manifests query

* rename entry to component

* add SaveManifests and GetManifests functions to handle storing and retrieving service-applied manifests

* rename `component` to `manifest` in insert statement and adjust field mappings in query

* add tests for SaveManifests and GetManifests to validate manifest storage and retrieval behavior

* update docs

* add GetPhaseHookDeletePolicy function to retrieve sync phase hook delete policy from annotations

* rename manifest storage to cleanup candidates and update related functions, queries, and tests

* replace direct schema initialization with GroupVersionKind method and refactor associated code

* refactor ToUnstructured logic into Component and update usage in applier to streamline resource conversion

* simplify resource existence check in applier by combining variable declaration and conditional

* simplify resource filtering by inverting conditional check to reduce nesting

* add UID field to cleanup candidate and update related queries and functions to handle unique resource identification

* replace `NewStoreKeyFromEntry` with `ToStoreKey` method in `Component` and `CleanupCandidate`, and refactor related logic for consistency and clarity

* refactor phase deletion logic to filter resources by sync phase directly in NewPhase for improved clarity and alignment with existing patterns

* rename cleanup candidates to processed hook components and update associated methods, queries, and tests for enhanced semantic clarity

* add status field to processed hook component and update queries, functions, and conflict resolution logic to manage and persist component state

* add ExpireProcessedHookComponents method to DatabaseStore and store interface to enable removal of processed hook components by service ID

* call ExpireProcessedHookComponents during service reconciliation to remove processed components and ensure consistency

* refactor SyncPhaseHookDeletePolicy handling to use switch-case for improved clarity and extensibility

* extend processed hook component logic with status-based handling and update related applier logic for resource filtering and deletion

* improve comment clarity in applier logic for resource deletion policies

* remove unnecessary comment in SetAnnotations call for streamlined code readability

* simplify applier logic by consolidating deletion policy checks and optimizing resource filtering and deletion handling

* clarify error log message in component saving logic to improve debugging

* refactor processed hook component saving logic to optimize checks, simplify queries, and enhance error handling

* extend processed hook component saving logic with status-based delete policy checks and update related error handling

* clarify log messages in processed hook saving logic for improved traceability and debugging

* extend `maybeSaveProcessedHookComponent` with state and serviceID parameters to refine delete policy checks and enhance logging accuracy

* refactor delete policy checks by introducing `HasStateDesiredByDeletePolicy` helper function and simplifying conditional logic in `maybeSaveProcessedHookComponent`

* fix import cycle

* refactor `maybeSaveProcessedHookComponent` to accept database connection as a parameter, removing redundant connection handling logic

* introduce `maybeSaveProcessedHookComponents` to handle bulk saving of processed hook components, replacing outdated TODO and improving logic readability

* remove redundant and outdated test cases for resource health and existence checks, replace with streamlined `createHookJob` utility, and refactor `TestComponentCache_ProcessedHookComponents`

* add comprehensive test cases for `GetProcessedHookComponents` to validate behavior with isolation, resource types, UIDs, and large datasets

* fix lint

* update pull request template

* support helm.sh/hook-delete-policy

* bump gql client v1.51.3

* optimize `maybeSaveProcessedHookComponents` by reducing redundant string operations and improving clarity with pre-allocated slice for value assembly

* replace fallthrough with explicit return for SyncPhaseSync in annotations

* replace fallthrough with call to defaultPhase in annotations

* recreate hooks if their manifest change

* fix cyclic dependency

* update comment

* mark service as failed only if it has errors, not warnings

* refactor

* do not sleep after last wave

* refactor

* fix query

* reformat

* fix manifest sha handling

* refactor SaveHookComponentWithManifestSHA signature and update call sites

* fix manifest sha handling

* add required checks

* support multiple phases

* fix multiple phases support

* add tests for phase recognition

* fix tests for phase recognition

* add tests for wave recognition

* fix delete filtering

* update comment

* update comment

* rename methods for consistency

* remove unused NewKeyFromEntry function

* rename methods and variables for consistency

* do not ignore helm hooks by default

* remove duplicated check

* replace `GetPhaseHookDeletePolicy` with `HasSyncPhaseHookDeletePolicy` for clarity and improve readability

* handle multiple delete policies

* support argo sync wave annotation

* skip applying hook resource if their manifest did not change

* reformat

* fix linter

* add continue

* fix multi-phase resource deletion

* fix for trivy

* fix import cycle

* fix import cycle

* improve logging

* fix install/upgrade bug

* fix install/upgrade bug

---------

Co-authored-by: Sebastian Florek <[email protected]>
Co-authored-by: Lukasz Zajaczkowski <[email protected]>

Author: 3 people

.github/PULL_REQUEST_TEMPLATE.md

@@ -24,6 +24,9 @@ Test environment: https://console.your-env.onplural.sh/
     - [ ] Service errors are reported properly and visible in the UI.
     - [ ] Service updates are reflected properly in the cluster.
     - [ ] Service resync triggers immediately and works as expected.
+    - [ ] Sync waves annotations are respected.
+    - [ ] Sync phases annotations are respected. Phases are executed in the correct order.
+    - [ ] Sync hook delete policies are respected. Resources are not recreated once they reach the desired state.
     - [ ] Service deletion works and cleanups resources properly.
     - [ ] Services can be recreated after deletion.
     - [ ] Service detachment works and keeps resources unaffected.

internal/controller/ingressreplica_controller_test.go

@@ -102,7 +102,7 @@ var _ = Describe("IngressReplica Controller", Ordered, func() {
 			err = kClient.Get(ctx, namespacedName, ingressReplica)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(SanitizeStatusConditions(ingressReplica.Status)).To(Equal(SanitizeStatusConditions(v1alpha1.Status{
-				SHA: lo.ToPtr("ACBBWIKK74ACGAK5NWAXYTTIYI2GDOSXGCJ65UGOLOPFCB24PKUQ===="),
+				SHA: lo.ToPtr("AIIbIUr_ACMBXW2BfE5owjRhulcwk-7QzlueUQdceqk"),
 				Conditions: []metav1.Condition{
 					{
 						Type:    v1alpha1.ReadyConditionType.String(),
@@ -132,7 +132,7 @@ var _ = Describe("IngressReplica Controller", Ordered, func() {
 			err = kClient.Get(ctx, namespacedName, ingressReplica)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(SanitizeStatusConditions(ingressReplica.Status)).To(Equal(SanitizeStatusConditions(v1alpha1.Status{
-				SHA: lo.ToPtr("ACBBWIKK74ACGAK5NWAXYTTIYI2GDOSXGCJ65UGOLOPFCB24PKUQ===="),
+				SHA: lo.ToPtr("AIIbIUr_ACMBXW2BfE5owjRhulcwk-7QzlueUQdceqk"),
 				Conditions: []metav1.Condition{
 					{
 						Type:    v1alpha1.ReadyConditionType.String(),

internal/utils/hash.go

@@ -2,20 +2,48 @@ package utils
 
 import (
 	"crypto/sha256"
-	"encoding/base32"
+	"encoding/base64"
 	"encoding/json"
+
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 )
 
-func HashObject(any interface{}) (string, error) {
-	out, err := json.Marshal(any)
+func HashObject(object any) (string, error) {
+	out, err := json.Marshal(object)
 	if err != nil {
 		return "", err
 	}
 	sha := sha256.Sum256(out)
-	return base32.StdEncoding.EncodeToString(sha[:]), nil
+	return base64.RawURLEncoding.EncodeToString(sha[:]), nil
 }
 
-func HashString(s string) string {
-	sha := sha256.Sum256([]byte(s))
-	return base32.StdEncoding.EncodeToString(sha[:])
+// HashResource calculates SHA for an unstructured resource.
+// It uses object metadata (name, namespace, labels, annotations, deletion timestamp)
+// and all other top-level fields except status.
+func HashResource(resource unstructured.Unstructured) (string, error) {
+	resourceCopy := resource.DeepCopy()
+	object := struct {
+		Name              string            `json:"name"`
+		Namespace         string            `json:"namespace"`
+		Labels            map[string]string `json:"labels"`
+		Annotations       map[string]string `json:"annotations"`
+		DeletionTimestamp string            `json:"deletionTimestamp"`
+		Other             map[string]any    `json:"other"`
+	}{
+		Name:        resourceCopy.GetName(),
+		Namespace:   resourceCopy.GetNamespace(),
+		Labels:      resourceCopy.GetLabels(),
+		Annotations: resourceCopy.GetAnnotations(),
+	}
+
+	deletionTimestamp := resourceCopy.GetDeletionTimestamp()
+	if deletionTimestamp != nil {
+		object.DeletionTimestamp = deletionTimestamp.String()
+	}
+
+	unstructured.RemoveNestedField(resourceCopy.Object, "metadata")
+	unstructured.RemoveNestedField(resourceCopy.Object, "status")
+	object.Other = resourceCopy.Object
+
+	return HashObject(object)
 }

internal/utils/kubernetes.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/cli-runtime/pkg/genericclioptions"
 	"k8s.io/client-go/rest"
@@ -17,6 +18,7 @@ import (
 	"k8s.io/kubectl/pkg/cmd/util"
 	ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	"sigs.k8s.io/yaml"
 
 	"github.com/pluralsh/deployment-operator/pkg/flowcontrol"
 
@@ -50,23 +52,6 @@ func TryAddControllerRef(ctx context.Context, client ctrlruntimeclient.Client, o
 	})
 }
 
-func TryToUpdate(ctx context.Context, client ctrlruntimeclient.Client, object ctrlruntimeclient.Object) error {
-	key := ctrlruntimeclient.ObjectKeyFromObject(object)
-
-	return retry.RetryOnConflict(retry.DefaultRetry, func() error {
-		original := object.DeepCopyObject().(ctrlruntimeclient.Object)
-		if err := client.Get(ctx, key, object); err != nil {
-			return fmt.Errorf("could not fetch current %s/%s state, got error: %w", object.GetName(), object.GetNamespace(), err)
-		}
-
-		if reflect.DeepEqual(object, original) {
-			return nil
-		}
-
-		return client.Patch(ctx, original, ctrlruntimeclient.MergeFrom(object))
-	})
-}
-
 func TryAddOwnerRef(ctx context.Context, client ctrlruntimeclient.Client, owner ctrlruntimeclient.Object, object ctrlruntimeclient.Object, scheme *runtime.Scheme) error {
 	key := ctrlruntimeclient.ObjectKeyFromObject(object)
 
@@ -228,3 +213,8 @@ func UnstructuredToConditions(c []interface{}) []metav1.Condition {
 	}
 	return conditions
 }
+
+func UnstructuredAsJSON(resource *unstructured.Unstructured) string {
+	data, _ := yaml.Marshal(resource)
+	return string(data)
+}

pkg/common/common.go

@@ -1,25 +1,19 @@
 package common
 
 import (
-	"context"
 	"math/rand"
 	"time"
 
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
-	ctrclient "sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/yaml"
 
 	"github.com/pluralsh/deployment-operator/cmd/agent/args"
-	"github.com/pluralsh/deployment-operator/internal/utils"
-	smcommon "github.com/pluralsh/deployment-operator/pkg/streamline/common"
 )
 
 const (
-	ManagedByLabel             = "plural.sh/managed-by"
-	AgentLabelValue            = "agent"
-	LastProgressTimeAnnotation = "plural.sh/last-progress-time"
+	ManagedByLabel  = "plural.sh/managed-by"
+	AgentLabelValue = "agent"
 )
 
 func ToUnstructured(obj runtime.Object) (*unstructured.Unstructured, error) {
@@ -53,48 +47,6 @@ func Unmarshal(s string) (map[string]interface{}, error) {
 	return result, nil
 }
 
-func GetLastProgressTimestamp(ctx context.Context, k8sClient ctrclient.Client, obj *unstructured.Unstructured) (progressTime metav1.Time, err error) {
-	progressTime = metav1.Now()
-
-	if obj.GetAnnotations() == nil {
-		obj.SetAnnotations(make(map[string]string))
-	}
-	annotations := obj.GetAnnotations()
-	timeStr, ok := annotations[LastProgressTimeAnnotation]
-
-	defer func() {
-		if !ok {
-			err = utils.TryToUpdate(ctx, k8sClient, obj)
-			if err != nil {
-				return
-			}
-			key := ctrclient.ObjectKeyFromObject(obj)
-			err = k8sClient.Get(ctx, key, obj)
-		}
-	}()
-
-	if !ok {
-		annotations[LastProgressTimeAnnotation] = progressTime.Format(time.RFC3339)
-		obj.SetAnnotations(annotations)
-		return
-	}
-	parsedTime, err := time.Parse(time.RFC3339, timeStr)
-	if err != nil {
-		return
-	}
-	progressTime = metav1.Time{Time: parsedTime}
-
-	return
-}
-
-func ServiceID(obj *unstructured.Unstructured) string {
-	if annotations := obj.GetAnnotations(); annotations != nil {
-		return annotations[smcommon.OwningInventoryKey]
-	}
-
-	return ""
-}
-
 // WithJitter adds a random jitter to the interval based on the global jitter factor.
 func WithJitter(interval time.Duration) time.Duration {
 	maxJitter := int64(float64(interval) * args.JitterFactor())

pkg/common/health.go

@@ -9,7 +9,6 @@ import (
 	"github.com/argoproj/argo-rollouts/pkg/apis/rollouts"
 	rolloutv1alpha1 "github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
 	flaggerv1beta1 "github.com/fluxcd/flagger/pkg/apis/flagger/v1beta1"
-	"github.com/pluralsh/deployment-operator/internal/utils"
 	appsv1 "k8s.io/api/apps/v1"
 	autoscalingv1 "k8s.io/api/autoscaling/v1"
 	autoscalingv2 "k8s.io/api/autoscaling/v2"
@@ -23,6 +22,8 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/kubectl/pkg/util/podutils"
+
+	"github.com/pluralsh/deployment-operator/internal/utils"
 )
 
 const (

pkg/controller/service/reconciler.go

@@ -134,7 +134,11 @@ func (s *ServiceReconciler) init() (*ServiceReconciler, error) {
 		return s.consoleClient.GetService(id)
 	})
 	s.manifestCache = manis.NewCache(s.manifestTTL, s.manifestTTLJitter, deployToken, s.consoleURL)
-	s.applier = applier.NewApplier(s.dynamicClient, s.discoveryCache, s.store, applier.WithWaveDelay(s.waveDelay), applier.WithFilter(applier.FilterCache, applier.CacheFilter()))
+	s.applier = applier.NewApplier(s.dynamicClient, s.discoveryCache, s.store,
+		applier.WithWaveDelay(s.waveDelay),
+		applier.WithFilter(applier.FilterCache, applier.CacheFilter()),
+		applier.WithFilter(applier.FilterSkip, applier.SkipFilter()),
+	)
 
 	return s, nil
 }
@@ -429,6 +433,11 @@ func (s *ServiceReconciler) Reconcile(ctx context.Context, id string) (result re
 			return ctrl.Result{}, err
 		}
 
+		if err = s.store.ExpireHookComponents(svc.ID); err != nil {
+			logger.Error(err, "failed to expire processed hook components", "service", svc.Name)
+			return ctrl.Result{}, err
+		}
+
 		// delete service when components len == 0 (no new statuses, inventory file is empty, all deleted)
 		if err := s.UpdateStatus(ctx, svc.ID, svc.Revision.ID, svc.Sha, lo.ToSlicePtr(components), []*console.ServiceErrorAttributes{}); err != nil {
 			logger.Error(err, "Failed to update service status, ignoring for now")

pkg/manifests/template/helm.go

@@ -137,7 +137,8 @@ func (h *helm) Render(svc *console.ServiceDeploymentForAgent, mapper meta.RESTMa
 	if err != nil {
 		return nil, err
 	}
-	if svc.Helm != nil && svc.Helm.IgnoreHooks != nil && !*svc.Helm.IgnoreHooks {
+
+	if svc.Helm != nil && !lo.FromPtr(svc.Helm.IgnoreHooks) {
 		for _, h := range rel.Hooks {
 			_, err = fmt.Fprintln(&buffer, "---")
 			if err != nil {