Fix OTel span attribute naming convention

api/v1/server/handlers/v1/tasks/list_logs.go

@@ -8,6 +8,7 @@ import (
 	"github.com/hatchet-dev/hatchet/pkg/repository/postgres/sqlchelpers"
 	v1 "github.com/hatchet-dev/hatchet/pkg/repository/v1"
 	"github.com/hatchet-dev/hatchet/pkg/repository/v1/sqlcv1"
+	"github.com/hatchet-dev/hatchet/pkg/telemetry"
 
 	transformers "github.com/hatchet-dev/hatchet/api/v1/server/oas/transformers/v1"
 )
@@ -17,12 +18,25 @@ func (t *TasksService) V1LogLineList(ctx echo.Context, request gen.V1LogLineList
 	tenantId := sqlchelpers.UUIDToStr(tenant.ID)
 	task := ctx.Get("task").(*sqlcv1.V1TasksOlap)
 
-	logLines, err := t.config.V1.Logs().ListLogLines(ctx.Request().Context(), tenantId, task.ID, task.InsertedAt, &v1.ListLogsOpts{})
+	reqCtx, span := telemetry.NewSpan(ctx.Request().Context(), "GET /api/v1/stable/tasks/{task}/logs")
+	defer span.End()
+
+	telemetry.WithAttributes(span,
+		telemetry.AttributeKV{Key: "tenant.id", Value: tenantId},
+		telemetry.AttributeKV{Key: "task.id", Value: task.ID},
+	)
+
+	logLines, err := t.config.V1.Logs().ListLogLines(reqCtx, tenantId, task.ID, task.InsertedAt, &v1.ListLogsOpts{})
 
 	if err != nil {
+		span.RecordError(err)
 		return nil, err
 	}
 
+	telemetry.WithAttributes(span,
+		telemetry.AttributeKV{Key: "log_lines.count", Value: len(logLines)},
+	)
+
 	rows := make([]gen.V1LogLine, len(logLines))
 
 	for i, log := range logLines {

api/v1/server/handlers/v1/workflow-runs/list.go

@@ -9,11 +9,11 @@ import (
 	"github.com/labstack/echo/v4"
 
 	"github.com/hatchet-dev/hatchet/api/v1/server/oas/gen"
-	"github.com/hatchet-dev/hatchet/internal/telemetry"
 	"github.com/hatchet-dev/hatchet/pkg/repository/postgres/dbsqlc"
 	"github.com/hatchet-dev/hatchet/pkg/repository/postgres/sqlchelpers"
 	v1 "github.com/hatchet-dev/hatchet/pkg/repository/v1"
 	"github.com/hatchet-dev/hatchet/pkg/repository/v1/sqlcv1"
+	"github.com/hatchet-dev/hatchet/pkg/telemetry"
 
 	transformers "github.com/hatchet-dev/hatchet/api/v1/server/oas/transformers/v1"
 )

cmd/hatchet-api/api/run.go

@@ -6,8 +6,8 @@ import (
 	"strings"
 
 	"github.com/hatchet-dev/hatchet/api/v1/server/run"
-	"github.com/hatchet-dev/hatchet/internal/telemetry"
 	"github.com/hatchet-dev/hatchet/pkg/config/loader"
+	"github.com/hatchet-dev/hatchet/pkg/telemetry"
 )
 
 func init() {

cmd/hatchet-engine/engine/run.go

@@ -28,11 +28,11 @@ import (
 	schedulerv1 "github.com/hatchet-dev/hatchet/internal/services/scheduler/v1"
 	"github.com/hatchet-dev/hatchet/internal/services/ticker"
 	"github.com/hatchet-dev/hatchet/internal/services/webhooks"
-	"github.com/hatchet-dev/hatchet/internal/telemetry"
 	"github.com/hatchet-dev/hatchet/pkg/config/loader"
 	"github.com/hatchet-dev/hatchet/pkg/config/server"
 	"github.com/hatchet-dev/hatchet/pkg/config/shared"
 	"github.com/hatchet-dev/hatchet/pkg/repository/cache"
+	"github.com/hatchet-dev/hatchet/pkg/telemetry"
 	"github.com/rs/zerolog"
 
 	"golang.org/x/sync/errgroup"

internal/msgqueue/postgres/msgqueue.go

@@ -13,10 +13,10 @@ import (
 	"github.com/hatchet-dev/hatchet/internal/cache"
 	"github.com/hatchet-dev/hatchet/internal/msgqueue"
 	"github.com/hatchet-dev/hatchet/internal/queueutils"
-	"github.com/hatchet-dev/hatchet/internal/telemetry"
 	"github.com/hatchet-dev/hatchet/pkg/logger"
 	"github.com/hatchet-dev/hatchet/pkg/repository"
 	"github.com/hatchet-dev/hatchet/pkg/repository/postgres/dbsqlc"
+	"github.com/hatchet-dev/hatchet/pkg/telemetry"
 )
 
 type PostgresMessageQueue struct {

internal/msgqueue/rabbitmq/rabbitmq.go

@@ -15,9 +15,9 @@ import (
 	"github.com/rs/zerolog"
 
 	"github.com/hatchet-dev/hatchet/internal/msgqueue"
-	"github.com/hatchet-dev/hatchet/internal/telemetry"
 	"github.com/hatchet-dev/hatchet/pkg/logger"
 	"github.com/hatchet-dev/hatchet/pkg/random"
+	"github.com/hatchet-dev/hatchet/pkg/telemetry"
 )
 
 const MAX_RETRY_COUNT = 15

internal/msgqueue/v1/postgres/msgqueue.go

@@ -14,10 +14,10 @@ import (
 	"github.com/hatchet-dev/hatchet/internal/cache"
 	msgqueue "github.com/hatchet-dev/hatchet/internal/msgqueue/v1"
 	"github.com/hatchet-dev/hatchet/internal/queueutils"
-	"github.com/hatchet-dev/hatchet/internal/telemetry"
 	"github.com/hatchet-dev/hatchet/pkg/logger"
 	"github.com/hatchet-dev/hatchet/pkg/repository"
 	"github.com/hatchet-dev/hatchet/pkg/repository/postgres/dbsqlc"
+	"github.com/hatchet-dev/hatchet/pkg/telemetry"
 )
 
 type PostgresMessageQueue struct {
@@ -111,7 +111,7 @@ func (p *PostgresMessageQueue) SendMessage(ctx context.Context, queue msgqueue.Q
 	ctx, span := telemetry.NewSpan(ctx, "PostgresMessageQueue.SendMessage")
 	defer span.End()
 
-	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: task.TenantID})
+	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: task.TenantID})
 
 	err := p.addMessage(ctx, queue, task)
 
@@ -128,7 +128,7 @@ func (p *PostgresMessageQueue) addMessage(ctx context.Context, queue msgqueue.Qu
 	ctx, span := telemetry.NewSpan(ctx, "add-message")
 	defer span.End()
 
-	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: task.TenantID})
+	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: task.TenantID})
 
 	// inject otel carrier into the message
 	if task.OtelCarrier == nil {

internal/msgqueue/v1/rabbitmq/rabbitmq.go

@@ -17,9 +17,9 @@ import (
 
 	msgqueue "github.com/hatchet-dev/hatchet/internal/msgqueue/v1"
 	"github.com/hatchet-dev/hatchet/internal/queueutils"
-	"github.com/hatchet-dev/hatchet/internal/telemetry"
 	"github.com/hatchet-dev/hatchet/pkg/logger"
 	"github.com/hatchet-dev/hatchet/pkg/random"
+	"github.com/hatchet-dev/hatchet/pkg/telemetry"
 )
 
 const MAX_RETRY_COUNT = 15
@@ -281,7 +281,7 @@ func (t *MessageQueueImpl) pubMessage(ctx context.Context, q msgqueue.Queue, msg
 	ctx, span := telemetry.NewSpanWithCarrier(ctx, "publish-message", otelCarrier)
 	defer span.End()
 
-	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: msg.TenantID})
+	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: msg.TenantID})
 
 	msg.SetOtelCarrier(otelCarrier)
 

