feat(aigw): replace custom admin monitor with one in func-e (#1341)

**Description**

This replaces our custom envoy admin monitor code now that func-e has
the same functionality.

**Related Issues/PRs (if applicable)**

functionality upstreamed (and more) here
tetratelabs/func-e#483

---------

Signed-off-by: Adrian Cole <[email protected]>

Author: codefromthecrypt

.github/workflows/build_and_test.yaml

@@ -363,7 +363,8 @@ jobs:
           FUNC_E_HOME: /tmp/envoy-gateway # hard-coded directory in EG
       - name: Install Goose
         env:
-          GOOSE_VERSION: v1.8.0
+          GOOSE_VERSION: v1.10.0
+          OS: Linux
         run: |
           curl -fsSL https://github.com/block/goose/releases/download/stable/download_cli.sh | CONFIGURE=false bash
       - env:

cmd/aigw/healthcheck.go

@@ -11,7 +11,7 @@ import (
 	"log/slog"
 	"time"
 
-	"github.com/envoyproxy/ai-gateway/internal/aigw"
+	"github.com/tetratelabs/func-e/experimental/admin"
 )
 
 // healthcheck performs looks up the Envoy subprocess, gets its admin port,
@@ -27,13 +27,12 @@ func healthcheck(ctx context.Context, _, stderr io.Writer) error {
 }
 
 func doHealthcheck(ctx context.Context, aigwPid int, logger *slog.Logger) error {
-	envoyAdmin, err := aigw.NewEnvoyAdminClient(ctx, aigwPid, 0)
-	if err != nil {
+	if adminClient, err := admin.NewAdminClient(ctx, aigwPid); err != nil {
 		logger.Error("Failed to find Envoy admin server", "error", err)
 		return err
-	} else if err = envoyAdmin.IsReady(ctx); err != nil {
-		logger.Error("Envoy admin server is not ready", "adminPort", envoyAdmin.Port(), "error", err)
+	} else if err = adminClient.IsReady(ctx); err != nil {
+		logger.Error("Envoy admin server is not ready", "adminPort", adminClient.Port(), "error", err)
 		return err
 	}
-	return err
+	return nil
 }

cmd/aigw/healthcheck_test.go

@@ -8,19 +8,16 @@ package main
 import (
 	"bytes"
 	"context"
-	"fmt"
+	"io"
 	"log/slog"
-	"net/http"
-	"net/http/httptest"
-	"net/url"
 	"os"
-	"os/exec"
-	"path/filepath"
-	"strconv"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/require"
+	func_e "github.com/tetratelabs/func-e"
+	"github.com/tetratelabs/func-e/api"
+	"github.com/tetratelabs/func-e/experimental/admin"
 )
 
 func Test_healthcheck(t *testing.T) {
@@ -38,38 +35,32 @@ func Test_healthcheck(t *testing.T) {
 	})
 
 	t.Run("returns nil when ready", func(t *testing.T) {
-		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-			require.Equal(t, "/ready", r.URL.Path)
-			w.WriteHeader(http.StatusOK)
-			_, _ = w.Write([]byte("live"))
-		}))
-		defer server.Close()
-
-		u, err := url.Parse(server.URL)
-		require.NoError(t, err)
-		port, err := strconv.Atoi(u.Port())
-		require.NoError(t, err)
-
-		adminFile := filepath.Join(t.TempDir(), "admin-address.txt")
-		require.NoError(t, os.WriteFile(adminFile, []byte(fmt.Sprintf("127.0.0.1:%d", port)), 0o600))
-
 		ctx, cancel := context.WithCancel(t.Context())
 		defer cancel()
 
-		cmdStr := fmt.Sprintf("sleep 30 && echo -- --admin-address-path %s", adminFile)
-		cmd := exec.CommandContext(ctx, "sh", "-c", cmdStr)
-		require.NoError(t, cmd.Start())
-		defer func() {
-			_ = cmd.Process.Kill()
-			_, _ = cmd.Process.Wait()
-		}()
+		var healthCheckErr error
+		var log bytes.Buffer
 
-		time.Sleep(100 * time.Millisecond)
+		// Even though AdminClient.IsReady exists, we don't have it injected in
+		// Docker. This intentionally ignores the parameter.
+		startupHook := func(ctx context.Context, _ admin.AdminClient, _ string) error {
+			logger := slog.New(slog.NewTextHandler(&log, nil))
+			healthCheckErr = doHealthcheck(ctx, pid, logger)
+			// Cancel immediately to stop Envoy and complete test quickly
+			cancel()
+			return nil
+		}
 
-		var buf bytes.Buffer
-		logger := slog.New(slog.NewTextHandler(&buf, nil))
-		err = doHealthcheck(t.Context(), pid, logger)
+		// Run with minimal Envoy config
+		err := func_e.Run(ctx, []string{
+			"--config-yaml",
+			"admin: {address: {socket_address: {address: '127.0.0.1', port_value: 0}}}",
+		}, api.Out(io.Discard), api.EnvoyOut(io.Discard), api.EnvoyErr(io.Discard), admin.WithStartupHook(startupHook))
+
+		// Expect nil error since Run returns nil on context cancellation (documented behavior)
 		require.NoError(t, err)
-		require.Empty(t, buf)
+
+		require.NoError(t, healthCheckErr)
+		require.Empty(t, log)
 	})
 }

