[metrics_generation_processor] Add metrics generation logic

processor/metricsgenerationprocessor/README.md

@@ -26,6 +26,9 @@ processors:
               # Name of the new metric. This is a required field.
             - name: <new_metric_name>
 
+              # Unit for the new metric being generated.
+              unit: <new_metric_unit>
+
               # type describes how the new metric will be generated. It can be one of `calculate` or `scale`.  calculate generates a metric applying the given operation on two operand metrics. scale operates only on operand1 metric to generate the new metric.
               type: {calculate, scale}
 
@@ -57,6 +60,7 @@ rules:
 # create pod.memory.usage.bytes from pod.memory.usage.megabytes
 rules:
     - name: pod.memory.usage.bytes
+      unit: Bytes
       type: scale
       metric1: pod.memory.usage.megabytes
       operation: multiply

processor/metricsgenerationprocessor/config.go

@@ -53,6 +53,9 @@ type Rule struct {
 	// Name of the new metric being generated. This is a required field.
 	Name string `mapstructure:"name"`
 
+	// Unit for the new metric being generated.
+	Unit string `mapstructure:"unit"`
+
 	// The rule type following which the new metric will be generated. This is a required field.
 	Type GenerationType `mapstructure:"type"`
 

processor/metricsgenerationprocessor/config_test.go

@@ -38,13 +38,15 @@ func TestLoadingFullConfig(t *testing.T) {
 				Rules: []Rule{
 					{
 						Name:      "new_metric",
+						Unit:      "percent",
 						Type:      "calculate",
 						Metric1:   "metric1",
 						Metric2:   "metric2",
 						Operation: "percent",
 					},
 					{
 						Name:      "new_metric",
+						Unit:      "unit",
 						Type:      "scale",
 						Metric1:   "metric1",
 						ScaleBy:   1000,

processor/metricsgenerationprocessor/factory.go

@@ -73,6 +73,7 @@ func buildInternalConfig(config *Config) []internalRule {
 	for i, rule := range config.Rules {
 		customRule := internalRule{
 			name:      rule.Name,
+			unit:      rule.Unit,
 			ruleType:  string(rule.Type),
 			metric1:   rule.Metric1,
 			metric2:   rule.Metric2,

processor/metricsgenerationprocessor/processor.go

@@ -32,6 +32,7 @@ var _ processorhelper.MProcessor = (*metricsGenerationProcessor)(nil)
 
 type internalRule struct {
 	name      string
+	unit      string
 	ruleType  string
 	metric1   string
 	metric2   string
@@ -53,6 +54,37 @@ func (mgp *metricsGenerationProcessor) Start(context.Context, component.Host) er
 
 // ProcessMetrics implements the MProcessor interface.
 func (mgp *metricsGenerationProcessor) ProcessMetrics(_ context.Context, md pdata.Metrics) (pdata.Metrics, error) {
+	resourceMetricsSlice := md.ResourceMetrics()
+
+	for i := 0; i < resourceMetricsSlice.Len(); i++ {
+		rm := resourceMetricsSlice.At(i)
+		nameToMetricMap := getNameToMetricMap(rm)
+
+		for _, rule := range mgp.rules {
+			operand2 := float64(0)
+			_, ok := nameToMetricMap[rule.metric1]
+			if !ok {
+				mgp.logger.Debug("Missing first metric", zap.String("metric_name", rule.metric1))
+				continue
+			}
+
+			if rule.ruleType == string(calculate) {
+				metric2, ok := nameToMetricMap[rule.metric2]
+				if !ok {
+					mgp.logger.Debug("Missing second metric", zap.String("metric_name", rule.metric2))
+					continue
+				}
+				operand2 = getMetricValue(metric2)
+				if operand2 <= 0 {
+					continue
+				}
+
+			} else if rule.ruleType == string(scale) {
+				operand2 = rule.scaleBy
+			}
+			generateMetrics(rm, operand2, rule, mgp.logger)
+		}
+	}
 	return md, nil
 }