consul/connect: add support for connect mesh gateways

api/services.go

@@ -281,17 +281,77 @@ func (cp *ConsulProxy) Canonicalize() {
 		cp.Upstreams = nil
 	}
 
+	for _, upstream := range cp.Upstreams {
+		upstream.Canonicalize()
+	}
+
 	if len(cp.Config) == 0 {
 		cp.Config = nil
 	}
 }
 
+// ConsulMeshGateway is used to configure mesh gateway usage when connecting to
+// a connect upstream in another datacenter.
+type ConsulMeshGateway struct {
+	// Mode configures how an upstream should be accessed with regard to using
+	// mesh gateways.
+	//
+	// local - the connect proxy makes outbound connections through mesh gateway
+	// originating in the same datacenter.
+	//
+	// remote - the connect proxy makes outbound connections to a mesh gateway
+	// in the destination datacenter.
+	//
+	// none (default) - no mesh gateway is used, the proxy makes outbound connections
+	// directly to destination services.
+	//
+	// https://www.consul.io/docs/connect/gateways/mesh-gateway#modes-of-operation
+	Mode string `mapstructure:"mode" hcl:"mode,optional"`
+}
+
+func (c *ConsulMeshGateway) Canonicalize() {
+	// Mode may be empty string, indicating behavior will defer to Consul
+	// service-defaults config entry.
+	return
+}
+
+func (c *ConsulMeshGateway) Copy() *ConsulMeshGateway {
+	if c == nil {
+		return nil
+	}
+
+	return &ConsulMeshGateway{
+		Mode: c.Mode,
+	}
+}
+
 // ConsulUpstream represents a Consul Connect upstream jobspec stanza.
 type ConsulUpstream struct {
-	DestinationName  string `mapstructure:"destination_name" hcl:"destination_name,optional"`
-	LocalBindPort    int    `mapstructure:"local_bind_port" hcl:"local_bind_port,optional"`
-	Datacenter       string `mapstructure:"datacenter" hcl:"datacenter,optional"`
-	LocalBindAddress string `mapstructure:"local_bind_address" hcl:"local_bind_address,optional"`
+	DestinationName  string             `mapstructure:"destination_name" hcl:"destination_name,optional"`
+	LocalBindPort    int                `mapstructure:"local_bind_port" hcl:"local_bind_port,optional"`
+	Datacenter       string             `mapstructure:"datacenter" hcl:"datacenter,optional"`
+	LocalBindAddress string             `mapstructure:"local_bind_address" hcl:"local_bind_address,optional"`
+	MeshGateway      *ConsulMeshGateway `mapstructure:"mesh_gateway" hcl:"mesh_gateway,block"`
+}
+
+func (cu *ConsulUpstream) Copy() *ConsulUpstream {
+	if cu == nil {
+		return nil
+	}
+	return &ConsulUpstream{
+		DestinationName:  cu.DestinationName,
+		LocalBindPort:    cu.LocalBindPort,
+		Datacenter:       cu.Datacenter,
+		LocalBindAddress: cu.LocalBindAddress,
+		MeshGateway:      cu.MeshGateway.Copy(),
+	}
+}
+
+func (cu *ConsulUpstream) Canonicalize() {
+	if cu == nil {
+		return
+	}
+	cu.MeshGateway.Canonicalize()
 }
 
 type ConsulExposeConfig struct {
@@ -326,8 +386,8 @@ type ConsulGateway struct {
 	// Terminating represents the Consul Configuration Entry for a Terminating Gateway.
 	Terminating *ConsulTerminatingConfigEntry `hcl:"terminating,block"`
 
-	// Mesh is not yet supported.
-	// Mesh *ConsulMeshConfigEntry
+	// Mesh indicates the Consul service should be a Mesh Gateway.
+	Mesh *ConsulMeshConfigEntry `hcl:"mesh,block"`
 }
 
 func (g *ConsulGateway) Canonicalize() {
@@ -643,6 +703,21 @@ func (e *ConsulTerminatingConfigEntry) Copy() *ConsulTerminatingConfigEntry {
 	}
 }
 
-// ConsulMeshConfigEntry is not yet supported.
-// type ConsulMeshConfigEntry struct {
-// }
+// ConsulMeshConfigEntry is a stub used to represent that the gateway service type
+// should be for a Mesh Gateway. Unlike Ingress and Terminating, there is no
+// actual Consul Config Entry type for mesh-gateway, at least for now. We still
+// create a type for future proofing, instead just using a bool for example.
+type ConsulMeshConfigEntry struct {
+	// nothing in here
+}
+
+func (e *ConsulMeshConfigEntry) Canonicalize() {
+	return
+}
+
+func (e *ConsulMeshConfigEntry) Copy() *ConsulMeshConfigEntry {
+	if e == nil {
+		return nil
+	}
+	return new(ConsulMeshConfigEntry)
+}

api/services_test.go

@@ -216,6 +216,56 @@ func TestService_Connect_ConsulProxy_Canonicalize(t *testing.T) {
 	})
 }
 
