Merge pull request crossplane#6407 from negz/pollution

Promote realtime composition to beta - don't poll in the XR reconciler

Author: negz

.github/workflows/ci.yml

@@ -227,7 +227,6 @@ jobs:
       matrix:
         test-suite:
           - base
-          - realtime-compositions
           - package-dependency-updates
           - package-signature-verification
 

apis/apiextensions/fn/proto/v1/run_function.pb.go

@@ -55,7 +55,7 @@ message RunFunctionRequest {
   // of the 'input' block of the relevant entry in a Composition's pipeline.
   optional google.protobuf.Struct input = 4;
 
-  // Optional context. Crossplane may pass arbitary contextual information to a
+  // Optional context. Crossplane may pass arbitrary contextual information to a
   // Function. A Function may also return context in its RunFunctionResponse,
   // and that context will be passed to subsequent Functions. Crossplane
   // discards all context returned by the last Function in the pipeline.
@@ -170,9 +170,9 @@ message ResponseMeta {
   // meta.tag of the corresponding RunFunctionRequest.
   string tag = 1;
 
-  // Time-to-live of this response. Deterministic Functions with no side-effects
-  // (e.g. simple templating Functions) may specify a TTL. Crossplane may choose
-  // to cache responses until the TTL expires.
+  // Time-to-live of this response. Crossplane will call the function again when
+  // the TTL expires. Crossplane may cache the response to avoid calling the
+  // function again until the TTL expires.
   optional google.protobuf.Duration ttl = 2;
 }
 

apis/apiextensions/fn/proto/v1/run_function.proto

@@ -57,7 +57,7 @@ message RunFunctionRequest {
   // of the 'input' block of the relevant entry in a Composition's pipeline.
   optional google.protobuf.Struct input = 4;
 
-  // Optional context. Crossplane may pass arbitary contextual information to a
+  // Optional context. Crossplane may pass arbitrary contextual information to a
   // Function. A Function may also return context in its RunFunctionResponse,
   // and that context will be passed to subsequent Functions. Crossplane
   // discards all context returned by the last Function in the pipeline.
@@ -172,9 +172,9 @@ message ResponseMeta {
   // meta.tag of the corresponding RunFunctionRequest.
   string tag = 1;
 
-  // Time-to-live of this response. Deterministic Functions with no side-effects
-  // (e.g. simple templating Functions) may specify a TTL. Crossplane may choose
-  // to cache responses until the TTL expires.
+  // Time-to-live of this response. Crossplane will call the function again when
+  // the TTL expires. Crossplane may cache the response to avoid calling the
+  // function again until the TTL expires.
   optional google.protobuf.Duration ttl = 2;
 }
 

apis/apiextensions/fn/proto/v1beta1/zz_generated_run_function.pb.go

@@ -116,14 +116,14 @@ type startCommand struct {
 	TLSClientCertsDir   string `env:"TLS_CLIENT_CERTS_DIR"   help:"The path of the folder which will store TLS client certificate of Crossplane."`
 
 	EnableExternalSecretStores      bool `group:"Alpha Features:" help:"Enable support for External Secret Stores."`
-	EnableRealtimeCompositions      bool `group:"Alpha Features:" help:"Enable support for realtime compositions, i.e. watching composed resources and reconciling compositions immediately when any of the composed resources is updated."`
 	EnableDependencyVersionUpgrades bool `group:"Alpha Features:" help:"Enable support for upgrading dependency versions when the parent package is updated."`
 	EnableSignatureVerification     bool `group:"Alpha Features:" help:"Enable support for package signature verification via ImageConfig API."`
 
 	EnableCompositionWebhookSchemaValidation bool `default:"true" group:"Beta Features:" help:"Enable support for Composition validation using schemas."`
 	EnableDeploymentRuntimeConfigs           bool `default:"true" group:"Beta Features:" help:"Enable support for Deployment Runtime Configs."`
 	EnableUsages                             bool `default:"true" group:"Beta Features:" help:"Enable support for deletion ordering and resource protection with Usages."`
 	EnableSSAClaims                          bool `default:"true" group:"Beta Features:" help:"Enable support for using Kubernetes server-side apply to sync claims with composite resources (XRs)."`
+	EnableRealtimeCompositions               bool `default:"true" group:"Beta Features:" help:"Enable support for realtime compositions, i.e. watching composed resources and reconciling compositions immediately when any of the composed resources is updated."`
 
 	// These are GA features that previously had alpha or beta feature flags.
 	// You can't turn off a GA feature. We maintain the flags to avoid breaking
@@ -244,14 +244,14 @@ func (c *startCommand) Run(s *runtime.Scheme, log logging.Logger) error { //noli
 		return errors.Wrap(err, "cannot load client TLS certificates")
 	}
 
-	m := xfn.NewMetrics()
-	metrics.Registry.MustRegister(m)
+	xfnm := xfn.NewPrometheusMetrics()
+	metrics.Registry.MustRegister(xfnm)
 
 	// We want all XR controllers to share the same gRPC clients.
 	functionRunner := xfn.NewPackagedFunctionRunner(mgr.GetClient(),
 		xfn.WithLogger(log),
 		xfn.WithTLSConfig(clienttls),
-		xfn.WithInterceptorCreators(m),
+		xfn.WithInterceptorCreators(xfnm),
 	)
 
 	// Periodically remove clients for Functions that no longer exist.
@@ -283,8 +283,8 @@ func (c *startCommand) Run(s *runtime.Scheme, log logging.Logger) error { //noli
 		}
 	}
 	if c.EnableRealtimeCompositions {
-		o.Features.Enable(features.EnableAlphaRealtimeCompositions)
-		log.Info("Alpha feature enabled", "flag", features.EnableAlphaRealtimeCompositions)
+		o.Features.Enable(features.EnableBetaRealtimeCompositions)
+		log.Info("Beta feature enabled", "flag", features.EnableBetaRealtimeCompositions)
 	}
 	if c.EnableDeploymentRuntimeConfigs {
 		o.Features.Enable(features.EnableBetaDeploymentRuntimeConfigs)
@@ -375,6 +375,9 @@ func (c *startCommand) Run(s *runtime.Scheme, log logging.Logger) error { //noli
 		return errors.Wrap(err, "cannot create uncached client for API extension controllers")
 	}
 
+	cem := engine.NewPrometheusMetrics()
+	metrics.Registry.MustRegister(cem)
+
 	// It's important the engine's client is wrapped with unstructured.NewClient
 	// because controller-runtime always caches *unstructured.Unstructured, not
 	// our wrapper types like *composite.Unstructured. This client takes care of
@@ -384,6 +387,7 @@ func (c *startCommand) Run(s *runtime.Scheme, log logging.Logger) error { //noli
 		unstructured.NewClient(cached),
 		unstructured.NewClient(uncached),
 		engine.WithLogger(log),
+		engine.WithMetrics(cem),
 	)
 
 	// TODO(negz): Garbage collect informers for CRs that are still defined

apis/apiextensions/fn/proto/v1beta1/zz_generated_run_function.proto

@@ -98,7 +98,7 @@ type CompositeSyncer interface {
 // composite resource (XR).
 type CompositeSyncerFn func(ctx context.Context, cm *claim.Unstructured, xr *composite.Unstructured) error
 
-// Sync the supplied claim with the supplied composite resource..
+// Sync the supplied claim with the supplied composite resource.
 func (fn CompositeSyncerFn) Sync(ctx context.Context, cm *claim.Unstructured, xr *composite.Unstructured) error {
 	return fn(ctx, cm, xr)
 }
@@ -188,9 +188,8 @@ type Reconciler struct {
 	composite crComposite
 	claim     crClaim
 
-	log          logging.Logger
-	record       event.Recorder
-	pollInterval time.Duration
+	log    logging.Logger
+	record event.Recorder
 }
 
 type crComposite struct {
@@ -275,17 +274,6 @@ func WithRecorder(er event.Recorder) ReconcilerOption {
 	}
 }
 
-// WithPollInterval specifies how long the Reconciler should wait before queueing
-// a new reconciliation after a successful reconcile. The Reconciler requeues
-// after a specified duration when it is not actively waiting for an external
-// operation, but wishes to check whether resources it does not have a watch on
-// (i.e. composed resources) need to be reconciled.
-func WithPollInterval(after time.Duration) ReconcilerOption {
-	return func(r *Reconciler) {
-		r.pollInterval = after
-	}
-}
-
 // NewReconciler returns a Reconciler that reconciles composite resource claims of
 // the supplied CompositeClaimKind with resources of the supplied CompositeKind.
 // The returned Reconciler will apply only the ObjectMetaConfigurator by

cmd/crossplane/core/core.go

@@ -22,15 +22,6 @@ import (
 	v1 "github.com/crossplane/crossplane/apis/apiextensions/v1"
 )
 
-// A CompositeResource is an output of the composition process.
-type CompositeResource struct { //nolint:revive // stick with CompositeResource
-	// Ready indicated whether the composite resource should be marked as
-	// ready or unready regardless of the state of the composed resoureces.
-	// If it is nil the readiness of the composite is determined by the
-	// readiness of the composed resources.
-	Ready *bool
-}
-
 // A ResourceName uniquely identifies the composed resource within a Composition
 // and within Composition Function gRPC calls. This is not the metadata.name of
 // the actual composed resource instance; rather it is the name of an entry in a

internal/controller/apiextensions/claim/reconciler.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"sort"
 	"strings"
+	"time"
 
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/types/known/structpb"
@@ -269,6 +270,11 @@ func (c *FunctionComposer) Compose(ctx context.Context, xr *composite.Unstructur
 		return CompositionResult{}, errors.Wrap(err, errBuildObserved)
 	}
 
+	// Time-to-live for this composition pipeline run. Each function returns
+	// a TTL. The pipeline's TTL will be the shortest non-zero TTL returned
+	// by any function. A TTL of zero means unlimited TTL.
+	var ttl time.Duration
+
 	// The Function pipeline starts with empty desired state.
 	d := &fnv1.State{}
 
@@ -319,6 +325,13 @@ func (c *FunctionComposer) Compose(ctx context.Context, xr *composite.Unstructur
 			return CompositionResult{}, errors.Wrapf(err, errFmtRunPipelineStep, fn.Step)
 		}
 
+		// If this Function specified a non-zero TTL that's less than
+		// the current recorded TTL for the pipeline, it's the new TTL
+		// for the pipeline.
+		if d := rsp.GetMeta().GetTtl().AsDuration(); d > 0 && (ttl == 0 || d < ttl) {
+			ttl = d
+		}
+
 		// Pass the desired state returned by this Function to the next one.
 		d = rsp.GetDesired()
 
@@ -426,16 +439,6 @@ func (c *FunctionComposer) Compose(ctx context.Context, xr *composite.Unstructur
 		}
 	}
 
-	compositeRes := CompositeResource{}
-
-	// Consider the explicit composite unready state in the function response:
-	switch d.GetComposite().GetReady() { //nolint:exhaustive // only check for false or true
-	case fnv1.Ready_READY_TRUE:
-		compositeRes.Ready = ptr.To(true)
-	case fnv1.Ready_READY_FALSE:
-		compositeRes.Ready = ptr.To(false)
-	}
-
 	// Garbage collect any observed resources that aren't part of our final
 	// desired state. We must do this before we update the XR's resource
 	// references to ensure that we don't forget and leak them if a delete
@@ -550,7 +553,26 @@ func (c *FunctionComposer) Compose(ctx context.Context, xr *composite.Unstructur
 		return CompositionResult{}, errors.Wrap(err, errApplyXRStatus)
 	}
 
-	return CompositionResult{ConnectionDetails: d.GetComposite().GetConnectionDetails(), Composite: compositeRes, Composed: resources, Events: events, Conditions: conditions}, nil
+	var ready *bool
+	switch d.GetComposite().GetReady() {
+	case fnv1.Ready_READY_TRUE:
+		ready = ptr.To(true)
+	case fnv1.Ready_READY_FALSE:
+		ready = ptr.To(false)
+	case fnv1.Ready_READY_UNSPECIFIED:
+		// Remains nil.
+	}
+
+	result := CompositionResult{
+		Composed:          resources,
+		ConnectionDetails: d.GetComposite().GetConnectionDetails(),
+		Ready:             ready,
+		Events:            events,
+		Conditions:        conditions,
+		TTL:               ttl,
+	}
+
+	return result, nil
 }
 
 // ComposedFieldOwnerName generates a unique field owner name

internal/controller/apiextensions/composite/composed.go

@@ -19,11 +19,13 @@ package composite
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/testing/protocmp"
+	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/structpb"
 	corev1 "k8s.io/api/core/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
@@ -698,6 +700,9 @@ func TestFunctionCompose(t *testing.T) {
 				},
 				r: FunctionRunnerFn(func(_ context.Context, _ string, _ *fnv1.RunFunctionRequest) (*fnv1.RunFunctionResponse, error) {
 					rsp := &fnv1.RunFunctionResponse{
+						Meta: &fnv1.ResponseMeta{
+							Ttl: durationpb.New(5 * time.Minute),
+						},
 						Desired: &fnv1.State{
 							Composite: &fnv1.Resource{
 								Resource: MustStruct(map[string]any{
@@ -890,6 +895,7 @@ func TestFunctionCompose(t *testing.T) {
 							Target: CompositionTargetCompositeAndClaim,
 						},
 					},
+					TTL: 5 * time.Minute,
 				},
 				err: nil,
 			},