[v14] integrations/operator: re-use the teleport client instead of creating a new one

integrations/operator/controllers/resources/github_connector_controller.go

@@ -34,41 +34,45 @@ type githubConnectorClient struct {
 
 // Get gets the Teleport github_connector of a given name
 func (r githubConnectorClient) Get(ctx context.Context, name string) (types.GithubConnector, error) {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
+	defer release()
 
 	github, err := teleportClient.GetGithubConnector(ctx, name, false /* with secrets*/)
 	return github, trace.Wrap(err)
 }
 
 // Create creates a Teleport github_connector
 func (r githubConnectorClient) Create(ctx context.Context, github types.GithubConnector) error {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	return trace.Wrap(teleportClient.UpsertGithubConnector(ctx, github))
 }
 
 // Update updates a Teleport github_connector
 func (r githubConnectorClient) Update(ctx context.Context, github types.GithubConnector) error {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	return trace.Wrap(teleportClient.UpsertGithubConnector(ctx, github))
 }
 
 // Delete deletes a Teleport github_connector
 func (r githubConnectorClient) Delete(ctx context.Context, name string) error {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	return trace.Wrap(teleportClient.DeleteGithubConnector(ctx, name))
 }

integrations/operator/controllers/resources/login_rule_controller.go

@@ -33,10 +33,11 @@ type loginRuleClient struct {
 
 // Get gets the Teleport login_rule of a given name
 func (l loginRuleClient) Get(ctx context.Context, name string) (*resourcesv1.LoginRuleResource, error) {
-	teleportClient, err := l.TeleportClientAccessor(ctx)
+	teleportClient, release, err := l.TeleportClientAccessor(ctx)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
+	defer release()
 
 	loginRule, err := teleportClient.GetLoginRule(ctx, name)
 	if err != nil {
@@ -48,32 +49,35 @@ func (l loginRuleClient) Get(ctx context.Context, name string) (*resourcesv1.Log
 
 // Create creates a Teleport login_rule
 func (l loginRuleClient) Create(ctx context.Context, resource *resourcesv1.LoginRuleResource) error {
-	teleportClient, err := l.TeleportClientAccessor(ctx)
+	teleportClient, release, err := l.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	_, err = teleportClient.CreateLoginRule(ctx, resource.LoginRule)
 	return trace.Wrap(err)
 }
 
 // Update updates a Teleport login_rule
 func (l loginRuleClient) Update(ctx context.Context, resource *resourcesv1.LoginRuleResource) error {
-	teleportClient, err := l.TeleportClientAccessor(ctx)
+	teleportClient, release, err := l.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	_, err = teleportClient.UpsertLoginRule(ctx, resource.LoginRule)
 	return trace.Wrap(err)
 }
 
 // Delete deletes a Teleport login_rule
 func (l loginRuleClient) Delete(ctx context.Context, name string) error {
-	teleportClient, err := l.TeleportClientAccessor(ctx)
+	teleportClient, release, err := l.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	return trace.Wrap(teleportClient.DeleteLoginRule(ctx, name))
 }

integrations/operator/controllers/resources/oidc_connector_controller.go

@@ -34,41 +34,45 @@ type oidcConnectorClient struct {
 
 // Get gets the Teleport oidc_connector of a given name
 func (r oidcConnectorClient) Get(ctx context.Context, name string) (types.OIDCConnector, error) {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
+	defer release()
 
 	oidc, err := teleportClient.GetOIDCConnector(ctx, name, false /* with secrets*/)
 	return oidc, trace.Wrap(err)
 }
 
 // Create creates a Teleport oidc_connector
 func (r oidcConnectorClient) Create(ctx context.Context, oidc types.OIDCConnector) error {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	return trace.Wrap(teleportClient.UpsertOIDCConnector(ctx, oidc))
 }
 
 // Update updates a Teleport oidc_connector
 func (r oidcConnectorClient) Update(ctx context.Context, oidc types.OIDCConnector) error {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	return trace.Wrap(teleportClient.UpsertOIDCConnector(ctx, oidc))
 }
 
 // Delete deletes a Teleport oidc_connector
 func (r oidcConnectorClient) Delete(ctx context.Context, name string) error {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	return trace.Wrap(teleportClient.DeleteOIDCConnector(ctx, name))
 }

integrations/operator/controllers/resources/okta_import_rule_controller.go

@@ -34,43 +34,47 @@ type oktaImportRuleClient struct {
 
 // Get gets the Teleport okta_import_rule of a given name
 func (r oktaImportRuleClient) Get(ctx context.Context, name string) (types.OktaImportRule, error) {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
+	defer release()
 
 	importRule, err := teleportClient.OktaClient().GetOktaImportRule(ctx, name)
 	return importRule, trace.Wrap(err)
 }
 
 // Create creates a Teleport okta_import_rule
 func (r oktaImportRuleClient) Create(ctx context.Context, importRule types.OktaImportRule) error {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	_, err = teleportClient.OktaClient().CreateOktaImportRule(ctx, importRule)
 	return trace.Wrap(err)
 }
 
 // Update updates a Teleport okta_import_rule
 func (r oktaImportRuleClient) Update(ctx context.Context, importRule types.OktaImportRule) error {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	_, err = teleportClient.OktaClient().UpdateOktaImportRule(ctx, importRule)
 	return trace.Wrap(err)
 }
 
 // Delete deletes a Teleport okta_import_rule
 func (r oktaImportRuleClient) Delete(ctx context.Context, name string) error {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	return trace.Wrap(teleportClient.OktaClient().DeleteOktaImportRule(ctx, name))
 }

integrations/operator/controllers/resources/provision_token_controller.go

@@ -31,41 +31,45 @@ type provisionTokenClient struct {
 
 // Get gets the Teleport provision token of a given name
 func (r provisionTokenClient) Get(ctx context.Context, name string) (types.ProvisionToken, error) {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
+	defer release()
 
 	token, err := teleportClient.GetToken(ctx, name)
 	return token, trace.Wrap(err)
 }
 
 // Create creates a Teleport provision token
 func (r provisionTokenClient) Create(ctx context.Context, token types.ProvisionToken) error {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	return trace.Wrap(teleportClient.UpsertToken(ctx, token))
 }
 
 // Update updates a Teleport provision token
 func (r provisionTokenClient) Update(ctx context.Context, token types.ProvisionToken) error {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	return trace.Wrap(teleportClient.UpsertToken(ctx, token))
 }
 
 // Delete deletes a Teleport provision token
 func (r provisionTokenClient) Delete(ctx context.Context, name string) error {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	return trace.Wrap(teleportClient.DeleteToken(ctx, name))
 }

integrations/operator/controllers/resources/role_controller.go

@@ -91,10 +91,12 @@ func (r *RoleReconciler) SetupWithManager(mgr ctrl.Manager) error {
 }
 
 func (r *RoleReconciler) Delete(ctx context.Context, obj kclient.Object) error {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
+
 	return teleportClient.DeleteRole(ctx, obj.GetName())
 }
 
@@ -123,7 +125,10 @@ func (r *RoleReconciler) Upsert(ctx context.Context, obj kclient.Object) error {
 
 	// Converting the Kubernetes resource into a Teleport one, checking potential ownership issues.
 	teleportResource := k8sResource.ToTeleport()
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
+	if err == nil {
+		defer release()
+	}
 	updateErr = updateStatus(updateStatusConfig{
 		ctx:         ctx,
 		client:      r.Client,

integrations/operator/controllers/resources/saml_connector_controller.go

@@ -34,41 +34,45 @@ type samlConnectorClient struct {
 
 // Get gets the Teleport saml_connector of a given name
 func (r samlConnectorClient) Get(ctx context.Context, name string) (types.SAMLConnector, error) {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
+	defer release()
 
 	saml, err := teleportClient.GetSAMLConnector(ctx, name, false /* with secrets*/)
 	return saml, trace.Wrap(err)
 }
 
 // Create creates a Teleport saml_connector
 func (r samlConnectorClient) Create(ctx context.Context, saml types.SAMLConnector) error {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	return trace.Wrap(teleportClient.UpsertSAMLConnector(ctx, saml))
 }
 
 // Update updates a Teleport saml_connector
 func (r samlConnectorClient) Update(ctx context.Context, saml types.SAMLConnector) error {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	return trace.Wrap(teleportClient.UpsertSAMLConnector(ctx, saml))
 }
 
 // Delete deletes a Teleport saml_connector
 func (r samlConnectorClient) Delete(ctx context.Context, name string) error {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	return trace.Wrap(teleportClient.DeleteSAMLConnector(ctx, name))
 }

integrations/operator/controllers/resources/testlib/env.go

@@ -45,6 +45,7 @@ import (
 	resourcesv3 "github.com/gravitational/teleport/integrations/operator/apis/resources/v3"
 	resourcesv5 "github.com/gravitational/teleport/integrations/operator/apis/resources/v5"
 	"github.com/gravitational/teleport/integrations/operator/controllers/resources"
+	"github.com/gravitational/teleport/integrations/operator/sidecar"
 	"github.com/gravitational/teleport/lib/modules"
 	"github.com/gravitational/teleport/lib/service/servicecfg"
 )
@@ -173,8 +174,8 @@ func (s *TestSetup) StartKubernetesOperator(t *testing.T) {
 	}
 
 	// We have to create a new Manager on each start because the Manager does not support to be restarted
-	clientAccessor := func(ctx context.Context) (*client.Client, error) {
-		return s.TeleportClient, nil
+	clientAccessor := func(ctx context.Context) (*sidecar.SyncClient, func(), error) {
+		return sidecar.NewSyncClient(s.TeleportClient), func() {}, nil
 	}
 
 	k8sManager, err := ctrl.NewManager(s.K8sRestConfig, ctrl.Options{

integrations/operator/controllers/resources/user_controller.go

@@ -34,41 +34,45 @@ type userClient struct {
 
 // Get gets the Teleport user of a given name
 func (r userClient) Get(ctx context.Context, name string) (types.User, error) {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
+	defer release()
 
 	user, err := teleportClient.GetUser(name, false /* with secrets*/)
 	return user, trace.Wrap(err)
 }
 
 // Create creates a Teleport user
 func (r userClient) Create(ctx context.Context, user types.User) error {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	return trace.Wrap(teleportClient.CreateUser(ctx, user))
 }
 
 // Update updates a Teleport user
 func (r userClient) Update(ctx context.Context, user types.User) error {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	return trace.Wrap(teleportClient.UpdateUser(ctx, user))
 }
 
 // Delete deletes a Teleport user
 func (r userClient) Delete(ctx context.Context, name string) error {
-	teleportClient, err := r.TeleportClientAccessor(ctx)
+	teleportClient, release, err := r.TeleportClientAccessor(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	defer release()
 
 	return trace.Wrap(teleportClient.DeleteUser(ctx, name))
 }