Traces and Metrics for Akka.Cluster DistributedPubSub Scenario

On this page we will see how Phobos handles publisher-subscriber scenario in distributed environment - using Distributed Publish Subscribe.

Code

Here is a code sample we will use:


var phobosBuilder = new PhobosConfigBuilder().WithTracing(t =>
{
    // As of https://github.com/petabridge/phobos/pull/779, these messages should be
    // filtered out automatically by Phobos.
    /*
    // Exclude traces with internal cluster messages
    t.AddIncludeMessageFilter(msg => !msg.GetType().Name.Contains("GossipTick") &&
                              !msg.GetType().Name.Contains("CurrentClusterState") &&
                              !msg.GetType().Name.Contains("Delta") &&
                              !msg.GetType().Name.Contains("Status"));
    */
});
var setup = Setup(Bootstrap(ClusterConfig), phobosBuilder);

// Create "publisher" system
using var sys1 = ActorSystem.Create("ClusterSystem", setup);
var node1 = sys1.ActorOf(Props.Create(() => new Node1Actor(TestActor)), "node1");

// Create "subscribers" system
using var sys2 = ActorSystem.Create("ClusterSystem", setup);
var node2 = sys2.ActorOf(Props.Create(() => new Node2Actor()), "node2");

// connect systems with each other
var cluster1 = Akka.Cluster.Cluster.Get(sys1);
await cluster1.JoinAsync(cluster1.SelfAddress);
await Akka.Cluster.Cluster.Get(sys2).JoinAsync(cluster1.SelfAddress);

// wait for subscriptions activated
await DistributedPubSub.Get(sys1).Mediator.Ask<SubscribeAck>(new Subscribe("strings", node1));
await DistributedPubSub.Get(sys2).Mediator.Ask<SubscribeAck>(new Subscribe("numbers", node2));

// wait until delivery is established
await AwaitAssertAsync(() =>
{
    node1.Tell(-1);
    ExpectMsg("-1");
}, TimeSpan.FromSeconds(10));

// publish and wait for messages
for (var i = 0; i < 10; ++i)
    node1.Tell(i);
for (var i = 0; i < 10; ++i)
    ExpectMsg(i.ToString());



private class Node1Actor : ReceiveActor
{
    public Node1Actor(IActorRef testActor)
    {
        Receive<int>(msg => DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish("numbers", msg)));
        Receive<string>(msg => testActor.Tell(msg));
    }
}

private class Node2Actor : ReceiveActor
{
    public Node2Actor()
    {
        Receive<int>(msg =>
            DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish("strings", msg.ToString())));
    }
}

Messages are published by publisher on one node, and received by subscriber on another node from numbers topic. And then subscriber is publishing converted message back to first node via strings topic.

Metrics

Unlike EventStream or Request-Response local scenarios, there are more actors created and messages published:

Context: Akka.NET

Counters

Name Tags Value
akka.actor.created actortype: Phobos.Actor.End2End.Tests.Standalone.DistributedPubSubSpecs+Node1Actor
akka_address: akka.tcp://[email protected]:port
 2
akka.messages.recv actortype: Akka.Cluster.Tools.PublishSubscribe.Internal.Topic
akka_address: akka.tcp://[email protected]:port
messagetype: Akka.Cluster.Tools.PublishSubscribe.Subscribe
 4
akka.actor.created actortype: Akka.Cluster.Tools.PublishSubscribe.Internal.Topic
akka_address: akka.tcp://[email protected]:port
 4
akka.messages.recv actortype: Phobos.Actor.End2End.Tests.Standalone.DistributedPubSubSpecs+Node1Actor
akka_address: akka.tcp://[email protected]:port
messagetype: string
 22
akka.messages.recv actortype: Phobos.Actor.End2End.Tests.Standalone.DistributedPubSubSpecs+Node1Actor
akka_address: akka.tcp://[email protected]:port
messagetype: int
 24
akka.messages.recv actortype: Akka.Cluster.Tools.PublishSubscribe.Internal.Topic
akka_address: akka.tcp://[email protected]:port
messagetype: string
 22
akka.messages.recv actortype: Akka.Cluster.Tools.PublishSubscribe.Internal.Topic
akka_address: akka.tcp://[email protected]:port
messagetype: int
 22
akka.actor.created actortype: Phobos.Actor.End2End.Tests.Standalone.DistributedPubSubSpecs+Node2Actor
akka_address: akka.tcp://[email protected]:port
 2
akka.messages.recv actortype: Akka.Cluster.Tools.PublishSubscribe.DistributedPubSubMediator
akka_address: akka.tcp://[email protected]:port
messagetype: Akka.Cluster.Tools.PublishSubscribe.Subscribe
 4
akka.actor.created actortype: Akka.Cluster.Tools.PublishSubscribe.DistributedPubSubMediator
akka_address: akka.tcp://[email protected]:port
 4
akka.messages.recv actortype: Phobos.Actor.End2End.Tests.Standalone.DistributedPubSubSpecs+Node2Actor
akka_address: akka.tcp://[email protected]:port
messagetype: int
 22
akka.messages.recv actortype: Akka.Cluster.Tools.PublishSubscribe.DistributedPubSubMediator
akka_address: akka.tcp://[email protected]:port
messagetype: Akka.Cluster.Tools.PublishSubscribe.Publish
 46
