What is fluentd and its support in PerfIt

After a few years of several contending products and tools in the tracing middleware space, it seems the industry has settled on fluentd. I believe this is probably because it has maintained technology-neutrality when it comes to ultimate destination of metrics and logs, i.e. storage/visualisation/search/alerting aspect.

The journey had started with Etsy’s statsd back in 2012. It allowed for collection of metrics mainly using UDP which started an important if not revolutionary trend. On the other hand, Zipkin project came out of internal works in Twitter inspired by Google’s Dapper paper.

Zipkin had poised to become the universal Open Source distributed tracing solution. It certainly inspired the OpenTracing initiative but the more OpenTracing grew, the more it dented the status of Zipkin, as OpenTracing improved on some of the shortcomings of Zipkin that were kept in order to maintain backward compatibility. I actually worked with Zipkin community and its amazing community leader, Adrian Cole, to bring Zipkin support to Azure Event Hubs. I also started adding support for Zipkin to the PerfIt project (I did not get a chance to use it in anger and I doubt anyone else used it). Regardless of its merits, drawbacks and its future, it couples middleware with storage and search and is not a pure middleware.

Another contender was LogStash which then morphed into Beats and was the default choice for collection of metrics and logs in Elasticsearch. But then again, these technologies are optimised for use with Elasticsearch and Kibana.

fluentd is ubiquitous

Now, with fluentd being a default choice with Kubernetes, it understandably has gained wide adoption. Having said that, I believe it says more about fluentd than Kubernetes as there is so much to love about fluentd since it:

• Is simple and straightforward
• Is Open Source
• Supports many inputs/transports (HTTP, UDP, file)
• Supports common output types (Elasticsearch, s3, kafka)
• Supports many formats/parsers (csv, json, nginx)
• Supports transformations and buffering
• Does not try to solve high availability and instead provides probes so that infrastructure can monitor
• Provides extension points using the plugin model

So simply, it can accept logs, metrics and traces from your applications, while running as a Kubernetes service, a sidecar to your application or a simple deamon, and then push them to your destination of choice so your application does not have to worry about buffering, retry, destinations, etc. Using UDP which is essentially fire and forget, you completely decouple trace collection from your code - practically zero to little overhead.

While it initially targeted as a Big Data transformation middleware, it is clear now that its strength will be in the observability space. That is why I believe it will be a glue that will be binding producers and consumers of observability data for the coming years. And that is the main reason I felt I had to support it in PerfIt.

PerfIt and fluentd

For those of you who are not familiar with PerfIt, I started the project more than 6 years ago for instrumenting .NET systems - in the absence of similar tool. Initially it was meant only for custom Windows performance counters but gradually grew to support ETW and finally with .NET Core to support other forms of transport such as Azure Event Hubs.

Now I support fluentd using UDP transport.

Getting started

I have blogged in the past on using PerfIt but essentially you create an instrumentor and instrument a piece of code:

var si = new SimpleInstrumentor(new InstrumentationInfo()
{
    Description = "Instruments your code!",
    Name = "general",
    CategoryName = "my app",
    InstanceName = "Sleep100"
});

si.Instrument(() =>
{
    Thread.Sleep(100);
});

You can also achieve this using AOP with attributes in ASP.NET Web Api or Core MVC.

You can register tracers which are essentially outputs/sinks for your captured metrics. Currently Azure EventHubs, Zipkin and fluend are supported - the latter is the new addition:

// add the tracer after creating instrumentor - sends UDP datagrams to localhost on port 5160
si.Tracers["fluentd"] = new UdpFluentdTracer("localhost", 5160);

That is really it! This means all your *sampled* traces will be sent to your fluentd. Easy peasy...

Business and Log Events, Azure EventHub and Psyfon

TLDR; If you need to send large number of events to Azure EventHub from a .NET process or passthru API, consider using psyfon.

Over the last two decades, many businesses have transformed themselves and modeled their processes and operations as software (bespoke or customising off-the-shelf products). These systems would turn business processes and transactions into data that can be stored, queried or exchanged - ROI for such data is very high and the challenges of building/evolving such systems have been widely known. These systems typically generate business events.

Businesses have been turning their attention to the next goal: capturing (and analysing in near real-time) the information that commonly not considered as valuable data, such as minute user interactions with sites/apps down to the level of mouse movements and scrolls, sensor outputs in vehicles or factories, CCTV streams from municipal cameras to predict/forecast traffic, shopper interactions/behaviour in supermarkets to gain insight/provide recommendations, etc. These systems generate what I - for better or worse - call log events which I have explained in the past here but would be useful to re-cap their differences with business events in the table below.

