feat(agent): add CRD gate implementation and tests

go/deployment-operator/go.mod

@@ -69,6 +69,7 @@ require (
 	github.com/yuin/gopher-lua v1.1.1
 	gitlab.com/gitlab-org/api/client-go v1.46.0
 	golang.org/x/oauth2 v0.35.0
+	golang.org/x/sync v0.21.0
 	golang.org/x/time v0.14.0
 	gopkg.in/yaml.v3 v3.0.1
 	gotest.tools/gotestsum v1.13.0
@@ -350,7 +351,6 @@ require (
 	golang.org/x/exp/typeparams v0.0.0-20260209203927-2842357ff358 // indirect
 	golang.org/x/mod v0.36.0 // indirect
 	golang.org/x/net v0.55.0 // indirect
-	golang.org/x/sync v0.21.0 // indirect
 	golang.org/x/sys v0.46.0 // indirect
 	golang.org/x/term v0.44.0 // indirect
 	golang.org/x/text v0.38.0 // indirect

go/deployment-operator/pkg/cache/discovery/cache.go

@@ -37,6 +37,9 @@ type Cache interface {
 	// MaybeResetRESTMapper resets the RESTMapper if the provided GVKs are CustomResourceDefinitions.
 	MaybeResetRESTMapper(...schema.GroupVersionKind)
 
+	// ResetRESTMapper resets cached discovery used by the RESTMapper.
+	ResetRESTMapper()
+
 	// GroupVersionKind returns the set of GroupVersionKinds in the cache.
 	GroupVersionKind() containers.Set[schema.GroupVersionKind]
 
@@ -151,6 +154,19 @@ func (in *cache) MaybeResetRESTMapper(crds ...schema.GroupVersionKind) {
 	}
 }
 
+func (in *cache) ResetRESTMapper() {
+	in.mu.Lock()
+	defer in.mu.Unlock()
+
+	if in.mapper == nil {
+		klog.V(log.LogLevelVerbose).ErrorS(fmt.Errorf("no RESTMapper provided, cannot reset"), "unable to reset RESTMapper")
+		return
+	}
+
+	meta.MaybeResetRESTMapper(in.mapper)
+	klog.V(log.LogLevelExtended).InfoS("resetting RESTMapper")
+}
+
 func (in *cache) RestMapping(gvk schema.GroupVersionKind) (*meta.RESTMapping, error) {
 	in.mu.Lock()
 	mapping, err := in.mapper.RESTMapping(gvk.GroupKind(), gvk.Version)

go/deployment-operator/pkg/cache/discovery/cache_test.go

@@ -0,0 +1,34 @@
+package discovery
+
+import (
+	"sync/atomic"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"k8s.io/apimachinery/pkg/api/meta"
+)
+
+type resettableMapper struct {
+	meta.RESTMapper
+	resetCalls atomic.Int32
+}
+
+func (in *resettableMapper) Reset() {
+	in.resetCalls.Add(1)
+}
+
+func TestResetRESTMapper(t *testing.T) {
+	t.Run("delegates to resettable mapper", func(t *testing.T) {
+		mapper := &resettableMapper{}
+		cache := NewCache(nil, mapper)
+
+		cache.ResetRESTMapper()
+
+		assert.Equal(t, int32(1), mapper.resetCalls.Load())
+	})
+
+	t.Run("ignores missing mapper", func(t *testing.T) {
+		cache := NewCache(nil, nil)
+		assert.NotPanics(t, cache.ResetRESTMapper)
+	})
+}

go/deployment-operator/pkg/common/crd.go

@@ -0,0 +1,54 @@
+package common
+
+import (
+	"github.com/samber/lo"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+var crdGroupKind = apiextensionsv1.SchemeGroupVersion.WithKind(CustomResourceDefinitionKind).GroupKind()
+
+func IsCRD(resource unstructured.Unstructured) bool {
+	return resource.GroupVersionKind().GroupKind() == crdGroupKind
+}
+
+func isStatusConditionTrue(resource unstructured.Unstructured, conditionType string) bool {
+	conditions, _, _ := unstructured.NestedSlice(resource.Object, "status", "conditions")
+	return meta.IsStatusConditionTrue(lo.FilterMap(conditions, func(condition any, _ int) (metav1.Condition, bool) {
+		value, ok := condition.(map[string]any)
+		if !ok {
+			return metav1.Condition{}, false
+		}
+
+		currentType, typeFound, _ := unstructured.NestedString(value, "type")
+		conditionStatus, statusFound, _ := unstructured.NestedString(value, "status")
+		return metav1.Condition{
+			Type:   currentType,
+			Status: metav1.ConditionStatus(conditionStatus),
+		}, typeFound && statusFound
+	}), conditionType)
+}
+
+func CRDEstablished(resource unstructured.Unstructured) bool {
+	return IsCRD(resource) && isStatusConditionTrue(resource, string(apiextensionsv1.Established))
+}
+
+func ServedCRDGVKs(resource unstructured.Unstructured) []schema.GroupVersionKind {
+	if !IsCRD(resource) {
+		return nil
+	}
+
+	crd := new(apiextensionsv1.CustomResourceDefinition)
+	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(resource.Object, crd); err != nil {
+		return nil
+	}
+
+	return lo.FilterMap(crd.Spec.Versions, func(version apiextensionsv1.CustomResourceDefinitionVersion, _ int) (schema.GroupVersionKind, bool) {
+		gvk := schema.GroupVersion{Group: crd.Spec.Group, Version: version.Name}.WithKind(crd.Spec.Names.Kind)
+		return gvk, version.Served && gvk.Group != "" && gvk.Version != "" && gvk.Kind != ""
+	})
+}

go/deployment-operator/pkg/common/crd_test.go

@@ -0,0 +1,40 @@
+package common
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+func TestCRDUtilities(t *testing.T) {
+	crd := unstructured.Unstructured{Object: map[string]any{
+		"apiVersion": "apiextensions.k8s.io/v1",
+		"kind":       "CustomResourceDefinition",
+		"spec": map[string]any{
+			"group": "example.com",
+			"names": map[string]any{"kind": "Widget", "plural": "widgets"},
+			"versions": []any{
+				map[string]any{"name": "v1", "served": true, "storage": true},
+				map[string]any{"name": "v2", "served": false, "storage": false},
+			},
+		},
+		"status": map[string]any{
+			"conditions": []any{map[string]any{"type": "Established", "status": "True"}},
+		},
+	}}
+
+	assert.True(t, IsCRD(crd))
+	assert.True(t, CRDEstablished(crd))
+	assert.Equal(t, []schema.GroupVersionKind{{Group: "example.com", Version: "v1", Kind: "Widget"}}, ServedCRDGVKs(crd))
+	assert.False(t, CRDEstablished(unstructured.Unstructured{}))
+
+	versions, _, _ := unstructured.NestedSlice(crd.Object, "spec", "versions")
+	versions = append(versions,
+		map[string]any{"name": "v3", "served": false},
+		map[string]any{"name": "", "served": true},
+	)
+	assert.NoError(t, unstructured.SetNestedSlice(crd.Object, versions, "spec", "versions"))
+	assert.Equal(t, []schema.GroupVersionKind{{Group: "example.com", Version: "v1", Kind: "Widget"}}, ServedCRDGVKs(crd))
+}

go/deployment-operator/pkg/manifests/template/common.go

@@ -16,13 +16,10 @@ import (
 	"sigs.k8s.io/kustomize/kyaml/kio/kioutil"
 	"sigs.k8s.io/kustomize/kyaml/yaml"
 
+	commonpkg "github.com/pluralsh/console/go/deployment-operator/pkg/common"
 	"github.com/pluralsh/console/go/deployment-operator/pkg/log"
 )
 
-var (
-	crdGK = schema.GroupKind{Group: "apiextensions.k8s.io", Kind: "CustomResourceDefinition"}
-)
-
 func setNamespaces(mapper meta.RESTMapper, objs []unstructured.Unstructured,
 	defaultNamespace string, enforceNamespace bool) ([]unstructured.Unstructured, error) {
 	// find any crds in the set of resources.
@@ -134,8 +131,7 @@ func IsCRD(u *unstructured.Unstructured) bool {
 	if u == nil {
 		return false
 	}
-	gvk := u.GroupVersionKind()
-	return crdGK == gvk.GroupKind()
+	return commonpkg.IsCRD(*u)
 }
 
 // LookupResourceScope tries to look up the scope of the type of the provided

go/deployment-operator/pkg/streamline/applier/applier.go

@@ -4,20 +4,20 @@ import (
 	"context"
 	"time"
 
-	"github.com/pluralsh/console/go/client"
-	"github.com/pluralsh/console/go/polly/containers"
 	"github.com/samber/lo"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/klog/v2"
 
+	"github.com/pluralsh/console/go/client"
 	"github.com/pluralsh/console/go/deployment-operator/internal/helpers"
 	discoverycache "github.com/pluralsh/console/go/deployment-operator/pkg/cache/discovery"
 	"github.com/pluralsh/console/go/deployment-operator/pkg/log"
 	smcommon "github.com/pluralsh/console/go/deployment-operator/pkg/streamline/common"
 	"github.com/pluralsh/console/go/deployment-operator/pkg/streamline/store"
+	"github.com/pluralsh/console/go/polly/containers"
 )
 
 type Applier struct {
@@ -41,6 +41,8 @@ func (in *Applier) Apply(ctx context.Context,
 	opts ...WaveProcessorOption,
 ) ([]client.ComponentAttributes, []client.ServiceErrorAttributes, error) {
 	resources = in.ensureServiceAnnotation(resources, service.ID)
+	gates := []Gate{newCRDGate(in.store, in.discoveryCache, withCRDGateResources(resources, lo.FromPtr(service.DryRun)))}
+	opts = append(opts, WithWaveGates(lo.Filter(gates, func(g Gate, _ int) bool { return g.Enabled() })...))
 
 	if err := in.store.SyncServiceComponents(service.ID, resources); err != nil {
 		return nil, nil, err
@@ -72,7 +74,7 @@ func (in *Applier) Apply(ctx context.Context,
 		}
 
 		now := time.Now()
-		syncPhase = lo.ToPtr(phase.Name())
+		syncPhase = new(phase.Name())
 
 		if !phase.HasWaves() {
 			klog.V(log.LogLevelDefault).InfoS(
@@ -134,7 +136,7 @@ func (in *Applier) Apply(ctx context.Context,
 			serviceErrorList = append(serviceErrorList, client.ServiceErrorAttributes{
 				Source:  string(phase.Name()),
 				Message: "waiting for resources to be ready",
-				Warning: lo.ToPtr(true),
+				Warning: new(true),
 			})
 			klog.V(log.LogLevelTrace).InfoS("waiting for resources to be ready", "phase", phase.Name())
 			break
@@ -198,8 +200,8 @@ func (in *Applier) Destroy(ctx context.Context, serviceID string) ([]client.Comp
 		err = in.client.
 			Resource(helpers.GVRFromGVK(live.GroupVersionKind())).
 			Namespace(live.GetNamespace()).Delete(ctx, live.GetName(), metav1.DeleteOptions{
-			GracePeriodSeconds: lo.ToPtr(int64(0)),
-			PropagationPolicy:  lo.ToPtr(metav1.DeletePropagationBackground),
+			GracePeriodSeconds: new(int64(0)),
+			PropagationPolicy:  new(metav1.DeletePropagationBackground),
 		})
 		if errors.IsNotFound(err) {
 			if err := in.store.DeleteComponent(smcommon.NewStoreKeyFromUnstructured(lo.FromPtr(live))); err != nil {