Change obsreport receiver calling pattern in signalfxreceiver, splunk…

…hecreceiver, statsdreceiver, and redisreceiver (open-telemetry#3521)

* Make changes to awsxrayreceiver

* Make changes to carbonreceiver

* Add entry to changelog

* Make changes to dockerstatsreceiver

* Add entry to changelog

* Makes changes to receiver pattern

* Fix test issue

* Add entry intro changelog

* Change receiver pattern for rest of receivers

* Add entry in changelog

CHANGELOG.md

@@ -34,7 +34,7 @@ The OpenTelemetry Collector Contrib contains everything in the [opentelemetry-co
 - `groupbytrace` processor: Added workers for queue processing (#2902)
 - `resourcedetection` processor: Add docker detector (#2775)
 - `tailsampling` processor: Support regex on span attribute filtering (#3335_
-- Change obsreport helpers for receiver to use the new pattern created in Collector (#3439,#3443,#3449,#3504)
+- Change obsreport helpers for receiver to use the new pattern created in Collector (#3439,#3443,#3449,#3504,#3521)
 
 ## 🧰 Bug fixes 🧰
 

receiver/redisreceiver/redis_runnable.go

@@ -29,6 +29,8 @@ import (
 
 var _ interval.Runnable = (*redisRunnable)(nil)
 
+const transport = "http" // todo verify this
+
 // Runs intermittently, fetching info from Redis, creating metrics/datapoints,
 // and feeding them to a metricsConsumer.
 type redisRunnable struct {
@@ -40,6 +42,7 @@ type redisRunnable struct {
  logger *zap.Logger
  timeBundle *timeBundle
  serviceName string
+ obsrecv *obsreport.Receiver
 }
 
 func newRedisRunnable(
@@ -57,6 +60,7 @@ func newRedisRunnable(
  redisSvc: newRedisSvc(client),
  metricsConsumer: metricsConsumer,
  logger: logger,
+ obsrecv: obsreport.NewReceiver(obsreport.ReceiverSettings{ReceiverID: id, Transport: transport}),
  }
 }
 
@@ -74,18 +78,17 @@ func (r *redisRunnable) Setup() error {
 // active Redis database, of which there can be 16.
 func (r *redisRunnable) Run() error {
  const dataFormat = "redis"
- const transport = "http" // todo verify this
- ctx := obsreport.StartMetricsReceiveOp(r.ctx, r.id, transport)
+ ctx := r.obsrecv.StartMetricsReceiveOp(r.ctx)
 
  inf, err := r.redisSvc.info()
  if err != nil {
- obsreport.EndMetricsReceiveOp(ctx, dataFormat, 0, err)
+ r.obsrecv.EndMetricsReceiveOp(ctx, dataFormat, 0, err)
  return nil
  }
 
  uptime, err := inf.getUptimeInSeconds()
  if err != nil {
- obsreport.EndMetricsReceiveOp(ctx, dataFormat, 0, err)
+ r.obsrecv.EndMetricsReceiveOp(ctx, dataFormat, 0, err)
  return nil
  }
 
@@ -121,7 +124,7 @@ func (r *redisRunnable) Run() error {
 
  err = r.metricsConsumer.ConsumeMetrics(r.ctx, pdm)
  _, numPoints := pdm.MetricAndDataPointCount()
- obsreport.EndMetricsReceiveOp(ctx, dataFormat, numPoints, err)
+ r.obsrecv.EndMetricsReceiveOp(ctx, dataFormat, numPoints, err)
 
  return nil
 }

receiver/signalfxreceiver/receiver.go

@@ -84,6 +84,7 @@ type sfxReceiver struct {
  metricsConsumer consumer.Metrics
  logsConsumer consumer.Logs
  server *http.Server
+ obsrecv *obsreport.Receiver
 }
 
 var _ component.MetricsReceiver = (*sfxReceiver)(nil)
@@ -93,9 +94,14 @@ func newReceiver(
  logger *zap.Logger,
  config Config,
 ) *sfxReceiver {
+ transport := "http"
+ if config.TLSSetting != nil {
+ transport = "https"
+ }
  r := &sfxReceiver{
- logger: logger,
- config: &config,
+ logger: logger,
+ config: &config,
+ obsrecv: obsreport.NewReceiver(obsreport.ReceiverSettings{ReceiverID: config.ID(), Transport: transport}),
  }
 
  return r
@@ -212,7 +218,7 @@ func (r *sfxReceiver) handleDatapointReq(resp http.ResponseWriter, req *http.Req
  }
 
  ctx := obsreport.ReceiverContext(req.Context(), r.config.ID(), transport)
- ctx = obsreport.StartMetricsReceiveOp(ctx, r.config.ID(), transport)
+ ctx = r.obsrecv.StartMetricsReceiveOp(ctx)
 
  if r.metricsConsumer == nil {
  r.failRequest(ctx, resp, http.StatusBadRequest, errMetricsNotConfigured, nil)
@@ -231,7 +237,7 @@ func (r *sfxReceiver) handleDatapointReq(resp http.ResponseWriter, req *http.Req
  }
 
  if len(msg.Datapoints) == 0 {
- obsreport.EndMetricsReceiveOp(ctx, typeStr, 0, nil)
+ r.obsrecv.EndMetricsReceiveOp(ctx, typeStr, 0, nil)
  resp.Write(okRespBody)
  return
  }
@@ -249,7 +255,7 @@ func (r *sfxReceiver) handleDatapointReq(resp http.ResponseWriter, req *http.Req
  }
 
  err := r.metricsConsumer.ConsumeMetrics(ctx, md)
- obsreport.EndMetricsReceiveOp(
+ r.obsrecv.EndMetricsReceiveOp(
  ctx,
  typeStr,
  len(msg.Datapoints),
@@ -265,7 +271,7 @@ func (r *sfxReceiver) handleEventReq(resp http.ResponseWriter, req *http.Request
  }
 
  ctx := obsreport.ReceiverContext(req.Context(), r.config.ID(), transport)
- ctx = obsreport.StartMetricsReceiveOp(ctx, r.config.ID(), transport)
+ ctx = r.obsrecv.StartMetricsReceiveOp(ctx)
 
  if r.logsConsumer == nil {
  r.failRequest(ctx, resp, http.StatusBadRequest, errLogsNotConfigured, nil)
@@ -284,7 +290,7 @@ func (r *sfxReceiver) handleEventReq(resp http.ResponseWriter, req *http.Request
  }
 
  if len(msg.Events) == 0 {
- obsreport.EndMetricsReceiveOp(ctx, typeStr, 0, nil)
+ r.obsrecv.EndMetricsReceiveOp(ctx, typeStr, 0, nil)
  resp.Write(okRespBody)
  return
  }
@@ -301,7 +307,7 @@ func (r *sfxReceiver) handleEventReq(resp http.ResponseWriter, req *http.Request
  }
 
  err := r.logsConsumer.ConsumeLogs(ctx, ld)
- obsreport.EndMetricsReceiveOp(
+ r.obsrecv.EndMetricsReceiveOp(
  ctx,
  typeStr,
  len(msg.Events),

receiver/splunkhecreceiver/receiver.go

@@ -79,6 +79,7 @@ type splunkReceiver struct {
  logsConsumer consumer.Logs
  metricsConsumer consumer.Metrics
  server *http.Server
+ obsrecv *obsreport.Receiver
 }
 
 var _ component.MetricsReceiver = (*splunkReceiver)(nil)
@@ -97,6 +98,11 @@ func NewMetricsReceiver(
  return nil, errEmptyEndpoint
  }
 
+ transport := "http"
+ if config.TLSSetting != nil {
+ transport = "https"
+ }
+
  r := &splunkReceiver{
  logger: logger,
  config: &config,
@@ -108,6 +114,7 @@ func NewMetricsReceiver(
  ReadHeaderTimeout: defaultServerTimeout,
  WriteTimeout: defaultServerTimeout,
  },
+ obsrecv: obsreport.NewReceiver(obsreport.ReceiverSettings{ReceiverID: config.ID(), Transport: transport}),
  }
 
  return r, nil
@@ -196,7 +203,7 @@ func (r *splunkReceiver) handleReq(resp http.ResponseWriter, req *http.Request)
 
  ctx := obsreport.ReceiverContext(req.Context(), r.config.ID(), transport)
  if r.logsConsumer == nil {
- ctx = obsreport.StartMetricsReceiveOp(ctx, r.config.ID(), transport)
+ ctx = r.obsrecv.StartMetricsReceiveOp(ctx)
  }
  reqPath := req.URL.Path
  if !r.config.pathGlob.Match(reqPath) {
@@ -275,7 +282,7 @@ func (r *splunkReceiver) consumeMetrics(ctx context.Context, events []*splunk.Ev
  md, _ := SplunkHecToMetricsData(r.logger, events, r.createResourceCustomizer(req))
 
  decodeErr := r.metricsConsumer.ConsumeMetrics(ctx, md)
- obsreport.EndMetricsReceiveOp(ctx, typeStr, len(events), decodeErr)
+ r.obsrecv.EndMetricsReceiveOp(ctx, typeStr, len(events), decodeErr)
 
  if decodeErr != nil {
  r.failRequest(ctx, resp, http.StatusInternalServerError, errInternalServerError, decodeErr)

receiver/statsdreceiver/reporter.go

@@ -31,6 +31,7 @@ type reporter struct {
  id config.ComponentID
  logger *zap.Logger
  sugaredLogger *zap.SugaredLogger // Used for generic debug logging
+ obsrecv *obsreport.Receiver
 }
 
 var _ transport.Reporter = (*reporter)(nil)
@@ -40,6 +41,7 @@ func newReporter(receiverID config.ComponentID, logger *zap.Logger) transport.Re
  id: receiverID,
  logger: logger,
  sugaredLogger: logger.Sugar(),
+ obsrecv: obsreport.NewReceiver(obsreport.ReceiverSettings{ReceiverID: receiverID, Transport: "tcp"}),
  }
 }
 
@@ -49,7 +51,7 @@ func newReporter(receiverID config.ComponentID, logger *zap.Logger) transport.Re
 // returned span.
 func (r *reporter) OnDataReceived(ctx context.Context) context.Context {
  ctx = obsreport.ReceiverContext(ctx, r.id, "tcp")
- return obsreport.StartMetricsReceiveOp(ctx, r.id, "tcp")
+ return r.obsrecv.StartMetricsReceiveOp(ctx)
 }
 
 // OnTranslationError is used to report a translation error from original
@@ -93,7 +95,7 @@ func (r *reporter) OnMetricsProcessed(
  })
  }
 
- obsreport.EndMetricsReceiveOp(ctx, "statsd", numReceivedMessages, err)
+ r.obsrecv.EndMetricsReceiveOp(ctx, "statsd", numReceivedMessages, err)
 }
 
 func (r *reporter) OnDebugf(template string, args ...interface{}) {