[chore] fix resourcetotelemetry usage in carbonexporter (open-telemet…

…ry#29888)

In
open-telemetry#29879
I forgot to actually plugin the logic, only the config was added there.
Added a test to confirm it.

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

exporter/carbonexporter/exporter.go

@@ -12,6 +12,8 @@ import (
  "go.opentelemetry.io/collector/exporter"
  "go.opentelemetry.io/collector/exporter/exporterhelper"
  "go.opentelemetry.io/collector/pdata/pmetric"
+
+ "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/resourcetotelemetry"
 )
 
 // newCarbonExporter returns a new Carbon exporter.
@@ -20,7 +22,7 @@ func newCarbonExporter(cfg *Config, set exporter.CreateSettings) (exporter.Metri
  connPool: newTCPConnPool(cfg.Endpoint, cfg.Timeout),
  }
 
- return exporterhelper.NewMetricsExporter(
+ exp, err := exporterhelper.NewMetricsExporter(
  context.TODO(),
  set,
  cfg,
@@ -29,6 +31,11 @@ func newCarbonExporter(cfg *Config, set exporter.CreateSettings) (exporter.Metri
  exporterhelper.WithQueue(cfg.QueueConfig),
  exporterhelper.WithRetry(cfg.RetryConfig),
  exporterhelper.WithShutdown(sender.Shutdown))
+ if err != nil {
+ return nil, err
+ }
+
+ return resourcetotelemetry.WrapMetricsExporter(cfg.ResourceToTelemetryConfig, exp), nil
 }
 
 // carbonSender is the struct tying the translation function and the TCP

exporter/carbonexporter/exporter_test.go

@@ -27,6 +27,7 @@ import (
  conventions "go.opentelemetry.io/collector/semconv/v1.9.0"
 
  "github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/testutil"
+ "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/resourcetotelemetry"
 )
 
 func TestNewWithDefaultConfig(t *testing.T) {
@@ -45,10 +46,31 @@ func TestConsumeMetricsNoServer(t *testing.T) {
  exportertest.NewNopCreateSettings())
  require.NoError(t, err)
  require.NoError(t, exp.Start(context.Background(), componenttest.NewNopHost()))
- require.Error(t, exp.ConsumeMetrics(context.Background(), generateLargeBatch()))
+ require.Error(t, exp.ConsumeMetrics(context.Background(), generateSmallBatch()))
  require.NoError(t, exp.Shutdown(context.Background()))
 }
 
+func TestConsumeMetricsWithResourceToTelemetry(t *testing.T) {
+ addr := testutil.GetAvailableLocalAddress(t)
+ cs := newCarbonServer(t, addr, "test_0;k0=v0;k1=v1;service.name=test_carbon 0")
+ // Each metric point will generate one Carbon line, set up the wait
+ // for all of them.
+ cs.start(t, 1)
+
+ exp, err := newCarbonExporter(
+ &Config{
+ TCPAddr: confignet.TCPAddr{Endpoint: addr},
+ TimeoutSettings: exporterhelper.TimeoutSettings{Timeout: 5 * time.Second},
+ ResourceToTelemetryConfig: resourcetotelemetry.Settings{Enabled: true},
+ },
+ exportertest.NewNopCreateSettings())
+ require.NoError(t, err)
+ require.NoError(t, exp.Start(context.Background(), componenttest.NewNopHost()))
+ require.NoError(t, exp.ConsumeMetrics(context.Background(), generateSmallBatch()))
+ assert.NoError(t, exp.Shutdown(context.Background()))
+ cs.shutdownAndVerify(t)
+}
+
 func TestConsumeMetrics(t *testing.T) {
  if runtime.GOOS == "windows" {
  t.Skip("skipping test on windows, see https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/10147")
@@ -94,7 +116,7 @@ func TestConsumeMetrics(t *testing.T) {
  for _, tt := range tests {
  t.Run(tt.name, func(t *testing.T) {
  addr := testutil.GetAvailableLocalAddress(t)
- cs := newCarbonServer(t, addr)
+ cs := newCarbonServer(t, addr, "")
  // Each metric point will generate one Carbon line, set up the wait
  // for all of them.
  cs.start(t, tt.numProducers*tt.writesPerProducer*tt.md.DataPointCount())
@@ -133,25 +155,21 @@ func TestConsumeMetrics(t *testing.T) {
 }
 
 func generateSmallBatch() pmetric.Metrics {
- metrics := pmetric.NewMetrics()
- m := metrics.ResourceMetrics().AppendEmpty().ScopeMetrics().AppendEmpty().Metrics().AppendEmpty()
- m.SetName("test_gauge")
- dp := m.SetEmptyGauge().DataPoints().AppendEmpty()
- dp.Attributes().PutStr("k0", "v0")
- dp.Attributes().PutStr("k1", "v1")
- dp.SetTimestamp(pcommon.NewTimestampFromTime(time.Now()))
- dp.SetDoubleValue(123)
- return metrics
+ return generateMetricsBatch(1)
 }
 
 func generateLargeBatch() pmetric.Metrics {
+ return generateMetricsBatch(1024)
+}
+
+func generateMetricsBatch(size int) pmetric.Metrics {
  ts := time.Now()
  metrics := pmetric.NewMetrics()
  rm := metrics.ResourceMetrics().AppendEmpty()
  rm.Resource().Attributes().PutStr(conventions.AttributeServiceName, "test_carbon")
  ms := rm.ScopeMetrics().AppendEmpty().Metrics()
 
- for i := 0; i < 1028; i++ {
+ for i := 0; i < size; i++ {
  m := ms.AppendEmpty()
  m.SetName("test_" + strconv.Itoa(i))
  dp := m.SetEmptyGauge().DataPoints().AppendEmpty()
@@ -165,19 +183,21 @@ func generateLargeBatch() pmetric.Metrics {
 }
 
 type carbonServer struct {
- ln *net.TCPListener
- doneServer *atomic.Bool
- wg sync.WaitGroup
+ ln *net.TCPListener
+ doneServer *atomic.Bool
+ wg sync.WaitGroup
+ expectedContainsValue string
 }
 
-func newCarbonServer(t *testing.T, addr string) *carbonServer {
+func newCarbonServer(t *testing.T, addr string, expectedContainsValue string) *carbonServer {
  laddr, err := net.ResolveTCPAddr("tcp", addr)
  require.NoError(t, err)
  ln, err := net.ListenTCP("tcp", laddr)
  require.NoError(t, err)
  return &carbonServer{
- ln: ln,
- doneServer: &atomic.Bool{},
+ ln: ln,
+ doneServer: &atomic.Bool{},
+ expectedContainsValue: expectedContainsValue,
  }
 }
 
@@ -198,14 +218,16 @@ func (cs *carbonServer) start(t *testing.T, numExpectedReq int) {
 
  reader := bufio.NewReader(conn)
  for {
- // Actual metric validation is done by other tests, here it
- // is just flow.
- _, err := reader.ReadBytes(byte('\n'))
+ buf, err := reader.ReadBytes(byte('\n'))
  if errors.Is(err, io.EOF) {
  return
  }
  require.NoError(t, err)
 
+ if cs.expectedContainsValue != "" {
+ assert.Contains(t, string(buf), cs.expectedContainsValue)
+ }
+
  cs.wg.Done()
  }
  }(conn)