feat: write listen files with actual address

cmd/status/root_test.go

@@ -46,7 +46,7 @@ func TestStatusCmd(t *testing.T) {
 			})
 
 			t.Run("case=block", func(t *testing.T) {
-				ctx := context.WithValue(context.Background(), client.ContextKeyTimeout, time.Millisecond)
+				ctx := context.WithValue(context.Background(), client.ContextKeyTimeout, 100*time.Millisecond)
 
 				l, err := net.Listen("tcp", "127.0.0.1:0")
 				require.NoError(t, err)

embedx/config.schema.json

@@ -211,6 +211,13 @@
               "title": "Host",
               "description": "The network interface to listen on."
             },
+            "write_listen_file": {
+              "type": "string",
+              "title": "Read Listen File",
+              "description": "The path to a file that will be created when the read API is ready to accept connections. The content of the file is the host:port of the read API. Use this to get the actual port when using port 0. The service might not yet be ready to accept connections when the file is created.",
+              "format": "uri",
+              "examples": ["file:///tmp/keto-read-api"]
+            },
             "cors": {
               "$ref": "#/definitions/cors"
             },
@@ -239,6 +246,13 @@
               "title": "Host",
               "description": "The network interface to listen on."
             },
+            "write_listen_file": {
+              "type": "string",
+              "title": "Write Listen File",
+              "description": "The path to a file that will be created when the write API is ready to accept connections. The content of the file is the host:port of the write API. Use this to get the actual port when using port 0. The service might not yet be ready to accept connections when the file is created.",
+              "format": "uri",
+              "examples": ["file:///tmp/keto-write-api"]
+            },
             "cors": {
               "$ref": "#/definitions/cors"
             },
@@ -267,6 +281,13 @@
               "title": "Host",
               "description": "The network interface to listen on."
             },
+            "write_listen_file": {
+              "type": "string",
+              "title": "Metrics Listen File",
+              "description": "The path to a file that will be created when the metrics API is ready to accept connections. The content of the file is the host:port of the metrics API. Use this to get the actual port when using port 0. The service might not yet be ready to accept connections when the file is created.",
+              "format": "uri",
+              "examples": ["file:///tmp/keto-metrics-api"]
+            },
             "cors": {
               "$ref": "#/definitions/cors"
             },
@@ -295,6 +316,13 @@
               "title": "Host",
               "description": "The network interface to listen on."
             },
+            "write_listen_file": {
+              "type": "string",
+              "title": "OPL Listen File",
+              "description": "The path to a file that will be created when the OPL API is ready to accept connections. The content of the file is the host:port of the OPL API. Use this to get the actual port when using port 0. The service might not yet be ready to accept connections when the file is created.",
+              "format": "uri",
+              "examples": ["file:///tmp/keto-opl-api"]
+            },
             "cors": {
               "$ref": "#/definitions/cors"
             },

go.mod

