[NET-5916] Fix locality-aware routing config and tests (CE) (#19483)

Fix locality-aware routing config and tests

agent/agent.go

@@ -812,6 +812,7 @@ func (a *Agent) Start(ctx context.Context) error {
 			Logger:          a.proxyConfig.Logger.Named("agent-state"),
 			Tokens:          a.baseDeps.Tokens,
 			NodeName:        a.config.NodeName,
+			NodeLocality:    a.config.StructLocality(),
 			ResyncFrequency: a.config.LocalProxyConfigResyncInterval,
 		},
 	)
@@ -3686,6 +3687,13 @@ func (a *Agent) loadServices(conf *config.RuntimeConfig, snap map[structs.CheckI
 		}
 
 		ns := service.NodeService()
+
+		// We currently do not persist locality inherited from the node service
+		// (it is inherited at runtime). See agent/proxycfg-sources/local/sync.go.
+		// To support locality-aware service discovery in the future, persisting
+		// this data may be necessary. This does not impact agent-less deployments
+		// because locality is explicitly set on service registration there.
+
 		chkTypes, err := service.CheckTypes()
 		if err != nil {
 			return fmt.Errorf("Failed to validate checks for service %q: %v", service.Name, err)

agent/agent_endpoint.go

@@ -1166,6 +1166,13 @@ func (s *HTTPHandlers) AgentRegisterService(resp http.ResponseWriter, req *http.
 
 	// Get the node service.
 	ns := args.NodeService()
+
+	// We currently do not persist locality inherited from the node service
+	// (it is inherited at runtime). See agent/proxycfg-sources/local/sync.go.
+	// To support locality-aware service discovery in the future, persisting
+	// this data may be necessary. This does not impact agent-less deployments
+	// because locality is explicitly set on service registration there.
+
 	if ns.Weights != nil {
 		if err := structs.ValidateWeights(ns.Weights); err != nil {
 			return nil, HTTPError{StatusCode: http.StatusBadRequest, Reason: fmt.Sprintf("Invalid Weights: %v", err)}

agent/config/builder.go

@@ -1732,10 +1732,21 @@ func (b *builder) serviceVal(v *ServiceDefinition) *structs.ServiceDefinition {
 		Checks:            checks,
 		Proxy:             b.serviceProxyVal(v.Proxy),
 		Connect:           b.serviceConnectVal(v.Connect),
+		Locality:          b.serviceLocalityVal(v.Locality),
 		EnterpriseMeta:    v.EnterpriseMeta.ToStructs(),
 	}
 }
 
+func (b *builder) serviceLocalityVal(l *Locality) *structs.Locality {
+	if l == nil {
+		return nil
+	}
+	return &structs.Locality{
+		Region: stringVal(l.Region),
+		Zone:   stringVal(l.Zone),
+	}
+}
+
 func (b *builder) serviceKindVal(v *string) structs.ServiceKind {
 	if v == nil {
 		return structs.ServiceKindTypical

agent/config/config.go

@@ -404,6 +404,7 @@ type ServiceDefinition struct {
 	EnableTagOverride *bool                     `mapstructure:"enable_tag_override"`
 	Proxy             *ServiceProxy             `mapstructure:"proxy"`
 	Connect           *ServiceConnect           `mapstructure:"connect"`
+	Locality          *Locality                 `mapstructure:"locality"`
 
 	EnterpriseMeta `mapstructure:",squash"`
 }

agent/config/runtime_test.go

@@ -6576,6 +6576,10 @@ func TestLoad_FullConfig(t *testing.T) {
 		KVMaxValueSize:        1234567800,
 		LeaveDrainTime:        8265 * time.Second,
 		LeaveOnTerm:           true,
+		Locality: &Locality{
+			Region: strPtr("us-east-2"),
+			Zone:   strPtr("us-east-2b"),
+		},
 		Logging: logging.Config{
 			LogLevel:       "k1zo9Spt",
 			LogJSON:        true,
@@ -6678,6 +6682,10 @@ func TestLoad_FullConfig(t *testing.T) {
 						},
 					},
 				},
+				Locality: &structs.Locality{
+					Region: "us-east-1",
+					Zone:   "us-east-1a",
+				},
 			},
 			{
 				ID:      "MRHVMZuD",
@@ -6836,6 +6844,10 @@ func TestLoad_FullConfig(t *testing.T) {
 				Connect: &structs.ServiceConnect{
 					Native: true,
 				},
+				Locality: &structs.Locality{
+					Region: "us-west-1",
+					Zone:   "us-west-1a",
+				},
 				Checks: structs.CheckTypes{
 					&structs.CheckType{
 						CheckID:    "Zv99e9Ka",

agent/config/testdata/full-config.hcl

@@ -317,6 +317,10 @@ limits {
         write_rate = 101.0
     }
 }
+locality = {
+    region = "us-east-2"
+    zone = "us-east-2b"
+}
 log_level = "k1zo9Spt"
 log_json = true
 max_query_time = "18237s"
@@ -510,6 +514,10 @@ service = {
     connect {
         native = true
     }
+    locality = {
+        region = "us-west-1"
+        zone = "us-west-1a"
+    }
 }
 services = [
     {
@@ -550,6 +558,10 @@ services = [
         connect {
             sidecar_service {}
         }
+        locality = {
+            region = "us-east-1"
+            zone = "us-east-1a"
+        }
     },
     {
         id = "MRHVMZuD"

agent/config/testdata/full-config.json

@@ -366,6 +366,10 @@
       "write_rate": 101.0
     }
   },
+  "locality": {
+    "region": "us-east-2",
+    "zone": "us-east-2b"
+  },
   "log_level": "k1zo9Spt",
   "log_json": true,
   "max_query_time": "18237s",
@@ -598,6 +602,10 @@
     ],
     "connect": {
       "native": true
+    },
+    "locality": {
+      "region": "us-west-1",
+      "zone": "us-west-1a"
     }
   },
   "services": [
@@ -649,6 +657,10 @@
       },
       "connect": {
         "sidecar_service": {}
+      },
+      "locality": {
+        "region": "us-east-1",
+        "zone": "us-east-1a"
       }
     },
     {

agent/proxycfg-sources/local/sync.go

@@ -5,9 +5,10 @@ package local
 
 import (
 	"context"
-	proxysnapshot "github.com/hashicorp/consul/internal/mesh/proxy-snapshot"
 	"time"
 
+	proxysnapshot "github.com/hashicorp/consul/internal/mesh/proxy-snapshot"
+
 	"github.com/hashicorp/go-hclog"
 
 	"github.com/hashicorp/consul/agent/local"
@@ -35,6 +36,9 @@ type SyncConfig struct {
 	// NodeName is the name of the local agent node.
 	NodeName string
 
+	// NodeLocality
+	NodeLocality *structs.Locality
+
 	// Logger will be used to write log messages.
 	Logger hclog.Logger
 
@@ -110,6 +114,14 @@ func sync(cfg SyncConfig) {
 			Token: "",
 		}
 
+		// We inherit the node's locality at runtime (not persisted).
+		// The service locality takes precedence if it was set directly during
+		// registration.
+		svc = svc.DeepCopy()
+		if svc.Locality == nil {
+			svc.Locality = cfg.NodeLocality
+		}
+
 		// TODO(banks): need to work out when to default some stuff. For example
 		// Proxy.LocalServicePort is practically necessary for any sidecar and can
 		// default to the port of the sidecar service, but only if it's already

agent/proxycfg-sources/local/sync_test.go

@@ -72,6 +72,10 @@ func TestSync(t *testing.T) {
 	go Sync(ctx, SyncConfig{
 		Manager: cfgMgr,
 		State:   state,
+		NodeLocality: &structs.Locality{
+			Region: "some-region",
+			Zone:   "some-zone",
+		},
 		Tokens: tokens,
 		Logger: hclog.NewNullLogger(),
 	})
@@ -107,6 +111,13 @@ func TestSync(t *testing.T) {
 	select {
 	case reg := <-registerCh:
 		require.Equal(t, serviceID, reg.service.ID)
+		require.Equal(t,
+			&structs.Locality{
+				Region: "some-region",
+				Zone:   "some-zone",
+			},
+			reg.service.Locality,
+		)
 		require.Equal(t, userToken, reg.token)
 	case <-time.After(100 * time.Millisecond):
 		t.Fatal("timeout waiting for service to be registered")

command/services/register/register_test.go

@@ -9,6 +9,7 @@ import (
 	"testing"
 
 	"github.com/hashicorp/consul/agent"
+	"github.com/hashicorp/consul/api"
 	"github.com/hashicorp/consul/sdk/testutil"
 	"github.com/mitchellh/cli"
 	"github.com/stretchr/testify/require"
@@ -75,7 +76,7 @@ func TestCommand_File(t *testing.T) {
 	ui := cli.NewMockUi()
 	c := New(ui)
 
-	contents := `{ "Service": { "Name": "web" } }`
+	contents := `{ "Service": { "Name": "web", "Locality": { "Region": "us-east-1", "Zone": "us-east-1a" } } }`
 	f := testFile(t, "json")
 	defer os.Remove(f.Name())
 	if _, err := f.WriteString(contents); err != nil {
@@ -93,8 +94,11 @@ func TestCommand_File(t *testing.T) {
 	require.NoError(t, err)
 	require.Len(t, svcs, 1)
 
-	svc := svcs["web"]
-	require.NotNil(t, svc)
+	require.NotNil(t, svcs["web"])
+
+	svc, _, err := client.Agent().Service("web", nil)
+	require.NoError(t, err)
+	require.Equal(t, &api.Locality{Region: "us-east-1", Zone: "us-east-1a"}, svc.Locality)
 }
 
 func TestCommand_Flags(t *testing.T) {

test/integration/connect/envoy/helpers.bash

@@ -328,6 +328,19 @@ function get_envoy_cluster_config {
   "
 }
 
+function get_envoy_endpoints_configs {
+  local HOSTPORT=$1
+  local CLUSTER_NAME=$2
+  run retry_default curl -s -f $HOSTPORT/config_dump?include_eds=on
+  [ "$status" -eq 0 ]
+  echo "$output" | jq --raw-output "
+    .configs[]
+    | select(.\"@type\" == \"type.googleapis.com/envoy.admin.v3.EndpointsConfigDump\")
+    | .dynamic_endpoint_configs[]
+    | .endpoint_config
+  "
+}
+
 function get_envoy_stats_flush_interval {
   local HOSTPORT=$1
   run retry_default curl -s -f $HOSTPORT/config_dump
@@ -344,7 +357,7 @@ function snapshot_envoy_admin {
   local OUTDIR="${LOG_DIR}/envoy-snapshots/${DC}/${ENVOY_NAME}"
 
   mkdir -p "${OUTDIR}"
-  docker_wget "$DC" "http://${HOSTPORT}/config_dump" -q -O - >"${OUTDIR}/config_dump.json"
+  docker_wget "$DC" "http://${HOSTPORT}/config_dump?include_eds=on" -q -O - >"${OUTDIR}/config_dump.json"
   docker_wget "$DC" "http://${HOSTPORT}/clusters?format=json" -q -O - >"${OUTDIR}/clusters.json"
   docker_wget "$DC" "http://${HOSTPORT}/stats" -q -O - >"${OUTDIR}/stats.txt"
   docker_wget "$DC" "http://${HOSTPORT}/stats/prometheus" -q -O - >"${OUTDIR}/stats_prometheus.txt"

test/integration/connect/envoy/helpers.windows.bash

@@ -389,7 +389,7 @@ function snapshot_envoy_admin {
   local OUTDIR="${LOG_DIR}/envoy-snapshots/${DC}/${ENVOY_NAME}"
 
   mkdir -p "${OUTDIR}"
-  docker_consul_exec "$DC" bash -c "curl -s http://${HOSTPORT}/config_dump" > "${OUTDIR}/config_dump.json"
+  docker_consul_exec "$DC" bash -c "curl -s http://${HOSTPORT}/config_dump?include_eds=on" > "${OUTDIR}/config_dump.json"
   docker_consul_exec "$DC" bash -c "curl -s http://${HOSTPORT}/clusters?format=json" > "${OUTDIR}/clusters.json"
   docker_consul_exec "$DC" bash -c "curl -s http://${HOSTPORT}/stats" > "${OUTDIR}/stats.txt"
   docker_consul_exec "$DC" bash -c "curl -s http://${HOSTPORT}/stats/prometheus"  > "${OUTDIR}/stats_prometheus.txt"

test/integration/consul-container/libs/cluster/agent.go

@@ -10,7 +10,6 @@ import (
 	"io"
 
 	jsonpatch "github.com/evanphx/json-patch"
-	agentconfig "github.com/hashicorp/consul/agent/config"
 	"github.com/hashicorp/consul/api"
 	"github.com/hashicorp/consul/lib/decode"
 	"github.com/hashicorp/hcl"
@@ -96,10 +95,6 @@ func (c Config) Clone() Config {
 	return c2
 }
 
-type decodeTarget struct {
-	agentconfig.Config `mapstructure:",squash"`
-}
-
 // MutatebyAgentConfig mutates config by applying the fields in the input hclConfig
 // Note that the precedence order is config > hclConfig, because user provider hclConfig
 // may not work with the testing environment, e.g., data dir, agent name, etc.
@@ -135,7 +130,10 @@ func convertHcl2Json(in string) (string, error) {
 		return "", err
 	}
 
-	var target decodeTarget
+	// We target an opaque map so that changes to config fields not yet present
+	// in a tagged version of `consul` (missing from latest released schema)
+	// can be used in tests.
+	var target map[string]any
 	var md mapstructure.Metadata
 	d, err := mapstructure.NewDecoder(&mapstructure.DecoderConfig{
 		DecodeHook: mapstructure.ComposeDecodeHookFunc(