internal/services/controllers/events/controller.go

@@ -13,11 +13,11 @@ import (
 	"github.com/hatchet-dev/hatchet/internal/datautils"
 	"github.com/hatchet-dev/hatchet/internal/msgqueue"
 	"github.com/hatchet-dev/hatchet/internal/services/shared/tasktypes"
-	"github.com/hatchet-dev/hatchet/internal/telemetry"
 	"github.com/hatchet-dev/hatchet/pkg/constants"
 	"github.com/hatchet-dev/hatchet/pkg/logger"
 	"github.com/hatchet-dev/hatchet/pkg/repository"
 	"github.com/hatchet-dev/hatchet/pkg/repository/postgres/sqlchelpers"
+	"github.com/hatchet-dev/hatchet/pkg/telemetry"
 )
 
 type EventsController interface {
@@ -165,7 +165,7 @@ func (ec *EventsControllerImpl) handleTask(ctx context.Context, task *msgqueue.M
 	err = ec.dv.DecodeAndValidate(task.Metadata, &metadata)
 
 	if err == nil {
-		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: metadata.TenantId})
+		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: metadata.TenantId})
 	} else {
 		return fmt.Errorf("could not decode task metadata: %w", err)
 	}

internal/services/controllers/jobs/controller.go

@@ -22,14 +22,14 @@ import (
 	"github.com/hatchet-dev/hatchet/internal/services/partition"
 	"github.com/hatchet-dev/hatchet/internal/services/shared/recoveryutils"
 	"github.com/hatchet-dev/hatchet/internal/services/shared/tasktypes"
-	"github.com/hatchet-dev/hatchet/internal/telemetry"
-	"github.com/hatchet-dev/hatchet/internal/telemetry/servertel"
 	"github.com/hatchet-dev/hatchet/pkg/config/shared"
 	hatcheterrors "github.com/hatchet-dev/hatchet/pkg/errors"
 	"github.com/hatchet-dev/hatchet/pkg/logger"
 	"github.com/hatchet-dev/hatchet/pkg/repository"
 	"github.com/hatchet-dev/hatchet/pkg/repository/postgres/dbsqlc"
 	"github.com/hatchet-dev/hatchet/pkg/repository/postgres/sqlchelpers"
+	"github.com/hatchet-dev/hatchet/pkg/telemetry"
+	"github.com/hatchet-dev/hatchet/pkg/telemetry/servertel"
 )
 
 type JobsController interface {
@@ -319,7 +319,7 @@ func (ec *JobsControllerImpl) handleJobRunQueued(ctx context.Context, task *msgq
 	err = ec.dv.DecodeAndValidate(task.Metadata, &metadata)
 
 	if err == nil {
-		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: metadata.TenantId})
+		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: metadata.TenantId})
 	} else {
 		return fmt.Errorf("could not decode job task metadata: %w", err)
 	}