akka.messages.recv actortype: Akka.Cluster.Tools.PublishSubscribe.DistributedPubSubMediator
akka_address: akka.tcp://[email protected]:port
messagetype: Akka.Cluster.Tools.PublishSubscribe.Internal.Subscribed
 4

Histograms

Name Tags Value
akka.messages.latency actortype: Akka.Cluster.Tools.PublishSubscribe.Internal.Topic
akka_address: akka.tcp://[email protected]:port
messagetype: Akka.Cluster.Tools.PublishSubscribe.Subscribe
 4
akka.messages.latency actortype: Phobos.Actor.End2End.Tests.Standalone.DistributedPubSubSpecs+Node1Actor
akka_address: akka.tcp://[email protected]:port
messagetype: string
 22
akka.messages.latency actortype: Phobos.Actor.End2End.Tests.Standalone.DistributedPubSubSpecs+Node1Actor
akka_address: akka.tcp://[email protected]:port
messagetype: int
 24
akka.messages.latency actortype: Akka.Cluster.Tools.PublishSubscribe.Internal.Topic
akka_address: akka.tcp://[email protected]:port
messagetype: string
 22
akka.messages.latency actortype: Akka.Cluster.Tools.PublishSubscribe.Internal.Topic
akka_address: akka.tcp://[email protected]:port
messagetype: int
 22
akka.messages.latency actortype: Akka.Cluster.Tools.PublishSubscribe.DistributedPubSubMediator
akka_address: akka.tcp://[email protected]:port
messagetype: Akka.Cluster.Tools.PublishSubscribe.Subscribe
 4
akka.messages.latency actortype: Phobos.Actor.End2End.Tests.Standalone.DistributedPubSubSpecs+Node2Actor
akka_address: akka.tcp://[email protected]:port
messagetype: int
 22
akka.messages.latency actortype: Akka.Cluster.Tools.PublishSubscribe.DistributedPubSubMediator
akka_address: akka.tcp://[email protected]:port
messagetype: Akka.Cluster.Tools.PublishSubscribe.Publish
 46
akka.messages.latency actortype: Akka.Cluster.Tools.PublishSubscribe.DistributedPubSubMediator
akka_address: akka.tcp://[email protected]:port
messagetype: Akka.Cluster.Tools.PublishSubscribe.Internal.Subscribed
 4

In addition to our Node1 and Node2 actors that we are creating ourselves, we can see two DistributedPubSubMediator actors created (one per each actor system in the cluster), and two Akka.Cluster.Tools.PublishSubscribe.Internal.Topic actors representing topics.

We can also see a set of messages coming into actor's mailboxes, including some internal onces like Akka.Cluster.Tools.PublishSubscribe.Publish.

Note: Some metrics are not displayed here, like GossipTick message metrics - these are service messages used for cluster nodes communication, and are not representative for this scenario.

Also, local port numbers are replaced with port here - you will have your own random ports displayed.

Traces

For each messages published, we have full-tracked message trace from publishing to subscriber's mailboxes through the actor systems boundaries. Here is a sample trace for the first message:

sequenceDiagram akka//TestSys/system/testActor->>/user/node1: akka.msg.recv int Note over akka//TestSys/system/testActor,/user/node1: SpanIndex: 0
ParentIndex: none
Tags:
  akka.actor.recv.msgType: int
Logs:
  message
  content: 0
/user/node1->>/system/distributedPubSubMediator: akka.msg.recv Akka.Cluster.Tools.PublishSubscribe.Publish Note over /user/node1,/system/distributedPubSubMediator: SpanIndex: 1
ParentIndex: 0
Tags:
  akka.actor.recv.msgType:
    Akka.Cluster.Tools.PublishSubscribe.Publish
Logs:
  waiting
  
  message
  content:
    Publish"topic=numbers, sendOneToEachGroup=False, message=0"
/user/node1->>/user/node2: akka.msg.recv int Note over /user/node1,/user/node2: SpanIndex: 2
ParentIndex: 1
Tags:
  akka.actor.recv.msgType: int
Logs:
  waiting
  
  message
  content: 0
/user/node2->>/system/distributedPubSubMediator: akka.msg.recv Akka.Cluster.Tools.PublishSubscribe.Publish Note over /user/node2,/system/distributedPubSubMediator: SpanIndex: 3
ParentIndex: 2
Tags:
  akka.actor.recv.msgType:
    Akka.Cluster.Tools.PublishSubscribe.Publish
Logs:
  waiting
  
  message
  content:
    Publish"topic=strings, sendOneToEachGroup=False, message=0"
/user/node2->>/user/node1: akka.msg.recv string Note over /user/node2,/user/node1: SpanIndex: 4
ParentIndex: 3
Tags:
  akka.actor.recv.msgType: string
Logs:
  waiting
  
  message
  content: 0

Here we can see that:

  1. Test actor is sending a message 0 to the publisher (node1 actor)
  2. Node1 is publishing a message - sending Publish envelope to system actor (it is DistributedPubSubMediator actor of his actor system), we can also see the actual message content in span logs - i.e. topic is set to numbers.
  3. Unwrapped message is forwarded to another system actor - Topic actor in this case, which in turn is forwarding it to the Node2 actor.
  4. After message conversion , Node2 is publishing wrapped message back to DistributedPubSubMediator which is forwarding it via another Topic actor to Node1.

As you can see, we have a pretty clear picture about where the messages goes and which actors are communicating. Isn't that great?