Makes changes to receiver pattern

go.sum

@@ -1383,6 +1383,8 @@ github.com/prometheus/common v0.18.0/go.mod h1:U+gB1OBLb1lF3O42bTCL+FK18tX9Oar16
 github.com/prometheus/common v0.23.0/go.mod h1:H6QK/N6XVT42whUeIdI3dp36w49c+/iMDk7UAI2qm7Q=
 github.com/prometheus/common v0.24.0 h1:aIycr3wRFxPUq8XlLQlGQ9aNXV3dFi5y62pe/SB262k=
 github.com/prometheus/common v0.24.0/go.mod h1:H6QK/N6XVT42whUeIdI3dp36w49c+/iMDk7UAI2qm7Q=
+github.com/prometheus/common v0.25.0 h1:IjJYZJCI8HZYtqA3xYwGyDzSCy1r4CA2GRh+4vdOmtE=
+github.com/prometheus/common v0.25.0/go.mod h1:H6QK/N6XVT42whUeIdI3dp36w49c+/iMDk7UAI2qm7Q=
 github.com/prometheus/exporter-toolkit v0.5.1/go.mod h1:OCkM4805mmisBhLmVFw858QYi3v0wKdY6/UxrT0pZVg=
 github.com/prometheus/procfs v0.0.0-20180125133057-cb4147076ac7/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
 github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=

receiver/k8sclusterreceiver/receiver.go

@@ -42,6 +42,7 @@ type kubernetesReceiver struct {
  logger *zap.Logger
  consumer consumer.Metrics
  cancel context.CancelFunc
+ obsrecv *obsreport.Receiver
 }
 
 func (kr *kubernetesReceiver) Start(ctx context.Context, host component.Host) error {
@@ -99,12 +100,12 @@ func (kr *kubernetesReceiver) dispatchMetrics(ctx context.Context) {
  now := time.Now()
  mds := kr.resourceWatcher.dataCollector.CollectMetricData(now)
 
- c := obsreport.StartMetricsReceiveOp(ctx, kr.config.ID(), transport)
+ c := kr.obsrecv.StartMetricsReceiveOp(ctx)
 
  _, numPoints := mds.MetricAndDataPointCount()
 
  err := kr.consumer.ConsumeMetrics(c, mds)
- obsreport.EndMetricsReceiveOp(c, typeStr, numPoints, err)
+ kr.obsrecv.EndMetricsReceiveOp(c, typeStr, numPoints, err)
 }
 
 // newReceiver creates the Kubernetes cluster receiver with the given configuration.
@@ -118,5 +119,6 @@ func newReceiver(
  logger: logger,
  config: config,
  consumer: consumer,
+ obsrecv: obsreport.NewReceiver(obsreport.ReceiverSettings{ReceiverID: config.ID(), Transport: transport}),
  }, nil
 }

receiver/kubeletstatsreceiver/runnable.go

@@ -35,6 +35,8 @@ import (
 
 var _ interval.Runnable = (*runnable)(nil)
 
+const transport = "http"
+
 type runnable struct {
  ctx context.Context
  receiverID config.ComponentID
@@ -47,6 +49,7 @@ type runnable struct {
  metricGroupsToCollect map[kubelet.MetricGroup]bool
  k8sAPIClient kubernetes.Interface
  cachedVolumeLabels map[string]map[string]string
+ obsrecv *obsreport.Receiver
 }
 
 func newRunnable(
@@ -66,6 +69,7 @@ func newRunnable(
  metricGroupsToCollect: rOptions.metricGroupsToCollect,
  k8sAPIClient: rOptions.k8sAPIClient,
  cachedVolumeLabels: make(map[string]map[string]string),
+ obsrecv: obsreport.NewReceiver(obsreport.ReceiverSettings{ReceiverID: rOptions.id, Transport: transport}),
  }
 }
 
@@ -77,7 +81,6 @@ func (r *runnable) Setup() error {
 }
 
 func (r *runnable) Run() error {
- const transport = "http"
  summary, err := r.statsProvider.StatsSummary()
  if err != nil {
  r.logger.Error("call to /stats/summary endpoint failed", zap.Error(err))
@@ -103,14 +106,14 @@ func (r *runnable) Run() error {
 
  var numPoints int
  ctx := obsreport.ReceiverContext(r.ctx, r.receiverID, transport)
- ctx = obsreport.StartMetricsReceiveOp(ctx, r.receiverID, transport)
+ ctx = r.obsrecv.StartMetricsReceiveOp(ctx)
  err = r.consumer.ConsumeMetrics(ctx, metrics)
  if err != nil {
  r.logger.Error("ConsumeMetricsData failed", zap.Error(err))
  } else {
  _, numPoints = metrics.MetricAndDataPointCount()
  }
- obsreport.EndMetricsReceiveOp(ctx, typeStr, numPoints, err)
+ r.obsrecv.EndMetricsReceiveOp(ctx, typeStr, numPoints, err)
 
  return nil
 }

receiver/sapmreceiver/trace_receiver.go

@@ -56,6 +56,8 @@ type sapmReceiver struct {
  // This defaultResponse is an optimization so we don't have to proto.Marshal the response
  // for every request. At some point this may be removed when there is actual content to return.
  defaultResponse []byte
+
+ obsrecv *obsreport.Receiver
 }
 
 // handleRequest parses an http request containing sapm and passes the trace data to the next consumer
@@ -71,7 +73,7 @@ func (sr *sapmReceiver) handleRequest(ctx context.Context, req *http.Request) er
  transport = "https"
  }
  ctx = obsreport.ReceiverContext(ctx, sr.config.ID(), transport)
- ctx = obsreport.StartTraceDataReceiveOp(ctx, sr.config.ID(), transport)
+ ctx = sr.obsrecv.StartTraceDataReceiveOp(ctx)
 
  td := jaegertranslator.ProtoBatchesToInternalTraces(sapm.Batches)
 
@@ -92,7 +94,7 @@ func (sr *sapmReceiver) handleRequest(ctx context.Context, req *http.Request) er
  err = fmt.Errorf("error passing trace data to next consumer: %v", err.Error())
  }
 
- obsreport.EndTraceDataReceiveOp(ctx, "protobuf", td.SpanCount(), err)
+ sr.obsrecv.EndTraceDataReceiveOp(ctx, "protobuf", td.SpanCount(), err)
  return err
 }
 
@@ -208,10 +210,15 @@ func New(
  if err != nil {
  return nil, fmt.Errorf("failed to marshal default response body for %v receiver: %w", config.ID(), err)
  }
+ transport := "http"
+ if config.TLSSetting != nil {
+ transport = "https"
+ }
  return &sapmReceiver{
  logger: params.Logger,
  config: config,
  nextConsumer: nextConsumer,
  defaultResponse: defaultResponseBytes,
+ obsrecv: obsreport.NewReceiver(obsreport.ReceiverSettings{ReceiverID: config.ID(), Transport: transport}),
  }, nil
 }