From e03d12a14ce6c8585277bb4af0dcd51b5c1f414e Mon Sep 17 00:00:00 2001 From: Jay Pipes Date: Wed, 21 Oct 2026 10:04:54 -0400 Subject: [PATCH] integrate kube-temporal base reconciler library Updates the `controller/namespace` package to use the `temporalio/kube-temporal/pkg/reconciler/base.Reconciler` struct and adds a resource-operator `pkg/reconciler/base.Reconciler` that builds on top of that and adds shared Temporal Cloud connection handling. Left the implementation of the `types.Resource.ResourceSetLatest`, `types.Resource.ResourceDelete` and `types.Resource.ResourceSync` as TODOs for Zain to tackle but left example unit tests and testutil infrastructure to make the common unit testing easier. Issue #7 Signed-off-by: Jay Pipes --- .mk/test.mk | 2 +- api/cloud/common/const.go | 6 + api/cloud/v1alpha1/connection.go | 19 ++ api/cloud/v1alpha1/credentials.go | 11 + cmd/resource-operator/main.go | 9 +- controller/namespace/controller.go | 57 ++-- controller/namespace/reconcile.go | 293 ++++++++-------- controller/namespace/reconcile_test.go | 130 ++++++++ controller/namespace/resource.go | 28 +- go.mod | 2 +- go.sum | 4 +- pkg/reconciler/base/reconciler.go | 185 +++++++++++ pkg/reconciler/base/reconciler_test.go | 212 ++++++++++++ pkg/temporal/cloud/connection/connection.go | 22 +- .../cloud/connection/connection_test.go | 47 ++- .../cloud/connection/{ => loader}/loader.go | 45 ++- .../connection/{ => loader}/loader_test.go | 131 ++++---- .../cloud/connection/manager/manager.go | 314 ++++++++++++++++++ .../cloud/connection/manager/manager_test.go | 287 ++++++++++++++++ pkg/testutil/kube.go | 27 ++ pkg/testutil/tcloud.go | 45 +++ 21 files changed, 1536 insertions(+), 340 deletions(-) create mode 100644 api/cloud/common/const.go create mode 100644 controller/namespace/reconcile_test.go create mode 100644 pkg/reconciler/base/reconciler.go create mode 100644 pkg/reconciler/base/reconciler_test.go rename pkg/temporal/cloud/connection/{ => loader}/loader.go (84%) rename pkg/temporal/cloud/connection/{ => loader}/loader_test.go (83%) create mode 100644 pkg/temporal/cloud/connection/manager/manager.go create mode 100644 pkg/temporal/cloud/connection/manager/manager_test.go create mode 100644 pkg/testutil/kube.go create mode 100644 pkg/testutil/tcloud.go diff --git a/.mk/test.mk b/.mk/test.mk index 781e6a2..004509c 100644 --- a/.mk/test.mk +++ b/.mk/test.mk @@ -30,7 +30,7 @@ test: test-unit ## Run all tests. .PHONY: test-unit test-unit: ## Run all unit tests. - @go test -v ./... + @go test -v ./... -coverprofile=cover.out .PHONY: test-cluster-delete test-cluster-delete: $(KIND) ## Delete the testing cluster. diff --git a/api/cloud/common/const.go b/api/cloud/common/const.go new file mode 100644 index 0000000..c81a72b --- /dev/null +++ b/api/cloud/common/const.go @@ -0,0 +1,6 @@ +package common + +const ( + AnnotationKeyPrefixTemporalCloud = "cloud.temporal.io/" + AnnotationKeyTemporalCloudHostPort = AnnotationKeyPrefixTemporalCloud + "hostport" +) diff --git a/api/cloud/v1alpha1/connection.go b/api/cloud/v1alpha1/connection.go index 1e60faf..a89ae12 100644 --- a/api/cloud/v1alpha1/connection.go +++ b/api/cloud/v1alpha1/connection.go @@ -1,10 +1,16 @@ package v1alpha1 import ( + "crypto/md5" + tcorev1 "github.com/temporalio/kube-temporal/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) +const ( + DefaultConnectionHostPort = "saas-api.tmprl.cloud:443" +) + // ConnectionSpec describes the desired configuration of a Connection to a // Temporal Cloud API server. It describes where credentials can be found for // instantiating a Temporal Cloud client object to that Temporal Cloud API @@ -31,6 +37,19 @@ type ConnectionSpec struct { Credentials CredentialsConfig `json:"credentials"` } +// Hash returns a deterministic MD5 hash of the ConnectionSpec's field +// components. +func (s ConnectionSpec) Hash() []byte { + out := md5.New() + out.Write([]byte{'|'}) + out.Write([]byte(s.HostPort)) + out.Write([]byte{'|'}) + out.Write(s.TLS.Hash()) + out.Write([]byte{'|'}) + out.Write(s.Credentials.Hash()) + return out.Sum(nil) +} + // ConnectionStatus defines the observed state of a Connection type ConnectionStatus struct { tcorev1.StatusBase `json:",inline"` diff --git a/api/cloud/v1alpha1/credentials.go b/api/cloud/v1alpha1/credentials.go index f34454a..bc5cf9e 100644 --- a/api/cloud/v1alpha1/credentials.go +++ b/api/cloud/v1alpha1/credentials.go @@ -1,6 +1,8 @@ package v1alpha1 import ( + "crypto/md5" + tcorev1 "github.com/temporalio/kube-temporal/api/core/v1" ) @@ -11,3 +13,12 @@ type CredentialsConfig struct { // API server using Temporal API Key authentication. APIKey tcorev1.APIKeyCredentialsConfig `json:"apiKey"` } + +// Hash returns a deterministic MD5 hash of the CredentialsConfig's field +// components. +func (c CredentialsConfig) Hash() []byte { + out := md5.New() + out.Write([]byte{'|'}) + out.Write(c.APIKey.Hash()) + return out.Sum(nil) +} diff --git a/cmd/resource-operator/main.go b/cmd/resource-operator/main.go index 6d55575..7eac77b 100644 --- a/cmd/resource-operator/main.go +++ b/cmd/resource-operator/main.go @@ -15,7 +15,7 @@ import ( ctrlrtlog "sigs.k8s.io/controller-runtime/pkg/log" "github.com/temporalio/kube-temporal/pkg/controller/config" - cloudv1alpha1 "github.com/temporalio/resource-operator/api/cloud/v1alpha1" + tcloudv1alpha1 "github.com/temporalio/resource-operator/api/cloud/v1alpha1" ctrlnamespace "github.com/temporalio/resource-operator/controller/namespace" "github.com/temporalio/resource-operator/pkg/version" ) @@ -31,7 +31,7 @@ var ( ) func init() { - _ = cloudv1alpha1.AddToScheme(scheme) + _ = tcloudv1alpha1.AddToScheme(scheme) } func main() { @@ -70,12 +70,13 @@ func main() { rootLogger := slog.New(logr.ToSlogHandler(ctrlrtlog.Log)) setupLog.Info("initializing controller") - cc := ctrlnamespace.New(cfg, rootLogger) + c := ctrlnamespace.New(cfg, rootLogger) - if err = cc.BindManager(mgr); err != nil { + if err = c.BindManager(mgr); err != nil { setupLog.Error( err, "failed binding controller to controller manager", + "controller", c.Name(), ) os.Exit(1) } diff --git a/controller/namespace/controller.go b/controller/namespace/controller.go index 95bbee5..8dd3971 100644 --- a/controller/namespace/controller.go +++ b/controller/namespace/controller.go @@ -3,13 +3,9 @@ package namespace import ( "log/slog" - "github.com/temporalio/kube-temporal/pkg/controller/base" + "github.com/temporalio/kube-temporal/pkg/controller" "github.com/temporalio/kube-temporal/pkg/controller/config" - "github.com/temporalio/kube-temporal/pkg/types" - ctrlrt "sigs.k8s.io/controller-runtime" - "sigs.k8s.io/controller-runtime/pkg/predicate" - - cloudv1alpha1 "github.com/temporalio/resource-operator/api/cloud/v1alpha1" + tcloudv1alpha1 "github.com/temporalio/resource-operator/api/cloud/v1alpha1" ) // +kubebuilder:rbac:groups=cloud.temporal.io,resources=namespaces,verbs=get;list;watch;create;update;patch;delete @@ -17,33 +13,36 @@ import ( // +kubebuilder:rbac:groups=cloud.temporal.io,resources=namespaces/finalizers,verbs=update const ( - controllerName = "namespace" + ControllerName = "namespace" ) -type NewOption func(*controller) - -// New returns a new [types.Controller] that handles [cloudv1alpha1.Namespace] -// resources. +// New returns a new [controller.Controller] that handles +// [cloudv1alpha1.Namespace] resources. func New( - cfg *config.Config, + cfg config.Config, logger *slog.Logger, -) types.Controller { - return &controller{ - Controller: base.New(cfg, logger), - } +) *controller.Controller { + return controller.New( + ControllerName, + cfg, + NewReconciler(), + controller.WithLogger(logger), + ) } -type controller struct { - base.Controller -} - -// BindManager binds the Controller to the supplied [ctrlrt.Manager]. -func (c *controller) BindManager(m ctrlrt.Manager) error { - c.Controller.Client = m.GetClient() - c.Controller.Reader = m.GetAPIReader() - builder := ctrlrt.NewControllerManagedBy(m). - Named(controllerName). - For(&cloudv1alpha1.Namespace{}). - WithEventFilter(predicate.GenerationChangedPredicate{}) - return builder.Complete(c) +// NewWithDefaultCloudConnectionSpec returns a new [controller.Controller] that +// handles [tcloudv1alpha1.Namespace] resources and passes in a +// [tcloudv1alpha1.ConnectionSpec] to instruct the reconciler how to connect to +// a Temporal Cloud API endpoint. +func NewWithDefaultCloudConnectionSpec( + cfg config.Config, + logger *slog.Logger, + spec tcloudv1alpha1.ConnectionSpec, +) *controller.Controller { + return controller.New( + ControllerName, + cfg, + NewReconciler(WithDefaultCloudConnectionSpec(spec)), + controller.WithLogger(logger), + ) } diff --git a/controller/namespace/reconcile.go b/controller/namespace/reconcile.go index 67535cc..4f59682 100644 --- a/controller/namespace/reconcile.go +++ b/controller/namespace/reconcile.go @@ -2,209 +2,188 @@ package namespace import ( "context" - "errors" "fmt" - "github.com/samber/lo" - "github.com/temporalio/kube-temporal/pkg/condition" - ctrlerrors "github.com/temporalio/kube-temporal/pkg/controller/errors" ctrlreqlog "github.com/temporalio/kube-temporal/pkg/controller/request/log" - "github.com/temporalio/kube-temporal/pkg/controller/requeue" "github.com/temporalio/kube-temporal/pkg/types" - apierrors "k8s.io/apimachinery/pkg/api/errors" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + tcclient "go.temporal.io/cloud-sdk/cloudclient" ctrlrt "sigs.k8s.io/controller-runtime" - "sigs.k8s.io/controller-runtime/pkg/client" + kclient "sigs.k8s.io/controller-runtime/pkg/client" + + tcloudv1alpha1 "github.com/temporalio/resource-operator/api/cloud/v1alpha1" + "github.com/temporalio/resource-operator/pkg/reconciler/base" ) -// Reconcile performs a full reconciliation for the object referred to by the -// Request. -// -// If the returned error is non-nil, the Result is ignored and the request will -// be requeued using exponential backoff. The only exception is if the error is -// a TerminalError in which case no requeuing happens. -// -// If the error is nil and the returned Result has a non-zero -// result.RequeueAfter, the request will be requeued after the specified -// duration. -// -// If the error is nil and result.RequeueAfter is zero and result.Requeue is -// true, the request will be requeued using exponential backoff. -func (c *controller) Reconcile( - ctx context.Context, - req ctrlrt.Request, -) (ctrlrt.Result, error) { - desired, err := c.resourceFromRequest(ctx, req) - if err != nil { - if apierrors.IsNotFound(err) { - // resource wasn't found. just ignore these. - return ctrlrt.Result{}, nil - } - return ctrlrt.Result{}, err +// NewrOption modifies the Reconciler returned from NewReconciler. +type NewReconcilerOption func(*Reconciler) + +// WithDefaultCloudConnectionSpec registers the supplied +// [tcloudv1alpha1.ConnectionSpec] object as the default cloud connection spec +// for the Reconciler. +func WithDefaultCloudConnectionSpec( + spec tcloudv1alpha1.ConnectionSpec, +) NewReconcilerOption { + return func(r *Reconciler) { + r.SetDefaultCloudConnectionSpec(spec) } +} - ctx = ctrlreqlog.ToContext(ctx, c.Logger(), desired) +// WithDefaultCloudClient registers the supplied [tcclient.Clinet] as the +// default cloud client fallback for the Reconciler. +func WithDefaultCloudClient(tcc tcclient.Client) NewReconcilerOption { + return func(r *Reconciler) { + r.SetDefaultCloudClient(tcc) + } +} - latest, err := c.reconcile(ctx, desired) - return c.handleReconcileError(ctx, desired, latest, err) +// WithKubeClient sets the supplied Kubernetes client for the Reconciler. +func WithKubeClient(kc kclient.Client) NewReconcilerOption { + return func(r *Reconciler) { + r.SetKubeClient(kc) + } } -// reconcile attempts to make the latest observed state of the resource match -// the latest desired state. -// -// It returns a copy of the resource that represents the latest observed state. -func (c *controller) reconcile( +// NewReconciler returns a new [types.Reconciler] that can Reconciler +// [ttcloudv1alpha1.Namespace] resources. +func NewReconciler(opts ...NewReconcilerOption) *Reconciler { + r := Reconciler{} + for _, opt := range opts { + opt(&r) + } + return &r +} + +// Reconciler wraps the base Reconciler and implements the [types.Reconciler] +// interface. +type Reconciler struct { + base.Reconciler +} + +// AdaptBuilder returns a [*ctrlrt.Builder] adapted from the supplied +// [*ctrlrt.Builder], typically calling the [ctrlrt.Builder.For] method with +// the Reconciler's managed Kind prototype. +func (r *Reconciler) AdaptBuilder(builder *ctrlrt.Builder) *ctrlrt.Builder { + return builder.For(&tcloudv1alpha1.Namespace{}) +} + +// ResourceFromRequest returns a Resource representing the requested Kubernetes +// namespaced object. +func (r *Reconciler) ResourceFromRequest( ctx context.Context, - desired types.Resource, + req ctrlrt.Request, ) (types.Resource, error) { - if desired.IsBeingDeleted() { - return c.resourceDelete(ctx, desired) - } - latest, err := c.resourceSync(ctx, desired) + cr := &tcloudv1alpha1.Namespace{} + err := r.KubeReadLatestInto(ctx, req.NamespacedName, cr) if err != nil { - return latest, err + return nil, fmt.Errorf("failed reading latest resource: %w", err) } - return latest, nil + return &res{cr}, nil } -// resourceSync ensures that the supplied Resource's backing Temporal Cloud -// Namespace matches the supplied desired state. +// ResourceSetLatest mutates the supplied [types.Resource] with the latest +// observed state. // -// It returns a copy of the resource that represents the latest observed state. -func (c *controller) resourceSync( +// Any error returned from this function is considered a runtime error and will +// cause the resource to contain a Stalled Condition with Status=True. +func (r *Reconciler) ResourceSetLatest( ctx context.Context, - desired types.Resource, -) (types.Resource, error) { + latest types.Resource, // mutated by this method +) error { var err error log := ctrlreqlog.FromContext(ctx) - exit := log.Trace("resource.sync") + exit := log.Trace("resource.set.latest") defer func() { exit(err) }() - var latest types.Resource // the newly created or mutated resource - - // We clear the resource's Conditions collection at the start of each - // reconcile loop to ensure that the Status.Conditions always contains the - // freshest indication of current state. - condition.Clear(desired) - defer func() { - c.conditionsSync(ctx, latest, err) - }() - return latest, nil + tcc, err := r.CloudClientFor(ctx, latest) + if err != nil { + return fmt.Errorf("failed getting tcloud client: %w", err) + } + _ = tcc + + // "pop" the underlying *tcloudv1alpha1.Namespace from the [types.Resource] + // wrapper. This is a pointer so by modifying it, you mutate the supplied + // `latest` parameter. + ns := latest.(*res).cr + _ = ns + + // TODO: call the Temporal Cloud API to fetch Namespace spec and status and + // mutate the supplied resource to set the spec and status fields + // accordingly. + // + // latestObserved, err := tcc.CloudService().GetNamespace(...) + // if err != nil { + // return fmt.Errorf("failed getting tcloud namespace: %w", err) + // } + // ns.Spec.ProjectID = latestObserved.ProjectId + // ... + + return err } -// resourceDelete ensures that the supplied Resource's backing Temporal Cluster -// is destroyed along with all child dependent resources. +// ResourceSyncFunc synchronizes any managed resources to match the supplied +// desired [types.Resource] state. +// +// Returns nil when the synchronization was successful, otherwise returns an +// error derived from `kube-tempoal/pkg/controller/errors.ErrResourceSync`. // -// Returns a copy of the resource with the latest state either right before -// deletion OR after a failed attempted deletion. -func (c *controller) resourceDelete( +// `desired` should NEVER be mutated by this callback. +// `latest` should NEVER be mutated by this callback. +// +// This method is only called when there was a difference detected between +// the desired and latest [types.Resource]. +func (r *Reconciler) ResourceSync( ctx context.Context, - current types.Resource, -) (types.Resource, error) { + desired types.Resource, + latest types.Resource, +) error { var err error log := ctrlreqlog.FromContext(ctx) - exit := log.Trace("resource.delete") + exit := log.Trace("resource.sync") defer func() { exit(err) }() - return nil, nil -} -// conditionsSync examines the supplied resource's collection of Condition -// objects and ensures that appropriate Ready and Terminal Conditions are -// present. -func (c *controller) conditionsSync( - ctx context.Context, - res types.Resource, - reconcileErr error, -) { - if lo.IsNil(res) { - return + tcc, err := r.CloudClientFor(ctx, desired) + if err != nil { + return fmt.Errorf("failed getting tcloud client: %w", err) } + _ = tcc - var err error - log := ctrlreqlog.FromContext(ctx) - exit := log.Trace("conditions.sync") - defer func() { - exit(err) - }() + // TODO: call the Temporal Cloud API to update Namespace desired + // configuration to match the desired resource. - if errors.Is(reconcileErr, &ctrlerrors.TerminalError{}) { - // If we got a terminal error while reconciling the resource, we set - // its Stalled condition to True to indicate that another reconcile - // will not change the status of the resource without a change to its - // desired state. - condition.SetStalled( - res, - metav1.ConditionFalse, - "TerminalError", - fmt.Sprintf( - "reconciliation encountered a terminal error: %s", - reconcileErr.Error(), - ), - ) - } else { - condition.SetStalled( - res, - metav1.ConditionFalse, - "ReconcileSucceeded", - "reconcile succeeded", - ) - } + return err } -// handleReconcileError will handle errors from reconcile handlers, which -// respects runtime errors. -// -// If the `latest` parameter is not nil, this function will ALWAYS patch the -// latest Status fields back to the Kubernetes API. -func (c *controller) handleReconcileError( +// ResourceDelete deletes any managed external resources for the supplied +// [types.Resource]. Note that the Reconciler does NOT need to remove the +// finalizer on the associated [client.Object]. The Controller handles that. +// All the Reconciler needs to do is delete any owned resources or +// externally-managed resources associated with the supplied [types.Resource]. +func (r *Reconciler) ResourceDelete( ctx context.Context, desired types.Resource, - latest types.Resource, - err error, -) (ctrlrt.Result, error) { - if lo.IsNotNil(latest) { - // The reconciliation loop may have returned an error, but if latest is - // not nil, there may be some changes available in the CR's Status - // struct (example: Conditions), and we want to make sure we save those - // changes before proceeding. - // - // It is okay to patch status when resource is not present due to deletion - // because a NotFound error is thrown which will be ignored. - _ = c.patchResourceStatus(ctx, desired, latest) - } - if err == nil || errors.Is(err, &ctrlerrors.TerminalError{}) { - // We don't requeue on a terminal error... - return ctrlrt.Result{}, nil - } +) error { + var err error + log := ctrlreqlog.FromContext(ctx) + exit := log.Trace("resource.delete") + defer func() { + exit(err) + }() - var requeueNeededAfter *requeue.RequeueNeededAfter - if errors.As(err, &requeueNeededAfter) { - after := requeueNeededAfter.Duration() - return ctrlrt.Result{RequeueAfter: after}, nil + tcc, err := r.CloudClientFor(ctx, desired) + if err != nil { + return fmt.Errorf("failed getting tcloud client: %w", err) } + _ = tcc - var requeueNeeded *requeue.RequeueNeeded - if errors.As(err, &requeueNeeded) { - return ctrlrt.Result{Requeue: true}, nil - } + // TODO: call the Temporal Cloud API to delete the Namespace. - return ctrlrt.Result{}, err + return err } -// patchResourceStatus patches the custom resource in the Kubernetes API to -// match the supplied latest resource. -func (c *controller) patchResourceStatus( - ctx context.Context, - desired types.Resource, - latest types.Resource, -) error { - dobj := desired.DeepCopy().ClientObject() - lobj := latest.DeepCopy().ClientObject() - patch := client.MergeFrom(dobj) - - return c.Client.Status().Patch(ctx, lobj, patch) -} +// compile-time check that Reconciler satisfies the [types.Reconciler] +// contract. +var _ types.Reconciler = (*Reconciler)(nil) diff --git a/controller/namespace/reconcile_test.go b/controller/namespace/reconcile_test.go new file mode 100644 index 0000000..e58996f --- /dev/null +++ b/controller/namespace/reconcile_test.go @@ -0,0 +1,130 @@ +package namespace_test + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + kttestutil "github.com/temporalio/kube-temporal/pkg/testutil" + "github.com/temporalio/kube-temporal/pkg/types" + tcloudv1alpha1 "github.com/temporalio/resource-operator/api/cloud/v1alpha1" + "github.com/temporalio/resource-operator/controller/namespace" + "github.com/temporalio/resource-operator/pkg/testutil" + ctrlrt "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +const ( + kubeNamespace = "kube-temporal" +) + +func TestReconciler_ResourceFromRequest(t *testing.T) { + tnsName := "tmprl-ns" + tns := &tcloudv1alpha1.Namespace{} + tns.Name = tnsName + tns.Namespace = kubeNamespace + + cases := []struct { + name string + client client.Client + req ctrlrt.Request + exp types.Resource + expErr string + }{ + { + name: "fail if no client", + client: func() client.Client { + return nil + }(), + expErr: "nil kube client", + }, + { + name: "fail if resource not found", + client: func() client.Client { + return testutil.NewFakeClient() + }(), + req: kttestutil.NewResourceRequest(tnsName, kubeNamespace), + expErr: "namespaces.cloud.temporal.io \"kube-temporal\" not found", + }, + { + name: "yields resource", + client: func() client.Client { + return testutil.NewFakeClient(tns) + }(), + req: kttestutil.NewResourceRequest(kubeNamespace, tnsName), + exp: namespace.NewResource(tns), + }, + } + for _, tc := range cases { + t.Run(tc.name, func(tt *testing.T) { + require := require.New(tt) + + kc := tc.client + r := namespace.NewReconciler() + r.SetKubeClient(kc) + r.SetKubeReader(kc) + + ctx := context.TODO() + + got, err := r.ResourceFromRequest(ctx, tc.req) + if tc.expErr != "" { + require.ErrorContains(err, tc.expErr) + return + } + + require.Nil(err) + if tc.exp != nil { + require.EqualExportedValues(tc.exp, got) + } + }) + } +} + +func TestReconciler_ResourceSetLatest(t *testing.T) { + tnsName := "tmprl-ns" + tns := &tcloudv1alpha1.Namespace{} + tns.Name = tnsName + tns.Namespace = kubeNamespace + + tcc, err := testutil.NewFakeCloudClient() + require.Nil(t, err) + + cases := []struct { + name string + r *namespace.Reconciler + res types.Resource + expErr string + }{ + { + name: "fail nil cloud connection manager or default client", + r: namespace.NewReconciler(), + res: namespace.NewResource(tns), + expErr: "connection manager nil and no default client set", + }, + { + name: "mutates latest resource", + r: namespace.NewReconciler( + namespace.WithKubeClient( + testutil.NewFakeClient(tns), + ), + namespace.WithDefaultCloudClient(*tcc), + ), + res: namespace.NewResource(tns), + }, + } + for _, tc := range cases { + t.Run(tc.name, func(tt *testing.T) { + require := require.New(tt) + + ctx := context.TODO() + + err := tc.r.ResourceSetLatest(ctx, tc.res) + if tc.expErr != "" { + require.ErrorContains(err, tc.expErr) + return + } + + require.Nil(err) + }) + } +} diff --git a/controller/namespace/resource.go b/controller/namespace/resource.go index 5e02d59..1ca0868 100644 --- a/controller/namespace/resource.go +++ b/controller/namespace/resource.go @@ -1,7 +1,6 @@ package namespace import ( - "context" "log/slog" "github.com/temporalio/kube-temporal/pkg/kube/transform" @@ -11,31 +10,16 @@ import ( "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" - ctrlrt "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" - cloudv1alpha1 "github.com/temporalio/resource-operator/api/cloud/v1alpha1" + tcloudv1alpha1 "github.com/temporalio/resource-operator/api/cloud/v1alpha1" ) -// resourceFromRequest returns a Resource representing the requested Kubernetes -// namespaced object. -func (c *controller) resourceFromRequest( - ctx context.Context, - req ctrlrt.Request, -) (types.Resource, error) { - cr := &cloudv1alpha1.Namespace{} - // We use APIReader in order to prevent stale reads. - if err := c.Reader.Get(ctx, req.NamespacedName, cr); err != nil { - return nil, err - } - return &res{cr}, nil -} - // res wraps a Namespace CR and implements the types.Resource // interface. type res struct { // cr is the Kubernetes-native custom resource. - cr *cloudv1alpha1.Namespace + cr *tcloudv1alpha1.Namespace } // IsBeingDeleted returns true if the Kubernetes resource has a non-zero @@ -59,7 +43,7 @@ func (r res) RuntimeObject() runtime.Object { // GroupVersionKind returns the GroupVersionKind that the Resource // represents. func (r res) GroupVersionKind() schema.GroupVersionKind { - return cloudv1alpha1.SchemeGroupVersion.WithKind("Namespace") + return tcloudv1alpha1.SchemeGroupVersion.WithKind("Namespace") } // Conditions returns the Resource's collection of Conditions from its status @@ -90,4 +74,8 @@ func (r *res) LogValue() slog.Value { return resource.LogValue(r) } -var _ types.Resource = (*res)(nil) +// NewResource returns the [types.Resource] wrapping the supplied +// [tcloudv1alpha1.Namespace] object. +func NewResource(cr *tcloudv1alpha1.Namespace) types.Resource { + return &res{cr} +} diff --git a/go.mod b/go.mod index cd7de3c..d09f514 100644 --- a/go.mod +++ b/go.mod @@ -7,7 +7,7 @@ require ( github.com/samber/lo v1.53.0 github.com/spf13/pflag v1.0.10 github.com/stretchr/testify v1.12.1 - github.com/temporalio/kube-temporal v0.3.0 + github.com/temporalio/kube-temporal v0.4.0 go.temporal.io/cloud-sdk v0.7.1 k8s.io/api v0.37.1 k8s.io/apimachinery v0.37.1 diff --git a/go.sum b/go.sum index 93f66a2..2a8ac40 100644 --- a/go.sum +++ b/go.sum @@ -136,8 +136,8 @@ github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+Q github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= -github.com/temporalio/kube-temporal v0.3.0 h1:9RFd81gh99kenk6cTwifbL+oyqpky0+QGwfkY05fatM= -github.com/temporalio/kube-temporal v0.3.0/go.mod h1:gJ+9ZxerlGn29Nn6VqAAXVeH73P31JuLUe3bQ+w4nwk= +github.com/temporalio/kube-temporal v0.4.0 h1:dvRaPoV6brzViW1QYG+1mPt5X/RCTTFn+ZjDgfpVZPs= +github.com/temporalio/kube-temporal v0.4.0/go.mod h1:QR80csThD/3jzUo2NS5wRoHReCQfEwTXUJiVEPP8Pbg= github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= diff --git a/pkg/reconciler/base/reconciler.go b/pkg/reconciler/base/reconciler.go new file mode 100644 index 0000000..48321c7 --- /dev/null +++ b/pkg/reconciler/base/reconciler.go @@ -0,0 +1,185 @@ +package base + +import ( + "context" + "fmt" + + ctrlreqlog "github.com/temporalio/kube-temporal/pkg/controller/request/log" + ktbase "github.com/temporalio/kube-temporal/pkg/reconciler/base" + "github.com/temporalio/kube-temporal/pkg/types" + tcloudservicev1 "go.temporal.io/cloud-sdk/api/cloudservice/v1" + tcclient "go.temporal.io/cloud-sdk/cloudclient" + "k8s.io/apimachinery/pkg/api/meta" + kclient "sigs.k8s.io/controller-runtime/pkg/client" + + tcloudcommon "github.com/temporalio/resource-operator/api/cloud/common" + tcloudv1alpha1 "github.com/temporalio/resource-operator/api/cloud/v1alpha1" + conmgr "github.com/temporalio/resource-operator/pkg/temporal/cloud/connection/manager" +) + +// NewOption modifies the Reconciler returned from New. +type NewOption func(*Reconciler) + +// WithDefaultCloudConnectionSpec registers the supplied +// [tcloudv1alpha1.ConnectionSpec] object as the default cloud connection spec +// for the Reconciler. +func WithDefaultCloudConnectionSpec( + spec tcloudv1alpha1.ConnectionSpec, +) NewOption { + return func(r *Reconciler) { + r.defaultCloudConnSpec = &spec + ccm := r.ccManager + if ccm == nil { + ccm = conmgr.New() + r.ccManager = ccm + } + // NOTE(jaypipes): Error-returning code paths in manager.Register are + // not hittable here, so safe to ignore errors. + _ = ccm.Register(&spec) + } +} + +// WithDefaultCloudClient registers the supplied [tcclient.Client] as the +// fallback default for the Reconciler. +func WithDefaultCloudClient(tcc tcclient.Client) NewOption { + return func(r *Reconciler) { + r.defaultCloudClient = &tcc + } +} + +// WithKubeClient sets the supplied Kubernetes client for the Reconciler. +func WithKubeClient(kc kclient.Client) NewOption { + return func(r *Reconciler) { + r.Reconciler.SetKubeClient(kc) + // Propogate our kube client down to the cloud connection manager + ccm := r.ccManager + if ccm == nil { + ccm = conmgr.New() + r.ccManager = ccm + } + ccm.SetKubeClient(kc) + } +} + +// New returns a new base Reconciler. +func New(opts ...NewOption) *Reconciler { + r := Reconciler{} + for _, opt := range opts { + opt(&r) + } + return &r +} + +// Reconciler is the base struct for all controller-specific reconcilers in +// resource-operator. It contains common methods that all controller-specific +// reconcilers need, for example, to connect to a Temporal Cloud API endpoint. +type Reconciler struct { + ktbase.Reconciler + ccManager *conmgr.Manager + + // defaultCloudConnSpec contains a [tcloudv1alpha1.ConnectionSpec] + // configured as a default for the Reconciler. + defaultCloudConnSpec *tcloudv1alpha1.ConnectionSpec + + // defaultCloudClient contains the [tcclient.Client] configured as the + // default fallback for the Reconciler. + defaultCloudClient *tcclient.Client +} + +// SetDefaultCloudConnectionSpec sets the [tcloudv1alpha1.ConnectionSpec] the +// Reconciler should use as a fallback default when fetching a Temporal Cloud +// client. +func (r *Reconciler) SetDefaultCloudConnectionSpec( + spec tcloudv1alpha1.ConnectionSpec, +) { + r.defaultCloudConnSpec = &spec +} + +// SetDefaultCloudClient sets the [tcclient.Client] the Reconciler should use +// as a fallback default. +func (r *Reconciler) SetDefaultCloudClient( + tcc tcclient.Client, +) { + r.defaultCloudClient = &tcc +} + +// CloudConnectionManager returns the Reconciler's manager of Temporal Cloud +// connections and clients. +func (r Reconciler) CloudConnectionManager() *conmgr.Manager { + return r.ccManager +} + +// SetCloudConnectionManager sets the Reconciler's manager of Temporal Cloud +// connections and clients. +func (r *Reconciler) SetCloudConnectionManager(ccm *conmgr.Manager) { + r.ccManager = ccm +} + +// CloudClientFor returns a [tcclient.Client] struct configured for the +// matching supplied subject. +// +// If a client is not found for the supplied subject, we fallback to any +// default client set in the Reconciler. +// +// If subject is a [tcloudv1.ConnectionSpec], we look up the client by the +// connection spec's hash key. +// +// If subject is a string, we look up the client by the HostPort. If no client +// is found, we find the first connection spec having that HostPort and then +// look up the client by that connection spec's hash key. +// +// If subject is a [types.Resource], we look for the +// cloud.temporal.io/hostport annotation and look up the client by the +// annotation value. If no client is found, we find the first connection spec +// having that annotation value and then look up the client by that connection +// spec's hash key. +func (r *Reconciler) CloudClientFor( + ctx context.Context, + subject any, +) (tcloudservicev1.CloudServiceClient, error) { + if r.ccManager == nil && r.defaultCloudClient == nil { + return nil, fmt.Errorf("connection manager nil and no default client set") + } + + var err error + log := ctrlreqlog.FromContext(ctx) + exit := log.Trace("cloud.client.for") + defer func() { + exit(err) + }() + + if r.ccManager == nil { + log.Debug( + "no connection manager, falling back to default cloud client", + ) + return r.defaultCloudClient.CloudService(), nil + } + + var cli *tcclient.Client + + switch subject := subject.(type) { + case string, *tcloudv1alpha1.ConnectionSpec: + cli, err = r.ccManager.Client(subject) + case types.Resource: + ro := subject.RuntimeObject() + metaAccessor := meta.NewAccessor() + annos, err := metaAccessor.Annotations(ro) + if err != nil { + return nil, fmt.Errorf("failed getting resource annotations: %w", err) + } + hp, ok := annos[tcloudcommon.AnnotationKeyTemporalCloudHostPort] + if !ok { + return nil, fmt.Errorf( + "no %q annotation on resource", + tcloudcommon.AnnotationKeyTemporalCloudHostPort, + ) + } + cli, err = r.ccManager.Client(hp) + default: + return nil, fmt.Errorf("unsupported type %T", subject) + } + if err != nil { + return nil, err + } + return cli.CloudService(), nil +} diff --git a/pkg/reconciler/base/reconciler_test.go b/pkg/reconciler/base/reconciler_test.go new file mode 100644 index 0000000..c1301fe --- /dev/null +++ b/pkg/reconciler/base/reconciler_test.go @@ -0,0 +1,212 @@ +package base_test + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" + tcorev1 "github.com/temporalio/kube-temporal/api/core/v1" + tcloudcommon "github.com/temporalio/resource-operator/api/cloud/common" + tcloudv1alpha1 "github.com/temporalio/resource-operator/api/cloud/v1alpha1" + "github.com/temporalio/resource-operator/controller/namespace" + "github.com/temporalio/resource-operator/pkg/reconciler/base" + conmgr "github.com/temporalio/resource-operator/pkg/temporal/cloud/connection/manager" + "github.com/temporalio/resource-operator/pkg/testutil" + tcloudservicev1 "go.temporal.io/cloud-sdk/api/cloudservice/v1" + "k8s.io/apimachinery/pkg/api/meta" +) + +const ( + kubeNamespace = "kube-temporal" +) + +func TestReconciler_CloudClientFor(t *testing.T) { + + tnsName := "tmprl-ns" + tns := func() *tcloudv1alpha1.Namespace { + ns := &tcloudv1alpha1.Namespace{} + ns.Name = tnsName + ns.Namespace = kubeNamespace + return ns + } + + tcc, err := testutil.NewFakeCloudClient() + require.Nil(t, err) + + apiKeyDir := t.TempDir() + apiKeyPath := filepath.Join(apiKeyDir, "api-key") + + err = os.WriteFile( + apiKeyPath, + []byte(testutil.FakeTemporalCloudAPIKeyContents), + 0644, + ) + require.Nil(t, err) + + cases := []struct { + name string + r *base.Reconciler + subject any + exp tcloudservicev1.CloudServiceClient + expErr string + }{ + { + name: "fail nil cloud connection manager or default client", + r: base.New(), + subject: namespace.NewResource(tns()), + expErr: "connection manager nil and no default client set", + }, + { + name: "yields default client when no manager", + r: base.New( + base.WithDefaultCloudClient(*tcc), + ), + subject: namespace.NewResource(tns()), + exp: tcc.CloudService(), + }, + { + name: "fail if subject not resource, string or conn spec", + r: base.New( + base.WithKubeClient( + testutil.NewFakeClient(tns()), + ), + ), + subject: int64(1), + expErr: "unsupported type int64", + }, + { + name: "fail if spec subject not in manager", + r: base.New( + base.WithDefaultCloudConnectionSpec( + tcloudv1alpha1.ConnectionSpec{ + HostPort: "host:port", + }, + ), + ), + subject: &tcloudv1alpha1.ConnectionSpec{ + HostPort: "differenthost:port", + Credentials: tcloudv1alpha1.CredentialsConfig{ + APIKey: tcorev1.APIKeyCredentialsConfig{ + Path: apiKeyPath, + }, + }, + }, + expErr: "no cloud client registered for spec", + }, + { + name: "spec subject yields registered client", + r: func() *base.Reconciler { + b := base.New( + base.WithKubeClient( + testutil.NewFakeClient(tns()), + ), + ) + ccm := b.CloudConnectionManager() + _ = ccm.Register( + &tcloudv1alpha1.ConnectionSpec{ + HostPort: "host:port", + Credentials: tcloudv1alpha1.CredentialsConfig{ + APIKey: tcorev1.APIKeyCredentialsConfig{ + Path: apiKeyPath, + }, + }, + }, + conmgr.RegisterWithClient(tcc), + ) + return b + }(), + subject: &tcloudv1alpha1.ConnectionSpec{ + HostPort: "host:port", + Credentials: tcloudv1alpha1.CredentialsConfig{ + APIKey: tcorev1.APIKeyCredentialsConfig{ + Path: apiKeyPath, + }, + }, + }, + exp: tcc.CloudService(), + }, + { + name: "string subject yields registered client", + r: func() *base.Reconciler { + b := base.New( + base.WithKubeClient( + testutil.NewFakeClient(tns()), + ), + ) + ccm := b.CloudConnectionManager() + _ = ccm.Register( + "host:port", + conmgr.RegisterWithClient(tcc), + ) + return b + }(), + subject: "host:port", + exp: tcc.CloudService(), + }, + { + name: "fail if resource subject missing hostport annotation", + r: func() *base.Reconciler { + b := base.New( + base.WithKubeClient( + testutil.NewFakeClient(tns()), + ), + ) + ccm := b.CloudConnectionManager() + _ = ccm.Register( + "host:port", + conmgr.RegisterWithClient(tcc), + ) + return b + }(), + subject: namespace.NewResource(tns()), + expErr: "no \"cloud.temporal.io/hostport\" annotation on resource", + }, + { + name: "resource subject with annotation yields registered client", + r: func() *base.Reconciler { + b := base.New( + base.WithKubeClient( + testutil.NewFakeClient(tns()), + ), + ) + ccm := b.CloudConnectionManager() + _ = ccm.Register( + "host:port", + conmgr.RegisterWithClient(tcc), + ) + return b + }(), + subject: func() any { + res := namespace.NewResource(tns()) + ro := res.RuntimeObject() + metaAccessor := meta.NewAccessor() + annos := map[string]string{ + tcloudcommon.AnnotationKeyTemporalCloudHostPort: "host:port", + } + metaAccessor.SetAnnotations(ro, annos) + return res + }(), + + exp: tcc.CloudService(), + }, + } + for _, tc := range cases { + t.Run(tc.name, func(tt *testing.T) { + require := require.New(tt) + + ctx := context.TODO() + + got, err := tc.r.CloudClientFor(ctx, tc.subject) + if tc.expErr != "" { + require.ErrorContains(err, tc.expErr) + return + } + + require.Nil(err) + require.NotNil(got) + require.EqualExportedValues(tc.exp, got) + }) + } +} diff --git a/pkg/temporal/cloud/connection/connection.go b/pkg/temporal/cloud/connection/connection.go index 1af4ae8..9c4a8c6 100644 --- a/pkg/temporal/cloud/connection/connection.go +++ b/pkg/temporal/cloud/connection/connection.go @@ -4,13 +4,15 @@ import ( "fmt" tcclient "go.temporal.io/cloud-sdk/cloudclient" + + "github.com/temporalio/resource-operator/pkg/temporal/cloud/connection/loader" ) // Connection wraps a Temporal client struct with an ability to load the client // from various sources, including a Kubernetes Secret object or local files. type Connection struct { tcc *tcclient.Client - loader *Loader + loader *loader.Loader } // Client returns the Temporal Cloud client associated with the Connection. @@ -22,7 +24,7 @@ func (c *Connection) Client() (*tcclient.Client, error) { return nil, fmt.Errorf("no Loader configured for Connection") } - clientOpts, err := c.loader.TemporalCloudClientOptions() + clientOpts, err := c.loader.ClientOptions() if err != nil { return nil, err } @@ -35,27 +37,23 @@ func (c *Connection) Client() (*tcclient.Client, error) { } // Loader returns the Loader associated with the Connection, if any. -func (c *Connection) Loader() *Loader { +func (c *Connection) Loader() *loader.Loader { return c.loader } -type newOptions struct { - loader *Loader -} - // NewOption allows modifying the returned Connection from New. type NewOption func(*Connection) // WithLoader modifies the Connection to use the supplied Loader. -func WithLoader(loader Loader) NewOption { +func WithLoader(loader loader.Loader) NewOption { return func(c *Connection) { c.loader = &loader } } -// WithTemporalCloudClient overrides the Connection's Temporal Cloud Client -// with the supplied object. -func WithTemporalCloudClient(tcc *tcclient.Client) NewOption { +// WithClient overrides the Connection's Temporal Cloud Client with the +// supplied object. +func WithClient(tcc *tcclient.Client) NewOption { return func(c *Connection) { c.tcc = tcc } @@ -70,7 +68,7 @@ func New(opts ...NewOption) (*Connection, error) { if (c.tcc == nil && c.loader == nil) || (c.tcc != nil && c.loader != nil) { return nil, fmt.Errorf( - "either use WithLoader or WithTemporalCloudClient", + "either use WithLoader or WithClient", ) } diff --git a/pkg/temporal/cloud/connection/connection_test.go b/pkg/temporal/cloud/connection/connection_test.go index 831a999..e9fd1dc 100644 --- a/pkg/temporal/cloud/connection/connection_test.go +++ b/pkg/temporal/cloud/connection/connection_test.go @@ -5,7 +5,8 @@ import ( "github.com/stretchr/testify/require" "github.com/temporalio/resource-operator/pkg/temporal/cloud/connection" - "github.com/temporalio/resource-operator/pkg/version" + "github.com/temporalio/resource-operator/pkg/temporal/cloud/connection/loader" + "github.com/temporalio/resource-operator/pkg/testutil" tcclient "go.temporal.io/cloud-sdk/cloudclient" ) @@ -13,57 +14,51 @@ const ( testCloudAPIKey = "test-api-key" ) -func newTCCOptions() *tcclient.Options { - return &tcclient.Options{ - UserAgent: version.UserAgent(), - } -} - func TestNew(t *testing.T) { - tccOpts := newTCCOptions() + tccOpts := testutil.NewCloudClientOptions() tccOpts.APIKey = testCloudAPIKey tcc, err := tcclient.New(*tccOpts) require.Nil(t, err) - loader := connection.NewLoader( - connection.WithTemporalCloudClientOptions(*tccOpts), + load := loader.New( + loader.WithClientOptions(*tccOpts), ) cases := []struct { name string args []connection.NewOption - expTCC *tcclient.Client - expLoader *connection.Loader + expClient *tcclient.Client + expLoader *loader.Loader expErr string }{ { name: "fail if no loader or tcclient provided", args: []connection.NewOption{}, - expErr: "either use WithLoader or WithTemporalCloudClient", + expErr: "either use WithLoader or WithClient", }, { name: "fail if both loader and tcclient provided", args: []connection.NewOption{ - connection.WithTemporalCloudClient(tcc), - connection.WithLoader(loader), + connection.WithClient(tcc), + connection.WithLoader(load), }, - expErr: "either use WithLoader or WithTemporalCloudClient", + expErr: "either use WithLoader or WithClient", }, { name: "temporal cloud client override", args: []connection.NewOption{ - connection.WithTemporalCloudClient(tcc), + connection.WithClient(tcc), }, - expTCC: tcc, + expClient: tcc, expLoader: nil, }, { name: "loader", args: []connection.NewOption{ - connection.WithLoader(loader), + connection.WithLoader(load), }, - expTCC: nil, - expLoader: &loader, + expClient: nil, + expLoader: &load, }, } for _, tc := range cases { @@ -75,22 +70,22 @@ func TestNew(t *testing.T) { return } require.NotNil(got) - if tc.expTCC != nil { + if tc.expClient != nil { gotTCC, err := got.Client() require.Nil(err) require.NotNil(gotTCC) - require.Equal(tc.expTCC, gotTCC) + require.Equal(tc.expClient, gotTCC) } gotLoader := got.Loader() if tc.expLoader == nil { require.Nil(gotLoader) } else { require.NotNil(gotLoader) - gotTCCOpts, err := gotLoader.TemporalCloudClientOptions() + gotTCCOpts, err := gotLoader.ClientOptions() require.Nil(err) - expTCCOpts, err := tc.expLoader.TemporalCloudClientOptions() + expClientOpts, err := tc.expLoader.ClientOptions() require.Nil(err) - require.EqualExportedValues(expTCCOpts, gotTCCOpts) + require.EqualExportedValues(expClientOpts, gotTCCOpts) } }) } diff --git a/pkg/temporal/cloud/connection/loader.go b/pkg/temporal/cloud/connection/loader/loader.go similarity index 84% rename from pkg/temporal/cloud/connection/loader.go rename to pkg/temporal/cloud/connection/loader/loader.go index ac38d28..6824bc5 100644 --- a/pkg/temporal/cloud/connection/loader.go +++ b/pkg/temporal/cloud/connection/loader/loader.go @@ -1,4 +1,4 @@ -package connection +package loader import ( "context" @@ -41,9 +41,9 @@ func (l Loader) KubeNamespace() string { return l.kns } -// TemporalCloudClientOptions returns the Temporal Cloud SDK client Options to -// use when dialing the Temporal Cloud client. -func (l *Loader) TemporalCloudClientOptions() (*tcclient.Options, error) { +// ClientOptions returns the Temporal Cloud SDK client Options to use when +// dialing the Temporal Cloud client. +func (l *Loader) ClientOptions() (*tcclient.Options, error) { if l.tccOpts != nil { return l.tccOpts, nil } @@ -51,7 +51,7 @@ func (l *Loader) TemporalCloudClientOptions() (*tcclient.Options, error) { return nil, fmt.Errorf("ConnectionSpec required") } return sync.OnceValues(func() (*tcclient.Options, error) { - tccOpts, err := l.cloudClientOptionsFromConnectionSpec() + tccOpts, err := l.clientOptionsFromConnectionSpec() if err != nil { return nil, err } @@ -60,12 +60,12 @@ func (l *Loader) TemporalCloudClientOptions() (*tcclient.Options, error) { })() } -// NewLoaderOption allows modifying the Loader returned from NewLoader. -type NewLoaderOption func(*Loader) +// NewOption allows modifying the Loader returned from New. +type NewOption func(*Loader) // WithKubeClient allows the caller to supply a Kubernetes client to use if // looking up credentials in Kubernetes Secrets or ConfigMaps. -func WithKubeClient(kc kclient.Client) NewLoaderOption { +func WithKubeClient(kc kclient.Client) NewOption { return func(l *Loader) { l.kc = kc } @@ -73,16 +73,16 @@ func WithKubeClient(kc kclient.Client) NewLoaderOption { // WithKubeNamespace allows the caller to supply a Kubernetes Namespace to use // if looking up credentials in Kubernetes Secrets or ConfigMaps. -func WithKubeNamespace(ns string) NewLoaderOption { +func WithKubeNamespace(ns string) NewOption { return func(l *Loader) { l.kns = ns } } -// WithTemporalCloudClientOptions allows the caller to override the Options -// supplied to the Temporal Cloud sdkclient.New function when constructing the -// Temporal Cloud Client. -func WithTemporalCloudClientOptions(tccOpts tcclient.Options) NewLoaderOption { +// WithClientOptions allows the caller to override the Options supplied to the +// Temporal Cloud sdkclient.New function when constructing the Temporal Cloud +// Client. +func WithClientOptions(tccOpts tcclient.Options) NewOption { return func(l *Loader) { l.tccOpts = &tccOpts } @@ -91,15 +91,15 @@ func WithTemporalCloudClientOptions(tccOpts tcclient.Options) NewLoaderOption { // WithConnectionSpec sets the Loader's ConnectionSpec which informs the Loader // where to find various credentials and connectivity information for the // Temporal sdkclient.Client. -func WithConnectionSpec(spec tcloudv1alpha1.ConnectionSpec) NewLoaderOption { +func WithConnectionSpec(spec tcloudv1alpha1.ConnectionSpec) NewOption { return func(l *Loader) { l.connSpec = &spec } } -// NewLoader returns a Loader that knows how to load a Connection from a +// New returns a Loader that knows how to load a Connection from a // variety of sources, including a ConnectionSpec. -func NewLoader(opts ...NewLoaderOption) Loader { +func New(opts ...NewOption) Loader { l := Loader{} for _, opt := range opts { opt(&l) @@ -108,9 +108,9 @@ func NewLoader(opts ...NewLoaderOption) Loader { return l } -// cloudClientOptionsFromConnectionSpec returns Temporal Cloud +// clientOptionsFromConnectionSpec returns Temporal Cloud // sdkclient.Options constructed from the Loader's ConnectionSpec. -func (l *Loader) cloudClientOptionsFromConnectionSpec() ( +func (l *Loader) clientOptionsFromConnectionSpec() ( *tcclient.Options, error, ) { spec := l.connSpec @@ -118,7 +118,7 @@ func (l *Loader) cloudClientOptionsFromConnectionSpec() ( return nil, fmt.Errorf("ConnectionSpec required") } - tccOpts, err := l.cloudClientOptionsFromAPIKeyCredentialsConfig( + tccOpts, err := l.clientOptionsFromAPIKeyCredentialsConfig( spec.Credentials.APIKey, ) if err != nil { @@ -159,10 +159,9 @@ func (l *Loader) cloudClientOptionsFromConnectionSpec() ( return tccOpts, nil } -// cloudClientOptionsFromAPIKeyCredentialsConfig returns the configured -// Temporal Cloud client.Options struct from the supplied -// APIKeyCredentialsConfig. -func (l Loader) cloudClientOptionsFromAPIKeyCredentialsConfig( +// clientOptionsFromAPIKeyCredentialsConfig returns the configured Temporal +// Cloud client.Options struct from the supplied APIKeyCredentialsConfig. +func (l Loader) clientOptionsFromAPIKeyCredentialsConfig( cfg tcorev1.APIKeyCredentialsConfig, ) (*tcclient.Options, error) { switch { diff --git a/pkg/temporal/cloud/connection/loader_test.go b/pkg/temporal/cloud/connection/loader/loader_test.go similarity index 83% rename from pkg/temporal/cloud/connection/loader_test.go rename to pkg/temporal/cloud/connection/loader/loader_test.go index df77034..844596b 100644 --- a/pkg/temporal/cloud/connection/loader_test.go +++ b/pkg/temporal/cloud/connection/loader/loader_test.go @@ -1,4 +1,4 @@ -package connection_test +package loader_test import ( "crypto/tls" @@ -10,9 +10,10 @@ import ( "github.com/stretchr/testify/require" tcorev1 "github.com/temporalio/kube-temporal/api/core/v1" - "github.com/temporalio/kube-temporal/pkg/testutil" + kttestutil "github.com/temporalio/kube-temporal/pkg/testutil" tcloudv1alpha1 "github.com/temporalio/resource-operator/api/cloud/v1alpha1" - "github.com/temporalio/resource-operator/pkg/temporal/cloud/connection" + "github.com/temporalio/resource-operator/pkg/temporal/cloud/connection/loader" + "github.com/temporalio/resource-operator/pkg/testutil" tcclient "go.temporal.io/cloud-sdk/cloudclient" kcorev1 "k8s.io/api/core/v1" kcrfake "sigs.k8s.io/controller-runtime/pkg/client/fake" @@ -30,29 +31,29 @@ var ( ) var validCAKeyPair = sync.OnceValues( - func() (*testutil.TLSKeyPair, error) { - return testutil.TLSNewCAKeyPair(tlsCAName) + func() (*kttestutil.TLSKeyPair, error) { + return kttestutil.TLSNewCAKeyPair(tlsCAName) }, ) func TestLoader_ClientOptions(t *testing.T) { - tccOpts := newTCCOptions() + tccOpts := testutil.NewCloudClientOptions() cases := []struct { name string - args []connection.NewLoaderOption + args []loader.NewOption exp *tcclient.Options expErr string }{ { name: "fail if no connection spec or tcclient opts", - args: []connection.NewLoaderOption{}, + args: []loader.NewOption{}, expErr: "ConnectionSpec required", }, { name: "tcclient opts override provided", - args: []connection.NewLoaderOption{ - connection.WithTemporalCloudClientOptions(*tccOpts), + args: []loader.NewOption{ + loader.WithClientOptions(*tccOpts), }, exp: tccOpts, }, @@ -60,8 +61,8 @@ func TestLoader_ClientOptions(t *testing.T) { for _, tc := range cases { t.Run(tc.name, func(tt *testing.T) { require := require.New(tt) - loader := connection.NewLoader(tc.args...) - got, err := loader.TemporalCloudClientOptions() + loader := loader.New(tc.args...) + got, err := loader.ClientOptions() if tc.expErr != "" { require.ErrorContains(err, tc.expErr) return @@ -130,14 +131,14 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { cases := []struct { name string - args []connection.NewLoaderOption + args []loader.NewOption exp *tcclient.Options expErr string }{ { name: "fail if no api key secret ref or path", - args: []connection.NewLoaderOption{ - connection.WithConnectionSpec( + args: []loader.NewOption{ + loader.WithConnectionSpec( tcloudv1alpha1.ConnectionSpec{}, ), }, @@ -145,8 +146,8 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { }, { name: "fail if api key secret no kube client", - args: []connection.NewLoaderOption{ - connection.WithConnectionSpec( + args: []loader.NewOption{ + loader.WithConnectionSpec( tcloudv1alpha1.ConnectionSpec{ Credentials: tcloudv1alpha1.CredentialsConfig{ APIKey: tcorev1.APIKeyCredentialsConfig{ @@ -165,11 +166,11 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { }, { name: "fail if api key secret no such secret", - args: []connection.NewLoaderOption{ - connection.WithKubeClient( + args: []loader.NewOption{ + loader.WithKubeClient( kcrfake.NewFakeClient(), ), - connection.WithConnectionSpec( + loader.WithConnectionSpec( tcloudv1alpha1.ConnectionSpec{ Credentials: tcloudv1alpha1.CredentialsConfig{ APIKey: tcorev1.APIKeyCredentialsConfig{ @@ -188,15 +189,15 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { }, { name: "fail if api key secret no such secret in namespace", - args: []connection.NewLoaderOption{ - connection.WithKubeNamespace(kubeNamespace), - connection.WithKubeClient( + args: []loader.NewOption{ + loader.WithKubeNamespace(kubeNamespace), + loader.WithKubeClient( // This is the non-namespaced secret, which should not // match the lookup with the kubeNamespace override from // the loader options. kcrfake.NewFakeClient(apiKeySecret), ), - connection.WithConnectionSpec( + loader.WithConnectionSpec( tcloudv1alpha1.ConnectionSpec{ Credentials: tcloudv1alpha1.CredentialsConfig{ APIKey: tcorev1.APIKeyCredentialsConfig{ @@ -215,11 +216,11 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { }, { name: "api key secret", - args: []connection.NewLoaderOption{ - connection.WithKubeClient( + args: []loader.NewOption{ + loader.WithKubeClient( kcrfake.NewFakeClient(apiKeySecret), ), - connection.WithConnectionSpec( + loader.WithConnectionSpec( tcloudv1alpha1.ConnectionSpec{ Credentials: tcloudv1alpha1.CredentialsConfig{ APIKey: tcorev1.APIKeyCredentialsConfig{ @@ -235,19 +236,19 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { ), }, exp: func() *tcclient.Options { - tcco := newTCCOptions() + tcco := testutil.NewCloudClientOptions() tcco.APIKey = apiKeySecretDataValue return tcco }(), }, { name: "api key secret in namespace", - args: []connection.NewLoaderOption{ - connection.WithKubeNamespace(kubeNamespace), - connection.WithKubeClient( + args: []loader.NewOption{ + loader.WithKubeNamespace(kubeNamespace), + loader.WithKubeClient( kcrfake.NewFakeClient(apiKeySecretWithNamespace), ), - connection.WithConnectionSpec( + loader.WithConnectionSpec( tcloudv1alpha1.ConnectionSpec{ Credentials: tcloudv1alpha1.CredentialsConfig{ APIKey: tcorev1.APIKeyCredentialsConfig{ @@ -263,18 +264,18 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { ), }, exp: func() *tcclient.Options { - tcco := newTCCOptions() + tcco := testutil.NewCloudClientOptions() tcco.APIKey = apiKeySecretDataValue return tcco }(), }, { name: "api key secret with custom host port", - args: []connection.NewLoaderOption{ - connection.WithKubeClient( + args: []loader.NewOption{ + loader.WithKubeClient( kcrfake.NewFakeClient(apiKeySecret), ), - connection.WithConnectionSpec( + loader.WithConnectionSpec( tcloudv1alpha1.ConnectionSpec{ HostPort: customHostPort, Credentials: tcloudv1alpha1.CredentialsConfig{ @@ -291,7 +292,7 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { ), }, exp: func() *tcclient.Options { - tcco := newTCCOptions() + tcco := testutil.NewCloudClientOptions() tcco.HostPort = customHostPort tcco.APIKey = apiKeySecretDataValue return tcco @@ -299,11 +300,11 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { }, { name: "api key secret with tls server name", - args: []connection.NewLoaderOption{ - connection.WithKubeClient( + args: []loader.NewOption{ + loader.WithKubeClient( kcrfake.NewFakeClient(apiKeySecret), ), - connection.WithConnectionSpec( + loader.WithConnectionSpec( tcloudv1alpha1.ConnectionSpec{ TLS: tcorev1.ServerTLSConfig{ ServerName: tlsServerName, @@ -322,7 +323,7 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { ), }, exp: func() *tcclient.Options { - tcco := newTCCOptions() + tcco := testutil.NewCloudClientOptions() tcco.TLSConfig = &tls.Config{ ServerName: tlsServerName, } @@ -332,8 +333,8 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { }, { name: "fail if api key path not found", - args: []connection.NewLoaderOption{ - connection.WithConnectionSpec( + args: []loader.NewOption{ + loader.WithConnectionSpec( tcloudv1alpha1.ConnectionSpec{ Credentials: tcloudv1alpha1.CredentialsConfig{ APIKey: tcorev1.APIKeyCredentialsConfig{ @@ -347,8 +348,8 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { }, { name: "api key path", - args: []connection.NewLoaderOption{ - connection.WithConnectionSpec( + args: []loader.NewOption{ + loader.WithConnectionSpec( tcloudv1alpha1.ConnectionSpec{ Credentials: tcloudv1alpha1.CredentialsConfig{ APIKey: tcorev1.APIKeyCredentialsConfig{ @@ -359,15 +360,15 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { ), }, exp: func() *tcclient.Options { - tcco := newTCCOptions() + tcco := testutil.NewCloudClientOptions() tcco.APIKey = apiKeySecretDataValue return tcco }(), }, { name: "api key path with tls server name", - args: []connection.NewLoaderOption{ - connection.WithConnectionSpec( + args: []loader.NewOption{ + loader.WithConnectionSpec( tcloudv1alpha1.ConnectionSpec{ TLS: tcorev1.ServerTLSConfig{ ServerName: tlsServerName, @@ -381,7 +382,7 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { ), }, exp: func() *tcclient.Options { - tcco := newTCCOptions() + tcco := testutil.NewCloudClientOptions() tcco.APIKey = apiKeySecretDataValue tcco.TLSConfig = &tls.Config{ ServerName: tlsServerName, @@ -391,11 +392,11 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { }, { name: "fail tls CA cert secret no such secret", - args: []connection.NewLoaderOption{ - connection.WithKubeClient( + args: []loader.NewOption{ + loader.WithKubeClient( kcrfake.NewFakeClient(), ), - connection.WithConnectionSpec( + loader.WithConnectionSpec( tcloudv1alpha1.ConnectionSpec{ Credentials: tcloudv1alpha1.CredentialsConfig{ APIKey: tcorev1.APIKeyCredentialsConfig{ @@ -419,11 +420,11 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { }, { name: "fail tls CA cert secret invalid cert", - args: []connection.NewLoaderOption{ - connection.WithKubeClient( + args: []loader.NewOption{ + loader.WithKubeClient( kcrfake.NewFakeClient(tlsCACertInvalidSecret), ), - connection.WithConnectionSpec( + loader.WithConnectionSpec( tcloudv1alpha1.ConnectionSpec{ Credentials: tcloudv1alpha1.CredentialsConfig{ APIKey: tcorev1.APIKeyCredentialsConfig{ @@ -447,11 +448,11 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { }, { name: "api key with tls CA cert opaque secret type", - args: []connection.NewLoaderOption{ - connection.WithKubeClient( + args: []loader.NewOption{ + loader.WithKubeClient( kcrfake.NewFakeClient(tlsCACertOpaqueSecret), ), - connection.WithConnectionSpec( + loader.WithConnectionSpec( tcloudv1alpha1.ConnectionSpec{ Credentials: tcloudv1alpha1.CredentialsConfig{ APIKey: tcorev1.APIKeyCredentialsConfig{ @@ -472,7 +473,7 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { ), }, exp: func() *tcclient.Options { - tcco := newTCCOptions() + tcco := testutil.NewCloudClientOptions() tcco.APIKey = apiKeySecretDataValue tcco.TLSConfig = &tls.Config{ RootCAs: poolWithValidCACert, @@ -482,11 +483,11 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { }, { name: "api key with tls CA cert tls secret tpe", - args: []connection.NewLoaderOption{ - connection.WithKubeClient( + args: []loader.NewOption{ + loader.WithKubeClient( kcrfake.NewFakeClient(tlsCACertTLSSecret), ), - connection.WithConnectionSpec( + loader.WithConnectionSpec( tcloudv1alpha1.ConnectionSpec{ Credentials: tcloudv1alpha1.CredentialsConfig{ APIKey: tcorev1.APIKeyCredentialsConfig{ @@ -511,7 +512,7 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { ), }, exp: func() *tcclient.Options { - tcco := newTCCOptions() + tcco := testutil.NewCloudClientOptions() tcco.APIKey = apiKeySecretDataValue tcco.TLSConfig = &tls.Config{ RootCAs: poolWithValidCACert, @@ -523,8 +524,8 @@ func TestLoader_ClientOptions_ControllerSpec_APIKeyCredentials(t *testing.T) { for _, tc := range cases { t.Run(tc.name, func(tt *testing.T) { require := require.New(tt) - loader := connection.NewLoader(tc.args...) - got, err := loader.TemporalCloudClientOptions() + loader := loader.New(tc.args...) + got, err := loader.ClientOptions() if tc.expErr != "" { require.ErrorContains(err, tc.expErr) return diff --git a/pkg/temporal/cloud/connection/manager/manager.go b/pkg/temporal/cloud/connection/manager/manager.go new file mode 100644 index 0000000..74bf925 --- /dev/null +++ b/pkg/temporal/cloud/connection/manager/manager.go @@ -0,0 +1,314 @@ +package manager + +import ( + "fmt" + "sync" + + "github.com/samber/lo" + tcloudv1alpha1 "github.com/temporalio/resource-operator/api/cloud/v1alpha1" + tcclient "go.temporal.io/cloud-sdk/cloudclient" + kclient "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/temporalio/resource-operator/pkg/temporal/cloud/connection" + "github.com/temporalio/resource-operator/pkg/temporal/cloud/connection/loader" +) + +// NewOption adapts the Manager returned from New. +type NewOption func(*Manager) + +// WithKubeClient sets the Manager's Kubernetes client. +func WithKubeClient(kc kclient.Client) NewOption { + return func(m *Manager) { + m.kc = kc + } +} + +// New returns a new Temporal Cloud connection manager. +func New(opts ...NewOption) *Manager { + m := Manager{ + hostPortConnSpecs: map[string][]*tcloudv1alpha1.ConnectionSpec{}, + hostPortClients: map[string]*tcclient.Client{}, + connSpecs: map[string]*tcloudv1alpha1.ConnectionSpec{}, + clients: map[string]*tcclient.Client{}, + } + for _, opt := range opts { + opt(&m) + } + return &m +} + +// Manager manages a cache of Temporal Cloud connection-related objects. +type Manager struct { + sync.RWMutex + // kc contains the Kubernetes client set on the Manager. + kc kclient.Client + // hostPortConnSpecs contains a map, keyed by HostPort string, of slices of + // [tcloudv1alpha1.ConnectionSpec] objects that contain instructions for + // creating cloud clients to that HostPort. + hostPortConnSpecs map[string][]*tcloudv1alpha1.ConnectionSpec + // hostPortClients contains a map, keyed by HostPort string, of + // [tcclient.Client] objects associated with the HostPort. + hostPortClients map[string]*tcclient.Client + // connSpecs contains the set of [tcloudv1alpha1.ConnectionSpec] objects + // associated with the hash of a [tcloudv1alpha1.ConnectionSpec]. + connSpecs map[string]*tcloudv1alpha1.ConnectionSpec + // clients contains the set of [tccClient.Client] objects associated + // with the hash of a [tcloudv1alpha1.ConnectionSpec]. + clients map[string]*tcclient.Client +} + +// SetKubeClient sets the Manager's Kubernetes client. +func (m *Manager) SetKubeClient(kc kclient.Client) { + m.kc = kc +} + +type registerOptions struct { + tcc *tcclient.Client + connSpecs []*tcloudv1alpha1.ConnectionSpec +} + +// RegisterOption allows callers to the Manager.Register function to control +// what is being registered with the manager. +type RegisterOption func(*registerOptions) + +// RegisterWithClient allows overriding the Temporal Cloud client associated +// with the Register subject. +func RegisterWithClient(tcc *tcclient.Client) RegisterOption { + return func(o *registerOptions) { + o.tcc = tcc + } +} + +// RegisterWithConnectionSpecs allows callers to the Manager.Register function +// to associate one or more [tcloudv1alpha1.ConnectionSpec] objects with the +// HostPort being registered. +func RegisterWithConnectionSpecs( + specs ...*tcloudv1alpha1.ConnectionSpec, +) RegisterOption { + return func(o *registerOptions) { + o.connSpecs = append(o.connSpecs, specs...) + } +} + +// Register safely adds the supplied subject to the Manager's various caches. +// +// If subject is a [tcloudv1alpha1.ConnectionSpec], the Manager map's the +// spec's host port to the connection spec. +// +// If subject is a string, it is assumed to be a HostPort and at least one +// RegisterOption should be provided that associates the HostPort to a Temporal +// Cloud client or a set of related connection specs. +func (m *Manager) Register(subject any, opts ...RegisterOption) error { + ro := registerOptions{} + for _, opt := range opts { + opt(&ro) + } + + specs := []*tcloudv1alpha1.ConnectionSpec{} + switch subject := subject.(type) { + case *tcloudv1alpha1.ConnectionSpec: + specHash := string(subject.Hash()) + m.RLock() + _, ok := m.connSpecs[specHash] + m.RUnlock() + + if ok { + // We already know about this subject + return nil + } + + specs = append(specs, subject) + case string: + m.RLock() + _, ok := m.hostPortConnSpecs[subject] + m.RUnlock() + if ok { + // We already know about this subject + return nil + } + + m.RLock() + // The HostPort may have been previously registered with no connection + // specs and just a pre-created Temporal Cloud Client. + _, ok = m.hostPortClients[subject] + m.RUnlock() + if ok { + // We already know about this subject + return nil + } + + if ro.tcc == nil && len(ro.connSpecs) == 0 { + return fmt.Errorf( + "connection specs or temporal cloud client " + + "required to register HostPort", + ) + } + + // Ensure that all the supplied connection specs actually use the + // subject HostPort. + for _, spec := range ro.connSpecs { + hp := spec.HostPort + if hp == "" { + hp = tcloudv1alpha1.DefaultConnectionHostPort + } + if hp != subject { + return fmt.Errorf( + "connection spec HostPort %q does not match subject %q", + hp, subject, + ) + } + } + specs = ro.connSpecs + default: + return fmt.Errorf("unsupported type %T", subject) + } + + m.Lock() + defer m.Unlock() + + if len(specs) == 0 { + // We shouldn't get here, but just in case... + if ro.tcc == nil { + return fmt.Errorf( + "connection specs or temporal cloud client " + + "required to register HostPort", + ) + } + m.hostPortClients[subject.(string)] = ro.tcc + return nil + } + + for _, spec := range specs { + hp := spec.HostPort + if hp == "" { + hp = tcloudv1alpha1.DefaultConnectionHostPort + } + + specHash := string(spec.Hash()) + + m.connSpecs[specHash] = spec + if ro.tcc != nil { + m.clients[specHash] = ro.tcc + } + + hpcs := m.hostPortConnSpecs[hp] + if len(hpcs) == 0 { + hpcs = []*tcloudv1alpha1.ConnectionSpec{} + } + hpcs = append(hpcs, spec) + m.hostPortConnSpecs[hp] = hpcs + if ro.tcc != nil { + m.hostPortClients[hp] = ro.tcc + } + } + return nil +} + +// Client returns a [tcclient.Client] struct configured for the +// matching the supplied subject. +// +// If a cached client is not found for the supplied subject, we attempt to +// create the client and cache the created client in the Manager. +// +// If subject is a [tcloudv1.ConnectionSpec], we look up the client by the +// connection spec's hash key. +// +// If subject is a string, we find the first known connection spec having that +// HostPort and then look up the client by that connection spec's hash key. +func (m *Manager) Client(subject any) (*tcclient.Client, error) { + var specHash string + var spec *tcloudv1alpha1.ConnectionSpec + + switch subject := subject.(type) { + case *tcloudv1alpha1.ConnectionSpec: + // Caller passed a ConnectionSpec. If we have a pre-created Temporal + // Cloud client for this connection spec, just return it. Otherwise, we + // fall through and attempt to create client from the supplied + // connection spec that contains instructions to create a client for + // this HostPort. + if lo.IsNil(subject) { + return nil, fmt.Errorf("nil cloud connection spec") + } + + specHash = string(subject.Hash()) + + m.RLock() + cli, ok := m.clients[specHash] + m.RUnlock() + if ok { + return cli, nil + } + + m.RLock() + _, ok = m.connSpecs[specHash] + m.RUnlock() + if !ok { + return nil, fmt.Errorf("no cloud client registered for spec") + } + spec = subject + case string: + // Caller passed a HostPort. If we have a pre-created Temporal Cloud + // client for this HostPort, just return it. Otherwise, we check to see + // if we have any connection specs that contain instructions to create + // a client for this HostPort. + + m.RLock() + cli, ok := m.hostPortClients[subject] + m.RUnlock() + if ok { + return cli, nil + } + + // We don't have a cached Temporal Cloud for this HostPort, so get the + // first ConnectionSpec associated with the supplied HostPort. + m.RLock() + specs, ok := m.hostPortConnSpecs[subject] + m.RUnlock() + if !ok || len(specs) == 0 { + return nil, fmt.Errorf( + "no client or connection specs found for HostPort %q", subject, + ) + } + spec = specs[0] + specHash = string(spec.Hash()) + default: + return nil, fmt.Errorf( + "unsupported type %T", subject, + ) + + } + + m.RLock() + cli, ok := m.clients[specHash] + m.RUnlock() + if ok { + return cli, nil + } + + // lazily create the Temporal Cloud client. + + m.Lock() + defer m.Unlock() + + conn, err := connection.New( + connection.WithLoader( + loader.New( + loader.WithKubeClient(m.kc), + loader.WithConnectionSpec(*spec), + ), + ), + ) + if err != nil { + return nil, fmt.Errorf( + "failed creating cloud connection: %w", err, + ) + } + tc, err := conn.Client() + if err != nil { + return nil, fmt.Errorf( + "failed getting cloud client: %w", err, + ) + } + m.clients[specHash] = tc + return tc, nil +} diff --git a/pkg/temporal/cloud/connection/manager/manager_test.go b/pkg/temporal/cloud/connection/manager/manager_test.go new file mode 100644 index 0000000..987dedd --- /dev/null +++ b/pkg/temporal/cloud/connection/manager/manager_test.go @@ -0,0 +1,287 @@ +package manager_test + +import ( + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" + tcorev1 "github.com/temporalio/kube-temporal/api/core/v1" + tcloudv1alpha1 "github.com/temporalio/resource-operator/api/cloud/v1alpha1" + "github.com/temporalio/resource-operator/pkg/temporal/cloud/connection/manager" + "github.com/temporalio/resource-operator/pkg/testutil" + tcclient "go.temporal.io/cloud-sdk/cloudclient" +) + +func TestManager_Register(t *testing.T) { + tcc, err := testutil.NewFakeCloudClient() + require.Nil(t, err) + + cases := []struct { + name string + newArgs []manager.NewOption + subject any + regOpts []manager.RegisterOption + expErr string + }{ + { + name: "fail if subject is neither string nor conn spec", + newArgs: []manager.NewOption{}, + subject: int64(1), + expErr: "unsupported type int64", + }, + { + name: "fail if string subject has no reg opts", + newArgs: []manager.NewOption{}, + subject: "host:port", + expErr: "connection specs or temporal cloud client required to register HostPort", + }, + { + name: "fail if string subject has unmatched conn spec", + newArgs: []manager.NewOption{}, + subject: "host:port", + regOpts: []manager.RegisterOption{ + manager.RegisterWithConnectionSpecs( + &tcloudv1alpha1.ConnectionSpec{ + HostPort: "differenthost:port", + }, + ), + }, + expErr: "connection spec HostPort \"differenthost:port\" does not match subject \"host:port\"", + }, + { + name: "string subject with conn spec", + newArgs: []manager.NewOption{}, + subject: "host:port", + regOpts: []manager.RegisterOption{ + manager.RegisterWithConnectionSpecs( + &tcloudv1alpha1.ConnectionSpec{ + HostPort: "host:port", + }, + ), + }, + }, + { + name: "string subject with client override", + newArgs: []manager.NewOption{}, + subject: "host:port", + regOpts: []manager.RegisterOption{ + manager.RegisterWithClient(tcc), + }, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(tt *testing.T) { + require := require.New(tt) + m := manager.New(tc.newArgs...) + + err := m.Register(tc.subject, tc.regOpts...) + if tc.expErr != "" { + require.ErrorContains(err, tc.expErr) + return + } + require.Nil(err) + }) + } +} + +type registerFunc func(*manager.Manager) error + +func TestManager_Client(t *testing.T) { + tcc, err := testutil.NewFakeCloudClient() + require.Nil(t, err) + + apiKeyDir := t.TempDir() + apiKeyPath := filepath.Join(apiKeyDir, "api-key") + + err = os.WriteFile( + apiKeyPath, + []byte(testutil.FakeTemporalCloudAPIKeyContents), + 0644, + ) + require.Nil(t, err) + + cases := []struct { + name string + m *manager.Manager + subject any + register []registerFunc + expErr string + exp *tcclient.Client + }{ + { + name: "fail if subject is neither string nor conn spec", + m: manager.New(), + subject: int64(1), + expErr: "unsupported type int64", + }, + { + name: "fail if string subject has no client or spec registered", + m: manager.New(), + subject: "host:port", + expErr: "no client or connection specs found for HostPort", + }, + { + name: "fails string subject with invalid conn spec", + m: manager.New(), + subject: "host:port", + register: []registerFunc{ + func(m *manager.Manager) error { + return m.Register( + "host:port", + manager.RegisterWithConnectionSpecs( + &tcloudv1alpha1.ConnectionSpec{ + HostPort: "host:port", + // This spec is not valid because it must contain + // APIKey credentials + }, + ), + ) + }, + }, + expErr: "APIKeyCredentialsConfig: either SecretRef or Path must be non-empty", + }, + { + name: "string subject yields registered client", + m: manager.New(), + subject: "host:port", + register: []registerFunc{ + func(m *manager.Manager) error { + return m.Register( + "host:port", + manager.RegisterWithClient(tcc), + ) + }, + }, + exp: tcc, + }, + { + name: "fails spec subject with invalid conn spec", + m: manager.New(), + subject: &tcloudv1alpha1.ConnectionSpec{ + HostPort: "host:port", + }, + register: []registerFunc{ + func(m *manager.Manager) error { + return m.Register( + "host:port", + manager.RegisterWithConnectionSpecs( + &tcloudv1alpha1.ConnectionSpec{ + HostPort: "host:port", + // This spec is not valid because it must contain + // APIKey credentials + }, + ), + ) + }, + }, + expErr: "APIKeyCredentialsConfig: either SecretRef or Path must be non-empty", + }, + { + name: "spec subject yields registered client", + m: manager.New(), + subject: &tcloudv1alpha1.ConnectionSpec{ + HostPort: "host:port", + }, + register: []registerFunc{ + func(m *manager.Manager) error { + return m.Register( + &tcloudv1alpha1.ConnectionSpec{ + HostPort: "host:port", + }, + manager.RegisterWithClient(tcc), + ) + }, + }, + exp: tcc, + }, + { + name: "spec subject yields registered client", + m: manager.New(), + subject: &tcloudv1alpha1.ConnectionSpec{ + HostPort: "host:port", + }, + register: []registerFunc{ + func(m *manager.Manager) error { + return m.Register( + &tcloudv1alpha1.ConnectionSpec{ + HostPort: "host:port", + }, + manager.RegisterWithClient(tcc), + ) + }, + }, + exp: tcc, + }, + { + name: "string subject constructs client from registered conn spec", + m: manager.New(), + subject: "host:port", + register: []registerFunc{ + func(m *manager.Manager) error { + return m.Register( + "host:port", + manager.RegisterWithConnectionSpecs( + &tcloudv1alpha1.ConnectionSpec{ + HostPort: "host:port", + Credentials: tcloudv1alpha1.CredentialsConfig{ + APIKey: tcorev1.APIKeyCredentialsConfig{ + Path: apiKeyPath, + }, + }, + }, + ), + ) + }, + }, + exp: tcc, + }, + { + name: "spec subject constructs client from registered conn spec", + m: manager.New(), + subject: &tcloudv1alpha1.ConnectionSpec{ + HostPort: "host:port", + Credentials: tcloudv1alpha1.CredentialsConfig{ + APIKey: tcorev1.APIKeyCredentialsConfig{ + Path: apiKeyPath, + }, + }, + }, + register: []registerFunc{ + func(m *manager.Manager) error { + return m.Register( + "host:port", + manager.RegisterWithConnectionSpecs( + &tcloudv1alpha1.ConnectionSpec{ + HostPort: "host:port", + Credentials: tcloudv1alpha1.CredentialsConfig{ + APIKey: tcorev1.APIKeyCredentialsConfig{ + Path: apiKeyPath, + }, + }, + }, + ), + ) + }, + }, + exp: tcc, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(tt *testing.T) { + require := require.New(tt) + + for _, reg := range tc.register { + err := reg(tc.m) + require.Nil(err) + } + got, err := tc.m.Client(tc.subject) + if tc.expErr != "" { + require.ErrorContains(err, tc.expErr) + return + } + require.Nil(err) + require.EqualExportedValues(tc.exp, got) + }) + } +} diff --git a/pkg/testutil/kube.go b/pkg/testutil/kube.go new file mode 100644 index 0000000..3fb88c0 --- /dev/null +++ b/pkg/testutil/kube.go @@ -0,0 +1,27 @@ +package testutil + +import ( + tcloudv1alpha1 "github.com/temporalio/resource-operator/api/cloud/v1alpha1" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + kcrfake "sigs.k8s.io/controller-runtime/pkg/client/fake" +) + +var ( + scheme = runtime.NewScheme() +) + +func init() { + _ = corev1.AddToScheme(scheme) + _ = tcloudv1alpha1.AddToScheme(scheme) +} + +// NewFakeClient returns a controller-runtime Client populated with the +// supplied objects. +func NewFakeClient(objs ...runtime.Object) client.Client { + return kcrfake.NewClientBuilder(). + WithScheme(scheme). + WithRuntimeObjects(objs...). + Build() +} diff --git a/pkg/testutil/tcloud.go b/pkg/testutil/tcloud.go new file mode 100644 index 0000000..edcc8a9 --- /dev/null +++ b/pkg/testutil/tcloud.go @@ -0,0 +1,45 @@ +package testutil + +import ( + tcorev1 "github.com/temporalio/kube-temporal/api/core/v1" + tcloudv1alpha1 "github.com/temporalio/resource-operator/api/cloud/v1alpha1" + "github.com/temporalio/resource-operator/pkg/version" + tcclient "go.temporal.io/cloud-sdk/cloudclient" +) + +const ( + FakeTemporalCloudHostPort = "fake.saas-api.tmprl.cloud:443" + FakeTemporalCloudAPIKeyContents = "indy!bad-dates!" +) + +// NewFakeCloudConnectionSpec returns a tcloudv1alpha1.ConnectionSpec for +// connecting to a fake Temporal Cloud API endpoint. +func NewFakeCloudConnectionSpec( + apiKeyPath string, +) *tcloudv1alpha1.ConnectionSpec { + return &tcloudv1alpha1.ConnectionSpec{ + HostPort: FakeTemporalCloudHostPort, + Credentials: tcloudv1alpha1.CredentialsConfig{ + APIKey: tcorev1.APIKeyCredentialsConfig{ + Path: apiKeyPath, + }, + }, + } +} + +// NewCloudClientOptions returns new Temporal Cloud client options configured +// with the UserAgent set to the testing user-agent. +func NewCloudClientOptions() *tcclient.Options { + return &tcclient.Options{ + UserAgent: version.UserAgent(), + } +} + +// NewFakeCloudClient returns a new Temporal Cloud client configured to connect +// to a fake Temporal Cloud API endpoint. +func NewFakeCloudClient() (*tcclient.Client, error) { + opts := NewCloudClientOptions() + opts.HostPort = FakeTemporalCloudHostPort + opts.APIKey = FakeTemporalCloudAPIKeyContents + return tcclient.New(*opts) +}