Update telemetry schema with workflows

Closes: GH-267
fluxninja · Sep 7, 2022 · e2ad20f · e2ad20f
1 parent 5f0cd38
commit e2ad20f
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 5 deletions.
diff --git a/pkg/otelcollector/consts.go b/pkg/otelcollector/consts.go
@@ -63,6 +63,10 @@ const (
 	ConcurrencyLimitersLabel = "concurrency_limiters"
 	// DroppingConcurrencyLimitersLabel describes rate limiters dropping the traffic.
 	DroppingConcurrencyLimitersLabel = "dropping_concurrency_limiters"
+	// WorkloadsLabel describes workloads matched to the traffic.
+	WorkloadsLabel = "workloads"
+	// DroppingWorkloadsLabel describes workloads dropping the traffic.
+	DroppingWorkloadsLabel = "dropping_workloads"
 	// FluxMetersLabel describes flux meters matched to the traffic.
 	FluxMetersLabel = "flux_meters"
 	// FlowLabelKeysLabel describes keys of flow labels matched to the traffic.

diff --git a/pkg/otelcollector/metricsprocessor/processor.go b/pkg/otelcollector/metricsprocessor/processor.go
@@ -176,21 +176,29 @@ func (p *metricsProcessor) addCheckResponseBasedLabels(attributes pcommon.Map, c
 			rawValue := []string{
 				fmt.Sprintf("%s:%v", metrics.PolicyNameLabel, decision.GetPolicyName()),
 				fmt.Sprintf("%s:%v", metrics.ComponentIndexLabel, decision.GetComponentIndex()),
-				fmt.Sprintf("%s:%v", metrics.WorkloadIndexLabel, cl.GetWorkloadIndex()),
 				fmt.Sprintf("%s:%v", metrics.PolicyHashLabel, decision.GetPolicyHash()),
 			}
 			value := strings.Join(rawValue, ",")
 			labels[otelcollector.ConcurrencyLimitersLabel].SliceVal().AppendEmpty().SetStringVal(value)
 			if decision.Dropped {
 				labels[otelcollector.DroppingConcurrencyLimitersLabel].SliceVal().AppendEmpty().SetStringVal(value)
 			}
+
+			workloadsRawValue := []string{
+				fmt.Sprintf("%s:%v", metrics.PolicyNameLabel, decision.GetPolicyName()),
+				fmt.Sprintf("%s:%v", metrics.ComponentIndexLabel, decision.GetComponentIndex()),
+				fmt.Sprintf("%s:%v", metrics.WorkloadIndexLabel, cl.GetWorkloadIndex()),
+				fmt.Sprintf("%s:%v", metrics.PolicyHashLabel, decision.GetPolicyHash()),
+			}
+			value = strings.Join(workloadsRawValue, ",")
+			labels[otelcollector.WorkloadsLabel].SliceVal().AppendEmpty().SetStringVal(value)
+			if decision.Dropped {
+				labels[otelcollector.DroppingWorkloadsLabel].SliceVal().AppendEmpty().SetStringVal(value)
+			}
 		}
 	}
 	for _, fluxMeter := range checkResponse.FluxMeters {
-		rawValue := []string{
-			fmt.Sprintf("%s:%v", metrics.FluxMeterNameLabel, fluxMeter.GetFluxMeterName()),
-		}
-		value := strings.Join(rawValue, ",")
+		value := fluxMeter.GetFluxMeterName()
 		labels[otelcollector.FluxMetersLabel].SliceVal().AppendEmpty().SetStringVal(value)
 	}