Stream Processing Access Logs: LoKI Stack
Airtel’s customer reach is tremendously vast, and so comes the chunk of data each application creates every day. Such huge numbers invite challenges of data storage, processing, archiving and presenting. In this post, we present our journey towards another out of the box solution at Airtel specifically for application access logs.
What are the access logs? A one-liner- “Trace of Footprints on application.” In brief, who visited, when and what on the application. These are logs created by application servers and gateways such as WebLogic, Tomcat, Nginx, Kong etc.
Our previous solution
The past couple of years, our existing solution was standard ELK stack, we had filebeat running on each application server, which tails for log files and sends them as events to logstash cluster, which parses and transforms each event (being each entry in the log file) and forwards it to an elastic-search cluster. From where we had custom jobs which read from ES and send them to our InfluxDb at intervals for time series plotting. The ELK pipeline is so far has become the industry standard and is used almost everywhere but there are cost and problems associated with this data pipeline.
The Problem(s)
Running with small microservices filebeat had no problems; challenges came in tailing with rotating log files for Nginx, Kong, apache and other high TPS applications. Additionally
- Maintaining such a long stack is cumbersome. Any new application onboarding is a pain as there have to be modifications done at each component of the pipeline.
- Most importantly maintaining a filebeat is a tedious and complicated task, there is no such thing as a perfect combination of right configurations of filebeat, and the issue of log delay is ubiquitous for all our applications and servers.
- Then comes the pain of logstash grok maintenance, we were maintaining multiple logstash clusters and configuration change has to be replicated for each of the logstash instances, and this is a cumbersome process.
- Since our final focal point is, Influx db, we have to take the additional server cost of setting up logstash and ES clusters.
- Debugging of log drop is a very complicated and time-consuming process due to this long stacked pipeline.
The Journey Begins
Introducing LoKI stack — Logstash/Kafka/Influx
To remediate the above issues, we started exploring options with one single goal: “To send application access logs to Influx db within less time, cost and number of hops.” Elaborately:
- Time: We needed real-time data from access log files to our focal point InfluxDb.
- Cost: We wish to reduce the number of servers and data storage. In our case, storage at ES and computation at logstash is a known overhead since we are using influx as final storage and custom jobs for data conversion.
- Hops: “At Airtel, we say more the number of bogeys in the train, more are the chances of loot”. To infer, more the number of middle-wares more are the points of failure. Hence, we wish to use as minimum storage and middle-wares in our pipeline to reduce the points of failure and save time for debugging and maintenance.
Additionally, we also needed a mechanism to replay the logs of a time quantum in case of any failure. This typically becomes very vexatious to do with the current stack.
The Criteria of our selection
In our goal of selecting components for the new pipeline, we divided our problem into 3 components with specific criteria:
The Forwarder: The role of this component is simple to forward the application logs and capability to perform stateless operations. We keep it simple to put as minimal load and resource consumption of the application server.
The Mediator: This is the most critical component as this will be the vantage point of all the application logs for ALL the applications. This component needs to be highly resilient, highly available, elastic and fault-tolerant. This component also needs to be fast and should be capable of working on high throughput consistently.
The Processor: This component is the part of the pipeline where all the logical part resides. Since we also have applications that work on almost 2500 Rps, we cannot directly push these logs into influxDb as it’s not meant for that space and cardinality of it would be sacrificed. That is the only reason why we maintain an ES cluster; the capability of ES to store data and provide aggregations on top of it. This aggregated data (in a timely fashion) is then pushed into the time-series databases such as influxDb thus maintaining the cardinality. Hence the combination of the Mediator and Processor should stand tall on these below requirements:
- Ability to store ephemeral data.
- Perform stateless and stateful operations.
- Able to ingest Tbs of Data daily with thousands of RPS.
- Asynchronous execution.
- Minimum maintenance and cost.
The Process of evaluation and Prospects
Our process of evaluation centered around two key metrics. We pay lesser focus on the forwarder that resides on the application server and more on the processor and mediator as they are the ones that do all critical work.
Hence, for our mediator, we started looking out for some queue-based solutions the benefits of which are in perfect alignment of our expectations of a mediator, i.e.,
- Highly available
- Fast and resilient
- Asynchronous
- Can be used as a storage
Hence our options were: RabbitMQ and Kafka. Only these two because we at present are using these in our application as well and are working phenomenally. Now the downsides of RabbitMQ (wrt. our requirements are as follows)
- Cannot be used as a storage as there is no message level TTL. Once the message is utilized, it’s consumed.
- Although it is fault-tolerant. Imagine a case wherein our processor is reading from the queue goes down after reading all data from the queue. There is no way to replay the from a point of time of failure.
In Kafka, these two capabilities are inbuilt. Kafka can be used as storage by giving the message (record) level TTL topic wise. Additionally, Kafka keeps track of how much a consumer has read the data. Hence, if a consumer wishes to read the data it has lost, it can merely ask Kafka from the last point of reading or if need be replayed the whole state.
The Architecture
The Forwarder
Since we decided to use Kafka as a mediator, our options for forwarder were filebeat, logstash and telegraf. As we already listed challenges on the filebeat above. We choose to compare Telegraf and Logstash and found out that in Higher TPS CPU usage of telegraf increases and we cannot accommodate that. Additionally, Worth pointing out is the fact that telegraf runs on Go and has little support for threading compared to logstash that comes bundled with the entire package. Moreover, logstash provide more customization when it comes to stateless processing of records.
Usage with Telegraf (~60%) Vs Logstash (~9%):
The Processor + Mediator
We could have used the confluent-connector APIs and influx sink connector already provided by the confluent stack, but the influx sink connector has limited capabilities when it comes to stateful processing as our use case involves time-sliced aggregations and transformations (to keep influx cardinality down); hence our options were Kafka Streams/Flink/Storm, and we choose Kafka Streams. Because
- Simple and easy interface with lambda support
- Loose coupling
- Low-level Processor API to facilitate custom Nodes and processing
- Custom State Store Interface
And many more.
The Idea
The idea is simple, we push data from files via logstash in Kafka as key-value messages; we can do so via a Kafka output plugin in logstash. From there we are consuming it into our custom Kafka stream application whose working can be summarized in 2 lines:
For a given key, read records with-in a window of 5 mins and initialize Buckets [see Bucketing( Part-2) ]. Calculate 50th, 75th, 90th, 95th and 99th mean. And output these metrics upon window close to another topic in Kafka to be read by final sinker (Telegraf). These windows are non-overlapping and hopping.
Quick Summary on Components
- Logstash: Transforms each message type to convert it to single type this is in alignment with the Kafka ecology since Kafka stores data in bytes, and there is no type in Kafka we are also taking the same approach and killing the “types” at the application server via logstash so that a homogeneous type reaches our stream application. The key in our case become the endpoint (url) from access logs. Value being rest of the metrices.
- Kafka Streams: Calculates aggregations upon consistent streaming data with stream time. Kafka provides ephemeral storage of event and fails support to replay the states, from any point of time.
- Telegraf: The final sink of aggregate data to appropriate influxDb application wise. Telegraf also provides stateless operation hence using it made us move all application to new pipeline with a blink with adding custom fields from logstash which we use as measurement name to InfluxDb.
With the above architecture we were able to:
- Fluidically onboard any new application.
- Migrate all other existing application in a blink.
- Dynamically route application to their destination influxDb instance on the fly.
- Process, monitor and visualize access logs in real-time.
- Replay logs from any point of time and
- Process more than 2 Billion events per day with just 3 servers hosting entire stack of kafka-brokers/zookeeper/kafka-streams application and telegraf.
With logstash residing at the application level we become in control of almost every possible stateless operation and transformation since our use cases involves: dropping some non-required items, removing some fields such as unique id in URLs (which increases influx cardinality) etc. Moreover, with kafka streams and telegraf we are able achieve entire state replay, we appended the timestamp in the aggregated data and used telegraf to override influx timestamp with this. Whenever we wish to replay, we simply rewinded the telegraf consumer group id to an offset/time.
Some Great Reads:
