System requirements

Functional:

• Collect logs from various sources: internal services, HTTP logs
• Provide long term storage for logs
• Support search on logs
• Compute near real-time aggregate metrics for services
• Generate alerts based on user specified rules
• Out of scope
  • Detect anomalies
  • Log retention

Non-Functional:

• High availability (99.99%)
• Eventual consistency is fine
• Highly scalable
• Fault tolerant
• Minimal latency generation of metrics and alerts: minutes (as opposed to seconds)
• Out of scope
  • Security

Capacity estimation

Storage

• Assume 10k services, each generates 10 lines of logs / sec. Each line of log on average is 200B.
• We have 10k * 10 * 200B * 86400 = 2TB / day. In one year it is 700TB. In 5 years 3.5PB.

Number of logs:

• 10k * 10 = 100k line/sec
• Within the limit of one kafka queue but we should partition it.

Connections

• Potentially 10k concurrent connections. Within the limit of 1 server, but we should partition it.

API design

// Search log

searchLog(query, limit, pageSize, pageToken) // consider ES-like query

// Get metrics

getMetrics(query) // consider PQL-like query

Database design

We choose several different storage:

• Use Cassandra for hot log storage, for its high write throughput and horizontal scalability
• Use S3 for cold storage
• Use ElasticSearch to index the logs for flexible search
• Use a time series DB like Prometheus for metrics

We implement a standard schemas for logs and metrics.

Log:

• timestamp
• serviceName
• serviceId
• region
• owner
• ...
• message

Metrics

• timestamp
• serviceName
• serviceId
• region
• owner
• metricName
• metricType

High-level design

LogAgent is deployed alongside with service (sidecar, or as a agent service in the same container) that ships logs to the LogQueue.

From the log queue, we have multiple consumers:

• LogService writes the log to hot storage (cassandra) and cold storage (blob)
• IndexService pushes the logs into ElasticSearch
• MetricsService pushes metrics data into a time series database like Prometheus

Client send queries to QueryService, which forwards the queries to ES or Prometheus.

Request flows

Ingestion flow

• LogAgent asynchronously writes log to Kafka queue
• LogService writes the log to hot storage (cassandra) and cold storage (blob)
• IndexService pushes the logs into ElasticSearch
• MetricsService pushes metrics data into a time series database like Prometheus

Query path

• For time-based log queries, it is served by Cassandra
• For keyword searches, the query goes to ElasticSearch
• For metric queries, it goes to Prometheus

Detailed component design

One key issue is that a hot service can generate disproportionate amount of logs. So sharding by service id is not sufficient. In addition to service id, we also shard by time interval. For example, we shard by every 1 hour interval. Assume 10% of the 10k services generate 90% of the logs, then each such service: 100 logs/sec and 20k / sec which should fit within one shard.

Trade offs/Tech choices

TSDB

• It is not efficient for high-dimensionality aggregations.
• It has a flexible query language, very compact (1-2 bytes per sample), and very flexible.
• For high-dimensionality, we should offload that to a different process, or perform pre-aggregation.

Failure scenarios/bottlenecks

Try to discuss as many failure scenarios/bottlenecks as possible.

Future improvements

What are some future improvements you would make? How would you mitigate the failure scenario(s) you described above?