Add MonitorHCPLink function

This function can be called in a goroutine to manage the lifecycle
of the HCP manager.

agent/hcp/link_monitor.go

@@ -0,0 +1,37 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package hcp
+
+import (
+	"context"
+
+	"github.com/hashicorp/go-hclog"
+
+	"github.com/hashicorp/consul/proto-public/pbresource"
+)
+
+// MonitorHCPLink monitors the status of the HCP Link and based on that, manages
+// the lifecycle of the HCP Manager. It's designed to be run in its own goroutine
+// for the life of a server agent. It should be run even if HCP is not configured
+// yet for servers. When an HCP Link is created, it will Start the Manager and
+// when an HCP Link is deleted, it will Stop the Manager.
+func MonitorHCPLink(ctx context.Context, logger hclog.Logger, hcpLinkEventCh chan *pbresource.WatchEvent, m Manager) {
+	for {
+		watchEvent := <-hcpLinkEventCh
+
+		if watchEvent.GetDelete() != nil {
+			logger.Debug("HCP Link deleted, stopping HCP manager")
+			err := m.Stop()
+			if err != nil {
+				logger.Error("error stopping HCP manager", "error", err)
+			}
+			continue
+		}
+
+		err := m.Start(ctx)
+		if err != nil {
+			logger.Error("error starting HCP manager", "error", err)
+		}
+	}
+}

agent/hcp/link_monitor_test.go

@@ -0,0 +1,33 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package hcp
+
+import (
+	"context"
+	"io"
+	"testing"
+
+	"github.com/stretchr/testify/mock"
+
+	"github.com/hashicorp/go-hclog"
+
+	"github.com/hashicorp/consul/proto-public/pbresource"
+)
+
+func TestMonitorHCPLink_Ok(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+
+	linkWatchCh := make(chan *pbresource.WatchEvent)
+	mgr := NewMockManager(t)
+	mgr.EXPECT().Start(mock.Anything).Return(nil).Once()
+	mgr.EXPECT().Stop().Return(nil).Once()
+
+	go MonitorHCPLink(ctx, hclog.New(&hclog.LoggerOptions{Output: io.Discard}), linkWatchCh, mgr)
+
+	linkWatchCh <- &pbresource.WatchEvent{Operation: pbresource.WatchEvent_OPERATION_UPSERT}
+	linkWatchCh <- &pbresource.WatchEvent{Operation: pbresource.WatchEvent_OPERATION_DELETE}
+
+	cancel()
+}

agent/hcp/manager.go

@@ -9,11 +9,12 @@ import (
 	"sync"
 	"time"
 
+	"github.com/hashicorp/go-hclog"
+
 	hcpclient "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/lib"
-	"github.com/hashicorp/go-hclog"
 )
 
 var (

agent/hcp/manager_test.go

@@ -9,16 +9,40 @@ import (
 	"testing"
 	"time"
 
-	hcpclient "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/sdk/testutil"
-	"github.com/hashicorp/go-hclog"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/net/context"
+
+	"github.com/hashicorp/go-hclog"
+
+	hcpclient "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/proto-public/pbresource"
+	"github.com/hashicorp/consul/sdk/testutil"
+	"github.com/hashicorp/consul/sdk/testutil/retry"
 )
 
+func TestManager_MonitorHCPLink(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+
+	linkWatchCh := make(chan *pbresource.WatchEvent)
+	mgr := NewManager(ManagerConfig{
+		Logger: hclog.New(&hclog.LoggerOptions{Output: io.Discard}),
+	})
+
+	require.False(t, mgr.isRunning())
+	go MonitorHCPLink(ctx, hclog.New(&hclog.LoggerOptions{Output: io.Discard}), linkWatchCh, mgr)
+
+	linkWatchCh <- &pbresource.WatchEvent{Operation: pbresource.WatchEvent_OPERATION_UPSERT}
+
+	// Validate that the HCP manager is started
+	retry.Run(t, func(r *retry.R) {
+		require.True(r, mgr.isRunning())
+	})
+}
+
 func TestManager_Start(t *testing.T) {
 	client := hcpclient.NewMockClient(t)
 	statusF := func(ctx context.Context) (hcpclient.ServerStatus, error) {

internal/hcp/watch_test.go

@@ -5,13 +5,14 @@ package hcp
 
 import (
 	"context"
-	"github.com/hashicorp/go-hclog"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 
+	"github.com/hashicorp/go-hclog"
+
 	svctest "github.com/hashicorp/consul/agent/grpc-external/services/resource/testing"
 	"github.com/hashicorp/consul/internal/controller"
 	"github.com/hashicorp/consul/internal/hcp/internal/types"
@@ -87,6 +88,7 @@ func (suite *controllerSuite) TestLinkWatch_Ok() {
 	}
 
 	_, err = suite.client.Delete(suite.ctx, &pbresource.DeleteRequest{Id: link.Id})
+	require.NoError(suite.T(), err)
 
 	select {
 	case watchEvent := <-linkWatchCh: