cli: fail and exit if ext-proc fails to start (envoyproxy#1134)

**Description**

When using the `aigw run` command, if the extproc fails to start, the
command does not return and hangs forever without giving any feedback to
the user unless the `--debug` flag is used.

This PR captures the extproc start errors and terminates the CLI process
to give proper feedback to the user, printing the error and exiting with
an error code:

```
$ ./out/aigw-darwin-arm64 run
looking up the latest patch for Envoy version 1.35
1.35.0 is already downloaded
starting: /tmp/envoy-gateway/versions/1.35.0/bin/envoy in run directory /tmp/envoy-gateway/runs/1756378634894657000
[2025-08-28 12:57:15.197][3950822][warning][config] [source/server/options_impl_platform_default.cc:9] CPU number provided by HW thread count (instead of cpuset).
2025/08/28 12:57:18 Error running: external processor run error: failed to start config watcher: failed to load initial config: failed to load config: cannot create backend auth handler: cannot retrieve AWS credentials: failed to refresh cached credentials, no EC2 IMDS role found, operation error ec2imds: GetMetadata, exceeded maximum number of attempts, 3, request send failed, Get "http://169.254.169.254/latest/meta-data/iam/security-credentials/": dial tcp 169.254.169.254:80: connect: host is down

$ echo $?
1
```

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

N/A

**Special notes for reviewers (if applicable)**

N/A

---------

Signed-off-by: Ignasi Barrera <[email protected]>

Author: nacx

cmd/aigw/main.go

@@ -44,7 +44,7 @@ type (
 type (
 	subCmdFn[T any] func(context.Context, T, io.Writer, io.Writer) error
 	translateFn     subCmdFn[cmdTranslate]
-	runFn           subCmdFn[cmdRun]
+	runFn           func(context.Context, cmdRun, runOpts, io.Writer, io.Writer) error
 )
 
 func main() {
@@ -84,7 +84,7 @@ func doMain(ctx context.Context, stdout, stderr io.Writer, args []string, exitFn
 			log.Fatalf("Error translating: %v", err)
 		}
 	case "run", "run <path>":
-		err = rf(ctx, c.Run, stdout, stderr)
+		err = rf(ctx, c.Run, runOpts{}, stdout, stderr)
 		if err != nil {
 			log.Fatalf("Error running: %v", err)
 		}

cmd/aigw/main_test.go

@@ -107,12 +107,12 @@ Flags:
 		{
 			name: "run no arg",
 			args: []string{"run"},
-			rf:   func(_ context.Context, _ cmdRun, _, _ io.Writer) error { return nil },
+			rf:   func(context.Context, cmdRun, runOpts, io.Writer, io.Writer) error { return nil },
 		},
 		{
 			name: "run help",
 			args: []string{"run", "--help"},
-			rf:   func(_ context.Context, _ cmdRun, _, _ io.Writer) error { return nil },
+			rf:   func(context.Context, cmdRun, runOpts, io.Writer, io.Writer) error { return nil },
 			expOut: `Usage: aigw run [<path>] [flags]
 
 Run the AI Gateway locally for given configuration.
@@ -133,15 +133,15 @@ Flags:
 		{
 			name: "run show default",
 			args: []string{"run", "--show-default"},
-			rf: func(_ context.Context, c cmdRun, _, _ io.Writer) error {
+			rf: func(_ context.Context, c cmdRun, _ runOpts, _, _ io.Writer) error {
 				require.True(t, c.ShowDefault)
 				return nil
 			},
 		},
 		{
 			name: "run with path",
 			args: []string{"run", "./path"},
-			rf: func(_ context.Context, c cmdRun, _, _ io.Writer) error {
+			rf: func(_ context.Context, c cmdRun, _ runOpts, _, _ io.Writer) error {
 				abs, err := filepath.Abs("./path")
 				require.NoError(t, err)
 				require.Equal(t, abs, c.Path)

cmd/aigw/run.go

@@ -35,22 +35,27 @@ import (
 	"github.com/envoyproxy/ai-gateway/internal/extensionserver"
 )
 
-// This is the default configuration for the AI Gateway when <path> parameter is not given.
-//
-//go:embed ai-gateway-default-resources.yaml
-var aiGatewayDefaultResources string
+var (
+	// This is the default configuration for the AI Gateway when <path> parameter is not given.
+	//
+	//go:embed ai-gateway-default-resources.yaml
+	aiGatewayDefaultResources string
 
-// This is the template for the Envoy Gateway configuration where PLACEHOLDER_TMPDIR will be replaced with the temporary
-// directory where the resources are written to.
-//
-//go:embed envoy-gateway-config.yaml
-var envoyGatewayConfigTemplate string
+	// This is the template for the Envoy Gateway configuration where PLACEHOLDER_TMPDIR will be replaced with the temporary
+	// directory where the resources are written to.
+	//
+	//go:embed envoy-gateway-config.yaml
+	envoyGatewayConfigTemplate string
+)
 
 const (
 	substitutionEnvAnnotationPrefix  = "substitution.aigw.run/env/"
 	substitutionFileAnnotationPrefix = "substitution.aigw.run/file/"
 )
 
+// errExtProcRun is returned when the external processor fails to run.
+var errExtProcRun = fmt.Errorf("external processor run error")
+
 type runCmdContext struct {
 	// isDebug true if the original `agw run` command is run with debug mode. Using this to
 	// set the log level of the external process currently. TODO: maybe simply expose the external process log level
@@ -69,12 +74,18 @@ type runCmdContext struct {
 	fakeClientSet *fake.Clientset
 }
 
+// runOpts are the options for the run command.
+type runOpts struct {
+	// udsPath is the path to the UDS socket used by the AI Gateway extproc.
+	udsPath string
+}
+
 // run starts the AI Gateway locally for a given configuration.
 //
 // This will create a temporary directory and a file:
 //  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, stdout, stderr io.Writer) error {
+func run(ctx context.Context, c cmdRun, o runOpts, stdout, stderr io.Writer) error {
 	if !c.Debug {
 		stderr = io.Discard
 	}
@@ -123,8 +134,11 @@ func run(ctx context.Context, c cmdRun, stdout, stderr io.Writer) error {
 	// Write the Envoy Gateway resources into a file under resourcesTmpdir.
 	resourceYamlPath := filepath.Join(resourcesTmpdir, "config.yaml")
 	stderrLogger.Info("Creating Envoy Gateway resource file", "path", resourceYamlPath)
-	udsPath := filepath.Join(tmpdir, "uds.sock")
-	_ = os.Remove(udsPath)
+	udsPath := o.udsPath
+	if udsPath == "" {
+		udsPath = filepath.Join(tmpdir, "uds.sock")
+		_ = os.Remove(udsPath)
+	}
 
 	// Do the translation of the given AI Gateway resources Yaml into Envoy Gateway resources and write them to the file.
 	resourcesBuf := &bytes.Buffer{}
@@ -133,7 +147,7 @@ func run(ctx context.Context, c cmdRun, stdout, stderr io.Writer) error {
 	if err != nil {
 		return err
 	}
-	fakeClient, err := runCtx.writeEnvoyResourcesAndRunExtProc(ctx, aiGatewayResourcesYaml)
+	fakeClient, extProxDone, err := runCtx.writeEnvoyResourcesAndRunExtProc(ctx, aiGatewayResourcesYaml)
 	if err != nil {
 		return fmt.Errorf("failed to write envoy resources and run extproc: %w", err)
 	}
@@ -150,9 +164,17 @@ func run(ctx context.Context, c cmdRun, stdout, stderr io.Writer) error {
 	extSrv := extensionserver.New(fakeClient, ctrl.Log, udsPath, true)
 	egextension.RegisterEnvoyGatewayExtensionServer(s, extSrv)
 	grpc_health_v1.RegisterHealthServer(s, extSrv)
+
+	serverCtx, serverCancel := context.WithCancel(ctx)
+
+	var extProcErr error
 	go func() {
-		<-ctx.Done()
+		select {
+		case <-ctx.Done():
+		case extProcErr = <-extProxDone:
+		}
 		s.GracefulStop()
+		serverCancel()
 	}()
 	go func() {
 		if err := s.Serve(lis); err != nil {
@@ -177,10 +199,11 @@ func run(ctx context.Context, c cmdRun, stdout, stderr io.Writer) error {
 		server.SetErr(io.Discard)
 	}
 	server.SetArgs([]string{"server", "--config-path", egConfigPath})
-	if err := server.ExecuteContext(ctx); err != nil {
+	if err := server.ExecuteContext(serverCtx); err != nil {
 		return fmt.Errorf("failed to execute server: %w", err)
 	}
-	return nil
+
+	return extProcErr
 }
 
 // readConfig returns config from the given path, substituting ENV variables
@@ -212,32 +235,32 @@ 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, error) {
+func (runCtx *runCmdContext) writeEnvoyResourcesAndRunExtProc(ctx context.Context, original string) (client.Client, <-chan error, error) {
 	aigwRoutes, aigwBackends, backendSecurityPolicies, gateways, secrets, err := collectObjects(original, runCtx.envoyGatewayResourcesOut, runCtx.stderrLogger)
 	if err != nil {
-		return nil, fmt.Errorf("error collecting: %w", err)
+		return nil, nil, fmt.Errorf("error collecting: %w", err)
 	}
 	if len(gateways) > 1 {
-		return nil, fmt.Errorf("multiple gateways are not supported: %s", gateways[0].Name)
+		return nil, nil, 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, fmt.Errorf("OIDC exchange token is not supported: %s", bsp.Name)
+			return nil, nil, 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, fmt.Errorf("failed to rewrite secret %s: %w", s.Name, err)
+			return nil, nil, fmt.Errorf("failed to rewrite secret %s: %w", s.Name, err)
 		}
 	}
 
 	fakeClient, _fakeClientSet, httpRoutes, eps, httpRouteFilter, backends, _, err := translateCustomResourceObjects(ctx, aigwRoutes, aigwBackends, backendSecurityPolicies, gateways, secrets, runCtx.stderrLogger)
 	if err != nil {
-		return nil, fmt.Errorf("error translating: %w", err)
+		return nil, nil, fmt.Errorf("error translating: %w", err)
 	}
 	runCtx.fakeClientSet = _fakeClientSet
 
@@ -260,27 +283,27 @@ func (runCtx *runCmdContext) writeEnvoyResourcesAndRunExtProc(ctx context.Contex
 		Secrets("").Get(ctx,
 		controller.FilterConfigSecretPerGatewayName(gw.Name, gw.Namespace), metav1.GetOptions{})
 	if err != nil {
-		return nil, fmt.Errorf("failed to get filter config secret: %w", err)
+		return nil, nil, fmt.Errorf("failed to get filter config secret: %w", err)
 	}
 
 	rawConfig, ok := filterConfigSecret.StringData[controller.FilterConfigKeyInSecret]
 	if !ok {
-		return nil, fmt.Errorf("failed to get filter config from secret: %w", err)
+		return nil, nil, 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, fmt.Errorf("failed to unmarshal filter config: %w", err)
+		return nil, nil, fmt.Errorf("failed to unmarshal filter config: %w", err)
 	}
 	runCtx.stderrLogger.Info("Running external process", "config", fc)
-	runCtx.mustStartExtProc(ctx, &fc)
-	return fakeClient, nil
+	done := runCtx.mustStartExtProc(ctx, &fc)
+	return fakeClient, done, nil
 }
 
 // mustStartExtProc starts the external process with the given working directory, port, and filter configuration.
 func (runCtx *runCmdContext) mustStartExtProc(
 	ctx context.Context,
 	filterCfg *filterapi.Config,
-) {
+) <-chan error {
 	marshaled, err := yaml.Marshal(filterCfg)
 	if err != nil {
 		panic(fmt.Sprintf("BUG: failed to marshal filter config: %v", err))
@@ -300,11 +323,16 @@ func (runCtx *runCmdContext) mustStartExtProc(
 	} else {
 		args = append(args, "--logLevel", "warn")
 	}
+
+	done := make(chan error)
 	go func() {
 		if err := mainlib.Main(ctx, args, os.Stderr); err != nil {
 			runCtx.stderrLogger.Error("Failed to run external processor", "error", err)
+			done <- fmt.Errorf("%w: %w", errExtProcRun, err)
 		}
+		close(done)
 	}()
+	return done
 }
 
 // mustClearSetOwnerReferencesAndStatusAndWriteObj clears the owner references and status of the given object, marshals it

cmd/aigw/run_test.go

@@ -47,7 +47,7 @@ func TestRun(t *testing.T) {
 	defer cancel()
 	done := make(chan struct{})
 	go func() {
-		require.NoError(t, run(ctx, cmdRun{Debug: true, Path: resourcePath}, os.Stdout, os.Stderr))
+		require.NoError(t, run(ctx, cmdRun{Debug: true, Path: resourcePath}, runOpts{}, os.Stdout, os.Stderr))
 		close(done)
 	}()
 	defer func() {
@@ -157,6 +157,26 @@ func TestRun(t *testing.T) {
 	})
 }
 
+func TestRunExtprocStartFailure(t *testing.T) {
+	var (
+		resourcePath, _ = setupDefaultAIGatewayResourcesWithAvailableCredentials(t)
+		errChan         = make(chan error)
+	)
+
+	go func() {
+		errChan <- run(t.Context(), cmdRun{Debug: true, Path: resourcePath}, runOpts{
+			udsPath: "/dev/null", // This will cause the external processor to fail to start.
+		}, os.Stdout, os.Stderr)
+	}()
+
+	select {
+	case <-time.After(10 * time.Second):
+		t.Fatalf("expected extproc start process to fail and return")
+	case err := <-errChan:
+		require.ErrorIs(t, err, errExtProcRun)
+	}
+}
+
 func TestRunCmdContext_writeEnvoyResourcesAndRunExtProc(t *testing.T) {
 	resourcePath, _ := setupDefaultAIGatewayResourcesWithAvailableCredentials(t)
 	runCtx := &runCmdContext{
@@ -170,29 +190,29 @@ func TestRunCmdContext_writeEnvoyResourcesAndRunExtProc(t *testing.T) {
 	content, err := os.ReadFile(resourcePath)
 	require.NoError(t, err)
 	ctx, cancel := context.WithCancel(t.Context())
-	_, err = runCtx.writeEnvoyResourcesAndRunExtProc(ctx, string(content))
+	_, done, err := runCtx.writeEnvoyResourcesAndRunExtProc(ctx, string(content))
 	require.NoError(t, err)
 	time.Sleep(1 * time.Second)
 	cancel()
 	// Wait for the external processor to stop.
-	time.Sleep(1 * time.Second)
+	require.NoError(t, <-done)
 }
 
 func Test_mustStartExtProc(t *testing.T) {
 	ctx, cancel := context.WithCancel(t.Context())
-	defer cancel()
+	t.Cleanup(cancel)
 	runCtx := &runCmdContext{
 		tmpdir: t.TempDir(),
 		// UNIX doesn't like a long UDS path, so we use a short one.
 		// https://unix.stackexchange.com/questions/367008/why-is-socket-path-length-limited-to-a-hundred-chars
 		udsPath:      filepath.Join("/tmp", "run.sock"),
 		stderrLogger: slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{})),
 	}
-	runCtx.mustStartExtProc(ctx, filterapi.MustLoadDefaultConfig())
+	done := runCtx.mustStartExtProc(ctx, filterapi.MustLoadDefaultConfig())
 	time.Sleep(1 * time.Second)
 	cancel()
 	// Wait for the external processor to stop.
-	time.Sleep(1 * time.Second)
+	require.NoError(t, <-done)
 }
 
 // checkIfOllamaReady checks if the Ollama server is ready and if the specified model is available.