WIP GRPC rate limiting

.changelog/533.txt

@@ -0,0 +1,3 @@
+```release-note:bug
+Use Consul Catalog API instead of Agent API to register APIGateway to prevent GRPC rate limits
+```

internal/commands/exec/exec.go

@@ -106,7 +106,7 @@ func RunExec(config ExecConfig) (ret int) {
 		config.GatewayConfig.Host,
 	)
 	if config.isTest {
-		registry = registry.WithTries(1)
+		registry = registry.WithRetries(1)
 	}
 
 	config.Logger.Trace("registering service")

internal/commands/exec/exec_test.go

@@ -371,8 +371,8 @@ func runMockConsulServer(t *testing.T, opts mockConsulOptions) *mockConsulServer
 
 	loginPath := "/v1/acl/login"
 	logoutPath := "/v1/acl/logout"
-	registerPath := "/v1/agent/service/register"
-	deregisterPath := "/v1/agent/service/deregister"
+	registerPath := "/v1/catalog/register"
+	deregisterPath := "/v1/catalog/deregister"
 	leafPath := "/v1/agent/connect/ca/leaf"
 	rootPath := "/v1/agent/connect/ca/roots"
 

internal/commands/server/k8s_e2e_test.go

@@ -378,8 +378,13 @@ func TestGatewayBasic(t *testing.T) {
 					Namespace: e2e.ConsulNamespace(ctx),
 				})
 				if err != nil {
+					fmt.Printf("ERROR: %#v", err)
 					return false
 				}
+				fmt.Printf("SERVICES: %d\n", len(services))
+				for _, service := range services {
+					fmt.Printf("%#v\n", service)
+				}
 				return len(services) == 0
 			}, checkTimeout, checkInterval, "consul service not deregistered in the allotted time")
 
@@ -427,7 +432,7 @@ func TestGatewayBasic(t *testing.T) {
 			require.Eventually(t, func() bool {
 				entries, _, err := client.ConfigEntries().List(api.IngressGateway, queryNamespace)
 				return err == nil && len(entries) == 1 && entries[0].GetName() == gatewayName
-			}, 5*time.Minute, checkInterval, "ingress-gateway config-entry not created in allotted time")
+			}, 90*time.Second, checkInterval, "ingress-gateway config-entry not created in allotted time")
 
 			// De-register Consul service
 			_, err := client.Catalog().Deregister(&api.CatalogDeregistration{
@@ -439,25 +444,15 @@ func TestGatewayBasic(t *testing.T) {
 			require.Eventually(t, func() bool {
 				services, _, err := client.Catalog().Service(gatewayName, "", queryNamespace)
 				return err == nil && len(services) == 0
-			}, 5*time.Minute, checkInterval, "service still returned after de-registering")
+			}, 90*time.Second, checkInterval, "service still returned after de-registering")
 
 			// Delete ingress-gateway config-entry
 			_, err = client.ConfigEntries().Delete(api.IngressGateway, gatewayName, &api.WriteOptions{Namespace: e2e.ConsulNamespace(ctx)})
 			require.NoError(t, err)
 			require.Eventually(t, func() bool {
 				entries, _, err := client.ConfigEntries().List(api.IngressGateway, queryNamespace)
 				return err == nil && len(entries) == 0
-			}, 5*time.Minute, checkInterval, "ingress-gateway config entry still returned after deleting")
-
-			// Check to make sure the controller recreates the service and config-entry in the background.
-			assert.Eventually(t, func() bool {
-				services, _, err := client.Catalog().Service(gatewayName, "", queryNamespace)
-				return err == nil && len(services) == 1
-			}, 5*time.Minute, checkInterval, "service not recreated after delete in allotted time")
-			assert.Eventually(t, func() bool {
-				entry, _, err := client.ConfigEntries().Get(api.IngressGateway, gatewayName, queryNamespace)
-				return err == nil && entry != nil
-			}, 5*time.Minute, checkInterval, "ingress-gateway config-entry not recreated after delete in allotted time")
+			}, 90*time.Second, checkInterval, "ingress-gateway config entry still returned after deleting")
 
 			// Clean up
 			require.NoError(t, resources.Delete(ctx, gw))
