feat(serviceintegration): adopt existing integrations instead of failing (#1042)

Author: vmyroslav

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## [MAJOR.MINOR.PATCH] - YYYY-MM-DD
 
+- Change `ServiceIntegration`: operator now adopts existing integrations instead of failing with conflict
 - Fix `PostgreSQL`: added retry logic for errors during upgrade task
 
 ## v0.33.1 - 2025-10-08

api/v1alpha1/serviceintegration_types.go

@@ -103,7 +103,9 @@ type ServiceIntegrationStatus struct {
 // +kubebuilder:object:root=true
 // +kubebuilder:subresource:status
 
-// ServiceIntegration is the Schema for the serviceintegrations API
+// ServiceIntegration is the Schema for the serviceintegrations API.
+//
+// info "Adoption of existing integrations": If a ServiceIntegration resource is created with configuration matching an existing Aiven integration (created outside the operator), the operator will adopt the existing integration.
 // +kubebuilder:printcolumn:name="Project",type="string",JSONPath=".spec.project"
 // +kubebuilder:printcolumn:name="Type",type="string",JSONPath=".spec.integrationType"
 // +kubebuilder:printcolumn:name="Source Service Name",type="string",JSONPath=".spec.sourceServiceName"

charts/aiven-operator-crds/templates/aiven.io_serviceintegrations.yaml

@@ -36,9 +36,11 @@ spec:
       name: v1alpha1
       schema:
         openAPIV3Schema:
-          description:
-            ServiceIntegration is the Schema for the serviceintegrations
-            API
+          description: |-
+            ServiceIntegration is the Schema for the serviceintegrations API.
+
+
+            info "Adoption of existing integrations": If a ServiceIntegration resource is created with configuration matching an existing Aiven integration (created outside the operator), the operator will adopt the existing integration.
           properties:
             apiVersion:
               description: |-

config/crd/bases/aiven.io_serviceintegrations.yaml

@@ -36,9 +36,11 @@ spec:
       name: v1alpha1
       schema:
         openAPIV3Schema:
-          description:
-            ServiceIntegration is the Schema for the serviceintegrations
-            API
+          description: |-
+            ServiceIntegration is the Schema for the serviceintegrations API.
+
+
+            info "Adoption of existing integrations": If a ServiceIntegration resource is created with configuration matching an existing Aiven integration (created outside the operator), the operator will adopt the existing integration.
           properties:
             apiVersion:
               description: |-

controllers/serviceintegration_controller.go

@@ -11,6 +11,7 @@ import (
 	avngen "github.com/aiven/go-client-codegen"
 	"github.com/aiven/go-client-codegen/handler/service"
 	"github.com/avast/retry-go"
+	"github.com/google/go-cmp/cmp"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -57,31 +58,43 @@ func (h ServiceIntegrationHandler) createOrUpdate(ctx context.Context, avnGen av
 	}
 
 	if si.Status.ID == "" {
-		userConfigMap, err := CreateUserConfiguration(userConfig)
+		existingIntegration, err := h.findExistingIntegration(ctx, avnGen, si)
 		if err != nil {
-			return err
+			return fmt.Errorf("failed to check for existing integration: %w", err)
 		}
 
-		integration, err := avnGen.ServiceIntegrationCreate(
-			ctx,
-			si.Spec.Project,
-			&service.ServiceIntegrationCreateIn{
-				DestEndpointId:   NilIfZero(si.Spec.DestinationEndpointID),
-				DestService:      NilIfZero(si.Spec.DestinationServiceName),
-				DestProject:      NilIfZero(si.Spec.DestinationProjectName),
-				IntegrationType:  si.Spec.IntegrationType,
-				SourceEndpointId: NilIfZero(si.Spec.SourceEndpointID),
-				SourceService:    NilIfZero(si.Spec.SourceServiceName),
-				SourceProject:    NilIfZero(si.Spec.SourceProjectName),
-				UserConfig:       &userConfigMap,
-			},
-		)
-		if err != nil {
-			return fmt.Errorf("cannot createOrUpdate service integration: %w", err)
+		if existingIntegration != nil {
+			si.Status.ID = existingIntegration.ServiceIntegrationId // adopt existing
+		} else {
+			userConfigMap, err := CreateUserConfiguration(userConfig)
+			if err != nil {
+				return err
+			}
+
+			integration, err := avnGen.ServiceIntegrationCreate(
+				ctx,
+				si.Spec.Project,
+				&service.ServiceIntegrationCreateIn{
+					DestEndpointId:   NilIfZero(si.Spec.DestinationEndpointID),
+					DestService:      NilIfZero(si.Spec.DestinationServiceName),
+					DestProject:      NilIfZero(si.Spec.DestinationProjectName),
+					IntegrationType:  si.Spec.IntegrationType,
+					SourceEndpointId: NilIfZero(si.Spec.SourceEndpointID),
+					SourceService:    NilIfZero(si.Spec.SourceServiceName),
+					SourceProject:    NilIfZero(si.Spec.SourceProjectName),
+					UserConfig:       &userConfigMap,
+				},
+			)
+			if err != nil {
+				return fmt.Errorf("cannot createOrUpdate service integration: %w", err)
+			}
+
+			si.Status.ID = integration.ServiceIntegrationId
+			return nil
 		}
+	}
 
-		si.Status.ID = integration.ServiceIntegrationId
-	} else {
+	if si.Status.ID != "" {
 		if !si.HasUserConfig() {
 			return nil
 		}
@@ -199,3 +212,79 @@ func (h ServiceIntegrationHandler) convert(i client.Object) (*v1alpha1.ServiceIn
 
 	return si, nil
 }
+
+// findExistingIntegration checks if an integration with matching configuration already exists on Aiven.
+func (h ServiceIntegrationHandler) findExistingIntegration(
+	ctx context.Context,
+	avnGen avngen.Client,
+	si *v1alpha1.ServiceIntegration,
+) (*service.ServiceIntegrationOut, error) {
+	if si.Spec.SourceServiceName == "" {
+		return nil, nil // integration with only endpoints, cannot list integrations
+	}
+
+	sourceProject := si.Spec.SourceProjectName
+	if sourceProject == "" {
+		sourceProject = si.Spec.Project
+	}
+
+	svc, err := avnGen.ServiceGet(ctx, sourceProject, si.Spec.SourceServiceName)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get service %s/%s: %w", sourceProject, si.Spec.SourceServiceName, err)
+	}
+
+	for _, integration := range svc.ServiceIntegrations {
+		if h.integrationMatches(&integration, si) {
+			return &integration, nil
+		}
+	}
+
+	return nil, nil
+}
+
+type integrationKey struct {
+	IntegrationType  service.IntegrationType
+	SourceService    string
+	SourceProject    string
+	SourceEndpointID *string
+	DestService      *string
+	DestProject      string
+	DestEndpointID   *string
+}
+
+func (h ServiceIntegrationHandler) integrationMatches(
+	existing *service.ServiceIntegrationOut,
+	desired *v1alpha1.ServiceIntegration,
+) bool {
+	sourceProject := desired.Spec.SourceProjectName
+	if sourceProject == "" {
+		sourceProject = desired.Spec.Project
+	}
+
+	destProject := desired.Spec.DestinationProjectName
+	if destProject == "" {
+		destProject = desired.Spec.Project
+	}
+
+	existingKey := integrationKey{
+		IntegrationType:  existing.IntegrationType,
+		SourceService:    existing.SourceService,
+		SourceProject:    existing.SourceProject,
+		SourceEndpointID: existing.SourceEndpointId,
+		DestService:      existing.DestService,
+		DestProject:      existing.DestProject,
+		DestEndpointID:   existing.DestEndpointId,
+	}
+
+	desiredKey := integrationKey{
+		IntegrationType:  desired.Spec.IntegrationType,
+		SourceService:    desired.Spec.SourceServiceName,
+		SourceProject:    sourceProject,
+		SourceEndpointID: NilIfZero(desired.Spec.SourceEndpointID),
+		DestService:      NilIfZero(desired.Spec.DestinationServiceName),
+		DestProject:      destProject,
+		DestEndpointID:   NilIfZero(desired.Spec.DestinationEndpointID),
+	}
+
+	return cmp.Equal(existingKey, desiredKey)
+}