consul/connect: in-place update service definition when connect upstreams are modified

This PR fixes a bug where modifying the upstreams of a Connect sidecar proxy
would not result Consul applying the changes, unless an additional change to
the job would trigger a task replacement (thus replacing the service definition).

The fix is to check if upstreams have been modified between Nomad's view of the
sidecar service definition, and the service definition for the sidecar that is
actually registered in Consul.

Fixes #8754

command/agent/consul/service_client.go

@@ -248,9 +248,74 @@ func sidecarTagsDifferent(parent, wanted, sidecar []string) bool {
 	return tagsDifferent(wanted, sidecar)
 }
 
+// proxyUpstreamsDifferent determines if the sidecar_service.proxy.upstreams
+// configurations are different between the desired sidecar service state, and
+// the actual sidecar service state currently registered in Consul.
+func proxyUpstreamsDifferent(wanted *api.AgentServiceConnect, sidecar *api.AgentServiceConnectProxyConfig) bool {
+	// There is similar code that already does this in Nomad's API package,
+	// however here we are operating on Consul API package structs, and they do not
+	// provide such helper functions.
+
+	getProxyUpstreams := func(pc *api.AgentServiceConnectProxyConfig) []api.Upstream {
+		switch {
+		case pc == nil:
+			return nil
+		case len(pc.Upstreams) == 0:
+			return nil
+		default:
+			return pc.Upstreams
+		}
+	}
+
+	getConnectUpstreams := func(sc *api.AgentServiceConnect) []api.Upstream {
+		switch {
+		case sc.SidecarService.Proxy == nil:
+			return nil
+		case len(sc.SidecarService.Proxy.Upstreams) == 0:
+			return nil
+		default:
+			return sc.SidecarService.Proxy.Upstreams
+		}
+	}
+
+	upstreamsDifferent := func(a, b []api.Upstream) bool {
+		if len(a) != len(b) {
+			return true
+		}
+
+		for i := 0; i < len(a); i++ {
+			A := a[i]
+			B := b[i]
+			switch {
+			case A.Datacenter != B.Datacenter:
+				return true
+			case A.DestinationName != B.DestinationName:
+				return true
+			case A.LocalBindAddress != B.LocalBindAddress:
+				return true
+			case A.LocalBindPort != B.LocalBindPort:
+				return true
+			case A.MeshGateway.Mode != B.MeshGateway.Mode:
+				return true
+			case !reflect.DeepEqual(A.Config, B.Config):
+				return true
+			}
+		}
+		return false
+	}
+
+	return upstreamsDifferent(
+		getConnectUpstreams(wanted),
+		getProxyUpstreams(sidecar),
+	)
+}
+
 // connectSidecarDifferent returns true if Nomad expects there to be a sidecar
 // hanging off the desired parent service definition on the Consul side, and does
 // not match with what Consul has.
