Ability to watch referenced or managed Kubernetes resources

Makefile

@@ -35,8 +35,8 @@ GOLANGCILINT_VERSION = 1.55.2
 
 # ====================================================================================
 # Setup Kubernetes tools
-KIND_VERSION = v0.18.0
-UP_VERSION = v0.21.0
+KIND_VERSION = v0.22.0
+UP_VERSION = v0.28.0
 UPTEST_VERSION = v0.9.0
 UP_CHANNEL = stable
 USE_HELM3 = true

apis/object/v1alpha1/types.go

@@ -124,6 +124,14 @@ type ObjectSpec struct {
 	ManagementPolicy `json:"managementPolicy,omitempty"`
 	References       []Reference `json:"references,omitempty"`
 	Readiness        Readiness   `json:"readiness,omitempty"`
+	// Watch enables watching the referenced or managed kubernetes resources.
+	//
+	// THIS IS AN ALPHA FIELD. Do not use it in production. It is not honored
+	// unless the relevant Crossplane feature flag is enabled, and may be
+	// changed or removed without notice.
+	// +optional
+	// +kubebuilder:default=false
+	Watch *bool `json:"watch,omitempty"`
 }
 
 // ReadinessPolicy defines how the Object's readiness condition should be computed.

apis/object/v1alpha2/types.go

@@ -97,6 +97,14 @@ type ObjectSpec struct {
 	ForProvider       ObjectParameters   `json:"forProvider"`
 	References        []Reference        `json:"references,omitempty"`
 	Readiness         Readiness          `json:"readiness,omitempty"`
+	// Watch enables watching the referenced or managed kubernetes resources.
+	//
+	// THIS IS AN ALPHA FIELD. Do not use it in production. It is not honored
+	// unless the relevant Crossplane feature flag is enabled, and may be
+	// changed or removed without notice.
+	// +optional
+	// +kubebuilder:default=false
+	Watch *bool `json:"watch,omitempty"`
 }
 
 // ReadinessPolicy defines how the Object's readiness condition should be computed.

cmd/provider/main.go

@@ -38,6 +38,7 @@ import (
 	"github.com/crossplane-contrib/provider-kubernetes/apis"
 	"github.com/crossplane-contrib/provider-kubernetes/apis/object/v1alpha1"
 	object "github.com/crossplane-contrib/provider-kubernetes/internal/controller"
+	"github.com/crossplane-contrib/provider-kubernetes/internal/features"
 
 	_ "k8s.io/client-go/plugin/pkg/client/auth"
 )
