[HCP Telemetry] Move first TelemetryConfig Fetch into the TelemetryConfigProvider (hashicorp#18318)

* Add Enabler interface to turn sink on/off

* Use h for hcpProviderImpl vars, fix PR feeback and fix errors

* Keep nil check in exporter and fix tests

* Clarify comment and fix function name

* Use disable instead of enable

* Fix errors nit in otlp_transform

* Add test for refreshInterval of updateConfig

* Add disabled field in MetricsConfig struct

* Fix PR feedback: improve comment and remove double colons

* Fix deps test which requires a maybe

* Update hcp-sdk-go to v0.61.0

* use disabled flag in telemetry_config.go

* Handle 4XX errors in telemetry_provider

* Fix deps test

* Check 4XX instead

* Run make go-mod-tidy

Author: Achooo

agent/hcp/client/telemetry_config.go

@@ -20,7 +20,7 @@ import (
 
 var (
 	// defaultMetricFilters is a regex that matches all metric names.
-	defaultMetricFilters = regexp.MustCompile(".+")
+	DefaultMetricFilters = regexp.MustCompile(".+")
 
 	// Validation errors for AgentTelemetryConfigOK response.
 	errMissingPayload         = errors.New("missing payload")
@@ -29,6 +29,7 @@ var (
 	errMissingMetricsConfig   = errors.New("missing metrics config")
 	errInvalidRefreshInterval = errors.New("invalid refresh interval")
 	errInvalidEndpoint        = errors.New("invalid metrics endpoint")
+	errEmptyEndpoint          = errors.New("empty metrics endpoint")
 )
 
 // TelemetryConfig contains configuration for telemetry data forwarded by Consul servers
@@ -43,18 +44,14 @@ type MetricsConfig struct {
 	Labels   map[string]string
 	Filters  *regexp.Regexp
 	Endpoint *url.URL
+	Disabled bool
 }
 
 // RefreshConfig contains configuration for the periodic fetch of configuration from HCP.
 type RefreshConfig struct {
 	RefreshInterval time.Duration
 }
 
-// MetricsEnabled returns true if metrics export is enabled, i.e. a valid metrics endpoint exists.
-func (t *TelemetryConfig) MetricsEnabled() bool {
-	return t.MetricsConfig.Endpoint != nil
-}
-
 // validateAgentTelemetryConfigPayload ensures the returned payload from HCP is valid.
 func validateAgentTelemetryConfigPayload(resp *hcptelemetry.AgentTelemetryConfigOK) error {
 	if resp.Payload == nil {
@@ -86,7 +83,7 @@ func convertAgentTelemetryResponse(ctx context.Context, resp *hcptelemetry.Agent
 	telemetryConfig := resp.Payload.TelemetryConfig
 	metricsEndpoint, err := convertMetricEndpoint(telemetryConfig.Endpoint, telemetryConfig.Metrics.Endpoint)
 	if err != nil {
-		return nil, errInvalidEndpoint
+		return nil, err
 	}
 
 	metricsFilters := convertMetricFilters(ctx, telemetryConfig.Metrics.IncludeList)
@@ -97,6 +94,7 @@ func convertAgentTelemetryResponse(ctx context.Context, resp *hcptelemetry.Agent
 			Endpoint: metricsEndpoint,
 			Labels:   metricLabels,
 			Filters:  metricsFilters,
+			Disabled: telemetryConfig.Metrics.Disabled,
 		},
 		RefreshConfig: &RefreshConfig{
 			RefreshInterval: refreshInterval,
@@ -114,9 +112,8 @@ func convertMetricEndpoint(telemetryEndpoint string, metricsEndpoint string) (*u
 		endpoint = metricsEndpoint
 	}
 
-	// If endpoint is empty, server not registered with CCM, no error returned.
 	if endpoint == "" {
-		return nil, nil
+		return nil, errEmptyEndpoint
 	}
 
 	// Endpoint from CTW has no metrics path, so it must be added.
@@ -145,15 +142,15 @@ func convertMetricFilters(ctx context.Context, payloadFilters []string) *regexp.
 
 	if len(validFilters) == 0 {
 		logger.Error("no valid filters")
-		return defaultMetricFilters
+		return DefaultMetricFilters
 	}
 
 	// Combine the valid regex strings with OR.
 	finalRegex := strings.Join(validFilters, "|")
 	composedRegex, err := regexp.Compile(finalRegex)
 	if err != nil {
 		logger.Error("failed to compile final regex", "error", err)
-		return defaultMetricFilters
+		return DefaultMetricFilters
 	}
 
 	return composedRegex

agent/hcp/client/telemetry_config_test.go

@@ -88,7 +88,6 @@ func TestConvertAgentTelemetryResponse(t *testing.T) {
 		resp                 *consul_telemetry_service.AgentTelemetryConfigOK
 		expectedTelemetryCfg *TelemetryConfig
 		wantErr              error
-		expectedEnabled      bool
 	}{
 		"success": {
 			resp: &consul_telemetry_service.AgentTelemetryConfigOK{
@@ -115,34 +114,6 @@ func TestConvertAgentTelemetryResponse(t *testing.T) {
 					RefreshInterval: 2 * time.Second,
 				},
 			},
-			expectedEnabled: true,
-		},
-		"successNoEndpoint": {
-			resp: &consul_telemetry_service.AgentTelemetryConfigOK{
-				Payload: &models.HashicorpCloudConsulTelemetry20230414AgentTelemetryConfigResponse{
-					TelemetryConfig: &models.HashicorpCloudConsulTelemetry20230414TelemetryConfig{
-						Endpoint: "",
-						Labels:   map[string]string{"test": "test"},
-						Metrics: &models.HashicorpCloudConsulTelemetry20230414TelemetryMetricsConfig{
-							IncludeList: []string{"test", "consul"},
-						},
-					},
-					RefreshConfig: &models.HashicorpCloudConsulTelemetry20230414RefreshConfig{
-						RefreshInterval: "2s",
-					},
-				},
-			},
-			expectedTelemetryCfg: &TelemetryConfig{
-				MetricsConfig: &MetricsConfig{
-					Endpoint: nil,
-					Labels:   map[string]string{"test": "test"},
-					Filters:  validTestFilters,
-				},
-				RefreshConfig: &RefreshConfig{
-					RefreshInterval: 2 * time.Second,
-				},
-			},
-			expectedEnabled: false,
 		},
 		"successBadFilters": {
 			resp: &consul_telemetry_service.AgentTelemetryConfigOK{
@@ -163,13 +134,12 @@ func TestConvertAgentTelemetryResponse(t *testing.T) {
 				MetricsConfig: &MetricsConfig{
 					Endpoint: validTestURL,
 					Labels:   map[string]string{"test": "test"},
-					Filters:  defaultMetricFilters,
+					Filters:  DefaultMetricFilters,
 				},
 				RefreshConfig: &RefreshConfig{
 					RefreshInterval: 2 * time.Second,
 				},
 			},
-			expectedEnabled: true,
 		},
 		"errorsWithInvalidRefreshInterval": {
 			resp: &consul_telemetry_service.AgentTelemetryConfigOK{
@@ -209,7 +179,6 @@ func TestConvertAgentTelemetryResponse(t *testing.T) {
 			}
 			require.NoError(t, err)
 			require.Equal(t, tc.expectedTelemetryCfg, telemetryCfg)
-			require.Equal(t, tc.expectedEnabled, telemetryCfg.MetricsEnabled())
 		})
 	}
 }
@@ -231,10 +200,10 @@ func TestConvertMetricEndpoint(t *testing.T) {
 			override: "https://override.com",
 			expected: "https://override.com/v1/metrics",
 		},
-		"noErrorWithEmptyEndpoints": {
+		"errorWithEmptyEndpoints": {
 			endpoint: "",
 			override: "",
-			expected: "",
+			wantErr:  errEmptyEndpoint,
 		},
 		"errorWithInvalidURL": {
 			endpoint: "     ",
@@ -252,12 +221,6 @@ func TestConvertMetricEndpoint(t *testing.T) {
 				return
 			}
 
-			if tc.expected == "" {
-				require.Nil(t, u)
-				require.NoError(t, err)
-				return
-			}
-
 			require.NotNil(t, u)
 			require.NoError(t, err)
 			require.Equal(t, tc.expected, u.String())
@@ -277,13 +240,13 @@ func TestConvertMetricFilters(t *testing.T) {
 	}{
 		"badFilterRegex": {
 			filters:             []string{"(*LF)"},
-			expectedRegexString: defaultMetricFilters.String(),
+			expectedRegexString: DefaultMetricFilters.String(),
 			matches:             []string{"consul.raft.peers", "consul.mem.heap_size"},
 			wantMatch:           true,
 		},
 		"emptyRegex": {
 			filters:             []string{},
-			expectedRegexString: defaultMetricFilters.String(),
+			expectedRegexString: DefaultMetricFilters.String(),
 			matches:             []string{"consul.raft.peers", "consul.mem.heap_size"},
 			wantMatch:           true,
 		},

agent/hcp/deps.go

@@ -6,19 +6,19 @@ package hcp
 import (
 	"context"
 	"fmt"
-	"time"
 
 	"github.com/armon/go-metrics"
-	hcpclient "github.com/hashicorp/consul/agent/hcp/client"
+	"github.com/hashicorp/go-hclog"
+
+	"github.com/hashicorp/consul/agent/hcp/client"
 	"github.com/hashicorp/consul/agent/hcp/config"
 	"github.com/hashicorp/consul/agent/hcp/scada"
 	"github.com/hashicorp/consul/agent/hcp/telemetry"
-	"github.com/hashicorp/go-hclog"
 )
 
 // Deps contains the interfaces that the rest of Consul core depends on for HCP integration.
 type Deps struct {
-	Client   hcpclient.Client
+	Client   client.Client
 	Provider scada.Provider
 	Sink     metrics.MetricSink
 }
@@ -27,7 +27,7 @@ func NewDeps(cfg config.CloudConfig, logger hclog.Logger) (Deps, error) {
 	ctx := context.Background()
 	ctx = hclog.WithContext(ctx, logger)
 
-	client, err := hcpclient.NewClient(cfg)
+	hcpClient, err := client.NewClient(cfg)
 	if err != nil {
 		return Deps{}, fmt.Errorf("failed to init client: %w", err)
 	}
@@ -37,50 +37,33 @@ func NewDeps(cfg config.CloudConfig, logger hclog.Logger) (Deps, error) {
 		return Deps{}, fmt.Errorf("failed to init scada: %w", err)
 	}
 
-	metricsClient, err := hcpclient.NewMetricsClient(ctx, &cfg)
+	metricsClient, err := client.NewMetricsClient(ctx, &cfg)
 	if err != nil {
 		logger.Error("failed to init metrics client", "error", err)
 		return Deps{}, fmt.Errorf("failed to init metrics client: %w", err)
 	}
 
-	sink, err := sink(ctx, client, metricsClient)
+	sink, err := sink(ctx, metricsClient, NewHCPProvider(ctx, hcpClient))
 	if err != nil {
 		// Do not prevent server start if sink init fails, only log error.
 		logger.Error("failed to init sink", "error", err)
 	}
 
 	return Deps{
-		Client:   client,
+		Client:   hcpClient,
 		Provider: provider,
 		Sink:     sink,
 	}, nil
 }
 
 // sink initializes an OTELSink which forwards Consul metrics to HCP.
-// The sink is only initialized if the server is registered with the management plane (CCM).
 // This step should not block server initialization, errors are returned, only to be logged.
 func sink(
 	ctx context.Context,
-	hcpClient hcpclient.Client,
 	metricsClient telemetry.MetricsClient,
+	cfgProvider *hcpProviderImpl,
 ) (metrics.MetricSink, error) {
-	logger := hclog.FromContext(ctx).Named("sink")
-	reqCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
-	defer cancel()
-
-	telemetryCfg, err := hcpClient.FetchTelemetryConfig(reqCtx)
-	if err != nil {
-		return nil, fmt.Errorf("failed to fetch telemetry config: %w", err)
-	}
-
-	if !telemetryCfg.MetricsEnabled() {
-		return nil, nil
-	}
-
-	cfgProvider, err := NewHCPProvider(ctx, hcpClient, telemetryCfg)
-	if err != nil {
-		return nil, fmt.Errorf("failed to init config provider: %w", err)
-	}
+	logger := hclog.FromContext(ctx)
 
 	reader := telemetry.NewOTELReader(metricsClient, cfgProvider)
 	sinkOpts := &telemetry.OTELSinkOpts{
@@ -90,7 +73,7 @@ func sink(
 
 	sink, err := telemetry.NewOTELSink(ctx, sinkOpts)
 	if err != nil {
-		return nil, fmt.Errorf("failed create OTELSink: %w", err)
+		return nil, fmt.Errorf("failed to create OTELSink: %w", err)
 	}
 
 	logger.Debug("initialized HCP metrics sink")

agent/hcp/deps_test.go

@@ -5,16 +5,10 @@ package hcp
 
 import (
 	"context"
-	"fmt"
-	"net/url"
-	"regexp"
 	"testing"
-	"time"
 
-	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
-	"github.com/hashicorp/consul/agent/hcp/client"
 	"github.com/hashicorp/consul/agent/hcp/telemetry"
 )
 
@@ -24,79 +18,11 @@ type mockMetricsClient struct {
 
 func TestSink(t *testing.T) {
 	t.Parallel()
-	for name, test := range map[string]struct {
-		expect       func(*client.MockClient)
-		wantErr      string
-		expectedSink bool
-	}{
-		"success": {
-			expect: func(mockClient *client.MockClient) {
-				u, _ := url.Parse("https://test.com/v1/metrics")
-				filters, _ := regexp.Compile("test")
-				mt := mockTelemetryConfig(1*time.Second, u, filters)
-				mockClient.EXPECT().FetchTelemetryConfig(mock.Anything).Return(mt, nil)
-			},
-			expectedSink: true,
-		},
-		"noSinkWhenFetchTelemetryConfigFails": {
-			expect: func(mockClient *client.MockClient) {
-				mockClient.EXPECT().FetchTelemetryConfig(mock.Anything).Return(nil, fmt.Errorf("fetch failed"))
-			},
-			wantErr: "failed to fetch telemetry config",
-		},
-		"noSinkWhenServerNotRegisteredWithCCM": {
-			expect: func(mockClient *client.MockClient) {
-				mt := mockTelemetryConfig(1*time.Second, nil, nil)
-				mockClient.EXPECT().FetchTelemetryConfig(mock.Anything).Return(mt, nil)
-			},
-		},
-		"noSinkWhenTelemetryConfigProviderInitFails": {
-			expect: func(mockClient *client.MockClient) {
-				u, _ := url.Parse("https://test.com/v1/metrics")
-				// Bad refresh interval forces ConfigProvider creation failure.
-				mt := mockTelemetryConfig(0*time.Second, u, nil)
-				mockClient.EXPECT().FetchTelemetryConfig(mock.Anything).Return(mt, nil)
-			},
-			wantErr: "failed to init config provider",
-		},
-	} {
-		test := test
-		t.Run(name, func(t *testing.T) {
-			t.Parallel()
-			c := client.NewMockClient(t)
-			mc := mockMetricsClient{}
 
-			test.expect(c)
-			ctx := context.Background()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s, err := sink(ctx, mockMetricsClient{}, &hcpProviderImpl{})
 
-			s, err := sink(ctx, c, mc)
-
-			if test.wantErr != "" {
-				require.NotNil(t, err)
-				require.Contains(t, err.Error(), test.wantErr)
-				require.Nil(t, s)
-				return
-			}
-
-			if !test.expectedSink {
-				require.Nil(t, s)
-				require.Nil(t, err)
-				return
-			}
-
-			require.NotNil(t, s)
-		})
-	}
-}
-
-func mockTelemetryConfig(refreshInterval time.Duration, metricsEndpoint *url.URL, filters *regexp.Regexp) *client.TelemetryConfig {
-	return &client.TelemetryConfig{
-		MetricsConfig: &client.MetricsConfig{
-			Endpoint: metricsEndpoint,
-			Filters:  filters,
-		},
-		RefreshConfig: &client.RefreshConfig{
-			RefreshInterval: refreshInterval,
-		},
-	}
+	require.NotNil(t, s)
+	require.NoError(t, err)
 }