feat: enable GRPC metrics (#1302)

* feat: enable GRPC metrics * fix: test and server registration * fix: GRPC metrics * fix: clean up for PR --------- Co-authored-by: Henning Perl <[email protected]>
ory · Apr 11, 2023 · 91c12c9 · 91c12c9
1 parent 38e955f
commit 91c12c9
Show file tree

Hide file tree

Showing 4 changed files with 76 additions and 25 deletions.
diff --git a/go.mod b/go.mod
@@ -19,10 +19,12 @@ require (
 	github.com/ory/herodot v0.9.13
 	github.com/ory/jsonschema/v3 v3.0.7
 	github.com/ory/keto/proto v0.10.0-alpha.0
-	github.com/ory/x v0.0.547
+	github.com/ory/x v0.0.550
 	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.3.0
+	github.com/prometheus/common v0.39.0
 	github.com/rs/cors v1.8.3
 	github.com/segmentio/objconv v1.0.1
 	github.com/sirupsen/logrus v1.9.0
@@ -96,6 +98,7 @@ require (
 	github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 // indirect
 	github.com/gorilla/css v1.0.0 // indirect
 	github.com/gorilla/websocket v1.5.0 // indirect
+	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.15.0 // indirect
 	github.com/hashicorp/go-cleanhttp v0.5.2 // indirect
 	github.com/hashicorp/go-hclog v1.4.0 // indirect
@@ -141,8 +144,6 @@ require (
 	github.com/pkg/profile v1.7.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_golang v1.14.0 // indirect
-	github.com/prometheus/client_model v0.3.0 // indirect
-	github.com/prometheus/common v0.39.0 // indirect
 	github.com/prometheus/procfs v0.9.0 // indirect
 	github.com/rogpeppe/go-internal v1.9.0 // indirect
 	github.com/seatgeek/logrus-gelf-formatter v0.0.0-20210414080842-5b05eb8ff761 // indirect

diff --git a/go.sum b/go.sum
@@ -354,6 +354,7 @@ github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/ad
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
 github.com/grpc-ecosystem/go-grpc-middleware v1.3.0 h1:+9834+KizmvFV7pXQGSXQTsaWhq2GjuNUt0aUU0YBYw=
 github.com/grpc-ecosystem/go-grpc-middleware v1.3.0/go.mod h1:z0ButlSOZa5vEBq9m2m2hlwIgKw+rp3sdCBRoJY+30Y=
+github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 h1:Ovs26xHkKqVztRpIrF/92BcuyuQ/YW4NSIpoGtfXNho=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk=
 github.com/grpc-ecosystem/grpc-gateway v1.9.0/go.mod h1:vNeuVxBJEsws4ogUvrchl83t/GYV9WGTSLVdBhOQFDY=
 github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw=
@@ -642,8 +643,8 @@ github.com/ory/herodot v0.9.13/go.mod h1:IWDs9kSvFQqw/cQ8zi5ksyYvITiUU4dI7glUrhZ
 github.com/ory/jsonschema/v3 v3.0.7 h1:GQ9qfZDiJqs4l2d3p56dozCChvejQFZyLKGHYzDzOSo=
 github.com/ory/jsonschema/v3 v3.0.7/go.mod h1:g8c8YOtN4TrR2wYeMdT02GDmzJDI0fEW2nI26BECafY=
 github.com/ory/viper v1.7.5/go.mod h1:ypOuyJmEUb3oENywQZRgeAMwqgOyDqwboO1tj3DjTaM=
-github.com/ory/x v0.0.547 h1:zDuxCi8xpnhSon/2qjjXw9xeYoBwckzV6UdM4dW4sls=
-github.com/ory/x v0.0.547/go.mod h1:sjmOyekJtJrlrq6wtYAKolLruBptf/qUrZ2oMrLEZkM=
+github.com/ory/x v0.0.550 h1:bcaOdUW/jHByoT8U6cfraBpRCtNtUB8lNBD97LTrB+Y=
+github.com/ory/x v0.0.550/go.mod h1:oRVemI3SQQOLvOCJWIRinHQKlgmay/NbwSyRUIsS/Yk=
 github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
 github.com/pascaldekloe/goe v0.1.0/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
 github.com/pborman/uuid v1.2.0/go.mod h1:X/NO0urCmaxf9VXbdlT7C2Yzkj2IKimNn4k+gtPdI/k=

diff --git a/internal/driver/daemon.go b/internal/driver/daemon.go
@@ -446,6 +446,7 @@ func (r *RegistryDefault) unaryInterceptors(ctx context.Context) []grpc.UnarySer
 	is = append(is,
 		herodot.UnaryErrorUnwrapInterceptor,
 		grpcLogrus.UnaryServerInterceptor(r.l.Entry),
+		r.pmm.UnaryServerInterceptor,
 	)
 	if r.sqaService != nil {
 		is = append(is, r.sqaService.UnaryInterceptor)
@@ -464,6 +465,7 @@ func (r *RegistryDefault) streamInterceptors(ctx context.Context) []grpc.StreamS
 	is = append(is,
 		herodot.StreamErrorUnwrapInterceptor,
 		grpcLogrus.StreamServerInterceptor(r.l.Entry),
+		r.pmm.StreamServerInterceptor,
 	)
 	if r.sqaService != nil {
 		is = append(is, r.sqaService.StreamInterceptor)
@@ -479,7 +481,8 @@ func (r *RegistryDefault) newGrpcServer(ctx context.Context) *grpc.Server {
 	if r.grpcTransportCredentials != nil {
 		opts = append(opts, grpc.Creds(r.grpcTransportCredentials))
 	}
-	return grpc.NewServer(opts...)
+	server := grpc.NewServer(opts...)
+	return server
 }
 
 func (r *RegistryDefault) ReadGRPCServer(ctx context.Context) *grpc.Server {
@@ -494,6 +497,7 @@ func (r *RegistryDefault) ReadGRPCServer(ctx context.Context) *grpc.Server {
 			h.RegisterReadGRPC(s)
 		}
 	}
+	r.pmm.Register(s)
 
 	return s
 }
@@ -510,6 +514,7 @@ func (r *RegistryDefault) WriteGRPCServer(ctx context.Context) *grpc.Server {
 			h.RegisterWriteGRPC(s)
 		}
 	}
+	r.pmm.Register(s)
 
 	return s
 }
@@ -526,6 +531,7 @@ func (r *RegistryDefault) OplGRPCServer(ctx context.Context) *grpc.Server {
 			h.RegisterSyntaxGRPC(s)
 		}
 	}
+	r.pmm.Register(s)
 
 	return s
 }

diff --git a/internal/driver/daemon_test.go b/internal/driver/daemon_test.go
@@ -5,12 +5,11 @@ package driver
 
 import (
 	"fmt"
-	"io"
 	"net/http"
-	"net/http/httptest"
 	"testing"
 
 	"github.com/phayes/freeport"
+	"github.com/prometheus/common/expfmt"
 	"github.com/stretchr/testify/assert"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
@@ -24,30 +23,64 @@ import (
 	"context"
 
 	prometheus "github.com/ory/x/prometheusx"
+	ioprometheusclient "github.com/prometheus/client_model/go"
 	"github.com/stretchr/testify/require"
-
-	"github.com/ory/keto/internal/x/dbx"
 )
 
-const (
-	promLogLine = "promhttp_metric_handler_requests_total"
-)
+func TestScrapingEndpoint(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
 
-func TestMetricsHandler(t *testing.T) {
-	for _, dsn := range dbx.GetDSNs(t, false) {
-		r := NewTestRegistry(t, dsn)
-		handler := r.metricsRouter(context.Background())
-		server := httptest.NewServer(handler)
-		defer server.Close()
+	port, err := freeport.GetFreePort()
+	require.NoError(t, err)
 
-		resp, err := http.Get(server.URL + prometheus.MetricsPrometheusPath)
-		require.NoError(t, err)
-		require.Equal(t, resp.StatusCode, http.StatusOK)
+	r := NewSqliteTestRegistry(t, false)
+	require.NoError(t, r.Config(ctx).Set(config.KeyWriteAPIPort, port))
 
-		body, err := io.ReadAll(resp.Body)
-		require.NoError(t, err)
-		require.Contains(t, string(body), promLogLine)
+	//metrics port
+	portMetrics, err := freeport.GetFreePort()
+	require.NoError(t, err)
+	require.NoError(t, r.Config(ctx).Set(config.KeyMetricsPort, portMetrics))
+
+	eg := errgroup.Group{}
+	doneShutdown := make(chan struct{})
+	eg.Go(r.serveWrite(ctx, doneShutdown))
+	eg.Go(r.serveMetrics(ctx, doneShutdown))
+
+	conn, err := grpc.DialContext(ctx, fmt.Sprintf("127.0.0.1:%d", port), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	defer conn.Close()
+
+	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() {
 	}
+
+	promresp, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d", portMetrics) + prometheus.MetricsPrometheusPath)
+	require.NoError(t, err)
+	require.EqualValues(t, http.StatusOK, promresp.StatusCode)
+
+	textParser := expfmt.TextParser{}
+	text, err := textParser.TextToMetricFamilies(promresp.Body)
+	require.NoError(t, err)
+	require.EqualValues(t, "grpc_server_handled_total", *text["grpc_server_handled_total"].Name)
+	require.EqualValues(t, "Check", getLabelValue("grpc_method", text["grpc_server_handled_total"].Metric))
+	require.EqualValues(t, "grpc.health.v1.Health", getLabelValue("grpc_service", text["grpc_server_handled_total"].Metric))
+
+	require.EqualValues(t, "grpc_server_msg_sent_total", *text["grpc_server_msg_sent_total"].Name)
+	require.EqualValues(t, "Check", getLabelValue("grpc_method", text["grpc_server_msg_sent_total"].Metric))
+	require.EqualValues(t, "grpc.health.v1.Health", getLabelValue("grpc_service", text["grpc_server_msg_sent_total"].Metric))
+
+	require.EqualValues(t, "grpc_server_msg_received_total", *text["grpc_server_msg_received_total"].Name)
+	require.EqualValues(t, "Check", getLabelValue("grpc_method", text["grpc_server_msg_received_total"].Metric))
+	require.EqualValues(t, "grpc.health.v1.Health", getLabelValue("grpc_service", text["grpc_server_msg_received_total"].Metric))
+
+	cancel()
+	<-doneShutdown
+	<-doneShutdown
+	require.NoError(t, eg.Wait())
 }
 
 func TestPanicRecovery(t *testing.T) {
@@ -101,3 +134,13 @@ func TestPanicRecovery(t *testing.T) {
 	<-doneShutdown
 	require.NoError(t, eg.Wait())
 }
+
+func getLabelValue(name string, metric []*ioprometheusclient.Metric) string {
+	for _, label := range metric[0].Label {
+		if *label.Name == name {
+			return *label.Value
+		}
+	}
+
+	return ""
+}