Support routing to Consul services imported from a peer (#406)

* Stub func for finding service imported from peer * Add peerName to MeshService CRD, fork service lookup when specified * Use "Peer" instead of "PeerName" for consistency * Add changelog entry * Return Consul resolution error when no matching service name from peer * Add unit test coverage for resolving of imported service from peer * Remove unnecessary mock generation target * go mod tidy * Consume latest tag of consul/api containing dependencies * Resolve dependency issues from merge * Add Peerings to Consul client interface * Adjust testing strategy to account for hot swap of Consul client
hashicorp · Nov 22, 2022 · 3fa4a6f · 3fa4a6f
1 parent ca999ea
commit 3fa4a6f
Show file tree

Hide file tree

Showing 15 changed files with 234 additions and 24 deletions.
diff --git a/.changelog/406.txt b/.changelog/406.txt
@@ -0,0 +1,3 @@
+```release-note:enhancement
+Allow MeshService CRD to reference a Consul service imported from a peer by specifying the peer's name
+```
diff --git a/config/crd/bases/api-gateway.consul.hashicorp.com_meshservices.yaml b/config/crd/bases/api-gateway.consul.hashicorp.com_meshservices.yaml
@@ -39,6 +39,11 @@ spec:
               name:
                 description: Name holds the service name for a Consul service.
                 type: string
+              peer:
+                description: Peer optionally specifies the name of the peer exporting
+                  the Consul service. If not specified, the Consul service is assumed
+                  to be in the local datacenter.
+                type: string
             type: object
         type: object
     served: true

diff --git a/internal/adapters/consul/sync_test.go b/internal/adapters/consul/sync_test.go
@@ -16,8 +16,8 @@ import (
 	"github.com/hashicorp/consul/sdk/testutil"
 
 	"github.com/hashicorp/consul-api-gateway/internal/common"
+	consultesting "github.com/hashicorp/consul-api-gateway/internal/consul"
 	"github.com/hashicorp/consul-api-gateway/internal/core"
-	consultesting "github.com/hashicorp/consul-api-gateway/internal/testing"
 )
 
 var (

diff --git a/internal/consul/certmanager_test.go b/internal/consul/certmanager_test.go
@@ -17,10 +17,10 @@ import (
 
 	"github.com/stretchr/testify/require"
 
-	consultesting "github.com/hashicorp/consul-api-gateway/internal/testing"
-	gwTesting "github.com/hashicorp/consul-api-gateway/internal/testing"
 	"github.com/hashicorp/consul/api"
 	"github.com/hashicorp/go-hclog"
+
+	gwTesting "github.com/hashicorp/consul-api-gateway/internal/testing"
 )
 
 func TestManage(t *testing.T) {
@@ -50,7 +50,7 @@ func TestManage(t *testing.T) {
 			options := DefaultCertManagerOptions()
 			options.Directory = directory
 
-			manager := NewCertManager(hclog.NewNullLogger(), consultesting.NewTestClient(server.consul), service, options)
+			manager := NewCertManager(hclog.NewNullLogger(), NewTestClient(server.consul), service, options)
 
 			manager.skipExtraFetch = true
 
@@ -110,7 +110,7 @@ func TestManage_Refresh(t *testing.T) {
 	server := runCertServer(t, 0, 0, service, 2)
 
 	options := DefaultCertManagerOptions()
-	manager := NewCertManager(hclog.NewNullLogger(), consultesting.NewTestClient(server.consul), service, options)
+	manager := NewCertManager(hclog.NewNullLogger(), NewTestClient(server.consul), service, options)
 
 	manager.skipExtraFetch = true
 

diff --git a/internal/consul/connection.go b/internal/consul/connection.go
@@ -4,12 +4,11 @@ import (
 	"context"
 	"crypto/tls"
 	"errors"
+	"fmt"
 	"strings"
 	"sync"
 	"time"
 
-	"fmt"
-
 	"github.com/hashicorp/consul-server-connection-manager/discovery"
 	"github.com/hashicorp/consul/api"
 	"github.com/hashicorp/go-hclog"
@@ -25,13 +24,18 @@ var (
 	globalWatcherMutex sync.Mutex
 )
 
+type PeeringClient interface {
+	Read(ctx context.Context, name string, q *api.QueryOptions) (*api.Peering, *api.QueryMeta, error)
+}
+
 type Client interface {
 	Agent() *api.Agent
 	ACL() *api.ACL
 	Catalog() *api.Catalog
 	ConfigEntries() *api.ConfigEntries
 	DiscoveryChain() *api.DiscoveryChain
 	Namespaces() *api.Namespaces
+	Peerings() PeeringClient
 
 	WatchServers(ctx context.Context) error
 
@@ -253,6 +257,13 @@ func (c *client) Namespaces() *api.Namespaces {
 	return c.client.Namespaces()
 }
 
+func (c *client) Peerings() PeeringClient {
+	c.mutex.RLock()
+	defer c.mutex.RUnlock()
+
+	return c.client.Peerings()
+}
+
 func (c *client) ACL() *api.ACL {
 	c.mutex.RLock()
 	defer c.mutex.RUnlock()

diff --git a/internal/consul/intentions_test.go b/internal/consul/intentions_test.go
@@ -9,10 +9,10 @@ import (
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
 
-	"github.com/hashicorp/consul-api-gateway/internal/consul/mocks"
-	consultesting "github.com/hashicorp/consul-api-gateway/internal/testing"
 	"github.com/hashicorp/consul/api"
 	"github.com/hashicorp/consul/sdk/testutil"
+
+	"github.com/hashicorp/consul-api-gateway/internal/consul/mocks"
 )
 
 type configEntryMatcher struct {
@@ -239,7 +239,7 @@ func TestIntentionsReconciler_Reconcile(t *testing.T) {
 			},
 		},
 	}
-	r := NewIntentionsReconciler(consultesting.NewTestClient(c), igw, testutil.Logger(t))
+	r := NewIntentionsReconciler(NewTestClient(c), igw, testutil.Logger(t))
 
 	require.NoError(r.Reconcile())
 

diff --git a/internal/consul/mocks/peerings.go b/internal/consul/mocks/peerings.go
diff --git a/internal/consul/peerings.go b/internal/consul/peerings.go
@@ -0,0 +1,13 @@
+package consul
+
+import (
+	"context"
+
+	"github.com/hashicorp/consul/api"
+)
+
+//go:generate mockgen -source ./peerings.go -destination ./mocks/peerings.go -package mocks Peerings
+
+type Peerings interface {
+	Read(context.Context, string, *api.QueryOptions) (*api.Peering, *api.QueryMeta, error)
+}
diff --git a/internal/consul/registration_test.go b/internal/consul/registration_test.go
@@ -13,10 +13,10 @@ import (
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/require"
 
-	consultesting "github.com/hashicorp/consul-api-gateway/internal/testing"
-	gwTesting "github.com/hashicorp/consul-api-gateway/internal/testing"
 	"github.com/hashicorp/consul/api"
 	"github.com/hashicorp/go-hclog"
+
+	gwTesting "github.com/hashicorp/consul-api-gateway/internal/testing"
 )
 
 func TestRegister(t *testing.T) {
@@ -55,7 +55,7 @@ func TestRegister(t *testing.T) {
 			}
 
 			server := runRegistryServer(t, test.failures, id)
-			registry := NewServiceRegistry(hclog.NewNullLogger(), consultesting.NewTestClient(server.consul), service, namespace, test.host).WithTries(maxAttempts)
+			registry := NewServiceRegistry(hclog.NewNullLogger(), NewTestClient(server.consul), service, namespace, test.host).WithTries(maxAttempts)
 
 			registry.backoffInterval = 0
 			registry.id = id
@@ -108,7 +108,7 @@ func TestDeregister(t *testing.T) {
 			}
 
 			server := runRegistryServer(t, test.failures, id)
-			registry := NewServiceRegistry(hclog.NewNullLogger(), consultesting.NewTestClient(server.consul), service, "", "").WithTries(maxAttempts)
+			registry := NewServiceRegistry(hclog.NewNullLogger(), NewTestClient(server.consul), service, "", "").WithTries(maxAttempts)
 			registry.backoffInterval = 0
 			registry.id = id
 			err := registry.Deregister(context.Background())

diff --git a/internal/testing/client.go → internal/consul/test_client.go b/internal/testing/client.go → internal/consul/test_client.go
@@ -1,14 +1,18 @@
-package testing
+package consul
 
 import (
 	"context"
 	"time"
 
 	"github.com/hashicorp/consul/api"
+
+	"github.com/hashicorp/consul-api-gateway/internal/consul/mocks"
 )
 
 type TestClient struct {
 	*api.Client
+
+	peerings *mocks.MockPeerings
 }
 
 func NewTestClient(c *api.Client) *TestClient {
@@ -32,3 +36,14 @@ func (c *TestClient) Wait(time.Duration) error {
 func (c *TestClient) Internal() *api.Client {
 	return c.Client
 }
+
+func (c *TestClient) Peerings() PeeringClient {
+	if c.peerings == nil {
+		return c.Client.Peerings()
+	}
+	return c.peerings
+}
+
+func (c *TestClient) SetPeerings(peerings *mocks.MockPeerings) {
+	c.peerings = peerings
+}
diff --git a/internal/k8s/service/resolver.go b/internal/k8s/service/resolver.go
@@ -13,6 +13,7 @@ import (
 	"github.com/hashicorp/go-hclog"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/utils/pointer"
 	gwv1alpha2 "sigs.k8s.io/gateway-api/apis/v1alpha2"
 
 	"github.com/hashicorp/consul-api-gateway/internal/common"
@@ -364,14 +365,26 @@ func (r *backendResolver) consulServiceForMeshService(ctx context.Context, names
 	// we do an inner retry since consul may take some time to sync
 	err = backoff.Retry(func() error {
 		r.logger.Trace("attempting to resolve global catalog service")
-		resolved, err = r.findCatalogService(service)
-		if err != nil {
-			r.logger.Trace("error resolving global catalog reference", "error", err)
-			return err
-		}
-		if resolved == nil {
-			return NewConsulResolutionError(fmt.Sprintf("consul service %s not found", namespacedName))
+
+		if pointer.StringDeref(service.Spec.Peer, "") != "" {
+			resolved, err = r.findPeerService(ctx, service)
+			if err != nil {
+				r.logger.Trace("error resolving imported service reference")
+				return err
+			} else if resolved == nil {
+				return NewConsulResolutionError(
+					fmt.Sprintf("imported consul service %s from peer %s not found", namespacedName, *service.Spec.Peer))
+			}
+		} else {
+			resolved, err = r.findCatalogService(service)
+			if err != nil {
+				r.logger.Trace("error resolving global catalog reference", "error", err)
+				return err
+			} else if resolved == nil {
+				return NewConsulResolutionError(fmt.Sprintf("consul service %s not found", namespacedName))
+			}
 		}
+
 		return nil
 	}, backoff.WithContext(backoff.WithMaxRetries(backoff.NewConstantBackOff(1*time.Second), 30), ctx))
 	if err != nil {
@@ -381,6 +394,37 @@ func (r *backendResolver) consulServiceForMeshService(ctx context.Context, names
 	return resolved, nil
 }
 
+func (r *backendResolver) findPeerService(ctx context.Context, service *apigwv1alpha1.MeshService) (*ResolvedReference, error) {
+	if pointer.StringDeref(service.Spec.Peer, "") == "" {
+		return nil, NewConsulResolutionError("peer name expected but not provided")
+	}
+
+	consulNamespace := r.mapper(service.Namespace)
+	consulName := service.Spec.Name
+	consulPeer := *service.Spec.Peer
+
+	peer, _, err := r.consul.Peerings().Read(ctx, consulPeer, &api.QueryOptions{Namespace: consulNamespace})
+	if err != nil {
+		r.logger.Trace("error resolving imported consul service reference", "error", err)
+		return nil, err
+	}
+
+	if peer == nil {
+		return nil, NewConsulResolutionError(fmt.Sprintf("no peer %q found", consulPeer))
+	}
+
+	for _, importedService := range peer.StreamStatus.ImportedServices {
+		if importedService == consulName {
+			return NewConsulServiceReference(&ConsulService{
+				Namespace: consulNamespace,
+				Name:      consulName,
+			}), nil
+		}
+	}
+
+	return nil, NewConsulResolutionError(fmt.Sprintf("no service %s found from peer %s", consulName, consulPeer))
+}
+
 func (r *backendResolver) findCatalogService(service *apigwv1alpha1.MeshService) (*ResolvedReference, error) {
 	consulNamespace := r.mapper(service.Namespace)
 	consulName := service.Spec.Name