Speaker A
Foreign.
Deep dive into Agoda's Log Castro, a high-volume log aggregation platform designed for real-time log management and analytics at scale.
Key Takeaways
- Log Castro enables centralized, real-time access to logs from thousands of instances across multiple regions.
- The platform supports complex queries and alerting with a focus on low-latency responses even at high data volumes.
- Kafka is used as a persistent ingestion buffer to ensure data durability and smooth processing.
- The system architecture balances real-time and historical data access with efficient caching and indexing.
- Replacing Elasticsearch/Kibana with Log Castro improves scalability and performance for Agoda’s observability needs.
What the video covers
- Introduction to Agoda's in-house log aggregation platform, Log Castro, designed to handle massive log volumes efficiently.
- Explanation of the need for log aggregation in large-scale deployments versus small-scale setups.
- Overview of functional requirements including real-time log access, alerting, dashboards, and post-processing capabilities.
- Discussion of hardware and performance requirements, including handling 150,000 logs per second and maintaining P99 response times under 5 seconds.
- Comparison with existing Elasticsearch and Kibana stack, highlighting the transition to Log Castro.
- Architecture overview: ingestion from Kafka, real-time processing, metadata storage in relational DB, and blob storage for actual logs.
- Description of system components such as real-time API, historical API, federated API, and roll-up auxiliary services.
- Details on data ingestion, storage formats, indexing strategies, caching mechanisms, and query optimization techniques.
- Challenges of cross-region data replication and network load management with multi-region federation.
- Insights into Log Castro’s design choices inspired by Grafana Loki and implemented in Rust.
Chapters
- 00:00Introduction and Overview of Log Aggregation
- 03:46Use Cases and Importance of Log Aggregation
- 07:04Performance and Scale Challenges
- 10:29Functional and Hardware Requirements
- 13:49Current System and Transition to Log Castro
- 17:14Log Ingestion Pipeline and Kafka Usage
- 21:29Log Castro Architecture and Components
- 26:00Data Storage, Indexing, and Query Optimization
Full Transcript — Download SRT & Markdown
Speaker A
[Music] Hello and welcome everyone in Agoda and our external guest to Agoda Take Talk. Today we will get to learn about Agoda's in-house log aggregation platform, Log Castro. Eugene here will guide us on how we can efficiently handle the management
Speaker A
of massive amounts of logs each day. I believe some of you may have heard about this before because Eugene has conducted a workshop last month on this topic. So please note that the content of this event will be quite similar to that
Speaker A
one, but we also think that this could be a good opportunity to extend the reach of our discussions to everyone here in Tech and also to a wider audience that join us today. So we hope you get to learn a lot of things. If you have any
Speaker A
questions along the way, please feel free to send in the chat box or wait until the end of the presentation where we will have a Q&A. Okay? Okay, so without further ado, please join me in welcoming
Speaker A
Eugene. Hi everyone, hope you're having an amazing day. Hopefully this meeting will be interesting for you. Right, let's start then. So first of all, who am I? My name is... I am a technical lead in Agoda working
Speaker A
on the observability platform. Basically, observability means everything starting from logs and metrics and something more, for example, anomaly detection. Today we are going to talk about logs and again, guys, please bear with me to the end. If you have any questions, we'll have a
Speaker A
separate Q&A session, so we will have enough time hopefully for everyone to answer your questions. But we will not have questions in the middle of the talk. Thank you for understanding. So let's start with
Speaker A
simple things. Basically, what is logging? Hopefully everybody knows that. Logs are just like a text representation of any kind of events which developers want to have later analyzable. We usually write them to either STDOUT or on
Speaker A
disk. But why do we need log aggregation? Basically, if you have just one simple server or your application is deployed on a small number of services, you can just go to the services directly or if you're using clouds you
Speaker A
can just open the console from one specific port and you can see what are the logs from this application from this particular instance. But for example, if you are deployed in large scale, for example of
Speaker A
the scale of Agoda, it will be quite problematic for you to get exactly the logs from, let's say, thousands of different instances of applications. So that's exactly why and which problem the log aggregation platform solves. It provides you centralized logs access,
Speaker A
meaning you can tell, I want my logs from application X for the last 15 minutes and I want to see what the logs, for example, for error did it have. So this is the first use case. Second use case, sometimes you just want to know how
Speaker A
many logs did you have, for example, how many errors did you have. This is quite important because having errors means something is wrong with your system. So you usually want to monitor that or sometimes you will want to have some log
Speaker A
patterns or like something else and you want to have monitoring over that and you want to have alerting over that. That is another thing which log aggregation is solving. So Log Castro, basically, Log Castro is Agoda's own log aggregation platform.
Speaker A
It's written in Rust and it's entirely inspired by Grafana Loki. Yeah, basically what we are going to talk about today, we will cover briefly the overall architecture and we'll go into each separate component and we'll discuss how do we store the data, how do
Speaker A
we ingest the data, what indexes do we use, which formats do we use on the storage, how do we transfer the data between different components of the system or the network, and some other interesting things. But first I would rather start with
Speaker A
functional requirements of this log aggregation platform. Basically, we wanted to have a platform which would provide people fluent real-time log access, meaning you just want to see something which happened on production like less than a minute ago,
Speaker A
obviously, but as fast as possible so you wouldn't need to go to the servers but you want to have your data accessible as soon as possible. Another thing obviously, you want to have the sourcing for alerts and dashboards like as we
Speaker A
discussed. You want to have graphs, like these fancy things which are hanging on the monitors throughout your rooms on the floors. Other things like if you are doing something interesting, for example, you're querying the data for last couple
Speaker A
of weeks, let's say, and you want to get some, for now, for example, how many particular events did you have throughout last some time. But another thing which we wanted to have, which you've seen in Grafana Loki, is the
Speaker A
real-time data post-processing, meaning if you have some logs, we want to do some additional filtration, we want to do some additional, for example, I see the URL and I want to parse particular user or like
Speaker A
parameter from the URL and I want to group by this parameter which I've seen so I would see how many particular occurrences of such variable I see in logs. So these are the functional requirements and our hardware requirements. Let's
Speaker A
start with the most controversial one which is 10 RPS. Doesn't sound that much, almost everyone having more, but if you will actually have a look what exactly the request for the system is, it might shock you because this request might require
Speaker A
traversing tens of gigabytes for each one, 10 gigabytes of data just to get your, just to get your response. For example, if you want to calculate number of logs or if you want to post-process and calculate, it might be really
Speaker A
challenging. Another thing, we have about 150,000 logs per second coming in different regions. And since we talked about regions, we also want to not overload our network by constantly replicating the stuff between the eight regions. So we want to also
Speaker A
have the cross-DC federation. And a bit more about our scale, in a day we have about 100 terabytes of logs and the most important one, we don't want our users to be able to get their coffee until we will be able to get the responses. So another requirement in our hard requirements, our P99 of the system overall should be less or equal to 5 seconds.
Speaker A
Foreign. Before going forward, I should tell as well what we are currently in the company using Elasticsearch with Kibana and we are replacing it with Log Castro and we are almost done. So, you can say it's almost a success story as well. So how the logs are going? Basically,
Speaker A
you have a server or like a pod, depends on where you deployed, you have a log agent which is getting the data from the servers and it sends to Kafka as our main source. We've chosen Kafka for
Speaker A
different reasons. The main reason is you can really operate with Kafka as a persistent storage for some time, for example for a day or for two, depends on your retention. And if you lose your data for the lost,
Speaker A
like for the recent amount of time, you can just edit the Kafka. So that's why Log Castro has no other mechanisms for heat ingestion apart from just ingesting from Kafka. Now about the architecture of the Log
Speaker A
Castro as is. So, this, this we have here a lot of errors, so let me just cover all of them separately. So about the data pipeline, how do we ingest? Basically, we ingest everything from Kafka and we have a core component of
Speaker A
the system which is ingesting the data and storing it further in storage and serving some requests which is a real-time. After that, after the real-time, save the metadata in our relational database and it saves the actual data in our
Speaker A
blob storage. After that, we have historical API which is serving the data for everything but older than 24 hours and you have federated API which is basically deciding where to go for this specific request. Is it refrigerated, historical, I worked to real-time, or to
Speaker A
both? Also, you have auxiliary thing which is called roll up.
Speaker A
basically these are the main components now let's go over uh each one of them and let's start with real time as I said real time in just the messages from Kafka also it serves the queries for the most
Speaker A
recent data because it's called real time and it stores the last 24 hours of data on disk before we can proceed to actually the real time and how it does everything and yada yada we should first start with
Speaker A
four Tower start talking with what our log message is so log message is looking something like that so it has a lot of different fields it has timestamp for example log level I don't know version and a lot of
Speaker A
different things related to to different applications different applications having the same logs but we will have this one as example so after the timestamp obviously we should notify a specific log and uh when did it occur we have application name
Speaker A
which is the primary key in our time series database in our log storage we use application name as a tablet identifier as well for example we can do really a lot of different things we can set up indexes separately we can set up
Speaker A
a quarter separately rate limits and a lot of different interesting things that's why we actually have something which uh our ours our system usually don't have or however don't impose people to have so we have primary key after that uh we are also mimicking the
Speaker A
time series database here so we have additional General index for all logs which is basically like has tags which are limited which are having limited cardinality cardinality for those who are not familiar with is just a number of unique values and these are the
Speaker A
things which are generally general for the for everybody in the system so everybody having log level everybody having server cluster and things like a logger name and exception type after that we have the actual messages so we have them split it into three
Speaker A
different things because people usually want to query them separately which are message exception message and a stack trace and each one of them can be very big and we can query each one of them separately on UI as well
Speaker A
also we have separate secondary indexes which are set up on demand for example some people want to trace their stuff uh talking about tracing as well so you can if you know what you have a lot of different unique values and you want to
Speaker A
make your index much faster you can ask and tell like guys I need additional index and we will create this index and it doesn't have a cardinality limitation so regardless of how many values how many unique values you have you can have
Speaker A
uh like you can have this index after that you can just label your data in a manner like you can put as many tags as you want and basically no limitation on the rest of the data is just something you want to
Speaker A
put into your data and you still will be able to search by it but the search will be quite slow uh so we have this log message and we are we will transform it into our data model data model called like log entry
Speaker A
it has serious time series id timestamp message and not index text uh so to get this model we are doing interning tags and we are creating the series so what does string entering mean basically since we know about
Speaker A
cardinality of my Intex is limited we can replace strings with integers string internal is also can be named as dictionary creation what does it mean it means what instead of operating the strings we can operate with integers and
Speaker A
operating with integers is much easier because they uh they have much less memory footprint and integers are basically just for whiteware non-essential so you can do much more things with integers than you can do with strings not going to dive into the
Speaker A
implementation we'll just say what is non-resizable highly concurrent open address hashmap uh and that's it you can read more in our uh Tech blog about our index index is effectively just a map of nested maps with the primary key of application name and the
Speaker A
tags when I say tag is just the concatenation of all tags which you have for specific log entry and the Series Sirius is basically imagine you have eight different tags and for each combination of each combination of this
Speaker A
unique text creates you the additional series since we don't have since we know what our serious count is less than five thousand five hundred thousand we can use brute force approach just a trade over for specific application in iterate
Speaker A
over series and it will be enough this is exactly the case where uh for brute force is enough smarter is not always better so after we figured out the basics we can now talk about actual real-time architecture so we have Kafka we have
Speaker A
some in memory components and on these components let's start before memory components and as one of the first ones we have in gestures and ingester is just an entity which reads data from Kafka and we have we start one adjuster per one partition
Speaker A
basically ingester means a thread and each adjuster reads the each separate partition from Kafka for those who are not familiar partition is just uh is just a unit of uh basically in Kafka you have several several topics and your
Speaker A
data is partitioned uh doesn't sound good enough partitioned in the partitions of uh data so what adjuster is doing first of all it obtains the message from Kafka parches the message it doesn't matter if it is a protot buff or if it is the open
Speaker A
Telemetry uh sorry if it is adjacent or if it is the otlp product above you get it you get some result out of it you in turn the tags internal attacks create the series and then you submit your log entry to
Speaker A
aggregation buffer it will cover it in a bit uh in a bit so after you ingester uh so for your injector to operate you need metadata because you want you need to get index ID and like series id and internal
Speaker A
strings uh that's for we have the metadata which is uh which we have one metadata block per one hour uh we have 24 blocks of such uh uh 24 blocks of such data in memory only first two blocks are mutable
Speaker A
and the mutable blocks aren't accepting new messages so it's a bit uh it's a bit tough to swallow and a lot of questions are arising first of all why exactly 24 hours uh we have a simple explanation to
Speaker A
that basically 85 percent of the queries are for last day and we decided to store such hot data nearby because you have a lot of use cases for such hot data for example investigation of production issues which are happening right now meaning like you
Speaker A
don't need to go to like uh to separate storage you don't want to face your locks from there you just store it somewhere nearby and you just your access to them will be super fast another question which is arising why do
Speaker A
we need this mutability and why only two blocks are immutable behind it we observed what 95 percent of locks are sent in last hour because usually you're just your logs are posted now like it always posted now meaning your logs should be like always
Speaker A
have a recent timestamp so 95 of the data has like is arriving at last hour but because we have we are not living in the ideal World sometimes the machine is lagging sometimes we have network issues sometimes slow collectors just having
Speaker A
too much stuff and we are and we are accepting the lag up to two hours because like 99.99 of messages are like uh being received in last 20 last two hours basically the five percent or like less a bit less than five percent are
Speaker A
being received in this uh second hour so why them utility at all uh uh in shorts it's much easier to operate immutable immutable data if you know what day your data is not changing you can just easily cache your stuff and
Speaker A
you're solving uh one of the hardest problems in the computer science uh right away and also uh no need to worry about the grades for shutdown Grace was shut down of stateful applications in my opinion it's the last uh unnamed or ever
Speaker A
unknown problem in computer science so if you want to mitigate the impact of your application stateful application dying and not being able to recover uh it's a big plus now we covered metadata blocks and adjuster now let's talk about
Speaker A
aggregation buffer and why do we need it before it we need to learn a new term called chunk and in shorts if you try to you're obviously not going to store your locks fully in memory because you have
Speaker A
hundreds of terabytes of them per day it will be just impractical so you're definitely going to write them on disk and if you're writing them on disk uh writing each message from disk separately is super slow so we group
Speaker A
them in memory and we store them in like we just called a batch of logs in chunks and chunks are being grouped by hourly block application which is the primary key and the log level so in result it looks like that and
Speaker A
another thing which I should mention is since we have a lot of different um a lot of different instances of your application running anywhere nobody guarantees what your clock are exactly aligned and what your messages in Kafka will be exactly ordered so this is
Speaker A
another problem why do we need chunks when you write chunk you receive them unorder it and after you've writing this Chunk on disk you can exactly tell what your uh what the what all of your logs in one chunks are ordered you can
Speaker A
utilize it in numerous number of numerous ways and which is really helpful so aggregation buffer looks like that you have a block for each block you have application or like multiple applications a lot of different applications and for each application
Speaker A
you might have a separate several different log levels um another several several facts about aggregation buffers you need to know is buffer has maximum size because if you will have too many log levels or like to have too many applications
Speaker A
otherwise you will just go over out of memory problem so to not have to not have this problem we limited the size of the buffer and uh basically if your buffer buffer after adding this a new log entry will grow too big we are going
Speaker A
to flash the biggest aggregation to disk also all data for specific block is being flushed to disk uh when we need to promote the block now let's go to on disk part because we finished with everything in memory
Speaker A
and let's now talk about how do we flush our data on disk I mentioned the chunks being written on disk multiple times and now let's let's talk about its format first of all Chan has two different components it has the data like actual
Speaker A
messages actual entries and it has the metadata metadata has a lot of different things and first of all it's obviously the minimum maximum timestamp so you would understand which exactly logs you need like which exactly chunks you need to
Speaker A
read if you are looking from let's say 1 pm to 5 PM you can you can kick out all the chunks which are outside of this time after that you have offsets of three different important things called uh
Speaker A
Bloom filter metadata like actual data offsets and serious timestamp meta offsets so what are these offsets offset is basically the actual physical uh location of this data where it was written so when I write this data on disk I get I get the position where I
Speaker A
started to write it on disk and how long how many bytes did it require to write it on disk and we have also some auxiliary stuff like seen serious and scene keys since serious means without going into date of
Speaker A
specific chunk you can know which series you already have so you would just like um before querying before actually doing any kind of additional i o for reading these messages you will um just read the series and we'll check
Speaker A
if you're like will you be able to find what you're looking for inside of this uh chunk or not and it goes both for Sirius and for an indexed Keys basically all of this metadata is supposed to reduce the number of actual physical
Speaker A
reads from disk as much as possible so more metadata you have let's less stuff you need to read and this comes in both ways so you need to balance on how many uh how many things in metadata you have
Speaker A
if you have too much metadata it means your stuff will be like your metadata will be to uh to um it will be too hard to store the memory basically okay too much information let's just say what messages is
Speaker A
something written on disk and these things we already into the right headlock of chunks basically we are writing it to disk but we are not kicking them out from memory we are leaving it there because we need it for
Speaker A
querying uh this is how our data format looks on disk I will just tell what we are not using any kind of existing uh product like any kind of existing formats like have retrieved or the protobuf just because we decided what these formats
Speaker A
are like pretty generic and we need something which will have a lead as little overhead as possible and we know what our schema for data for actual messages are not changing very often in fact we didn't change it in last year so
Speaker A
we decided to write our own format and it works pretty well for sequential reads also we didn't just write it on disk because like logs are excessive and they can be compressed very well so we decided to apply aussie4
Speaker A
compression to the top and now you have probably the question why do we have all Z4 compression although we have much more efficient compressions for example that STD and like perhaps even Snappy so this is the answer why ztcg isn't is
Speaker A
indeed much more efficient for the camper in case of uh compression so compression means if you have the data for example 100 megabytes compressed should be somewhere like 50 for example megabytes and as you see on the right graph that
Speaker A
STD is instead about 12 or 15 percent more efficient like the data after a zcg is about uh 12 less than after lz4 but if you will also check the second graph decompressing blocks is much more efficient and much faster with lc4
Speaker A
that's why we decided what we will buy the bullet and take the network overhead which we need to have to transfer this locks for the reading node but next time when it will be cached on disk we will
Speaker A
read it much faster when it would be with Z City so your so your call data will have a like pretty bad access time but when you when it is cached it will be much better than it is if
Speaker A
so yeah basically you just write this uh uh Bloom filters series ah yeah and Bloom filters which I didn't cover blue filters is just a basic data structure which which can tell you exactly uh something exists in a data set or not if
Speaker A
it says it exists in the data set it means it can be false positive but if it tells you it doesn't exist in a data set it definitely doesn't exist in the data set you don't need to go where
Speaker A
that's additional style of additional type of metadata which we have so uh data blocks is just the stripes it can be represented on discussed stripes and they are just appending our chunk which is serialized appending one after an hour after another many more chunks
Speaker A
and each stripe has the size in our configuration can be configured up to one gigabyte when this size is exceeded we are creating more blocks so why are we doing it basically we are like we want to be as nice to our S3 storage as
Speaker A
possible so we don't want to have a lot of files on our storage so we decided to organize the block storage on top of block storage and when it will be in Block storage we will um like if you have a big file and you
Speaker A
want only chunk two you can and if you know the offset of data for chunk two you can you can tell exactly hey give me the data from chunk 2 starting from offset a and giving it the number of
Speaker A
bytes B this way you will get your data from one offset to another set and this is how it effectively looks like you have just a lot of different threads building the stripes after the stripe is finished you're starting to
Speaker A
build another strain okay uh now let's talk about querying basically the querying means like you store the data it's very well like all good but nobody cares if you're studying the storing the date if you don't a lot
Speaker A
to play it right so for querying we are using grafana Loki local as well because we find its possibilities of post-processing as very nice and you also don't have for example in Lucin we had a we had a problem but you can do a
Speaker A
lot of things in multiple ways and it's really confusing the users and sometimes if you are doing if you are doing something there is a chance what you are you want something but you are not getting something what you want with local sorry
Speaker A
with Loki log ql it either works either doesn't so it's pretty simple and we like it in that case okay so for querying the logs uh imagine you have like four blocks and you have a query starting from from 12
Speaker A
to 2 PM uh so first thing which you need to do you need to identify the number of blocks which you have which are like intersecting to this time range after you found the number of blocks which are
Speaker A
intersecting with this time range you find all chunks which are applicable for these blocks after that you need to in turn your query in turning the query means you find this series applicable to the query from index so you query the index and
Speaker A
you say hey like give me all series for each of these blocks which I can find the data for my query and it gives you like for example a set of series one two three four five six after it you filter
Speaker A
in all the chunks which are under these blocks which have series one two three four five six so you're reducing the number of blocks which you need to read super nice you're reducing the number of iOS IO everybody's happy
Speaker A
after you reduce the number of challenge by intersectionate with the series you're filtering chunks intersecting with the time period so meaning you can have chunk which is for example your your query is starting from 12 30 to 130 but your chunk is started
Speaker A
starts from 12 5 and to 12 15. so you will reduce the number of chunks even more after that you will apply some additional optimization from the metadata which I just talked about uh you filter it by syntax and Bloom
Speaker A
filters you just check if Bloom filter exists and if Bloom filter has the data which is requested after you reduced it as well uh you basically have like a least number of chunks possible which is definitely having the data for your query after you
Speaker A
have this number of uh like after you have these chunks you need to implement like you actually need to read the logs no way around it you need to read the logs and execute your query after you executed the your query on each of the
Speaker A
chunk you aggregated to the level of blocks for example you had the five chunks here or four chance on the left you aggregate them together and have a have a result for the block after it you have a result per block
Speaker A
you aggregate results per block together basically it's uh it's an absolute basic mapreduce uh and this is how our querying works now let's talk about the types of queries in local first of all you have histogram queries meaning when you want
Speaker A
to get the counts of some data occurrences for example you want to get the number of Errors for last five hours and when you have actually the log queries where you need to get the text and you need to get the whole data
Speaker A
um blog queries what we can say about it first of all result is text this is this means what you will fetch a lot of data even up to hundreds of megabytes sometimes if you have a big loss
Speaker A
another important fact user never wants to have a lot of locks you don't want to read the whole database and basically user will just walk away and close the browser and we'll tell how bad your system is so you just need to give them
Speaker A
a thousand of lines of logs usually or less or more like ten thousand hundred but it always is limited number which is applicable to the query after that you don't need to process the all blocks meaning if I am creating the logs which
Speaker A
are like which you have a lot of for last year you don't need to iterate over the whole data for last year you just need to check the data which you have like for the last five minutes and you
Speaker A
will probably get this 100 likes of Lock and you will return it back and that's why the creating of logs is sequential on Block Level so you're not going to to query the logs from next block if you
Speaker A
didn't finish the querying for specific for one block so if you're querying for last year I will first query now minus one hour then now minus two hours and now minus three hours now about histogram queries uh result is
Speaker A
a Time series basically it just means that you have very small results it is it will be almost always in kilobytes unless you have a lot of series um but it means that you need to process all the data if you query the accounts
Speaker A
for last year it means you need to calculate the data for last year no way around it and the volume of data can be obviously enormous that's why we parallelize it as hard as possible like you have that like one year of data you
Speaker A
run it on many executors many threads just make it finished as fast as possible okay now let's talk about how do we actually scale the real time um let's talk about uh let's play a small game of system design you have a
Speaker A
small like you have an instance of real time you can you can have a you you can have a instance of real time after it you know what it needs to be queried by Federated API and it needs to ingest the
Speaker A
data from Kafka what's the problem here obviously you have a downtime during the deployment um when you do that okay guys sorry there are a lot of questions I will go down go for them at the end just just bear with me for 10
Speaker A
more minutes uh basically if you have just one note you will have definitely the downtime during the deployment so you definitely it cannot just obviously have one single real-time lot so what you obviously do you will just plug in in our instance
Speaker A
you will have the primary and the backup so what is good and bad about it primary and backup consume the same traffic under different consumer groups and consumer group in Kafka is just a mechanism how you can have exactly the
Speaker A
same data for like consumed multiple times and the traffic fall back to backup in case the primary is unavailable okay we solved it so now you can just uh you can redeploy your application congrats what's wrong with this here
Speaker A
yeah it's not scalable we are talking about hundreds of terabytes of data per day you definitely cannot process it on one note so what can we do here uh yeah we just split the whole data whole stream into shards and each chart will
Speaker A
have primary and backup each chart will consume the same like each chart will consume some partitions and all shards together will consume the whole volume of data from Kafka uh artists in the same charts are consuming the same
Speaker A
topics and partitions and Federated API we will query all shards to serve the request so now we can redeploy now we can maintain the volume and everybody is Happy well no not really uh you have a hardware duplication it is impossible to
Speaker A
scale uh it is impossible to scale querying separately for a magician and in case you're deploying being deployed on pods and or VMS VM disk size is limited so if you have so many so much data for last day let's say you have
Speaker A
like more than us and you have like a petabyte of data per day you will be like in a world of trouble with this approach because you just will need so much VMS to store your data for last day
Speaker A
but uh it it is so bad but it is something we are working with right now and it is it is how it is deployed in production and it is something some idea of improvement for the future for us
Speaker A
now um let's talk about historical uh historical is just a service which serves the queries for data which is older than 24 hours uh he'll quickly go through that uh not so important I mean the data and data
Speaker A
flow the map reduce the querying flow is exactly the same in real time only thing matters is how you query the data like how do you provide the interfaces implementation to the historical basically first thing you do you get the
Speaker A
blocks metadata metadata meaning how many blocks and partitions did you have for last uh for your time period of your query after you got this you definitely cache it because you don't want to code postgres for every for for your for your
Speaker A
queries for all of them you cash into memory after you obtain the blocks blocks uh rows you get the metadata for them you get the interners and everything after you got this you get the metadata of chunks and you cache all of this in memory
Speaker A
after you have after you identify it with chunks you need to read you fetch the logs from chunks and go into chunks as I'm going for data from self every time as I said it's not the best idea so
Speaker A
we decided to Cache it on disk as well historical and we have many more caches we have Block Level response cache and we have chunk level response cache and we have some more caches which I'm not going to
Speaker A
talk today and the morale here is if you have a network uh related service or the data reading node just cache as much as you can and it will make you be happy and it will help you to to survive the
Speaker A
workload caching is the king this is the morale okay and the last component bear with me guys five six more minutes please uh Federated APA Federated API is the component which distributes the queries between real-time and historical and
Speaker A
also it executes the cross DC requests so why do we need close DC basically we don't have any kind of cross DC replication because it will be first of all it's obviously too hard and we don't want to do that but it's not really the
Speaker A
actual reason if you're going to replicate all the logs you will burn the network it everybody will not be happy so you so we query blocks cross DC only in case we are required meaning if you need to get the data from all four or 10
Speaker A
data centers you need to go and distribute this query to all of our data centers now let's talk about local query processing so imagine you have a good query for last four days uh when you have a date in a query for
Speaker A
last four days you obviously go to cross the scene which I just discussed but in local DC you do service Discovery call because you might you you have a lot of different Services after you identified the services which you need to go you
Speaker A
created the RT cluster for last day meaning you query all artists which we discussed just now after it uh using the service Discovery you identify how many historicals you have and you split this request basically last day goes to real time and
Speaker A
the rest of the data the rest of the time range goes to historicals and here is the spiciest decision we made basically for each of the requests we are querying all historicals and for each request we are adding the
Speaker A
ID of specific instance so for example you have servers ABC so you saw them a gets ID one and C gets ID free and also you send the total number of total number of instances free what next so you send query to each historical
Speaker A
each historical receives instance ID and total number of servers for each chunk you remember this mapreduce histogram right like mapreduce picture when you identify the chunk versus additional value when you figure out do you need to read chunk or not on each historical you
Speaker A
calculate the hash of the chunk metadata and you decide which chart like which instance out of the ring you have total number of instances right in the current instance you identify does this instance ID like does the current instance which
Speaker A
I have ID for uh is actually the opener of this chunk if like um if it is the opener of this chunk uh for this hashing which I have so like imagine imagine you have like the same thing in the same way as
Speaker A
Cassandra does that does the instance selection using consistent hashing but we are doing it a bit in a weird uh in the viewer structure so and we are also using Rendezvous hashing not consistent hashing because not going to dive into details because
Speaker A
we don't have time the Rendezvous hashing is better if you have small number of instances and if your and if you need better like even spreading of the workload so using this determination you will each historical will read only only its
Speaker A
part of the chunks and you will get this the whole data like for example you have 25 chunks five instances and each chunk is going to read exactly I'm sorry each historical will read exactly five chunks so you obligate this data in each
Speaker A
historical you get it back to Federated API and using this formula Federated API we will have the data for like for the all historical time range what's the problem with this approach let's start with the good part so quite
Speaker A
straightforward it gives you the best possible cash hit rates and each instance receives even workload Enfield API doesn't know anything about the metadata as well like super simple that's that's exactly the reason why we did pick it up but the bad side is
Speaker A
querying all instances for each request like for example if you have 10 instances you need to query 10 instances if you have hundred instances you will need to query hundred instances another problem is all historicals need to Cache
Speaker A
the metadata that's something really big and this is really problematic so for example if your metadata will grow too big you will need to uh you will need to vertically scale each historical because horizontal scaling will not help
Speaker A
okay thank you for bearing to me with me to the end here comes the summary uh the strong sides of the system real-time logs access because the data stored on the same ingesting notes uh as we're being queried the access is super
Speaker A
fast no network is required uh indexing we have additional scaling as well as the low cardinality indexes and unlimited cardinality indexes scalability we can scale pretty well also efficient resource utilization since this whole thing is written in Rust we are quite resource efficient and
Speaker A
we don't worry need to worry about things like boxing and garbage collection and other things from java and the data compression since we are using dictionaries since we are using compression we are and we are using seriouses we are quite resource
Speaker A
efficient and data compression is pretty good something we need to improve uh real-time and ingestion basically all the lists from the backlog we need to improve to the real time another problem is in historical logs must be downloaded from S3 before they
Speaker A
can be processed not so much you can do about it but currently if you Group by by specific field or if you are interested only in one field during the histogram aggregation you will still need to read the whole uh the whole message
Speaker A
and it means you will load much more data when you actually need to serve your query and this is something we are going to mitigate in a sooner future by uh onboarding to parquet form all right that's it from my site please
Speaker A
submit your feedback and none uh let's start with the Q a session okay thank you Eugene um great sharing so now we uh open four questions Eugene you want to start from the questions in the chat first so we have questions from Eugene as well
Speaker A
um yes thank you none I see the question regarding the compression speed measurements are those numbers for just one CPU yes this is these numbers are just for one CPU uh basically I mean uh I mean you have a lot of
Speaker A
different threats and workers going on and it means like yeah but they will have the same um same problems or the same pros and cons basically for each other algorithm can we have both histograms and logs uh query both added to some dashboard uh
Speaker A
yes you definitely can and it will be just a separate uh separate queries to the to the system but you can definitely do it on the user side you can do it how does the river sorting reverse ordering sorting Works once you need to
Speaker A
read all blocks uh to the end uh yes and no it's actually the great question I really like it so for the Sorting of blocks uh we will just start like for example you query the data for last year we will find the last block
Speaker A
which you need and we will yes and we will basically have a queue uh like basically the Heap you can say we will read the Heap and we will identify the last logs in each block and yeah that's how we are going to do
Speaker A
it but order of blocks will be absolutely different if you uh if you will fetch the Sorting from the start so yeah basically this is how it works but it's definitely less efficient than uh getting the first looks
Speaker A
teams data compression requires a lot of CPU work using multi-course helps to improve the compression speed uh yeah I mean we definitely can explore more algorithms I guess but we didn't uh we didn't explore the internals of others uh algorithm compression
Speaker A
algorithms and I think we are currently deployed I mean I think currently we're all uh use only single thread for compression um so why not use Loki uh other things what Loki can do so yeah that's also the
Speaker A
great question uh so first of all first of all I want to stress out what the grafana lock is a great tool but it had some some limitations because of which we couldn't uh we couldn't use it in
Speaker A
Agoda first of all for example in a in Loki we couldn't have the custom indexing on Trace IDs so in Loki if you would need to query buy some Trace ID for let's say six months of data you
Speaker A
would need to read the whole six months of data in our case you just need to uh you can just use um you can just use Bloom filters and in that case you will need to just edit Bloom filters and we are actually
Speaker A
currently working on additional levels of indexing so because we are almost on our system we can customize our customize our indexing as much as possible that was the core idea behind writing our own system it's just like it
Speaker A
utilizes the same Locale it utilizes more or less the same principles but it requires a bit a different workload okay guys any other questions any other questions you can unmute yourself as well all right if not then we will end the
Speaker A
session here thank you so much Eugene for the great presentation today and thank you everyone for tuning in um hope you have learned something useful before you go I have some news for a Golden's um before uh so yeah we just created
Speaker A
this group on gitlab called Tech culture where everyone can suggest Tech talk topics in the future and upvote for topics that excite you the most so um we can work together to schedule the talks um yeah so please do visit the group and
Speaker A
explore it or reach out to me if you have any questions um we will also have another initiative coming up this month so please stay tuned for the announcement so that's a wrap for today thank you everyone for
Speaker A
attending um don't forget to Pro feedback as well um have a good rest of the week thank you so much guys [Music]
Topics:log aggregationAgodaLog Castroreal-time loggingobservabilityKafkaGrafana Lokilog managementhigh volume logsRust
![[YTP] Jordan Peterson: Rat King — Transcript](https://i.ytimg.com/vi/0qUoznGIe1M/maxresdefault.jpg)
![Nick & Charlie ‣ their story [+heartstopper forever] — Transcript](https://i.ytimg.com/vi/p4TiW1TZM3w/maxresdefault.jpg)
