Fix kubeletstats logging and self reported metrics (#357)

* Add logging on errors in Runnable

Kubeletstats receiver was missing log statements on
important errors.

* Fix self reported metrics

* Fix obsreport calls

Author: pmcollins

receiver/kubeletstatsreceiver/receiver.go

@@ -38,7 +38,7 @@ type receiver struct {
 
 // Creates and starts the kubelet stats runnable.
 func (r *receiver) Start(ctx context.Context, host component.Host) error {
-	runnable := newRunnable(ctx, r.consumer, r.rest, r.logger)
+	runnable := newRunnable(ctx, r.cfg.Name(), r.consumer, r.rest, r.logger)
 
 	cfg := r.cfg.(*Config)
 	r.runner = interval.NewRunner(cfg.CollectionInterval, runnable)

receiver/kubeletstatsreceiver/runnable.go

@@ -29,24 +29,27 @@ import (
 var _ interval.Runnable = (*runnable)(nil)
 
 type runnable struct {
-	ctx        context.Context
-	provider   *kubelet.StatsProvider
-	consumer   consumer.MetricsConsumerOld
-	logger     *zap.Logger
-	restClient kubelet.RestClient
+	ctx          context.Context
+	receiverName string
+	provider     *kubelet.StatsProvider
+	consumer     consumer.MetricsConsumerOld
+	logger       *zap.Logger
+	restClient   kubelet.RestClient
 }
 
 func newRunnable(
 	ctx context.Context,
+	receiverName string,
 	consumer consumer.MetricsConsumerOld,
 	restClient kubelet.RestClient,
 	logger *zap.Logger,
 ) *runnable {
 	return &runnable{
-		ctx:        ctx,
-		consumer:   consumer,
-		restClient: restClient,
-		logger:     logger,
+		ctx:          ctx,
+		receiverName: receiverName,
+		consumer:     consumer,
+		restClient:   restClient,
+		logger:       logger,
 	}
 }
 
@@ -57,19 +60,24 @@ func (r *runnable) Setup() error {
 }
 
 func (r *runnable) Run() error {
-	const dataformat = "kubelet"
 	const transport = "http"
-	ctx := obsreport.StartMetricsReceiveOp(r.ctx, dataformat, transport)
 	summary, err := r.provider.StatsSummary()
 	if err != nil {
-		obsreport.EndMetricsReceiveOp(ctx, dataformat, 0, 0, err)
+		r.logger.Error("StatsSummary failed", zap.Error(err))
 		return nil
 	}
 	mds := kubelet.MetricsData(summary, typeStr)
+	ctx := obsreport.ReceiverContext(r.ctx, typeStr, transport, r.receiverName)
 	for _, md := range mds {
-		err = r.consumer.ConsumeMetricsData(r.ctx, *md)
-		numTimeSeries, numPoints := obsreport.CountMetricPoints(*md)
-		obsreport.EndMetricsReceiveOp(ctx, dataformat, numTimeSeries, numPoints, err)
+		ctx = obsreport.StartMetricsReceiveOp(ctx, typeStr, transport)
+		err = r.consumer.ConsumeMetricsData(ctx, *md)
+		var numTimeSeries, numPoints int
+		if err != nil {
+			r.logger.Error("ConsumeMetricsData failed", zap.Error(err))
+		} else {
+			numTimeSeries, numPoints = obsreport.CountMetricPoints(*md)
+		}
+		obsreport.EndMetricsReceiveOp(ctx, typeStr, numTimeSeries, numPoints, err)
 	}
 	return nil
 }

receiver/kubeletstatsreceiver/runnable_test.go

@@ -16,20 +16,25 @@ package kubeletstatsreceiver
 
 import (
 	"context"
+	"errors"
 	"io/ioutil"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/consumer/consumerdata"
 	"go.uber.org/zap"
+	"go.uber.org/zap/zaptest/observer"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kubeletstatsreceiver/kubelet"
 )
 
+const dataLen = 19
+
 func TestRunnable(t *testing.T) {
 	consumer := &fakeConsumer{}
 	r := newRunnable(
 		context.Background(),
+		"",
 		consumer,
 		&fakeRestClient{},
 		zap.NewNop(),
@@ -38,29 +43,97 @@ func TestRunnable(t *testing.T) {
 	require.NoError(t, err)
 	err = r.Run()
 	require.NoError(t, err)
-	require.Equal(t, 19, len(consumer.mds))
+	require.Equal(t, dataLen, len(consumer.mds))
+}
+
+func TestClientErrors(t *testing.T) {
+	tests := []struct {
+		name    string
+		fail    bool
+		numLogs int
+	}{
+		{"no error", false, 0},
+		{"rest error", true, 1},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			core, observedLogs := observer.New(zap.ErrorLevel)
+			r := newRunnable(
+				context.Background(),
+				"",
+				&fakeConsumer{},
+				&fakeRestClient{fail: test.fail},
+				zap.New(core),
+			)
+			err := r.Setup()
+			require.NoError(t, err)
+			err = r.Run()
+			require.NoError(t, err)
+			require.Equal(t, test.numLogs, observedLogs.Len())
+		})
+	}
+}
+
+func TestConsumerErrors(t *testing.T) {
+	tests := []struct {
+		name    string
+		fail    bool
+		numLogs int
+	}{
+		{"no error", false, 0},
+		{"consume error", true, dataLen},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			core, observedLogs := observer.New(zap.ErrorLevel)
+			r := newRunnable(
+				context.Background(),
+				"",
+				&fakeConsumer{fail: test.fail},
+				&fakeRestClient{},
+				zap.New(core),
+			)
+			err := r.Setup()
+			require.NoError(t, err)
+			err = r.Run()
+			require.NoError(t, err)
+			require.Equal(t, test.numLogs, observedLogs.Len())
+		})
+	}
 }
 
 type fakeConsumer struct {
-	mds []consumerdata.MetricsData
+	fail bool
+	mds  []consumerdata.MetricsData
 }
 
 func (c *fakeConsumer) ConsumeMetricsData(
 	ctx context.Context,
 	md consumerdata.MetricsData,
 ) error {
+	if c.fail {
+		return errors.New("")
+	}
 	c.mds = append(c.mds, md)
 	return nil
 }
 
 var _ kubelet.RestClient = (*fakeRestClient)(nil)
 
-type fakeRestClient struct{}
+type fakeRestClient struct {
+	fail bool
+}
 
 func (f *fakeRestClient) StatsSummary() ([]byte, error) {
+	if f.fail {
+		return nil, errors.New("")
+	}
 	return ioutil.ReadFile("testdata/stats-summary.json")
 }
 
 func (f *fakeRestClient) Pods() ([]byte, error) {
+	if f.fail {
+		return nil, errors.New("")
+	}
 	return ioutil.ReadFile("testdata/pods.json")
 }