Skip to content

Commit

Permalink
feat: write listen files with actual address (#1607)
Browse files Browse the repository at this point in the history 
This change improves spinning up test Keto servers that use port `0`. A new config value enables to set a file path where the server writes the actual address it listens on after it was assigned a random free port by the OS.
  • Loading branch information
@zepatrik
zepatrik authored Dec 19, 2024
1 parent a72dc58 commit 0ba58c7
Show file tree
Hide file tree
Showing 11 changed files with 254 additions and 126 deletions.
2 changes: 1 addition & 1 deletion cmd/status/root_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
28 changes: 28 additions & 0 deletions embedx/config.schema.json
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
},
Expand Down Expand Up @@ -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"
},
Expand Down Expand Up @@ -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"
},
Expand Down Expand Up @@ -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"
},
Expand Down
2 changes: 1 addition & 1 deletion go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
40 changes: 26 additions & 14 deletions internal/driver/config/provider.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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)
Expand Down
4 changes: 3 additions & 1 deletion internal/driver/config/provider_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
82 changes: 54 additions & 28 deletions internal/driver/daemon.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ package driver

import (
"context"
"fmt"
"net"
"net/http"
"os"
Expand Down Expand Up @@ -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)
Expand All @@ -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()
Expand All @@ -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)
}
}

Expand All @@ -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)
}
}

Expand All @@ -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)
}
}

Expand All @@ -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
Expand All @@ -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
}
Expand Down Expand Up @@ -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{}
Expand All @@ -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")
}
})

Expand All @@ -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 {
Expand Down
Loading

0 comments on commit 0ba58c7

Please sign in to comment.