Allow configuration of upstream connection limits in Envoy

agent/xds/clusters.go

@@ -234,6 +234,9 @@ func (s *Server) makeUpstreamClusterForPreparedQuery(upstream structs.Upstream,
 					},
 				},
 			},
+			CircuitBreakers: &envoycluster.CircuitBreakers{
+				Thresholds: makeThresholdsIfNeeded(cfg.Limits),
+			},
 			// Having an empty config enables outlier detection with default config.
 			OutlierDetection: &envoycluster.OutlierDetection{},
 		}
@@ -339,6 +342,9 @@ func (s *Server) makeUpstreamClustersForDiscoveryChain(
 					},
 				},
 			},
+			CircuitBreakers: &envoycluster.CircuitBreakers{
+				Thresholds: makeThresholdsIfNeeded(cfg.Limits),
+			},
 			// Having an empty config enables outlier detection with default config.
 			OutlierDetection: &envoycluster.OutlierDetection{},
 		}
@@ -449,3 +455,27 @@ func (s *Server) makeMeshGatewayCluster(clusterName string, cfgSnap *proxycfg.Co
 		OutlierDetection: &envoycluster.OutlierDetection{},
 	}, nil
 }
+
+func makeThresholdsIfNeeded(limits UpstreamLimits) []*envoycluster.CircuitBreakers_Thresholds {
+	var empty UpstreamLimits
+	// Make sure to not create any thresholds when passed the zero-value in order
+	// to rely on Envoy defaults
+	if limits == empty {
+		return nil
+	}
+
+	threshold := &envoycluster.CircuitBreakers_Thresholds{}
+	// Likewise, make sure to not set any threshold values on the zero-value in
+	// order to rely on Envoy defaults
+	if limits.MaxConnections != nil {
+		threshold.MaxConnections = makeUint32Value(*limits.MaxConnections)
+	}
+	if limits.MaxPendingRequests != nil {
+		threshold.MaxPendingRequests = makeUint32Value(*limits.MaxPendingRequests)
+	}
+	if limits.MaxConcurrentRequests != nil {
+		threshold.MaxRequests = makeUint32Value(*limits.MaxConcurrentRequests)
+	}
+
+	return []*envoycluster.CircuitBreakers_Thresholds{threshold}
+}

agent/xds/clusters_test.go

@@ -111,6 +111,59 @@ func TestClustersFromSnapshot(t *testing.T) {
 				snap.Proxy.Upstreams[0].Config["connect_timeout_ms"] = 2345
 			},
 		},
