feat: emit events through tracing (#1244)

cmd/server/serve.go

@@ -23,7 +23,7 @@ ORY Keto can be configured using environment variables as well as a configuratio
 on configuration options, open the configuration documentation:
 
 >> https://www.ory.sh/keto/docs/reference/configuration <<`,
-		RunE: func(cmd *cobra.Command, args []string) error {
+		RunE: func(cmd *cobra.Command, _ []string) error {
 			reg, err := helpers.NewRegistry(cmd, opts)
 			if err != nil {
 				return err

internal/check/engine.go

@@ -7,10 +7,8 @@ import (
 	"context"
 
 	"github.com/ory/herodot"
+	"github.com/ory/x/otelx"
 	"github.com/pkg/errors"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/codes"
-	"go.opentelemetry.io/otel/trace"
 
 	"github.com/ory/keto/internal/check/checkgroup"
 	"github.com/ory/keto/internal/driver/config"
@@ -19,6 +17,7 @@ import (
 	"github.com/ory/keto/internal/persistence"
 	"github.com/ory/keto/internal/relationtuple"
 	"github.com/ory/keto/internal/x"
+	"github.com/ory/keto/internal/x/events"
 	"github.com/ory/keto/internal/x/graph"
 	"github.com/ory/keto/ketoapi"
 )
@@ -35,6 +34,8 @@ type (
 		persistence.Provider
 		config.Provider
 		x.LoggerProvider
+		x.TracingProvider
+		x.NetworkIDProvider
 	}
 
 	EngineOpt func(*Engine)
@@ -70,15 +71,10 @@ func (e *Engine) CheckIsMember(ctx context.Context, r *relationTuple, restDepth
 // the object in the namespace either directly or indirectly and returns a check
 // result.
 func (e *Engine) CheckRelationTuple(ctx context.Context, r *relationTuple, restDepth int) (res checkgroup.Result) {
-	ctx, span := trace.SpanFromContext(ctx).TracerProvider().Tracer("keto/internal/check").Start(ctx, "Engine.CheckRelationTuple")
-	defer func() {
-		if res.Err != nil {
-			span.SetStatus(codes.Error, res.Err.Error())
-		} else {
-			span.SetAttributes(attribute.String("membership", res.Membership.String()))
-		}
-		span.End()
-	}()
+	ctx, span := e.d.Tracer(ctx).Tracer().Start(ctx, "Engine.CheckRelationTuple")
+	defer otelx.End(span, &res.Err)
+
+	events.Add(ctx, e.d, events.PermissionsChecked)
 
 	// global max-depth takes precedence when it is the lesser or if the request
 	// max-depth is less than or equal to 0

internal/check/engine_test.go

@@ -30,6 +30,7 @@ type deps struct {
 	configProvider
 	x.LoggerProvider
 	x.TracingProvider
+	x.NetworkIDProvider
 }
 
 func newDepsProvider(t testing.TB, namespaces []*namespace.Namespace, pageOpts ...x.PaginationOptionSetter) *deps {
@@ -38,11 +39,12 @@ func newDepsProvider(t testing.TB, namespaces []*namespace.Namespace, pageOpts .
 	mr := relationtuple.NewManagerWrapper(t, reg, pageOpts...)
 
 	return &deps{
-		ManagerWrapper:  mr,
-		Provider:        reg,
-		configProvider:  reg,
-		LoggerProvider:  reg,
-		TracingProvider: reg,
+		ManagerWrapper:    mr,
+		Provider:          reg,
+		configProvider:    reg,
+		LoggerProvider:    reg,
+		TracingProvider:   reg,
+		NetworkIDProvider: reg,
 	}
 }
 
@@ -292,7 +294,7 @@ func TestEngine(t *testing.T) {
 			{Name: "obj"},
 			{Name: "org"},
 		})
-		//require.NoError(t, reg.RelationTupleManager().WriteRelationTuples(ctx, &writeRel, &orgOwnerRel, &userMembershipRel))
+		// require.NoError(t, reg.RelationTupleManager().WriteRelationTuples(ctx, &writeRel, &orgOwnerRel, &userMembershipRel))
 		insertFixtures(t, reg.RelationTupleManager(), []string{
 			"obj:object#write@obj:object#owner",
 			"obj:object#owner@org:organization#member",

internal/check/handler.go

@@ -112,6 +112,7 @@ func (h *Handler) getCheckNoStatus(w http.ResponseWriter, r *http.Request, _ htt
 // Check Permission Or Error Request Parameters
 //
 // swagger:parameters checkPermissionOrError
+// nolint:deadcode,unused
 type checkPermissionOrError struct {
 	// in: query
 	MaxDepth int `json:"max-depth"`
@@ -176,6 +177,7 @@ func (h *Handler) getCheck(ctx context.Context, q url.Values) (bool, error) {
 // Check Permission using Post Request Parameters
 //
 // swagger:parameters postCheckPermission
+// nolint:deadcode,unused
 type postCheckPermission struct {
 	// in: query
 	MaxDepth int `json:"max-depth"`
@@ -187,6 +189,7 @@ type postCheckPermission struct {
 // Check Permission using Post Request Body
 //
 // swagger:model postCheckPermissionBody
+// nolint:deadcode,unused
 type postCheckPermissionBody struct {
 	ketoapi.RelationQuery
 }
@@ -221,7 +224,9 @@ func (h *Handler) postCheckNoStatus(w http.ResponseWriter, r *http.Request, _ ht
 // Post Check Permission Or Error Request Parameters
 //
 // swagger:parameters postCheckPermissionOrError
+// nolint:deadcode,unused
 type postCheckPermissionOrError struct {
+	// nolint:deadcode,unused
 	// in: query
 	MaxDepth int `json:"max-depth"`
 
@@ -232,6 +237,7 @@ type postCheckPermissionOrError struct {
 // Post Check Permission Or Error Body
 //
 // swagger:model postCheckPermissionOrErrorBody
+// nolint:deadcode,unused
 type postCheckPermissionOrErrorBody struct {
 	ketoapi.RelationQuery
 }

internal/driver/registry_default.go

@@ -9,6 +9,7 @@ import (
 	"sync"
 
 	"github.com/gobuffalo/pop/v6"
+	"github.com/gofrs/uuid"
 	"github.com/ory/herodot"
 	"github.com/ory/x/dbal"
 	"github.com/ory/x/fsx"
@@ -68,6 +69,7 @@ type (
 		handlers       []Handler
 		sqaService     *metricsx.Service
 		tracer         *otelx.Tracer
+		tracerWrapper  ketoctx.TracerWrapper
 		pmm            *prometheus.MetricsManager
 		metricsHandler *prometheus.Handler
 
@@ -148,6 +150,12 @@ func (r *RegistryDefault) Tracer(ctx context.Context) *otelx.Tracer {
 		if err != nil {
 			r.Logger().WithError(err).Fatalf("Unable to initialize Tracer.")
 		}
+
+		// Wrap the tracer if required
+		if r.tracerWrapper != nil {
+			t = r.tracerWrapper(t)
+		}
+
 		r.tracer = t
 	}
 
@@ -203,6 +211,13 @@ func (r *RegistryDefault) Persister() persistence.Persister {
 	return r.p
 }
 
+func (r *RegistryDefault) NetworkID(ctx context.Context) uuid.UUID {
+	if r.p == nil {
+		panic("no persister, but expected to have one")
+	}
+	return r.p.NetworkID(ctx)
+}
+
 func (r *RegistryDefault) Traverser() relationtuple.Traverser {
 	if r.traverser == nil {
 		panic("no traverser, but expected to have one")

internal/driver/registry_factory.go

@@ -78,6 +78,7 @@ func NewDefaultRegistry(ctx context.Context, flags *pflag.FlagSet, withoutNetwor
 	r := &RegistryDefault{
 		c:                         c,
 		l:                         l,
+		tracerWrapper:             options.TracerWrapper,
 		ctxer:                     options.Contextualizer(),
 		defaultUnaryInterceptors:  options.GRPCUnaryInterceptors(),
 		defaultStreamInterceptors: options.GRPCStreamInterceptors(),

internal/expand/engine.go

@@ -6,20 +6,23 @@ package expand
 import (
 	"context"
 
-	"github.com/ory/keto/ketoapi"
+	"github.com/ory/x/otelx"
 
 	"github.com/ory/keto/internal/driver/config"
+	"github.com/ory/keto/internal/relationtuple"
 	"github.com/ory/keto/internal/x"
+	"github.com/ory/keto/internal/x/events"
 	"github.com/ory/keto/internal/x/graph"
-
-	"github.com/ory/keto/internal/relationtuple"
+	"github.com/ory/keto/ketoapi"
 )
 
 type (
 	EngineDependencies interface {
 		relationtuple.ManagerProvider
 		config.Provider
 		x.LoggerProvider
+		x.TracingProvider
+		x.NetworkIDProvider
 	}
 	Engine struct {
 		d EngineDependencies
@@ -35,7 +38,16 @@ func NewEngine(d EngineDependencies) *Engine {
 	}
 }
 
-func (e *Engine) BuildTree(ctx context.Context, subject relationtuple.Subject, restDepth int) (*relationtuple.Tree, error) {
+func (e *Engine) BuildTree(ctx context.Context, subject relationtuple.Subject, restDepth int) (t *relationtuple.Tree, err error) {
+	ctx, span := e.d.Tracer(ctx).Tracer().Start(ctx, "Engine.BuildTree")
+	defer otelx.End(span, &err)
+	events.Add(ctx, e.d, events.PermissionsExpanded)
+
+	t, err = e.buildTreeRecursive(ctx, subject, restDepth)
+	return
+}
+
+func (e *Engine) buildTreeRecursive(ctx context.Context, subject relationtuple.Subject, restDepth int) (*relationtuple.Tree, error) {
 	// global max-depth takes precedence when it is the lesser or if the request max-depth is less than or equal to 0
 	if globalMaxDepth := e.d.Config(ctx).MaxReadDepth(); restDepth <= 0 || globalMaxDepth < restDepth {
 		restDepth = globalMaxDepth
@@ -89,7 +101,7 @@ func (e *Engine) BuildTree(ctx context.Context, subject relationtuple.Subject, r
 
 		children := make([]*relationtuple.Tree, len(rels))
 		for ri, r := range rels {
-			child, err := e.BuildTree(ctx, r.Subject, restDepth-1)
+			child, err := e.buildTreeRecursive(ctx, r.Subject, restDepth-1)
 			if err != nil {
 				return nil, err
 			}

internal/expand/engine_test.go

@@ -27,24 +27,30 @@ import (
 	"github.com/ory/keto/internal/driver"
 )
 
-type configProvider = config.Provider
-type loggerProvider = x.LoggerProvider
+type (
+	configProvider = config.Provider
+	loggerProvider = x.LoggerProvider
+)
 
 // deps is defined to capture engine dependencies in a single struct
 type deps struct {
 	*relationtuple.ManagerWrapper // managerProvider
 	configProvider
 	loggerProvider
+	x.TracingProvider
+	x.NetworkIDProvider
 }
 
 func newTestEngine(t *testing.T, namespaces []*namespace.Namespace, paginationOpts ...x.PaginationOptionSetter) (*relationtuple.ManagerWrapper, *expand.Engine) {
 	innerReg := driver.NewSqliteTestRegistry(t, false)
 	require.NoError(t, innerReg.Config(context.Background()).Set(config.KeyNamespaces, namespaces))
 	reg := relationtuple.NewManagerWrapper(t, innerReg, paginationOpts...)
 	e := expand.NewEngine(&deps{
-		ManagerWrapper: reg,
-		configProvider: innerReg,
-		loggerProvider: innerReg,
+		ManagerWrapper:    reg,
+		configProvider:    innerReg,
+		loggerProvider:    innerReg,
+		TracingProvider:   innerReg,
+		NetworkIDProvider: innerReg,
 	})
 	return reg, e
 }

internal/relationtuple/handler.go

@@ -6,9 +6,8 @@ package relationtuple
 import (
 	"google.golang.org/grpc"
 
-	rts "github.com/ory/keto/proto/ory/keto/relation_tuples/v1alpha2"
-
 	"github.com/ory/keto/internal/x"
+	rts "github.com/ory/keto/proto/ory/keto/relation_tuples/v1alpha2"
 )
 
 type (
@@ -17,6 +16,8 @@ type (
 		MapperProvider
 		x.LoggerProvider
 		x.WriterProvider
+		x.TracingProvider
+		x.NetworkIDProvider
 	}
 	handler struct {
 		d handlerDeps

internal/relationtuple/transact_server.go

@@ -8,20 +8,18 @@ import (
 	"encoding/json"
 	"net/http"
 
-	"github.com/ory/keto/internal/x/validate"
-	"github.com/ory/keto/ketoapi"
-
-	rts "github.com/ory/keto/proto/ory/keto/relation_tuples/v1alpha2"
-
 	"github.com/julienschmidt/httprouter"
 	"github.com/ory/herodot"
 	"github.com/pkg/errors"
-)
 
-var (
-	_ rts.WriteServiceServer = (*handler)(nil)
+	"github.com/ory/keto/internal/x/events"
+	"github.com/ory/keto/internal/x/validate"
+	"github.com/ory/keto/ketoapi"
+	rts "github.com/ory/keto/proto/ory/keto/relation_tuples/v1alpha2"
 )
 
+var _ rts.WriteServiceServer = (*handler)(nil)
+
 func protoTuplesWithAction(deltas []*rts.RelationTupleDelta, action rts.RelationTupleDelta_Action) (filtered []*ketoapi.RelationTuple, err error) {
 	for _, d := range deltas {
 		if d.Action == action {
@@ -36,6 +34,8 @@ func protoTuplesWithAction(deltas []*rts.RelationTupleDelta, action rts.Relation
 }
 
 func (h *handler) TransactRelationTuples(ctx context.Context, req *rts.TransactRelationTuplesRequest) (*rts.TransactRelationTuplesResponse, error) {
+	events.Add(ctx, h.d, events.RelationtuplesChanged)
+
 	insertTuples, err := protoTuplesWithAction(req.RelationTupleDeltas, rts.RelationTupleDelta_ACTION_INSERT)
 	if err != nil {
 		return nil, err
@@ -66,6 +66,8 @@ func (h *handler) TransactRelationTuples(ctx context.Context, req *rts.TransactR
 }
 
 func (h *handler) DeleteRelationTuples(ctx context.Context, req *rts.DeleteRelationTuplesRequest) (*rts.DeleteRelationTuplesResponse, error) {
+	events.Add(ctx, h.d, events.RelationtuplesDeleted)
+
 	var q ketoapi.RelationQuery
 
 	switch {
@@ -91,6 +93,7 @@ func (h *handler) DeleteRelationTuples(ctx context.Context, req *rts.DeleteRelat
 // Create Relationship Request Parameters
 //
 // swagger:parameters createRelationship
+// nolint:deadcode,unused
 type createRelationship struct {
 	// in: body
 	Body createRelationshipBody
@@ -99,6 +102,7 @@ type createRelationship struct {
 // Create Relationship Request Body
 //
 // swagger:model createRelationshipBody
+// nolint:deadcode,unused
 type createRelationshipBody struct {
 	ketoapi.RelationQuery
 }
@@ -124,6 +128,8 @@ type createRelationshipBody struct {
 func (h *handler) createRelation(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
 	ctx := r.Context()
 
+	events.Add(ctx, h.d, events.RelationtuplesCreated)
+
 	var rt ketoapi.RelationTuple
 	if err := json.NewDecoder(r.Body).Decode(&rt); err != nil {
 		h.d.Writer().WriteError(w, r, errors.WithStack(herodot.ErrBadRequest.WithError(err.Error())))
@@ -176,6 +182,8 @@ func (h *handler) createRelation(w http.ResponseWriter, r *http.Request, _ httpr
 func (h *handler) deleteRelations(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
 	ctx := r.Context()
 
+	events.Add(ctx, h.d, events.RelationtuplesDeleted)
+
 	if err := validate.All(r,
 		validate.NoExtraQueryParams(ketoapi.RelationQueryKeys...),
 		validate.QueryParamsContainsOneOf(ketoapi.NamespaceKey),
@@ -244,6 +252,8 @@ func internalTuplesWithAction(deltas []*ketoapi.PatchDelta, action ketoapi.Patch
 func (h *handler) patchRelationTuples(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
 	ctx := r.Context()
 
+	events.Add(ctx, h.d, events.RelationtuplesChanged)
+
 	var deltas []*ketoapi.PatchDelta
 	if err := json.NewDecoder(r.Body).Decode(&deltas); err != nil {
 		h.d.Writer().WriteError(w, r, herodot.ErrBadRequest.WithError(err.Error()))