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 Oct 18, 2024
1 parent dd4979b commit b0bb1f2
Show file tree
Hide file tree
Showing 11 changed files with 204 additions and 84 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
8 changes: 4 additions & 4 deletions go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,9 +22,8 @@ require (
github.com/ory/herodot v0.10.3-0.20230626083119-d7e5192f0d88
github.com/ory/jsonschema/v3 v3.0.8
github.com/ory/keto/proto v0.13.0-alpha.0
github.com/ory/x v0.0.647
github.com/ory/x v0.0.662
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.55.0
Expand All @@ -34,7 +33,7 @@ require (
github.com/soheilhy/cmux v0.1.5
github.com/spf13/cobra v1.8.0
github.com/spf13/pflag v1.0.5
github.com/stretchr/testify v1.9.0
github.com/stretchr/testify v1.9.1-0.20240613200951-b074924938f8
github.com/tidwall/gjson v1.17.0
github.com/tidwall/sjson v1.2.5
github.com/urfave/negroni v1.0.0
Expand Down Expand Up @@ -146,11 +145,12 @@ require (
github.com/nyaruka/phonenumbers v1.1.8 // indirect
github.com/opencontainers/go-digest v1.0.0 // indirect
github.com/opencontainers/image-spec v1.1.0 // indirect
github.com/opencontainers/runc v1.1.13 // indirect
github.com/opencontainers/runc v1.1.14 // indirect
github.com/openzipkin/zipkin-go v0.4.2 // indirect
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.19.1 // indirect
Expand Down
12 changes: 6 additions & 6 deletions go.sum
View file
Original file line number Diff line number Diff line change
Expand Up @@ -471,8 +471,8 @@ github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8
github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
github.com/opencontainers/image-spec v1.1.0 h1:8SG7/vwALn54lVB/0yZ/MMwhFrPYtpEHQb2IpWsCzug=
github.com/opencontainers/image-spec v1.1.0/go.mod h1:W4s4sFTMaBeK1BQLXbG4AdM2szdn85PY75RI83NrTrM=
github.com/opencontainers/runc v1.1.13 h1:98S2srgG9vw0zWcDpFMn5TRrh8kLxa/5OFUstuUhmRs=
github.com/opencontainers/runc v1.1.13/go.mod h1:R016aXacfp/gwQBYw2FDGa9m+n6atbLWrYY8hNMT/sA=
github.com/opencontainers/runc v1.1.14 h1:rgSuzbmgz5DUJjeSnw337TxDbRuqjs6iqQck/2weR6w=
github.com/opencontainers/runc v1.1.14/go.mod h1:E4C2z+7BxR7GHXp0hAY53mek+x49X1LjPNeMTfRGvOA=
github.com/openzipkin/zipkin-go v0.4.2 h1:zjqfqHjUpPmB3c1GlCvvgsM1G4LkvqQbBDueDOCg/jA=
github.com/openzipkin/zipkin-go v0.4.2/go.mod h1:ZeVkFjuuBiSy13y8vpSDCjMi9GoI3hPpCJSBx/EYFhY=
github.com/ory/analytics-go/v5 v5.0.1 h1:LX8T5B9FN8KZXOtxgN+R3I4THRRVB6+28IKgKBpXmAM=
Expand All @@ -487,8 +487,8 @@ github.com/ory/herodot v0.10.3-0.20230626083119-d7e5192f0d88 h1:J0CIFKdpUeqKbVMw
github.com/ory/herodot v0.10.3-0.20230626083119-d7e5192f0d88/go.mod h1:MMNmY6MG1uB6fnXYFaHoqdV23DTWctlPsmRCeq/2+wc=
github.com/ory/jsonschema/v3 v3.0.8 h1:Ssdb3eJ4lDZ/+XnGkvQS/te0p+EkolqwTsDOCxr/FmU=
github.com/ory/jsonschema/v3 v3.0.8/go.mod h1:ZPzqjDkwd3QTnb2Z6PAS+OTvBE2x5i6m25wCGx54W/0=
github.com/ory/x v0.0.647 h1:DVKgA3ykMB9qXuMdSl5C8SFWr3yw7Xe8jpSm0+iqGeU=
github.com/ory/x v0.0.647/go.mod h1:M+0EAXo7DT7Z2/Yrzvh4mgxOoV1fGI1jOKyAJ72d4Qs=
github.com/ory/x v0.0.662 h1:Qah5/f63Kr33Wcqm1S79adcYHaoWeiCDMQiDLuUXBqU=
github.com/ory/x v0.0.662/go.mod h1:tS0FyZXpVeKd1lCcFgV/Rb1GlccI/Xq8DraFS+lmIt8=
github.com/pelletier/go-toml v1.9.5 h1:4yBQzkHv+7BHq2PQUZF3Mx0IYxG7LsP222s7Agd3ve8=
github.com/pelletier/go-toml v1.9.5/go.mod h1:u1nR/EPcESfeI/szUZKdtJ0xRNbUoANCkoOuaOx1Y+c=
github.com/pelletier/go-toml/v2 v2.1.0 h1:FnwAJ4oYMvbT/34k9zzHuZNrhlz48GB3/s6at6/MHO4=
Expand Down Expand Up @@ -587,8 +587,8 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
github.com/stretchr/testify v1.9.1-0.20240613200951-b074924938f8 h1:lMSOB3NNaQcNDK9eCjQb0LFfpXW03lzF2SN6vUziALo=
github.com/stretchr/testify v1.9.1-0.20240613200951-b074924938f8/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8=
github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU=
github.com/tidwall/gjson v1.14.2/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
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 b0bb1f2

Please sign in to comment.