fix and unify IDs

Author: tarokkk

pkg/resources/model/system.go

@@ -190,7 +190,8 @@ func FlowDispatcher(flowCr interface{}) (*CommonFlow, error) {
 				},
 			}
 		}
-		flow, err := types.NewFlow(matches, f.Name, f.Namespace)
+		id := fmt.Sprintf("clusterflow" + ":" + f.Namespace + ":" + f.Name)
+		flow, err := types.NewFlow(matches, id, f.Name, f.Namespace)
 		if err != nil {
 			return nil, err
 		}
@@ -223,7 +224,8 @@ func FlowDispatcher(flowCr interface{}) (*CommonFlow, error) {
 				},
 			}
 		}
-		flow, err := types.NewFlow(matches, f.Name, f.Namespace)
+		id := fmt.Sprintf("flow" + ":" + f.Namespace + ":" + f.Name)
+		flow, err := types.NewFlow(matches, id, f.Name, f.Namespace)
 		commonFlow.Flow = flow
 		if err != nil {
 			return nil, err
@@ -240,15 +242,22 @@ func (l *LoggingResources) CreateFlowFromCustomResource(flowCr interface{}) (*ty
 	}
 	flow := commonFlow.Flow
 	outputs := []types.Output{}
+	var flowType string
+	if commonFlow.Scope != "" {
+		flowType = "flow"
+	} else {
+		flowType = "clusterflow"
+	}
 	var multierr error
 FindOutputForAllRefs:
 	for _, outputRef := range commonFlow.OutputRefs {
 		// only namespaced flows should use namespaced outputs
 		if commonFlow.Scope != "" {
 			for _, output := range l.Outputs {
 				// only an output from the same namespace can be used with a matching name
+				// flow -> output (matching)
 				if output.Namespace == commonFlow.Scope && outputRef == output.Name {
-					outputId := commonFlow.Scope + "_" + commonFlow.Name + "_" + output.Name
+					outputId := fmt.Sprintf("%s:%s:%s:output:%s:%s", flowType, commonFlow.Namespace, commonFlow.Name, output.Namespace, output.Name)
 					plugin, err := plugins.CreateOutput(output.Spec, outputId, secret.NewSecretLoader(l.client, output.Namespace, fluentd.OutputSecretPath, l.Secrets))
 					if err != nil {
 						multierr = errors.Combine(multierr, errors.WrapIff(err, "failed to create configured output %s", outputRef))
@@ -261,7 +270,9 @@ FindOutputForAllRefs:
 		}
 		for _, clusterOutput := range l.ClusterOutputs {
 			if outputRef == clusterOutput.Name {
-				outputId := commonFlow.Namespace + "_" + commonFlow.Name + "_" + clusterOutput.Name
+				// flow, clusterflow -> clusterOutput
+				// diff flow / clusterflow based on scope
+				outputId := fmt.Sprintf("%s:%s:%s:clusteroutput:%s:%s", flowType, commonFlow.Namespace, commonFlow.Name, clusterOutput.Namespace, clusterOutput.Name)
 				plugin, err := plugins.CreateOutput(clusterOutput.Spec.OutputSpec, outputId, secret.NewSecretLoader(l.client, clusterOutput.Namespace, fluentd.OutputSecretPath, l.Secrets))
 				if err != nil {
 					multierr = errors.Combine(multierr, errors.WrapIff(err, "failed to create configured output %s", outputRef))
@@ -278,7 +289,7 @@ FindOutputForAllRefs:
 	// Filter
 	var filters []types.Filter
 	for i, f := range commonFlow.Filters {
-		id := fmt.Sprintf("%s_%s_%d", commonFlow.Namespace, commonFlow.Name, i)
+		id := fmt.Sprintf("%s:%s:%s:%d", flowType, commonFlow.Namespace, commonFlow.Name, i)
 		filter, err := plugins.CreateFilter(f, id, secret.NewSecretLoader(l.client, commonFlow.Namespace, fluentd.OutputSecretPath, l.Secrets))
 		if err != nil {
 			multierr = errors.Combine(multierr, errors.WrapIff(err, "failed to create filter with index %d for flow %s", i, commonFlow.Name))

pkg/sdk/model/filter/concat.go

@@ -101,7 +101,7 @@ func (c *Concat) ToDirective(secretLoader secret.SecretLoader, id string) (types
 			Type:      pluginType,
 			Directive: "filter",
 			Tag:       "**",
-			Id:        id + "_" + pluginType,
+			Id:        id,
 		},
 	}
 	concatConfig := c.DeepCopy()

pkg/sdk/model/filter/dedot.go

@@ -80,6 +80,6 @@ func (c *DedotFilterConfig) ToDirective(secretLoader secret.SecretLoader, id str
 		Type:      pluginType,
 		Directive: "filter",
 		Tag:       "**",
-		Id:        id + "_" + pluginType,
+		Id:        id,
 	}, c, secretLoader)
 }

pkg/sdk/model/filter/detect_exceptions.go

@@ -96,13 +96,12 @@ type _expDetectExceptions interface{}
 
 func (d *DetectExceptions) ToDirective(secretLoader secret.SecretLoader, id string) (types.Directive, error) {
 	pluginType := "detect_exceptions"
-	pluginID := id + "_" + pluginType
 	detector := &types.OutputPlugin{
 		PluginMeta: types.PluginMeta{
 			Type:      pluginType,
 			Directive: "match",
 			Tag:       "kubernetes.**",
-			Id:        pluginID,
+			Id:        id,
 		},
 	}
 	detect := d.DeepCopy()

pkg/sdk/model/filter/geoip.go

@@ -97,7 +97,7 @@ func (g *GeoIP) ToDirective(secretLoader secret.SecretLoader, id string) (types.
 			Type:      pluginType,
 			Directive: "filter",
 			Tag:       "**",
-			Id:        id + "_" + pluginType,
+			Id:        id,
 		},
 	}
 	if params, err := types.NewStructToStringMapper(secretLoader).StringsMap(g); err != nil {

pkg/sdk/model/filter/grep.go

@@ -298,7 +298,7 @@ func (g *GrepConfig) ToDirective(secretLoader secret.SecretLoader, id string) (t
 			Type:      pluginType,
 			Directive: "filter",
 			Tag:       "**",
-			Id:        id + "_" + pluginType,
+			Id:        id,
 		},
 	}
 	if len(g.Regexp) > 0 {

pkg/sdk/model/filter/parser.go

@@ -191,7 +191,7 @@ func (p *ParserConfig) ToDirective(secretLoader secret.SecretLoader, id string)
 			Type:      pluginType,
 			Directive: "filter",
 			Tag:       "**",
-			Id:        id + "_" + pluginType,
+			Id:        id,
 		},
 	}
 	parserConfig := p.DeepCopy()

pkg/sdk/model/filter/prometheus.go

@@ -156,7 +156,7 @@ func (p *PrometheusConfig) ToDirective(secretLoader secret.SecretLoader, id stri
 			Type:      pluginType,
 			Directive: "filter",
 			Tag:       "**",
-			Id:        id + "_" + pluginType,
+			Id:        id,
 		},
 	}
 

pkg/sdk/model/filter/record_modifier.go

@@ -106,7 +106,7 @@ func (r *RecordModifier) ToDirective(secretLoader secret.SecretLoader, id string
 			Type:      pluginType,
 			Directive: "filter",
 			Tag:       "**",
-			Id:        id + "_" + pluginType,
+			Id:        id,
 		},
 	}
 	if params, err := types.NewStructToStringMapper(secretLoader).StringsMap(r); err != nil {

pkg/sdk/model/filter/record_transformer.go

@@ -103,7 +103,7 @@ func (r *RecordTransformer) ToDirective(secretLoader secret.SecretLoader, id str
 			Type:      pluginType,
 			Directive: "filter",
 			Tag:       "**",
-			Id:        id + "_" + pluginType,
+			Id:        id,
 		},
 	}
 	if params, err := types.NewStructToStringMapper(secretLoader).StringsMap(r); err != nil {