agent/consul: support catalog registration with Connect native

agent/consul/catalog_endpoint_test.go

@@ -444,6 +444,36 @@ service "foo" {
 	assert.Nil(msgpackrpc.CallWithCodec(codec, "Catalog.Register", &args, &out))
 }
 
+func TestCatalog_Register_ConnectNative(t *testing.T) {
+	t.Parallel()
+
+	assert := assert.New(t)
+	dir1, s1 := testServer(t)
+	defer os.RemoveAll(dir1)
+	defer s1.Shutdown()
+	codec := rpcClient(t, s1)
+	defer codec.Close()
+
+	args := structs.TestRegisterRequest(t)
+	args.Service.ConnectNative = true
+
+	// Register
+	var out struct{}
+	assert.Nil(msgpackrpc.CallWithCodec(codec, "Catalog.Register", &args, &out))
+
+	// List
+	req := structs.ServiceSpecificRequest{
+		Datacenter:  "dc1",
+		ServiceName: args.Service.Service,
+	}
+	var resp structs.IndexedServiceNodes
+	assert.Nil(msgpackrpc.CallWithCodec(codec, "Catalog.ServiceNodes", &req, &resp))
+	assert.Len(resp.ServiceNodes, 1)
+	v := resp.ServiceNodes[0]
+	assert.Equal(structs.ServiceKindTypical, v.ServiceKind)
+	assert.True(v.ServiceConnectNative)
+}
+
 func TestCatalog_Deregister(t *testing.T) {
 	t.Parallel()
 	dir1, s1 := testServer(t)
@@ -1877,6 +1907,37 @@ service "foo" {
 	assert.Equal("foo-proxy", v.ServiceName)
 }
 
+func TestCatalog_ListServiceNodes_ConnectNative(t *testing.T) {
+	t.Parallel()
+
+	assert := assert.New(t)
+	dir1, s1 := testServer(t)
+	defer os.RemoveAll(dir1)
+	defer s1.Shutdown()
+	codec := rpcClient(t, s1)
+	defer codec.Close()
+
+	testrpc.WaitForLeader(t, s1.RPC, "dc1")
+
+	// Register the service
+	args := structs.TestRegisterRequest(t)
+	args.Service.ConnectNative = true
+	var out struct{}
+	assert.Nil(msgpackrpc.CallWithCodec(codec, "Catalog.Register", args, &out))
+
+	// List
+	req := structs.ServiceSpecificRequest{
+		Datacenter:  "dc1",
+		ServiceName: args.Service.Service,
+		TagFilter:   false,
+	}
+	var resp structs.IndexedServiceNodes
+	assert.Nil(msgpackrpc.CallWithCodec(codec, "Catalog.ServiceNodes", &req, &resp))
+	assert.Len(resp.ServiceNodes, 1)
+	v := resp.ServiceNodes[0]
+	assert.Equal(args.Service.ConnectNative, v.ServiceConnectNative)
+}
+
 func TestCatalog_NodeServices(t *testing.T) {
 	t.Parallel()
 	dir1, s1 := testServer(t)
@@ -1958,6 +2019,36 @@ func TestCatalog_NodeServices_ConnectProxy(t *testing.T) {
 	assert.Equal(args.Service.ProxyDestination, v.ProxyDestination)
 }
 
+func TestCatalog_NodeServices_ConnectNative(t *testing.T) {
+	t.Parallel()
+
+	assert := assert.New(t)
+	dir1, s1 := testServer(t)
+	defer os.RemoveAll(dir1)
+	defer s1.Shutdown()
+	codec := rpcClient(t, s1)
+	defer codec.Close()
+
+	testrpc.WaitForLeader(t, s1.RPC, "dc1")
+
+	// Register the service
+	args := structs.TestRegisterRequest(t)
+	var out struct{}
+	assert.Nil(msgpackrpc.CallWithCodec(codec, "Catalog.Register", args, &out))
+
+	// List
+	req := structs.NodeSpecificRequest{
+		Datacenter: "dc1",
+		Node:       args.Node,
+	}
+	var resp structs.IndexedNodeServices
+	assert.Nil(msgpackrpc.CallWithCodec(codec, "Catalog.NodeServices", &req, &resp))
+
+	assert.Len(resp.NodeServices.Services, 1)
+	v := resp.NodeServices.Services[args.Service.Service]
+	assert.Equal(args.Service.ConnectNative, v.ConnectNative)
+}
+
 // Used to check for a regression against a known bug
 func TestCatalog_Register_FailedCase1(t *testing.T) {
 	t.Parallel()

agent/structs/structs.go

@@ -432,6 +432,7 @@ type ServiceNode struct {
 	ServicePort              int
 	ServiceEnableTagOverride bool
 	ServiceProxyDestination  string
+	ServiceConnectNative     bool
 
 	RaftIndex
 }
@@ -460,6 +461,7 @@ func (s *ServiceNode) PartialClone() *ServiceNode {
 		ServiceMeta:              nsmeta,
 		ServiceEnableTagOverride: s.ServiceEnableTagOverride,
 		ServiceProxyDestination:  s.ServiceProxyDestination,
+		ServiceConnectNative:     s.ServiceConnectNative,
 		RaftIndex: RaftIndex{
 			CreateIndex: s.CreateIndex,
 			ModifyIndex: s.ModifyIndex,
@@ -479,6 +481,7 @@ func (s *ServiceNode) ToNodeService() *NodeService {
 		Meta:              s.ServiceMeta,
 		EnableTagOverride: s.ServiceEnableTagOverride,
 		ProxyDestination:  s.ServiceProxyDestination,
+		ConnectNative:     s.ServiceConnectNative,
 		RaftIndex: RaftIndex{
 			CreateIndex: s.CreateIndex,
 			ModifyIndex: s.ModifyIndex,
@@ -526,6 +529,9 @@ type NodeService struct {
 	// earlier than their target services.
 	ProxyDestination string
 
+	// ConnectNative is true if this service speaks the Connect protocol.
+	ConnectNative bool
+
 	RaftIndex
 }
 
@@ -567,7 +573,8 @@ func (s *NodeService) IsSame(other *NodeService) bool {
 		!reflect.DeepEqual(s.Meta, other.Meta) ||
 		s.EnableTagOverride != other.EnableTagOverride ||
 		s.Kind != other.Kind ||
-		s.ProxyDestination != other.ProxyDestination {
+		s.ProxyDestination != other.ProxyDestination ||
+		s.ConnectNative != other.ConnectNative {
 		return false
 	}
 
@@ -590,6 +597,7 @@ func (s *NodeService) ToServiceNode(node string) *ServiceNode {
 		ServiceMeta:              s.Meta,
 		ServiceEnableTagOverride: s.EnableTagOverride,
 		ServiceProxyDestination:  s.ProxyDestination,
+		ServiceConnectNative:     s.ConnectNative,
 		RaftIndex: RaftIndex{
 			CreateIndex: s.CreateIndex,
 			ModifyIndex: s.ModifyIndex,

agent/structs/structs_test.go

@@ -331,6 +331,7 @@ func TestStructs_NodeService_IsSame(t *testing.T) {
 	check(func() { other.EnableTagOverride = false }, func() { other.EnableTagOverride = true })
 	check(func() { other.Kind = ServiceKindConnectProxy }, func() { other.Kind = "" })
 	check(func() { other.ProxyDestination = "" }, func() { other.ProxyDestination = "db" })
+	check(func() { other.ConnectNative = true }, func() { other.ConnectNative = false })
 }
 
 func TestStructs_HealthCheck_IsSame(t *testing.T) {