Use dynamic config reload for subnet labels

Use explicit dynamic vs static config flag

Refactor flp_pipeline_builder.go to break down stages config into
smaller functions

Author: jotak

internal/controller/flp/flp_common_objects.go

@@ -266,16 +266,17 @@ func metricsSettings(desired *flowslatest.FlowCollectorSpec, vol *volumes.Builde
 	return metricsSettings
 }
 
-func getStaticJSONConfig(desired *flowslatest.FlowCollectorSpec, vol *volumes.Builder, promTLS *flowslatest.CertificateReference, pipeline *PipelineBuilder, dynCMName string) (string, error) {
+func getJSONConfigs(desired *flowslatest.FlowCollectorSpec, vol *volumes.Builder, promTLS *flowslatest.CertificateReference, pipeline *PipelineBuilder, dynCMName string) (string, string, error) {
 	metricsSettings := metricsSettings(desired, vol, promTLS)
 	advancedConfig := helper.GetAdvancedProcessorConfig(desired)
+	static, dynamic := pipeline.GetSplitStageParams()
 	config := map[string]interface{}{
 		"log-level": desired.Processor.LogLevel,
 		"health": map[string]interface{}{
 			"port": *advancedConfig.HealthPort,
 		},
 		"pipeline":        pipeline.GetStages(),
-		"parameters":      pipeline.GetStaticStageParams(),
+		"parameters":      static,
 		"metricsSettings": metricsSettings,
 		"dynamicParameters": config.DynamicParameters{
 			Namespace: desired.Namespace,
@@ -288,22 +289,19 @@ func getStaticJSONConfig(desired *flowslatest.FlowCollectorSpec, vol *volumes.Bu
 			"port": *advancedConfig.ProfilePort,
 		}
 	}
-	bs, err := json.Marshal(config)
+	jsonStatic, err := json.Marshal(config)
 	if err != nil {
-		return "", err
+		return "", "", err
 	}
-	return string(bs), nil
-}
 
-func getDynamicJSONConfig(pipeline *PipelineBuilder) (string, error) {
-	config := map[string]interface{}{
-		"parameters": pipeline.GetDynamicStageParams(),
+	config = map[string]interface{}{
+		"parameters": dynamic,
 	}
-	bs, err := json.Marshal(config)
+	jsonDynamic, err := json.Marshal(config)
 	if err != nil {
-		return "", err
+		return "", "", err
 	}
-	return string(bs), nil
+	return string(jsonStatic), string(jsonDynamic), nil
 }
 
 func promService(desired *flowslatest.FlowCollectorSpec, svcName, namespace, appLabel string) *corev1.Service {

internal/controller/flp/flp_monolith_objects.go

@@ -121,38 +121,30 @@ func (b *monolithBuilder) deployment(annotations map[string]string) *appsv1.Depl
 }
 
 func (b *monolithBuilder) configMaps() (*corev1.ConfigMap, string, *corev1.ConfigMap, error) {
-	grpcStage := newGRPCPipeline(b.desired, &b.volumes)
-	pipeline := newPipelineBuilder(
+	pipeline, err := createPipeline(
 		b.desired,
 		b.flowMetrics,
 		b.fcSlices,
 		b.detectedSubnets,
 		b.info.Loki,
 		b.info.ClusterInfo.GetID(),
 		&b.volumes,
-		&grpcStage,
+		newGRPCPipeline(b.desired, &b.volumes),
 	)
-	err := pipeline.AddProcessorStages()
 	if err != nil {
 		return nil, "", nil, err
 	}
 
-	// Get static CM
-	data, err := getStaticJSONConfig(b.desired, &b.volumes, b.promTLS, &pipeline, monoDynConfigMap)
+	// Get static and dynamic CM
+	static, dynamic, err := getJSONConfigs(b.desired, &b.volumes, b.promTLS, pipeline, monoDynConfigMap)
 	if err != nil {
 		return nil, "", nil, err
 	}
-	staticCM, digest, err := configMap(monoConfigMap, b.info.Namespace, data, monoName)
+	staticCM, digest, err := configMap(monoConfigMap, b.info.Namespace, static, monoName)
 	if err != nil {
 		return nil, "", nil, err
 	}
-
-	// Get dynamic CM (hot reload)
-	data, err = getDynamicJSONConfig(&pipeline)
-	if err != nil {
-		return nil, "", nil, err
-	}
-	dynamicCM, _, err := configMap(monoDynConfigMap, b.info.Namespace, data, monoName)
+	dynamicCM, _, err := configMap(monoDynConfigMap, b.info.Namespace, dynamic, monoName)
 	if err != nil {
 		return nil, "", nil, err
 	}