Loading...
Let's consider examples from large-scale services such as Netflix.
We will consider only 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 2 hours per user:
79,020 bytes per hour x 2 hours=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()
Based on our capacity estimation, we can use S3 for long-term cold storage and Cassandra for hot, real-time storage. Cassandra’s partitioning capabilities can bucket logs into fixed intervals (e.g., 5 minutes), enabling efficient writes and queries for recent logs. To handle full-text search, we integrate Elasticsearch, optimized for querying and analyzing large-scale text data.
Key attributes:
We can partition by service_id and time interval, minimizing costly distributed queries. Logs can be denormalized by log_level and pre-aggregated for metrics like error counts.
Elasticsearch enables full-text search using a inverted index for log messages. Logs can be stored in time-based indices (e.g., logs-2024.12.13) for efficient retrieval and lifecycle management. Index Lifecycle Management automates archival and retention policies, while bulk indexing batches writes to improve performance.
Log Ingestion:
Log Processing:
Log Query:
Full-Text Search:
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.
This 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.4 TB). If we partition by 5 minute interval we have 288 buckets for a service.
Assuming we store the recommended 100 MB per partition (Cassandra's 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 when 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.
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.
Kafka
Kafka is used to handle sudden bursts of log traffic. Its distributed architecture allows it to process millions of events per second, ensuring scalability and resilience during high-traffic periods. Kafka acts as a buffer, decoupling log producers from downstream consumers and maintaining system stability under heavy loads.
Watermarking
During data aggregation, logs may arrive late or out of order, leading to potential errors in aggregated metrics. This issue can be mitigated using watermarking, where the system tolerates late-arriving messages up to a certain threshold. The downside of watermarking is that it introduces latency, as the system waits for a defined time before processing logs to ensure completeness.
Replication
The Shard Router and Metadata Store are critical components of the system. If either fails, it could cause system downtime. To mitigate this risk, these services must be designed to be highly available, employing techniques such as replication across multiple nodes or regions to ensure fault tolerance and continuity.
Pagination
Querying recent data may still involve accessing a large volume of rows. Querying all data at once is inefficient, especially when data spans multiple partitions. To prevent overwhelming the system with massive lookups, queries should be paginated, limiting results to manageable chunks (e.g., fetching the first 100 rows using a cursor). This approach ensures better performance and minimizes the risk of system overload.
Multi-tenant support
We can extend our design to support multi-tenancy by incorporating tenant_id into the partition key. This approach ensures that data from multiple tenants is stored logically isolated while still sharing the underlying infrastructure for scalability and efficiency.
Alert manager
We can explore setting up an alert manager to alert users if certain error number goes over the set threshold.