fix: Merge integration subscriptions into one, apply newly supplied s…

…ubscriptions if existing ones are empty (#8597)
keptn · Aug 8, 2022 · 1cdd2a9 · 1cdd2a9
1 parent defe0b6
commit 1cdd2a9
Show file tree

Hide file tree

Showing 2 changed files with 184 additions and 12 deletions.
diff --git a/shipyard-controller/handler/uniformintegrationhandler.go b/shipyard-controller/handler/uniformintegrationhandler.go
@@ -130,16 +130,7 @@ func (rh *UniformIntegrationHandler) Register(c *gin.Context) {
 	logger.Debugf("Uniform:Register(): Checking for existing integration for %s", integrationInfo)
 	if err == nil && len(existingIntegrations) > 0 {
 		logger.Debugf("Uniform:Register(): Found existing integration for %s with id %s", integrationInfo, existingIntegrations[0].ID)
-		integration.ID = existingIntegrations[0].ID
-
-		err2 := rh.updateExistingIntegration(integration)
-		if err2 != nil {
-			SetInternalServerErrorResponse(c, err2.Error())
-			return
-		}
-		c.JSON(http.StatusOK, &models.RegisterResponse{
-			ID: integration.ID,
-		})
+		rh.updateIntegration(c, existingIntegrations, integrationInfo, integration)
 		return
 	}
 
@@ -175,7 +166,7 @@ func (rh *UniformIntegrationHandler) Register(c *gin.Context) {
 	if err != nil {
 		// if the integration already exists, update only needed fields
 		if errors.Is(err, db.ErrUniformRegistrationAlreadyExists) {
-			err2 := rh.updateExistingIntegration(integration)
+			err2 := rh.updateIntegrationMetadata(integration)
 			if err2 != nil {
 				SetInternalServerErrorResponse(c, err2.Error())
 				return
@@ -194,7 +185,30 @@ func (rh *UniformIntegrationHandler) Register(c *gin.Context) {
 	})
 }
 
-func (rh *UniformIntegrationHandler) updateExistingIntegration(integration *apimodels.Integration) error {
+func (rh *UniformIntegrationHandler) updateIntegration(c *gin.Context, existingIntegrations []apimodels.Integration, integrationInfo string, newIntegration *apimodels.Integration) {
+	var existingIntegration *apimodels.Integration
+	// if we get multiple results, merge them into one - this can be the case during an upgrade where a new version of an integration
+	// re-registered itself while the shipyard controller was not running the latest version yet
+
+	existingIntegration, err := rh.mergeIntegrationSubscriptions(existingIntegrations, newIntegration)
+	if err != nil {
+		SetInternalServerErrorResponse(c, err.Error())
+		return
+	}
+
+	logger.Debugf("Uniform:Register(): Found existing integration for %s with id %s", integrationInfo, existingIntegrations[0].ID)
+	newIntegration.ID = existingIntegration.ID
+
+	if err = rh.updateIntegrationMetadata(newIntegration); err != nil {
+		SetInternalServerErrorResponse(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, &models.RegisterResponse{
+		ID: newIntegration.ID,
+	})
+}
+
+func (rh *UniformIntegrationHandler) updateIntegrationMetadata(integration *apimodels.Integration) error {
 	var err error
 	result, err := rh.uniformRepo.GetUniformIntegrations(models.GetUniformIntegrationsParams{ID: integration.ID})
 
@@ -485,3 +499,71 @@ func (rh *UniformIntegrationHandler) GetSubscriptions(c *gin.Context) {
 	}
 	c.JSON(http.StatusOK, subscriptions)
 }
+
+func (rh *UniformIntegrationHandler) mergeIntegrationSubscriptions(integrations []apimodels.Integration, newIntegration *apimodels.Integration) (*apimodels.Integration, error) {
+	// first, determine the target integration - i.e. the one which was seen most recently
+	targetIntegration, err := getMostRecentIntegration(integrations)
+	if err != nil {
+		return nil, err
+	}
+
+	// put the subscriptions of the outdated integrations into the target integration
+	// only update the subscriptions if we actually took some subscriptions
+	updateSubscriptions := false
+	for _, integration := range integrations {
+		if len(integration.Subscriptions) > 0 && integration.ID != targetIntegration.ID {
+			subscriptionsAdded := false
+			targetIntegration.Subscriptions, subscriptionsAdded = adoptSubscriptions(targetIntegration.Subscriptions, integration.Subscriptions)
+			if subscriptionsAdded {
+				updateSubscriptions = true
+			}
+		}
+	}
+
+	// check if the target integration has no subscriptions' property set - if yes, apply the initial subscriptions from the newly registered integration
+	if targetIntegration.Subscriptions == nil || len(targetIntegration.Subscriptions) == 0 {
+		targetIntegration.Subscriptions = newIntegration.Subscriptions
+		updateSubscriptions = true
+	}
+
+	if updateSubscriptions {
+		if err := rh.uniformRepo.CreateOrUpdateUniformIntegration(*targetIntegration); err != nil {
+			return nil, err
+		}
+	}
+	return targetIntegration, nil
+}
+
+func adoptSubscriptions(target []apimodels.EventSubscription, subscriptions []apimodels.EventSubscription) ([]apimodels.EventSubscription, bool) {
+	addedSubscription := false
+	for _, sub := range subscriptions {
+		skipSubscription := false
+		for _, existingSubscription := range target {
+			// if the subscription is already present, we don't add it
+			if existingSubscription.ID == sub.ID {
+				skipSubscription = true
+				break
+			}
+		}
+		if skipSubscription {
+			continue
+		}
+		addedSubscription = true
+		target = append(target, sub)
+	}
+	return target, addedSubscription
+}
+
+func getMostRecentIntegration(integrations []apimodels.Integration) (*apimodels.Integration, error) {
+	if len(integrations) == 0 {
+		return nil, errors.New("list of integrations is empty")
+	}
+	targetIntegration := integrations[0]
+
+	for _, integration := range integrations {
+		if integration.MetaData.LastSeen.After(targetIntegration.MetaData.LastSeen) {
+			targetIntegration = integration
+		}
+	}
+	return &targetIntegration, nil
+}
diff --git a/shipyard-controller/handler/uniformintegrationhandler_test.go b/shipyard-controller/handler/uniformintegrationhandler_test.go
@@ -8,6 +8,7 @@ import (
 	"net/http/httptest"
 	"reflect"
 	"testing"
+	"time"
 
 	"github.com/gin-gonic/gin"
 	apimodels "github.com/keptn/go-utils/pkg/api/models"
@@ -269,6 +270,95 @@ func TestUniformIntegrationHandler_Register(t *testing.T) {
 
 }
 
+func TestUniformIntegrationHandler_RegisterMergeIntegrationInstancesNoExistingSubscriptions(t *testing.T) {
+	mockRepo := &db_mock.UniformRepoMock{
+		CreateUniformIntegrationFunc: func(integration apimodels.Integration) error { return db.ErrUniformRegistrationAlreadyExists },
+		CreateOrUpdateUniformIntegrationFunc: func(integration apimodels.Integration) error {
+			return nil
+		},
+		UpdateLastSeenFunc: func(integrationID string) (*apimodels.Integration, error) {
+			return nil, nil
+		},
+		UpdateVersionInfoFunc: func(integrationID string, integrationVersion string, distributorVersion string) (*apimodels.Integration, error) {
+			return nil, nil
+		},
+		GetUniformIntegrationsFunc: func(filter models.GetUniformIntegrationsParams) ([]apimodels.Integration, error) {
+			return []apimodels.Integration{
+				{
+					ID:   "123",
+					Name: "my-integration",
+					MetaData: apimodels.MetaData{
+						Hostname:           "my-host",
+						DistributorVersion: "0.16.0",
+						KubernetesMetaData: apimodels.KubernetesMetaData{
+							Namespace: "my-namespace",
+						},
+						LastSeen: time.Now(),
+					},
+					Subscriptions: []apimodels.EventSubscription{},
+				},
+				{
+					ID:   "456",
+					Name: "my-integration",
+					MetaData: apimodels.MetaData{
+						Hostname:           "my-host",
+						DistributorVersion: "0.17.0",
+						KubernetesMetaData: apimodels.KubernetesMetaData{
+							Namespace: "my-namespace",
+						},
+						LastSeen: time.Now().Add(10 * time.Second),
+					},
+					Subscriptions: []apimodels.EventSubscription{},
+				},
+			}, nil
+		},
+	}
+
+	h := handler.NewUniformIntegrationHandler(mockRepo)
+
+	router := gin.Default()
+	router.POST("/uniform/registration", func(c *gin.Context) {
+		h.Register(c)
+	})
+
+	myIntegration := &apimodels.Integration{
+		Name: "my-integration",
+		MetaData: apimodels.MetaData{
+			Hostname:           "my-host",
+			DistributorVersion: "0.17.0",
+			KubernetesMetaData: apimodels.KubernetesMetaData{
+				Namespace: "my-namespace",
+			},
+		},
+		Subscriptions: []apimodels.EventSubscription{
+			{
+				Event: "sh.keptn.event.test.triggered",
+			},
+		},
+	}
+	payload, _ := json.Marshal(myIntegration)
+
+	request := httptest.NewRequest("POST", "/uniform/registration", bytes.NewBuffer(payload))
+
+	resp := performRequest(router, request)
+
+	require.Equal(t, http.StatusOK, resp.Code)
+
+	require.Len(t, mockRepo.CreateOrUpdateUniformIntegrationCalls(), 1)
+
+	// the update should have been performed for the integration which has been seen most recently
+	require.Equal(t, "456", mockRepo.CreateOrUpdateUniformIntegrationCalls()[0].Integration.ID)
+
+	require.Len(t, mockRepo.CreateOrUpdateUniformIntegrationCalls()[0].Integration.Subscriptions, 1)
+
+	// in this case we want to apply the subscriptions from the new integration, since the existing integrations did not have any subscriptions
+	require.ElementsMatch(t, mockRepo.CreateOrUpdateUniformIntegrationCalls()[0].Integration.Subscriptions, []apimodels.EventSubscription{
+		{
+			Event: "sh.keptn.event.test.triggered",
+		},
+	})
+}
+
 func TestUniformIntegrationKeepAlive(t *testing.T) {
 
 	existingIntegration := &apimodels.Integration{