@@ -377,7 +377,7 @@ func (ec *JobsControllerImpl) handleJobRunCancelled(ctx context.Context, task *m
 	err = ec.dv.DecodeAndValidate(task.Metadata, &metadata)
 
 	if err == nil {
-		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: metadata.TenantId})
+		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: metadata.TenantId})
 	} else {
 		return fmt.Errorf("could not decode job task metadata: %w", err)
 	}
@@ -434,7 +434,7 @@ func (ec *JobsControllerImpl) handleStepRunRetry(ctx context.Context, task *msgq
 	err = ec.dv.DecodeAndValidate(task.Metadata, &metadata)
 
 	if err == nil {
-		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: metadata.TenantId})
+		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: metadata.TenantId})
 	} else {
 		return fmt.Errorf("could not decode job task metadata: %w", err)
 	}
@@ -504,7 +504,7 @@ func (ec *JobsControllerImpl) handleStepRunReplay(ctx context.Context, task *msg
 	err = ec.dv.DecodeAndValidate(task.Metadata, &metadata)
 
 	if err == nil {
-		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: metadata.TenantId})
+		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: metadata.TenantId})
 	} else {
 		return fmt.Errorf("could not decode job task metadata: %w", err)
 	}
@@ -602,7 +602,7 @@ func (ec *JobsControllerImpl) handleStepRunQueued(ctx context.Context, task *msg
 	err = ec.dv.DecodeAndValidate(task.Metadata, &metadata)
 
 	if err == nil {
-		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: metadata.TenantId})
+		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: metadata.TenantId})
 	} else {
 		return fmt.Errorf("could not decode job task metadata: %w", err)
 	}
