Code review updates

CHANGELOG.md

@@ -1,5 +1,11 @@
 ## UNRELEASED
 
+IMPROVEMENTS:
+
+* Add an ability to configure the synthetic Consul node name where catalog sync registers services. [[GH-312](https://github.com/hashicorp/consul-k8s/pull/312)]
+  * Sync: Add `-consul-node-name` flag to the `sync-catalog` command to configure the Consul node name for syncing services to Consul.
+  * ACLs: Add `-sync-consul-node-name` flag to the server-acl-init command so that it can create correct policy for the sync catalog.
+
 ## 0.18.1 (August 10, 2020)
 
 BUG FIXES:

catalog/to-consul/resource.go

@@ -115,6 +115,9 @@ type ServiceResource struct {
 	// `k8s-default` namespace.
 	K8SNSMirroringPrefix string
 
+	// The Consul node name to register service with.
+	ConsulNodeName string
+
 	// serviceLock must be held for any read/write to these maps.
 	serviceLock sync.RWMutex
 
@@ -332,7 +335,7 @@ func (t *ServiceResource) generateRegistrations(key string) {
 	// shallow copied for each instance.
 	baseNode := consulapi.CatalogRegistration{
 		SkipNodeUpdate: true,
-		Node:           t.Syncer.ConsulNode(),
+		Node:           t.ConsulNodeName,
 		Address:        "127.0.0.1",
 		NodeMeta: map[string]string{
 			ConsulSourceKey: ConsulSourceValue,

catalog/to-consul/resource_test.go

@@ -252,6 +252,34 @@ func TestServiceResource_addK8SNamespaceWithPrefix(t *testing.T) {
 	})
 }
 
+// Test that when consul node name is set to a non-default value,
+// services are synced to that node.
+func TestServiceResource_ConsulNodeName(t *testing.T) {
+	t.Parallel()
+	client := fake.NewSimpleClientset()
+	syncer := newTestSyncer()
+	serviceResource := defaultServiceResource(client, syncer)
+	serviceResource.ConsulNodeName = "test-node"
+
+	// Start the controller
+	closer := controller.TestControllerRun(&serviceResource)
+	defer closer()
+
+	// Insert an LB service with the sync=true
+	svc := lbService("foo", "namespace", "1.2.3.4")
+	_, err := client.CoreV1().Services("namespace").Create(context.Background(), svc, metav1.CreateOptions{})
+	require.NoError(t, err)
+
+	// Verify that the service name has k8s namespace appended with an '-'
+	retry.Run(t, func(r *retry.R) {
+		syncer.Lock()
+		defer syncer.Unlock()
+		actual := syncer.Registrations
+		require.Len(r, actual, 1)
+		require.Equal(r, actual[0].Node, "test-node")
+	})
+}
+
 // Test k8s namespace suffix is not appended
 // when the service name annotation is provided
 func TestServiceResource_addK8SNamespaceWithNameAnnotation(t *testing.T) {
@@ -1569,5 +1597,6 @@ func defaultServiceResource(client kubernetes.Interface, syncer Syncer) ServiceR
 		Syncer:                syncer,
 		AllowK8sNamespacesSet: mapset.NewSet("*"),
 		DenyK8sNamespacesSet:  mapset.NewSet(),
+		ConsulNodeName:        ConsulSyncNodeName,
 	}
 }

catalog/to-consul/syncer.go

@@ -28,10 +28,6 @@ const (
 type Syncer interface {
 	// Sync is called to sync the full set of registrations.
 	Sync([]*api.CatalogRegistration)
-
-	// ConsulNode returns the name of the Consul node where services
-	// will be synced.
-	ConsulNode() string
 }
 
 // ConsulSyncer is a Syncer that takes the set of registrations and
@@ -69,7 +65,7 @@ type ConsulSyncer struct {
 	// ConsulK8STag is the tag value for services registered.
 	ConsulK8STag string
 
-	// The Consul node name to register for this syncer.
+	// The Consul node name to register services with.
 	ConsulNodeName string
 
 	// ConsulNodeServicesClient is used to list services for a node. We use a
@@ -101,10 +97,6 @@ type ConsulSyncer struct {
 	watchers map[string]map[string]context.CancelFunc
 }
 
-func (s *ConsulSyncer) ConsulNode() string {
-	return s.ConsulNodeName
-}
-
 // Sync implements Syncer
 func (s *ConsulSyncer) Sync(rs []*api.CatalogRegistration) {
 	// Grab the lock so we can replace the sync state
@@ -196,7 +188,7 @@ func (s *ConsulSyncer) watchReapableServices(ctx context.Context) {
 		var meta *api.QueryMeta
 		err := backoff.Retry(func() error {
 			var err error
-			services, meta, err = s.ConsulNodeServicesClient.NodeServices(s.ConsulK8STag, s.ConsulNode(), *opts)
+			services, meta, err = s.ConsulNodeServicesClient.NodeServices(s.ConsulK8STag, s.ConsulNodeName, *opts)
 			return err
 		}, backoff.WithContext(backoff.NewExponentialBackOff(), ctx))
 

catalog/to-consul/syncer_test.go

@@ -2,6 +2,7 @@ package catalog
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"testing"
@@ -64,61 +65,67 @@ func TestConsulSyncer_register(t *testing.T) {
 // Test that the syncer reaps individual invalid service instances.
 func TestConsulSyncer_reapServiceInstance(t *testing.T) {
 	t.Parallel()
-	require := require.New(t)
 
-	// Set up server, client, syncer
-	a, err := testutil.NewTestServerConfigT(t, nil)
-	require.NoError(err)
-	defer a.Stop()
+	for _, node := range []string{ConsulSyncNodeName, "test-node"} {
+		name := fmt.Sprintf("consul node name: %s", node)
+		t.Run(name, func(t *testing.T) {
+			require := require.New(t)
 
-	client, err := api.NewClient(&api.Config{
-		Address: a.HTTPAddr,
-	})
-	require.NoError(err)
+			// Set up server, client, syncer
+			a, err := testutil.NewTestServerConfigT(t, nil)
+			require.NoError(err)
+			defer a.Stop()
 
-	s, closer := testConsulSyncer(client)
-	defer closer()
+			client, err := api.NewClient(&api.Config{
+				Address: a.HTTPAddr,
+			})
+			require.NoError(err)
 
-	// Sync
-	s.Sync([]*api.CatalogRegistration{
-		testRegistration(ConsulSyncNodeName, "bar", "default"),
-	})
+			s, closer := testConsulSyncer(client)
+			defer closer()
 
-	// Wait for the first service
-	retry.Run(t, func(r *retry.R) {
-		services, _, err := client.Catalog().Service("bar", "", nil)
-		if err != nil {
-			r.Fatalf("err: %s", err)
-		}
-		if len(services) != 1 {
-			r.Fatal("service not found or too many")
-		}
-	})
+			// Sync
+			s.Sync([]*api.CatalogRegistration{
+				testRegistration(node, "bar", "default"),
+			})
 
-	// Create an invalid service directly in Consul
-	svc := testRegistration(ConsulSyncNodeName, "bar", "default")
-	svc.Service.ID = serviceID("k8s-sync", "bar2")
-	_, err = client.Catalog().Register(svc, nil)
-	require.NoError(err)
+			// Wait for the first service
+			retry.Run(t, func(r *retry.R) {
+				services, _, err := client.Catalog().Service("bar", "", nil)
+				if err != nil {
+					r.Fatalf("err: %s", err)
+				}
+				if len(services) != 1 {
+					r.Fatal("service not found or too many")
+				}
+			})
 
-	// Valid service should exist
-	var service *api.CatalogService
-	retry.Run(t, func(r *retry.R) {
-		services, _, err := client.Catalog().Service("bar", "", nil)
-		if err != nil {
-			r.Fatalf("err: %s", err)
-		}
-		if len(services) != 1 {
-			r.Fatal("service not found or too many")
-		}
-		service = services[0]
-	})
+			// Create an invalid service directly in Consul
+			svc := testRegistration(node, "bar", "default")
+			svc.Service.ID = serviceID(node, "bar2")
+			_, err = client.Catalog().Register(svc, nil)
+			require.NoError(err)
+
+			// Valid service should exist
+			var service *api.CatalogService
+			retry.Run(t, func(r *retry.R) {
+				services, _, err := client.Catalog().Service("bar", "", nil)
+				if err != nil {
+					r.Fatalf("err: %s", err)
+				}
+				if len(services) != 1 {
+					r.Fatal("service not found or too many")
+				}
+				service = services[0]
+			})
 
-	// Verify the settings
-	require.Equal(serviceID("k8s-sync", "bar"), service.ServiceID)
-	require.Equal("k8s-sync", service.Node)
-	require.Equal("bar", service.ServiceName)
-	require.Equal("127.0.0.1", service.Address)
+			// Verify the settings
+			require.Equal(serviceID(node, "bar"), service.ServiceID)
+			require.Equal(node, service.Node)
+			require.Equal("bar", service.ServiceName)
+			require.Equal("127.0.0.1", service.Address)
+		})
+	}
 }
 
 // Test that the syncer reaps services not registered by us that are tagged

catalog/to-consul/testing.go

@@ -13,13 +13,8 @@ const (
 // testSyncer implements Syncer for tests, giving easy access to the
 // set of registrations.
 type testSyncer struct {
-	sync.Mutex     // Lock should be held while accessing Registrations
-	Registrations  []*api.CatalogRegistration
-	ConsulNodeName string // Consul node name to register for this syncer
-}
-
-func (s *testSyncer) ConsulNode() string {
-	return s.ConsulNodeName
+	sync.Mutex    // Lock should be held while accessing Registrations
+	Registrations []*api.CatalogRegistration
 }
 
 // Sync implements Syncer
@@ -30,5 +25,5 @@ func (s *testSyncer) Sync(rs []*api.CatalogRegistration) {
 }
 
 func newTestSyncer() *testSyncer {
-	return &testSyncer{ConsulNodeName: "k8s-sync"}
+	return &testSyncer{}
 }

subcommand/server-acl-init/command.go

@@ -767,20 +767,20 @@ func (c *Command) validateFlags() error {
 	// For the Consul node name to be discoverable via DNS, it must contain only
 	// dashes and alphanumeric characters. Length is also constrained.
 	// These restrictions match those defined in Consul's agent definition.
-	var InvalidDnsRe = regexp.MustCompile(`[^A-Za-z0-9\\-]+`)
-	const MaxDNSLabelLength = 63
+	var invalidDnsRe = regexp.MustCompile(`[^A-Za-z0-9\\-]+`)
+	const maxDNSLabelLength = 63
 
-	if InvalidDnsRe.MatchString(c.flagSyncConsulNodeName) {
-		return fmt.Errorf("Node name will not be discoverable "+
+	if invalidDnsRe.MatchString(c.flagSyncConsulNodeName) {
+		return fmt.Errorf("-sync-consul-node-name=%s is invalid: node name will not be discoverable "+
 			"via DNS due to invalid characters. Valid characters include "+
-			"all alpha-numerics and dashes. sync-consul-node-name=%s",
+			"all alpha-numerics and dashes",
 			c.flagSyncConsulNodeName,
 		)
 	}
-	if len(c.flagSyncConsulNodeName) > MaxDNSLabelLength {
-		return fmt.Errorf("Node name will not be discoverable "+
+	if len(c.flagSyncConsulNodeName) > maxDNSLabelLength {
+		return fmt.Errorf("-sync-consul-node-name=%s is invalid: node name will not be discoverable "+
 			"via DNS due to it being too long. Valid lengths are between "+
-			"1 and 63 bytes. sync-consul-node-name=%s",
+			"1 and 63 bytes",
 			c.flagSyncConsulNodeName,
 		)
 	}

subcommand/server-acl-init/command_test.go

@@ -62,17 +62,17 @@ func TestRun_FlagValidation(t *testing.T) {
 				"-resource-prefix=prefix",
 				"-sync-consul-node-name=Speci@l_Chars",
 			},
-			ExpErr: "Node name will not be discoverable via DNS due to invalid characters. Valid characters include " +
-				"all alpha-numerics and dashes. sync-consul-node-name=Speci@l_Chars",
+			ExpErr: "-sync-consul-node-name=Speci@l_Chars is invalid: node name will not be discoverable "+
+				"via DNS due to invalid characters. Valid characters include all alpha-numerics and dashes",
 		},
 		{
 			Flags: []string{
 				"-server-address=localhost",
 				"-resource-prefix=prefix",
 				"-sync-consul-node-name=5r9OPGfSRXUdGzNjBdAwmhCBrzHDNYs4XjZVR4wp7lSLIzqwS0ta51nBLIN0TMPV-too-long",
 			},
-			ExpErr: "Node name will not be discoverable via DNS due to it being too long. Valid lengths are between " +
-				"1 and 63 bytes. sync-consul-node-name=5r9OPGfSRXUdGzNjBdAwmhCBrzHDNYs4XjZVR4wp7lSLIzqwS0ta51nBLIN0TMPV-too-long",
+			ExpErr: "-sync-consul-node-name=5r9OPGfSRXUdGzNjBdAwmhCBrzHDNYs4XjZVR4wp7lSLIzqwS0ta51nBLIN0TMPV-too-long is invalid: node name will not be discoverable "+
+				"via DNS due to it being too long. Valid lengths are between 1 and 63 bytes",
 		},
 	}
 
@@ -895,7 +895,7 @@ func TestRun_BindingRuleUpdates(t *testing.T) {
 	firstRunArgs := append(commonArgs,
 		"-acl-binding-rule-selector=serviceaccount.name!=default",
 	)
-	// Our second run, we change the binding rule selector.
+	// On the second run, we change the binding rule selector.
 	secondRunArgs := append(commonArgs,
 		"-acl-binding-rule-selector=serviceaccount.name!=changed",
 	)
@@ -955,7 +955,6 @@ func TestRun_BindingRuleUpdates(t *testing.T) {
 func TestRun_SyncPolicyUpdates(t *testing.T) {
 	t.Parallel()
 	k8s, testSvr := completeSetup(t)
-	setUpK8sServiceAccount(t, k8s)
 	defer testSvr.Stop()
 	require := require.New(t)
 
@@ -970,7 +969,7 @@ func TestRun_SyncPolicyUpdates(t *testing.T) {
 	firstRunArgs := append(commonArgs,
 		"-sync-consul-node-name=k8s-sync",
 	)
-	// Our second run, we change the binding rule selector.
+	// On the second run, we change the sync node name.
 	secondRunArgs := append(commonArgs,
 		"-sync-consul-node-name=new-node-name",
 	)

subcommand/server-acl-init/create_or_update.go

@@ -108,7 +108,7 @@ func (c *Command) createOrUpdateACLPolicy(policy api.ACLPolicy, consulClient *ap
 	// settings, the policies associated with their ACL tokens will need to be
 	// updated to be namespace aware.
 	// Allowing the Consul node name to be configurable also requires any sync
-	// token to be updated in case the node name has changed.
+	// policy to be updated in case the node name has changed.
 	if isPolicyExistsErr(err, policy.Name) {
 		if c.flagEnableNamespaces || c.flagCreateSyncToken {
 			c.log.Info(fmt.Sprintf("Policy %q already exists, updating", policy.Name))

subcommand/sync-catalog/command.go

@@ -280,6 +280,7 @@ func (c *Command) Run(args []string) int {
 				ConsulDestinationNamespace: c.flagConsulDestinationNamespace,
 				EnableK8SNSMirroring:       c.flagEnableK8SNSMirroring,
 				K8SNSMirroringPrefix:       c.flagK8SNSMirroringPrefix,
+				ConsulNodeName:             c.flagConsulNodeName,
 			},
 		}
 
@@ -392,20 +393,18 @@ func (c *Command) validateFlags() error {
 	// For the Consul node name to be discoverable via DNS, it must contain only
 	// dashes and alphanumeric characters. Length is also constrained.
 	// These restrictions match those defined in Consul's agent definition.
-	var InvalidDnsRe = regexp.MustCompile(`[^A-Za-z0-9\\-]+`)
-	const MaxDNSLabelLength = 63
+	var invalidDnsRe = regexp.MustCompile(`[^A-Za-z0-9\\-]+`)
+	const maxDNSLabelLength = 63
 
-	if InvalidDnsRe.MatchString(c.flagConsulNodeName) {
-		return fmt.Errorf("Node name will not be discoverable "+
-			"via DNS due to invalid characters. Valid characters include "+
-			"all alpha-numerics and dashes. consul-node-name=%s",
+	if invalidDnsRe.MatchString(c.flagConsulNodeName) {
+		return fmt.Errorf("-consul-node-name=%s is invalid: node name will not be discoverable "+
+			"via DNS due to invalid characters. Valid characters include all alpha-numerics and dashes",
 			c.flagConsulNodeName,
 		)
 	}
-	if len(c.flagConsulNodeName) > MaxDNSLabelLength {
-		return fmt.Errorf("Node name will not be discoverable "+
-			"via DNS due to it being too long. Valid lengths are between "+
-			"1 and 63 bytes. consul-node-name=%s",
+	if len(c.flagConsulNodeName) > maxDNSLabelLength {
+		return fmt.Errorf("-consul-node-name=%s is invalid: node name will not be discoverable "+
+			"via DNS due to it being too long. Valid lengths are between 1 and 63 bytes",
 			c.flagConsulNodeName,
 		)
 	}

subcommand/sync-catalog/command_test.go

@@ -26,13 +26,13 @@ func TestRun_FlagValidation(t *testing.T) {
 	}{
 		{
 			Flags: []string{"-consul-node-name=Speci@l_Chars"},
-			ExpErr: "Node name will not be discoverable via DNS due to invalid characters. Valid characters include " +
-				"all alpha-numerics and dashes. consul-node-name=Speci@l_Chars",
+			ExpErr: "-consul-node-name=Speci@l_Chars is invalid: node name will not be discoverable "+
+				"via DNS due to invalid characters. Valid characters include all alpha-numerics and dashes",
 		},
 		{
 			Flags: []string{"-consul-node-name=5r9OPGfSRXUdGzNjBdAwmhCBrzHDNYs4XjZVR4wp7lSLIzqwS0ta51nBLIN0TMPV-too-long"},
-			ExpErr: "Node name will not be discoverable via DNS due to it being too long. Valid lengths are between " +
-				"1 and 63 bytes. consul-node-name=5r9OPGfSRXUdGzNjBdAwmhCBrzHDNYs4XjZVR4wp7lSLIzqwS0ta51nBLIN0TMPV-too-long",
+			ExpErr: "-consul-node-name=5r9OPGfSRXUdGzNjBdAwmhCBrzHDNYs4XjZVR4wp7lSLIzqwS0ta51nBLIN0TMPV-too-long is invalid: node name will not be discoverable "+
+				"via DNS due to it being too long. Valid lengths are between 1 and 63 bytes",
 		},
 	}