Skip to content

Commit

Permalink
feat: write listen files with actual address
Browse files Browse the repository at this point in the history
  • Loading branch information
@zepatrik
zepatrik committed Dec 18, 2024
1 parent cc558dc commit ba8c4db
Show file tree
Hide file tree
Showing 10 changed files with 199 additions and 79 deletions.
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
32 changes: 18 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,18 @@ func (k *Config) Set(key string, v any) error {
return nil
}

func (k *Config) addressFor(endpoint EndpointType) string {
func (k *Config) addressFor(endpoint EndpointType) (string, 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() (string, string) { return k.addressFor(EndpointRead) }
func (k *Config) WriteAPIListenOn() (string, string) { return k.addressFor(EndpointWrite) }
func (k *Config) MetricsListenOn() (string, string) { return k.addressFor(EndpointMetrics) }
func (k *Config) OPLSyntaxAPIListenOn() (string, 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)
}
37 changes: 30 additions & 7 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 @@ -163,7 +164,8 @@ func (r *RegistryDefault) serveRead(ctx context.Context, done chan<- struct{}) e
rt = otelx.TraceHandler(rt, otelhttp.WithTracerProvider(tracer.Provider()))
}

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)
}

func (r *RegistryDefault) serveWrite(ctx context.Context, done chan<- struct{}) error {
Expand All @@ -173,7 +175,8 @@ func (r *RegistryDefault) serveWrite(ctx context.Context, done chan<- struct{})
rt = otelx.TraceHandler(rt, otelhttp.WithTracerProvider(tracer.Provider()))
}

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)
}

func (r *RegistryDefault) serveOPLSyntax(ctx context.Context, done chan<- struct{}) error {
Expand All @@ -183,23 +186,29 @@ func (r *RegistryDefault) serveOPLSyntax(ctx context.Context, done chan<- struct
rt = otelx.TraceHandler(rt, otelhttp.WithTracerProvider(tracer.Provider()))
}

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)
}

