Replace telegraf prometheus plugin (#1709)

Author: dricross

plugins/plugins.go

@@ -7,6 +7,7 @@ import (
 	//Enable cloudwatch-agent process plugins
 	_ "github.com/aws/amazon-cloudwatch-agent/plugins/processors/ecsdecorator"
 	_ "github.com/aws/amazon-cloudwatch-agent/plugins/processors/k8sdecorator"
+	_ "github.com/aws/amazon-cloudwatch-agent/plugins/processors/prometheusadapter"
 
 	// Enabled cloudwatch-agent input plugins
 	_ "github.com/aws/amazon-cloudwatch-agent/plugins/inputs/logfile"

plugins/processors/prometheusadapter/config.go

@@ -0,0 +1,17 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package prometheusadapter
+
+import (
+	"go.opentelemetry.io/collector/component"
+)
+
+type Config struct{}
+
+// Verify Config implements Processor interface.
+var _ component.Config = (*Config)(nil)
+
+func (cfg *Config) Validate() error {
+	return nil
+}

plugins/processors/prometheusadapter/config_test.go

@@ -0,0 +1,18 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package prometheusadapter
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/confmap"
+)
+
+func TestUnmarshalDefaultConfig(t *testing.T) {
+	factory := NewFactory()
+	cfg := factory.CreateDefaultConfig()
+	assert.NoError(t, confmap.New().Unmarshal(cfg))
+	assert.Equal(t, factory.CreateDefaultConfig(), cfg)
+}

plugins/processors/prometheusadapter/factory.go

@@ -0,0 +1,56 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package prometheusadapter
+
+import (
+	"context"
+	"fmt"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/processor"
+	"go.opentelemetry.io/collector/processor/processorhelper"
+)
+
+const (
+	stability = component.StabilityLevelBeta
+)
+
+var (
+	TypeStr, _            = component.NewType("prometheusadapter")
+	processorCapabilities = consumer.Capabilities{MutatesData: true}
+)
+
+func NewFactory() processor.Factory {
+	return processor.NewFactory(
+		TypeStr,
+		createDefaultConfig,
+		processor.WithMetrics(createMetricsProcessor, stability))
+}
+
+func createDefaultConfig() component.Config {
+	return &Config{}
+}
+
+func createMetricsProcessor(
+	ctx context.Context,
+	set processor.Settings,
+	cfg component.Config,
+	nextConsumer consumer.Metrics,
+) (processor.Metrics, error) {
+	processorConfig, ok := cfg.(*Config)
+	if !ok {
+		return nil, fmt.Errorf("configuration parsing error")
+	}
+
+	metricsProcessor := newPrometheusAdapterProcessor(processorConfig, set.Logger)
+
+	return processorhelper.NewMetrics(
+		ctx,
+		set,
+		cfg,
+		nextConsumer,
+		metricsProcessor.processMetrics,
+		processorhelper.WithCapabilities(processorCapabilities))
+}

plugins/processors/prometheusadapter/factory_test.go

@@ -0,0 +1,45 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package prometheusadapter
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/pipeline"
+	"go.opentelemetry.io/collector/processor/processortest"
+)
+
+func TestCreateDefaultConfig(t *testing.T) {
+	factory := NewFactory()
+	require.NotNil(t, factory)
+
+	cfg := factory.CreateDefaultConfig()
+	assert.NotNil(t, cfg, "failed to create default config")
+	assert.NoError(t, componenttest.CheckConfigStruct(cfg))
+}
+
+func TestCreateProcessor(t *testing.T) {
+	factory := NewFactory()
+	require.NotNil(t, factory)
+
+	cfg := factory.CreateDefaultConfig()
+	setting := processortest.NewNopSettings()
+
+	tProcessor, err := factory.CreateTraces(context.Background(), setting, cfg, consumertest.NewNop())
+	assert.Equal(t, err, pipeline.ErrSignalNotSupported)
+	assert.Nil(t, tProcessor)
+
+	mProcessor, err := factory.CreateMetrics(context.Background(), setting, cfg, consumertest.NewNop())
+	assert.NoError(t, err)
+	assert.NotNil(t, mProcessor)
+
+	lProcessor, err := factory.CreateLogs(context.Background(), setting, cfg, consumertest.NewNop())
+	assert.Equal(t, err, pipeline.ErrSignalNotSupported)
+	assert.Nil(t, lProcessor)
+}

plugins/processors/prometheusadapter/processor.go

@@ -0,0 +1,28 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package prometheusadapter
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.uber.org/zap"
+)
+
+type prometheusAdapterProcessor struct {
+	*Config
+	logger *zap.Logger
+}
+
+func newPrometheusAdapterProcessor(config *Config, logger *zap.Logger) *prometheusAdapterProcessor {
+	d := &prometheusAdapterProcessor{
+		Config: config,
+		logger: logger,
+	}
+	return d
+}
+
+func (d *prometheusAdapterProcessor) processMetrics(_ context.Context, md pmetric.Metrics) (pmetric.Metrics, error) {
+	return md, nil
+}

plugins/processors/prometheusadapter/processor_test.go

@@ -0,0 +1,18 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package prometheusadapter
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.uber.org/zap"
+)
+
+func TestProcessMetricsForPrometheusAdapter(t *testing.T) {
+	logger, _ := zap.NewDevelopment()
+	pap := newPrometheusAdapterProcessor(createDefaultConfig().(*Config), logger)
+
+	assert.NotNil(t, pap)
+}

receiver/adapter/plugins_test.go

@@ -124,7 +124,7 @@ func scrapeMetrics(as *assert.Assertions, ctx context.Context, receiver *Adapted
 }
 
 func validateMetricName(as *assert.Assertions, plugin string, expectedResourceMetricsName []string, actualOtelSlMetrics pmetric.MetricSlice) {
-	as.Equal(len(expectedResourceMetricsName), actualOtelSlMetrics.Len(), "Number of metrics did not match!")
+	as.Equal(len(expectedResourceMetricsName), actualOtelSlMetrics.Len(), "Number of metrics did not match! expected %+v\nactual: %+v", expectedResourceMetricsName, actualOtelSlMetrics)
 
 	matchMetrics := actualOtelSlMetrics.Len()
 	for _, expectedMetric := range expectedResourceMetricsName {

service/defaultcomponents/components.go

@@ -65,6 +65,7 @@ import (
 	"github.com/aws/amazon-cloudwatch-agent/plugins/processors/ec2tagger"
 	"github.com/aws/amazon-cloudwatch-agent/plugins/processors/gpuattributes"
 	"github.com/aws/amazon-cloudwatch-agent/plugins/processors/kueueattributes"
+	"github.com/aws/amazon-cloudwatch-agent/plugins/processors/prometheusadapter"
 	"github.com/aws/amazon-cloudwatch-agent/processor/rollupprocessor"
 	"github.com/aws/amazon-cloudwatch-agent/receiver/awsebsnvmereceiver"
 )
@@ -112,6 +113,7 @@ func Factories() (otelcol.Factories, error) {
 		metricsgenerationprocessor.NewFactory(),
 		metricstransformprocessor.NewFactory(),
 		probabilisticsamplerprocessor.NewFactory(),
+		prometheusadapter.NewFactory(),
 		resourceprocessor.NewFactory(),
 		resourcedetectionprocessor.NewFactory(),
 		rollupprocessor.NewFactory(),

service/defaultcomponents/components_test.go

@@ -57,6 +57,7 @@ func TestComponents(t *testing.T) {
 		"k8sattributes",
 		"memory_limiter",
 		"metricstransform",
+		"prometheusadapter",
 		"resourcedetection",
 		"resource",
 		"rollup",