Add brokers metrics and metrics metadata to Kafkametrics

receiver/kafkametricsreceiver/broker_scraper.go

@@ -16,11 +16,15 @@ package kafkametricsreceiver
 
 import (
 	"context"
+	"time"
 
 	"github.com/Shopify/sarama"
 	"go.opentelemetry.io/collector/consumer/pdata"
+	"go.opentelemetry.io/collector/consumer/simple"
 	"go.opentelemetry.io/collector/receiver/scraperhelper"
 	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkametricsreceiver/internal/metadata"
 )
 
 type brokerScraper struct {
@@ -40,13 +44,20 @@ func (s *brokerScraper) shutdown(context.Context) error {
 	return nil
 }
 
-func (s *brokerScraper) scrape(context.Context) (pdata.MetricSlice, error) {
-	metrics := pdata.NewMetricSlice()
-	// implement brokers metrics
-	return metrics, nil
+func (s *brokerScraper) scrape(context.Context) (pdata.ResourceMetricsSlice, error) {
+	brokers := s.client.Brokers()
+	metrics := simple.Metrics{
+		Metrics:                    pdata.NewMetrics(),
+		Timestamp:                  time.Now(),
+		MetricFactoriesByName:      metadata.M.FactoriesByName(),
+		InstrumentationLibraryName: "otelcol/kafkametrics",
+	}
+	metrics.AddGaugeDataPoint(metadata.M.KafkaBrokers.Name(), int64(len(brokers)))
+
+	return metrics.Metrics.ResourceMetrics(), nil
 }
 
-func createBrokerScraper(_ context.Context, config Config, saramaConfig *sarama.Config, logger *zap.Logger) (scraperhelper.MetricsScraper, error) {
+func createBrokerScraper(_ context.Context, config Config, saramaConfig *sarama.Config, logger *zap.Logger) (scraperhelper.ResourceMetricsScraper, error) {
 	client, err := newSaramaClient(config.Brokers, saramaConfig)
 	if err != nil {
 		return nil, err
@@ -56,7 +67,7 @@ func createBrokerScraper(_ context.Context, config Config, saramaConfig *sarama.
 		logger: logger,
 		config: config,
 	}
-	ms := scraperhelper.NewMetricsScraper(
+	ms := scraperhelper.NewResourceMetricsScraper(
 		s.Name(),
 		s.scrape,
 		scraperhelper.WithShutdown(s.shutdown),

receiver/kafkametricsreceiver/broker_scraper_test.go

@@ -82,6 +82,7 @@ func TestBrokerScraper_scrape(t *testing.T) {
 	r := sarama.NewBroker(testBroker)
 	testBrokers := make([]*sarama.Broker, 1)
 	testBrokers[0] = r
+	client.brokers = testBrokers
 	client.Mock.On("Brokers").Return(testBrokers)
 	bs := brokerScraper{
 		client: client,
@@ -90,6 +91,10 @@ func TestBrokerScraper_scrape(t *testing.T) {
 	}
 	ms, err := bs.scrape(context.Background())
 	assert.Nil(t, err)
+	expectedDp := int64(len(testBrokers))
+	receivedMetrics := ms.At(0).InstrumentationLibraryMetrics().At(0).Metrics().At(0)
+	receivedDp := receivedMetrics.IntGauge().DataPoints().At(0).Value()
+	assert.Equal(t, expectedDp, receivedDp)
 	assert.NotNil(t, ms)
 }
 

receiver/kafkametricsreceiver/codegen.go

@@ -0,0 +1,19 @@
+// Copyright 2020 The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// +build !windows
+
+//go:generate mdatagen metadata.yaml
+
+package kafkametricsreceiver

receiver/kafkametricsreceiver/internal/metadata/.nocover

@@ -0,0 +1 @@
+We test the generated code in the package it is used.