feat: improve sentry and logging

.trunk/configs/cspell.json

@@ -169,6 +169,7 @@
     "sandboxed",
     "santhosh",
     "schemagen",
+    "sentryhttp",
     "sjson",
     "somedbtype",
     "sqlclient",

CHANGELOG.md

@@ -9,6 +9,7 @@ NOTE: all releases may include dependency updates, not specifically mentioned
 - feat: integrate try-as library [#912](https://github.com/hypermodeinc/modus/pull/912)
 - fix: check topic actor status before publishing events [#918](https://github.com/hypermodeinc/modus/pull/918)
 - feat: update health endpoint [#924](https://github.com/hypermodeinc/modus/pull/924)
+- feat: improve sentry and logging [#925](https://github.com/hypermodeinc/modus/pull/925)
 
 ## 2025-06-25 - Runtime v0.18.2
 

runtime/actors/actorsystem.go

@@ -48,15 +48,15 @@ func Initialize(ctx context.Context) {
 
 	actorSystem, err := goakt.NewActorSystem("modus", opts...)
 	if err != nil {
-		logger.Fatal(ctx).Err(err).Msg("Failed to create actor system.")
+		logger.Fatal(ctx, err).Msg("Failed to create actor system.")
 	}
 
 	if err := startActorSystem(ctx, actorSystem); err != nil {
-		logger.Fatal(ctx).Err(err).Msg("Failed to start actor system.")
+		logger.Fatal(ctx, err).Msg("Failed to start actor system.")
 	}
 
 	if err := actorSystem.Inject(&wasmAgentInfo{}); err != nil {
-		logger.Fatal(ctx).Err(err).Msg("Failed to inject wasm agent info into actor system.")
+		logger.Fatal(ctx, err).Msg("Failed to inject wasm agent info into actor system.")
 	}
 
 	_actorSystem = actorSystem
@@ -67,12 +67,12 @@ func Initialize(ctx context.Context) {
 }
 
 func startActorSystem(ctx context.Context, actorSystem goakt.ActorSystem) error {
-	maxRetries := getIntFromEnv("MODUS_ACTOR_SYSTEM_START_MAX_RETRIES", 5)
-	retryInterval := getDurationFromEnv("MODUS_ACTOR_SYSTEM_START_RETRY_INTERVAL_SECONDS", 2, time.Second)
+	maxRetries := utils.GetIntFromEnv("MODUS_ACTOR_SYSTEM_START_MAX_RETRIES", 5)
+	retryInterval := utils.GetDurationFromEnv("MODUS_ACTOR_SYSTEM_START_RETRY_INTERVAL_SECONDS", 2, time.Second)
 
 	for i := range maxRetries {
 		if err := actorSystem.Start(ctx); err != nil {
-			logger.Warn(ctx).Err(err).Int("attempt", i+1).Msgf("Failed to start actor system, retrying in %s...", retryInterval)
+			logger.Warn(ctx, err).Int("attempt", i+1).Msgf("Failed to start actor system, retrying in %s...", retryInterval)
 			time.Sleep(retryInterval)
 			retryInterval *= 2 // Exponential backoff
 			continue
@@ -96,15 +96,15 @@ func loadAgentActors(ctx context.Context, plugin *plugins.Plugin) error {
 	for _, pid := range actors {
 		if a, ok := pid.Actor().(*wasmAgentActor); ok {
 			if err := goakt.Tell(ctx, pid, &messages.RestartAgent{}); err != nil {
-				logger.Err(ctx, err).Str("agent_id", a.agentId).Msg("Failed to send restart agent message to actor.")
+				logger.Error(ctx, err).Str("agent_id", a.agentId).Msg("Failed to send restart agent message to actor.")
 			}
 		}
 	}
 
 	// do this in a goroutine to avoid blocking the cluster engine startup
 	go func() {
 		if err := restoreAgentActors(ctx, plugin.Name()); err != nil {
-			logger.Err(ctx, err).Msg("Failed to restore agent actors.")
+			logger.Error(ctx, err).Msg("Failed to restore agent actors.")
 		}
 	}()
 
@@ -133,11 +133,11 @@ func restoreAgentActors(ctx context.Context, pluginName string) error {
 	for _, agent := range agents {
 		actorName := getActorName(agent.Id)
 		if exists, err := _actorSystem.ActorExists(ctx, actorName); err != nil {
-			logger.Err(ctx, err).Msgf("Failed to check if actor %s exists.", actorName)
+			logger.Error(ctx, err).Msgf("Failed to check if actor %s exists.", actorName)
 		} else if !exists {
 			err := spawnActorForAgent(ctx, pluginName, agent.Id, agent.Name, false)
 			if err != nil {
-				logger.Err(ctx, err).Msgf("Failed to spawn actor for agent %s.", agent.Id)
+				logger.Error(ctx, err).Msgf("Failed to spawn actor for agent %s.", agent.Id)
 			}
 		}
 	}
@@ -159,7 +159,7 @@ func beforeShutdown(ctx context.Context) error {
 			if actor.status == AgentStatusRunning {
 				ctx := actor.augmentContext(ctx, pid)
 				if err := actor.suspendAgent(ctx); err != nil {
-					logger.Err(ctx, err).Str("agent_id", actor.agentId).Msg("Failed to suspend agent actor.")
+					logger.Error(ctx, err).Str("agent_id", actor.agentId).Msg("Failed to suspend agent actor.")
 				}
 			}
 		}
@@ -169,7 +169,7 @@ func beforeShutdown(ctx context.Context) error {
 	for _, pid := range actors {
 		if _, ok := pid.Actor().(*subscriptionActor); ok && pid.IsRunning() {
 			if err := pid.Shutdown(ctx); err != nil {
-				logger.Err(ctx, err).Msgf("Failed to shutdown actor %s.", pid.Name())
+				logger.Error(ctx, err).Msgf("Failed to shutdown actor %s.", pid.Name())
 			}
 		}
 	}
@@ -201,7 +201,7 @@ func Shutdown(ctx context.Context) {
 	}
 
 	if err := _actorSystem.Stop(ctx); err != nil {
-		logger.Err(ctx, err).Msg("Failed to shutdown actor system.")
+		logger.Error(ctx, err).Msg("Failed to shutdown actor system.")
 	}
 
 	logger.Info(ctx).Msg("Actor system shutdown complete.")

runtime/actors/cluster.go

@@ -52,7 +52,7 @@ func clusterOptions(ctx context.Context) []goakt.Option {
 
 	disco, err := newDiscoveryProvider(ctx, clusterMode, discoveryPort)
 	if err != nil {
-		logger.Fatal(ctx).Err(err).Msg("Failed to create cluster discovery provider.")
+		logger.Fatal(ctx, err).Msg("Failed to create cluster discovery provider.")
 	}
 
 	return []goakt.Option{
@@ -166,9 +166,9 @@ func clusterPorts() (discoveryPort, remotingPort, peersPort int) {
 
 	// Get default ports dynamically, but use environment variables if set
 	ports := dynaport.Get(3)
-	discoveryPort = getIntFromEnv("MODUS_CLUSTER_DISCOVERY_PORT", ports[0])
-	remotingPort = getIntFromEnv("MODUS_CLUSTER_REMOTING_PORT", ports[1])
-	peersPort = getIntFromEnv("MODUS_CLUSTER_PEERS_PORT", ports[2])
+	discoveryPort = utils.GetIntFromEnv("MODUS_CLUSTER_DISCOVERY_PORT", ports[0])
+	remotingPort = utils.GetIntFromEnv("MODUS_CLUSTER_REMOTING_PORT", ports[1])
+	peersPort = utils.GetIntFromEnv("MODUS_CLUSTER_PEERS_PORT", ports[2])
 
 	return
 }
@@ -179,7 +179,7 @@ func clusterPorts() (discoveryPort, remotingPort, peersPort int) {
 // This value is also used for a sleep both on system startup and when spawning a new agent actor,
 // so it needs to be low enough to not be noticed by the user.
 func peerSyncInterval() time.Duration {
-	return getDurationFromEnv("MODUS_CLUSTER_PEER_SYNC_SECONDS", 1, time.Second)
+	return utils.GetDurationFromEnv("MODUS_CLUSTER_PEER_SYNC_SECONDS", 1, time.Second)
 }
 
 // nodesSyncInterval returns the interval at which the cluster forces a resync of the list of active nodes across the cluster.
@@ -189,7 +189,7 @@ func peerSyncInterval() time.Duration {
 // On each interval, the node will sync its list of nodes with the cluster, and update its local state accordingly.
 // The default is 10 seconds, which is a reasonable balance between responsiveness and network overhead.
 func nodesSyncInterval() time.Duration {
-	return getDurationFromEnv("MODUS_CLUSTER_NODES_SYNC_SECONDS", 10, time.Second)
+	return utils.GetDurationFromEnv("MODUS_CLUSTER_NODES_SYNC_SECONDS", 10, time.Second)
 }
 
 // partitionCount returns the number of partitions the cluster will use for actor distribution.
@@ -199,19 +199,19 @@ func nodesSyncInterval() time.Duration {
 // We'll use a slightly lower default of 13, which is still a prime number and should work well for most clusters.
 // The GoAkt default is 271, but this has been found to lead to other errors in practice.
 func partitionCount() uint64 {
-	return uint64(getIntFromEnv("MODUS_CLUSTER_PARTITION_COUNT", 13))
+	return uint64(utils.GetIntFromEnv("MODUS_CLUSTER_PARTITION_COUNT", 13))
 }
 
 // readTimeout returns the duration to wait for a cluster read operation before timing out.
 // The default is 1 second, which should usually not need to be changed.
 func readTimeout() time.Duration {
-	return getDurationFromEnv("MODUS_CLUSTER_READ_TIMEOUT_SECONDS", 1, time.Second)
+	return utils.GetDurationFromEnv("MODUS_CLUSTER_READ_TIMEOUT_SECONDS", 1, time.Second)
 }
 
 // writeTimeout returns the duration to wait for a cluster write operation before timing out.
 // The default is 1 second, which should usually not need to be changed.
 func writeTimeout() time.Duration {
-	return getDurationFromEnv("MODUS_CLUSTER_WRITE_TIMEOUT_SECONDS", 1, time.Second)
+	return utils.GetDurationFromEnv("MODUS_CLUSTER_WRITE_TIMEOUT_SECONDS", 1, time.Second)
 }
 
 func getPodLabels() map[string]string {

runtime/actors/misc.go

@@ -12,11 +12,8 @@ package actors
 import (
 	"context"
 	"fmt"
-	"os"
-	"strconv"
 	"time"
 
-	"github.com/hypermodeinc/modus/runtime/logger"
 	"github.com/hypermodeinc/modus/runtime/utils"
 	goakt "github.com/tochemey/goakt/v3/actor"
 
@@ -60,30 +57,3 @@ func ask(ctx context.Context, actorName string, message proto.Message, timeout t
 	}
 	return nil, fmt.Errorf("failed to get address or PID for actor %s", actorName)
 }
-
-// Retrieves an integer value from an environment variable.
-func getIntFromEnv(envVar string, defaultValue int) int {
-	str := os.Getenv(envVar)
-	if str == "" {
-		return defaultValue
-	}
-
-	value, err := strconv.Atoi(str)
-	if err != nil || value <= 0 {
-		logger.Warnf("Invalid value for %s. Using %d instead.", envVar, defaultValue)
-		return defaultValue
-	}
-
-	return value
-}
-
-// Retrieves a duration value from an environment variable.
-func getDurationFromEnv(envVar string, defaultValue int, unit time.Duration) time.Duration {
-	intVal := getIntFromEnv(envVar, defaultValue)
-	if intVal <= 0 {
-		duration := time.Duration(defaultValue) * unit
-		logger.Warnf("Invalid value for %s. Using %s instead.", envVar, duration)
-		return duration
-	}
-	return time.Duration(intVal) * unit
-}

runtime/actors/subscriber.go

@@ -83,11 +83,11 @@ func SubscribeForAgentEvents(ctx context.Context, agentId string, update func(da
 
 		unsubscribe := &goaktpb.Unsubscribe{Topic: topic}
 		if err := subActor.Tell(ctx, _actorSystem.TopicActor(), unsubscribe); err != nil {
-			logger.Err(ctx, err).Msg("Failed to unsubscribe from topic")
+			logger.Error(ctx, err).Msg("Failed to unsubscribe from topic")
 		}
 
 		if err := subActor.Shutdown(ctx); err != nil {
-			logger.Err(ctx, err).Msg("Failed to shut down subscription actor")
+			logger.Error(ctx, err).Msg("Failed to shut down subscription actor")
 		}
 	}()
 

runtime/actors/wasmagent.go

@@ -15,6 +15,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/getsentry/sentry-go"
 	"github.com/hypermodeinc/modus/runtime/db"
 	"github.com/hypermodeinc/modus/runtime/logger"
 	"github.com/hypermodeinc/modus/runtime/messages"
@@ -37,12 +38,16 @@ type wasmAgentActor struct {
 	host         wasmhost.WasmHost
 	module       wasm.Module
 	buffers      utils.OutputBuffers
+	sentryHub    *sentry.Hub
 	initializing bool
 }
 
 func (a *wasmAgentActor) PreStart(ac *goakt.Context) error {
 	ctx := ac.Context()
 
+	a.sentryHub = sentry.CurrentHub().Clone()
+	ctx = sentry.SetHubOnContext(ctx, a.sentryHub)
+
 	span, ctx := utils.NewSentrySpanForCurrentFunc(ctx)
 	defer span.Finish()
 
@@ -139,13 +144,14 @@ func (a *wasmAgentActor) Receive(rc *goakt.ReceiveContext) {
 
 func (a *wasmAgentActor) PostStop(ac *goakt.Context) error {
 	ctx := ac.Context()
+	ctx = sentry.SetHubOnContext(ctx, a.sentryHub)
 	span, ctx := utils.NewSentrySpanForCurrentFunc(ctx)
 	defer span.Finish()
 
 	// suspend the agent if it's not already suspended or terminated
 	if a.status != AgentStatusSuspended && a.status != AgentStatusTerminated {
 		if err := a.suspendAgent(ctx); err != nil {
-			logger.Err(ctx, err).Msg("Error suspending agent.")
+			logger.Error(ctx, err).Msg("Error suspending agent.")
 			// don't return on error - we'll still try to deactivate the agent
 		}
 	}
@@ -194,7 +200,7 @@ func (a *wasmAgentActor) handleAgentRequest(ctx context.Context, rc *goakt.Recei
 				response.Data = result
 			default:
 				err := fmt.Errorf("unexpected result type: %T", result)
-				logger.Err(ctx, err).Msg("Error handling message.")
+				logger.Error(ctx, err).Msg("Error handling message.")
 				return err
 			}
 		}
@@ -209,7 +215,7 @@ func (a *wasmAgentActor) handleAgentRequest(ctx context.Context, rc *goakt.Recei
 
 	// save the state after handling the message to ensure the state is up to date in case of hard termination
 	if err := a.saveState(ctx); err != nil {
-		logger.Err(ctx, err).Msg("Error saving agent state.")
+		logger.Error(ctx, err).Msg("Error saving agent state.")
 	}
 
 	return nil
@@ -299,6 +305,9 @@ func (a *wasmAgentActor) augmentContext(ctx context.Context, pid *goakt.PID) con
 	if ctx.Value(pidContextKey{}) == nil {
 		ctx = context.WithValue(ctx, pidContextKey{}, pid)
 	}
+	if !sentry.HasHubOnContext(ctx) {
+		ctx = sentry.SetHubOnContext(ctx, a.sentryHub)
+	}
 	return ctx
 }
 

runtime/app/app.go

@@ -77,7 +77,7 @@ func SetShuttingDown() {
 }
 
 // GetRootSourcePath returns the root path of the source code.
-// It is used to trim the paths in stack traces when included in telemetry.
+// It is used to trim the paths in stack traces when included in utils.
 func GetRootSourcePath() string {
 	_, filename, _, ok := runtime.Caller(0)
 	if !ok {

runtime/db/db.go

@@ -51,15 +51,15 @@ func Stop(ctx context.Context) {
 
 func logDbWarningOrError(ctx context.Context, err error, msg string) {
 	if _, ok := err.(*pgconn.ConnectError); ok {
-		logger.Warn(ctx).Err(err).Msgf("Database connection error. %s", msg)
+		logger.Warn(ctx, err).Msgf("Database connection error. %s", msg)
 	} else if errors.Is(err, errDbNotConfigured) {
 		if !useModusDB() {
 			logger.Warn(ctx).Msgf("Database has not been configured. %s", msg)
 		}
 	} else {
 		// not really an error, but we log it as such
 		// but user-visible so it doesn't flag in Sentry
-		logger.Err(ctx, err).Bool("user_visible", true).Msg(msg)
+		logger.Error(ctx, err).Bool("user_visible", true).Msg(msg)
 	}
 }
 
@@ -71,7 +71,7 @@ func Initialize(ctx context.Context) {
 	// this will initialize the pool and start the worker
 	_, err := globalRuntimePostgresWriter.GetPool(ctx)
 	if err != nil {
-		logger.Warn(ctx).Err(err).Msg("Metadata database is not available.")
+		logger.Warn(ctx, err).Msg("Metadata database is not available.")
 	}
 	go globalRuntimePostgresWriter.worker(ctx)
 }

runtime/db/modusdb.go

@@ -40,7 +40,7 @@ func InitModusDb(ctx context.Context) {
 	}
 
 	if eng, err := modusgraph.NewEngine(modusgraph.NewDefaultConfig(dataDir)); err != nil {
-		logger.Fatal(ctx).Err(err).Msg("Failed to initialize the local modusGraph database.")
+		logger.Fatal(ctx, err).Msg("Failed to initialize the local modusGraph database.")
 	} else {
 		GlobalModusDbEngine = eng
 	}
@@ -58,15 +58,15 @@ func addToGitIgnore(ctx context.Context, rootPath, contents string) {
 	// if .gitignore file does not exist, create it and add contents to it
 	if _, err := os.Stat(gitIgnorePath); errors.Is(err, os.ErrNotExist) {
 		if err := os.WriteFile(gitIgnorePath, []byte(contents+"\n"), 0644); err != nil {
-			logger.Err(ctx, err).Msg("Failed to create .gitignore file.")
+			logger.Error(ctx, err).Msg("Failed to create .gitignore file.")
 		}
 		return
 	}
 
 	// check if contents are already in the .gitignore file
 	file, err := os.Open(gitIgnorePath)
 	if err != nil {
-		logger.Err(ctx, err).Msg("Failed to open .gitignore file.")
+		logger.Error(ctx, err).Msg("Failed to open .gitignore file.")
 		return
 	}
 	defer file.Close()
@@ -80,11 +80,11 @@ func addToGitIgnore(ctx context.Context, rootPath, contents string) {
 	// contents are not in the file, so append them
 	file, err = os.OpenFile(gitIgnorePath, os.O_APPEND|os.O_WRONLY, 0644)
 	if err != nil {
-		logger.Err(ctx, err).Msg("Failed to open .gitignore file.")
+		logger.Error(ctx, err).Msg("Failed to open .gitignore file.")
 		return
 	}
 	defer file.Close()
 	if _, err := file.WriteString("\n" + contents + "\n"); err != nil {
-		logger.Err(ctx, err).Msg("Failed to append " + contents + " to .gitignore file.")
+		logger.Error(ctx, err).Msg("Failed to append " + contents + " to .gitignore file.")
 	}
 }

runtime/dgraphclient/dgraph.go

@@ -63,7 +63,7 @@ func (dc *dgraphConnector) execute(ctx context.Context, req *Request) (*Response
 		tx := dc.dgClient.NewReadOnlyTxn()
 		defer func() {
 			if err := tx.Discard(ctx); err != nil {
-				logger.Warn(ctx).Err(err).Msg("Error discarding transaction.")
+				logger.Warn(ctx, err).Msg("Error discarding transaction.")
 				return
 			}
 
@@ -86,7 +86,7 @@ func (dc *dgraphConnector) execute(ctx context.Context, req *Request) (*Response
 	tx := dc.dgClient.NewTxn()
 	defer func() {
 		if err := tx.Discard(ctx); err != nil {
-			logger.Warn(ctx).Err(err).Msg("Error discarding transaction.")
+			logger.Warn(ctx, err).Msg("Error discarding transaction.")
 			return
 		}