[receiver/solacereceiver]: Updated the format for generated metrics and included new Solace receiver metric attribute (#34541)

**Description:** Updated the format for generated metrics. Included a
`receiver_name` attribute that identifies the Solace receiver that
generated the metrics.

**Link to tracking Issue:** N/A

**Testing:** Updated unit tests to accommodate passing of attributes for
metrics

**Documentation:** N/A

Author: oodigie

.chloggen/fix_metrics_prefix.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: solacereceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Updated the format for generated metrics. Included a `receiver_name` attribute that identifies the Solace receiver that generated the metrics"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [34541]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

receiver/solacereceiver/receiver.go

@@ -15,6 +15,8 @@ import (
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/receiver"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/solacereceiver/internal/metadata"
@@ -38,6 +40,10 @@ const (
 	flowControlStateControlled
 )
 
+const (
+	brokerComponenteNameAttr = "receiver_name"
+)
+
 // solaceTracesReceiver uses azure AMQP to consume and handle telemetry data from SOlace. Implements receiver.Traces
 type solaceTracesReceiver struct {
 	// config is the receiver.Config instance used to build the receiver
@@ -56,6 +62,8 @@ type solaceTracesReceiver struct {
 	terminating *atomic.Bool
 	// retryTimeout is the timeout between connection attempts
 	retryTimeout time.Duration
+	// Other Attributes including the ID of the receiver Solace broker's component name
+	metricAttrs attribute.Set
 }
 
 // newTracesReceiver creates a new solaceTraceReceiver as a receiver.Traces
@@ -73,7 +81,18 @@ func newTracesReceiver(config *Config, set receiver.Settings, nextConsumer consu
 		return nil, err
 	}
 
-	unmarshaller := newTracesUnmarshaller(set.Logger, telemetryBuilder)
+	// solaceBrokerAttrs - including the component name of the connected Solace broker
+	receiverName := set.ID.Name()
+	if receiverName != "" {
+		receiverName = "solace/" + receiverName
+	} else {
+		receiverName = "solace"
+	}
+	solaceBrokerAttrs := attribute.NewSet(
+		attribute.String(brokerComponenteNameAttr, receiverName),
+	)
+
+	unmarshaller := newTracesUnmarshaller(set.Logger, telemetryBuilder, solaceBrokerAttrs)
 
 	return &solaceTracesReceiver{
 		config:            config,
@@ -85,13 +104,15 @@ func newTracesReceiver(config *Config, set receiver.Settings, nextConsumer consu
 		factory:           factory,
 		retryTimeout:      1 * time.Second,
 		terminating:       &atomic.Bool{},
+		metricAttrs:       solaceBrokerAttrs,
 	}, nil
 }
 
 // Start implements component.Receiver::Start
 func (s *solaceTracesReceiver) Start(ctx context.Context, _ component.Host) error {
-	s.telemetryBuilder.SolacereceiverReceiverStatus.Record(ctx, int64(receiverStateStarting))
-	s.telemetryBuilder.SolacereceiverReceiverFlowControlStatus.Record(ctx, int64(flowControlStateClear))
+	// set the component name for the connected Solace broker
+	s.telemetryBuilder.SolacereceiverReceiverStatus.Record(ctx, int64(receiverStateStarting), metric.WithAttributeSet(s.metricAttrs))
+	s.telemetryBuilder.SolacereceiverReceiverFlowControlStatus.Record(ctx, int64(flowControlStateClear), metric.WithAttributeSet(s.metricAttrs))
 	var cancelableContext context.Context
 	cancelableContext, s.cancel = context.WithCancel(context.Background())
 
@@ -108,13 +129,14 @@ func (s *solaceTracesReceiver) Shutdown(_ context.Context) error {
 	if s.cancel == nil {
 		return nil
 	}
+	// set the component name for the connected Solace broker
 	s.terminating.Store(true)
-	s.telemetryBuilder.SolacereceiverReceiverStatus.Record(context.Background(), int64(receiverStateTerminating))
+	s.telemetryBuilder.SolacereceiverReceiverStatus.Record(context.Background(), int64(receiverStateTerminating), metric.WithAttributeSet(s.metricAttrs))
 	s.settings.Logger.Info("Shutdown waiting for all components to complete")
 	s.cancel() // cancels the context passed to the reconnection loop
 	s.shutdownWaitGroup.Wait()
 	s.settings.Logger.Info("Receiver shutdown successfully")
-	s.telemetryBuilder.SolacereceiverReceiverStatus.Record(context.Background(), int64(receiverStateTerminated))
+	s.telemetryBuilder.SolacereceiverReceiverStatus.Record(context.Background(), int64(receiverStateTerminated), metric.WithAttributeSet(s.metricAttrs))
 	return nil
 }
 
@@ -130,7 +152,7 @@ func (s *solaceTracesReceiver) connectAndReceive(ctx context.Context) {
 	disable := false
 
 	// indicate we are in connecting state at the start
-	s.telemetryBuilder.SolacereceiverReceiverStatus.Record(context.Background(), int64(receiverStateConnecting))
+	s.telemetryBuilder.SolacereceiverReceiverStatus.Record(context.Background(), int64(receiverStateConnecting), metric.WithAttributeSet(s.metricAttrs))
 
 reconnectionLoop:
 	for !disable {
@@ -156,7 +178,7 @@ reconnectionLoop:
 
 			if err := service.dial(ctx); err != nil {
 				s.settings.Logger.Debug("Encountered error while connecting messaging service", zap.Error(err))
-				s.telemetryBuilder.SolacereceiverFailedReconnections.Add(ctx, 1)
+				s.telemetryBuilder.SolacereceiverFailedReconnections.Add(ctx, 1, metric.WithAttributeSet(s.metricAttrs))
 				return
 			}
 			// dial was successful, record the connected state
@@ -165,7 +187,7 @@ reconnectionLoop:
 			if err := s.receiveMessages(ctx, service); err != nil {
 				s.settings.Logger.Debug("Encountered error while receiving messages", zap.Error(err))
 				if errors.Is(err, errUpgradeRequired) {
-					s.telemetryBuilder.SolacereceiverNeedUpgrade.Record(ctx, 1)
+					s.telemetryBuilder.SolacereceiverNeedUpgrade.Record(ctx, 1, metric.WithAttributeSet(s.metricAttrs))
 					disable = true
 					return
 				}
@@ -182,7 +204,7 @@ reconnectionLoop:
 // this state transition were to happen, it would be short lived.
 func (s *solaceTracesReceiver) recordConnectionState(state receiverState) {
 	if !s.terminating.Load() {
-		s.telemetryBuilder.SolacereceiverReceiverStatus.Record(context.Background(), int64(state))
+		s.telemetryBuilder.SolacereceiverReceiverStatus.Record(context.Background(), int64(state), metric.WithAttributeSet(s.metricAttrs))
 	}
 }
 
@@ -220,18 +242,18 @@ func (s *solaceTracesReceiver) receiveMessage(ctx context.Context, service messa
 		}
 	}()
 	// message received successfully
-	s.telemetryBuilder.SolacereceiverReceivedSpanMessages.Add(ctx, 1)
+	s.telemetryBuilder.SolacereceiverReceivedSpanMessages.Add(ctx, 1, metric.WithAttributeSet(s.metricAttrs))
 	// unmarshal the message. unmarshalling errors are not fatal unless the version is unknown
 	traces, unmarshalErr := s.unmarshaller.unmarshal(msg)
 	if unmarshalErr != nil {
 		s.settings.Logger.Error("Encountered error while unmarshalling message", zap.Error(unmarshalErr))
-		s.telemetryBuilder.SolacereceiverFatalUnmarshallingErrors.Add(ctx, 1)
+		s.telemetryBuilder.SolacereceiverFatalUnmarshallingErrors.Add(ctx, 1, metric.WithAttributeSet(s.metricAttrs))
 		if errors.Is(unmarshalErr, errUpgradeRequired) {
 			disposition = service.failed // if we don't know the version, reject the trace message since we will disable the receiver
 			return unmarshalErr
 		}
-		s.telemetryBuilder.SolacereceiverDroppedSpanMessages.Add(ctx, 1) // if the error is some other unmarshalling error, we will ack the message and drop the content
-		return nil                                                       // don't propagate error, but don't continue forwarding traces
+		s.telemetryBuilder.SolacereceiverDroppedSpanMessages.Add(ctx, 1, metric.WithAttributeSet(s.metricAttrs)) // if the error is some other unmarshalling error, we will ack the message and drop the content
+		return nil                                                                                               // don't propagate error, but don't continue forwarding traces
 	}
 
 	var flowControlCount int64
@@ -245,10 +267,10 @@ flowControlLoop:
 				s.settings.Logger.Info("Encountered temporary error while forwarding traces to next receiver, will allow redelivery", zap.Error(forwardErr))
 				// handle flow control metrics
 				if flowControlCount == 0 {
-					s.telemetryBuilder.SolacereceiverReceiverFlowControlStatus.Record(ctx, int64(flowControlStateControlled))
+					s.telemetryBuilder.SolacereceiverReceiverFlowControlStatus.Record(ctx, int64(flowControlStateControlled), metric.WithAttributeSet(s.metricAttrs))
 				}
 				flowControlCount++
-				s.telemetryBuilder.SolacereceiverReceiverFlowControlRecentRetries.Record(ctx, flowControlCount)
+				s.telemetryBuilder.SolacereceiverReceiverFlowControlRecentRetries.Record(ctx, flowControlCount, metric.WithAttributeSet(s.metricAttrs))
 				// Backpressure scenario. For now, we are only delayed retry, eventually we may need to handle this
 				delayTimer := time.NewTimer(s.config.Flow.DelayedRetry.Delay)
 				select {
@@ -261,21 +283,21 @@ flowControlLoop:
 				}
 			} else { // error is permanent, we want to accept the message and increment the number of dropped messages
 				s.settings.Logger.Warn("Encountered permanent error while forwarding traces to next receiver, will swallow trace", zap.Error(forwardErr))
-				s.telemetryBuilder.SolacereceiverDroppedSpanMessages.Add(ctx, 1)
+				s.telemetryBuilder.SolacereceiverDroppedSpanMessages.Add(ctx, 1, metric.WithAttributeSet(s.metricAttrs))
 				break flowControlLoop
 			}
 		} else {
 			// no forward error
-			s.telemetryBuilder.SolacereceiverReportedSpans.Add(ctx, int64(traces.SpanCount()))
+			s.telemetryBuilder.SolacereceiverReportedSpans.Add(ctx, int64(traces.SpanCount()), metric.WithAttributeSet(s.metricAttrs))
 			break flowControlLoop
 		}
 	}
 	// Make sure to clear the stats no matter what, unless we were interrupted in which case we should preserve the last state
 	if flowControlCount != 0 {
-		s.telemetryBuilder.SolacereceiverReceiverFlowControlStatus.Record(ctx, int64(flowControlStateClear))
-		s.telemetryBuilder.SolacereceiverReceiverFlowControlTotal.Add(ctx, 1)
+		s.telemetryBuilder.SolacereceiverReceiverFlowControlStatus.Record(ctx, int64(flowControlStateClear), metric.WithAttributeSet(s.metricAttrs))
+		s.telemetryBuilder.SolacereceiverReceiverFlowControlTotal.Add(ctx, 1, metric.WithAttributeSet(s.metricAttrs))
 		if flowControlCount == 1 {
-			s.telemetryBuilder.SolacereceiverReceiverFlowControlWithSingleSuccessfulRetry.Add(ctx, 1)
+			s.telemetryBuilder.SolacereceiverReceiverFlowControlWithSingleSuccessfulRetry.Add(ctx, 1, metric.WithAttributeSet(s.metricAttrs))
 		}
 	}
 	return nil

receiver/solacereceiver/unmarshaller.go

@@ -9,6 +9,7 @@ import (
 
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.opentelemetry.io/otel/attribute"
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/solacereceiver/internal/metadata"
@@ -22,18 +23,20 @@ type tracesUnmarshaller interface {
 }
 
 // newUnmarshalleer returns a new unmarshaller ready for message unmarshalling
-func newTracesUnmarshaller(logger *zap.Logger, telemetryBuilder *metadata.TelemetryBuilder) tracesUnmarshaller {
+func newTracesUnmarshaller(logger *zap.Logger, telemetryBuilder *metadata.TelemetryBuilder, metricAttrs attribute.Set) tracesUnmarshaller {
 	return &solaceTracesUnmarshaller{
 		logger:           logger,
 		telemetryBuilder: telemetryBuilder,
 		// v1 unmarshaller is implemented by solaceMessageUnmarshallerV1
 		receiveUnmarshallerV1: &brokerTraceReceiveUnmarshallerV1{
 			logger:           logger,
 			telemetryBuilder: telemetryBuilder,
+			metricAttrs:      metricAttrs,
 		},
 		egressUnmarshallerV1: &brokerTraceEgressUnmarshallerV1{
 			logger:           logger,
 			telemetryBuilder: telemetryBuilder,
+			metricAttrs:      metricAttrs,
 		},
 	}
 }

receiver/solacereceiver/unmarshaller_egress.go

@@ -11,6 +11,8 @@ import (
 
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
 	"go.uber.org/zap"
 	"google.golang.org/protobuf/proto"
 
@@ -21,6 +23,7 @@ import (
 type brokerTraceEgressUnmarshallerV1 struct {
 	logger           *zap.Logger
 	telemetryBuilder *metadata.TelemetryBuilder
+	metricAttrs      attribute.Set // othere Otel attributes (to add to the metrics)
 }
 
 func (u *brokerTraceEgressUnmarshallerV1) unmarshal(message *inboundMessage) (ptrace.Traces, error) {
@@ -68,9 +71,9 @@ func (u *brokerTraceEgressUnmarshallerV1) mapEgressSpan(spanData *egress_v1.Span
 			u.mapTransactionEvent(transactionEvent, clientSpan.Events().AppendEmpty())
 		}
 	} else {
-		// unknown span type, drop the span
-		u.logger.Warn("Received egress span with unknown span type, is the collector out of date?")
-		u.telemetryBuilder.SolacereceiverDroppedEgressSpans.Add(context.Background(), 1)
+		// malformed/incomplete egress span received, drop the span
+		u.logger.Warn("Received egress span with no span type, could be malformed egress span?")
+		u.telemetryBuilder.SolacereceiverDroppedEgressSpans.Add(context.Background(), 1, metric.WithAttributeSet(u.metricAttrs))
 	}
 }
 
@@ -143,7 +146,7 @@ func (u *brokerTraceEgressUnmarshallerV1) mapSendSpan(sendSpan *egress_v1.SpanDa
 		attributes.PutStr(sourceKindKey, queueKind)
 	default:
 		u.logger.Warn(fmt.Sprintf("Unknown source type %T", casted))
-		u.telemetryBuilder.SolacereceiverRecoverableUnmarshallingErrors.Add(context.Background(), 1)
+		u.telemetryBuilder.SolacereceiverRecoverableUnmarshallingErrors.Add(context.Background(), 1, metric.WithAttributeSet(u.metricAttrs))
 		name = unknownSendName
 	}
 	span.SetName(name + sendNameSuffix)
@@ -196,7 +199,7 @@ func (u *brokerTraceEgressUnmarshallerV1) mapTransactionEvent(transactionEvent *
 		// Set the name to the unknown transaction event type to ensure forward compat.
 		name = fmt.Sprintf("Unknown Transaction Event (%s)", transactionEvent.GetType().String())
 		u.logger.Warn(fmt.Sprintf("Received span with unknown transaction event %s", transactionEvent.GetType()))
-		u.telemetryBuilder.SolacereceiverRecoverableUnmarshallingErrors.Add(context.Background(), 1)
+		u.telemetryBuilder.SolacereceiverRecoverableUnmarshallingErrors.Add(context.Background(), 1, metric.WithAttributeSet(u.metricAttrs))
 	}
 	clientEvent.SetName(name)
 	clientEvent.SetTimestamp(pcommon.Timestamp(transactionEvent.TimeUnixNano))
@@ -212,7 +215,7 @@ func (u *brokerTraceEgressUnmarshallerV1) mapTransactionEvent(transactionEvent *
 	default:
 		initiator = fmt.Sprintf("Unknown Transaction Initiator (%s)", transactionEvent.GetInitiator().String())
 		u.logger.Warn(fmt.Sprintf("Received span with unknown transaction initiator %s", transactionEvent.GetInitiator()))
-		u.telemetryBuilder.SolacereceiverRecoverableUnmarshallingErrors.Add(context.Background(), 1)
+		u.telemetryBuilder.SolacereceiverRecoverableUnmarshallingErrors.Add(context.Background(), 1, metric.WithAttributeSet(u.metricAttrs))
 	}
 	clientEvent.Attributes().PutStr(transactionInitiatorEventKey, initiator)
 	// conditionally set the error description if one occurred, otherwise omit
@@ -233,7 +236,7 @@ func (u *brokerTraceEgressUnmarshallerV1) mapTransactionEvent(transactionEvent *
 		clientEvent.Attributes().PutStr(transactionXIDEventKey, xidString)
 	default:
 		u.logger.Warn(fmt.Sprintf("Unknown transaction ID type %T", transactionID))
-		u.telemetryBuilder.SolacereceiverRecoverableUnmarshallingErrors.Add(context.Background(), 1)
+		u.telemetryBuilder.SolacereceiverRecoverableUnmarshallingErrors.Add(context.Background(), 1, metric.WithAttributeSet(u.metricAttrs))
 	}
 }
 

receiver/solacereceiver/unmarshaller_egress_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	"go.uber.org/zap"
 
@@ -315,7 +316,8 @@ func TestEgressUnmarshallerEgressSpan(t *testing.T) {
 							IsMonotonic: true,
 							DataPoints: []metricdata.DataPoint[int64]{
 								{
-									Value: 1,
+									Value:      1,
+									Attributes: u.metricAttrs,
 								},
 							},
 						},
@@ -438,7 +440,8 @@ func TestEgressUnmarshallerSendSpanAttributes(t *testing.T) {
 						IsMonotonic: true,
 						DataPoints: []metricdata.DataPoint[int64]{
 							{
-								Value: tt.expectedUnmarshallingErrors,
+								Value:      tt.expectedUnmarshallingErrors,
+								Attributes: u.metricAttrs,
 							},
 						},
 					},
@@ -649,7 +652,8 @@ func TestEgressUnmarshallerTransactionEvent(t *testing.T) {
 						IsMonotonic: true,
 						DataPoints: []metricdata.DataPoint[int64]{
 							{
-								Value: tt.expectedUnmarshallingErrors,
+								Value:      tt.expectedUnmarshallingErrors,
+								Attributes: u.metricAttrs,
 							},
 						},
 					},
@@ -664,5 +668,6 @@ func newTestEgressV1Unmarshaller(t *testing.T) (*brokerTraceEgressUnmarshallerV1
 	tt := setupTestTelemetry()
 	builder, err := metadata.NewTelemetryBuilder(tt.NewSettings().TelemetrySettings)
 	require.NoError(t, err)
-	return &brokerTraceEgressUnmarshallerV1{zap.NewNop(), builder}, tt
+	metricAttr := attribute.NewSet(attribute.String("receiver_name", tt.NewSettings().ID.Name()))
+	return &brokerTraceEgressUnmarshallerV1{zap.NewNop(), builder, metricAttr}, tt
 }