Update carbon and wavefront to the new interface (#744)

Signed-off-by: Bogdan Drutu <[email protected]>

cmd/otelcontribcol/components.go

@@ -77,8 +77,8 @@ func components() (component.Factories, error) {
  collectdreceiver.NewFactory(),
  sapmreceiver.NewFactory(),
  signalfxreceiver.NewFactory(),
- &carbonreceiver.Factory{},
- &wavefrontreceiver.Factory{},
+ carbonreceiver.NewFactory(),
+ wavefrontreceiver.NewFactory(),
  redisreceiver.NewFactory(),
  kubeletstatsreceiver.NewFactory(),
  simpleprometheusreceiver.NewFactory(),

receiver/carbonreceiver/config_test.go

@@ -33,7 +33,7 @@ func TestLoadConfig(t *testing.T) {
  factories, err := componenttest.ExampleComponents()
  assert.Nil(t, err)
 
- factory := &Factory{}
+ factory := NewFactory()
  factories.Receivers[configmodels.Type(typeStr)] = factory
  cfg, err := configtest.LoadConfigFile(
  t, path.Join(".", "testdata", "config.yaml"), factories,

receiver/carbonreceiver/factory.go

@@ -20,11 +20,10 @@ import (
 
  "github.com/spf13/viper"
  "go.opentelemetry.io/collector/component"
- "go.opentelemetry.io/collector/config/configerror"
  "go.opentelemetry.io/collector/config/configmodels"
  "go.opentelemetry.io/collector/config/confignet"
  "go.opentelemetry.io/collector/consumer"
- "go.uber.org/zap"
+ "go.opentelemetry.io/collector/receiver/receiverhelper"
 
  "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/carbonreceiver/protocol"
  "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/carbonreceiver/transport"
@@ -37,57 +36,50 @@ const (
  typeStr = "carbon"
 )
 
-// Factory is the factory for carbon receiver.
-type Factory struct {
+// NewFactory creates a factory for Carbon receiver.
+func NewFactory() component.ReceiverFactory {
+ return receiverhelper.NewFactory(
+ typeStr,
+ createDefaultConfig,
+ receiverhelper.WithCustomUnmarshaler(customUnmarshaler),
+ receiverhelper.WithMetrics(createMetricsReceiver))
 }
 
-var _ component.ReceiverFactoryOld = (*Factory)(nil)
-
-// Type gets the type of the Receiver config created by this factory.
-func (f *Factory) Type() configmodels.Type {
- return configmodels.Type(typeStr)
-}
-
-// CustomUnmarshaler returns the custom function to handle the special settings
-// used on the receiver.
-func (f *Factory) CustomUnmarshaler() component.CustomUnmarshaler {
- return func(sourceViperSection *viper.Viper, intoCfg interface{}) error {
- if sourceViperSection == nil {
- // The section is empty nothing to do, using the default config.
- return nil
- }
-
- // Unmarshal but not exact yet so the different keys under config do not
- // trigger errors, this is needed so that the types of protocol and transport
- // are read.
- if err := sourceViperSection.Unmarshal(intoCfg); err != nil {
- return err
- }
+func customUnmarshaler(sourceViperSection *viper.Viper, intoCfg interface{}) error {
+ if sourceViperSection == nil {
+ // The section is empty nothing to do, using the default config.
+ return nil
+ }
 
- // Unmarshal the protocol, so the type of config can be properly set.
- rCfg := intoCfg.(*Config)
- vParserCfg := sourceViperSection.Sub(parserConfigSection)
- if vParserCfg != nil {
- if err := protocol.LoadParserConfig(vParserCfg, rCfg.Parser); err != nil {
- return fmt.Errorf(
- "error on %q section for %s: %v",
- parserConfigSection,
- rCfg.Name(),
- err)
- }
- }
+ // Unmarshal but not exact yet so the different keys under config do not
+ // trigger errors, this is needed so that the types of protocol and transport
+ // are read.
+ if err := sourceViperSection.Unmarshal(intoCfg); err != nil {
+ return err
+ }
 
- // Unmarshal exact to validate the config keys.
- if err := sourceViperSection.UnmarshalExact(intoCfg); err != nil {
- return err
+ // Unmarshal the protocol, so the type of config can be properly set.
+ rCfg := intoCfg.(*Config)
+ vParserCfg := sourceViperSection.Sub(parserConfigSection)
+ if vParserCfg != nil {
+ if err := protocol.LoadParserConfig(vParserCfg, rCfg.Parser); err != nil {
+ return fmt.Errorf(
+ "error on %q section for %s: %v",
+ parserConfigSection,
+ rCfg.Name(),
+ err)
  }
+ }
 
- return nil
+ // Unmarshal exact to validate the config keys.
+ if err := sourceViperSection.UnmarshalExact(intoCfg); err != nil {
+ return err
  }
+
+ return nil
 }
 
-// CreateDefaultConfig creates the default configuration for Carbon receiver.
-func (f *Factory) CreateDefaultConfig() configmodels.Receiver {
+func createDefaultConfig() configmodels.Receiver {
  return &Config{
  ReceiverSettings: configmodels.ReceiverSettings{
  TypeVal: configmodels.Type(typeStr),
@@ -105,25 +97,13 @@ func (f *Factory) CreateDefaultConfig() configmodels.Receiver {
  }
 }
 
-// CreateTraceReceiver creates a trace receiver based on provided config.
-func (f *Factory) CreateTraceReceiver(
- ctx context.Context,
- logger *zap.Logger,
- cfg configmodels.Receiver,
- consumer consumer.TraceConsumerOld,
-) (component.TraceReceiver, error) {
-
- return nil, configerror.ErrDataTypeIsNotSupported
-}
-
-// CreateMetricsReceiver creates a metrics receiver based on provided config.
-func (f *Factory) CreateMetricsReceiver(
- ctx context.Context,
- logger *zap.Logger,
+func createMetricsReceiver(
+ _ context.Context,
+ params component.ReceiverCreateParams,
  cfg configmodels.Receiver,
- consumer consumer.MetricsConsumerOld,
+ consumer consumer.MetricsConsumer,
 ) (component.MetricsReceiver, error) {
 
  rCfg := cfg.(*Config)
- return New(logger, *rCfg, consumer)
+ return New(params.Logger, *rCfg, consumer)
 }

receiver/carbonreceiver/factory_test.go

@@ -19,43 +19,25 @@ import (
  "testing"
 
  "github.com/stretchr/testify/assert"
+ "go.opentelemetry.io/collector/component"
  "go.opentelemetry.io/collector/config/configcheck"
- "go.opentelemetry.io/collector/config/configerror"
- "go.opentelemetry.io/collector/consumer"
- "go.opentelemetry.io/collector/consumer/consumerdata"
+ "go.opentelemetry.io/collector/exporter/exportertest"
  "go.uber.org/zap"
 )
 
 func TestCreateDefaultConfig(t *testing.T) {
- factory := &Factory{}
+ factory := NewFactory()
  cfg := factory.CreateDefaultConfig()
  assert.NotNil(t, cfg, "failed to create default config")
  assert.NoError(t, configcheck.ValidateConfig(cfg))
 }
 
-type mockMetricsConsumer struct {
-}
-
-var _ (consumer.MetricsConsumerOld) = (*mockMetricsConsumer)(nil)
-
-func (m *mockMetricsConsumer) ConsumeMetricsData(ctx context.Context, md consumerdata.MetricsData) error {
- return nil
-}
-
 func TestCreateReceiver(t *testing.T) {
- factory := &Factory{}
- cfg := factory.CreateDefaultConfig().(*Config)
+ cfg := createDefaultConfig().(*Config)
  cfg.Endpoint = "localhost:0" // Endpoint is required, not going to be used here.
 
- tReceiver, err := factory.CreateMetricsReceiver(context.Background(), zap.NewNop(), cfg, &mockMetricsConsumer{})
- assert.Nil(t, err, "receiver creation failed")
- assert.NotNil(t, tReceiver, "receiver creation failed")
-
- tReceiver, err = factory.CreateMetricsReceiver(context.Background(), zap.NewNop(), cfg, &mockMetricsConsumer{})
- assert.Nil(t, err, "receiver creation failed")
+ params := component.ReceiverCreateParams{Logger: zap.NewNop()}
+ tReceiver, err := createMetricsReceiver(context.Background(), params, cfg, exportertest.NewNopMetricsExporter())
+ assert.NoError(t, err)
  assert.NotNil(t, tReceiver, "receiver creation failed")
-
- mReceiver, err := factory.CreateTraceReceiver(context.Background(), zap.NewNop(), cfg, nil)
- assert.Equal(t, err, configerror.ErrDataTypeIsNotSupported)
- assert.Nil(t, mReceiver)
 }

receiver/carbonreceiver/receiver.go

@@ -45,7 +45,7 @@ type carbonReceiver struct {
  server transport.Server
  reporter transport.Reporter
  parser protocol.Parser
- nextConsumer consumer.MetricsConsumerOld
+ nextConsumer consumer.MetricsConsumer
 
  startOnce sync.Once
  stopOnce sync.Once
@@ -57,7 +57,7 @@ var _ component.MetricsReceiver = (*carbonReceiver)(nil)
 func New(
  logger *zap.Logger,
  config Config,
- nextConsumer consumer.MetricsConsumerOld,
+ nextConsumer consumer.MetricsConsumer,
 ) (component.MetricsReceiver, error) {
 
  if nextConsumer == nil {

receiver/carbonreceiver/receiver_test.go

@@ -30,6 +30,7 @@ import (
  "go.opentelemetry.io/collector/config/configmodels"
  "go.opentelemetry.io/collector/config/confignet"
  "go.opentelemetry.io/collector/consumer"
+ "go.opentelemetry.io/collector/consumer/pdatautil"
  "go.opentelemetry.io/collector/exporter/exportertest"
  "go.opentelemetry.io/collector/testutil"
  "go.uber.org/zap"
@@ -40,10 +41,10 @@ import (
 )
 
 func Test_carbonreceiver_New(t *testing.T) {
- defaultConfig := (&Factory{}).CreateDefaultConfig().(*Config)
+ defaultConfig := createDefaultConfig().(*Config)
  type args struct {
  config Config
- nextConsumer consumer.MetricsConsumerOld
+ nextConsumer consumer.MetricsConsumer
  }
  tests := []struct {
  name string
@@ -54,7 +55,7 @@ func Test_carbonreceiver_New(t *testing.T) {
  name: "default_config",
  args: args{
  config: *defaultConfig,
- nextConsumer: new(exportertest.SinkMetricsExporterOld),
+ nextConsumer: exportertest.NewNopMetricsExporter(),
  },
  },
  {
@@ -68,7 +69,7 @@ func Test_carbonreceiver_New(t *testing.T) {
  },
  TCPIdleTimeout: defaultConfig.TCPIdleTimeout,
  },
- nextConsumer: new(exportertest.SinkMetricsExporterOld),
+ nextConsumer: exportertest.NewNopMetricsExporter(),
  },
  },
  {
@@ -84,7 +85,7 @@ func Test_carbonreceiver_New(t *testing.T) {
  config: Config{
  ReceiverSettings: configmodels.ReceiverSettings{},
  },
- nextConsumer: new(exportertest.SinkMetricsExporterOld),
+ nextConsumer: exportertest.NewNopMetricsExporter(),
  },
  wantErr: errEmptyEndpoint,
  },
@@ -104,7 +105,7 @@ func Test_carbonreceiver_New(t *testing.T) {
  Config: &protocol.PlaintextConfig{},
  },
  },
- nextConsumer: new(exportertest.SinkMetricsExporterOld),
+ nextConsumer: exportertest.NewNopMetricsExporter(),
  },
  wantErr: errors.New("unsupported transport \"unknown_transp\" for receiver \"invalid_transport_rcv\""),
  },
@@ -130,7 +131,7 @@ func Test_carbonreceiver_New(t *testing.T) {
  },
  },
  },
- nextConsumer: new(exportertest.SinkMetricsExporterOld),
+ nextConsumer: exportertest.NewNopMetricsExporter(),
  },
  },
  {
@@ -150,7 +151,7 @@ func Test_carbonreceiver_New(t *testing.T) {
  Config: &protocol.PlaintextConfig{},
  },
  },
- nextConsumer: new(exportertest.SinkMetricsExporterOld),
+ nextConsumer: exportertest.NewNopMetricsExporter(),
  },
  wantErr: errors.New("invalid idle timeout: -1s"),
  },
@@ -184,7 +185,7 @@ func Test_carbonreceiver_EndToEnd(t *testing.T) {
  {
  name: "default_config",
  configFn: func() *Config {
- return (&Factory{}).CreateDefaultConfig().(*Config)
+ return createDefaultConfig().(*Config)
  },
  clientFn: func(t *testing.T) *client.Graphite {
  c, err := client.NewGraphite(client.TCP, host, port)
@@ -195,7 +196,7 @@ func Test_carbonreceiver_EndToEnd(t *testing.T) {
  {
  name: "default_config_udp",
  configFn: func() *Config {
- cfg := (&Factory{}).CreateDefaultConfig().(*Config)
+ cfg := createDefaultConfig().(*Config)
  cfg.Transport = "udp"
  return cfg
  },
@@ -210,7 +211,7 @@ func Test_carbonreceiver_EndToEnd(t *testing.T) {
  t.Run(tt.name, func(t *testing.T) {
  cfg := tt.configFn()
  cfg.Endpoint = addr
- sink := new(exportertest.SinkMetricsExporterOld)
+ sink := new(exportertest.SinkMetricsExporter)
  rcv, err := New(zap.NewNop(), *cfg, sink)
  require.NoError(t, err)
  r := rcv.(*carbonReceiver)
@@ -237,9 +238,11 @@ func Test_carbonreceiver_EndToEnd(t *testing.T) {
  mr.WaitAllOnMetricsProcessedCalls()
 
  mdd := sink.AllMetrics()
- require.Equal(t, 1, len(mdd))
- require.Equal(t, 1, len(mdd[0].Metrics))
- metric := mdd[0].Metrics[0]
+ require.Len(t, mdd, 1)
+ ocmd := pdatautil.MetricsToMetricsData(mdd[0])
+ require.Len(t, ocmd, 1)
+ require.Len(t, ocmd[0].Metrics, 1)
+ metric := ocmd[0].Metrics[0]
  assert.Equal(t, carbonMetric.Name, metric.GetMetricDescriptor().GetName())
  tss := metric.GetTimeseries()
  require.Equal(t, 1, len(tss))

receiver/carbonreceiver/transport/server.go

@@ -36,7 +36,7 @@ type Server interface {
  // the Parser and passed to the next consumer.
  ListenAndServe(
  p protocol.Parser,
- mc consumer.MetricsConsumerOld,
+ mc consumer.MetricsConsumer,
  r Reporter,
  ) error