Loading...
let's consider examples from large-scale services such as Netflix:
Let consider their streaming log volume. Assume each user generates
500 bytes (buffering) + 600 bytes (bitrate)+200 bytes (playback controls)+17 bytes (device)=1,317bytes per minute.
Assume 230 million subscribers and assuming 10% are active daily, we estimate:
1,317 bytes per minute x 60minutes=79,020 bytes per hour.
assume average usage of 2hours per user
79,020 bytes per hour x 2hours=158,040 bytes per user.
23 million users x 158,040 bytes per user= 3.635 TB.
Assume a 20% buffer: 3.635 TB x 1.2=4.362 TB/day
Netflix could generate approximately 4.4 TB of logs daily just from streaming logs alone
These numbers illustrate that certain services can plausibly generate terabytes of logs daily.
We would probably need an object/blob store system like s3 to indefinitely store all the logs.
A distributed storage like Cassandra can be used to store hot data.
Search()
ViewLogOverPeriod()
ViewLogByLogLevel()
Database choice
As mentioned in the capacity estimation our strategy is to use S3 for cold long term storage while using Cassandra for hot storage. We can use Cassandra's partitioning capabilities to bucket the data by intervals (Per 5 minute).
One thing we would also need to consider is handling full text search. We would need to use elastic search.
Data Model
service_id
log_message
time_stamp
log_level
Based on this data model, we would partition the data by serviceId so that we don't have to perform distributed queries which are quite expensive.
denormalize by log_level
pre aggregate logs such as error count, success count and etc.
For elasticsearch we would use a reverse index on the log_message
log aggregation flow (write path)
view log over period (read path)
search (read path)
With this in mind our high level diagram would look as follow
message queue
Log collection service
See high level design read and write paths.
One challenge with a logging system is the potential for hot partitions, where a single service generates a disproportionate amount of log data. If we partition data solely by serviceId, certain high-traffic services could overwhelm specific partitions, leading to performance bottlenecks and storage inefficiencies.
To mitigate this, we can further distribute partitions by time intervals, creating a composite partition key such as (serviceId, interval_start_time). This approach ensures data is distributed more evenly across partitions and reduces the likelihood of oversize partitions.
It also allows us to efficiently query for a particular time interval for a specific service.
However, as mentioned in our capacity estimation certain services such as netflix's streaming logs could generate terabytes of data per day (4.4tb)
If we partition by 5 minute interval we have 288 buckets for a service.
Assuming we store the recommended 100 mb per partition (Cassandra recommendation), that would amount to:
288 buckets/day x 100 MB/partition = about 30 GB/day/service, which is not enough for services producing over 1 TB of log data daily.
So even with partitioning by service and time interval (e.g., 5-minute buckets), a high-traffic service could still lead to hotspots or excessively large partitions.
To address this, we need to allow shard counts to scale with traffic for each service. Dynamic sharding is needed.
Logging platforms like Datadog and Splunk use custom data stores optimized for dynamic sharding. While building such a system is impractical in an interview, we can approximate their behavior using existing technologies.
Cassandra allows partitioning by compound keys, enabling efficient data distribution. A dedicated shard router can manage shard assignments at the application layer. The router consults a metadata store to determine shard counts and shard-to-time mappings for each service.
Metadata Store
A high-consistency metadata store (e.g., FoundationDB) tracks shard assignments for each service and time range:
CREATE TABLE shard_metadata (
service_id TEXT,
time_range_start TIMESTAMP,
time_range_end TIMESTAMP,
shard_count INT,
shards LIST<INT>,
PRIMARY KEY ((service_id), time_range_start)
);
Shard Router
Logs are written to Cassandra using a schema that supports dynamic sharding. Noticed we added a shard_id field to the table.
CREATE TABLE logs (
service_id TEXT,
time_range_start TIMESTAMP,
shard_id INT,
log_timestamp TIMESTAMP,
log_message TEXT,
PRIMARY KEY ((service_id, time_range_start, shard_id), log_timestamp)
);
On write we use consistent hashing to evenly distribute writes across shards. This ensures balanced load and avoids overloading specific shards.
INSERT INTO logs (
service_id,
time_range_start,
shard_id,
log_timestamp,
log_message
) VALUES (
'auth-service',
'2024-12-13T11:05:00',
3,
'2024-12-13T11:05:37',
'User login successful for user_id=12345'
);
Each shard is mapped to a specific time range, and during read operations, only the shards relevant to that interval are queried via the mapping stored in the centralized metadata store.
To minimize query latency, each Shard Router node maintains an in-memory cache of the metadata it needs. Since shard placements are small and immutable for specific time intervals, they can be efficiently cached locally. Updates to shard metadata (e.g., changes in the number of shards or assignments) are managed by a central Sharding Allocator. These changes are broadcast to all Shard Router nodes, ensuring their caches remain up-to-date and preventing stale metadata.
Sharding Allocator
The Sharding Allocator determines which shards are allocated to each service for a given time range. Assignments are based on traffic volume, service-specific requirements, and the number of available shards in the system. The sharding allocator is also responsible for pre populating the metadata table upon first registration of the service and future time periods.
While this design reduces query fan-out, a challenge still arises when querying long time intervals, as it involves multi-partition queries across many shards, increasing latency. To mitigate this, we can use a hybrid approach:
By combining a strongly consistent metadata store, in-memory caching, and a hybrid storage model, this design ensures high query performance and scalability for both real-time and historical log analysis.
Our final design would look like below:
Alternative Architectures
Some systems choose to write logs directly to object storage (e.g., Amazon S3, Azure Blob Storage) before downstream services ingest the data. This approach is commonly used for batch processing or analytical workloads where low latency is not a critical requirement. Some examples include ETL pipelines or data warehousing workloads operate on a periodic schedule (e.g., hourly or daily) and do not require immediate log processing. The advantage of this approach is that logs can be stored in cheaper storage tiers, reducing the overall cost for batch workloads.
While pre aggregating the date it's possible that certain logs arrive late and out of order. This causes error during aggregation. One mitigation is to use watermarking by tolerating late arrival of message. This downside of this approach is that it would introduce some latency if we use watermarking.
The shard router and metadata store are central services. If they fail then our system could crash. We can mitigate this by making these high avaialble via replicas.
pagination
It would be a bad idea to query all data at once. We should limit the queries via pagination (eg first 100 from cursor). This would ensure our system doesn't get overwhelm with massive look ups.
kafka is use to handle sudden burst of traffic. Kafka is able to handle millions of events per second.
Alert manager