+func TestService_Connect_ConsulUpstream_Copy(t *testing.T) {
+	t.Parallel()
+
+	t.Run("nil upstream", func(t *testing.T) {
+		cu := (*ConsulUpstream)(nil)
+		result := cu.Copy()
+		require.Nil(t, result)
+	})
+
+	t.Run("complete upstream", func(t *testing.T) {
+		cu := &ConsulUpstream{
+			DestinationName:  "dest1",
+			Datacenter:       "dc2",
+			LocalBindPort:    2000,
+			LocalBindAddress: "10.0.0.1",
+			MeshGateway:      &ConsulMeshGateway{Mode: "remote"},
+		}
+		result := cu.Copy()
+		require.Equal(t, cu, result)
+	})
+}
+
+func TestService_Connect_ConsulUpstream_Canonicalize(t *testing.T) {
+	t.Parallel()
+
+	t.Run("nil upstream", func(t *testing.T) {
+		cu := (*ConsulUpstream)(nil)
+		cu.Canonicalize()
+		require.Nil(t, cu)
+	})
+
+	t.Run("complete", func(t *testing.T) {
+		cu := &ConsulUpstream{
+			DestinationName:  "dest1",
+			Datacenter:       "dc2",
+			LocalBindPort:    2000,
+			LocalBindAddress: "10.0.0.1",
+			MeshGateway:      &ConsulMeshGateway{Mode: ""},
+		}
+		cu.Canonicalize()
+		require.Equal(t, &ConsulUpstream{
+			DestinationName:  "dest1",
+			Datacenter:       "dc2",
+			LocalBindPort:    2000,
+			LocalBindAddress: "10.0.0.1",
+			MeshGateway:      &ConsulMeshGateway{Mode: ""},
+		}, cu)
+	})
+}
+
 func TestService_Connect_proxy_settings(t *testing.T) {
 	t.Parallel()
 
@@ -508,3 +558,75 @@ func TestService_ConsulTerminatingConfigEntry_Copy(t *testing.T) {
 		require.Equal(t, entry, result)
 	})
 }
+
+func TestService_ConsulMeshConfigEntry_Canonicalize(t *testing.T) {
+	t.Parallel()
+
+	t.Run("nil", func(t *testing.T) {
+		ce := (*ConsulMeshConfigEntry)(nil)
+		ce.Canonicalize()
+		require.Nil(t, ce)
+	})
+
+	t.Run("instantiated", func(t *testing.T) {
+		ce := new(ConsulMeshConfigEntry)
+		ce.Canonicalize()
+		require.NotNil(t, ce)
+	})
+}
+
+func TestService_ConsulMeshConfigEntry_Copy(t *testing.T) {
+	t.Parallel()
+
+	t.Run("nil", func(t *testing.T) {
+		ce := (*ConsulMeshConfigEntry)(nil)
+		ce2 := ce.Copy()
+		require.Nil(t, ce2)
+	})
+
+	t.Run("instantiated", func(t *testing.T) {
+		ce := new(ConsulMeshConfigEntry)
+		ce2 := ce.Copy()
+		require.NotNil(t, ce2)
+	})
+}
+
+func TestService_ConsulMeshGateway_Canonicalize(t *testing.T) {
+	t.Parallel()
+
+	t.Run("nil", func(t *testing.T) {
+		c := (*ConsulMeshGateway)(nil)
+		c.Canonicalize()
+		require.Nil(t, c)
+	})
+
+	t.Run("unset mode", func(t *testing.T) {
+		c := &ConsulMeshGateway{Mode: ""}
+		c.Canonicalize()
+		require.Equal(t, "", c.Mode)
+	})
+
+	t.Run("set mode", func(t *testing.T) {
+		c := &ConsulMeshGateway{Mode: "remote"}
+		c.Canonicalize()
+		require.Equal(t, "remote", c.Mode)
+	})
+}
+
+func TestService_ConsulMeshGateway_Copy(t *testing.T) {
+	t.Parallel()
+
+	t.Run("nil", func(t *testing.T) {
+		c := (*ConsulMeshGateway)(nil)
+		result := c.Copy()
+		require.Nil(t, result)
+	})
+
+	t.Run("instantiated", func(t *testing.T) {
+		c := &ConsulMeshGateway{
+			Mode: "local",
+		}
+		result := c.Copy()
+		require.Equal(t, c, result)
+	})
+}

client/allocrunner/taskrunner/envoy_bootstrap_hook.go

