agent/local: anti-entropy for connect proxy services

agent/agent_endpoint_test.go

@@ -1365,6 +1365,39 @@ func TestAgent_RegisterService_InvalidAddress(t *testing.T) {
 	}
 }
 
+func TestAgent_RegisterService_ConnectProxy(t *testing.T) {
+	t.Parallel()
+
+	assert := assert.New(t)
+	a := NewTestAgent(t.Name(), "")
+	defer a.Shutdown()
+
+	args := &structs.ServiceDefinition{
+		Kind:             structs.ServiceKindConnectProxy,
+		Name:             "connect-proxy",
+		Port:             8000,
+		ProxyDestination: "db",
+		Check: structs.CheckType{
+			TTL: 15 * time.Second,
+		},
+	}
+
+	req, _ := http.NewRequest("PUT", "/v1/agent/service/register?token=abc123", jsonReader(args))
+	resp := httptest.NewRecorder()
+	obj, err := a.srv.AgentRegisterService(resp, req)
+	assert.Nil(err)
+	assert.Nil(obj)
+
+	// Ensure the servie
+	svc, ok := a.State.Services()["connect-proxy"]
+	assert.True(ok, "has service")
+	assert.Equal(structs.ServiceKindConnectProxy, svc.Kind)
+	assert.Equal("db", svc.ProxyDestination)
+
+	// Ensure the token was configured
+	assert.Equal("abc123", a.State.ServiceToken("connect-proxy"))
+}
+
 func TestAgent_DeregisterService(t *testing.T) {
 	t.Parallel()
 	a := NewTestAgent(t.Name(), "")

agent/local/state_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/hashicorp/consul/testutil/retry"
 	"github.com/hashicorp/consul/types"
 	"github.com/pascaldekloe/goe/verify"
+	"github.com/stretchr/testify/assert"
 )
 
 func TestAgentAntiEntropy_Services(t *testing.T) {
@@ -224,6 +225,145 @@ func TestAgentAntiEntropy_Services(t *testing.T) {
 	}
 }
 
