chore(api): don't send event to public event integrations

Signed-off-by: francois  samin <[email protected]>

engine/api/event/event.go

@@ -19,7 +19,6 @@ import (
 
 // cache with go cache
 var brokersConnectionCache = gocache.New(10*time.Minute, 6*time.Hour)
-var publicBrokersConnectionCache = []Broker{}
 var hostname, cdsname string
 var brokers []Broker
 var subscribers []chan<- sdk.Event
@@ -45,30 +44,6 @@ func getBroker(ctx context.Context, t string, option interface{}) (Broker, error
 	return nil, fmt.Errorf("invalid Broker Type %s", t)
 }
 
-// ResetPublicIntegrations load all integration of type Event and creates kafka brokers
-func ResetPublicIntegrations(ctx context.Context, db *gorp.DbMap) error {
-	publicBrokersConnectionCache = []Broker{}
-	filterType := sdk.IntegrationTypeEvent
-	integrations, err := integration.LoadPublicModelsByTypeWithDecryption(db, &filterType)
-	if err != nil {
-		return sdk.WrapError(err, "cannot load public models for event type")
-	}
-
-	for _, integration := range integrations {
-		for _, cfg := range integration.PublicConfigurations {
-			kafkaCfg := getKafkaConfig(cfg)
-			kafkaBroker, err := getBroker(ctx, "kafka", kafkaCfg)
-			if err != nil {
-				return sdk.WrapError(err, "cannot get broker for %s and user %s", cfg["broker url"].Value, cfg["username"].Value)
-			}
-
-			publicBrokersConnectionCache = append(publicBrokersConnectionCache, kafkaBroker)
-		}
-	}
-
-	return nil
-}
-
 func getKafkaConfig(cfg sdk.IntegrationConfig) KafkaConfig {
 	kafkaCfg := KafkaConfig{
 		Enabled:         true,
@@ -138,7 +113,7 @@ func Initialize(ctx context.Context, db *gorp.DbMap, cache Store) error {
 		}
 	}
 
-	return ResetPublicIntegrations(ctx, db)
+	return nil
 }
 
 // Subscribe to CDS events
@@ -163,15 +138,10 @@ func DequeueEvent(ctx context.Context, db *gorp.DbMap) {
 			s <- e
 		}
 
-		// Send into public brokers
-		for _, b := range publicBrokersConnectionCache {
-			if err := b.sendEvent(&e); err != nil {
-				log.Warn(ctx, "Error while sending message [%s: %s/%s/%s/%s/%s]: %s", e.EventType, e.ProjectKey, e.WorkflowName, e.ApplicationName, e.PipelineName, e.EnvironmentName, err)
-			}
-		}
 		for _, eventIntegrationID := range e.EventIntegrationsID {
 			brokerConnectionKey := strconv.FormatInt(eventIntegrationID, 10)
 			brokerConnection, ok := brokersConnectionCache.Get(brokerConnectionKey)
+			var brokerConfig KafkaConfig
 			if !ok {
 				projInt, err := integration.LoadProjectIntegrationByIDWithClearPassword(ctx, db, eventIntegrationID)
 				if err != nil {
@@ -194,6 +164,7 @@ func DequeueEvent(ctx context.Context, db *gorp.DbMap) {
 					continue
 				}
 				brokerConnection = kafkaBroker
+				brokerConfig = kafkaCfg
 			}
 
 			broker, ok := brokerConnection.(Broker)
@@ -203,6 +174,7 @@ func DequeueEvent(ctx context.Context, db *gorp.DbMap) {
 			}
 
 			// Send into external brokers
+			log.Info(ctx, "sending event %q to %s", e.EventType, brokerConfig.BrokerAddresses)
 			if err := broker.sendEvent(&e); err != nil {
 				log.Warn(ctx, "Error while sending message [%s: %s/%s/%s/%s/%s]: %s", e.EventType, e.ProjectKey, e.WorkflowName, e.ApplicationName, e.PipelineName, e.EnvironmentName, err)
 			}

engine/api/integration.go

@@ -70,11 +70,6 @@ func (api *API) postIntegrationModelHandler() service.Handler {
 
 		if m.Public {
 			go propagatePublicIntegrationModel(ctx, api.mustDB(), api.Cache, *m, getAPIConsumer(ctx))
-			if m.Event {
-				if err := event.ResetPublicIntegrations(ctx, api.mustDB()); err != nil {
-					return sdk.WrapError(err, "error while resetting public integrations")
-				}
-			}
 		}
 
 		return service.WriteJSON(w, m, http.StatusCreated)
@@ -127,12 +122,6 @@ func (api *API) putIntegrationModelHandler() service.Handler {
 			api.GoRoutines.Exec(ctx, "propagatePublicIntegrationModel", func(ctx context.Context) {
 				propagatePublicIntegrationModel(ctx, api.mustDB(), api.Cache, *m, getAPIConsumer(ctx))
 			})
-
-			if m.Event {
-				if err := event.ResetPublicIntegrations(ctx, api.mustDB()); err != nil {
-					return sdk.WrapError(err, "error while resetting public integrations")
-				}
-			}
 		}
 
 		return service.WriteJSON(w, m, http.StatusOK)
@@ -231,13 +220,6 @@ func (api *API) deleteIntegrationModelHandler() service.Handler {
 			return sdk.WrapError(err, "Unable to commit tx")
 		}
 
-		if old.Event && old.Public {
-			// reset outside the transaction
-			if err := event.ResetPublicIntegrations(ctx, api.mustDB()); err != nil {
-				return sdk.WrapError(err, "error while resetting public integrations")
-			}
-		}
-
 		return nil
 	}
 }

engine/api/integration/dao_model.go

@@ -39,46 +39,6 @@ func LoadModels(db gorp.SqlExecutor) ([]sdk.IntegrationModel, error) {
 	return res, nil
 }
 
-func LoadPublicModelsByTypeWithDecryption(db gorp.SqlExecutor, integrationType *sdk.IntegrationType) ([]sdk.IntegrationModel, error) {
-	q := "SELECT * from integration_model WHERE public = true"
-	if integrationType != nil {
-		switch *integrationType {
-		case sdk.IntegrationTypeEvent:
-			q += " AND integration_model.event = true"
-		case sdk.IntegrationTypeCompute:
-			q += " AND integration_model.compute = true"
-		case sdk.IntegrationTypeStorage:
-			q += " AND integration_model.storage = true"
-		case sdk.IntegrationTypeHook:
-			q += " AND integration_model.hook = true"
-		case sdk.IntegrationTypeDeployment:
-			q += " AND integration_model.deployment = true"
-		}
-	}
-
-	query := gorpmapping.NewQuery(q)
-	var pms integrationModelSlice
-
-	if err := gorpmapping.GetAll(context.Background(), db, query, &pms, gorpmapping.GetOptions.WithDecryption); err != nil {
-		return nil, err
-	}
-
-	var res []sdk.IntegrationModel
-	for _, pm := range pms {
-		isValid, err := gorpmapping.CheckSignature(pm, pm.Signature)
-		if err != nil {
-			return nil, err
-		}
-		if !isValid {
-			log.Error(context.Background(), "integration.LoadModel> model  %d data corrupted", pm.ID)
-			continue
-		}
-		res = append(res, pm.IntegrationModel)
-	}
-
-	return res, nil
-}
-
 // LoadModel Load a integration model by its ID
 func LoadModel(ctx context.Context, db gorp.SqlExecutor, modelID int64) (sdk.IntegrationModel, error) {
 	query := gorpmapping.NewQuery("SELECT * from integration_model where id = $1").Args(modelID)

engine/api/integration/dao_model_test.go

@@ -42,8 +42,4 @@ func TestCRUDModel(t *testing.T) {
 	require.NoError(t, err)
 
 	assert.True(t, len(models) > 1)
-
-	filter := sdk.IntegrationTypeEvent
-	_, err = LoadPublicModelsByTypeWithDecryption(db, &filter)
-	require.NoError(t, err)
 }