cmd/aigw/run.go

@@ -19,11 +19,11 @@ import (
 	"strings"
 	"time"
 
-	egv1a1 "github.com/envoyproxy/gateway/api/v1alpha1"
 	"github.com/envoyproxy/gateway/cmd/envoy-gateway/root"
 	egextension "github.com/envoyproxy/gateway/proto/extension"
 	"github.com/go-logr/logr"
 	"github.com/tetratelabs/func-e/api"
+	"github.com/tetratelabs/func-e/experimental/admin"
 	"github.com/tetratelabs/func-e/experimental/middleware"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/health/grpc_health_v1"
@@ -35,7 +35,6 @@ import (
 	gwapiv1 "sigs.k8s.io/gateway-api/apis/v1"
 	"sigs.k8s.io/yaml"
 
-	"github.com/envoyproxy/ai-gateway/internal/aigw"
 	"github.com/envoyproxy/ai-gateway/internal/controller"
 	"github.com/envoyproxy/ai-gateway/internal/extensionserver"
 	"github.com/envoyproxy/ai-gateway/internal/filterapi"
@@ -92,9 +91,6 @@ type runOpts struct {
 //  1. ${os.TempDir}/envoy-gateway-config.yaml: This contains the configuration for the Envoy Gateway agent to run, derived from envoyGatewayConfig.
 //  2. ${os.TempDir}/envoy-ai-gateway-resources: This will contain the EG resource generated by the translation and deployed by EG.
 func run(ctx context.Context, c cmdRun, o runOpts, stdout, stderr io.Writer) error {
-	redirectEnvoyStdio := newEnvoyRunMiddleware(stdout, stderr)
-	ctx = middleware.WithRunMiddleware(ctx, redirectEnvoyStdio)
-
 	start := time.Now()
 	var debugLogger *slog.Logger
 	if c.Debug {
@@ -161,7 +157,7 @@ func run(ctx context.Context, c cmdRun, o runOpts, stdout, stderr io.Writer) err
 	if err != nil {
 		return err
 	}
-	fakeClient, extProxDone, envoyAdminPort, listenerPort, err := runCtx.writeEnvoyResourcesAndRunExtProc(ctx, aiGatewayResourcesYaml)
+	fakeClient, extProxDone, listenerPort, err := runCtx.writeEnvoyResourcesAndRunExtProc(ctx, aiGatewayResourcesYaml)
 	if err != nil {
 		return fmt.Errorf("failed to write envoy resources and run extproc: %w", err)
 	}
@@ -170,6 +166,10 @@ func run(ctx context.Context, c cmdRun, o runOpts, stdout, stderr io.Writer) err
 		return fmt.Errorf("failed to write file %s: %w", resourceYamlPath, err)
 	}
 
+	// Set up middleware with startup hook now that we know listenerPort
+	redirectEnvoyStdio := newEnvoyRunMiddleware(start, listenerPort, stdout, stderr)
+	ctx = middleware.WithRunMiddleware(ctx, redirectEnvoyStdio)
+
 	lis, err := net.Listen("tcp", "localhost:1061")
 	if err != nil {
 		return fmt.Errorf("failed to listen: %w", err)
@@ -214,65 +214,40 @@ func run(ctx context.Context, c cmdRun, o runOpts, stdout, stderr io.Writer) err
 	}
 	server.SetArgs([]string{"server", "--config-path", egConfigPath})
 
-	// Start a monitoring goroutine to poll Envoy's readiness. This starts
-	// before the server to ensure we don't miss the readiness window.
-	go func() {
-		envoyAdmin, err := aigw.NewEnvoyAdminClient(ctx, os.Getpid(), envoyAdminPort)
-		if err != nil {
-			debugLogger.Error("Failed to find Envoy admin server", "error", err)
-			serverCancel() // Likely a crashed envoy process
-			return
-		}
-		debugLogger.Info("Found Envoy admin server", "adminPort", envoyAdmin.Port())
-		if err = pollEnvoyReady(ctx, debugLogger, envoyAdmin, 100*time.Millisecond); err != nil {
-			return
-		}
-		c.AdminPort = envoyAdminPort // write back for testing
-		// Print a status message without any timestamp formatting
-		startDuration := time.Since(start).Round(100 * time.Millisecond)
-		_, _ = fmt.Fprintf(stderr, "Envoy AI Gateway listening on http://localhost:%d (admin http://localhost:%d) after %v\n",
-			listenerPort, envoyAdmin.Port(), startDuration)
-	}()
-
 	// Start the gateway server. This will block until the server is stopped.
+	// The startup hook (configured via middleware) will print the status message when Envoy is ready.
 	if err := server.ExecuteContext(serverCtx); err != nil {
 		return fmt.Errorf("failed to execute server: %w", err)
 	}
 	return extProcErr
 }
 
-// newEnvoyRunMiddleware sets options for running Envoy and returns a context to
-// propagate them to func-e/
-func newEnvoyRunMiddleware(stdout, stderr io.Writer) func(next api.RunFunc) api.RunFunc {
+// newEnvoyRunMiddleware sets options for running Envoy and returns a middleware
+// that configures Envoy I/O and sets up a startup hook to print the ready message.
+func newEnvoyRunMiddleware(start time.Time, listenerPort int, stdout, stderr io.Writer) func(next api.RunFunc) api.RunFunc {
+	// Define startup hook that will be called when Envoy admin is ready
+	startupHook := func(_ context.Context, adminClient admin.AdminClient, _ string) error {
+		// Print a status message without any timestamp formatting
+		startDuration := time.Since(start).Round(100 * time.Millisecond)
+		_, _ = fmt.Fprintf(stderr, "Envoy AI Gateway listening on http://localhost:%d (admin http://localhost:%d) after %v\n",
+			listenerPort, adminClient.Port(), startDuration)
+		return nil
+	}
+
 	// aigw is primarily an Envoy controller, so ensure its output is visible
-	overrides := []api.RunOption{api.EnvoyOut(stdout), api.EnvoyErr(stderr)}
+	overrides := []api.RunOption{
+		api.EnvoyOut(stdout),
+		api.EnvoyErr(stderr),
+		admin.WithStartupHook(startupHook),
+	}
+
 	return func(next api.RunFunc) api.RunFunc {
 		return func(ctx context.Context, args []string, options ...api.RunOption) error {
 			return next(ctx, args, append(options, overrides...)...)
 		}
 	}
 }
 
-// pollEnvoyReady polls Envoy's readiness until it is ready or the context is done.
-func pollEnvoyReady(ctx context.Context, l *slog.Logger, envoyAdmin aigw.EnvoyAdminClient, interval time.Duration) error {
-	t := time.NewTicker(interval)
-	defer t.Stop()
-
-	for {
-		select {
-		case <-ctx.Done():
-			return ctx.Err()
-		case <-t.C:
-			if err := envoyAdmin.IsReady(ctx); err == nil {
-				l.Info("Envoy is ready!")
-				return nil
-			} else {
-				l.Info("Waiting for Envoy to be ready...", "err", err)
-			}
-		}
-	}
-}
-
 // recreateDir removes the directory at the given path and creates a new one.
 func recreateDir(path string) error {
 	err := os.RemoveAll(path)
@@ -288,33 +263,33 @@ func recreateDir(path string) error {
 
 // writeEnvoyResourcesAndRunExtProc reads all resources from the given string, writes them to the output file, and runs
 // external processes for EnvoyExtensionPolicy resources.
-func (runCtx *runCmdContext) writeEnvoyResourcesAndRunExtProc(ctx context.Context, original string) (client.Client, <-chan error, int, int, error) {
-	aigwRoutes, mcpRoutes, aigwBackends, backendSecurityPolicies, backendTLSPolicies, gateways, secrets, envoyProxies, err := collectObjects(original, runCtx.envoyGatewayResourcesOut, runCtx.stderrLogger)
+func (runCtx *runCmdContext) writeEnvoyResourcesAndRunExtProc(ctx context.Context, original string) (client.Client, <-chan error, int, error) {
+	aigwRoutes, mcpRoutes, aigwBackends, backendSecurityPolicies, backendTLSPolicies, gateways, secrets, _, err := collectObjects(original, runCtx.envoyGatewayResourcesOut, runCtx.stderrLogger)
 	if err != nil {
-		return nil, nil, 0, 0, fmt.Errorf("error collecting: %w", err)
+		return nil, nil, 0, fmt.Errorf("error collecting: %w", err)
 	}
 	if len(gateways) > 1 {
-		return nil, nil, 0, 0, fmt.Errorf("multiple gateways are not supported: %s", gateways[0].Name)
+		return nil, nil, 0, fmt.Errorf("multiple gateways are not supported: %s", gateways[0].Name)
 	}
 	for _, bsp := range backendSecurityPolicies {
 		spec := bsp.Spec
 		if spec.AWSCredentials != nil && spec.AWSCredentials.OIDCExchangeToken != nil {
 			// TODO: We can make it work by generalizing the rotation logic.
-			return nil, nil, 0, 0, fmt.Errorf("OIDC exchange token is not supported: %s", bsp.Name)
+			return nil, nil, 0, fmt.Errorf("OIDC exchange token is not supported: %s", bsp.Name)
 		}
 	}
 
 	// Do the substitution for the secrets.
 	for _, s := range secrets {
 		if err = runCtx.rewriteSecretWithAnnotatedLocation(s); err != nil {
-			return nil, nil, 0, 0, fmt.Errorf("failed to rewrite secret %s: %w", s.Name, err)
+			return nil, nil, 0, fmt.Errorf("failed to rewrite secret %s: %w", s.Name, err)
 		}
 	}
 
 	var secretList *corev1.SecretList
 	fakeClient, _fakeClientSet, httpRoutes, eps, httpRouteFilters, backends, secretList, backendTrafficPolicies, securityPolicies, err := translateCustomResourceObjects(ctx, aigwRoutes, mcpRoutes, aigwBackends, backendSecurityPolicies, backendTLSPolicies, gateways, secrets, runCtx.stderrLogger)
 	if err != nil {
-		return nil, nil, 0, 0, fmt.Errorf("error translating: %w", err)
+		return nil, nil, 0, fmt.Errorf("error translating: %w", err)
 	}
 	runCtx.fakeClientSet = _fakeClientSet
 
@@ -338,7 +313,7 @@ func (runCtx *runCmdContext) writeEnvoyResourcesAndRunExtProc(ctx context.Contex
 	}
 	gw := gateways[0]
 	if len(gw.Spec.Listeners) == 0 {
-		return nil, nil, 0, 0, fmt.Errorf("gateway %s has no listeners configured", gw.Name)
+		return nil, nil, 0, fmt.Errorf("gateway %s has no listeners configured", gw.Name)
 	}
 	runCtx.mustClearSetOwnerReferencesAndStatusAndWriteObj(&gw.TypeMeta, gw)
 	for _, ep := range eps.Items {
@@ -349,20 +324,20 @@ func (runCtx *runCmdContext) writeEnvoyResourcesAndRunExtProc(ctx context.Contex
 		Secrets("").Get(ctx,
 		controller.FilterConfigSecretPerGatewayName(gw.Name, gw.Namespace), metav1.GetOptions{})
 	if err != nil {
-		return nil, nil, 0, 0, fmt.Errorf("failed to get filter config secret: %w", err)
+		return nil, nil, 0, fmt.Errorf("failed to get filter config secret: %w", err)
 	}
 
 	rawConfig, ok := filterConfigSecret.StringData[controller.FilterConfigKeyInSecret]
 	if !ok {
-		return nil, nil, 0, 0, fmt.Errorf("failed to get filter config from secret: %w", err)
+		return nil, nil, 0, fmt.Errorf("failed to get filter config from secret: %w", err)
 	}
 	var fc filterapi.Config
 	if err = yaml.Unmarshal([]byte(rawConfig), &fc); err != nil {
-		return nil, nil, 0, 0, fmt.Errorf("failed to unmarshal filter config: %w", err)
+		return nil, nil, 0, fmt.Errorf("failed to unmarshal filter config: %w", err)
 	}
 	runCtx.stderrLogger.Info("Running external process", "config", fc)
 	done := runCtx.mustStartExtProc(ctx, &fc)
-	return fakeClient, done, runCtx.tryFindEnvoyAdminPort(gw, envoyProxies), runCtx.tryFindEnvoyListenerPort(gw), nil
+	return fakeClient, done, runCtx.tryFindEnvoyListenerPort(gw), nil
 }
 
 // mustStartExtProc starts the external process with the given working directory, port, and filter configuration.
@@ -486,54 +461,6 @@ func (runCtx *runCmdContext) tryFindEnvoyListenerPort(gw *gwapiv1.Gateway) int {
 	return int(gw.Spec.Listeners[0].Port)
 }
 
-// tryFindEnvoyAdminPort tries to find the port where the Envoy Admin interface is listening to.
-// By default, Envoy Gateway assigns a random port to the Envoy Admin interface, and we may not be able to find it. This method
-// attempts to find an EnvoyProxy instance attached to the standalone Gateway, and reads the bootstrap config to check if there
-// is a custom port configured for the admin interface.
-// If there is no EnvoyProxy or the admin port is not configured, this returns 0.
-func (runCtx *runCmdContext) tryFindEnvoyAdminPort(gw *gwapiv1.Gateway, proxies []*egv1a1.EnvoyProxy) int {
-	if gw.Spec.Infrastructure == nil ||
-		gw.Spec.Infrastructure.ParametersRef == nil ||
-		gw.Spec.Infrastructure.ParametersRef.Kind != "EnvoyProxy" ||
-		gw.Spec.Infrastructure.ParametersRef.Name == "" {
-		return 0
-	}
-
-	var bootstrap *egv1a1.ProxyBootstrap
-	for _, p := range proxies {
-		if p.Name == gw.Spec.Infrastructure.ParametersRef.Name {
-			bootstrap = p.Spec.Bootstrap
-			break
-		}
-	}
-
-	if bootstrap == nil || bootstrap.Value == nil {
-		return 0
-	}
-
-	type adminSettings struct {
-		Admin struct {
-			Address struct {
-				SocketAddress struct {
-					Address   string `json:"address"`
-					PortValue int    `json:"port_value"`
-				} `json:"socket_address,omitempty"`
-			} `json:"address"`
-		} `json:"admin"`
-	}
-
-	var admin adminSettings
-	if err := yaml.Unmarshal([]byte(*bootstrap.Value), &admin); err != nil {
-		runCtx.stderrLogger.Error("Failed to read EnvoyProxy bootstrap settings", "error", err)
-		return 0
-	}
-
-	if admin.Admin.Address.SocketAddress.Address == "" {
-		return 0
-	}
-	return admin.Admin.Address.SocketAddress.PortValue
-}
-
 func maybeResolveHome(p string) string {
 	if strings.HasPrefix(p, "~/") {
 		home, err := os.UserHomeDir()