fix: added kafka_topic and kube_cluster in labels to support list fil…

…ters (#78) * fix: fixed checks for list firehose handler * fix: fixed lint errors * feat(firehose): use same label builder for create/edit operations --------- Co-authored-by: Ayushi Sharma <[email protected]> Co-authored-by: Stewart Jingga <[email protected]>
goto · Oct 24, 2023 · cf39840 · cf39840
1 parent 8f39ef9
commit cf39840
Show file tree

Hide file tree

Showing 3 changed files with 34 additions and 50 deletions.
diff --git a/internal/server/v1/firehose/constant.go b/internal/server/v1/firehose/constant.go
@@ -17,12 +17,11 @@ const (
 // Some of firehose Configs used/modified in more than one place by dex
 // Refer https://goto.github.io/firehose/advance/generic/
 const (
-	configSourceKafkaTopic         = "SOURCE_KAFKA_TOPIC"
-	configSourceKafkaBrokers       = "SOURCE_KAFKA_BROKERS"
-	configSourceKafkaConsumerGroup = "SOURCE_KAFKA_CONSUMER_GROUP_ID"
-	configSinkType                 = "SINK_TYPE"
-	configStreamName               = "STREAM_NAME"
-	configStencilURL               = "SCHEMA_REGISTRY_STENCIL_URLS"
+	configSourceKafkaTopic   = "SOURCE_KAFKA_TOPIC"
+	configSourceKafkaBrokers = "SOURCE_KAFKA_BROKERS"
+	configSinkType           = "SINK_TYPE"
+	configStreamName         = "STREAM_NAME"
+	configStencilURL         = "SCHEMA_REGISTRY_STENCIL_URLS"
 )
 
 const (

diff --git a/internal/server/v1/firehose/crud.go b/internal/server/v1/firehose/crud.go
@@ -64,14 +64,7 @@ func (api *firehoseAPI) handleCreate(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	def.Labels = cloneAndMergeMaps(def.Labels, map[string]string{
-		labelTitle:       *def.Title,
-		labelGroup:       groupID,
-		labelTeam:        groupSlug,
-		labelStream:      *def.Configs.StreamName,
-		labelDescription: def.Description,
-		labelSinkType:    def.Configs.EnvVars["SINK_TYPE"],
-	})
+	def.Labels = cloneAndMergeMaps(def.Labels, api.buildLabels(def, groupSlug))
 
 	prj, err := project.GetProject(ctx, def.Project, api.Shield)
 	if err != nil {
@@ -180,43 +173,31 @@ func (api *firehoseAPI) handleList(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	includeEnv := []string{
-		configSinkType,
-		configSourceKafkaTopic,
-		configSourceKafkaConsumerGroup,
-	}
-
 	topicName := q.Get("topic_name")
 	kubeCluster := q.Get("kube_cluster")
 	sinkTypes := sinkTypeSet(q.Get("sink_type"))
-	var arr []models.Firehose
+
+	arr := []models.Firehose{}
 	for _, res := range rpcResp.GetResources() {
 		def, err := mapEntropyResourceToFirehose(res)
 		if err != nil {
 			utils.WriteErr(w, err)
 			return
 		}
 
-		if kubeCluster != "" && *def.Configs.KubeCluster != kubeCluster {
+		if kubeCluster != "" && def.Labels[labelKubeCluster] != kubeCluster {
 			continue
 		}
 
-		if topicName != "" && def.Configs.EnvVars[configSourceKafkaTopic] != topicName {
+		if topicName != "" && def.Labels[labelTopic] != topicName {
 			continue
 		}
 
-		_, include := sinkTypes[def.Configs.EnvVars[configSinkType]]
+		_, include := sinkTypes[def.Labels[labelSinkType]]
 		if len(sinkTypes) > 0 && !include {
 			continue
 		}
 
-		// only return selected keys to reduce list response-size.
-		returnEnv := map[string]string{}
-		for _, key := range includeEnv {
-			returnEnv[key] = def.Configs.EnvVars[key]
-		}
-		def.Configs.EnvVars = returnEnv
-
 		arr = append(arr, def)
 	}
 
@@ -264,20 +245,15 @@ func (api *firehoseAPI) handleUpdate(w http.ResponseWriter, r *http.Request) {
 		utils.WriteErr(w, err)
 		return
 	}
-	labels := cloneAndMergeMaps(existingFirehose.Labels, map[string]string{
-		labelGroup: groupID,
-		labelTeam:  groupSlug,
-	})
-	if updates.Description != "" {
-		labels[labelDescription] = existingFirehose.Description
-	}
 
 	err = api.buildEnvVars(r.Context(), &existingFirehose, reqCtx.UserID, false)
 	if err != nil {
 		utils.WriteErr(w, fmt.Errorf("error building env vars: %w", err))
 		return
 	}
 
+	labels := cloneAndMergeMaps(existingFirehose.Labels, api.buildLabels(existingFirehose, groupSlug))
+
 	cfgStruct, err := makeConfigStruct(existingFirehose.Configs)
 	if err != nil {
 		utils.WriteErr(w, err)
@@ -335,19 +311,16 @@ func (api *firehoseAPI) handlePartialUpdate(w http.ResponseWriter, r *http.Reque
 		return
 	}
 
-	labels := existing.Labels
-	if req.Description != "" {
-		labels[labelDescription] = req.Description
-	}
+	var groupSlug string
 	if req.Group != "" {
 		groupID := req.Group
-		groupSlug, err := api.getGroupSlug(ctx, groupID)
+		groupSlug, err = api.getGroupSlug(ctx, groupID)
 		if err != nil {
 			utils.WriteErr(w, err)
 			return
 		}
-		labels[labelGroup] = groupID
-		labels[labelTeam] = groupSlug
+	} else {
+		groupSlug = existing.Labels[labelGroup]
 	}
 
 	if req.Configs.Stopped != nil {
@@ -360,7 +333,6 @@ func (api *firehoseAPI) handlePartialUpdate(w http.ResponseWriter, r *http.Reque
 
 	if req.Configs.StreamName != "" {
 		existing.Configs.StreamName = &req.Configs.StreamName
-		labels[labelStream] = req.Configs.StreamName
 	}
 
 	if req.Configs.Replicas > 0 {
@@ -383,10 +355,6 @@ func (api *firehoseAPI) handlePartialUpdate(w http.ResponseWriter, r *http.Reque
 		}
 	}
 
-	if existing.Configs.EnvVars["SINK_TYPE"] != req.Configs.EnvVars["SINK_TYPE"] {
-		labels[labelSinkType] = req.Configs.EnvVars["SINK_TYPE"]
-	}
-
 	existing.Configs.EnvVars = cloneAndMergeMaps(
 		existing.Configs.EnvVars,
 		req.Configs.EnvVars,
@@ -400,6 +368,8 @@ func (api *firehoseAPI) handlePartialUpdate(w http.ResponseWriter, r *http.Reque
 		return
 	}
 
+	labels := cloneAndMergeMaps(existing.Labels, api.buildLabels(existing, groupSlug))
+
 	cfgStruct, err := makeConfigStruct(existing.Configs)
 	if err != nil {
 		utils.WriteErr(w, err)
@@ -436,6 +406,19 @@ func (api *firehoseAPI) handlePartialUpdate(w http.ResponseWriter, r *http.Reque
 	utils.WriteJSON(w, http.StatusOK, updatedFirehose)
 }
 
+func (*firehoseAPI) buildLabels(firehose models.Firehose, groupSlug string) map[string]string {
+	return map[string]string{
+		labelTitle:       *firehose.Title,
+		labelGroup:       string(*firehose.Group),
+		labelTeam:        groupSlug,
+		labelStream:      *firehose.Configs.StreamName,
+		labelDescription: firehose.Description,
+		labelSinkType:    firehose.Configs.EnvVars[configSinkType],
+		labelTopic:       firehose.Configs.EnvVars[configSourceKafkaTopic],
+		labelKubeCluster: *firehose.Configs.KubeCluster,
+	}
+}
+
 func (api *firehoseAPI) handleGetHistory(w http.ResponseWriter, r *http.Request) {
 	urn := chi.URLParam(r, pathParamURN)
 

diff --git a/internal/server/v1/firehose/mappings.go b/internal/server/v1/firehose/mappings.go
@@ -28,6 +28,8 @@ const (
 	labelStream      = "stream_name"
 	labelDescription = "description"
 	labelSinkType    = "sink_type"
+	labelKubeCluster = "kube_cluster"
+	labelTopic       = "topic"
 )
 
 var nonAlphaNumPattern = regexp.MustCompile("[^a-zA-Z0-9]+")