While log events could have been historically stored and then analysed in batch mode, there is growing need to make some sense of the data in real-time in addition to in-depth analysis in offline mode. That is essentially stream processing.

Stream processing is hard. Building resilient processes to be able to reliably process tons of data in parallel while handling back-pressure, point failures, peaks of activity - all of which with few seconds or even sub-second latency - is not trivial. There are such systems already available such as Apache Flink, Kafka Streams or Spark Streaming. These systems typically work on top of an Event Store such as Kafka or Azure EventHubs.

Azure EventHub has been built for publishing and consuming events at high-scale. The design is not dissimilar to that of Kafka: a replicated/Highly-Available log per arbitrary (but constant) number of partitions where ordering can be guaranteed only at the partition level. You can read from the beginning of the log or from any point in the stream but remembering where you last read events from (checkpointing) is completely left to the consumers.

Typically only a single consumer is meant to read from a partition hence having more partitions is important for improving scalability. In terms of publishing, this is much more laxed: a high number of producers can send events to EventHub.

How does EventHub assign events to partitions? You can optionally send a Partition Key which gets hashed and used for assigning to partitions. To make sure you get the best out of your system, the Partition Key needs to be evenly distributed. If you are sending device events, you would most likely use the DeviceId. For customer events, Customer ID is a natural choice. This will ensure all events for a device or customer are ordered according to the time they are arrived at the EventHub.

Usually there are data pipelines that receive and funnel the incoming data (usually through a passthru API) to these stores but the key point is these data pipelines exhibit the same challenges shared by the stream processing. While initially exposing EventHub directly to the outside world was advocated by Microsoft, you would most likely want to hide your EventHub behind a passthru API that does authentication and optimises delivery of the events to the EventHub by batching. This layer is also useful to handle back-pressure by buffering events so you can deal with spikes gracefully. One thing you cannot do here at the API is to keep the caller waiting for event to be successfully committed to the EventHub for a few reasons. First of all, EventHub can sometimes have latency in the order of 100-150ms. While this is completely acceptable for most purposes, (other than High-Frequency Trading!), keeping clients waiting means more power consumption for publishers many of whom are phones and other low-power devices, sending many events per hour. Another reason is that EventHub works best if you send events in batches hence waiting until your buffer is full and then committing the batch of events.

Batching is already supported built-in with the EventHub:

var batch = new EventDataBatch("myPartitionKey");
batch.TryAdd(eventData); // keep adding until method returns false 
await client.SenAsync(batch);

But did you notice something? All events within a batch must have the same partition key. While it is understandable Microsoft made this decision for performance reasons - since all such events will be sent to the same partition otherwise batch has to wait for all partitions involved to respond successfully - it essentially renders batching remarkably less useful. As said earlier, Partition Key must have widely diverse value such as Customer ID or Device ID. There is no guarantee that an event arrived from a customer at the API is followed by enough events from the same customer in a reasonable amount of time to fill the batch and make batching worthwhile - let alone those events arriving at exactly the same web-head.

Solution is to essentially send the events directly to partitions. But the hashing takes place at the EventHub, how could we know what Partition Key gets allocated to which partition? This implementation is opaque and is not possible to reproduce it outside EventHub. That is why we have to hash the Partition Keyes ourselves and send batches directy to the partitions. All we need is a hashing algorithm capable of uniformly hash Partition Keyes across partitions. It turns out that most hashing algorithms including MD5 can easily achieve this, although some might be cryptographically broekn. MD5 is a very quick and efficient algorithm hence is a good fit.

Psyfon

Now, all of what I have said so far - batching, buffering and hashing - have been implemented in an Open Source project called Psyfon. Using this library supporting both .NET Standard 2.0 and .NET 4.52, all you have to do is to create a single instance of BufferingEventDispatcher per process, start it and add events to it:

var singletonDispatcher = BufferingEventDispatcher("<connection string>");
singletonDispatcher.Start();

// and somewhere else in the code where events generated
var ed = new EventData(mySerialisedEventAsByteArray);
singletonDispatcher.Add(ed);

You can set a maximum byte size (according to the size of your events) and maximum number of seconds before committing the batches, whichever is reached earlier batch will be committed to the partition. I have tested it under high scale and essentially a single process had no issue sending 5000 EPS to EventHub. I will be publishing the results of a more extended test soon.

While my use case was a passthru API, this cane be equally used for dispatching monitoring and instrumentation events to the EventHub. PerfIt, another Open Source library will benefit from this very soon - watch the space.