diff --git a/pkg/authorization/gateway_authorizer.go b/pkg/authorization/gateway_authorizer.go index 9e133de37f..d6d1bd466e 100644 --- a/pkg/authorization/gateway_authorizer.go +++ b/pkg/authorization/gateway_authorizer.go @@ -6,7 +6,6 @@ import ( log "github.com/sirupsen/logrus" "github.com/superplanehq/superplane/pkg/models" - "github.com/superplanehq/superplane/pkg/telemetry" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) @@ -51,12 +50,7 @@ func (a *GatewayAuthorizer) AuthorizeHTTP( return nil, status.Error(codes.NotFound, "Not found") } - var allowed bool - err := telemetry.RunSpan(ctx, "auth.check_permission", func(ctx context.Context) error { - var checkErr error - allowed, checkErr = checkOrganizationPermission(ctx, a.auth, userID, organizationID, rule.Resource, rule.Action) - return checkErr - }) + allowed, err := checkOrganizationPermission(ctx, a.auth, userID, organizationID, rule.Resource, rule.Action) if err != nil { return nil, err } diff --git a/pkg/authorization/organization_permissions.go b/pkg/authorization/organization_permissions.go index 3f3fc9bcde..6fd8773280 100644 --- a/pkg/authorization/organization_permissions.go +++ b/pkg/authorization/organization_permissions.go @@ -4,6 +4,8 @@ import ( "context" "sync" "time" + + "github.com/superplanehq/superplane/pkg/telemetry" ) const organizationPermissionCacheTTL = 15 * time.Minute @@ -62,6 +64,10 @@ type organizationPermissionChecker interface { } func checkOrganizationPermission(ctx context.Context, auth organizationPermissionChecker, userID, orgID, resource, action string) (bool, error) { + var err error + ctx, done := telemetry.Span(ctx, "auth.check_organization_permission") + defer done(&err) + if allowed, ok := cachedOrganizationPermissionAllowed(userID, orgID, resource, action); ok && allowed { return true, nil } diff --git a/pkg/grpc/actions/canvases/common.go b/pkg/grpc/actions/canvases/common.go index 79eed8d5af..0c1578abc0 100644 --- a/pkg/grpc/actions/canvases/common.go +++ b/pkg/grpc/actions/canvases/common.go @@ -6,7 +6,7 @@ import ( "github.com/google/uuid" "github.com/superplanehq/superplane/pkg/grpc/actions/canvases/changesets" - "github.com/superplanehq/superplane/pkg/grpc/errors" + grpcerrors "github.com/superplanehq/superplane/pkg/grpc/errors" "github.com/superplanehq/superplane/pkg/models" pb "github.com/superplanehq/superplane/pkg/protos/canvases" "github.com/superplanehq/superplane/pkg/telemetry" @@ -15,87 +15,68 @@ import ( const canvasNameAlreadyExistsMessage = "Canvas with the same name already exists" -func checkCanvasExistence(ctx context.Context, db *gorm.DB, orgID, canvasID uuid.UUID) error { - return telemetry.RunSpan(ctx, "canvases.check_canvas_existence", func(ctx context.Context) error { - exists, err := models.CheckCanvasExistence(db, orgID, canvasID) - if err != nil { - return err - } - if !exists { - return gorm.ErrRecordNotFound - } +func checkCanvasExistence(ctx context.Context, db *gorm.DB, orgID, canvasID uuid.UUID) (err error) { + ctx, done := telemetry.Span(ctx, "canvases.check_canvas_existence") + defer done(&err) - return nil - }) + exists, err := models.CheckCanvasExistence(db.WithContext(ctx), orgID, canvasID) + if err != nil { + return err + } + if !exists { + return gorm.ErrRecordNotFound + } + + return nil +} + +func loadCanvas(ctx context.Context, db *gorm.DB, orgID, canvasID uuid.UUID) (canvas *models.Canvas, err error) { + ctx, done := telemetry.Span(ctx, "canvases.find_canvas") + defer done(&err) + + return models.FindCanvasInTransaction(db.WithContext(ctx), orgID, canvasID) } -func loadCanvas(ctx context.Context, db *gorm.DB, orgID, canvasID uuid.UUID) (*models.Canvas, error) { - var canvas *models.Canvas - err := telemetry.RunSpan(ctx, "canvases.find_canvas", func(ctx context.Context) error { - var findErr error - canvas, findErr = models.FindCanvasInTransaction(db, orgID, canvasID) - return findErr - }) +func loadLiveCanvasVersion(ctx context.Context, db *gorm.DB, canvas *models.Canvas) (liveVersion *models.CanvasVersion, err error) { + ctx, done := telemetry.Span(ctx, "canvases.load_live_version") + defer done(&err) + + return models.FindLiveCanvasVersionByCanvasInTransaction(db.WithContext(ctx), canvas) +} + +func loadCanvasStatus(ctx context.Context, db *gorm.DB, canvasID uuid.UUID) (canvasStatus *pb.Canvas_Status, err error) { + ctx, done := telemetry.Span(ctx, "canvases.load_status") + defer done(&err) + + lastExecutions, err := models.FindLastExecutionPerNode(db.WithContext(ctx), canvasID) if err != nil { return nil, err } - return canvas, nil -} + executionResources, err := LoadNodeExecutionResources(db.WithContext(ctx), lastExecutions) + if err != nil { + return nil, err + } -func loadLiveCanvasVersion(ctx context.Context, db *gorm.DB, canvas *models.Canvas) (*models.CanvasVersion, error) { - var liveVersion *models.CanvasVersion - err := telemetry.RunSpan(ctx, "canvases.load_live_version", func(ctx context.Context) error { - var loadErr error - liveVersion, loadErr = models.FindLiveCanvasVersionByCanvasInTransaction(db, canvas) - return loadErr - }) + serializedExecutions, err := SerializeNodeExecutions(lastExecutions, executionResources) if err != nil { return nil, err } - return liveVersion, nil -} + lastEvents, err := models.FindLastEventPerNode(db.WithContext(ctx), canvasID) + if err != nil { + return nil, err + } -func loadCanvasStatus(ctx context.Context, db *gorm.DB, canvasID uuid.UUID) (*pb.Canvas_Status, error) { - var canvasStatus *pb.Canvas_Status - err := telemetry.RunSpan(ctx, "canvases.load_status", func(ctx context.Context) error { - lastExecutions, loadErr := models.FindLastExecutionPerNode(db, canvasID) - if loadErr != nil { - return loadErr - } - - executionResources, loadErr := LoadNodeExecutionResources(db, lastExecutions) - if loadErr != nil { - return loadErr - } - - serializedExecutions, loadErr := SerializeNodeExecutions(lastExecutions, executionResources) - if loadErr != nil { - return loadErr - } - - lastEvents, loadErr := models.FindLastEventPerNode(db, canvasID) - if loadErr != nil { - return loadErr - } - - serializedEvents, loadErr := SerializeCanvasEvents(lastEvents) - if loadErr != nil { - return loadErr - } - - canvasStatus = &pb.Canvas_Status{ - LastExecutions: serializedExecutions, - LastEvents: serializedEvents, - } - return nil - }) + serializedEvents, err := SerializeCanvasEvents(lastEvents) if err != nil { return nil, err } - return canvasStatus, nil + return &pb.Canvas_Status{ + LastExecutions: serializedExecutions, + LastEvents: serializedEvents, + }, nil } func ensureCanvasNameAvailableInTransaction( diff --git a/pkg/grpc/actions/canvases/describe_canvas.go b/pkg/grpc/actions/canvases/describe_canvas.go index f570b97f84..ed97d6c2ce 100644 --- a/pkg/grpc/actions/canvases/describe_canvas.go +++ b/pkg/grpc/actions/canvases/describe_canvas.go @@ -11,8 +11,16 @@ import ( pb "github.com/superplanehq/superplane/pkg/protos/canvases" "github.com/superplanehq/superplane/pkg/registry" "github.com/superplanehq/superplane/pkg/telemetry" + "gorm.io/gorm" ) +func loadCanvasCreator(ctx context.Context, db *gorm.DB, canvas *models.Canvas) (user *models.User, err error) { + ctx, done := telemetry.Span(ctx, "canvases.load_creator") + defer done(&err) + + return models.FindMaybeDeletedUserByIDInTransaction(db.WithContext(ctx), canvas.OrganizationID.String(), canvas.CreatedBy.String()) +} + func DescribeCanvas(ctx context.Context, registry *registry.Registry, organizationID string, id string) (*pb.DescribeCanvasResponse, error) { canvasID, err := uuid.Parse(id) if err != nil { @@ -29,11 +37,7 @@ func DescribeCanvas(ctx context.Context, registry *registry.Registry, organizati var user *models.User if canvas.CreatedBy != nil { - err = telemetry.RunSpan(ctx, "canvases.load_creator", func(ctx context.Context) error { - var loadErr error - user, loadErr = models.FindMaybeDeletedUserByIDInTransaction(db, canvas.OrganizationID.String(), canvas.CreatedBy.String()) - return loadErr - }) + user, err = loadCanvasCreator(ctx, db, canvas) if err != nil { return nil, err } diff --git a/pkg/grpc/actions/canvases/list_canvas_memories.go b/pkg/grpc/actions/canvases/list_canvas_memories.go index 212cf7d8bd..4b4e9d55a2 100644 --- a/pkg/grpc/actions/canvases/list_canvas_memories.go +++ b/pkg/grpc/actions/canvases/list_canvas_memories.go @@ -36,12 +36,7 @@ func ListCanvasMemories(ctx context.Context, registry *registry.Registry, organi return nil, grpcerrors.Internal(err, "failed to load canvas") } - var records []models.CanvasMemory - err = telemetry.RunSpan(ctx, "memories.list", func(ctx context.Context) error { - var listErr error - records, listErr = models.ListCanvasMemoriesInTransaction(database.DB(ctx), canvasUUID) - return listErr - }) + records, err := listCanvasMemories(ctx, canvasUUID) if err != nil { return nil, grpcerrors.Internal(err, "failed to list canvas memories") } @@ -56,27 +51,29 @@ func ListCanvasMemories(ctx context.Context, registry *registry.Registry, organi }, nil } -func serializeCanvasMemories(ctx context.Context, records []models.CanvasMemory) ([]*pb.CanvasMemory, error) { - var items []*pb.CanvasMemory - err := telemetry.RunSpan(ctx, "memories.serialize", func(ctx context.Context) error { - items = make([]*pb.CanvasMemory, 0, len(records)) - for _, record := range records { - item, itemErr := canvasMemoryToProto(record) - if itemErr != nil { - return itemErr - } - - items = append(items, item) - } +func listCanvasMemories(ctx context.Context, canvasUUID uuid.UUID) (records []models.CanvasMemory, err error) { + ctx, done := telemetry.Span(ctx, "memories.list") + defer done(&err) + + return models.ListCanvasMemoriesInTransaction(database.DB(ctx), canvasUUID) +} + +func serializeCanvasMemories(ctx context.Context, records []models.CanvasMemory) (items []*pb.CanvasMemory, err error) { + ctx, done := telemetry.Span(ctx, "memories.serialize") + defer done(&err) - if span := trace.SpanFromContext(ctx); span.IsRecording() { - span.SetAttributes(attribute.Int("memories.count", len(records))) + items = make([]*pb.CanvasMemory, 0, len(records)) + for _, record := range records { + item, itemErr := canvasMemoryToProto(record) + if itemErr != nil { + return nil, itemErr } - return nil - }) - if err != nil { - return nil, err + items = append(items, item) + } + + if span := trace.SpanFromContext(ctx); span.IsRecording() { + span.SetAttributes(attribute.Int("memories.count", len(records))) } return items, nil diff --git a/pkg/grpc/actions/canvases/list_canvas_versions.go b/pkg/grpc/actions/canvases/list_canvas_versions.go index c9c35addbd..93bd58cdcc 100644 --- a/pkg/grpc/actions/canvases/list_canvas_versions.go +++ b/pkg/grpc/actions/canvases/list_canvas_versions.go @@ -2,6 +2,7 @@ package canvases import ( "context" + "time" "github.com/google/uuid" "github.com/superplanehq/superplane/pkg/authentication" @@ -54,25 +55,7 @@ func ListCanvasVersionsPaginated( limit = getCanvasVersionLimit(limit) beforeTime := getBefore(before) - var publishedVersions []models.CanvasVersion - var publishedCount int64 - err = telemetry.RunSpan(ctx, "canvases.list_published_versions", func(ctx context.Context) error { - return database.DB(ctx).Transaction(func(tx *gorm.DB) error { - versions, versionsErr := models.ListPublishedCanvasVersionsInTransaction(tx, canvasUUID, int(limit), beforeTime) - if versionsErr != nil { - return versionsErr - } - publishedVersions = versions - - count, countErr := models.CountPublishedCanvasVersionsInTransaction(tx, canvasUUID) - if countErr != nil { - return countErr - } - publishedCount = count - - return nil - }) - }) + publishedVersions, publishedCount, err := listPublishedCanvasVersions(ctx, canvasUUID, int(limit), beforeTime) if err != nil { return nil, grpcerrors.Internal(err, "failed to list canvas versions") } @@ -98,25 +81,7 @@ func listDraftCanvasVersions( limit = getCanvasVersionLimit(limit) beforeTime := getBefore(before) - var draftVersions []models.CanvasVersion - var draftCount int64 - err := telemetry.RunSpan(ctx, "canvases.list_draft_versions", func(ctx context.Context) error { - return database.DB(ctx).Transaction(func(tx *gorm.DB) error { - versions, versionsErr := models.ListDraftBranchesForCanvasInTransaction(tx, canvasID, ownerID, int(limit), beforeTime) - if versionsErr != nil { - return versionsErr - } - draftVersions = versions - - count, countErr := models.CountDraftBranchesForCanvasInTransaction(tx, canvasID, ownerID) - if countErr != nil { - return countErr - } - draftCount = count - - return nil - }) - }) + draftVersions, draftCount, err := listDraftCanvasVersionsInTransaction(ctx, canvasID, ownerID, int(limit), beforeTime) if err != nil { return nil, grpcerrors.Internal(err, "failed to list canvas versions") } @@ -168,3 +133,39 @@ func getLastDraftCanvasVersionTimestamp(versions []models.CanvasVersion) *timest return timestamppb.New(*lastVersion.UpdatedAt) } + +func listPublishedCanvasVersions(ctx context.Context, canvasUUID uuid.UUID, limit int, beforeTime *time.Time) (versions []models.CanvasVersion, count int64, err error) { + ctx, done := telemetry.Span(ctx, "canvases.list_published_versions") + defer done(&err) + + err = database.DB(ctx).Transaction(func(tx *gorm.DB) error { + var txErr error + versions, txErr = models.ListPublishedCanvasVersionsInTransaction(tx, canvasUUID, limit, beforeTime) + if txErr != nil { + return txErr + } + + count, txErr = models.CountPublishedCanvasVersionsInTransaction(tx, canvasUUID) + return txErr + }) + + return versions, count, err +} + +func listDraftCanvasVersionsInTransaction(ctx context.Context, canvasID, ownerID uuid.UUID, limit int, beforeTime *time.Time) (versions []models.CanvasVersion, count int64, err error) { + ctx, done := telemetry.Span(ctx, "canvases.list_draft_versions") + defer done(&err) + + err = database.DB(ctx).Transaction(func(tx *gorm.DB) error { + var txErr error + versions, txErr = models.ListDraftBranchesForCanvasInTransaction(tx, canvasID, ownerID, limit, beforeTime) + if txErr != nil { + return txErr + } + + count, txErr = models.CountDraftBranchesForCanvasInTransaction(tx, canvasID, ownerID) + return txErr + }) + + return versions, count, err +} diff --git a/pkg/grpc/actions/canvases/list_runs.go b/pkg/grpc/actions/canvases/list_runs.go index 8b106e6cc5..3d9ace2d59 100644 --- a/pkg/grpc/actions/canvases/list_runs.go +++ b/pkg/grpc/actions/canvases/list_runs.go @@ -3,6 +3,8 @@ package canvases import ( "context" "fmt" + "time" + "github.com/google/uuid" "github.com/superplanehq/superplane/pkg/database" "github.com/superplanehq/superplane/pkg/grpc/errors" @@ -23,24 +25,12 @@ func ListRuns(ctx context.Context, registry *registry.Registry, canvasID uuid.UU return nil, err } - db := database.DB(ctx) - - var runs []models.CanvasRun - err = telemetry.RunSpan(ctx, "runs.list", func(ctx context.Context) error { - var listErr error - runs, listErr = models.ListCanvasRunsInTransaction(database.DB(ctx), canvasID, int(limit), beforeTime, filters) - return listErr - }) + runs, err := listCanvasRuns(ctx, canvasID, int(limit), beforeTime, filters) if err != nil { return nil, err } - var count int64 - err = telemetry.RunSpan(ctx, "runs.count", func(ctx context.Context) error { - var countErr error - count, countErr = models.CountCanvasRunsInTransaction(database.DB(ctx), canvasID, filters) - return countErr - }) + count, err := countCanvasRuns(ctx, canvasID, filters) if err != nil { return nil, err } @@ -50,22 +40,12 @@ func ListRuns(ctx context.Context, registry *registry.Registry, canvasID uuid.UU runIDs[i] = run.ID } - var rootEventsByRunID map[string]models.CanvasEvent - err = telemetry.RunSpan(ctx, "runs.load_root_events", func(ctx context.Context) error { - var loadErr error - rootEventsByRunID, loadErr = listRootEventsForRuns(ctx, canvasID, runIDs) - return loadErr - }) + rootEventsByRunID, err := loadRootEventsForRunsSpan(ctx, canvasID, runIDs) if err != nil { return nil, err } - var executions []models.CanvasNodeExecution - err = telemetry.RunSpan(ctx, "runs.load_executions", func(ctx context.Context) error { - var loadErr error - executions, loadErr = models.ListExecutionsForRunsInTransaction(db, canvasID, runIDs) - return loadErr - }) + executions, err := listExecutionsForRuns(ctx, canvasID, runIDs) if err != nil { return nil, err } @@ -249,21 +229,46 @@ func serializeCanvasRuns( runs []models.CanvasRun, rootEventsByRunID map[string]models.CanvasEvent, executionsByRunID map[string][]models.CanvasNodeExecution, -) ([]*pb.CanvasRun, error) { - var serialized []*pb.CanvasRun - err := telemetry.RunSpan(ctx, "runs.serialize", func(ctx context.Context) error { - var serErr error - serialized, serErr = SerializeCanvasRuns(runs, rootEventsByRunID, executionsByRunID) - - if span := trace.SpanFromContext(ctx); span.IsRecording() { - span.SetAttributes(attribute.Int("runs.count", len(runs))) - } +) (serialized []*pb.CanvasRun, err error) { + ctx, done := telemetry.Span(ctx, "runs.serialize") + defer done(&err) - return serErr - }) + serialized, err = SerializeCanvasRuns(runs, rootEventsByRunID, executionsByRunID) if err != nil { return nil, err } + if span := trace.SpanFromContext(ctx); span.IsRecording() { + span.SetAttributes(attribute.Int("runs.count", len(runs))) + } + return serialized, nil } + +func listCanvasRuns(ctx context.Context, canvasID uuid.UUID, limit int, beforeTime *time.Time, filters models.CanvasRunFilters) (runs []models.CanvasRun, err error) { + ctx, done := telemetry.Span(ctx, "runs.list") + defer done(&err) + + return models.ListCanvasRunsInTransaction(database.DB(ctx), canvasID, limit, beforeTime, filters) +} + +func countCanvasRuns(ctx context.Context, canvasID uuid.UUID, filters models.CanvasRunFilters) (count int64, err error) { + ctx, done := telemetry.Span(ctx, "runs.count") + defer done(&err) + + return models.CountCanvasRunsInTransaction(database.DB(ctx), canvasID, filters) +} + +func loadRootEventsForRunsSpan(ctx context.Context, canvasID uuid.UUID, runIDs []uuid.UUID) (events map[string]models.CanvasEvent, err error) { + ctx, done := telemetry.Span(ctx, "runs.load_root_events") + defer done(&err) + + return listRootEventsForRuns(ctx, canvasID, runIDs) +} + +func listExecutionsForRuns(ctx context.Context, canvasID uuid.UUID, runIDs []uuid.UUID) (executions []models.CanvasNodeExecution, err error) { + ctx, done := telemetry.Span(ctx, "runs.load_executions") + defer done(&err) + + return models.ListExecutionsForRunsInTransaction(database.DB(ctx), canvasID, runIDs) +} diff --git a/pkg/grpc/actions/canvases/serialization.go b/pkg/grpc/actions/canvases/serialization.go index 15fb0e36bd..20ac5f02fc 100644 --- a/pkg/grpc/actions/canvases/serialization.go +++ b/pkg/grpc/actions/canvases/serialization.go @@ -234,16 +234,9 @@ func serializeCanvas( liveVersion *models.CanvasVersion, user *models.User, status *pb.Canvas_Status, -) (*pb.Canvas, error) { - var proto *pb.Canvas - err := telemetry.RunSpan(ctx, "canvases.serialize", func(ctx context.Context) error { - var serErr error - proto, serErr = SerializeCanvas(canvas, liveVersion, user, status) - return serErr - }) - if err != nil { - return nil, err - } +) (proto *pb.Canvas, err error) { + ctx, done := telemetry.Span(ctx, "canvases.serialize") + defer done(&err) - return proto, nil + return SerializeCanvas(canvas, liveVersion, user, status) } diff --git a/pkg/grpc/actions/canvases/version_serialization.go b/pkg/grpc/actions/canvases/version_serialization.go index 9402c20aed..99a0619571 100644 --- a/pkg/grpc/actions/canvases/version_serialization.go +++ b/pkg/grpc/actions/canvases/version_serialization.go @@ -112,24 +112,23 @@ func ownersByIDForCanvasVersions(ctx context.Context, orgID string, versions []m } func serializeCanvasVersions(ctx context.Context, versions []models.CanvasVersion, organizationID string) []*pb.CanvasVersion { - var protoVersions []*pb.CanvasVersion - _ = telemetry.RunSpan(ctx, "canvases.serialize_versions", func(ctx context.Context) error { - ownersByID, err := ownersByIDForCanvasVersions(ctx, organizationID, versions) - if err != nil { - ownersByID = nil - } + var err error + ctx, done := telemetry.Span(ctx, "canvases.serialize_versions") + defer done(&err) - protoVersions = make([]*pb.CanvasVersion, 0, len(versions)) - for i := range versions { - protoVersions = append(protoVersions, SerializeCanvasVersion(&versions[i], organizationID, ownersByID)) - } + ownersByID, ownersErr := ownersByIDForCanvasVersions(ctx, organizationID, versions) + if ownersErr != nil { + ownersByID = nil + } - if span := trace.SpanFromContext(ctx); span.IsRecording() { - span.SetAttributes(attribute.Int("canvases.version_count", len(versions))) - } + protoVersions := make([]*pb.CanvasVersion, 0, len(versions)) + for i := range versions { + protoVersions = append(protoVersions, SerializeCanvasVersion(&versions[i], organizationID, ownersByID)) + } - return nil - }) + if span := trace.SpanFromContext(ctx); span.IsRecording() { + span.SetAttributes(attribute.Int("canvases.version_count", len(versions))) + } return protoVersions } diff --git a/pkg/grpc/actions/me/get_user.go b/pkg/grpc/actions/me/get_user.go index 3c28bbe110..2e7acfc279 100644 --- a/pkg/grpc/actions/me/get_user.go +++ b/pkg/grpc/actions/me/get_user.go @@ -4,6 +4,7 @@ import ( "context" "errors" + "github.com/google/uuid" "github.com/superplanehq/superplane/pkg/authentication" "github.com/superplanehq/superplane/pkg/authorization" "github.com/superplanehq/superplane/pkg/database" @@ -52,11 +53,7 @@ func GetUser(ctx context.Context, authService authorization.Authorization, inclu } var roles []*authorization.RoleDefinition - err = telemetry.RunSpan(ctx, "auth.load_user_roles", func(ctx context.Context) error { - var loadErr error - roles, loadErr = authService.GetUserRolesForOrg(ctx, userID, user.OrganizationID.String()) - return loadErr - }) + roles, err = loadUserRoles(ctx, authService, userID, user.OrganizationID) if err != nil { return nil, grpcerrors.Internal(err, "failed to get user roles") } @@ -86,11 +83,7 @@ func GetUser(ctx context.Context, authService authorization.Authorization, inclu userProto.Permissions = permissions var groups []string - err = telemetry.RunSpan(ctx, "auth.load_user_groups", func(ctx context.Context) error { - var loadErr error - groups, loadErr = authService.GetUserGroups(ctx, user.OrganizationID.String(), models.DomainTypeOrganization, userID) - return loadErr - }) + groups, err = loadUserGroups(ctx, authService, userID, user.OrganizationID) if err != nil { return nil, grpcerrors.Internal(err, "failed to get user groups") } @@ -102,13 +95,11 @@ func GetUser(ctx context.Context, authService authorization.Authorization, inclu }, nil } -func loadUser(ctx context.Context, orgID, userID string) (*models.User, error) { - var user *models.User - err := telemetry.RunSpan(ctx, "auth.load_user", func(ctx context.Context) error { - var loadErr error - user, loadErr = models.FindActiveUserByIDInTransaction(database.DB(ctx), orgID, userID) - return loadErr - }) +func loadUser(ctx context.Context, orgID, userID string) (user *models.User, err error) { + ctx, done := telemetry.Span(ctx, "auth.load_user") + defer done(&err) + + user, err = models.FindActiveUserByIDInTransaction(database.DB(ctx), orgID, userID) if err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return nil, grpcerrors.NotFound(err, "user not found") @@ -119,3 +110,17 @@ func loadUser(ctx context.Context, orgID, userID string) (*models.User, error) { return user, nil } + +func loadUserRoles(ctx context.Context, authService authorization.Authorization, userID string, organizationID uuid.UUID) (roles []*authorization.RoleDefinition, err error) { + ctx, done := telemetry.Span(ctx, "auth.load_user_roles") + defer done(&err) + + return authService.GetUserRolesForOrg(ctx, userID, organizationID.String()) +} + +func loadUserGroups(ctx context.Context, authService authorization.Authorization, userID string, organizationID uuid.UUID) (groups []string, err error) { + ctx, done := telemetry.Span(ctx, "auth.load_user_groups") + defer done(&err) + + return authService.GetUserGroups(ctx, organizationID.String(), models.DomainTypeOrganization, userID) +} diff --git a/pkg/grpc/actions/organizations/describe_organization.go b/pkg/grpc/actions/organizations/describe_organization.go index 250a23f873..1c7b38fe14 100644 --- a/pkg/grpc/actions/organizations/describe_organization.go +++ b/pkg/grpc/actions/organizations/describe_organization.go @@ -15,12 +15,7 @@ import ( ) func DescribeOrganization(ctx context.Context, orgID string) (*pb.DescribeOrganizationResponse, error) { - var organization *models.Organization - err := telemetry.RunSpan(ctx, "organizations.load", func(ctx context.Context) error { - var loadErr error - organization, loadErr = models.FindOrganizationByIDInTransaction(database.DB(ctx), orgID) - return loadErr - }) + organization, err := loadOrganization(ctx, orgID) if err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return nil, grpcerrors.NotFound(err, "organization not found") @@ -47,3 +42,10 @@ func DescribeOrganization(ctx context.Context, orgID string) (*pb.DescribeOrgani return response, nil } + +func loadOrganization(ctx context.Context, orgID string) (organization *models.Organization, err error) { + ctx, done := telemetry.Span(ctx, "organizations.load") + defer done(&err) + + return models.FindOrganizationByIDInTransaction(database.DB(ctx), orgID) +} diff --git a/pkg/grpc/actions/organizations/describe_usage.go b/pkg/grpc/actions/organizations/describe_usage.go index 6a5a002e9b..32d15dbfa1 100644 --- a/pkg/grpc/actions/organizations/describe_usage.go +++ b/pkg/grpc/actions/organizations/describe_usage.go @@ -26,12 +26,7 @@ func DescribeUsage(ctx context.Context, usageService usage.Service, orgID string } readStartedAt := time.Now() - var limits *usagepb.OrganizationLimits - err := telemetry.RunSpan(ctx, "usage.load_limits", func(ctx context.Context) error { - var loadErr error - limits, loadErr = describeUsageLimits(ctx, usageService, orgID) - return loadErr - }) + limits, err := loadUsageLimits(ctx, usageService, orgID) if err != nil { return nil, err } @@ -40,12 +35,7 @@ func DescribeUsage(ctx context.Context, usageService usage.Service, orgID string log.Warnf("Failed to persist usage limits cache for organization %s: %v", orgID, err) } - var orgUsage *usagepb.OrganizationUsage - err = telemetry.RunSpan(ctx, "usage.load_metrics", func(ctx context.Context) error { - var loadErr error - orgUsage, loadErr = describeUsageMetrics(ctx, usageService, orgID) - return loadErr - }) + orgUsage, err := loadUsageMetrics(ctx, usageService, orgID) if err != nil { return nil, err } @@ -56,13 +46,29 @@ func DescribeUsage(ctx context.Context, usageService usage.Service, orgID string go usage.ReconcileCanvasCount(orgID, orgUsage.GetCanvases()) - var response *pb.DescribeUsageResponse - _ = telemetry.RunSpan(ctx, "usage.build_response", func(ctx context.Context) error { - response = buildDescribeUsageResponse(limits, orgUsage) - return nil - }) + return describeUsageResponse(ctx, limits, orgUsage), nil +} + +func loadUsageLimits(ctx context.Context, usageService usage.Service, orgID string) (limits *usagepb.OrganizationLimits, err error) { + ctx, done := telemetry.Span(ctx, "usage.load_limits") + defer done(&err) + + return describeUsageLimits(ctx, usageService, orgID) +} + +func loadUsageMetrics(ctx context.Context, usageService usage.Service, orgID string) (orgUsage *usagepb.OrganizationUsage, err error) { + ctx, done := telemetry.Span(ctx, "usage.load_metrics") + defer done(&err) + + return describeUsageMetrics(ctx, usageService, orgID) +} + +func describeUsageResponse(ctx context.Context, limits *usagepb.OrganizationLimits, orgUsage *usagepb.OrganizationUsage) *pb.DescribeUsageResponse { + var err error + ctx, done := telemetry.Span(ctx, "usage.build_response") + defer done(&err) - return response, nil + return buildDescribeUsageResponse(limits, orgUsage) } func buildDescribeUsageResponse(limits *usagepb.OrganizationLimits, orgUsage *usagepb.OrganizationUsage) *pb.DescribeUsageResponse { diff --git a/pkg/public/admin.go b/pkg/public/admin.go index af88a34a60..407dbcf755 100644 --- a/pkg/public/admin.go +++ b/pkg/public/admin.go @@ -350,54 +350,16 @@ func (s *Server) adminListOrganizations(w http.ResponseWriter, r *http.Request) search, limit, offset := parsePagination(r) sortBy, sortDirection := parseSorting(r) - var organizations []models.OrganizationWithCounts - var total int64 - err := telemetry.RunSpan(ctx, "organizations.list", func(ctx context.Context) error { - var listErr error - organizations, total, listErr = models.ListAllOrganizations(search, limit, offset, sortBy, sortDirection) - return listErr - }) + organizations, total, err := listAllOrganizations(ctx, search, limit, offset, sortBy, sortDirection) if err != nil { log.Errorf("admin: failed to list organizations: %v", err) http.Error(w, "Failed to list organizations", http.StatusInternalServerError) return } - type orgItem struct { - ID string `json:"id"` - Name string `json:"name"` - Description string `json:"description"` - CanvasCount int64 `json:"canvas_count"` - MemberCount int64 `json:"member_count"` - CreatedAt *string `json:"created_at,omitempty"` - } - - var items []orgItem - _ = telemetry.RunSpan(ctx, "organizations.serialize", func(ctx context.Context) error { - items = make([]orgItem, 0, len(organizations)) - for _, org := range organizations { - item := orgItem{ - ID: org.ID.String(), - Name: org.Name, - Description: org.Description, - CanvasCount: org.CanvasCount, - MemberCount: org.MemberCount, - } - - if org.CreatedAt != nil { - formatted := org.CreatedAt.Format(time.RFC3339) - item.CreatedAt = &formatted - } - - items = append(items, item) - } - - if span := trace.SpanFromContext(ctx); span.IsRecording() { - span.SetAttributes(attribute.Int("organizations.count", len(items))) - } + type orgItem = adminOrgItem - return nil - }) + items := serializeAdminOrganizations(ctx, organizations) w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(paginatedResponse{ @@ -771,3 +733,49 @@ func (s *Server) adminDisableOrgExperimentalFeature(w http.ResponseWriter, r *ht w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(map[string]string{"status": "disabled"}) } + +type adminOrgItem struct { + ID string `json:"id"` + Name string `json:"name"` + Description string `json:"description"` + CanvasCount int64 `json:"canvas_count"` + MemberCount int64 `json:"member_count"` + CreatedAt *string `json:"created_at,omitempty"` +} + +func listAllOrganizations(ctx context.Context, search string, limit, offset int, sortBy, sortDirection string) (organizations []models.OrganizationWithCounts, total int64, err error) { + ctx, done := telemetry.Span(ctx, "organizations.list") + defer done(&err) + + return models.ListAllOrganizations(search, limit, offset, sortBy, sortDirection) +} + +func serializeAdminOrganizations(ctx context.Context, organizations []models.OrganizationWithCounts) []adminOrgItem { + var err error + ctx, done := telemetry.Span(ctx, "organizations.serialize") + defer done(&err) + + items := make([]adminOrgItem, 0, len(organizations)) + for _, org := range organizations { + item := adminOrgItem{ + ID: org.ID.String(), + Name: org.Name, + Description: org.Description, + CanvasCount: org.CanvasCount, + MemberCount: org.MemberCount, + } + + if org.CreatedAt != nil { + formatted := org.CreatedAt.Format(time.RFC3339) + item.CreatedAt = &formatted + } + + items = append(items, item) + } + + if span := trace.SpanFromContext(ctx); span.IsRecording() { + span.SetAttributes(attribute.Int("organizations.count", len(items))) + } + + return items +} diff --git a/pkg/public/middleware/gateway_trace.go b/pkg/public/middleware/gateway_trace.go index 89678d334b..5cca1d4fb0 100644 --- a/pkg/public/middleware/gateway_trace.go +++ b/pkg/public/middleware/gateway_trace.go @@ -22,16 +22,17 @@ func TraceGatewayServe(ctx context.Context, w http.ResponseWriter, gateway http. return } - _ = telemetry.RunSpan(ctx, "grpc_gateway.serve", func(ctx context.Context) error { - capture := &tracedGatewayResponseWriter{ - ResponseWriter: w, - ctx: ctx, - statusCode: http.StatusOK, - } - gateway.ServeHTTP(capture, r.WithContext(ctx)) - capture.finish() - return nil - }) + var err error + ctx, done := telemetry.Span(ctx, "grpc_gateway.serve") + defer done(&err) + + capture := &tracedGatewayResponseWriter{ + ResponseWriter: w, + ctx: ctx, + statusCode: http.StatusOK, + } + gateway.ServeHTTP(capture, r.WithContext(ctx)) + capture.finish() } /* diff --git a/pkg/public/repository_file_download.go b/pkg/public/repository_file_download.go index afc9a1678b..5c3d556381 100644 --- a/pkg/public/repository_file_download.go +++ b/pkg/public/repository_file_download.go @@ -57,17 +57,7 @@ func (s *Server) handleRepositoryFileDownload(w http.ResponseWriter, r *http.Req return } - var allowed bool - err = telemetry.RunSpan(ctx, "repository.check_permission", func(ctx context.Context) error { - var checkErr error - allowed, checkErr = s.authService.CheckOrganizationPermission(ctx, - user.ID.String(), - user.OrganizationID.String(), - "canvases", - "read", - ) - return checkErr - }) + allowed, err := s.checkRepositoryReadPermission(ctx, user) if err != nil { log.Errorf("Failed to check permission: %v", err) http.Error(w, "Unauthorized", http.StatusForbidden) @@ -80,44 +70,13 @@ func (s *Server) handleRepositoryFileDownload(w http.ResponseWriter, r *http.Req return } - var canvas *models.Canvas - err = telemetry.RunSpan(ctx, "repository.find_canvas", func(ctx context.Context) error { - var findErr error - canvas, findErr = models.FindCanvasInTransaction(database.DB(ctx), user.OrganizationID, canvasID) - return findErr - }) + canvas, err := s.findRepositoryCanvas(ctx, user, canvasID) if err != nil { http.Error(w, "Canvas not found", http.StatusNotFound) return } - err = telemetry.RunSpan(ctx, "repository.read_file", func(ctx context.Context) error { - if span := trace.SpanFromContext(ctx); span.IsRecording() { - span.SetAttributes( - attribute.String("repository.file_path", path), - attribute.Bool("repository.staged", stage), - ) - if versionID != "" { - span.SetAttributes(attribute.String("repository.version_id", versionID)) - } - } - - if canvases.IsRepositorySpecFilePath(path) { - return s.writeRepositorySpecFile(ctx, w, user, canvas, canvasID, path, versionID, stage) - } - - if stage && versionID != "" { - written, readErr := s.tryWriteStagedRepositoryFile(ctx, w, user, canvas, canvasID, path, versionID) - if readErr != nil { - return readErr - } - if written { - return nil - } - } - - return s.writeRepositoryGitFile(ctx, w, user.OrganizationID, canvas, canvasID, path) - }) + err = s.readRepositoryFile(ctx, w, user, canvas, canvasID, path, versionID, stage) if errors.Is(err, errRepositoryFileNotFound) { http.Error(w, "File not found", http.StatusNotFound) return @@ -168,6 +127,65 @@ func (s *Server) writeRepositorySpecFile( return nil } +func (s *Server) checkRepositoryReadPermission(ctx context.Context, user *models.User) (allowed bool, err error) { + ctx, done := telemetry.Span(ctx, "repository.check_permission") + defer done(&err) + + return s.authService.CheckOrganizationPermission(ctx, + user.ID.String(), + user.OrganizationID.String(), + "canvases", + "read", + ) +} + +func (s *Server) findRepositoryCanvas(ctx context.Context, user *models.User, canvasID uuid.UUID) (canvas *models.Canvas, err error) { + ctx, done := telemetry.Span(ctx, "repository.find_canvas") + defer done(&err) + + return models.FindCanvasInTransaction(database.DB(ctx), user.OrganizationID, canvasID) +} + +func (s *Server) readRepositoryFile( + ctx context.Context, + w http.ResponseWriter, + user *models.User, + canvas *models.Canvas, + canvasID uuid.UUID, + path string, + versionID string, + stage bool, +) (err error) { + ctx, done := telemetry.Span(ctx, "repository.read_file") + defer done(&err) + + if span := trace.SpanFromContext(ctx); span.IsRecording() { + span.SetAttributes( + attribute.String("repository.file_path", path), + attribute.Bool("repository.staged", stage), + ) + if versionID != "" { + span.SetAttributes(attribute.String("repository.version_id", versionID)) + } + } + + if canvases.IsRepositorySpecFilePath(path) { + return s.writeRepositorySpecFile(ctx, w, user, canvas, canvasID, path, versionID, stage) + } + + if stage && versionID != "" { + written, readErr := s.tryWriteStagedRepositoryFile(ctx, w, user, canvas, canvasID, path, versionID) + if readErr != nil { + return readErr + } + if written { + return nil + } + } + + return s.writeRepositoryGitFile(ctx, w, user.OrganizationID, canvas, canvasID, path) +} + func (s *Server) tryWriteStagedRepositoryFile( ctx context.Context, w http.ResponseWriter, diff --git a/pkg/telemetry/span.go b/pkg/telemetry/span.go index 214455c33c..270d8bf208 100644 --- a/pkg/telemetry/span.go +++ b/pkg/telemetry/span.go @@ -2,9 +2,11 @@ package telemetry import ( "context" + "errors" "go.opentelemetry.io/otel/codes" "go.opentelemetry.io/otel/trace" + "gorm.io/gorm" ) func StartSpan(ctx context.Context, name string, opts ...trace.SpanStartOption) (context.Context, trace.Span) { @@ -15,15 +17,24 @@ func StartSpan(ctx context.Context, name string, opts ...trace.SpanStartOption) return Tracer().Start(ctx, name, opts...) } -func RunSpan(ctx context.Context, name string, fn func(context.Context) error) error { - ctx, span := StartSpan(ctx, name) - defer span.End() +// Span starts a span and returns ctx plus a done callback for defer done(&err). +func Span(ctx context.Context, name string, opts ...trace.SpanStartOption) (context.Context, func(*error)) { + ctx, span := StartSpan(ctx, name, opts...) + return ctx, func(err *error) { FinishSpan(span, err) } +} - err := fn(ctx) - if err != nil { - span.RecordError(err) - span.SetStatus(codes.Error, err.Error()) +func FinishSpan(span trace.Span, err *error) { + if err != nil && *err != nil && shouldMarkSpanError(*err) { + span.RecordError(*err) + span.SetStatus(codes.Error, (*err).Error()) } + span.End() +} + +func shouldMarkSpanError(err error) bool { + return !isRecordNotFound(err) +} - return err +func isRecordNotFound(err error) bool { + return errors.Is(err, gorm.ErrRecordNotFound) } diff --git a/pkg/telemetry/span_test.go b/pkg/telemetry/span_test.go new file mode 100644 index 0000000000..0469451593 --- /dev/null +++ b/pkg/telemetry/span_test.go @@ -0,0 +1,73 @@ +package telemetry + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + "gorm.io/gorm" +) + +func TestSpanDoneRecordNotFoundDoesNotMarkSpanError(t *testing.T) { + exporter := tracetest.NewInMemoryExporter() + cleanup := ConfigureTestTracerProvider(exporter) + t.Cleanup(cleanup) + + run := func() (err error) { + _, done := Span(context.Background(), "test.find") + defer done(&err) + + return gorm.ErrRecordNotFound + } + + require.ErrorIs(t, run(), gorm.ErrRecordNotFound) + + spans := exporter.GetSpans() + require.Len(t, spans, 1) + assert.Equal(t, codes.Unset, spans[0].Status.Code) +} + +func TestSpanDoneWrappedRecordNotFoundDoesNotMarkSpanError(t *testing.T) { + exporter := tracetest.NewInMemoryExporter() + cleanup := ConfigureTestTracerProvider(exporter) + t.Cleanup(cleanup) + + wrapped := errors.Join(errors.New("canvas lookup"), gorm.ErrRecordNotFound) + run := func() (err error) { + _, done := Span(context.Background(), "test.find") + defer done(&err) + + return wrapped + } + + require.Error(t, run()) + + spans := exporter.GetSpans() + require.Len(t, spans, 1) + assert.Equal(t, codes.Unset, spans[0].Status.Code) +} + +func TestSpanDoneOtherErrorMarksSpanError(t *testing.T) { + exporter := tracetest.NewInMemoryExporter() + cleanup := ConfigureTestTracerProvider(exporter) + t.Cleanup(cleanup) + + dbDown := errors.New("db down") + run := func() (err error) { + _, done := Span(context.Background(), "test.query") + defer done(&err) + + return dbDown + } + + require.ErrorIs(t, run(), dbDown) + + spans := exporter.GetSpans() + require.Len(t, spans, 1) + assert.Equal(t, codes.Error, spans[0].Status.Code) + assert.Equal(t, "db down", spans[0].Status.Description) +}