Unix Domain Socket support for upstreams and downstreams

.changelog/9981.txt

@@ -0,0 +1,3 @@
+```release-note:feature
+connect: add support for unix domain sockets addresses for service upstreams and downstreams
+```

agent/agent_endpoint_test.go

@@ -399,7 +399,7 @@ func TestAgent_Service(t *testing.T) {
 		Service:     "web-sidecar-proxy",
 		Port:        8000,
 		Proxy:       expectProxy.ToAPI(),
-		ContentHash: "eb557bc310d4f8a0",
+		ContentHash: "35ad6dd5b1ff8d18",
 		Weights: api.AgentWeights{
 			Passing: 1,
 			Warning: 1,
@@ -413,7 +413,7 @@ func TestAgent_Service(t *testing.T) {
 	// Copy and modify
 	updatedResponse := *expectedResponse
 	updatedResponse.Port = 9999
-	updatedResponse.ContentHash = "d61c11f438c7eb02"
+	updatedResponse.ContentHash = "8e407e299ec9eba"
 
 	// Simple response for non-proxy service registered in TestAgent config
 	expectWebResponse := &api.AgentService{

agent/config/builder.go

@@ -14,6 +14,7 @@ import (
 	"reflect"
 	"regexp"
 	"sort"
+	"strconv"
 	"strings"
 	"time"
 
@@ -1640,6 +1641,14 @@ func (b *builder) serviceVal(v *ServiceDefinition) *structs.ServiceDefinition {
 	if err := structs.ValidateWeights(serviceWeights); err != nil {
 		b.err = multierror.Append(fmt.Errorf("Invalid weight definition for service %s: %s", stringVal(v.Name), err))
 	}
+
+	if (v.Port != nil || v.Address != nil) && (v.SocketPath != nil) {
+		b.err = multierror.Append(
+			fmt.Errorf("service %s cannot have both socket path %s and address/port",
+				stringVal(v.Name), stringVal(v.SocketPath)), b.err)
+
+	}
+
 	return &structs.ServiceDefinition{
 		Kind:              kind,
 		ID:                stringVal(v.ID),
@@ -1649,6 +1658,7 @@ func (b *builder) serviceVal(v *ServiceDefinition) *structs.ServiceDefinition {
 		TaggedAddresses:   b.svcTaggedAddresses(v.TaggedAddresses),
 		Meta:              meta,
 		Port:              intVal(v.Port),
+		SocketPath:        stringVal(v.SocketPath),
 		Token:             stringVal(v.Token),
 		EnableTagOverride: boolVal(v.EnableTagOverride),
 		Weights:           serviceWeights,
@@ -1687,6 +1697,7 @@ func (b *builder) serviceProxyVal(v *ServiceProxy) *structs.ConnectProxyConfig {
 		DestinationServiceID:   stringVal(v.DestinationServiceID),
 		LocalServiceAddress:    stringVal(v.LocalServiceAddress),
 		LocalServicePort:       intVal(v.LocalServicePort),
+		LocalServiceSocketPath: stringVal(&v.LocalServiceSocketPath),
 		Config:                 v.Config,
 		Upstreams:              b.upstreamsVal(v.Upstreams),
 		MeshGateway:            b.meshGatewayConfVal(v.MeshGateway),
@@ -1706,6 +1717,8 @@ func (b *builder) upstreamsVal(v []Upstream) structs.Upstreams {
 			Datacenter:           stringVal(u.Datacenter),
 			LocalBindAddress:     stringVal(u.LocalBindAddress),
 			LocalBindPort:        intVal(u.LocalBindPort),
+			LocalBindSocketPath:  stringVal(u.LocalBindSocketPath),
+			LocalBindSocketMode:  b.unixPermissionsVal("local_bind_socket_mode", u.LocalBindSocketMode),
 			Config:               u.Config,
 			MeshGateway:          b.meshGatewayConfVal(u.MeshGateway),
 		}
@@ -1892,6 +1905,18 @@ func uint64Val(v *uint64) uint64 {
 	return *v
 }
 
+// Expect an octal permissions string, e.g. 0644
+func (b *builder) unixPermissionsVal(name string, v *string) string {
+	if v == nil {
+		return ""
+	}
+	if _, err := strconv.ParseUint(*v, 8, 32); err == nil {
+		return *v
+	}
+	b.err = multierror.Append(b.err, fmt.Errorf("%s: invalid mode: %s", name, *v))
+	return "0"
+}
+
 func (b *builder) portVal(name string, v *int) int {
 	if v == nil || *v <= 0 {
 		return -1

agent/config/builder_test.go

@@ -175,6 +175,30 @@ func TestBuilder_BuildAndValidate_NodeName(t *testing.T) {
 	}
 }
 
+func TestBuilder_unixPermissionsVal(t *testing.T) {
+
+	b, _ := newBuilder(LoadOpts{
+		FlagValues: Config{
+			NodeName: pString("foo"),
+			DataDir:  pString("dir"),
+		},
+	})
+
+	goodmode := "666"
+	badmode := "9666"
+
+	patchLoadOptsShims(&b.opts)
+	require.NoError(t, b.err)
+	_ = b.unixPermissionsVal("local_bind_socket_mode", &goodmode)
+	require.NoError(t, b.err)
+	require.Len(t, b.Warnings, 0)
+
+	_ = b.unixPermissionsVal("local_bind_socket_mode", &badmode)
+	require.NotNil(t, b.err)
+	require.Contains(t, b.err.Error(), "local_bind_socket_mode: invalid mode")
+	require.Len(t, b.Warnings, 0)
+}
+
 func patchLoadOptsShims(opts *LoadOpts) {
 	if opts.hostname == nil {
 		opts.hostname = func() (string, error) {

agent/config/config.go

@@ -375,6 +375,7 @@ type ServiceDefinition struct {
 	TaggedAddresses   map[string]ServiceAddress `mapstructure:"tagged_addresses"`
 	Meta              map[string]string         `mapstructure:"meta"`
 	Port              *int                      `mapstructure:"port"`
+	SocketPath        *string                   `mapstructure:"socket_path"`
 	Check             *CheckDefinition          `mapstructure:"check"`
 	Checks            []CheckDefinition         `mapstructure:"checks"`
 	Token             *string                   `mapstructure:"token"`
@@ -461,6 +462,10 @@ type ServiceProxy struct {
 	// (DestinationServiceID is set) but otherwise will be ignored.
 	LocalServicePort *int `mapstructure:"local_service_port"`
 
+	// LocalServiceSocketPath is the socket of the local service instance. It is optional
+	// and should only be specified for "side-car" style proxies.
+	LocalServiceSocketPath string `mapstructure:"local_service_socket_path"`
+
 	// TransparentProxy configuration.
 	TransparentProxy *TransparentProxyConfig `mapstructure:"transparent_proxy"`
 
@@ -503,14 +508,21 @@ type Upstream struct {
 	// datacenter.
 	Datacenter *string `mapstructure:"datacenter"`
 
+	// It would be worth thinking about a separate structure for these four items,
+	// unifying under address as something like "unix:/tmp/foo", "tcp:localhost:80" could make sense
 	// LocalBindAddress is the ip address a side-car proxy should listen on for
-	// traffic destined for this upstream service. Default if empty is 127.0.0.1.
+	// traffic destined for this upstream service. Default if empty and local bind socket
+	// is not present is 127.0.0.1.
 	LocalBindAddress *string `mapstructure:"local_bind_address"`
 
 	// LocalBindPort is the ip address a side-car proxy should listen on for traffic
 	// destined for this upstream service. Required.
 	LocalBindPort *int `mapstructure:"local_bind_port"`
 
+	// These are exclusive with LocalBindAddress/LocalBindPort. These are created under our control.
+	LocalBindSocketPath *string `mapstructure:"local_bind_socket_path"`
+	LocalBindSocketMode *string `mapstructure:"local_bind_socket_mode"`
+
 	// Config is an opaque config that is specific to the proxy process being run.
 	// It can be used to pass arbitrary configuration for this specific upstream
 	// to the proxy.

agent/config/runtime_test.go

@@ -2590,6 +2590,11 @@ func TestLoad_IntegrationWithFlags(t *testing.T) {
 										{
 											"destination_name": "db",
 											"local_bind_port": 7000
+										},
+										{
+											"destination_name": "db2",
+											"local_bind_socket_path": "/tmp/socketpath",
+											"local_bind_socket_mode": "0644"
 										}
 									]
 								}
@@ -2631,6 +2636,11 @@ func TestLoad_IntegrationWithFlags(t *testing.T) {
 										destination_name = "db"
 										local_bind_port = 7000
 									},
+									{
+									    destination_name = "db2",
+									    local_bind_socket_path = "/tmp/socketpath",
+									    local_bind_socket_mode = "0644"
+									}
 								]
 							}
 						}
@@ -2675,6 +2685,12 @@ func TestLoad_IntegrationWithFlags(t *testing.T) {
 										DestinationName: "db",
 										LocalBindPort:   7000,
 									},
+									structs.Upstream{
+										DestinationType:     "service",
+										DestinationName:     "db2",
+										LocalBindSocketPath: "/tmp/socketpath",
+										LocalBindSocketMode: "0644",
+									},
 								},
 							},
 							Weights: &structs.Weights{

agent/config/testdata/full-config.hcl

@@ -362,6 +362,7 @@ service = {
     address = "cOlSOhbp"
     token = "msy7iWER"
     port = 24237
+    socket_path = "/tmp/rc78ap"
     weights = {
         passing = 100,
         warning = 1
@@ -455,6 +456,7 @@ services = [
         address = "9RhqPSPB"
         token = "myjKJkWH"
         port = 72219
+        socket_path = "/foo/bar/sock_7IszXMQ1"
         enable_tag_override = true
         check = {
             id = "qmfeO5if"
@@ -561,6 +563,7 @@ services = [
             destination_service_id = "6L6BVfgH-id"
             local_service_address = "127.0.0.2"
             local_service_port = 23759
+            local_service_socket_path = "/foo/bar/local"
             config {
                 cedGGtZf = "pWrUNiWw"
             }
@@ -578,6 +581,8 @@ services = [
                     destination_name = "KSd8HsRl"
                     local_bind_port = 11884
                     local_bind_address = "127.24.88.0"
+                    local_bind_socket_path = "/foo/bar/upstream"
+                    local_bind_socket_mode = "0600"
                 },
             ]
             expose {

agent/config/testdata/full-config.json

@@ -358,6 +358,7 @@
     "address": "cOlSOhbp",
     "token": "msy7iWER",
     "port": 24237,
+    "socket_path": "/tmp/rc78ap",
     "weights": {
       "passing": 100,
       "warning": 1
@@ -452,6 +453,7 @@
       "address": "9RhqPSPB",
       "token": "myjKJkWH",
       "port": 72219,
+      "socket_path":"/foo/bar/sock_7IszXMQ1",
       "enable_tag_override": true,
       "check": {
         "id": "qmfeO5if",
@@ -561,6 +563,7 @@
         "destination_service_name": "6L6BVfgH",
         "local_service_address": "127.0.0.2",
         "local_service_port": 23759,
+        "local_service_socket_path": "/foo/bar/local",
         "expose": {
           "checks": true,
           "paths": [
@@ -589,7 +592,9 @@
             "destination_namespace": "9nakw0td",
             "destination_type": "prepared_query",
             "local_bind_address": "127.24.88.0",
-            "local_bind_port": 11884
+            "local_bind_port": 11884,
+            "local_bind_socket_path":  "/foo/bar/upstream",
+            "local_bind_socket_mode":  "0600"
           }
         ]
       }

agent/proxycfg/manager_test.go

@@ -235,6 +235,7 @@ func TestManager_BasicLifecycle(t *testing.T) {
 						UpstreamConfig: map[string]*structs.Upstream{
 							upstreams[0].Identifier(): &upstreams[0],
 							upstreams[1].Identifier(): &upstreams[1],
+							upstreams[2].Identifier(): &upstreams[2],
 						},
 					},
 					PreparedQueryEndpoints: map[string]structs.CheckServiceNodes{},
@@ -290,6 +291,7 @@ func TestManager_BasicLifecycle(t *testing.T) {
 						UpstreamConfig: map[string]*structs.Upstream{
 							upstreams[0].Identifier(): &upstreams[0],
 							upstreams[1].Identifier(): &upstreams[1],
+							upstreams[2].Identifier(): &upstreams[2],
 						},
 					},
 					PreparedQueryEndpoints: map[string]structs.CheckServiceNodes{},

agent/proxycfg/state.go

@@ -1709,6 +1709,8 @@ func (s *state) handleUpdateIngressGateway(u cache.UpdateEvent, snap *ConfigSnap
 	return nil
 }
 
+// Note: Ingress gateways are always bound to ports and never unix sockets.
+// This means LocalBindPort is the only possibility
 func makeUpstream(g *structs.GatewayService) structs.Upstream {
 	upstream := structs.Upstream{
 		DestinationName:      g.Service.Name,

agent/proxycfg/testing.go

@@ -1639,6 +1639,7 @@ func testConfigSnapshotIngressGateway(
 	additionalEntries ...structs.ConfigEntry,
 ) *ConfigSnapshot {
 	roots, leaf := TestCerts(t)
+
 	snap := &ConfigSnapshot{
 		Kind:       structs.ServiceKindIngressGateway,
 		Service:    "ingress-gateway",

agent/sidecar_service.go

@@ -98,11 +98,20 @@ func (a *Agent) sidecarServiceFromNodeService(ns *structs.NodeService, token str
 	if sidecar.Proxy.DestinationServiceID == "" {
 		sidecar.Proxy.DestinationServiceID = ns.ID
 	}
-	if sidecar.Proxy.LocalServiceAddress == "" {
-		sidecar.Proxy.LocalServiceAddress = "127.0.0.1"
-	}
-	if sidecar.Proxy.LocalServicePort < 1 {
-		sidecar.Proxy.LocalServicePort = ns.Port
+
+	// Fill defaults from NodeService if none of the address components are present.
+	// This really argues for a refactoring to a more generalized 'address' concept.
+	if sidecar.Proxy.LocalServiceSocketPath == "" && (sidecar.Proxy.LocalServiceAddress == "" || sidecar.Proxy.LocalServicePort < 1) {
+		if ns.SocketPath != "" {
+			sidecar.Proxy.LocalServiceSocketPath = ns.SocketPath
+		} else {
+			if sidecar.Proxy.LocalServiceAddress == "" {
+				sidecar.Proxy.LocalServiceAddress = "127.0.0.1"
+			}
+			if sidecar.Proxy.LocalServicePort < 1 {
+				sidecar.Proxy.LocalServicePort = ns.Port
+			}
+		}
 	}
 
 	// Allocate port if needed (min and max inclusive).