[HCP Observability] Add custom metrics for OTEL sink, improve logging, upgrade modules and cleanup metrics client

agent/hcp/client/metrics_client.go

@@ -144,16 +144,8 @@ func (o *otlpClient) ExportMetrics(ctx context.Context, protoMetrics *metricpb.R
 		return fmt.Errorf("failed to export metrics: %v", err)
 	}
 
-	if respData.Len() != 0 {
-		var respProto colmetricpb.ExportMetricsServiceResponse
-		if err := proto.Unmarshal(respData.Bytes(), &respProto); err != nil {
-			return fmt.Errorf("failed to export metrics: %v", err)
-		}
-
-		if respProto.PartialSuccess != nil {
-			msg := respProto.PartialSuccess.GetErrorMessage()
-			return fmt.Errorf("failed to export metrics: partial success: %s", msg)
-		}
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("failed to export metrics: code %d: %s", resp.StatusCode, string(body))
 	}
 
 	return nil

agent/hcp/client/metrics_client_test.go

@@ -72,7 +72,7 @@ func TestExportMetrics(t *testing.T) {
 		},
 		"failsWithNonRetryableError": {
 			status:  http.StatusBadRequest,
-			wantErr: "failed to export metrics",
+			wantErr: "failed to export metrics: code 400",
 		},
 	} {
 		t.Run(name, func(t *testing.T) {
@@ -82,12 +82,6 @@ func TestExportMetrics(t *testing.T) {
 				require.Equal(t, r.Header.Get("Authorization"), "Bearer test-token")
 
 				body := colpb.ExportMetricsServiceResponse{}
-
-				if test.wantErr != "" {
-					body.PartialSuccess = &colpb.ExportMetricsPartialSuccess{
-						ErrorMessage: "partial failure",
-					}
-				}
 				bytes, err := proto.Marshal(&body)
 
 				require.NoError(t, err)

agent/hcp/deps.go

@@ -27,15 +27,17 @@ type Deps struct {
 func NewDeps(cfg config.CloudConfig, logger hclog.Logger, nodeID types.NodeID) (d Deps, err error) {
 	d.Client, err = hcpclient.NewClient(cfg)
 	if err != nil {
+		logger.Error("failed to init HCP deps - client:", "error", err)
 		return
 	}
 
-	d.Provider, err = scada.New(cfg, logger.Named("hcp.scada"))
+	d.Provider, err = scada.New(cfg, logger.Named("scada"))
 	if err != nil {
+		logger.Error("failed to init HCP deps - scada:", "error", err)
 		return
 	}
 
-	d.Sink = sink(d.Client, &cfg, logger, nodeID)
+	d.Sink = sink(d.Client, &cfg, logger.Named("sink"), nodeID)
 
 	return
 }
@@ -86,5 +88,7 @@ func sink(hcpClient hcpclient.Client, cfg hcpclient.CloudConfig, logger hclog.Lo
 		return nil
 	}
 
+	logger.Info("Initialized HCP Metrics Sink")
+
 	return sink
 }

agent/hcp/telemetry/custom_metrics.go

@@ -0,0 +1,14 @@
+package telemetry
+
+// Keys for custom Go Metrics metrics emitted only for the OTEL
+// export (exporter.go) and transform (transform.go) failures and successes.
+// These enable us to monitor OTEL operations.
+var (
+	transformFailureMetric []string = []string{"hcp", "otel", "transform", "failure"}
+
+	exportSuccessMetric []string = []string{"hcp", "otel", "exporter", "export", "sucess"}
+	exportFailureMetric []string = []string{"hcp", "otel", "exporter", "export", "failure"}
+
+	exporterShutdownMetric   []string = []string{"hcp", "otel", "exporter", "shutdown"}
+	exporterForceFlushMetric []string = []string{"hcp", "otel", "exporter", "force_flush"}
+)

agent/hcp/telemetry/otel_exporter.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"net/url"
 
+	goMetrics "github.com/armon/go-metrics"
 	"go.opentelemetry.io/otel/sdk/metric"
 	"go.opentelemetry.io/otel/sdk/metric/aggregation"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
@@ -56,17 +57,24 @@ func (e *OTELExporter) Export(ctx context.Context, metrics *metricdata.ResourceM
 	if isEmpty(otlpMetrics) {
 		return nil
 	}
-	return e.client.ExportMetrics(ctx, otlpMetrics, e.endpoint.String())
+	err := e.client.ExportMetrics(ctx, otlpMetrics, e.endpoint.String())
+	if err != nil {
+		goMetrics.IncrCounter(exportFailureMetric, 1)
+		return err
+	}
+
+	goMetrics.IncrCounter(exportSuccessMetric, 1)
+	return nil
 }
 
 // ForceFlush is a no-op, as the MetricsClient client holds no state.
 func (e *OTELExporter) ForceFlush(ctx context.Context) error {
-	// TODO: Emit metric when this operation occurs.
+	goMetrics.IncrCounter(exporterForceFlushMetric, 1)
 	return ctx.Err()
 }
 
 // Shutdown is a no-op, as the MetricsClient is a HTTP client that requires no graceful shutdown.
 func (e *OTELExporter) Shutdown(ctx context.Context) error {
-	// TODO: Emit metric when this operation occurs.
+	goMetrics.IncrCounter(exporterShutdownMetric, 1)
 	return ctx.Err()
 }

agent/hcp/telemetry/otel_exporter_test.go

@@ -4,8 +4,11 @@ import (
 	"context"
 	"fmt"
 	"net/url"
+	"strings"
 	"testing"
+	"time"
 
+	"github.com/armon/go-metrics"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/otel/sdk/metric"
 	"go.opentelemetry.io/otel/sdk/metric/aggregation"
@@ -16,6 +19,14 @@ import (
 	"github.com/hashicorp/consul/agent/hcp/client"
 )
 
+type mockMetricsClient struct {
+	exportErr error
+}
+
+func (m *mockMetricsClient) ExportMetrics(ctx context.Context, protoMetrics *metricpb.ResourceMetrics, endpoint string) error {
+	return m.exportErr
+}
+
 func TestTemporality(t *testing.T) {
 	t.Parallel()
 	exp := &OTELExporter{}
@@ -50,14 +61,6 @@ func TestAggregation(t *testing.T) {
 	}
 }
 
-type mockMetricsClient struct {
-	exportErr error
-}
-
-func (m *mockMetricsClient) ExportMetrics(ctx context.Context, protoMetrics *metricpb.ResourceMetrics, endpoint string) error {
-	return m.exportErr
-}
-
 func TestExport(t *testing.T) {
 	t.Parallel()
 	for name, test := range map[string]struct {
@@ -111,6 +114,58 @@ func TestExport(t *testing.T) {
 	}
 }
 
+// TestExport_CustomMetrics tests that a custom metric (hcp.otel.exporter.*) is emitted
+// for exporter operations. This test cannot be run in parallel as the metrics.NewGlobal()
+// sets a shared global sink.
+func TestExport_CustomMetrics(t *testing.T) {
+	for name, tc := range map[string]struct {
+		client    client.MetricsClient
+		metricKey []string
+	}{
+		"exportSuccess": {
+			client:    &mockMetricsClient{},
+			metricKey: exportSuccessMetric,
+		},
+		"exportFailure": {
+			client: &mockMetricsClient{
+				exportErr: fmt.Errorf("failed to export metrics"),
+			},
+			metricKey: exportFailureMetric,
+		},
+		"shutdown": {
+			metricKey: exporterShutdownMetric,
+		},
+		"forceFlush": {
+			metricKey: exporterForceFlushMetric,
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			// Init global sink.
+			serviceName := "test.transform"
+			cfg := metrics.DefaultConfig(serviceName)
+			cfg.EnableHostname = false
+
+			sink := metrics.NewInmemSink(10*time.Second, 10*time.Second)
+			metrics.NewGlobal(cfg, sink)
+
+			// Perform operation that emits metric.
+			exp := NewOTELExporter(tc.client, &url.URL{})
+			performExporterOperation(exp, name)
+
+			// Collect sink metrics.
+			intervals := sink.Data()
+			require.Len(t, intervals, 1)
+			key := serviceName + "." + strings.Join(tc.metricKey, ".")
+			sv := intervals[0].Counters[key]
+
+			// Verify count for transform failure metric.
+			require.NotNil(t, sv)
+			require.NotNil(t, sv.AggregateSample)
+			require.Equal(t, 1, sv.AggregateSample.Count)
+		})
+	}
+}
+
 func TestForceFlush(t *testing.T) {
 	t.Parallel()
 	exp := &OTELExporter{}
@@ -137,3 +192,15 @@ func mutateMetrics(m []metricdata.ScopeMetrics) *metricdata.ResourceMetrics {
 		ScopeMetrics: m,
 	}
 }
+
+func performExporterOperation(exp metric.Exporter, operation string) {
+	ctx := context.Background()
+	switch operation {
+	case "forceFlush":
+		exp.ForceFlush(ctx)
+	case "shutdown":
+		exp.Shutdown(ctx)
+	default:
+		exp.Export(ctx, inputResourceMetrics)
+	}
+}

agent/hcp/telemetry/otlp_transform.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"fmt"
 
+	goMetrics "github.com/armon/go-metrics"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	cpb "go.opentelemetry.io/proto/otlp/common/v1"
@@ -70,7 +71,7 @@ func metricsToPB(metrics []metricdata.Metrics) []*mpb.Metric {
 	for _, m := range metrics {
 		o, err := metricTypeToPB(m)
 		if err != nil {
-			// TODO: Emit metric when a transformation failure occurs.
+			goMetrics.IncrCounter(transformFailureMetric, 1)
 			continue
 		}
 		out = append(out, o)