func (r *RegistryDefault) serveMetrics(ctx context.Context, done chan<- struct{}) error {
ctx, cancel := context.WithCancel(ctx)
defer cancel()

addr, listenFile := r.Config(ctx).MetricsListenOn()
l, err := listenAndWriteFile(ctx, addr, listenFile)
if err != nil {
return err
}

//nolint:gosec // graceful.WithDefaults already sets a timeout
s := graceful.WithDefaults(&http.Server{
Handler: r.metricsRouter(ctx),
Addr: r.Config(ctx).MetricsListenOn(),
})

eg := &errgroup.Group{}

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 @@ -224,8 +233,8 @@ func (r *RegistryDefault) serveMetrics(ctx context.Context, done chan<- struct{}
return eg.Wait()
}

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 @@ -324,6 +333,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
42 changes: 20 additions & 22 deletions internal/driver/daemon_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@ import (
"testing"
"time"

"github.com/phayes/freeport"
"github.com/prometheus/common/expfmt"
"github.com/stretchr/testify/assert"
"golang.org/x/sync/errgroup"
Expand All @@ -19,8 +18,6 @@ import (
grpcHealthV1 "google.golang.org/grpc/health/grpc_health_v1"
"google.golang.org/grpc/status"

"github.com/ory/keto/internal/driver/config"

"context"

prometheus "github.com/ory/x/prometheusx"
Expand All @@ -29,19 +26,13 @@ import (
)

func TestScrapingEndpoint(t *testing.T) {
t.Parallel()

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

port, err := freeport.GetFreePort()
require.NoError(t, err)

r := NewSqliteTestRegistry(t, false)
require.NoError(t, r.Config(ctx).Set(config.KeyWriteAPIPort, port))

//metrics port
portMetrics, err := freeport.GetFreePort()
require.NoError(t, err)
require.NoError(t, r.Config(ctx).Set(config.KeyMetricsPort, portMetrics))
getAddr := UseDynamicPorts(ctx, t, r)

eg := errgroup.Group{}
doneShutdown := make(chan struct{})
Expand All @@ -52,8 +43,13 @@ func TestScrapingEndpoint(t *testing.T) {
return r.serveMetrics(ctx, doneShutdown)
})

require.EventuallyWithT(t, func(t *assert.CollectT) {
conn, err := grpc.DialContext(ctx, fmt.Sprintf("127.0.0.1:%d", port), grpc.WithTransportCredentials(insecure.NewCredentials()))
_, writePort, _ := getAddr(t, "write")
_, metricsPort, _ := getAddr(t, "metrics")

t.Logf("write port: %s, metrics port: %s", writePort, metricsPort)

assert.EventuallyWithT(t, func(t *assert.CollectT) {
conn, err := grpc.DialContext(ctx, fmt.Sprintf("127.0.0.1:%s", writePort), grpc.WithTransportCredentials(insecure.NewCredentials()))
require.NoError(t, err)
defer conn.Close()

Expand All @@ -63,9 +59,9 @@ func TestScrapingEndpoint(t *testing.T) {
require.NoError(t, watcher.CloseSend())
for err := status.Error(codes.Unavailable, "init"); status.Code(err) != codes.Unavailable; _, err = watcher.Recv() {
}
}, 2*time.Second, 100*time.Millisecond)
}, 2*time.Second, 10*time.Millisecond)

promresp, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d", portMetrics) + prometheus.MetricsPrometheusPath)
promresp, err := http.Get(fmt.Sprintf("http://127.0.0.1:%s", metricsPort) + prometheus.MetricsPrometheusPath)
require.NoError(t, err)
require.EqualValues(t, http.StatusOK, promresp.StatusCode)

Expand All @@ -91,6 +87,8 @@ func TestScrapingEndpoint(t *testing.T) {
}

func TestPanicRecovery(t *testing.T) {
t.Parallel()

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

Expand All @@ -100,31 +98,31 @@ func TestPanicRecovery(t *testing.T) {
streamPanicInterceptor := func(context.Context, interface{}, *grpc.UnaryServerInfo, grpc.UnaryHandler) (interface{}, error) {
panic("test panic")
}
port, err := freeport.GetFreePort()
require.NoError(t, err)

r := NewSqliteTestRegistry(t, false, WithGRPCUnaryInterceptors(unaryPanicInterceptor), WithGRPCUnaryInterceptors(streamPanicInterceptor))
require.NoError(t, r.Config(ctx).Set(config.KeyWriteAPIPort, port))
getAddr := UseDynamicPorts(ctx, t, r)

eg := errgroup.Group{}
doneShutdown := make(chan struct{})
eg.Go(func() error {
return r.serveWrite(ctx, doneShutdown)
})

conn, err := grpc.DialContext(ctx, fmt.Sprintf("127.0.0.1:%d", port), grpc.WithTransportCredentials(insecure.NewCredentials()))
_, port, _ := getAddr(t, "write")

conn, err := grpc.DialContext(ctx, fmt.Sprintf("127.0.0.1:%s", port), grpc.WithTransportCredentials(insecure.NewCredentials()))
require.NoError(t, err)
defer conn.Close()

require.EventuallyWithT(t, func(t *assert.CollectT) {
assert.EventuallyWithT(t, func(t *assert.CollectT) {
cl := grpcHealthV1.NewHealthClient(conn)

watcher, err := cl.Watch(ctx, &grpcHealthV1.HealthCheckRequest{})
require.NoError(t, err)
require.NoError(t, watcher.CloseSend())
for err := status.Error(codes.Unavailable, "init"); status.Code(err) != codes.Unavailable; _, err = watcher.Recv() {
}
}, 2*time.Second, 100*time.Millisecond)
}, 2*time.Second, 10*time.Millisecond)

cl := grpcHealthV1.NewHealthClient(conn)
// we want to ensure the server is still running after the panic
Expand Down
Loading

0 comments on commit ba8c4db

Please sign in to comment.