@@ -699,7 +699,7 @@ func (ec *JobsControllerImpl) runStepRunReassignTenant(ctx context.Context, tena
 	ctx, span := telemetry.NewSpan(ctx, "handle-step-run-reassign")
 	defer span.End()
 
-	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: tenantId})
+	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: tenantId})
 
 	_, stepRunsToFail, err := ec.repo.StepRun().ListStepRunsToReassign(ctx, tenantId)
 
@@ -732,7 +732,7 @@ func (ec *JobsControllerImpl) queueStepRun(ctx context.Context, tenantId, stepId
 	ctx, span := telemetry.NewSpan(ctx, "queue-step-run")
 	defer span.End()
 
-	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: tenantId})
+	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: tenantId})
 
 	// add the rendered data to the step run
 	stepRun, err := ec.repo.StepRun().GetStepRunForEngine(ctx, tenantId, stepRunId)
@@ -944,7 +944,7 @@ func (ec *JobsControllerImpl) handleStepRunStarted(ctx context.Context, task *ms
 	err = ec.dv.DecodeAndValidate(task.Metadata, &metadata)
 
 	if err == nil {
-		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: metadata.TenantId})
+		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: metadata.TenantId})
 	} else {
 		return fmt.Errorf("could not decode step run started task metadata: %w", err)
 	}
@@ -981,7 +981,7 @@ func (ec *JobsControllerImpl) handleStepRunAcked(ctx context.Context, task *msgq
 	err = ec.dv.DecodeAndValidate(task.Metadata, &metadata)
 
 	if err == nil {
-		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: metadata.TenantId})
+		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: metadata.TenantId})
 	} else {
 		return fmt.Errorf("could not decode step run started task metadata: %w", err)
 	}
@@ -1018,7 +1018,7 @@ func (ec *JobsControllerImpl) handleStepRunFinished(ctx context.Context, task *m
 	err = ec.dv.DecodeAndValidate(task.Metadata, &metadata)
 
 	if err == nil {
-		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: metadata.TenantId})
+		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: metadata.TenantId})
 	} else {
 		return fmt.Errorf("could not decode step run finished task metadata: %w", err)
 	}
@@ -1088,7 +1088,7 @@ func (ec *JobsControllerImpl) handleStepRunFailed(ctx context.Context, task *msg
 	err = ec.dv.DecodeAndValidate(task.Metadata, &metadata)
 
 	if err == nil {
-		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: metadata.TenantId})
+		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: metadata.TenantId})
 	} else {
 		return fmt.Errorf("could not decode step run failed task metadata: %w", err)
 	}
@@ -1218,7 +1218,7 @@ func (ec *JobsControllerImpl) handleStepRunTimedOut(ctx context.Context, task *m
 	err = ec.dv.DecodeAndValidate(task.Metadata, &metadata)
 
 	if err == nil {
-		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: metadata.TenantId})
+		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: metadata.TenantId})
 	} else {
 		return fmt.Errorf("could not decode step run timed out task metadata: %w", err)
 	}
@@ -1242,7 +1242,7 @@ func (ec *JobsControllerImpl) handleStepRunCancel(ctx context.Context, task *msg
 	err = ec.dv.DecodeAndValidate(task.Metadata, &metadata)
 
 	if err == nil {
-		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: metadata.TenantId})
+		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: metadata.TenantId})
 	} else {
 		return fmt.Errorf("could not decode step run notify cancel task metadata: %w", err)
 	}
@@ -1254,7 +1254,7 @@ func (ec *JobsControllerImpl) cancelStepRun(ctx context.Context, tenantId, stepR
 	ctx, span := telemetry.NewSpan(ctx, "cancel-step-run")
 	defer span.End()
 
-	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: tenantId})
+	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: tenantId})
 
 	// cancel current step run
 	now := time.Now().UTC()

internal/services/controllers/jobs/queue.go