+//
+// This is used to determine if the connect sidecar service registration should be
+// updated - potentially (but not necessarily) in-place.
 func connectSidecarDifferent(wanted *api.AgentServiceRegistration, sidecar *api.AgentService) bool {
 	if wanted.Connect != nil && wanted.Connect.SidecarService != nil {
 		if sidecar == nil {
@@ -262,6 +327,11 @@ func connectSidecarDifferent(wanted *api.AgentServiceRegistration, sidecar *api.
 			// tags on the nomad definition have been modified
 			return true
 		}
+
+		if proxyUpstreamsDifferent(wanted.Connect, sidecar.Proxy) {
+			// proxy upstreams on the nomad definition have been modified
+			return true
+		}
 	}
 
 	// Either Nomad does not expect there to be a sidecar_service, or there is

command/agent/consul/service_client_test.go

@@ -1,7 +1,6 @@
 package consul
 
 import (
-	"reflect"
 	"testing"
 	"time"
 
@@ -12,30 +11,40 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-var (
+func TestSyncLogic_agentServiceUpdateRequired(t *testing.T) {
+	t.Parallel()
+
 	// the service as known by nomad
-	wanted = api.AgentServiceRegistration{
-		Kind:              "",
-		ID:                "aca4c175-1778-5ef4-0220-2ab434147d35",
-		Name:              "myservice",
-		Tags:              []string{"a", "b"},
-		Port:              9000,
-		Address:           "1.1.1.1",
-		EnableTagOverride: true,
-		Meta:              map[string]string{"foo": "1"},
-		Connect: &api.AgentServiceConnect{
-			Native: false,
-			SidecarService: &api.AgentServiceRegistration{
-				Kind: "connect-proxy",
-				ID:   "_nomad-task-8e8413af-b5bb-aa67-2c24-c146c45f1ec9-group-mygroup-myservice-9001-sidecar-proxy",
-				Name: "name-sidecar-proxy",
-				Tags: []string{"x", "y", "z"},
+	wanted := func() api.AgentServiceRegistration {
+		return api.AgentServiceRegistration{
+			Kind:              "",
+			ID:                "aca4c175-1778-5ef4-0220-2ab434147d35",
+			Name:              "myservice",
+			Tags:              []string{"a", "b"},
+			Port:              9000,
+			Address:           "1.1.1.1",
+			EnableTagOverride: true,
+			Meta:              map[string]string{"foo": "1"},
+			Connect: &api.AgentServiceConnect{
+				Native: false,
+				SidecarService: &api.AgentServiceRegistration{
+					Kind: "connect-proxy",
+					ID:   "_nomad-task-8e8413af-b5bb-aa67-2c24-c146c45f1ec9-group-mygroup-myservice-9001-sidecar-proxy",
+					Name: "name-sidecar-proxy",
+					Tags: []string{"x", "y", "z"},
+					Proxy: &api.AgentServiceConnectProxyConfig{
+						Upstreams: []api.Upstream{{
+							Datacenter:      "dc1",
+							DestinationName: "dest1",
+						}},
+					},
+				},
 			},
-		},
+		}
 	}
 
 	// the service (and + connect proxy) as known by consul
-	existing = &api.AgentService{
+	existing := &api.AgentService{
 		Kind:              "",
 		ID:                "aca4c175-1778-5ef4-0220-2ab434147d35",
 		Service:           "myservice",
@@ -46,16 +55,18 @@ var (
 		Meta:              map[string]string{"foo": "1"},
 	}
 
-	sidecar = &api.AgentService{
+	sidecar := &api.AgentService{
 		Kind:    "connect-proxy",
 		ID:      "_nomad-task-8e8413af-b5bb-aa67-2c24-c146c45f1ec9-group-mygroup-myservice-9001-sidecar-proxy",
 		Service: "myservice-sidecar-proxy",
 		Tags:    []string{"x", "y", "z"},
+		Proxy: &api.AgentServiceConnectProxyConfig{
+			Upstreams: []api.Upstream{{
+				Datacenter:      "dc1",
+				DestinationName: "dest1",
+			}},
+		},
 	}
-)
-
-func TestSyncLogic_agentServiceUpdateRequired(t *testing.T) {
-	t.Parallel()
 
 	// By default wanted and existing match. Each test should modify wanted in
 	// 1 way, and / or configure the type of sync operation that is being
@@ -69,7 +80,7 @@ func TestSyncLogic_agentServiceUpdateRequired(t *testing.T) {
 		exp bool,
 		reason syncReason,
 		tweak tweaker) {
-		result := agentServiceUpdateRequired(reason, tweak(wanted), existing, sidecar)
+		result := agentServiceUpdateRequired(reason, tweak(wanted()), existing, sidecar)
 		require.Equal(t, exp, result)
 	}
 
@@ -128,6 +139,40 @@ func TestSyncLogic_agentServiceUpdateRequired(t *testing.T) {
 		})
 	})
 
+	t.Run("different sidecar upstream", func(t *testing.T) {
+		try(t, true, syncNewOps, func(w asr) *asr {
+			w.Connect.SidecarService.Proxy.Upstreams[0].DestinationName = "dest2"
+			return &w
+		})
+	})
+
+	t.Run("remove sidecar upstream", func(t *testing.T) {
+		try(t, true, syncNewOps, func(w asr) *asr {
+			w.Connect.SidecarService.Proxy.Upstreams = nil
+			return &w
+		})
+	})
+
+	t.Run("additional sidecar upstream", func(t *testing.T) {
+		try(t, true, syncNewOps, func(w asr) *asr {
+			w.Connect.SidecarService.Proxy.Upstreams = append(
+				w.Connect.SidecarService.Proxy.Upstreams,
+				api.Upstream{
+					Datacenter:      "dc2",
+					DestinationName: "dest2",
+				},
+			)
+			return &w
+		})
+	})
+
+	t.Run("nil proxy block", func(t *testing.T) {
+		try(t, true, syncNewOps, func(w asr) *asr {
+			w.Connect.SidecarService.Proxy = nil
+			return &w
+		})
+	})
+
 	t.Run("different tags syncNewOps eto=true", func(t *testing.T) {
 		// sync is required even though eto=true, because NewOps indicates the
 		// service definition  in nomad has changed (e.g. job run a modified job)
@@ -165,12 +210,12 @@ func TestSyncLogic_agentServiceUpdateRequired(t *testing.T) {
 	})
 
 	// for remaining tests, EnableTagOverride = false
-	wanted.EnableTagOverride = false
 	existing.EnableTagOverride = false
 
 	t.Run("different tags syncPeriodic eto=false", func(t *testing.T) {
 		// sync is required because eto=false and the tags do not match
 		try(t, true, syncPeriodic, func(w asr) *asr {
+			w.EnableTagOverride = false
 			w.Tags = []string{"other", "tags"}
 			return &w
 		})
@@ -179,6 +224,7 @@ func TestSyncLogic_agentServiceUpdateRequired(t *testing.T) {
 	t.Run("different tags syncNewOps eto=false", func(t *testing.T) {
 		// sync is required because eto=false and the tags do not match
 		try(t, true, syncNewOps, func(w asr) *asr {
+			w.EnableTagOverride = false
 			w.Tags = []string{"other", "tags"}
 			return &w
 		})
@@ -188,6 +234,7 @@ func TestSyncLogic_agentServiceUpdateRequired(t *testing.T) {
 		// like the parent service, sync is required because eto=false and the
 		// sidecar's tags do not match
 		try(t, true, syncPeriodic, func(w asr) *asr {
+			w.EnableTagOverride = false
 			w.Connect.SidecarService.Tags = []string{"other", "tags"}
 			return &w
 		})
@@ -197,6 +244,7 @@ func TestSyncLogic_agentServiceUpdateRequired(t *testing.T) {
 		// like the parent service, sync is required because eto=false and the
 		// sidecar's tags do not match
 		try(t, true, syncNewOps, func(w asr) *asr {
+			w.EnableTagOverride = false
 			w.Connect.SidecarService.Tags = []string{"other", "tags"}
 			return &w
 		})
@@ -312,8 +360,10 @@ func TestSyncLogic_maybeTweakTags_emptySC(t *testing.T) {
 	// side (i.e. are we checking multiple nil pointers).
 
 	try := func(asr *api.AgentServiceRegistration) {
+		existing := &api.AgentService{Tags: []string{"a", "b"}}
+		sidecar := &api.AgentService{Tags: []string{"a", "b"}}
 		maybeTweakTags(asr, existing, sidecar)
-		require.False(t, reflect.DeepEqual([]string{"original"}, asr.Tags))
+		require.False(t, !tagsDifferent([]string{"original"}, asr.Tags))
 	}
 
 	try(&api.AgentServiceRegistration{
@@ -414,3 +464,138 @@ func TestServiceRegistration_CheckOnUpdate(t *testing.T) {
 		}
 	}
 }
+
+func TestSyncLogic_proxyUpstreamsDifferent(t *testing.T) {
+	t.Parallel()
+
+	upstream1 := func() api.Upstream {
+		return api.Upstream{
+			Datacenter:       "sfo",
+			DestinationName:  "billing",
+			LocalBindAddress: "127.0.0.1",
+			LocalBindPort:    5050,
+			MeshGateway: api.MeshGatewayConfig{
+				Mode: "remote",
+			},
+			Config: map[string]interface{}{"foo": 1},
+		}
+	}
+
+	upstream2 := func() api.Upstream {
+		return api.Upstream{
+			Datacenter:       "ny",
+			DestinationName:  "metrics",
+			LocalBindAddress: "127.0.0.1",
+			LocalBindPort:    6060,
+			MeshGateway: api.MeshGatewayConfig{
+				Mode: "local",
+			},
+			Config: nil,
+		}
+	}
+
+	newASC := func() *api.AgentServiceConnect {
+		return &api.AgentServiceConnect{
+			SidecarService: &api.AgentServiceRegistration{
+				Proxy: &api.AgentServiceConnectProxyConfig{
+					Upstreams: []api.Upstream{
+						upstream1(),
+						upstream2(),
+					},
+				},
+			},
+		}
+	}
+
+	original := newASC()
+
+	t.Run("same", func(t *testing.T) {
+		require.False(t, proxyUpstreamsDifferent(original, newASC().SidecarService.Proxy))
+	})
+
+	type proxy = *api.AgentServiceConnectProxyConfig
+	type tweaker = func(proxy)
+
+	try := func(t *testing.T, desc string, tweak tweaker) {
+		t.Run(desc, func(t *testing.T) {
+			p := newASC().SidecarService.Proxy
+			tweak(p)
+			require.True(t, proxyUpstreamsDifferent(original, p))
+		})
+	}
+
+	try(t, "empty upstreams", func(p proxy) {
+		p.Upstreams = make([]api.Upstream, 0)
+	})
+
+	try(t, "missing upstream", func(p proxy) {
+		p.Upstreams = []api.Upstream{
+			upstream1(),
+		}
+	})
+
+	try(t, "extra upstream", func(p proxy) {
+		p.Upstreams = []api.Upstream{
+			upstream1(),
+			upstream2(),
+			{
+				Datacenter:      "north",
+				DestinationName: "dest3",
+			},
+		}
+	})
+
+	try(t, "different datacenter", func(p proxy) {
+		diff := upstream2()
+		diff.Datacenter = "south"
+		p.Upstreams = []api.Upstream{
+			upstream1(),
+			diff,
+		}
+	})
+
+	try(t, "different destination", func(p proxy) {
+		diff := upstream2()
+		diff.DestinationName = "sink"
+		p.Upstreams = []api.Upstream{
+			upstream1(),
+			diff,
+		}
+	})
+
+	try(t, "different local_bind_address", func(p proxy) {
+		diff := upstream2()
+		diff.LocalBindAddress = "10.0.0.1"
+		p.Upstreams = []api.Upstream{
+			upstream1(),
+			diff,
+		}
+	})
+
+	try(t, "different local_bind_port", func(p proxy) {
+		diff := upstream2()
+		diff.LocalBindPort = 9999
+		p.Upstreams = []api.Upstream{
+			upstream1(),
+			diff,
+		}
+	})
+
+	try(t, "different mesh gateway mode", func(p proxy) {
+		diff := upstream2()
+		diff.MeshGateway.Mode = "none"
+		p.Upstreams = []api.Upstream{
+			upstream1(),
+			diff,
+		}
+	})
+
+	try(t, "different config", func(p proxy) {
+		diff := upstream1()
+		diff.Config = map[string]interface{}{"foo": 2}
+		p.Upstreams = []api.Upstream{
+			diff,
+			upstream2(),
+		}
+	})
+}