@@ -26,7 +26,6 @@ require (
 	github.com/ory/keto/proto v0.13.0-alpha.0
 	github.com/ory/x v0.0.677
 	github.com/pelletier/go-toml v1.9.5
-	github.com/phayes/freeport v0.0.0-20220201140144-74d24b5ae9f5
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_model v0.6.1
 	github.com/prometheus/common v0.61.0
@@ -153,6 +152,7 @@ require (
 	github.com/ory/dockertest/v3 v3.11.0 // indirect
 	github.com/ory/go-acc v0.2.9-0.20230103102148-6b1c9a70dbbe // indirect
 	github.com/pelletier/go-toml/v2 v2.1.0 // indirect
+	github.com/phayes/freeport v0.0.0-20220201140144-74d24b5ae9f5 // indirect
 	github.com/pkg/profile v1.7.0 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/prometheus/client_golang v1.20.4 // indirect

internal/driver/config/provider.go

@@ -43,14 +43,18 @@ const (
 	KeyBatchCheckMaxBatchSize         = "limit.max_batch_check_size"
 	KeyBatchCheckParallelizationLimit = "limit.batch_check_max_parallelization"
 
-	KeyReadAPIHost      = "serve." + string(EndpointRead) + ".host"
-	KeyReadAPIPort      = "serve." + string(EndpointRead) + ".port"
-	KeyWriteAPIHost     = "serve." + string(EndpointWrite) + ".host"
-	KeyWriteAPIPort     = "serve." + string(EndpointWrite) + ".port"
-	KeyOPLSyntaxAPIHost = "serve." + string(EndpointOPLSyntax) + ".host"
-	KeyOPLSyntaxAPIPort = "serve." + string(EndpointOPLSyntax) + ".port"
-	KeyMetricsHost      = "serve." + string(EndpointMetrics) + ".host"
-	KeyMetricsPort      = "serve." + string(EndpointMetrics) + ".port"
+	KeyReadAPIHost         = "serve." + string(EndpointRead) + ".host"
+	KeyReadAPIPort         = "serve." + string(EndpointRead) + ".port"
+	KeyReadAPIListenFile   = "serve." + string(EndpointRead) + ".write_listen_file"
+	KeyWriteAPIHost        = "serve." + string(EndpointWrite) + ".host"
+	KeyWriteAPIPort        = "serve." + string(EndpointWrite) + ".port"
+	KeyWriteAPIListenFile  = "serve." + string(EndpointWrite) + ".write_listen_file"
+	KeyOPLSyntaxAPIHost    = "serve." + string(EndpointOPLSyntax) + ".host"
+	KeyOPLSyntaxAPIPort    = "serve." + string(EndpointOPLSyntax) + ".port"
+	KeyOPLSyntaxListenFile = "serve." + string(EndpointOPLSyntax) + ".write_listen_file"
+	KeyMetricsHost         = "serve." + string(EndpointMetrics) + ".host"
+	KeyMetricsPort         = "serve." + string(EndpointMetrics) + ".port"
+	KeyMetricsListenFile   = "serve." + string(EndpointMetrics) + ".write_listen_file"
 
 	KeyNamespaces                       = "namespaces"
 	KeyNamespacesExperimentalStrictMode = KeyNamespaces + ".experimental_strict_mode"
@@ -167,18 +171,26 @@ func (k *Config) Set(key string, v any) error {
 	return nil
 }
 
-func (k *Config) addressFor(endpoint EndpointType) string {
+func (k *Config) addressFor(endpoint EndpointType) (addr string, listenFile string) {
 	return fmt.Sprintf(
 		"%s:%d",
 		k.p.StringF("serve."+string(endpoint)+".host", ""),
 		k.p.IntF("serve."+string(endpoint)+".port", 0),
-	)
+	), k.p.StringF("serve."+string(endpoint)+".write_listen_file", "")
 }
 
-func (k *Config) ReadAPIListenOn() string      { return k.addressFor(EndpointRead) }
-func (k *Config) WriteAPIListenOn() string     { return k.addressFor(EndpointWrite) }
-func (k *Config) MetricsListenOn() string      { return k.addressFor(EndpointMetrics) }
-func (k *Config) OPLSyntaxAPIListenOn() string { return k.addressFor(EndpointOPLSyntax) }
+func (k *Config) ReadAPIListenOn() (addr string, listenFile string) {
+	return k.addressFor(EndpointRead)
+}
+func (k *Config) WriteAPIListenOn() (addr string, listenFile string) {
+	return k.addressFor(EndpointWrite)
+}
+func (k *Config) MetricsListenOn() (addr string, listenFile string) {
+	return k.addressFor(EndpointMetrics)
+}
+func (k *Config) OPLSyntaxAPIListenOn() (addr string, listenFile string) {
+	return k.addressFor(EndpointOPLSyntax)
+}
 
 func (k *Config) MaxReadDepth() int {
 	return k.p.Int(KeyLimitMaxReadDepth)

internal/driver/config/provider_test.go

@@ -283,5 +283,7 @@ func TestProvider_DefaultReadAPIListenOn(t *testing.T) {
 	)
 	require.NoError(t, err)
 
-	assert.Equal(t, ":4466", config.ReadAPIListenOn())
+	addr, listenFile := config.ReadAPIListenOn()
+	assert.Equal(t, ":4466", addr)
+	assert.Zero(t, listenFile)
 }

internal/driver/daemon.go

@@ -5,6 +5,7 @@ package driver
 
 import (
 	"context"
+	"fmt"
 	"net"
 	"net/http"
 	"os"
@@ -106,7 +107,14 @@ func (r *RegistryDefault) ServeAll(ctx context.Context) error {
 	innerCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-	doneShutdown := make(chan struct{}, 3)
+	serveFuncs := []func(context.Context, chan<- struct{}) func() error{
+		r.serveRead,
+		r.serveWrite,
+		r.serveOPLSyntax,
+		r.serveMetrics,
+	}
+
+	doneShutdown := make(chan struct{}, len(serveFuncs))
 
 	go func() {
 		osSignals := make(chan os.Signal, 1)
@@ -118,33 +126,30 @@ func (r *RegistryDefault) ServeAll(ctx context.Context) error {
 		case <-innerCtx.Done():
 		}
 
-		ctx, cancel := context.WithTimeout(context.Background(), graceful.DefaultShutdownTimeout)
+		ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), graceful.DefaultShutdownTimeout)
 		defer cancel()
 
-		nWaitingForShutdown := cap(doneShutdown)
-		select {
-		case <-ctx.Done():
-			return
-		case <-doneShutdown:
-			nWaitingForShutdown--
-			if nWaitingForShutdown == 0 {
-				// graceful shutdown done
+		nWaitingForShutdown := len(serveFuncs)
+		for {
+			select {
+			case <-ctx.Done():
 				return
+			case <-doneShutdown:
+				nWaitingForShutdown--
+				if nWaitingForShutdown == 0 {
+					// graceful shutdown done
+					return
+				}
 			}
 		}
 	}()
 
 	eg := &errgroup.Group{}
 
 	// We need to separate the setup (invoking the functions that return the serve functions) from running the serve
-	// functions to mitigate race contitions in the HTTP router.
-	for _, serve := range []func() error{
-		r.serveRead(innerCtx, doneShutdown),
-		r.serveWrite(innerCtx, doneShutdown),
-		r.serveOPLSyntax(innerCtx, doneShutdown),
-		r.serveMetrics(innerCtx, doneShutdown),
-	} {
-		eg.Go(serve)
+	// functions to mitigate race conditions in the HTTP router.
+	for _, serve := range serveFuncs {
+		eg.Go(serve(innerCtx, doneShutdown))
 	}
 
 	return eg.Wait()
@@ -158,7 +163,8 @@ func (r *RegistryDefault) serveRead(ctx context.Context, done chan<- struct{}) f
 	}
 
 	return func() error {
-		return multiplexPort(ctx, r.Logger().WithField("endpoint", "read"), r.Config(ctx).ReadAPIListenOn(), rt, s, done)
+		addr, listenFile := r.Config(ctx).ReadAPIListenOn()
+		return multiplexPort(ctx, r.Logger().WithField("endpoint", "read"), addr, listenFile, rt, s, done)
 	}
 }
 
@@ -170,7 +176,8 @@ func (r *RegistryDefault) serveWrite(ctx context.Context, done chan<- struct{})
 	}
 
 	return func() error {
-		return multiplexPort(ctx, r.Logger().WithField("endpoint", "write"), r.Config(ctx).WriteAPIListenOn(), rt, s, done)
+		addr, listenFile := r.Config(ctx).WriteAPIListenOn()
+		return multiplexPort(ctx, r.Logger().WithField("endpoint", "write"), addr, listenFile, rt, s, done)
 	}
 }
 
@@ -182,7 +189,8 @@ func (r *RegistryDefault) serveOPLSyntax(ctx context.Context, done chan<- struct
 	}
 
 	return func() error {
-		return multiplexPort(ctx, r.Logger().WithField("endpoint", "opl"), r.Config(ctx).OPLSyntaxAPIListenOn(), rt, s, done)
+		addr, listenFile := r.Config(ctx).OPLSyntaxAPIListenOn()
+		return multiplexPort(ctx, r.Logger().WithField("endpoint", "opl"), addr, listenFile, rt, s, done)
 	}
 }
 
@@ -192,16 +200,20 @@ func (r *RegistryDefault) serveMetrics(ctx context.Context, done chan<- struct{}
 	//nolint:gosec // graceful.WithDefaults already sets a timeout
 	s := graceful.WithDefaults(&http.Server{
 		Handler: r.metricsRouter(ctx),
-		Addr:    r.Config(ctx).MetricsListenOn(),
 	})
 
 	return func() error {
 		defer cancel()
-
 		eg := &errgroup.Group{}
 
+		addr, listenFile := r.Config(ctx).MetricsListenOn()
+		l, err := listenAndWriteFile(ctx, addr, listenFile)
+		if err != nil {
+			return err
+		}
+
 		eg.Go(func() error {
-			if err := s.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) {
+			if err := s.Serve(l); !errors.Is(err, http.ErrServerClosed) {
 				return errors.WithStack(err)
 			}
 			return nil
@@ -227,8 +239,8 @@ func (r *RegistryDefault) serveMetrics(ctx context.Context, done chan<- struct{}
 	}
 }
 
-func multiplexPort(ctx context.Context, log *logrusx.Logger, addr string, router http.Handler, grpcS *grpc.Server, done chan<- struct{}) error {
-	l, err := (&net.ListenConfig{}).Listen(ctx, "tcp", addr)
+func multiplexPort(ctx context.Context, log *logrusx.Logger, addr, listenFile string, router http.Handler, grpcS *grpc.Server, done chan<- struct{}) error {
+	l, err := listenAndWriteFile(ctx, addr, listenFile)
 	if err != nil {
 		return err
 	}
@@ -281,7 +293,7 @@ func multiplexPort(ctx context.Context, log *logrusx.Logger, addr string, router
 
 		<-ctx.Done()
 
-		ctx, cancel := context.WithTimeout(context.Background(), graceful.DefaultShutdownTimeout)
+		ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), graceful.DefaultShutdownTimeout)
 		defer cancel()
 
 		shutdownEg := errgroup.Group{}
@@ -304,7 +316,7 @@ func multiplexPort(ctx context.Context, log *logrusx.Logger, addr string, router
 				return nil
 			case <-ctx.Done():
 				grpcS.Stop()
-				return errors.New("graceful stop of gRPC server canceled, had to force it")
+				return errors.New("graceful stop of gRPC server timed out, had to force it")
 			}
 		})
 
@@ -327,6 +339,20 @@ func (r *RegistryDefault) allHandlers() []Handler {
 	return r.handlers
 }
 
+func listenAndWriteFile(ctx context.Context, addr, listenFile string) (net.Listener, error) {
+	l, err := (&net.ListenConfig{}).Listen(ctx, "tcp", addr)
+	if err != nil {
+		return nil, errors.WithStack(fmt.Errorf("unable to listen on %q: %w", addr, err))
+	}
+	const filePrefix = "file://"
+	if strings.HasPrefix(listenFile, filePrefix) {
+		if err := os.WriteFile(listenFile[len(filePrefix):], []byte(l.Addr().String()), 0600); err != nil {
+			return nil, errors.WithStack(fmt.Errorf("unable to write listen file %q: %w", listenFile, err))
+		}
+	}
+	return l, nil
+}
+
 func (r *RegistryDefault) ReadRouter(ctx context.Context) http.Handler {
 	n := negroni.New()
 	for _, f := range r.defaultHttpMiddlewares {