feat: add a metric to ActorSystem and PID (#594)

Author: Tochemey

actors/actor_system.go

@@ -67,6 +67,9 @@ import (
 
 // ActorSystem defines the contract of an actor system
 type ActorSystem interface {
+	// Metric returns the actor system metric.
+	// The metric does not include any cluster data
+	Metric(ctx context.Context) *Metric
 	// Name returns the actor system name
 	Name() string
 	// Actors returns the list of Actors that are alive on a given running node.
@@ -286,6 +289,9 @@ type actorSystem struct {
 	rebalancing     *atomic.Bool
 	rebalanceLocker *sync.Mutex
 	shutdownHooks   []ShutdownHook
+
+	actorsCounter      *atomic.Uint64
+	deadlettersCounter *atomic.Uint64
 }
 
 var (
@@ -332,6 +338,8 @@ func NewActorSystem(name string, opts ...Option) (ActorSystem, error) {
 		shutdownHooks:          make([]ShutdownHook, 0),
 		nodesInprocess:         goset.NewSet[string](),
 		rebalanceLocker:        &sync.Mutex{},
+		actorsCounter:          atomic.NewUint64(0),
+		deadlettersCounter:     atomic.NewUint64(0),
 	}
 
 	system.started.Store(false)
@@ -456,6 +464,20 @@ func (x *actorSystem) Stop(ctx context.Context) error {
 	return x.shutdown(ctx)
 }
 
+// Metrics returns the actor system metrics.
+// The metrics does not include any cluster data
+func (x *actorSystem) Metric(ctx context.Context) *Metric {
+	if x.started.Load() {
+		x.getSetDeadlettersCount(ctx)
+		return &Metric{
+			deadlettersCount: int64(x.deadlettersCounter.Load()),
+			actorsCount:      int64(x.actorsCounter.Load()),
+			uptime:           x.Uptime(),
+		}
+	}
+	return nil
+}
+
 // Running returns true when the actor system is running
 func (x *actorSystem) Running() bool {
 	return x.started.Load()
@@ -597,7 +619,7 @@ func (x *actorSystem) InCluster() bool {
 // NumActors returns the total number of active actors on a given running node.
 // This does not account for the total number of actors in the cluster
 func (x *actorSystem) NumActors() uint64 {
-	return uint64(len(x.Actors()))
+	return x.actorsCounter.Load()
 }
 
 // Spawn creates or returns the instance of a given actor in the system
@@ -625,6 +647,7 @@ func (x *actorSystem) Spawn(ctx context.Context, name string, actor Actor, opts
 		return nil, err
 	}
 
+	x.actorsCounter.Inc()
 	// add the given actor to the tree and supervise it
 	_ = x.actors.AddNode(x.userGuardian, pid)
 	x.actors.AddWatcher(pid, x.janitor)
@@ -661,6 +684,7 @@ func (x *actorSystem) SpawnNamedFromFunc(ctx context.Context, name string, recei
 		return nil, err
 	}
 
+	x.actorsCounter.Inc()
 	_ = x.actors.AddNode(x.userGuardian, pid)
 	x.actors.AddWatcher(pid, x.janitor)
 	x.broadcastActor(pid)
@@ -691,8 +715,8 @@ func (x *actorSystem) Kill(ctx context.Context, name string) error {
 	pidNode, exist := x.actors.GetNode(actorAddress.String())
 	if exist {
 		pid := pidNode.GetValue()
-		// stop the given actor. No need to record error in the span context
-		// because the shutdown method is taking care of that
+		// decrement the actors count since we are stopping the actor
+		x.actorsCounter.Dec()
 		return pid.Shutdown(ctx)
 	}
 
@@ -1226,7 +1250,7 @@ func (x *actorSystem) handleRemoteTell(ctx context.Context, to *PID, message pro
 	return Tell(ctx, to, message)
 }
 
-// getRootGuardian returns the system root guardian
+// getRootGuardian returns the system rootGuardian guardian
 func (x *actorSystem) getRootGuardian() *PID {
 	x.locker.Lock()
 	rootGuardian := x.rootGuardian
@@ -1779,16 +1803,16 @@ func (x *actorSystem) setHostPort() error {
 	return nil
 }
 
-// spawnRootGuardian creates the root guardian
+// spawnRootGuardian creates the rootGuardian guardian
 func (x *actorSystem) spawnRootGuardian(ctx context.Context) error {
 	var err error
 	actorName := x.reservedName(rootGuardianType)
 	x.rootGuardian, err = x.configPID(ctx, actorName, newRootGuardian())
 	if err != nil {
-		return fmt.Errorf("actor=%s failed to start root guardian: %w", actorName, err)
+		return fmt.Errorf("actor=%s failed to start rootGuardian guardian: %w", actorName, err)
 	}
 
-	// rootGuardian is the root node of the actors tree
+	// rootGuardian is the rootGuardian node of the actors tree
 	_ = x.actors.AddNode(NoSender, x.rootGuardian)
 	return nil
 }
@@ -1880,7 +1904,7 @@ func (x *actorSystem) spawnRebalancer(ctx context.Context) error {
 // spawnDeadletters creates the deadletters synthetic actor
 func (x *actorSystem) spawnDeadletters(ctx context.Context) error {
 	var err error
-	actorName := x.reservedName(deadletters)
+	actorName := x.reservedName(deadlettersType)
 	x.deadletters, err = x.configPID(ctx,
 		actorName,
 		newDeadLetters(),
@@ -1935,6 +1959,25 @@ func (x *actorSystem) cleanupCluster(ctx context.Context, actorNames []string) e
 	return eg.Wait()
 }
 
+// getSetDeadlettersCount gets and sets the deadletter count
+func (x *actorSystem) getSetDeadlettersCount(ctx context.Context) {
+	var (
+		to      = x.getDeadletters()
+		from    = x.getSystemGuardian()
+		message = new(internalpb.GetDeadlettersCount)
+	)
+	if to.IsRunning() {
+		// ask the deadletter actor for the count
+		// using the default ask timeout
+		// note: no need to check for error because this call is internal
+		message, _ := from.Ask(ctx, to, message, DefaultAskTimeout)
+		// cast the response received from the deadletters
+		deadlettersCount := message.(*internalpb.DeadlettersCount)
+		// set the counter
+		x.deadlettersCounter.Store(uint64(deadlettersCount.GetTotalCount()))
+	}
+}
+
 func isReservedName(name string) bool {
 	return strings.HasPrefix(name, systemNamePrefix)
 }

actors/actor_system_test.go

@@ -79,6 +79,7 @@ func TestActorSystem(t *testing.T) {
 		actorRef, err := sys.Spawn(ctx, "Test", actor)
 		assert.Error(t, err)
 		assert.EqualError(t, err, ErrActorSystemNotStarted.Error())
+		assert.Nil(t, sys.Metric(ctx))
 		assert.Nil(t, actorRef)
 		assert.Zero(t, sys.Uptime())
 	})
@@ -952,12 +953,14 @@ func TestActorSystem(t *testing.T) {
 		err = sys.Unsubscribe(consumer)
 		require.NoError(t, err)
 
-		t.Cleanup(
-			func() {
-				err = sys.Stop(ctx)
-				assert.NoError(t, err)
-			},
-		)
+		metric := sys.Metric(ctx)
+		require.NotNil(t, metric)
+		require.EqualValues(t, 1, metric.ActorsCount())
+		require.EqualValues(t, 5, metric.DeadlettersCount())
+		require.NotZero(t, metric.Uptime())
+
+		err = sys.Stop(ctx)
+		assert.NoError(t, err)
 	})
 	t.Run("With deadletters subscription when not started", func(t *testing.T) {
 		sys, _ := NewActorSystem("testSys", WithLogger(log.DiscardLogger))
@@ -1738,4 +1741,5 @@ func TestActorSystem(t *testing.T) {
 		// shutdown the nats server gracefully
 		srv.Shutdown()
 	})
+	t.Run("With Metrics", func(t *testing.T) {})
 }

actors/metric.go

@@ -0,0 +1,106 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2022-2025  Arsene Tochemey Gandote
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package actors
+
+import "time"
+
+// Metric defines the actor system metric
+type Metric struct {
+	// DeadlettersCount returns the total number of deadletters
+	deadlettersCount int64
+	// ActorsCount returns the total number of actors in the system
+	actorsCount int64
+	// Uptime returns the number of seconds since the actor/system started
+	uptime int64
+}
+
+// DeadlettersCount returns the total number of deadletters
+func (m Metric) DeadlettersCount() int64 {
+	return m.deadlettersCount
+}
+
+// ActorsCount returns the total number of actors either in the system
+// or the total number of child actor given a specific PID
+func (m Metric) ActorsCount() int64 {
+	return m.actorsCount
+}
+
+// Uptime returns the number of seconds since the actor/system started
+func (m Metric) Uptime() int64 {
+	return m.uptime
+}
+
+// ActorMetric defines actor specific metrics
+type ActorMetric struct {
+	// DeadlettersCount returns the total number of deadletters
+	deadlettersCount uint64
+	// ActorsCount returns the total number of child actor given a specific PID
+	childrenCount uint64
+	// Uptime returns the number of seconds since the actor/system started
+	uptime int64
+	// LastProcessingDuration returns the duration of the latest message processed
+	latestProcessedDuration time.Duration
+	// RestartCount returns the total number of re-starts by the given PID
+	restartCount uint64
+	// ProcessedCount returns the total number of messages processed at a given time
+	processedCount uint64
+	// StashSize returns the stash size at a given time
+	stashSize uint64
+}
+
+// LatestProcessedDuration returns the duration of the latest message processed duration
+func (x ActorMetric) LatestProcessedDuration() time.Duration {
+	return x.latestProcessedDuration
+}
+
+// RestartCount returns the total number of re-starts by the given PID
+func (x ActorMetric) RestartCount() uint64 {
+	return x.restartCount
+}
+
+// DeadlettersCount returns the total number of deadletters
+func (x ActorMetric) DeadlettersCount() uint64 {
+	return x.deadlettersCount
+}
+
+// ChidrenCount returns the total number of child actor given a specific PID
+func (x ActorMetric) ChidrenCount() uint64 {
+	return x.childrenCount
+}
+
+// Uptime returns the number of seconds since the actor/system started
+func (x ActorMetric) Uptime() int64 {
+	return x.uptime
+}
+
+// ProcessedCount returns the total number of messages processed at a given time
+func (x ActorMetric) ProcessedCount() uint64 {
+	return x.processedCount
+}
+
+// StashSize returns the stash size at a given time
+func (x ActorMetric) StashSize() uint64 {
+	return x.stashSize
+}

actors/pid.go

@@ -220,6 +220,32 @@ func newPID(ctx context.Context, address *address.Address, actor Actor, opts ...
 	return pid, nil
 }
 
+// Metric returns the actor system metrics.
+// The metric does not include any cluster data
+func (pid *PID) Metric(ctx context.Context) *ActorMetric {
+	if pid.IsRunning() {
+		var (
+			uptime                  = pid.Uptime()
+			latestProcessedDuration = pid.LatestProcessedDuration()
+			childrenCount           = pid.ChildrenCount()
+			deadlettersCount        = pid.getDeadlettersCount(ctx)
+			restartCount            = pid.RestartCount()
+			processedCount          = pid.ProcessedCount() - 1 // 1 because of the PostStart message
+			stashSize               = pid.StashSize()
+		)
+		return &ActorMetric{
+			deadlettersCount:        uint64(deadlettersCount),
+			childrenCount:           uint64(childrenCount),
+			uptime:                  uptime,
+			latestProcessedDuration: latestProcessedDuration,
+			restartCount:            uint64(restartCount),
+			processedCount:          uint64(processedCount),
+			stashSize:               stashSize,
+		}
+	}
+	return nil
+}
+
 // Uptime returns the number of seconds since the actor started
 func (pid *PID) Uptime() int64 {
 	if pid.IsRunning() {
@@ -1768,3 +1794,25 @@ func (pid *PID) suspend(reason string) {
 		Reason:      reason,
 	})
 }
+
+// getDeadlettersCount gets deadletters
+func (pid *PID) getDeadlettersCount(ctx context.Context) int64 {
+	var (
+		name    = pid.Name()
+		to      = pid.ActorSystem().getDeadletters()
+		from    = pid.ActorSystem().getSystemGuardian()
+		message = &internalpb.GetDeadlettersCount{
+			ActorId: &name,
+		}
+	)
+	if to.IsRunning() {
+		// ask the deadletter actor for the count
+		// using the default ask timeout
+		// note: no need to check for error because this call is internal
+		message, _ := from.Ask(ctx, to, message, DefaultAskTimeout)
+		// cast the response received from the deadletters
+		deadlettersCount := message.(*internalpb.DeadlettersCount)
+		return deadlettersCount.GetTotalCount()
+	}
+	return 0
+}

actors/pid_test.go

@@ -72,6 +72,8 @@ func TestReceive(t *testing.T) {
 		require.NoError(t, err)
 		assert.NotNil(t, pid)
 
+		lib.Pause(time.Second)
+
 		// let us send 10 messages to the actor
 		count := 10
 		for i := 0; i < count; i++ {
@@ -89,6 +91,16 @@ func TestReceive(t *testing.T) {
 		assert.Zero(t, pid.ChildrenCount())
 		assert.NotZero(t, pid.LatestProcessedDuration())
 		assert.EqualValues(t, 10, pid.ProcessedCount()-1) // 1 because of the PostStart message
+		metric := pid.Metric(ctx)
+		require.NotNil(t, metric)
+		assert.NotZero(t, metric.LatestProcessedDuration())
+		assert.Zero(t, metric.ChidrenCount())
+		assert.EqualValues(t, 10, metric.ProcessedCount())
+		assert.NotZero(t, metric.Uptime())
+		assert.Zero(t, metric.RestartCount())
+		assert.Zero(t, metric.StashSize())
+		assert.Zero(t, metric.DeadlettersCount())
+
 		// stop the actor
 		err = pid.Shutdown(ctx)
 		assert.NoError(t, err)
@@ -1514,6 +1526,8 @@ func TestSpawnChild(t *testing.T) {
 		lib.Pause(time.Second)
 
 		require.Len(t, parent.Children(), 1)
+		metric := parent.Metric(ctx)
+		require.EqualValues(t, 1, metric.ChidrenCount())
 
 		// stop the child actor
 		require.NoError(t, child.Shutdown(ctx))