consul/connect: add support for connect mesh gateways

This PR implements first-class support for Nomad running Consul
Connect Mesh Gateways. Mesh gateways enable services in the Connect
mesh to make cross-DC connections via gateways, where each datacenter
may not have full node interconnectivity.

Consul docs with more information:
https://www.consul.io/docs/connect/gateways/mesh-gateway

The following group level service block can be used to establish
a Connect mesh gateway.

service {
  connect {
    gateway {
      mesh {
        // no configuration
      }
    }
  }
}

Services can make use of a mesh gateway by configuring so in their
upstream blocks, e.g.

service {
  connect {
    sidecar_service {
      proxy {
        upstreams {
          destination_name = "<service>"
          local_bind_port  = <port>
          datacenter       = "<datacenter>"
          mesh_gateway {
            mode = "<mode>"
          }
        }
      }
    }
  }
}

Typical use of a mesh gateway is to create a bridge between datacenters.
A mesh gateway should then be configured with a service port that is
mapped from a host_network configured on a WAN interface in Nomad agent
config, e.g.

client {
  host_network "public" {
    interface = "eth1"
  }
}

Create a port mapping in the group.network block for use by the mesh
gateway service from the public host_network, e.g.

network {
  mode = "bridge"
  port "mesh_wan" {
    host_network = "public"
  }
}

Use this port label for the service.port of the mesh gateway, e.g.

service {
  name = "mesh-gateway"
  port = "mesh_wan"
  connect {
    gateway {
      mesh {}
    }
  }
}

Currently Envoy is the only supported gateway implementation in Consul.
By default Nomad client will run the latest official Envoy docker image
supported by the local Consul agent. The Envoy task can be customized
by setting `meta.connect.gateway_image` in agent config or by setting
the `connect.sidecar_task` block.

Gateways require Consul 1.8.0+, enforced by the Nomad scheduler.

Closes #9446

api/services.go

@@ -281,17 +281,81 @@ func (cp *ConsulProxy) Canonicalize() {
 		cp.Upstreams = nil
 	}
 
+	for i := 0; i < len(cp.Upstreams); i++ {
+		cp.Upstreams[i].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() {
+	if c == nil {
+		return
+	}
+
+	if c.Mode == "" {
+		c.Mode = "none"
+	}
+}
+
+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 +390,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 +707,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

@@ -214,6 +214,68 @@ func TestService_Connect_ConsulProxy_Canonicalize(t *testing.T) {
 		require.Nil(t, cp.Upstreams)
 		require.Nil(t, cp.Config)
 	})
+
+	t.Run("fix upstream mesh_gateway", func(t *testing.T) {
+		cp := &ConsulProxy{
+			Upstreams: []*ConsulUpstream{{
+				MeshGateway: &ConsulMeshGateway{
+					Mode: "",
+				}},
+			},
+		}
+		cp.Canonicalize()
+		require.Equal(t, "none", cp.Upstreams[0].MeshGateway.Mode)
+	})
+}
+
+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("fix mesh_gateway", 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: "none"},
+		}, cu)
+	})
 }
 
 func TestService_Connect_proxy_settings(t *testing.T) {
@@ -508,3 +570,73 @@ 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, &ConsulMeshGateway{
+			Mode: "none",
+		}, c)
+	})
+}
+
+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"))
+}