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/clients.go

@@ -0,0 +1,14 @@
+/*
+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 clients

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

@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Crossplane Authors.
+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
@@ -11,7 +11,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package clients
+package kube
 
 import (
 	"context"
@@ -41,76 +41,24 @@ const (
 	errInjectAzureCredentials   = "failed to wrap REST client with Azure Application Credentials"
 )
 
-// NewRESTConfig returns a rest config given a secret with connection information.
-func NewRESTConfig(kubeconfig []byte) (*rest.Config, error) {
-	ac, err := clientcmd.Load(kubeconfig)
+// ClientForProvider returns the client and *rest.config for the given provider
+// config.
+func ClientForProvider(ctx context.Context, inclusterClient client.Client, providerConfigName string) (client.Client, *rest.Config, error) { //nolint:gocyclo
+	rc, err := configForProvider(ctx, inclusterClient, providerConfigName)
 	if err != nil {
-		return nil, errors.Wrap(err, "failed to load kubeconfig")
+		return nil, nil, errors.Wrapf(err, "cannot get REST config for provider %q", providerConfigName)
 	}
-	return restConfigFromAPIConfig(ac)
-}
-
-// NewKubeClient returns a kubernetes client given a secret with connection
-// information.
-func NewKubeClient(config *rest.Config) (client.Client, error) {
-	kc, err := client.New(config, client.Options{})
+	k, err := client.New(rc, client.Options{})
 	if err != nil {
-		return nil, errors.Wrap(err, "cannot create Kubernetes client")
-	}
-
-	return kc, nil
-}
-
-func restConfigFromAPIConfig(c *api.Config) (*rest.Config, error) {
-	if c.CurrentContext == "" {
-		return nil, errors.New("currentContext not set in kubeconfig")
-	}
-	ctx := c.Contexts[c.CurrentContext]
-	cluster := c.Clusters[ctx.Cluster]
-	if cluster == nil {
-		return nil, errors.Errorf("cluster for currentContext (%s) not found", c.CurrentContext)
-	}
-	user := c.AuthInfos[ctx.AuthInfo]
-	if user == nil {
-		// We don't require a user because it's possible user
-		// authorization configuration will be loaded from a separate
-		// set of identity credentials (e.g. Google Application Creds).
-		user = &api.AuthInfo{}
-	}
-	config := &rest.Config{
-		Host:            cluster.Server,
-		Username:        user.Username,
-		Password:        user.Password,
-		BearerToken:     user.Token,
-		BearerTokenFile: user.TokenFile,
-		Impersonate: rest.ImpersonationConfig{
-			UserName: user.Impersonate,
-			Groups:   user.ImpersonateGroups,
-			Extra:    user.ImpersonateUserExtra,
-		},
-		AuthProvider: user.AuthProvider,
-		ExecProvider: user.Exec,
-		TLSClientConfig: rest.TLSClientConfig{
-			Insecure:   cluster.InsecureSkipTLSVerify,
-			ServerName: cluster.TLSServerName,
-			CertData:   user.ClientCertificateData,
-			KeyData:    user.ClientKeyData,
-			CAData:     cluster.CertificateAuthorityData,
-		},
+		return nil, nil, errors.Wrapf(err, "cannot create Kubernetes client for provider %q", providerConfigName)
 	}
-
-	// NOTE(tnthornton): these values match the burst and QPS values in kubectl.
-	// xref: https://github.com/kubernetes/kubernetes/pull/105520
-	config.Burst = 300
-	config.QPS = 50
-
-	return config, nil
+	return k, rc, nil
 }
 
-// ClientForProvider returns the client for the given provider config
-func ClientForProvider(ctx context.Context, inclusterClient client.Client, providerConfigName string) (client.Client, error) { //nolint:gocyclo
+// ConfigForProvider returns the *rest.config for the given provider config.
+func configForProvider(ctx context.Context, local client.Client, providerConfigName string) (*rest.Config, error) { // nolint:gocyclo
 	pc := &v1alpha1.ProviderConfig{}
-	if err := inclusterClient.Get(ctx, types.NamespacedName{Name: providerConfigName}, pc); err != nil {
+	if err := local.Get(ctx, types.NamespacedName{Name: providerConfigName}, pc); err != nil {
 		return nil, errors.Wrap(err, errGetPC)
 	}
 
@@ -124,12 +72,17 @@ func ClientForProvider(ctx context.Context, inclusterClient client.Client, provi
 			return nil, errors.Wrap(err, errCreateRestConfig)
 		}
 	default:
-		kc, err := resource.CommonCredentialExtractor(ctx, cd.Source, inclusterClient, cd.CommonCredentialSelectors)
+		kc, err := resource.CommonCredentialExtractor(ctx, cd.Source, local, cd.CommonCredentialSelectors)
 		if err != nil {
 			return nil, errors.Wrap(err, errGetCreds)
 		}
 
-		if rc, err = NewRESTConfig(kc); err != nil {
+		ac, err := clientcmd.Load(kc)
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to load kubeconfig")
+		}
+
+		if rc, err = fromAPIConfig(ac); err != nil {
 			return nil, errors.Wrap(err, errCreateRestConfig)
 		}
 	}
@@ -143,7 +96,7 @@ func ClientForProvider(ctx context.Context, inclusterClient client.Client, provi
 					return nil, errors.Wrap(err, errInjectGoogleCredentials)
 				}
 			default:
-				creds, err := resource.CommonCredentialExtractor(ctx, id.Source, inclusterClient, id.CommonCredentialSelectors)
+				creds, err := resource.CommonCredentialExtractor(ctx, id.Source, local, id.CommonCredentialSelectors)
 				if err != nil {
 					return nil, errors.Wrap(err, errExtractGoogleCredentials)
 				}
@@ -158,7 +111,7 @@ func ClientForProvider(ctx context.Context, inclusterClient client.Client, provi
 				return nil, errors.Errorf("%s is not supported as identity source for identity type %s",
 					xpv1.CredentialsSourceInjectedIdentity, v1alpha1.IdentityTypeAzureServicePrincipalCredentials)
 			default:
-				creds, err := resource.CommonCredentialExtractor(ctx, id.Source, inclusterClient, id.CommonCredentialSelectors)
+				creds, err := resource.CommonCredentialExtractor(ctx, id.Source, local, id.CommonCredentialSelectors)
 				if err != nil {
 					return nil, errors.Wrap(err, errExtractAzureCredentials)
 				}
@@ -172,5 +125,51 @@ func ClientForProvider(ctx context.Context, inclusterClient client.Client, provi
 		}
 	}
 
-	return NewKubeClient(rc)
+	return rc, nil
+}
+
+func fromAPIConfig(c *api.Config) (*rest.Config, error) {
+	if c.CurrentContext == "" {
+		return nil, errors.New("currentContext not set in kubeconfig")
+	}
+	ctx := c.Contexts[c.CurrentContext]
+	cluster := c.Clusters[ctx.Cluster]
+	if cluster == nil {
+		return nil, errors.Errorf("cluster for currentContext (%s) not found", c.CurrentContext)
+	}
+	user := c.AuthInfos[ctx.AuthInfo]
+	if user == nil {
+		// We don't require a user because it's possible user
+		// authorization configuration will be loaded from a separate
+		// set of identity credentials (e.g. Google Application Creds).
+		user = &api.AuthInfo{}
+	}
+	config := &rest.Config{
+		Host:            cluster.Server,
+		Username:        user.Username,
+		Password:        user.Password,
+		BearerToken:     user.Token,
+		BearerTokenFile: user.TokenFile,
+		Impersonate: rest.ImpersonationConfig{
+			UserName: user.Impersonate,
+			Groups:   user.ImpersonateGroups,
+			Extra:    user.ImpersonateUserExtra,
+		},
+		AuthProvider: user.AuthProvider,
+		ExecProvider: user.Exec,
+		TLSClientConfig: rest.TLSClientConfig{
+			Insecure:   cluster.InsecureSkipTLSVerify,
+			ServerName: cluster.TLSServerName,
+			CertData:   user.ClientCertificateData,
+			KeyData:    user.ClientKeyData,
+			CAData:     cluster.CertificateAuthorityData,
+		},
+	}
+
+	// NOTE(tnthornton): these values match the burst and QPS values in kubectl.
+	// xref: https://github.com/kubernetes/kubernetes/pull/105520
+	config.Burst = 300
+	config.QPS = 50
+
+	return config, nil
 }