Feat: expand kafka broker metrics

.chloggen/expand-kafka-broker-metrics.yaml

@@ -0,0 +1,16 @@
+# 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: kafkametricsreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "expanding the broker metrics that are scraped."
+
+# One or more tracking issues related to the change
+issues: [14166]
+
+# (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:

receiver/kafkametricsreceiver/broker_scraper.go

@@ -26,6 +26,20 @@ type brokerScraper struct {
 	mb           *metadata.MetricsBuilder
 }
 
+type saramaMetrics map[string]map[string]interface{} // saramaMetrics is a map of metric name to tags
+
+var nrMetricsPrefix = [...]string{
+	"consumer-fetch-rate-for-broker-",
+	"incoming-byte-rate-for-broker-",
+	"outgoing-byte-rate-for-broker-",
+	"request-rate-for-broker-",
+	"response-rate-for-broker-",
+	"response-size-for-broker-",
+	"request-size-for-broker-",
+	"requests-in-flight-for-broker-",
+	"request-latency-in-ms-for-broker-",
+}
+
 func (s *brokerScraper) Name() string {
 	return brokersScraperName
 }
@@ -42,6 +56,53 @@ func (s *brokerScraper) shutdown(context.Context) error {
 	return nil
 }
 
+func (s *brokerScraper) scrapeMetric(now pcommon.Timestamp, allMetrics saramaMetrics, brokerID int64, prefix string) {
+	key := fmt.Sprint(prefix, brokerID)
+
+	if metric, ok := allMetrics[key]; ok {
+		switch prefix {
+		case "consumer-fetch-rate-for-broker-":
+			if v, ok := metric["mean.rate"].(float64); ok {
+				s.mb.RecordMessagingKafkaBrokerConsumerFetchRateDataPoint(now, v, brokerID)
+			}
+		case "incoming-byte-rate-for-broker-":
+			if v, ok := metric["mean.rate"].(float64); ok {
+				s.mb.RecordMessagingKafkaBrokerIncomingByteRateDataPoint(now, v, brokerID)
+			}
+		case "outgoing-byte-rate-for-broker-":
+			if v, ok := metric["mean.rate"].(float64); ok {
+				s.mb.RecordMessagingKafkaBrokerOutgoingByteRateDataPoint(now, v, brokerID)
+			}
+		case "request-rate-for-broker-":
+			if v, ok := metric["mean.rate"].(float64); ok {
+				s.mb.RecordMessagingKafkaBrokerRequestRateDataPoint(now, v, brokerID)
+			}
+		case "response-rate-for-broker-":
+			if v, ok := metric["mean.rate"].(float64); ok {
+				s.mb.RecordMessagingKafkaBrokerResponseRateDataPoint(now, v, brokerID)
+			}
+		case "response-size-for-broker-":
+			if v, ok := metric["mean"].(float64); ok {
+				s.mb.RecordMessagingKafkaBrokerResponseSizeDataPoint(now, v, brokerID)
+			}
+		case "request-size-for-broker-":
+			if v, ok := metric["mean"].(float64); ok {
+				s.mb.RecordMessagingKafkaBrokerRequestSizeDataPoint(now, v, brokerID)
+			}
+		case "requests-in-flight-for-broker-":
+			if v, ok := metric["count"].(int64); ok {
+				s.mb.RecordMessagingKafkaBrokerRequestsInFlightDataPoint(now, v, brokerID)
+			}
+		case "request-latency-in-ms-for-broker-":
+			if v, ok := metric["mean"].(float64); ok {
+				s.mb.RecordMessagingKafkaBrokerRequestLatencyDataPoint(now, v, brokerID)
+			}
+		default:
+			fmt.Printf("undefined for prefix %s\n", prefix)
+		}
+	}
+}
+
 func (s *brokerScraper) scrape(context.Context) (pmetric.Metrics, error) {
 	if s.client == nil {
 		client, err := newSaramaClient(s.config.Brokers, s.saramaConfig)
@@ -53,7 +114,26 @@ func (s *brokerScraper) scrape(context.Context) (pmetric.Metrics, error) {
 
 	brokers := s.client.Brokers()
 
-	s.mb.RecordKafkaBrokersDataPoint(pcommon.NewTimestampFromTime(time.Now()), int64(len(brokers)))
+	allMetrics := make(map[string]map[string]interface{})
+
+	if s.saramaConfig != nil {
+		allMetrics = s.saramaConfig.MetricRegistry.GetAll()
+	}
+
+	now := pcommon.NewTimestampFromTime(time.Now())
+	for _, broker := range brokers {
+		brokerID := int64(broker.ID())
+		for _, prefix := range nrMetricsPrefix {
+			s.scrapeMetric(now, allMetrics, brokerID, prefix)
+		}
+	}
+
+	brokerCount := int64(len(brokers))
+	// kafka.brokers is deprecated. This should be removed in a future release.
+	s.mb.RecordKafkaBrokersDataPoint(now, brokerCount)
+
+	// messaging.kafka.broker.count should replace kafka.brokers.
+	s.mb.RecordMessagingKafkaBrokerCountDataPoint(now, brokerCount)
 
 	return s.mb.Emit(), nil
 }

receiver/kafkametricsreceiver/broker_scraper_test.go

@@ -104,10 +104,38 @@ func TestBrokerScraper_scrape(t *testing.T) {
 	require.NoError(t, bs.start(context.Background(), componenttest.NewNopHost()))
 	md, err := bs.scrape(context.Background())
 	assert.NoError(t, err)
-	expectedDp := int64(len(testBrokers))
-	receivedMetrics := md.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(0)
-	receivedDp := receivedMetrics.Sum().DataPoints().At(0).IntValue()
-	assert.Equal(t, expectedDp, receivedDp)
+	require.Equal(t, 1, md.ResourceMetrics().Len())
+	require.Equal(t, 1, md.ResourceMetrics().At(0).ScopeMetrics().Len())
+	ms := md.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics()
+	for i := 0; i < ms.Len(); i++ {
+		m := ms.At(i)
+		switch m.Name() {
+		case "kafka.brokers":
+			assert.Equal(t, m.Sum().DataPoints().At(0).IntValue(), int64(len(testBrokers)))
+		case "kafka.broker.count":
+			assert.Equal(t, m.Sum().DataPoints().At(0).IntValue(), int64(len(testBrokers)))
+		case "kafka.broker.consumer_fetch_rate":
+			assert.Equal(t, m.Gauge().DataPoints().At(0).DoubleValue(), int64(len(testBrokers)))
+		case "kafka.broker.incoming_byte_rate":
+			assert.Equal(t, m.Gauge().DataPoints().At(0).DoubleValue(), int64(len(testBrokers)))
+		case "kafka.broker.outgoing_byte_rate":
+			assert.Equal(t, m.Gauge().DataPoints().At(0).DoubleValue(), int64(len(testBrokers)))
+		case "kafka.broker.request_latency":
+			assert.Equal(t, m.Gauge().DataPoints().At(0).DoubleValue(), int64(len(testBrokers)))
+		case "kafka.broker.response_rate":
+			assert.Equal(t, m.Gauge().DataPoints().At(0).DoubleValue(), int64(len(testBrokers)))
+		case "kafka.broker.response_size":
+			assert.Equal(t, m.Gauge().DataPoints().At(0).DoubleValue(), int64(len(testBrokers)))
+		case "kafka.broker.request_rate":
+			assert.Equal(t, m.Gauge().DataPoints().At(0).DoubleValue(), int64(len(testBrokers)))
+		case "kafka.broker.request_size":
+			assert.Equal(t, m.Gauge().DataPoints().At(0).DoubleValue(), int64(len(testBrokers)))
+		case "kafka.broker.requests_in_flight":
+			assert.Equal(t, m.Sum().DataPoints().At(0).IntValue(), int64(len(testBrokers)))
+		case "kafka.broker.consumer_fetch_count":
+			assert.Equal(t, m.Sum().DataPoints().At(0).IntValue(), int64(len(testBrokers)))
+		}
+	}
 }
 
 func TestBrokersScraper_createBrokerScraper(t *testing.T) {

receiver/kafkametricsreceiver/documentation.md

@@ -14,11 +14,11 @@ metrics:
 
 ### kafka.brokers
 
-Number of brokers in the cluster.
+[DEPRACATED] Number of brokers in the cluster.
 
 | Unit | Metric Type | Value Type | Aggregation Temporality | Monotonic |
 | ---- | ----------- | ---------- | ----------------------- | --------- |
-| {brokers} | Sum | Int | Cumulative | false |
+| {broker} | Sum | Int | Cumulative | false |
 
 ### kafka.consumer_group.lag
 
@@ -169,3 +169,161 @@ Number of partitions in topic.
 | Name | Description | Values |
 | ---- | ----------- | ------ |
 | topic | The ID (integer) of a topic | Any Str |
+
+## Optional Metrics
+
+The following metrics are not emitted by default. Each of them can be enabled by applying the following configuration:
+
+```yaml
+metrics:
+  <metric_name>:
+    enabled: true
+```
+
+### messaging.kafka.broker.consumer_fetch_count
+
+Count of consumer fetches
+
+| Unit | Metric Type | Value Type | Aggregation Temporality | Monotonic |
+| ---- | ----------- | ---------- | ----------------------- | --------- |
+| {fetches} | Sum | Int | Cumulative | false |
+
+#### Attributes
+
+| Name | Description | Values |
+| ---- | ----------- | ------ |
+| broker | The ID (integer) of a broker | Any Int |
+
+### messaging.kafka.broker.consumer_fetch_rate
+
+Average consumer fetch Rate
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| {fetches}/s | Gauge | Double |
+
+#### Attributes
+
+| Name | Description | Values |
+| ---- | ----------- | ------ |
+| broker | The ID (integer) of a broker | Any Int |
+
+### messaging.kafka.broker.count
+
+Number of brokers in the cluster.
+
+| Unit | Metric Type | Value Type | Aggregation Temporality | Monotonic |
+| ---- | ----------- | ---------- | ----------------------- | --------- |
+| {broker} | Sum | Int | Cumulative | true |
+
+### messaging.kafka.broker.incoming_byte_rate
+
+Average incoming Byte Rate in bytes/second
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| By/s | Gauge | Double |
+
+#### Attributes
+
+| Name | Description | Values |
+| ---- | ----------- | ------ |
+| broker | The ID (integer) of a broker | Any Int |
+
+### messaging.kafka.broker.outgoing_byte_rate
+
+Average outgoing Byte Rate in bytes/second.
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| 1 | Gauge | Double |
+
+#### Attributes
+
+| Name | Description | Values |
+| ---- | ----------- | ------ |
+| broker | The ID (integer) of a broker | Any Int |
+
+### messaging.kafka.broker.request_latency
+
+Average request latency in seconds
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| s | Gauge | Double |
+
+#### Attributes
+
+| Name | Description | Values |
+| ---- | ----------- | ------ |
+| broker | The ID (integer) of a broker | Any Int |
+
+### messaging.kafka.broker.request_rate
+
+Average request rate per second.
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| {requests}/s | Gauge | Double |
+
+#### Attributes
+
+| Name | Description | Values |
+| ---- | ----------- | ------ |
+| broker | The ID (integer) of a broker | Any Int |
+
+### messaging.kafka.broker.request_size
+
+Average request size in bytes
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| By | Gauge | Double |
+
+#### Attributes
+
+| Name | Description | Values |
+| ---- | ----------- | ------ |
+| broker | The ID (integer) of a broker | Any Int |
+
+### messaging.kafka.broker.requests_in_flight
+
+Requests in flight
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| {requests} | Gauge | Int |
+
+#### Attributes
+
+| Name | Description | Values |
+| ---- | ----------- | ------ |
+| broker | The ID (integer) of a broker | Any Int |
+
+### messaging.kafka.broker.response_rate
+
+Average response rate per second
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| {response}/s | Gauge | Double |
+
+#### Attributes
+
+| Name | Description | Values |
+| ---- | ----------- | ------ |
+| broker | The ID (integer) of a broker | Any Int |
+
+### messaging.kafka.broker.response_size
+
+Average response size in bytes
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| By | Gauge | Double |
+
+#### Attributes
+
+| Name | Description | Values |
+| ---- | ----------- | ------ |
+| broker | The ID (integer) of a broker | Any Int |

receiver/kafkametricsreceiver/integration_test.go

@@ -67,13 +67,41 @@ func TestIntegration(t *testing.T) {
 				rCfg.Brokers = []string{fmt.Sprintf("%s:%s",
 					ci.HostForNamedContainer(t, "kafka"),
 					ci.MappedPortForNamedContainer(t, "kafka", kafkaPort))}
-				rCfg.Scrapers = []string{"brokers", "consumers", "topics"}
+				rCfg.Scrapers = []string{
+					"brokers",
+					"consumers",
+					"topics",
+				}
+				rCfg.Metrics.KafkaBrokers.Enabled = false
+				rCfg.Metrics.KafkaConsumerGroupLag.Enabled = true
+				rCfg.Metrics.KafkaConsumerGroupLagSum.Enabled = true
+				rCfg.Metrics.KafkaConsumerGroupMembers.Enabled = true
+				rCfg.Metrics.KafkaConsumerGroupOffset.Enabled = true
+				rCfg.Metrics.KafkaConsumerGroupOffsetSum.Enabled = true
+				rCfg.Metrics.KafkaPartitionCurrentOffset.Enabled = true
+				rCfg.Metrics.KafkaPartitionOldestOffset.Enabled = true
+				rCfg.Metrics.KafkaPartitionReplicas.Enabled = true
+				rCfg.Metrics.KafkaPartitionReplicasInSync.Enabled = true
+				rCfg.Metrics.KafkaTopicPartitions.Enabled = true
+				rCfg.Metrics.MessagingKafkaBrokerConsumerFetchCount.Enabled = true
+				rCfg.Metrics.MessagingKafkaBrokerConsumerFetchRate.Enabled = true
+				rCfg.Metrics.MessagingKafkaBrokerCount.Enabled = true
+				rCfg.Metrics.MessagingKafkaBrokerIncomingByteRate.Enabled = true
+				rCfg.Metrics.MessagingKafkaBrokerOutgoingByteRate.Enabled = true
+				rCfg.Metrics.MessagingKafkaBrokerRequestLatency.Enabled = true
+				rCfg.Metrics.MessagingKafkaBrokerRequestRate.Enabled = true
+				rCfg.Metrics.MessagingKafkaBrokerRequestSize.Enabled = true
+				rCfg.Metrics.MessagingKafkaBrokerRequestsInFlight.Enabled = true
+				rCfg.Metrics.MessagingKafkaBrokerResponseRate.Enabled = true
+				rCfg.Metrics.MessagingKafkaBrokerResponseSize.Enabled = true
 			}),
+
 		// scraperinttest.WriteExpected(), // TODO remove
 		scraperinttest.WithCompareOptions(
 			// pmetrictest.IgnoreMetricValues(),
 			pmetrictest.IgnoreStartTimestamp(),
 			pmetrictest.IgnoreTimestamp(),
 		),
+		scraperinttest.WriteExpected(),
 	).Run(t)
 }