[CC-7044] Start HCP manager as part of link creation

.changelog/20312.txt

@@ -0,0 +1,6 @@
+```release-note:feature
+cloud: Adds new API/CLI to initiate and manage link cluster to HCP Consul Central
+```
+```release-note:breaking-change
+telemetry: Adds fix to always default to prefixing gauge-type metrics with the hostname of the Consul agent, even if only the default metric sinks are enabled.
+```

agent/agent.go

@@ -1295,7 +1295,7 @@ func (a *Agent) listenHTTP() ([]apiServer, error) {
 	}
 
 	httpAddrs := a.config.HTTPAddrs
-	if a.config.IsCloudEnabled() {
+	if a.scadaProvider != nil {
 		httpAddrs = append(httpAddrs, scada.CAPCoreAPI)
 	}
 

agent/agent_test.go

@@ -37,7 +37,6 @@ import (
 	"github.com/hashicorp/serf/coordinate"
 	"github.com/hashicorp/serf/serf"
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/sync/errgroup"
 	"golang.org/x/time/rate"
@@ -6338,12 +6337,8 @@ func TestAgent_scadaProvider(t *testing.T) {
 	require.NoError(t, err)
 	defer require.NoError(t, l.Close())
 
-	pvd.EXPECT().UpdateMeta(mock.Anything).Once()
-	pvd.EXPECT().Start().Return(nil).Once()
 	pvd.EXPECT().Listen(scada.CAPCoreAPI.Capability()).Return(l, nil).Once()
 	pvd.EXPECT().Stop().Return(nil).Once()
-	pvd.EXPECT().SessionStatus().Return("test")
-	pvd.EXPECT().UpdateHCPConfig(mock.Anything).Return(nil).Once()
 	a := TestAgent{
 		OverrideDeps: func(deps *BaseDeps) {
 			deps.HCP.Provider = pvd

agent/config/builder.go

@@ -1112,8 +1112,8 @@ func (b *builder) build() (rt RuntimeConfig, err error) {
 		LocalProxyConfigResyncInterval:    30 * time.Second,
 	}
 
-	// host metrics are enabled by default if consul is configured with HashiCorp Cloud Platform integration
-	rt.Telemetry.EnableHostMetrics = boolValWithDefault(c.Telemetry.EnableHostMetrics, rt.IsCloudEnabled())
+	// host metrics are enabled by default to support HashiCorp Cloud Platform integration
+	rt.Telemetry.EnableHostMetrics = boolValWithDefault(c.Telemetry.EnableHostMetrics, true)
 
 	rt.TLS, err = b.buildTLSConfig(rt, c.TLS)
 	if err != nil {

agent/consul/server.go

@@ -595,14 +595,15 @@ func NewServer(config *Config, flat Deps, externalGRPCServer *grpc.Server,
 	})
 
 	s.hcpManager = hcp.NewManager(hcp.ManagerConfig{
-		CloudConfig:       s.config.Cloud,
-		Client:            flat.HCP.Client,
+		CloudConfig:       flat.HCP.Config,
 		StatusFn:          s.hcpServerStatus(flat),
 		Logger:            logger.Named("hcp_manager"),
 		SCADAProvider:     flat.HCP.Provider,
 		TelemetryProvider: flat.HCP.TelemetryProvider,
 		ManagementTokenUpserterFn: func(name, secretId string) error {
-			if s.IsLeader() {
+			// Check the state of the server before attempting to upsert the token. Otherwise,
+			// the upsert will fail and log errors that do not require action from the user.
+			if s.config.ACLsEnabled && s.IsLeader() && s.InPrimaryDatacenter() {
 				// Idea for improvement: Upsert a token with a well-known accessorId here instead
 				// of a randomly generated one. This would prevent any possible insertion collision between
 				// this and the insertion that happens during the ACL initialization process (initializeACLs function)
@@ -953,15 +954,6 @@ func NewServer(config *Config, flat Deps, externalGRPCServer *grpc.Server,
 	// Start the metrics handlers.
 	go s.updateMetrics()
 
-	// Now we are setup, configure the HCP manager
-	go func() {
-		err := s.hcpManager.Run(&lib.StopChannelContext{StopCh: shutdownCh})
-		if err != nil {
-			logger.Error("error starting HCP manager, some HashiCorp Cloud Platform functionality has been disabled",
-				"error", err)
-		}
-	}()
-
 	err = s.runEnterpriseRateLimiterConfigEntryController()
 	if err != nil {
 		return nil, err
@@ -995,6 +987,7 @@ func (s *Server) registerControllers(deps Deps, proxyUpdater ProxyUpdater) error
 		HCPAllowV2ResourceApis: s.hcpAllowV2Resources,
 		CloudConfig:            deps.HCP.Config,
 		DataDir:                deps.HCP.DataDir,
+		HCPManager:             s.hcpManager,
 	})
 
 	// When not enabled, the v1 tenancy bridge is used by default.

agent/consul/server_test.go

@@ -39,6 +39,7 @@ import (
 	external "github.com/hashicorp/consul/agent/grpc-external"
 	grpcmiddleware "github.com/hashicorp/consul/agent/grpc-middleware"
 	hcpclient "github.com/hashicorp/consul/agent/hcp/client"
+	hcpconfig "github.com/hashicorp/consul/agent/hcp/config"
 	"github.com/hashicorp/consul/agent/leafcert"
 	"github.com/hashicorp/consul/agent/metadata"
 	"github.com/hashicorp/consul/agent/rpc/middleware"
@@ -2097,6 +2098,8 @@ func TestServer_Peering_LeadershipCheck(t *testing.T) {
 
 func TestServer_hcpManager(t *testing.T) {
 	_, conf1 := testServerConfig(t)
+
+	// Configure the server for the StatusFn
 	conf1.BootstrapExpect = 1
 	conf1.RPCAdvertise = &net.TCPAddr{IP: []byte{127, 0, 0, 2}, Port: conf1.RPCAddr.Port}
 	hcp1 := hcpclient.NewMockClient(t)
@@ -2106,17 +2109,36 @@ func TestServer_hcpManager(t *testing.T) {
 		require.Equal(t, status.LanAddress, "127.0.0.2")
 	}).Call.Return(nil)
 
+	// Configure the server for the ManagementTokenUpserterFn
+	conf1.ACLsEnabled = true
+
 	deps1 := newDefaultDeps(t, conf1)
-	deps1.HCP.Client = hcp1
 	s1, err := newServerWithDeps(t, conf1, deps1)
 	if err != nil {
 		t.Fatalf("err: %v", err)
 	}
 	defer s1.Shutdown()
 	require.NotNil(t, s1.hcpManager)
 	waitForLeaderEstablishment(t, s1)
+
+	// Update the HCP manager and start it
+	token, err := uuid.GenerateUUID()
+	require.NoError(t, err)
+	s1.hcpManager.UpdateConfig(hcp1, hcpconfig.CloudConfig{
+		ManagementToken: token,
+	})
+	err = s1.hcpManager.Start(context.Background())
+	require.NoError(t, err)
+
+	// Validate that the server status pushed as expected
 	hcp1.AssertExpectations(t)
 
+	// Validate that the HCP token has been created as expected
+	retry.Run(t, func(r *retry.R) {
+		_, createdToken, err := s1.fsm.State().ACLTokenGetBySecret(nil, token, nil)
+		require.NoError(r, err)
+		require.NotNil(r, createdToken)
+	})
 }
 
 func TestServer_addServerTLSInfo(t *testing.T) {

agent/hcp/config/config.go

@@ -71,3 +71,50 @@ func (c *CloudConfig) HCPConfig(opts ...hcpcfg.HCPConfigOption) (hcpcfg.HCPConfi
 func (c *CloudConfig) IsConfigured() bool {
 	return c.ResourceID != "" && c.ClientID != "" && c.ClientSecret != ""
 }
+
+// Merge returns a cloud configuration that is the combined the values of
+// two configurations.
+func Merge(o CloudConfig, n CloudConfig) CloudConfig {
+	c := o
+	if n.ResourceID != "" {
+		c.ResourceID = n.ResourceID
+	}
+
+	if n.ClientID != "" {
+		c.ClientID = n.ClientID
+	}
+
+	if n.ClientSecret != "" {
+		c.ClientSecret = n.ClientSecret
+	}
+
+	if n.Hostname != "" {
+		c.Hostname = n.Hostname
+	}
+
+	if n.AuthURL != "" {
+		c.AuthURL = n.AuthURL
+	}
+
+	if n.ScadaAddress != "" {
+		c.ScadaAddress = n.ScadaAddress
+	}
+
+	if n.ManagementToken != "" {
+		c.ManagementToken = n.ManagementToken
+	}
+
+	if n.TLSConfig != nil {
+		c.TLSConfig = n.TLSConfig
+	}
+
+	if n.NodeID != "" {
+		c.NodeID = n.NodeID
+	}
+
+	if n.NodeName != "" {
+		c.NodeName = n.NodeName
+	}
+
+	return c
+}

agent/hcp/config/config_test.go

@@ -0,0 +1,82 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package config
+
+import (
+	"crypto/tls"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestMerge(t *testing.T) {
+	oldCfg := CloudConfig{
+		ResourceID:      "old-resource-id",
+		ClientID:        "old-client-id",
+		ClientSecret:    "old-client-secret",
+		Hostname:        "old-hostname",
+		AuthURL:         "old-auth-url",
+		ScadaAddress:    "old-scada-address",
+		ManagementToken: "old-token",
+		TLSConfig: &tls.Config{
+			ServerName: "old-server-name",
+		},
+		NodeID:   "old-node-id",
+		NodeName: "old-node-name",
+	}
+
+	newCfg := CloudConfig{
+		ResourceID:      "new-resource-id",
+		ClientID:        "new-client-id",
+		ClientSecret:    "new-client-secret",
+		Hostname:        "new-hostname",
+		AuthURL:         "new-auth-url",
+		ScadaAddress:    "new-scada-address",
+		ManagementToken: "new-token",
+		TLSConfig: &tls.Config{
+			ServerName: "new-server-name",
+		},
+		NodeID:   "new-node-id",
+		NodeName: "new-node-name",
+	}
+
+	for name, tc := range map[string]struct {
+		newCfg      CloudConfig
+		expectedCfg CloudConfig
+	}{
+		"Empty": {
+			newCfg:      CloudConfig{},
+			expectedCfg: oldCfg,
+		},
+		"All": {
+			newCfg:      newCfg,
+			expectedCfg: newCfg,
+		},
+		"Partial": {
+			newCfg: CloudConfig{
+				ResourceID:      newCfg.ResourceID,
+				ClientID:        newCfg.ClientID,
+				ClientSecret:    newCfg.ClientSecret,
+				ManagementToken: newCfg.ManagementToken,
+			},
+			expectedCfg: CloudConfig{
+				ResourceID:      newCfg.ResourceID,
+				ClientID:        newCfg.ClientID,
+				ClientSecret:    newCfg.ClientSecret,
+				ManagementToken: newCfg.ManagementToken,
+				Hostname:        oldCfg.Hostname,
+				AuthURL:         oldCfg.AuthURL,
+				ScadaAddress:    oldCfg.ScadaAddress,
+				TLSConfig:       oldCfg.TLSConfig,
+				NodeID:          oldCfg.NodeID,
+				NodeName:        oldCfg.NodeName,
+			},
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			merged := Merge(oldCfg, tc.newCfg)
+			require.Equal(t, tc.expectedCfg, merged)
+		})
+	}
+}

agent/hcp/deps.go

@@ -19,9 +19,8 @@ import (
 // Deps contains the interfaces that the rest of Consul core depends on for HCP integration.
 type Deps struct {
 	Config            config.CloudConfig
-	Client            client.Client
 	Provider          scada.Provider
-	Sink              metrics.MetricSink
+	Sink              metrics.ShutdownSink
 	TelemetryProvider *hcpProviderImpl
 	DataDir           string
 }
@@ -30,11 +29,6 @@ func NewDeps(cfg config.CloudConfig, logger hclog.Logger, dataDir string) (Deps,
 	ctx := context.Background()
 	ctx = hclog.WithContext(ctx, logger)
 
-	hcpClient, err := client.NewClient(cfg)
-	if err != nil {
-		return Deps{}, fmt.Errorf("failed to init client: %w", err)
-	}
-
 	provider, err := scada.New(logger.Named("scada"))
 	if err != nil {
 		return Deps{}, fmt.Errorf("failed to init scada: %w", err)
@@ -56,7 +50,6 @@ func NewDeps(cfg config.CloudConfig, logger hclog.Logger, dataDir string) (Deps,
 
 	return Deps{
 		Config:            cfg,
-		Client:            hcpClient,
 		Provider:          provider,
 		Sink:              sink,
 		TelemetryProvider: metricsProvider,
@@ -70,7 +63,7 @@ func sink(
 	ctx context.Context,
 	metricsClient telemetry.MetricsClient,
 	cfgProvider *hcpProviderImpl,
-) (metrics.MetricSink, error) {
+) (metrics.ShutdownSink, error) {
 	logger := hclog.FromContext(ctx)
 
 	reader := telemetry.NewOTELReader(metricsClient, cfgProvider)