@@ -629,6 +624,18 @@ func TestHTTPRouteFlattening(t *testing.T) {
 			err = resources.Create(ctx, route)
 			require.NoError(t, err)
 
+			// Verify HTTPRoute has updated its status
+			check := createConditionsCheck([]meta.Condition{
+				{
+					Type: rstatus.RouteConditionAccepted, Status: "True", Reason: rstatus.RouteConditionReasonAccepted,
+				},
+				{
+					Type: rstatus.RouteConditionResolvedRefs, Status: "True", Reason: rstatus.RouteConditionReasonResolvedRefs,
+				},
+			})
+			require.Eventually(t, httpRouteStatusCheck(ctx, resources, gatewayName, routeOneName, namespace, check), checkTimeout, checkInterval, "route status not set in allotted time")
+			require.Eventually(t, httpRouteStatusCheck(ctx, resources, gatewayName, routeTwoName, namespace, check), checkTimeout, checkInterval, "route status not set in allotted time")
+
 			checkRoute(t, checkPort, "/v2/test", httpResponse{
 				StatusCode: http.StatusOK,
 				Body:       serviceTwo.Name,

internal/consul/auth.go

@@ -25,7 +25,7 @@ type Authenticator struct {
 
 	method          string
 	namespace       string
-	tries           uint64
+	retries         uint64
 	backoffInterval time.Duration
 }
 
@@ -36,13 +36,13 @@ func NewAuthenticator(logger hclog.Logger, consul *api.Client, method, namespace
 		logger:          logger,
 		method:          method,
 		namespace:       namespace,
-		tries:           defaultMaxAttempts,
+		retries:         defaultMaxRetries,
 		backoffInterval: defaultBackoffInterval,
 	}
 }
 
-func (a *Authenticator) WithTries(tries uint64) *Authenticator {
-	a.tries = tries
+func (a *Authenticator) WithRetries(retries uint64) *Authenticator {
+	a.retries = retries
 	return a
 }
 
@@ -58,7 +58,7 @@ func (a *Authenticator) Authenticate(ctx context.Context, service, bearerToken s
 			a.logger.Error("error authenticating", "error", err)
 		}
 		return err
-	}, backoff.WithContext(backoff.WithMaxRetries(backoff.NewConstantBackOff(a.backoffInterval), a.tries), ctx))
+	}, backoff.WithContext(backoff.WithMaxRetries(backoff.NewConstantBackOff(a.backoffInterval), a.retries), ctx))
 	return token, err
 }
 

internal/consul/auth_test.go

@@ -30,7 +30,7 @@ func TestAuthenticate(t *testing.T) {
 		service      string
 		expectedMeta string
 		failures     uint64
-		maxAttempts  uint64
+		maxRetries   uint64
 		fail         bool
 	}{{
 		name:         "success-no-namespace",
@@ -46,25 +46,25 @@ func TestAuthenticate(t *testing.T) {
 		service:      "consul-api-gateway-test",
 		expectedMeta: "consul-api-gateway-test",
 		failures:     3,
-		maxAttempts:  3,
+		maxRetries:   3,
 	}, {
-		name:        "retry-failure",
-		service:     "consul-api-gateway-test",
-		failures:    3,
-		maxAttempts: 2,
-		fail:        true,
+		name:       "retry-failure",
+		service:    "consul-api-gateway-test",
+		failures:   3,
+		maxRetries: 2,
+		fail:       true,
 	}} {
 		t.Run(test.name, func(t *testing.T) {
 			server := runACLServer(t, test.failures)
 			method := gwTesting.RandomString()
 			token := gwTesting.RandomString()
 
-			maxAttempts := defaultMaxAttempts
-			if test.maxAttempts > 0 {
-				maxAttempts = test.maxAttempts
+			maxRetries := defaultMaxRetries
+			if test.maxRetries > 0 {
+				maxRetries = test.maxRetries
 			}
 
-			auth := NewAuthenticator(hclog.NewNullLogger(), server.consul, method, test.namespace).WithTries(maxAttempts)
+			auth := NewAuthenticator(hclog.NewNullLogger(), server.consul, method, test.namespace).WithRetries(maxRetries)
 			auth.backoffInterval = 0
 			consulToken, err := auth.Authenticate(context.Background(), test.service, token)
 

internal/consul/common.go

@@ -8,6 +8,6 @@ import (
 )
 
 const (
-	defaultMaxAttempts     = uint64(30)
+	defaultMaxRetries      = uint64(30)
 	defaultBackoffInterval = 1 * time.Second
 )