@@ -50,15 +51,17 @@ const (
 
 func main() {
 	var (
-		app                      = kingpin.New(filepath.Base(os.Args[0]), "Template support for Crossplane.").DefaultEnvars()
-		debug                    = app.Flag("debug", "Run with debug logging.").Short('d').Bool()
-		syncInterval             = app.Flag("sync", "Controller manager sync period such as 300ms, 1.5h, or 2h45m").Short('s').Default("1h").Duration()
-		pollInterval             = app.Flag("poll", "Poll interval controls how often an individual resource should be checked for drift.").Default("10m").Duration()
-		pollJitterPercentage     = app.Flag("poll-jitter-percentage", "Percentage of jitter to apply to poll interval. It cannot be negative, and must be less than 100.").Default("10").Uint()
-		leaderElection           = app.Flag("leader-election", "Use leader election for the controller manager.").Short('l').Default("false").Envar("LEADER_ELECTION").Bool()
-		maxReconcileRate         = app.Flag("max-reconcile-rate", "The number of concurrent reconciliations that may be running at one time.").Default("100").Int()
+		app                  = kingpin.New(filepath.Base(os.Args[0]), "Template support for Crossplane.").DefaultEnvars()
+		debug                = app.Flag("debug", "Run with debug logging.").Short('d').Bool()
+		syncInterval         = app.Flag("sync", "Controller manager sync period such as 300ms, 1.5h, or 2h45m").Short('s').Default("1h").Duration()
+		pollInterval         = app.Flag("poll", "Poll interval controls how often an individual resource should be checked for drift.").Default("10m").Duration()
+		pollJitterPercentage = app.Flag("poll-jitter-percentage", "Percentage of jitter to apply to poll interval. It cannot be negative, and must be less than 100.").Default("10").Uint()
+		leaderElection       = app.Flag("leader-election", "Use leader election for the controller manager.").Short('l').Default("false").Envar("LEADER_ELECTION").Bool()
+		maxReconcileRate     = app.Flag("max-reconcile-rate", "The number of concurrent reconciliations that may be running at one time.").Default("100").Int()
+		sanitizeSecrets      = app.Flag("sanitize-secrets", "when enabled, redacts Secret data from Object status").Default("false").Envar("SANITIZE_SECRETS").Bool()
+
 		enableManagementPolicies = app.Flag("enable-management-policies", "Enable support for Management Policies.").Default("true").Envar("ENABLE_MANAGEMENT_POLICIES").Bool()
-		sanitizeSecrets          = app.Flag("sanitize-secrets", "when enabled, redacts Secret data from Object status").Default("false").Envar("SANITIZE_SECRETS").Bool()
+		enableWatches            = app.Flag("enable-watches", "Enable support for watching resources.").Default("false").Envar("ENABLE_WATCHES").Bool()
 	)
 	kingpin.MustParse(app.Parse(os.Args[1:]))
 
@@ -135,6 +138,11 @@ func main() {
 		log.Info("Beta feature enabled", "flag", feature.EnableBetaManagementPolicies)
 	}
 
+	if *enableWatches {
+		o.Features.Enable(features.EnableAlphaWatches)
+		log.Info("Alpha feature enabled", "flag", features.EnableAlphaWatches)
+	}
+
 	// NOTE(lsviben): We are registering the conversion webhook with v1alpha1
 	// Object. As far as I can see and based on some tests, it doesn't matter
 	// which version we use here. Leaving it as v1alpha1 as it will be easy to

examples/object/object.yaml

@@ -3,6 +3,10 @@ kind: Object
 metadata:
   name: sample-namespace
 spec:
+  # Watch for changes to the Namespace object.
+  # Watching resources is an alpha feature and needs to be enabled with --enable-watches
+  # in the provider to get this configuration working.
+  # watch: true
   forProvider:
     manifest:
       apiVersion: v1

examples/object/references/patches-from-resource.yaml

@@ -4,6 +4,10 @@ kind: Object
 metadata:
   name: foo
 spec:
+  # Watch for changes to the Namespace object.
+  # Watching resources is an alpha feature and needs to be enabled with --enable-watches
+  # in the provider to get this configuration working.
+  # watch: true
   references:
   # Use patchesFrom to patch field from other k8s resource to this object
   - patchesFrom:

internal/clients/client.go → internal/clients/clients.go

@@ -107,8 +107,33 @@ func restConfigFromAPIConfig(c *api.Config) (*rest.Config, error) {
 	return config, nil
 }
 
+// RestConfigGetter is an interface that provides a REST config.
+type RestConfigGetter interface {
+	// GetConfig returns an initialized Config
+	GetConfig() *rest.Config
+}
+
+// ClusterClient is a client that can be used to interact with a Kubernetes
+// cluster including getting its rest config.
+type ClusterClient interface {
+	client.Client
+	resource.Applicator
+	RestConfigGetter
+}
+
+// ApplicatorClientWithConfig is a ClusterClient that also has a rest config.
+type ApplicatorClientWithConfig struct {
+	resource.ClientApplicator
+	config *rest.Config
+}
+
+// GetConfig returns the rest config for the client.
+func (c *ApplicatorClientWithConfig) GetConfig() *rest.Config {
+	return c.config
+}
+
 // ClientForProvider returns the client for the given provider config
-func ClientForProvider(ctx context.Context, inclusterClient client.Client, providerConfigName string) (client.Client, error) { //nolint:gocyclo
+func ClientForProvider(ctx context.Context, inclusterClient client.Client, providerConfigName string) (ClusterClient, error) { //nolint:gocyclo
 	pc := &v1alpha1.ProviderConfig{}
 	if err := inclusterClient.Get(ctx, types.NamespacedName{Name: providerConfigName}, pc); err != nil {
 		return nil, errors.Wrap(err, errGetPC)
@@ -171,6 +196,15 @@ func ClientForProvider(ctx context.Context, inclusterClient client.Client, provi
 			return nil, errors.Errorf("unknown identity type: %s", id.Type)
 		}
 	}
-
-	return NewKubeClient(rc)
+	k, err := NewKubeClient(rc)
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot create Kubernetes client")
+	}
+	return &ApplicatorClientWithConfig{
+		ClientApplicator: resource.ClientApplicator{
+			Client:     k,
+			Applicator: resource.NewAPIPatchingApplicator(k),
+		},
+		config: rc,
+	}, nil
 }

internal/controller/object/indexes.go

@@ -0,0 +1,130 @@
+/*
+Copyright 2024 The Crossplane Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package object
+
+import (
+	"context"
+	"fmt"
+
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/util/workqueue"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	runtimeevent "sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	"github.com/crossplane/crossplane-runtime/pkg/logging"
+
+	"github.com/crossplane-contrib/provider-kubernetes/apis/object/v1alpha2"
+)
+
+const (
+	// resourceRefGVKsIndex is an index of all GroupKinds that
+	// are in use by an Object.
+	resourceRefGVKsIndex = "objectsRefsGVKs"
+	// resourceRefsIndex is an index of resourceRefs that are referenced or
+	// managed by an Object.
+	resourceRefsIndex = "objectsRefs"
+)
+
+var (
+	_ client.IndexerFunc = IndexByProviderGVK
+	_ client.IndexerFunc = IndexByProviderNamespacedNameGVK
+)
+
+// IndexByProviderGVK assumes the passed object is an Object. It returns keys
+// with "ProviderConfig + GVK" for every resource referenced or managed by the
+// Object.
+func IndexByProviderGVK(o client.Object) []string {
+	obj, ok := o.(*v1alpha2.Object)
+	if !ok {
+		return nil // should never happen
+	}
+
+	// Index references.
+	refs := obj.Spec.References
+	keys := make([]string, 0, len(refs))
+	for _, ref := range refs {
+		refAPIVersion, refKind, _, _ := getReferenceInfo(ref)
+		group, version := parseAPIVersion(refAPIVersion)
+		providerConfig := "" // references are always local (i.e. on the control plane), which we represent as an empty provider config.
+		keys = append(keys, refKeyProviderGVK(providerConfig, refKind, group, version))
+	}
+
+	// Index the desired object.
+	// We don't expect errors here, as the getDesired function is already called
+	// in the reconciler and the desired object already validated.
+	d, _ := getDesired(obj)
+	keys = append(keys, refKeyProviderGVK(obj.Spec.ProviderConfigReference.Name, d.GetKind(), d.GroupVersionKind().Group, d.GroupVersionKind().Version)) // unification is done by the informer.
+
+	// unification is done by the informer.
+	return keys
+}
+
+func refKeyProviderGVK(providerConfig, kind, group, version string) string {
+	return fmt.Sprintf("%s.%s.%s.%s", providerConfig, kind, group, version)
+}
+
+// IndexByProviderNamespacedNameGVK assumes the passed object is an Object. It
+// returns keys with "ProviderConfig + NamespacedName + GVK" for every resource
+// referenced or managed by the Object.
+func IndexByProviderNamespacedNameGVK(o client.Object) []string {
+	obj, ok := o.(*v1alpha2.Object)
+	if !ok {
+		return nil // should never happen
+	}
+
+	// Index references.
+	refs := obj.Spec.References
+	keys := make([]string, 0, len(refs))
+	for _, ref := range refs {
+		refAPIVersion, refKind, refNamespace, refName := getReferenceInfo(ref)
+		providerConfig := "" // references are always local (i.e. on the control plane), which we represent as an empty provider config.
+		keys = append(keys, refKeyProviderNamespacedNameGVK(providerConfig, refNamespace, refName, refKind, refAPIVersion))
+	}
+
+	// Index the desired object.
+	// We don't expect errors here, as the getDesired function is already called
+	// in the reconciler and the desired object already validated.
+	d, _ := getDesired(obj)
+	keys = append(keys, refKeyProviderNamespacedNameGVK(obj.Spec.ProviderConfigReference.Name, d.GetNamespace(), d.GetName(), d.GetKind(), d.GetAPIVersion())) // unification is done by the informer.
+
+	return keys
+}
+
+func refKeyProviderNamespacedNameGVK(providerConfig, ns, name, kind, apiVersion string) string {
+	return fmt.Sprintf("%s.%s.%s.%s.%s", providerConfig, name, ns, kind, apiVersion)
+}
+
+func enqueueObjectsForReferences(ca cache.Cache, log logging.Logger) func(ctx context.Context, ev runtimeevent.GenericEvent, q workqueue.RateLimitingInterface) {
+	return func(ctx context.Context, ev runtimeevent.GenericEvent, q workqueue.RateLimitingInterface) {
+		pc, _ := ctx.Value(keyProviderConfigName).(string)
+		rGVK := ev.Object.GetObjectKind().GroupVersionKind()
+		key := refKeyProviderNamespacedNameGVK(pc, ev.Object.GetNamespace(), ev.Object.GetName(), rGVK.Kind, rGVK.GroupVersion().String())
+
+		objects := v1alpha2.ObjectList{}
+		if err := ca.List(ctx, &objects, client.MatchingFields{resourceRefsIndex: key}); err != nil {
+			log.Debug("cannot list objects related to a reference change", "error", err, "fieldSelector", resourceRefsIndex+"="+key)
+			return
+		}
+		// queue those Objects for reconciliation
+		for _, o := range objects.Items {
+			log.Info("Enqueueing Object because referenced resource changed", "name", o.GetName(), "referencedGVK", rGVK.String(), "referencedName", ev.Object.GetName(), "providerConfig", pc)
+			q.Add(reconcile.Request{NamespacedName: types.NamespacedName{Name: o.GetName()}})
+		}
+	}
+}