+		{
+			name:   "custom-limits-max-connections-only",
+			create: proxycfg.TestConfigSnapshot,
+			setup: func(snap *proxycfg.ConfigSnapshot) {
+				for i := range snap.Proxy.Upstreams {
+					// We check if Config is nil because the prepared_query upstream is
+					// initialized without a Config map. Use Upstreams[i] syntax to
+					// modify the actual ConfigSnapshot instead of copying the Upstream
+					// in the range.
+					if snap.Proxy.Upstreams[i].Config == nil {
+						snap.Proxy.Upstreams[i].Config = map[string]interface{}{}
+					}
+
+					snap.Proxy.Upstreams[i].Config["limits"] = map[string]interface{}{
+						"max_connections": 500,
+					}
+				}
+			},
+		},
+		{
+			name:   "custom-limits-set-to-zero",
+			create: proxycfg.TestConfigSnapshot,
+			setup: func(snap *proxycfg.ConfigSnapshot) {
+				for i := range snap.Proxy.Upstreams {
+					if snap.Proxy.Upstreams[i].Config == nil {
+						snap.Proxy.Upstreams[i].Config = map[string]interface{}{}
+					}
+
+					snap.Proxy.Upstreams[i].Config["limits"] = map[string]interface{}{
+						"max_connections":         0,
+						"max_pending_requests":    0,
+						"max_concurrent_requests": 0,
+					}
+				}
+			},
+		},
+		{
+			name:   "custom-limits",
+			create: proxycfg.TestConfigSnapshot,
+			setup: func(snap *proxycfg.ConfigSnapshot) {
+				for i := range snap.Proxy.Upstreams {
+					if snap.Proxy.Upstreams[i].Config == nil {
+						snap.Proxy.Upstreams[i].Config = map[string]interface{}{}
+					}
+
+					snap.Proxy.Upstreams[i].Config["limits"] = map[string]interface{}{
+						"max_connections":         500,
+						"max_pending_requests":    600,
+						"max_concurrent_requests": 700,
+					}
+				}
+			},
+		},
 		{
 			name:   "connect-proxy-with-chain",
 			create: proxycfg.TestConfigSnapshotDiscoveryChain,
@@ -312,6 +365,9 @@ func expectClustersJSONResources(t *testing.T, snap *proxycfg.ConfigSnapshot, to
 				},
 				"outlierDetection": {
 
+				},
+				"circuitBreakers": {
+
 				},
 				"altStatName": "db.default.dc1.internal.11111111-2222-3333-4444-555555555555.consul",
 				"commonLbConfig": {
@@ -334,6 +390,9 @@ func expectClustersJSONResources(t *testing.T, snap *proxycfg.ConfigSnapshot, to
 				},
 				"outlierDetection": {
 
+				},
+				"circuitBreakers": {
+
 				},
 				"connectTimeout": "5s",
 				"tlsContext": ` + expectedUpstreamTLSContextJSON(t, snap, "geo-cache.default.dc1.query.11111111-2222-3333-4444-555555555555.consul") + `

agent/xds/config.go

@@ -104,6 +104,25 @@ func ParseMeshGatewayConfig(m map[string]interface{}) (MeshGatewayConfig, error)
 	return cfg, err
 }
 
+// UpstreamLimits describes the limits that are associated with a specific
+// upstream of a service instance.
+type UpstreamLimits struct {
+	// MaxConnections is the maximum number of connections the local proxy can
+	// make to the upstream service.
+	MaxConnections *int `mapstructure:"max_connections"`
+
+	// MaxPendingRequests is the maximum number of requests that will be queued
+	// waiting for an available connection. This is mostly applicable to HTTP/1.1
+	// clusters since all HTTP/2 requests are streamed over a single
+	// connection.
+	MaxPendingRequests *int `mapstructure:"max_pending_requests"`
+
+	// MaxConcurrentRequests is the maximum number of in-flight requests that will be allowed
+	// to the upstream cluster at a point in time. This is mostly applicable to HTTP/2
+	// clusters since all HTTP/1.1 requests are limited by MaxConnections.
+	MaxConcurrentRequests *int `mapstructure:"max_concurrent_requests"`
+}
+
 // UpstreamConfig describes the keys we understand from
 // Connect.Proxy.Upstream[*].Config.
 type UpstreamConfig struct {
@@ -131,6 +150,10 @@ type UpstreamConfig struct {
 	// ConnectTimeoutMs is the number of milliseconds to timeout making a new
 	// connection to this upstream. Defaults to 5000 (5 seconds) if not set.
 	ConnectTimeoutMs int `mapstructure:"connect_timeout_ms"`
+
+	// Limits are the set of limits that are applied to the proxy for a specific upstream of a
+	// service instance.
+	Limits UpstreamLimits `mapstructure:"limits"`
 }
 
 func ParseUpstreamConfigNoDefaults(m map[string]interface{}) (UpstreamConfig, error) {

agent/xds/config_test.go

@@ -205,6 +205,44 @@ func TestParseUpstreamConfig(t *testing.T) {
 				Protocol:         "tcp",
 			},
 		},
+		{
+			name: "connect limits map",
+			input: map[string]interface{}{
+				"limits": map[string]interface{}{
+					"max_connections":         50,
+					"max_pending_requests":    60,
+					"max_concurrent_requests": 70,
+				},
+			},
+			want: UpstreamConfig{
+				ConnectTimeoutMs: 5000,
+				Protocol:         "tcp",
+				Limits: UpstreamLimits{
+					MaxConnections:        intPointer(50),
+					MaxPendingRequests:    intPointer(60),
+					MaxConcurrentRequests: intPointer(70),
+				},
+			},
+		},
+		{
+			name: "connect limits map zero",
+			input: map[string]interface{}{
+				"limits": map[string]interface{}{
+					"max_connections":         0,
+					"max_pending_requests":    0,
+					"max_concurrent_requests": 0,
+				},
+			},
+			want: UpstreamConfig{
+				ConnectTimeoutMs: 5000,
+				Protocol:         "tcp",
+				Limits: UpstreamLimits{
+					MaxConnections:        intPointer(0),
+					MaxPendingRequests:    intPointer(0),
+					MaxConcurrentRequests: intPointer(0),
+				},
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -214,3 +252,7 @@ func TestParseUpstreamConfig(t *testing.T) {
 		})
 	}
 }
+
+func intPointer(i int) *int {
+	return &i
+}

agent/xds/testdata/clusters/connect-proxy-with-chain-and-failover.golden

@@ -14,6 +14,9 @@
         }
       },
       "connectTimeout": "33s",
