tmp

agent/consul/server.go

@@ -8,6 +8,7 @@ import (
 	"crypto/x509"
 	"errors"
 	"fmt"
+	"github.com/hashicorp/consul/agent/hcp/bootstrap"
 	"io"
 	"net"
 	"os"
@@ -954,6 +955,8 @@ func NewServer(config *Config, flat Deps, externalGRPCServer *grpc.Server,
 			}
 			return nil
 		},
+		LoadManagementToken: bootstrap.LoadManagementToken,
+		DataDir:             flat.HCP.DataDir,
 	})
 
 	// Now we are setup, configure the HCP manager

agent/hcp/manager.go

@@ -40,8 +40,10 @@ type ManagerConfig struct {
 	// Idempotent function to upsert the HCP management token. This will be called periodically in
 	// the manager's main loop.
 	ManagementTokenUpserterFn ManagementTokenUpserter
+	LoadManagementToken       func(ctx context.Context, logger hclog.Logger, client hcpclient.Client, dataDir string) (string, error)
 	MinInterval               time.Duration
 	MaxInterval               time.Duration
+	DataDir                   string
 
 	Logger hclog.Logger
 }
@@ -108,24 +110,33 @@ func NewManager(cfg ManagerConfig) *HCPManager {
 // yet for servers since an HCP Link can be added at any time, at which point,
 // it will Start.
 func (m *HCPManager) Run(ctx context.Context) error {
-	for {
-		time.Sleep(10 * time.Second)
+	m.cfgMu.RLock()
+	cfg := m.cfg
+	m.cfgMu.RUnlock()
+
+	time.Sleep(10 * time.Second)
+
+	watchClient, err := cfg.ResourceServiceClient.WatchList(ctx, &pbresource.WatchListRequest{
+		Type:       pbhcp.LinkType,
+		NamePrefix: hcp.HCPLinkName,
+	})
+	if err != nil {
+		// TODO... what do we do
+		m.logger.Error("error watching resource service client, stopping HCP manager")
+		return err
+	}
 
+	for {
 		m.cfgMu.RLock()
 		cfg := m.cfg
 		m.cfgMu.RUnlock()
 
-		rsp, err := cfg.ResourceServiceClient.Read(ctx, &pbresource.ReadRequest{
-			Id: &pbresource.ID{
-				Name: hcp.HCPLinkName,
-				Type: pbhcp.LinkType,
-			},
-		})
+		// Block until any new events for Link object
+		watchEvent, err := watchClient.Recv()
 
 		if err != nil {
 			if status.Code(err) == codes.NotFound {
-				m.logger.Trace("link not found")
-				m.logger.Trace("stopping manager")
+				m.logger.Trace("link not found, stopping HCP manager")
 				m.Stop()
 				continue
 			} else {
@@ -134,40 +145,50 @@ func (m *HCPManager) Run(ctx context.Context) error {
 			}
 		}
 
-		if !m.isRunning() {
-			res := rsp.Resource
-			var link pbhcp.Link
-			if err := res.Data.UnmarshalTo(&link); err != nil {
-				m.logger.Error("error unmarshalling link data", "error", err)
-				continue
-			}
+		res := watchEvent.Resource
+		var link pbhcp.Link
+		if err := res.Data.UnmarshalTo(&link); err != nil {
+			m.logger.Error("error unmarshalling link data", "error", err)
+			continue
+		}
 
-			// Update the HCP manager configuration with the link values
-			// Merge the link data with the existing cloud config so that we only overwrite the
-			// fields that are provided by the link. This ensures that:
-			// 1. The HCP configuration (i.e., how to connect to HCP) is preserved
-			// 2. The Consul agent's node ID and node name are preserved
-			existingCfg := m.GetCloudConfig()
-			newCfg := config.CloudConfig{
-				ResourceID:   link.ResourceId,
-				ClientID:     link.ClientId,
-				ClientSecret: link.ClientSecret,
-			}
-			mergedCfg := config.Merge(existingCfg, newCfg)
-			hcpClient, err := cfg.HCPClientFn(mergedCfg)
+		// Update the HCP manager configuration with the link values
+		// Merge the link data with the existing cloud config so that we only overwrite the
+		// fields that are provided by the link. This ensures that:
+		// 1. The HCP configuration (i.e., how to connect to HCP) is preserved
+		// 2. The Consul agent's node ID and node name are preserved
+		existingCfg := m.GetCloudConfig()
+		newCfg := config.CloudConfig{
+			ResourceID:   link.ResourceId,
+			ClientID:     link.ClientId,
+			ClientSecret: link.ClientSecret,
+		}
+		mergedCfg := config.Merge(existingCfg, newCfg)
+		hcpClient, err := cfg.HCPClientFn(mergedCfg)
+		if err != nil {
+			m.logger.Error("error creating HCP client", "error", err)
+			continue
+		}
+
+		// Load the management token if access is not set to read-only. Read-only clusters
+		// will not have a management token provided by HCP.
+		var token string
+		if link.GetAccessLevel() != pbhcp.AccessLevel_ACCESS_LEVEL_UNSPECIFIED &&
+			link.GetAccessLevel() != pbhcp.AccessLevel_ACCESS_LEVEL_GLOBAL_READ_ONLY {
+			//token, err = bootstrap.LoadManagementToken(ctx, m.logger, hcpClient, cfg.DataDir)
+			token, err = cfg.LoadManagementToken(ctx, m.logger, hcpClient, cfg.DataDir)
 			if err != nil {
-				m.logger.Error("error creating HCP client", "error", err)
-				return err
+				m.logger.Error("error loading management token", "error", err)
+				continue
 			}
+		}
 
-			// TODO: Add token to HCP Link
-			// mergedCfg.ManagementToken = token
-			m.UpdateConfig(hcpClient, mergedCfg)
+		mergedCfg.ManagementToken = token
+		m.UpdateConfig(hcpClient, mergedCfg)
 
-			err = m.Start(ctx)
-			if err != nil {
-				m.logger.Error("error starting HCP manager", "error", err)
-			}
+		err = m.Start(ctx)
+		if err != nil {
+			m.logger.Error("error starting HCP manager", "error", err)
 		}
 	}
 }