@@ -164,8 +164,18 @@ func (envoyBootstrapHook) Name() string {
 }
 
 func isConnectKind(kind string) bool {
-	kinds := []string{structs.ConnectProxyPrefix, structs.ConnectIngressPrefix, structs.ConnectTerminatingPrefix}
-	return helper.SliceStringContains(kinds, kind)
+	switch kind {
+	case structs.ConnectProxyPrefix:
+		return true
+	case structs.ConnectIngressPrefix:
+		return true
+	case structs.ConnectTerminatingPrefix:
+		return true
+	case structs.ConnectMeshPrefix:
+		return true
+	default:
+		return false
+	}
 }
 
 func (_ *envoyBootstrapHook) extractNameAndKind(kind structs.TaskKind) (string, string, error) {
@@ -183,7 +193,7 @@ func (_ *envoyBootstrapHook) extractNameAndKind(kind structs.TaskKind) (string,
 	return serviceKind, serviceName, nil
 }
 
-func (h *envoyBootstrapHook) lookupService(svcKind, svcName, tgName string, taskEnv *taskenv.TaskEnv) (*structs.Service, error) {
+func (h *envoyBootstrapHook) lookupService(svcKind, svcName string, taskEnv *taskenv.TaskEnv) (*structs.Service, error) {
 	tg := h.alloc.Job.LookupTaskGroup(h.alloc.TaskGroup)
 	interpolatedServices := taskenv.InterpolateServices(taskEnv, tg.Services)
 
@@ -221,7 +231,7 @@ func (h *envoyBootstrapHook) Prestart(ctx context.Context, req *ifs.TaskPrestart
 		return err
 	}
 
-	service, err := h.lookupService(serviceKind, serviceName, h.alloc.TaskGroup, req.TaskEnv)
+	service, err := h.lookupService(serviceKind, serviceName, req.TaskEnv)
 	if err != nil {
 		return err
 	}
@@ -404,6 +414,11 @@ func (h *envoyBootstrapHook) proxyServiceID(group string, service *structs.Servi
 	return agentconsul.MakeAllocServiceID(h.alloc.ID, "group-"+group, service)
 }
 
+// newEnvoyBootstrapArgs is used to prepare for the invocation of the
+// 'consul connect envoy' command with arguments which will bootstrap the connect
+// proxy or gateway.
+//
+// https://www.consul.io/commands/connect/envoy#consul-connect-envoy
 func (h *envoyBootstrapHook) newEnvoyBootstrapArgs(
 	group string, service *structs.Service,
 	grpcAddr, envoyAdminBind, siToken, filepath string,
@@ -416,19 +431,23 @@ func (h *envoyBootstrapHook) newEnvoyBootstrapArgs(
 	)
 
 	namespace = h.getConsulNamespace()
+	id := h.proxyServiceID(group, service)
 
 	switch {
 	case service.Connect.HasSidecar():
-		sidecarForID = h.proxyServiceID(group, service)
+		sidecarForID = id
 	case service.Connect.IsIngress():
-		proxyID = h.proxyServiceID(group, service)
+		proxyID = id
 		gateway = "ingress"
 	case service.Connect.IsTerminating():
-		proxyID = h.proxyServiceID(group, service)
+		proxyID = id
 		gateway = "terminating"
+	case service.Connect.IsMesh():
+		proxyID = id
+		gateway = "mesh"
 	}
 
-	h.logger.Debug("bootstrapping envoy",
+	h.logger.Info("bootstrapping envoy",
 		"sidecar_for", service.Name, "bootstrap_file", filepath,
 		"sidecar_for_id", sidecarForID, "grpc_addr", grpcAddr,
 		"admin_bind", envoyAdminBind, "gateway", gateway,

client/allocrunner/taskrunner/envoy_bootstrap_hook_test.go

@@ -193,6 +193,25 @@ func TestEnvoyBootstrapHook_envoyBootstrapArgs(t *testing.T) {
 			"-proxy-id", "_nomad-task-803cb569-881c-b0d8-9222-360bcc33157e-group-ig-ig-8080",
 		}, result)
 	})
+
+	t.Run("mesh gateway", func(t *testing.T) {
+		ebArgs := envoyBootstrapArgs{
+			consulConfig:   consulPlainConfig,
+			grpcAddr:       "1.1.1.1",
+			envoyAdminBind: "localhost:3333",
+			gateway:        "my-mesh-gateway",
+			proxyID:        "_nomad-task-803cb569-881c-b0d8-9222-360bcc33157e-group-mesh-mesh-8080",
+		}
+		result := ebArgs.args()
+		require.Equal(t, []string{"connect", "envoy",
+			"-grpc-addr", "1.1.1.1",
+			"-http-addr", "2.2.2.2",
+			"-admin-bind", "localhost:3333",
+			"-bootstrap",
+			"-gateway", "my-mesh-gateway",
+			"-proxy-id", "_nomad-task-803cb569-881c-b0d8-9222-360bcc33157e-group-mesh-mesh-8080",
+		}, result)
+	})
 }
 
 func TestEnvoyBootstrapHook_envoyBootstrapEnv(t *testing.T) {
@@ -809,3 +828,12 @@ func TestTaskRunner_EnvoyBootstrapHook_grpcAddress(t *testing.T) {
 		require.Equal(t, "127.0.0.1:8502", hostH.grpcAddress(nil))
 	})
 }
+
+func TestTaskRunner_EnvoyBootstrapHook_isConnectKind(t *testing.T) {
+	require.True(t, isConnectKind(structs.ConnectProxyPrefix))
+	require.True(t, isConnectKind(structs.ConnectIngressPrefix))
+	require.True(t, isConnectKind(structs.ConnectTerminatingPrefix))
+	require.True(t, isConnectKind(structs.ConnectMeshPrefix))
+	require.False(t, isConnectKind(""))
+	require.False(t, isConnectKind("something"))
+}