Monitoring cluster nodes in Rx

Ok. So we've now defined an observable query that tells us the connectivity of a heart beat / keep alive stream, based on some timeout interval.

IObservable<bool> IsConnected<T>(IObservable<T> heartbeats, TimeSpan timeout)

This allows us to monitor a single point of failure, or in the context of a client session, connectivity to the client. This is useful, but what if we are in a clustered environment, where 5 different servers are publishing their heart beats to the same channel.

Over time our event stream might look something like this;

TIME            Event (node id)
04:49:53       ponting
04:49:53       hayden
04:49:53       martyn
04:49:53       gilchrist
04:49:53       warne
04:49:54       ponting
04:49:54       gilchrist
04:49:54       hayden
04:49:54       martyn
04:49:54       warne

What would be useful is if we could separate this into connectivity status for each server. We already have a query for a single point of failure, so let's focus on breaking this stream into notifications for each node.

Ultimately we need a keyed observable for each unique node id. This is exactly what group by does, returning an observable of grouped observables.

IObservable<IGroupedObservable<TKey, TSource>> GroupBy<TSource, TKey>(...)

We can "compose" the grouped observable with the query we've already defined and it will give us the connectivity status for that node!

from nodeId in source
group nodeId by nodeId into grp
from isConnected in IsConnected(grp, timeout)
select new { nodeId = grp.Key, isConnected }

Outputting something like this;

{ nodeId = hayden, isConnected = True }
{ nodeId = martyn, isConnected = True }
{ nodeId = ponting, isConnected = True }
{ nodeId = warne, isConnected = True }
{ nodeId = gilchrist, isConnected = True }
*warne crashes + time passes*
{ nodeId = warne, isConnected = False }

I think this is a great example of composition in Rx.

Next up "Does my cluster have Quorum?"