status-im
/
consul
mirror of https://github.com/status-im/consul.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

[NET-5217] [OSS] Derive sidecar proxy locality from parent service (#18437) 

* Add logging to locality policy application

In OSS, this is currently a no-op.

* Inherit locality when registering sidecars

When sidecar locality is not explicitly configured, inherit locality
from the proxied service.
This commit is contained in:
Michael Zalimeni 2023-08-10 14:00:44 -04:00 committed by GitHub
parent bee12c6b1f
commit 05604eeec1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 273 additions and 99 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.changelog/18437.txt Normal file
View File

@ -0,0 +1,3 @@
```release-note:bug
Inherit locality from services when registering sidecar proxies.
```

6
agent/sidecar_service.go
View File

@ -81,6 +81,12 @@ func sidecarServiceFromNodeService(ns *structs.NodeService, token string) (*stru
sidecar.Tags = append(sidecar.Tags, ns.Tags...)
}
// Copy the locality from the original service if locality was not provided
if sidecar.Locality == nil && ns.Locality != nil {
tmp := *ns.Locality
sidecar.Locality = &tmp
}
// Flag this as a sidecar - this is not persisted in catalog but only needed
// in local agent state to disambiguate lineage when deregistering the parent
// service later.

55
agent/sidecar_service_test.go
View File

@ -134,13 +134,17 @@ func TestAgent_sidecarServiceFromNodeService(t *testing.T) {
wantToken: "custom-token",
},
{
name: "inherit tags and meta",
name: "inherit locality, tags and meta",
sd: &structs.ServiceDefinition{
ID: "web1",
Name: "web",
Port: 1111,
Tags: []string{"foo"},
Meta: map[string]string{"foo": "bar"},
Locality: &structs.Locality{
Region: "us-east-1",
Zone: "us-east-1a",
},
Connect: &structs.ServiceConnect{
SidecarService: &structs.ServiceDefinition{},
},
@ -153,6 +157,55 @@ func TestAgent_sidecarServiceFromNodeService(t *testing.T) {
Port: 0,
Tags: []string{"foo"},
Meta: map[string]string{"foo": "bar"},
Locality: &structs.Locality{
Region: "us-east-1",
Zone: "us-east-1a",
},
LocallyRegisteredAsSidecar: true,
Proxy: structs.ConnectProxyConfig{
DestinationServiceName: "web",
DestinationServiceID: "web1",
LocalServiceAddress: "127.0.0.1",
LocalServicePort: 1111,
},
},
wantChecks: nil,
},
{
name: "retain locality, tags and meta if explicitly configured",
sd: &structs.ServiceDefinition{
ID: "web1",
Name: "web",
Port: 1111,
Tags: []string{"foo"},
Meta: map[string]string{"foo": "bar"},
Locality: &structs.Locality{
Region: "us-east-1",
Zone: "us-east-1a",
},
Connect: &structs.ServiceConnect{
SidecarService: &structs.ServiceDefinition{
Tags: []string{"bar"},
Meta: map[string]string{"baz": "qux"},
Locality: &structs.Locality{
Region: "us-east-2",
Zone: "us-east-2a",
},
},
},
},
wantNS: &structs.NodeService{
EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(),
Kind: structs.ServiceKindConnectProxy,
ID: "web1-sidecar-proxy",
Service: "web-sidecar-proxy",
Port: 0,
Tags: []string{"bar"},
Meta: map[string]string{"baz": "qux"},
Locality: &structs.Locality{
Region: "us-east-2",
Zone: "us-east-2a",
},
LocallyRegisteredAsSidecar: true,
Proxy: structs.ConnectProxyConfig{
DestinationServiceName: "web",

14
agent/xds/endpoints.go
View File

@ -6,6 +6,7 @@ package xds
import (
"errors"
"fmt"
"github.com/hashicorp/go-hclog"
"strconv"
envoy_cluster_v3 "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
@ -136,6 +137,7 @@ func (s *ResourceGenerator) endpointsFromSnapshotConnectProxy(cfgSnap *proxycfg.
endpoints, ok := cfgSnap.ConnectProxy.PreparedQueryEndpoints[uid]
if ok {
la := makeLoadAssignment(
s.Logger,
cfgSnap,
clusterName,
nil,
@ -161,6 +163,7 @@ func (s *ResourceGenerator) endpointsFromSnapshotConnectProxy(cfgSnap *proxycfg.
endpoints, ok := cfgSnap.ConnectProxy.DestinationGateways.Get(uid)
if ok {
la := makeLoadAssignment(
s.Logger,
cfgSnap,
name,
nil,
@ -229,6 +232,7 @@ func (s *ResourceGenerator) endpointsFromSnapshotMeshGateway(cfgSnap *proxycfg.C
clusterName := connect.GatewaySNI(key.Datacenter, key.Partition, cfgSnap.Roots.TrustDomain)
la := makeLoadAssignment(
s.Logger,
cfgSnap,
clusterName,
nil,
@ -246,6 +250,7 @@ func (s *ResourceGenerator) endpointsFromSnapshotMeshGateway(cfgSnap *proxycfg.C
clusterName := cfgSnap.ServerSNIFn(key.Datacenter, "")
la := makeLoadAssignment(
s.Logger,
cfgSnap,
clusterName,
nil,
@ -418,6 +423,7 @@ func (s *ResourceGenerator) endpointsFromServicesAndResolvers(
for subsetName, groups := range clusterEndpoints {
clusterName := connect.ServiceSNI(svc.Name, subsetName, svc.NamespaceOrDefault(), svc.PartitionOrDefault(), cfgSnap.Datacenter, cfgSnap.Roots.TrustDomain)
la := makeLoadAssignment(
s.Logger,
cfgSnap,
clusterName,
nil,
@ -455,6 +461,7 @@ func (s *ResourceGenerator) makeEndpointsForOutgoingPeeredServices(
groups := []loadAssignmentEndpointGroup{{Endpoints: serviceGroup.Nodes, OnlyPassing: false}}
la := makeLoadAssignment(
s.Logger,
cfgSnap,
clusterName,
nil,
@ -619,6 +626,7 @@ func (s *ResourceGenerator) makeUpstreamLoadAssignmentForPeerService(
return la, nil
}
la = makeLoadAssignment(
s.Logger,
cfgSnap,
clusterName,
nil,
@ -641,6 +649,7 @@ func (s *ResourceGenerator) makeUpstreamLoadAssignmentForPeerService(
return nil, nil
}
la = makeLoadAssignment(
s.Logger,
cfgSnap,
clusterName,
nil,
@ -773,6 +782,7 @@ func (s *ResourceGenerator) endpointsFromDiscoveryChain(
}
la := makeLoadAssignment(
s.Logger,
cfgSnap,
clusterName,
ti.PrioritizeByLocality,
@ -861,7 +871,7 @@ type loadAssignmentEndpointGroup struct {
OverrideHealth envoy_core_v3.HealthStatus
}
func makeLoadAssignment(cfgSnap *proxycfg.ConfigSnapshot, clusterName string, policy *structs.DiscoveryPrioritizeByLocality, endpointGroups []loadAssignmentEndpointGroup, localKey proxycfg.GatewayKey) *envoy_endpoint_v3.ClusterLoadAssignment {
func makeLoadAssignment(logger hclog.Logger, cfgSnap *proxycfg.ConfigSnapshot, clusterName string, policy *structs.DiscoveryPrioritizeByLocality, endpointGroups []loadAssignmentEndpointGroup, localKey proxycfg.GatewayKey) *envoy_endpoint_v3.ClusterLoadAssignment {
cla := &envoy_endpoint_v3.ClusterLoadAssignment{
ClusterName: clusterName,
Endpoints: make([]*envoy_endpoint_v3.LocalityLbEndpoints, 0, len(endpointGroups)),
@ -878,7 +888,7 @@ func makeLoadAssignment(cfgSnap *proxycfg.ConfigSnapshot, clusterName string, po
var priority uint32
for _, endpointGroup := range endpointGroups {
endpointsByLocality, err := groupedEndpoints(cfgSnap.ServiceLocality, policy, endpointGroup.Endpoints)
endpointsByLocality, err := groupedEndpoints(logger, cfgSnap.ServiceLocality, policy, endpointGroup.Endpoints)
if err != nil {
continue

3
agent/xds/endpoints_test.go
View File

@ -4,6 +4,7 @@
package xds
import (
"github.com/hashicorp/go-hclog"
"path/filepath"
"sort"
"testing"
@ -213,6 +214,7 @@ func Test_makeLoadAssignment(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := makeLoadAssignment(
hclog.NewNullLogger(),
&proxycfg.ConfigSnapshot{ServiceLocality: tt.locality},
tt.clusterName,
nil,
@ -223,6 +225,7 @@ func Test_makeLoadAssignment(t *testing.T) {
if tt.locality == nil {
got := makeLoadAssignment(
hclog.NewNullLogger(),
&proxycfg.ConfigSnapshot{ServiceLocality: &structs.Locality{Region: "us-west-1", Zone: "us-west-1a"}},
tt.clusterName,
nil,

6
agent/xds/locality_policy.go
View File

@ -5,16 +5,18 @@ package xds
import (
"fmt"
"github.com/hashicorp/go-hclog"
"github.com/hashicorp/consul/agent/structs"
)
func groupedEndpoints(locality *structs.Locality, policy *structs.DiscoveryPrioritizeByLocality, csns structs.CheckServiceNodes) ([]structs.CheckServiceNodes, error) {
func groupedEndpoints(logger hclog.Logger, locality *structs.Locality, policy *structs.DiscoveryPrioritizeByLocality, csns structs.CheckServiceNodes) ([]structs.CheckServiceNodes, error) {
switch {
case policy == nil || policy.Mode == "" || policy.Mode == "none":
return []structs.CheckServiceNodes{csns}, nil
case policy.Mode == "failover":
return prioritizeByLocalityFailover(locality, csns), nil
log := logger.Named("locality")
return prioritizeByLocalityFailover(log, locality, csns), nil
default:
return nil, fmt.Errorf("unexpected priortize-by-locality mode %q", policy.Mode)
}

3
agent/xds/locality_policy_oss.go
View File

@ -8,8 +8,9 @@ package xds
import (
"github.com/hashicorp/consul/agent/structs"
"github.com/hashicorp/go-hclog"
)
func prioritizeByLocalityFailover(locality *structs.Locality, csns structs.CheckServiceNodes) []structs.CheckServiceNodes {
func prioritizeByLocalityFailover(_ hclog.Logger, _ *structs.Locality, _ structs.CheckServiceNodes) []structs.CheckServiceNodes {
return nil
}

156
command/connect/envoy/envoy.go
View File

@ -344,15 +344,16 @@ func (c *cmd) run(args []string) int {
}
}
var svcForSidecar api.AgentService
if c.proxyID == "" {
switch {
case c.sidecarFor != "":
proxyID, err := proxyCmd.LookupProxyIDForSidecar(c.client, c.sidecarFor)
svcForSidecar, err := proxyCmd.LookupServiceForSidecar(c.client, c.sidecarFor)
if err != nil {
c.UI.Error(err.Error())
return 1
}
c.proxyID = proxyID
c.proxyID = svcForSidecar.ID
case c.gateway != "" && !c.register:
gatewaySvc, err := proxyCmd.LookupGatewayProxy(c.client, c.gatewayKind)
@ -394,77 +395,13 @@ func (c *cmd) run(args []string) int {
return 1
}
taggedAddrs := make(map[string]api.ServiceAddress)
lanAddr := c.lanAddress.Value()
if lanAddr.Address != "" {
taggedAddrs[structs.TaggedAddressLAN] = lanAddr
}
wanAddr := c.wanAddress.Value()
if wanAddr.Address != "" {
taggedAddrs[structs.TaggedAddressWAN] = wanAddr
}
tcpCheckAddr := lanAddr.Address
if tcpCheckAddr == "" {
// fallback to localhost as the gateway has to reside in the same network namespace
// as the agent
tcpCheckAddr = "127.0.0.1"
}
var proxyConf *api.AgentServiceConnectProxyConfig
if len(c.bindAddresses.value) > 0 {
// override all default binding rules and just bind to the user-supplied addresses
proxyConf = &api.AgentServiceConnectProxyConfig{
Config: map[string]interface{}{
"envoy_gateway_no_default_bind": true,
"envoy_gateway_bind_addresses": c.bindAddresses.value,
},
}
} else if canBind(lanAddr) && canBind(wanAddr) {
// when both addresses are bindable then we bind to the tagged addresses
// for creating the envoy listeners
proxyConf = &api.AgentServiceConnectProxyConfig{
Config: map[string]interface{}{
"envoy_gateway_no_default_bind": true,
"envoy_gateway_bind_tagged_addresses": true,
},
}
} else if !canBind(lanAddr) && lanAddr.Address != "" {
c.UI.Error(fmt.Sprintf("The LAN address %q will not be bindable. Either set a bindable address or override the bind addresses with -bind-address", lanAddr.Address))
svc, err := c.proxyRegistration(&svcForSidecar)
if err != nil {
c.UI.Error(err.Error())
return 1
}
var meta map[string]string
if c.exposeServers {
meta = map[string]string{structs.MetaWANFederationKey: "1"}
}
// API gateways do not have a default listener or ready endpoint,
// so adding any check to the registration will fail
var check *api.AgentServiceCheck
if c.gatewayKind != api.ServiceKindAPIGateway {
check = &api.AgentServiceCheck{
Name: fmt.Sprintf("%s listening", c.gatewayKind),
TCP: ipaddr.FormatAddressPort(tcpCheckAddr, lanAddr.Port),
Interval: "10s",
DeregisterCriticalServiceAfter: c.deregAfterCritical,
}
}
svc := api.AgentServiceRegistration{
Kind: c.gatewayKind,
Name: c.gatewaySvcName,
ID: c.proxyID,
Address: lanAddr.Address,
Port: lanAddr.Port,
Meta: meta,
TaggedAddresses: taggedAddrs,
Proxy: proxyConf,
Check: check,
}
if err := c.client.Agent().ServiceRegister(&svc); err != nil {
if err := c.client.Agent().ServiceRegister(svc); err != nil {
c.UI.Error(fmt.Sprintf("Error registering service %q: %s", svc.Name, err))
return 1
}
@ -542,6 +479,85 @@ func (c *cmd) run(args []string) int {
return 0
}
func (c *cmd) proxyRegistration(svcForSidecar *api.AgentService) (*api.AgentServiceRegistration, error) {
taggedAddrs := make(map[string]api.ServiceAddress)
lanAddr := c.lanAddress.Value()
if lanAddr.Address != "" {
taggedAddrs[structs.TaggedAddressLAN] = lanAddr
}
wanAddr := c.wanAddress.Value()
if wanAddr.Address != "" {
taggedAddrs[structs.TaggedAddressWAN] = wanAddr
}
tcpCheckAddr := lanAddr.Address
if tcpCheckAddr == "" {
// fallback to localhost as the gateway has to reside in the same network namespace
// as the agent
tcpCheckAddr = "127.0.0.1"
}
var proxyConf *api.AgentServiceConnectProxyConfig
if len(c.bindAddresses.value) > 0 {
// override all default binding rules and just bind to the user-supplied addresses
proxyConf = &api.AgentServiceConnectProxyConfig{
Config: map[string]interface{}{
"envoy_gateway_no_default_bind": true,
"envoy_gateway_bind_addresses": c.bindAddresses.value,
},
}
} else if canBind(lanAddr) && canBind(wanAddr) {
// when both addresses are bindable then we bind to the tagged addresses
// for creating the envoy listeners
proxyConf = &api.AgentServiceConnectProxyConfig{
Config: map[string]interface{}{
"envoy_gateway_no_default_bind": true,
"envoy_gateway_bind_tagged_addresses": true,
},
}
} else if !canBind(lanAddr) && lanAddr.Address != "" {
return nil, fmt.Errorf("The LAN address %q will not be bindable. Either set a bindable address or override the bind addresses with -bind-address", lanAddr.Address)
}
var meta map[string]string
if c.exposeServers {
meta = map[string]string{structs.MetaWANFederationKey: "1"}
}
// API gateways do not have a default listener or ready endpoint,
// so adding any check to the registration will fail
var check *api.AgentServiceCheck
if c.gatewayKind != api.ServiceKindAPIGateway {
check = &api.AgentServiceCheck{
Name: fmt.Sprintf("%s listening", c.gatewayKind),
TCP: ipaddr.FormatAddressPort(tcpCheckAddr, lanAddr.Port),
Interval: "10s",
DeregisterCriticalServiceAfter: c.deregAfterCritical,
}
}
// If registering a sidecar for an existing service, inherit the
// locality of that service if it was explicitly configured.
var locality *api.Locality
if c.sidecarFor != "" {
locality = svcForSidecar.Locality
}
return &api.AgentServiceRegistration{
Kind: c.gatewayKind,
Name: c.gatewaySvcName,
ID: c.proxyID,
Address: lanAddr.Address,
Port: lanAddr.Port,
Meta: meta,
TaggedAddresses: taggedAddrs,
Proxy: proxyConf,
Check: check,
Locality: locality,
}, nil
}
var errUnsupportedOS = errors.New("envoy: not implemented on this operating system")
func (c *cmd) findBinary() (string, error) {

77
command/connect/envoy/envoy_test.go
View File

@ -1399,6 +1399,83 @@ func TestEnvoy_GatewayRegistration(t *testing.T) {
}
}
func TestEnvoy_proxyRegistration(t *testing.T) {
t.Parallel()
type args struct {
svcForProxy api.AgentService
cmdFn func(*cmd)
}
cases := []struct {
name string
args args
testFn func(*testing.T, args, *api.AgentServiceRegistration)
}{
{
"locality is inherited from proxied service if configured and using sidecarFor",
args{
svcForProxy: api.AgentService{
ID: "my-svc",
Locality: &api.Locality{
Region: "us-east-1",
Zone: "us-east-1a",
},
},
cmdFn: func(c *cmd) {
c.sidecarFor = "my-svc"
},
},
func(t *testing.T, args args, r *api.AgentServiceRegistration) {
assert.NotNil(t, r.Locality)
assert.Equal(t, args.svcForProxy.Locality, r.Locality)
},
},
{
"locality is not inherited if not using sidecarFor",
args{
svcForProxy: api.AgentService{
ID: "my-svc",
Locality: &api.Locality{
Region: "us-east-1",
Zone: "us-east-1a",
},
},
},
func(t *testing.T, args args, r *api.AgentServiceRegistration) {
assert.Nil(t, r.Locality)
},
},
{
"locality is not set if not configured for proxied service",
args{
svcForProxy: api.AgentService{},
cmdFn: func(c *cmd) {
c.sidecarFor = "my-svc"
},
},
func(t *testing.T, args args, r *api.AgentServiceRegistration) {
assert.Nil(t, r.Locality)
},
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
ui := cli.NewMockUi()
c := New(ui)
if tc.args.cmdFn != nil {
tc.args.cmdFn(c)
}
result, err := c.proxyRegistration(&tc.args.svcForProxy)
assert.NoError(t, err)
tc.testFn(t, tc.args, result)
})
}
}
// testMockAgent combines testMockAgentProxyConfig and testMockAgentSelf,
// routing /agent/service/... requests to testMockAgentProxyConfig,
// routing /catalog/node-services/... requests to testMockCatalogNodeServiceList

35
command/connect/proxy/proxy.go
View File

@ -215,40 +215,42 @@ func (c *cmd) Run(args []string) int {
return 0
}
func (c *cmd) lookupProxyIDForSidecar(client *api.Client) (string, error) {
return LookupProxyIDForSidecar(client, c.sidecarFor)
func (c *cmd) lookupServiceForSidecar(client *api.Client) (*api.AgentService, error) {
return LookupServiceForSidecar(client, c.sidecarFor)
}
// LookupProxyIDForSidecar finds candidate local proxy registrations that are a
// sidecar for the given service. It will return an ID if and only if there is
// exactly one registered connect proxy with `Proxy.DestinationServiceID` set to
// LookupServiceForSidecar finds candidate local proxy registrations that are a
// sidecar for the given service. It will return that service if and only if there
// is exactly one registered connect proxy with `Proxy.DestinationServiceID` set to
// the specified service ID.
//
// This is exported to share it with the connect envoy command.
func LookupProxyIDForSidecar(client *api.Client, sidecarFor string) (string, error) {
func LookupServiceForSidecar(client *api.Client, sidecarFor string) (*api.AgentService, error) {
svcs, err := client.Agent().Services()
if err != nil {
return "", fmt.Errorf("Failed looking up sidecar proxy info for %s: %s",
return nil, fmt.Errorf("Failed looking up sidecar proxy info for %s: %s",
sidecarFor, err)
}
var proxyIDs []string
var matched []*api.AgentService
var matchedProxyIDs []string
for _, svc := range svcs {
if svc.Kind == api.ServiceKindConnectProxy && svc.Proxy != nil &&
strings.EqualFold(svc.Proxy.DestinationServiceID, sidecarFor) {
proxyIDs = append(proxyIDs, svc.ID)
matched = append(matched, svc)
matchedProxyIDs = append(matchedProxyIDs, svc.ID)
}
}
if len(proxyIDs) == 0 {
return "", fmt.Errorf("No sidecar proxy registered for %s", sidecarFor)
if len(matched) == 0 {
return nil, fmt.Errorf("No sidecar proxy registered for %s", sidecarFor)
}
if len(proxyIDs) > 1 {
return "", fmt.Errorf("More than one sidecar proxy registered for %s.\n"+
if len(matched) > 1 {
return nil, fmt.Errorf("More than one sidecar proxy registered for %s.\n"+
" Start proxy with -proxy-id and one of the following IDs: %s",
sidecarFor, strings.Join(proxyIDs, ", "))
sidecarFor, strings.Join(matchedProxyIDs, ", "))
}
return proxyIDs[0], nil
return matched[0], nil
}
// LookupGatewayProxy finds the gateway service registered with the local
@ -285,10 +287,11 @@ func (c *cmd) configWatcher(client *api.Client) (proxyImpl.ConfigWatcher, error)
// Running as a sidecar, we need to find the proxy-id for the requested
// service
var err error
c.proxyID, err = c.lookupProxyIDForSidecar(client)
svc, err := c.lookupServiceForSidecar(client)
if err != nil {
return nil, err
}
c.proxyID = svc.ID
c.UI.Info("Configuration mode: Agent API")
c.UI.Info(fmt.Sprintf(" Sidecar for ID: %s", c.sidecarFor))