Add exporting of Envoy metrics to HCP for linked clusters

.changelog/370.txt

@@ -0,0 +1,3 @@
+```release-note:feature
+Add metrics exporting directly to HCP when configured in core.
+```

.changelog/406.txt

@@ -0,0 +1,3 @@
+```release-note:improvement
+Upgrade consul/proto-public.
+```

go.mod

@@ -11,18 +11,28 @@ require (
 	github.com/hashi-derek/grpc-proxy v0.0.0-20231207191910-191266484d75
 	github.com/hashicorp/consul-server-connection-manager v0.1.6
 	github.com/hashicorp/consul/proto-public v0.4.1
+	github.com/hashicorp/go-cleanhttp v0.5.2
 	github.com/hashicorp/go-hclog v1.2.2
 	github.com/hashicorp/go-multierror v1.1.1
 	github.com/hashicorp/go-rootcerts v1.0.2
+	github.com/hashicorp/go-uuid v1.0.3
+	github.com/hashicorp/hcp-sdk-go v0.77.0
 	github.com/mitchellh/mapstructure v1.5.0
-	github.com/prometheus/client_golang v1.13.0
-	github.com/stretchr/testify v1.8.3
-	google.golang.org/grpc v1.56.3
-	google.golang.org/protobuf v1.30.0
+	github.com/prometheus/client_golang v1.17.0
+	github.com/prometheus/client_model v0.5.0
+	github.com/prometheus/common v0.44.0
+	github.com/stretchr/testify v1.8.4
+	go.opentelemetry.io/collector/pdata v1.0.0
+	golang.org/x/net v0.18.0
+	golang.org/x/oauth2 v0.14.0
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d
+	google.golang.org/grpc v1.59.0
+	google.golang.org/protobuf v1.31.0
 )
 
 require (
-	github.com/DataDog/datadog-go v3.2.0+incompatible // indirect
+	github.com/DataDog/datadog-go v4.8.2+incompatible // indirect
+	github.com/Microsoft/go-winio v0.6.1 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cenkalti/backoff/v4 v4.1.3 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
@@ -35,26 +45,28 @@ require (
 	github.com/hashicorp/errwrap v1.0.0 // indirect
 	github.com/hashicorp/go-immutable-radix v1.0.0 // indirect
 	github.com/hashicorp/go-netaddrs v0.1.0 // indirect
-	github.com/hashicorp/go-uuid v1.0.2 // indirect
 	github.com/hashicorp/golang-lru v0.5.1 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/mattn/go-colorable v0.1.12 // indirect
 	github.com/mattn/go-isatty v0.0.14 // indirect
-	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
+	github.com/mitchellh/colorstring v0.0.0-20190213212951-d06e56a500db // indirect
 	github.com/mitchellh/go-homedir v1.1.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
-	github.com/prometheus/client_model v0.2.0 // indirect
-	github.com/prometheus/common v0.37.0 // indirect
-	github.com/prometheus/procfs v0.8.0 // indirect
+	github.com/prometheus/procfs v0.11.1 // indirect
+	github.com/rogpeppe/go-internal v1.11.0 // indirect
+	github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966 // indirect
 	github.com/stretchr/objx v0.5.0 // indirect
+	github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb // indirect
+	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/exp v0.0.0-20230817173708-d852ddb80c63 // indirect
-	golang.org/x/net v0.17.0 // indirect
-	golang.org/x/sync v0.3.0 // indirect
+	golang.org/x/mod v0.12.0 // indirect
 	golang.org/x/sys v0.15.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20230526203410-71b5a4ffd15e // indirect
+	golang.org/x/tools v0.12.1-0.20230815132531-74c255bcf846 // indirect
+	google.golang.org/appengine v1.6.7 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect

pkg/consuldp/consul_dataplane.go

@@ -16,11 +16,13 @@ import (
 	"github.com/hashicorp/consul-server-connection-manager/discovery"
 	"github.com/hashicorp/consul/proto-public/pbdataplane"
 	"github.com/hashicorp/consul/proto-public/pbdns"
+	"github.com/hashicorp/consul/proto-public/pbresource"
 	"github.com/hashicorp/go-hclog"
 	"google.golang.org/grpc"
 
 	"github.com/hashicorp/consul-dataplane/pkg/dns"
 	"github.com/hashicorp/consul-dataplane/pkg/envoy"
+	"github.com/hashicorp/consul-dataplane/pkg/hcp/telemetry"
 	metricscache "github.com/hashicorp/consul-dataplane/pkg/metrics-cache"
 )
 
@@ -43,6 +45,7 @@ type ConsulDataplane struct {
 	cfg             *Config
 	serverConn      *grpc.ClientConn
 	dpServiceClient pbdataplane.DataplaneServiceClient
+	resourceClient  pbresource.ResourceServiceClient
 	xdsServer       *xdsServer
 	aclToken        string
 	metricsConfig   *metricsConfig
@@ -175,6 +178,7 @@ func (cdp *ConsulDataplane) Run(ctx context.Context) error {
 	cdp.serverConn = state.GRPCConn
 	cdp.aclToken = state.Token
 	cdp.dpServiceClient = pbdataplane.NewDataplaneServiceClient(state.GRPCConn)
+	cdp.resourceClient = pbresource.NewResourceServiceClient(state.GRPCConn)
 
 	err = cdp.setupXDSServer()
 	if err != nil {
@@ -216,6 +220,13 @@ func (cdp *ConsulDataplane) Run(ctx context.Context) error {
 		return err
 	}
 
+	go telemetry.NewHCPExporter(
+		cdp.resourceClient,
+		cdp.logger.Named("hcp_telemetry"),
+		fmt.Sprintf("%s:%d", cdp.cfg.Envoy.AdminBindAddress, cdp.cfg.Envoy.AdminBindPort),
+		cdp.cfg.Proxy.ProxyID,
+	).Run(ctx)
+
 	doneCh := make(chan error)
 	go func() {
 		select {

pkg/consuldp/metrics_test.go

@@ -51,7 +51,7 @@ func assertServerMatchesExpected(t *testing.T, server *net.UDPConn, buf []byte,
 	n, _ := server.Read(buf)
 	msg := string(buf[:n])
 
-	require.Equal(t, msg, expected, fmt.Sprintf("Line %s does not match expected: %s", msg, expected))
+	require.Equal(t, expected, msg, fmt.Sprintf("Line %s does not match expected: %s", msg, expected))
 
 }
 
@@ -317,14 +317,14 @@ func TestMetricsDatadogWithoutGlobalTags(t *testing.T) {
 		Tags     []metrics.Label
 		Expected string
 	}{
-		{"SetGauge", []string{"foo", "bar"}, float32(42), emptyTags, fmt.Sprintf("%v.foo.bar:42|g", prefix)},
-		{"SetGauge", []string{"foo", "bar", "baz"}, float32(42), emptyTags, fmt.Sprintf("%v.foo.bar.baz:42|g", prefix)},
-		{"AddSample", []string{"sample", "thing"}, float32(4), emptyTags, fmt.Sprintf("%v.sample.thing:4.000000|ms", prefix)},
-		{"IncrCounter", []string{"count", "me"}, float32(3), emptyTags, fmt.Sprintf("%v.count.me:3|c", prefix)},
-
-		{"SetGauge", []string{"foo", "baz"}, float32(42), []metrics.Label{{Name: tag, Value: ""}}, fmt.Sprintf("%v.foo.baz:42|g|#my_tag", prefix)},
-		{"SetGauge", []string{"foo", "baz"}, float32(42), []metrics.Label{{Name: tag, Value: tagValue}}, fmt.Sprintf("%v.foo.baz:42|g|#my_tag:my_value", prefix)},
-		{"SetGauge", []string{"foo", "bar"}, float32(42), []metrics.Label{{Name: tag, Value: tagValue}, {Name: "other_tag", Value: "other_value"}}, fmt.Sprintf("%v.foo.bar:42|g|#my_tag:my_value,other_tag:other_value", prefix)},
+		{"SetGauge", []string{"foo", "bar"}, float32(42), emptyTags, fmt.Sprintf("%v.foo.bar:42|g\n", prefix)},
+		{"SetGauge", []string{"foo", "bar", "baz"}, float32(42), emptyTags, fmt.Sprintf("%v.foo.bar.baz:42|g\n", prefix)},
+		{"AddSample", []string{"sample", "thing"}, float32(4), emptyTags, fmt.Sprintf("%v.sample.thing:4.000000|ms\n", prefix)},
+		{"IncrCounter", []string{"count", "me"}, float32(3), emptyTags, fmt.Sprintf("%v.count.me:3|c\n", prefix)},
+
+		{"SetGauge", []string{"foo", "baz"}, float32(42), []metrics.Label{{Name: tag, Value: ""}}, fmt.Sprintf("%v.foo.baz:42|g|#my_tag\n", prefix)},
+		{"SetGauge", []string{"foo", "baz"}, float32(42), []metrics.Label{{Name: tag, Value: tagValue}}, fmt.Sprintf("%v.foo.baz:42|g|#my_tag:my_value\n", prefix)},
+		{"SetGauge", []string{"foo", "bar"}, float32(42), []metrics.Label{{Name: tag, Value: tagValue}, {Name: "other_tag", Value: "other_value"}}, fmt.Sprintf("%v.foo.bar:42|g|#my_tag:my_value,other_tag:other_value\n", prefix)},
 	}
 
 	server, buf := setupTestServerAndBuffer(t)
@@ -384,14 +384,14 @@ func TestMetricsDatadogWithGlobalTags(t *testing.T) {
 		Tags     []metrics.Label
 		Expected string
 	}{
-		{"SetGauge", []string{"foo", "bar"}, float32(42), emptyTags, fmt.Sprintf("%v.foo.bar:42|g|#%v", prefix, globalTags)},
-		{"SetGauge", []string{"foo", "bar", "baz"}, float32(42), emptyTags, fmt.Sprintf("%v.foo.bar.baz:42|g|#%v", prefix, globalTags)},
-		{"AddSample", []string{"sample", "thing"}, float32(4), emptyTags, fmt.Sprintf("%v.sample.thing:4.000000|ms|#%v", prefix, globalTags)},
-		{"IncrCounter", []string{"count", "me"}, float32(3), emptyTags, fmt.Sprintf("%v.count.me:3|c|#%v", prefix, globalTags)},
-
-		{"SetGauge", []string{"foo", "baz"}, float32(42), []metrics.Label{{Name: tag, Value: ""}}, fmt.Sprintf("%v.foo.baz:42|g|#%v,my_tag", prefix, globalTags)},
-		{"SetGauge", []string{"foo", "baz"}, float32(42), []metrics.Label{{Name: tag, Value: tagValue}}, fmt.Sprintf("%v.foo.baz:42|g|#%v,my_tag:my_value", prefix, globalTags)},
-		{"SetGauge", []string{"foo", "bar"}, float32(42), []metrics.Label{{Name: tag, Value: tagValue}, {Name: "other_tag", Value: "other_value"}}, fmt.Sprintf("%v.foo.bar:42|g|#%v,my_tag:my_value,other_tag:other_value", prefix, globalTags)},
+		{"SetGauge", []string{"foo", "bar"}, float32(42), emptyTags, fmt.Sprintf("%v.foo.bar:42|g|#%v\n", prefix, globalTags)},
+		{"SetGauge", []string{"foo", "bar", "baz"}, float32(42), emptyTags, fmt.Sprintf("%v.foo.bar.baz:42|g|#%v\n", prefix, globalTags)},
+		{"AddSample", []string{"sample", "thing"}, float32(4), emptyTags, fmt.Sprintf("%v.sample.thing:4.000000|ms|#%v\n", prefix, globalTags)},
+		{"IncrCounter", []string{"count", "me"}, float32(3), emptyTags, fmt.Sprintf("%v.count.me:3|c|#%v\n", prefix, globalTags)},
+
+		{"SetGauge", []string{"foo", "baz"}, float32(42), []metrics.Label{{Name: tag, Value: ""}}, fmt.Sprintf("%v.foo.baz:42|g|#%v,my_tag\n", prefix, globalTags)},
+		{"SetGauge", []string{"foo", "baz"}, float32(42), []metrics.Label{{Name: tag, Value: tagValue}}, fmt.Sprintf("%v.foo.baz:42|g|#%v,my_tag:my_value\n", prefix, globalTags)},
+		{"SetGauge", []string{"foo", "bar"}, float32(42), []metrics.Label{{Name: tag, Value: tagValue}, {Name: "other_tag", Value: "other_value"}}, fmt.Sprintf("%v.foo.bar:42|g|#%v,my_tag:my_value,other_tag:other_value\n", prefix, globalTags)},
 	}
 
 	server, buf := setupTestServerAndBuffer(t)

pkg/hcp/telemetry/doc.go

@@ -0,0 +1,9 @@
+// Package telemetry is for scraping and sending Envoy metrics to HCP in OTLP format.
+//
+// This exists to let customers send mesh telemetry to HCP without having to deploy
+// a dedicated telemetry collector. Configuration for the collection, filtering, and exporting
+// of Envoy metrics is queried in the hcp.v2.TelemetryState resource in Consul.
+//
+// A lot of the package is concerned with converting prometheus-format metrics into their
+// OTLP equivalent.
+package telemetry

pkg/hcp/telemetry/exporter.go

@@ -0,0 +1,70 @@
+package telemetry
+
+import (
+	"context"
+	"time"
+
+	"github.com/hashicorp/consul/proto-public/pbresource"
+	"github.com/hashicorp/go-hclog"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+)
+
+const exporterDefaultScrapeInterval = time.Minute
+
+// Exporter is a telemetry exporter that is specific to HCP. If enabled via the hcp.v2.TelemetryState resource,
+// this exporter periodically scrapes Envoy metrics from its admin endpoint and pushes the metrics up to HCP.
+type Exporter struct {
+	envoyProxyID   string
+	logger         hclog.Logger
+	scrapeInterval time.Duration
+	scraper        scraper
+	stateTracker   stateTracker
+}
+
+// NewHCPExporter creates a new HCP telemetry exporter.
+func NewHCPExporter(resourceClient pbresource.ResourceServiceClient, logger hclog.Logger, envoyAdminHostPort, envoyProxyID string) *Exporter {
+	return &Exporter{
+		envoyProxyID:   envoyProxyID,
+		logger:         logger,
+		scrapeInterval: exporterDefaultScrapeInterval,
+		scraper:        newScraper(envoyAdminHostPort, logger.Named("scraper")),
+		stateTracker:   newStateTracker(resourceClient, logger.Named("state_tracker")),
+	}
+}
+
+// Run starts the exporter's exporter goroutine that periodically scrapes Envoy metrics and pushes them to HCP.
+func (w *Exporter) Run(ctx context.Context) {
+	go w.stateTracker.Run(ctx) // start syncing state from consul.
+	ticker := time.NewTicker(w.scrapeInterval)
+
+	for {
+		select {
+		case <-ticker.C:
+			state, ok := w.stateTracker.GetState()
+			if !ok || state.disabled {
+				w.logger.Debug("metric exporting disabled")
+				continue
+			}
+
+			labels := pcommon.NewMap()
+			for k, v := range state.labels {
+				labels.PutStr(k, v)
+			}
+			labels.PutStr("node.id", w.envoyProxyID)
+
+			metrics, err := w.scraper.scrape(ctx, state.includeList, labels)
+			if err != nil {
+				w.logger.Error("failed to scrape envoy stats", "error", err)
+				continue
+			}
+
+			if err := state.client.ExportMetrics(ctx, metrics); err != nil {
+				w.logger.Error("failed to export metrics", "error", err)
+			} else {
+				w.logger.Debug("exported metrics to hcp", "metrics_count", metrics.MetricCount())
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}