+func TestAgentAntiEntropy_Services_ConnectProxy(t *testing.T) {
+	t.Parallel()
+
+	assert := assert.New(t)
+	a := &agent.TestAgent{Name: t.Name()}
+	a.Start()
+	defer a.Shutdown()
+
+	// Register node info
+	var out struct{}
+	args := &structs.RegisterRequest{
+		Datacenter: "dc1",
+		Node:       a.Config.NodeName,
+		Address:    "127.0.0.1",
+	}
+
+	// Exists both same (noop)
+	srv1 := &structs.NodeService{
+		Kind:             structs.ServiceKindConnectProxy,
+		ID:               "mysql-proxy",
+		Service:          "mysql-proxy",
+		Port:             5000,
+		ProxyDestination: "db",
+	}
+	a.State.AddService(srv1, "")
+	args.Service = srv1
+	assert.Nil(a.RPC("Catalog.Register", args, &out))
+
+	// Exists both, different (update)
+	srv2 := &structs.NodeService{
+		ID:               "redis-proxy",
+		Service:          "redis-proxy",
+		Port:             8000,
+		Kind:             structs.ServiceKindConnectProxy,
+		ProxyDestination: "redis",
+	}
+	a.State.AddService(srv2, "")
+
+	srv2_mod := new(structs.NodeService)
+	*srv2_mod = *srv2
+	srv2_mod.Port = 9000
+	args.Service = srv2_mod
+	assert.Nil(a.RPC("Catalog.Register", args, &out))
+
+	// Exists local (create)
+	srv3 := &structs.NodeService{
+		ID:               "web-proxy",
+		Service:          "web-proxy",
+		Port:             80,
+		Kind:             structs.ServiceKindConnectProxy,
+		ProxyDestination: "web",
+	}
+	a.State.AddService(srv3, "")
+
+	// Exists remote (delete)
+	srv4 := &structs.NodeService{
+		ID:               "lb-proxy",
+		Service:          "lb-proxy",
+		Port:             443,
+		Kind:             structs.ServiceKindConnectProxy,
+		ProxyDestination: "lb",
+	}
+	args.Service = srv4
+	assert.Nil(a.RPC("Catalog.Register", args, &out))
+
+	// Exists local, in sync, remote missing (create)
+	srv5 := &structs.NodeService{
+		ID:               "cache-proxy",
+		Service:          "cache-proxy",
+		Port:             11211,
+		Kind:             structs.ServiceKindConnectProxy,
+		ProxyDestination: "cache-proxy",
+	}
+	a.State.SetServiceState(&local.ServiceState{
+		Service: srv5,
+		InSync:  true,
+	})
+
+	assert.Nil(a.State.SyncFull())
+
+	var services structs.IndexedNodeServices
+	req := structs.NodeSpecificRequest{
+		Datacenter: "dc1",
+		Node:       a.Config.NodeName,
+	}
+	assert.Nil(a.RPC("Catalog.NodeServices", &req, &services))
+
+	// We should have 5 services (consul included)
+	assert.Len(services.NodeServices.Services, 5)
+
+	// All the services should match
+	for id, serv := range services.NodeServices.Services {
+		serv.CreateIndex, serv.ModifyIndex = 0, 0
+		switch id {
+		case "mysql-proxy":
+			assert.Equal(srv1, serv)
+		case "redis-proxy":
+			assert.Equal(srv2, serv)
+		case "web-proxy":
+			assert.Equal(srv3, serv)
+		case "cache-proxy":
+			assert.Equal(srv5, serv)
+		case structs.ConsulServiceID:
+			// ignore
+		default:
+			t.Fatalf("unexpected service: %v", id)
+		}
+	}
+
+	assert.Nil(servicesInSync(a.State, 4))
+
+	// Remove one of the services
+	a.State.RemoveService("cache-proxy")
+	assert.Nil(a.State.SyncFull())
+	assert.Nil(a.RPC("Catalog.NodeServices", &req, &services))
+
+	// We should have 4 services (consul included)
+	assert.Len(services.NodeServices.Services, 4)
+
+	// All the services should match
+	for id, serv := range services.NodeServices.Services {
+		serv.CreateIndex, serv.ModifyIndex = 0, 0
+		switch id {
+		case "mysql-proxy":
+			assert.Equal(srv1, serv)
+		case "redis-proxy":
+			assert.Equal(srv2, serv)
+		case "web-proxy":
+			assert.Equal(srv3, serv)
+		case structs.ConsulServiceID:
+			// ignore
+		default:
+			t.Fatalf("unexpected service: %v", id)
+		}
+	}
+
+	assert.Nil(servicesInSync(a.State, 3))
+}
+
 func TestAgentAntiEntropy_EnableTagOverride(t *testing.T) {
 	t.Parallel()
 	a := &agent.TestAgent{Name: t.Name()}

agent/structs/service_definition.go

@@ -2,6 +2,7 @@ package structs
 
 // ServiceDefinition is used to JSON decode the Service definitions
 type ServiceDefinition struct {
+	Kind              ServiceKind
 	ID                string
 	Name              string
 	Tags              []string
@@ -12,10 +13,12 @@ type ServiceDefinition struct {
 	Checks            CheckTypes
 	Token             string
 	EnableTagOverride bool
+	ProxyDestination  string
 }
 
 func (s *ServiceDefinition) NodeService() *NodeService {
 	ns := &NodeService{
+		Kind:              s.Kind,
 		ID:                s.ID,
 		Service:           s.Name,
 		Tags:              s.Tags,
@@ -23,6 +26,7 @@ func (s *ServiceDefinition) NodeService() *NodeService {
 		Meta:              s.Meta,
 		Port:              s.Port,
 		EnableTagOverride: s.EnableTagOverride,
+		ProxyDestination:  s.ProxyDestination,
 	}
 	if ns.ID == "" && ns.Service != "" {
 		ns.ID = ns.Service

agent/structs/testing_service_definition.go

@@ -0,0 +1,13 @@
+package structs
+
+import (
+	"github.com/mitchellh/go-testing-interface"
+)
+
+// TestServiceDefinitionProxy returns a ServiceDefinition for a proxy.
+func TestServiceDefinitionProxy(t testing.T) *ServiceDefinition {
+	return &ServiceDefinition{
+		Kind:             ServiceKindConnectProxy,
+		ProxyDestination: "db",
+	}
+}