@@ -15,11 +15,11 @@ import (
 	"github.com/hatchet-dev/hatchet/internal/services/partition"
 	"github.com/hatchet-dev/hatchet/internal/services/shared/recoveryutils"
 	"github.com/hatchet-dev/hatchet/internal/services/shared/tasktypes"
-	"github.com/hatchet-dev/hatchet/internal/telemetry"
 	hatcheterrors "github.com/hatchet-dev/hatchet/pkg/errors"
 	"github.com/hatchet-dev/hatchet/pkg/repository"
 	"github.com/hatchet-dev/hatchet/pkg/repository/postgres/dbsqlc"
 	"github.com/hatchet-dev/hatchet/pkg/repository/postgres/sqlchelpers"
+	"github.com/hatchet-dev/hatchet/pkg/telemetry"
 )
 
 type queue struct {
@@ -193,7 +193,7 @@ func (q *queue) handleCheckQueue(ctx context.Context, task *msgqueue.Message) er
 	err := q.dv.DecodeAndValidate(task.Metadata, &metadata)
 
 	if err == nil {
-		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: metadata.TenantId})
+		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: metadata.TenantId})
 	} else {
 		return fmt.Errorf("could not decode check queue metadata: %w", err)
 	}
@@ -235,7 +235,7 @@ func (q *queue) processStepRunUpdatesV2(ctx context.Context, tenantId string) (b
 	ctx, span := telemetry.NewSpan(ctx, "process-step-run-updates-v2")
 	defer span.End()
 
-	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: tenantId})
+	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: tenantId})
 
 	dbCtx, cancel := context.WithTimeout(ctx, 300*time.Second)
 	defer cancel()
@@ -355,7 +355,7 @@ func (q *queue) processStepRunTimeouts(ctx context.Context, tenantId string) (bo
 	ctx, span := telemetry.NewSpan(ctx, "handle-step-run-timeout")
 	defer span.End()
 
-	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: tenantId})
+	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: tenantId})
 
 	shouldContinue, stepRuns, err := q.repo.StepRun().ListStepRunsToTimeout(ctx, tenantId)
 
@@ -376,7 +376,7 @@ func (q *queue) processStepRunTimeouts(ctx context.Context, tenantId string) (bo
 		scheduleCtx, span := telemetry.NewSpan(scheduleCtx, "handle-step-run-timeout-step-run")
 		defer span.End()
 
-		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: tenantId})
+		telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: tenantId})
 
 		for i := range group {
 			stepRunCp := group[i]
@@ -408,7 +408,7 @@ func (q *queue) processStepRunRetries(ctx context.Context, tenantId string) (boo
 	ctx, span := telemetry.NewSpan(ctx, "handle-step-run-timeout")
 	defer span.End()
 
-	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: tenantId})
+	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: tenantId})
 
 	shouldContinue, err := q.repo.StepRun().RetryStepRuns(ctx, tenantId)
 

internal/services/controllers/retention/events.go

@@ -5,9 +5,9 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/hatchet-dev/hatchet/internal/telemetry"
 	"github.com/hatchet-dev/hatchet/pkg/repository/postgres/dbsqlc"
 	"github.com/hatchet-dev/hatchet/pkg/repository/postgres/sqlchelpers"
+	"github.com/hatchet-dev/hatchet/pkg/telemetry"
 )
 
 func (rc *RetentionControllerImpl) runDeleteExpiredEvents(ctx context.Context) func() {
@@ -45,7 +45,7 @@ func (wc *RetentionControllerImpl) runDeleteExpiredEventsTenant(ctx context.Cont
 
 	tenantId := sqlchelpers.UUIDToStr(tenant.ID)
 
-	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: tenantId})
+	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: tenantId})
 
 	createdBefore, err := GetDataRetentionExpiredTime(tenant.DataRetentionPeriod)
 
@@ -80,7 +80,7 @@ func (wc *RetentionControllerImpl) runClearDeletedEventsPayloadTenant(ctx contex
 
 	tenantId := sqlchelpers.UUIDToStr(tenant.ID)
 
-	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant_id", Value: tenantId})
+	telemetry.WithAttributes(span, telemetry.AttributeKV{Key: "tenant.id", Value: tenantId})
 
 	// keep deleting until the context is done
 	for {