logger service key

Author: maxdml

dbos/conductor.go

@@ -102,7 +102,7 @@ func NewConductor(config ConductorConfig, dbosCtx *dbosContext) (*Conductor, err
 		pingInterval:  _PING_INTERVAL,
 		pingTimeout:   _PING_TIMEOUT,
 		reconnectWait: _INITIAL_RECONNECT_WAIT,
-		logger:        dbosCtx.logger,
+		logger:        dbosCtx.logger.With("service", "conductor"),
 	}
 
 	// Start with needsReconnect set to true so we connect on first run
@@ -114,8 +114,6 @@ func NewConductor(config ConductorConfig, dbosCtx *dbosContext) (*Conductor, err
 // Shutdown gracefully conductor
 func (c *Conductor) Shutdown(timeout time.Duration) {
 	c.stopOnce.Do(func() {
-		c.logger.Info("Shutting down conductor")
-
 		if c.pingCancel != nil {
 			c.pingCancel()
 		}

dbos/dbos.go

@@ -314,7 +314,7 @@ func NewDBOSContext(inputConfig Config) (DBOSContext, error) {
 	initExecutor.logger.Info("System database initialized")
 
 	// Initialize the queue runner and register DBOS internal queue
-	initExecutor.queueRunner = newQueueRunner()
+	initExecutor.queueRunner = newQueueRunner(initExecutor.logger)
 	NewWorkflowQueue(initExecutor, _DBOS_INTERNAL_QUEUE_NAME)
 
 	// Initialize conductor if API key is provided
@@ -414,8 +414,9 @@ func (c *dbosContext) Launch() error {
 // 2. Waits for the queue runner to complete processing
 // 3. Stops the workflow scheduler and waits for scheduled jobs to finish
 // 4. Shuts down the system database connection pool and notification listener
-// 5. Shuts down the admin server
-// 6. Marks the context as not launched
+// 5. Shuts down conductor
+// 6. Shuts down the admin server
+// 7. Marks the context as not launched
 //
 // Each step respects the provided timeout. If any component doesn't shut down within the timeout,
 // a warning is logged and the shutdown continues to the next component.

dbos/queue.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/base64"
 	"encoding/gob"
+	"log/slog"
 	"math"
 	"math/rand"
 	"time"
@@ -132,6 +133,8 @@ func NewWorkflowQueue(dbosCtx DBOSContext, name string, options ...QueueOption)
 }
 
 type queueRunner struct {
+	logger *slog.Logger
+
 	// Queue runner iteration parameters
 	baseInterval    float64
 	minInterval     float64
@@ -148,7 +151,7 @@ type queueRunner struct {
 	completionChan chan struct{}
 }
 
-func newQueueRunner() *queueRunner {
+func newQueueRunner(logger *slog.Logger) *queueRunner {
 	return &queueRunner{
 		baseInterval:          1.0,
 		minInterval:           1.0,
@@ -159,6 +162,7 @@ func newQueueRunner() *queueRunner {
 		jitterMax:             1.05,
 		workflowQueueRegistry: make(map[string]WorkflowQueue),
 		completionChan:        make(chan struct{}, 1),
+		logger:                logger.With("service", "queue_runner"),
 	}
 }
 
@@ -193,31 +197,31 @@ func (qr *queueRunner) run(ctx *dbosContext) {
 						hasBackoffError = true
 					}
 				} else {
-					ctx.logger.Error("Error dequeuing workflows from queue", "queue_name", queueName, "error", err)
+					qr.logger.Error("Error dequeuing workflows from queue", "queue_name", queueName, "error", err)
 				}
 				continue
 			}
 
 			if len(dequeuedWorkflows) > 0 {
-				ctx.logger.Debug("Dequeued workflows from queue", "queue_name", queueName, "workflows", dequeuedWorkflows)
+				qr.logger.Debug("Dequeued workflows from queue", "queue_name", queueName, "workflows", dequeuedWorkflows)
 			}
 			for _, workflow := range dequeuedWorkflows {
 				// Find the workflow in the registry
 
 				wfName, ok := ctx.workflowCustomNametoFQN.Load(workflow.name)
 				if !ok {
-					ctx.logger.Error("Workflow not found in registry", "workflow_name", workflow.name)
+					qr.logger.Error("Workflow not found in registry", "workflow_name", workflow.name)
 					continue
 				}
 
 				registeredWorkflowAny, exists := ctx.workflowRegistry.Load(wfName.(string))
 				if !exists {
-					ctx.logger.Error("workflow function not found in registry", "workflow_name", workflow.name)
+					qr.logger.Error("workflow function not found in registry", "workflow_name", workflow.name)
 					continue
 				}
 				registeredWorkflow, ok := registeredWorkflowAny.(workflowRegistryEntry)
 				if !ok {
-					ctx.logger.Error("invalid workflow registry entry type", "workflow_name", workflow.name)
+					qr.logger.Error("invalid workflow registry entry type", "workflow_name", workflow.name)
 					continue
 				}
 
@@ -226,20 +230,20 @@ func (qr *queueRunner) run(ctx *dbosContext) {
 				if len(workflow.input) > 0 {
 					inputBytes, err := base64.StdEncoding.DecodeString(workflow.input)
 					if err != nil {
-						ctx.logger.Error("failed to decode input for workflow", "workflow_id", workflow.id, "error", err)
+						qr.logger.Error("failed to decode input for workflow", "workflow_id", workflow.id, "error", err)
 						continue
 					}
 					buf := bytes.NewBuffer(inputBytes)
 					dec := gob.NewDecoder(buf)
 					if err := dec.Decode(&input); err != nil {
-						ctx.logger.Error("failed to decode input for workflow", "workflow_id", workflow.id, "error", err)
+						qr.logger.Error("failed to decode input for workflow", "workflow_id", workflow.id, "error", err)
 						continue
 					}
 				}
 
 				_, err := registeredWorkflow.wrappedFunction(ctx, input, WithWorkflowID(workflow.id))
 				if err != nil {
-					ctx.logger.Error("Error running queued workflow", "error", err)
+					qr.logger.Error("Error running queued workflow", "error", err)
 				}
 			}
 		}
@@ -260,7 +264,7 @@ func (qr *queueRunner) run(ctx *dbosContext) {
 		// Sleep with jittered interval, but allow early exit on context cancellation
 		select {
 		case <-ctx.Done():
-			ctx.logger.Info("Queue runner stopping due to context cancellation", "cause", context.Cause(ctx))
+			qr.logger.Info("Queue runner stopping due to context cancellation", "cause", context.Cause(ctx))
 			qr.completionChan <- struct{}{}
 			return
 		case <-time.After(sleepDuration):

dbos/system_database.go

@@ -243,7 +243,7 @@ func newSystemDatabase(ctx context.Context, databaseURL string, logger *slog.Log
 		notificationListenerConnection: notificationListenerConnection,
 		notificationsMap:               notificationsMap,
 		notificationLoopDone:           make(chan struct{}),
-		logger:                         logger,
+		logger:                         logger.With("service", "system_database"),
 	}, nil
 }
 
@@ -860,7 +860,6 @@ func (s *sysDB) garbageCollectWorkflows(ctx context.Context, input garbageCollec
 	return nil
 }
 
-
 func (s *sysDB) resumeWorkflow(ctx context.Context, workflowID string) error {
 	tx, err := s.pool.Begin(ctx)
 	if err != nil {