+      "circuitBreakers": {
+
+      },
       "tlsContext": {
         "commonTlsContext": {
           "tlsParams": {
@@ -58,6 +61,9 @@
         }
       },
       "connectTimeout": "5s",
+      "circuitBreakers": {
+
+      },
       "tlsContext": {
         "commonTlsContext": {
           "tlsParams": {

agent/xds/testdata/clusters/connect-proxy-with-chain-and-overrides.golden

@@ -14,6 +14,9 @@
         }
       },
       "connectTimeout": "66s",
+      "circuitBreakers": {
+
+      },
       "tlsContext": {
         "commonTlsContext": {
           "tlsParams": {
@@ -61,6 +64,9 @@
         }
       },
       "connectTimeout": "5s",
+      "circuitBreakers": {
+
+      },
       "tlsContext": {
         "commonTlsContext": {
           "tlsParams": {

agent/xds/testdata/clusters/connect-proxy-with-chain-external-sni.golden

@@ -14,6 +14,9 @@
         }
       },
       "connectTimeout": "33s",
+      "circuitBreakers": {
+
+      },
       "tlsContext": {
         "commonTlsContext": {
           "tlsParams": {
@@ -58,6 +61,9 @@
         }
       },
       "connectTimeout": "5s",
+      "circuitBreakers": {
+
+      },
       "tlsContext": {
         "commonTlsContext": {
           "tlsParams": {

agent/xds/testdata/clusters/connect-proxy-with-chain.golden

@@ -14,6 +14,9 @@
         }
       },
       "connectTimeout": "33s",
+      "circuitBreakers": {
+
+      },
       "tlsContext": {
         "commonTlsContext": {
           "tlsParams": {
@@ -58,6 +61,9 @@
         }
       },
       "connectTimeout": "5s",
+      "circuitBreakers": {
+
+      },
       "tlsContext": {
         "commonTlsContext": {
           "tlsParams": {

agent/xds/testdata/clusters/connect-proxy-with-tcp-chain-double-failover-through-local-gateway-triggered.golden

@@ -14,6 +14,9 @@
         }
       },
       "connectTimeout": "33s",
+      "circuitBreakers": {
+
+      },
       "tlsContext": {
         "commonTlsContext": {
           "tlsParams": {
@@ -58,6 +61,9 @@
         }
       },
       "connectTimeout": "5s",
+      "circuitBreakers": {
+
+      },
       "tlsContext": {
         "commonTlsContext": {
           "tlsParams": {

agent/xds/testdata/clusters/connect-proxy-with-tcp-chain-double-failover-through-local-gateway.golden

@@ -14,6 +14,9 @@
         }
       },
       "connectTimeout": "33s",
+      "circuitBreakers": {
+
+      },
       "tlsContext": {
         "commonTlsContext": {
           "tlsParams": {
@@ -58,6 +61,9 @@
         }
       },
       "connectTimeout": "5s",
+      "circuitBreakers": {
+
+      },
       "tlsContext": {
         "commonTlsContext": {
           "tlsParams": {

agent/xds/testdata/clusters/connect-proxy-with-tcp-chain-double-failover-through-remote-gateway-triggered.golden

@@ -14,6 +14,9 @@
         }
       },
       "connectTimeout": "33s",
+      "circuitBreakers": {
+
+      },
       "tlsContext": {
         "commonTlsContext": {
           "tlsParams": {
@@ -58,6 +61,9 @@
         }
       },
       "connectTimeout": "5s",
+      "circuitBreakers": {
+
+      },
       "tlsContext": {
         "commonTlsContext": {
           "tlsParams": {

agent/xds/testdata/clusters/connect-proxy-with-tcp-chain-double-failover-through-remote-gateway.golden

@@ -14,6 +14,9 @@
         }
       },
       "connectTimeout": "33s",
+      "circuitBreakers": {
+
+      },
       "tlsContext": {
         "commonTlsContext": {
           "tlsParams": {
@@ -58,6 +61,9 @@
         }
       },
       "connectTimeout": "5s",
+      "circuitBreakers": {
+
+      },
       "tlsContext": {
         "commonTlsContext": {
           "tlsParams": {

agent/xds/testdata/clusters/connect-proxy-with-tcp-chain-failover-through-local-gateway-triggered.golden

@@ -14,6 +14,9 @@
         }
       },
       "connectTimeout": "33s",
+      "circuitBreakers": {
+
+      },
       "tlsContext": {
         "commonTlsContext": {
           "tlsParams": {
@@ -58,6 +61,9 @@
         }
       },
       "connectTimeout": "5s",
+      "circuitBreakers": {
+
+      },
       "tlsContext": {
         "commonTlsContext": {
           "tlsParams": {