Design A Top K Request Analysis System - System Design

System requirements

Functional:

A service which returns the top K requests' information in last 'T' time
The K and T values are configurable(would be part of the request params)

Non-Functional:

The ingestion volume is large
The system is of desired high availability, we are fine with eventual-consistency
The system is required to be of low-latency.

Capacity estimation

Assumption:
- Min value of T is 1 minute and max value is in days(let say 30 days at max) for the scope of the problem(we can discuss later if we want to have broader scope).
- Max. value of K would be 100 as I am assuming that we want this information to be shown to the customer, this doesn't make sense to show like top 10k, 1M requests.
- The write/update TPS is higher as compared to the read one.
- We can assume that every second let say we are serving 100 requests, so daily it would be somewhere around: 10 Million requests per day.
- The number of read calls per second could be assumed to be of the rate of 1 TPS.

API design

POST: captureTheRequest(requestObject, epoch_timestamp)
- requestObject: It contains details like: {requestId, requestTargetEndOperation, response_status_code, caller_details)
  - It is requestTargetEndOperation that we want to get the top-k trending informations
  - requestId: this is unique for each request
  - caller_details: It contains informations about the caller of the operations
  - response_status_code: It involves the response_status_code(it is out of the scope of the current problem to filter out the requests with only a particular response_status, but we can keep it for the later purpose)
- Response: 201, created
Get: getTopKRequestsFor(K_number, T, time_units)
- K_number: we want top-k requests
- T: we want to find top requests per T time.
- time_units: it could minutes, hours, days.
  - If it is minutes: the T would be in between: {1, 59}
  - If it is in hours: the T would be in between: {1, 23}
  - If it is in days: the T would be in between: {1,30}
- Response: code: 200, response-body: JSON: List<requestTargetEndOperation>
  - requestTargetEndOperation is a string indicating the Top-k operations name.
  - every element in the list is unique(as it is picking one operation only once)

Database design

We will have following tables primarly:

MinutelyTrendingTable
HourlyTrendingTable
DailyTrendingTable

The table details would be something like: {requestTargetEndOperation, frequency}

For MinutelyTrendingTable: It will have the information like: {current_time_stamp_till_minute, requestTargetEndOperation, count} -- >retention-policy: 61 minutes.

For HourlyTrendingTable: It will have the information like: {current_time_stamp_till_minute, requestTargetEndOperation, count}-- >retention-policy: 25 hours, it could have max. records: 100

For DailyTrendingTable: It will have the information like: {current_time_stamp_till_hour, requestTargetEndOperation, count}-- >retention-policy: 31 days for now, , it could have max. records: 100

For each of the table time_stamp would be the partition-key(such that we can set retention-policy on the key) and requestTargetEndOperation as the sort key, hence if we want to find the frequency of an operation for a particular timestamp: we can query like: get me the count from

where timestamp="" and operation=""

We could use only one table for above three as well, but for the readibility purpose and to have separate retention-policies, for now, it is advised to go separately, later we can discuss if we want a single one.

As the table structure looks like a key-values kind of data, we could leverage both sql or DynamoDB for the purpose, Due to simplere data-queries, not being complex, higher scalability for DDB, non-requirement of ACID properties, we can leverage DDB in the above case.

There is one more table: current_minute_record

This will keep the live records of the current minute, its data would be like: timestamp_till_minute, requestTargetEndOperation, count, For this we can have either in-memory Redis server(for speed) or can have it in some key-value Database like DDB. Since, redis offers faster reading, we can have redis for this. If somehow this operations fails for some of the requests, this should be fine as we are more concerned about availability and fine with eventual-consistency.

, we could always have one re-consillation job.

High-level design

ApiGateway:

This acts as the gateway for us, it will provide:
- AuthZ/ AuthN, act as a reverse-proxy, can throttle the requests as well.

FeedService:

The service is behind the LB as write volume is high, this will provide both the above operations to the customer..
For the getTopKRequestsFor() operations:
- It queries the respective DB, minute-based or hour-based or daily based to get the top-K records for a particular time-period, It also leverages the current_minute_record to finally calculate the top-k requests. As in case of:
  - minuteTable: pre-computed[User could pass at max. 59 minutes, so we could have at max: 59*100(5900 records)] + current minute record(would be 1000 keeping tps of 100 in mind), we can use the approach of merge-sort merging to get top K post(complexiy is O(K)) in memory.
    - If the users passed the input as last
  - hourTable: User could pass at max. 23 hours, so we could have at max: pre-computed[23 * 100(2300 records)] + curent_minute record(1000, tps is 100) ==> 3300 records.
  - Daily table: user could pass at max. 30 days, so it could have data like 30*100(3000 records) + minuteTables(at max. 58 minute, 58 * 100 = 5800 records) + curent_minute record(1000, tps is 100) ==> 9800 records. We could easily, calculate the top K using the merging logic of sorted list here.. This could be acheived in memory and
- We can cache this response records to avoid future recalculation(cache-record: <start-timestamp_till_minute, end-timestamp_till_minute: List<top_request_operations>), in case of a cache. We can enhance this information getting whatever is saved in the cache for a start_time_stamp and then only query the DDB with the timestamps >= end_timestamp(but ofcourse, this would be a bit complex and can be kept out of the scope for now).
For the captureTheRequest(): It will populate the current_minute_record table and calls ingestion service via kafka to ingest this which in turn dumps it in minutely partitioned folders in S3(we can have retention of 1/2 years there for now).
We will have following airflow jobs:
- Minute based job: runs every minute to populate the minutelyTrendingTable with the top K requests from the S3 record for the minute. It processes t-1th minute record.
- Hour based job: runs every minute to populate the hourlyTrendingTable with the top K requests from the minutelyTrendingTable for the hour ending with t-1th minute. It processes t-1th hour record.
- Daily based job: runs every hour to populate the dailyTrendingTable with the top K requests from the hourlyTrendingTable for the day ending with t-1th hour. It processes t-1th day record.

Request flows

For the getTopKRequestsFor() operations:
- It queries the respective DB, minute-based or hour-based or daily based to get the top-K records for a particular time-period, It also leverages the current_minute_record to finally calculate the top-k requests. As in case of:
  - minuteTable: pre-computed[User could pass at max. 59 minutes, so we could have at max: 59*100(5900 records)] + current minute record(would be 1000 keeping tps of 100 in mind), we can use the approach of merge-sort merging to get top K post(complexiy is O(K)) in memory.
    - If the users passed the input as last
  - hourTable: User could pass at max. 23 hours, so we could have at max: pre-computed[23 * 100(2300 records)] + curent_minute record(1000, tps is 100) ==> 3300 records.
  - Daily table: user could pass at max. 30 days, so it could have data like 30*100(3000 records) + minuteTables(at max. 58 minute, 58 * 100 = 5800 records) + curent_minute record(1000, tps is 100) ==> 9800 records. We could easily, calculate the top K using the merging logic of sorted list here.. This could be acheived in memory and
- We can cache this response records to avoid future recalculation(cache-record: <start-timestamp_till_minute, end-timestamp_till_minute: List<top_request_operations>), in case of a cache. We can enhance this information getting whatever is saved in the cache for a start_time_stamp and then only query the DDB with the timestamps >= end_timestamp(but ofcourse, this would be a bit complex and can be kept out of the scope for now).
For the captureTheRequest(): It will populate the current_minute_record table and calls ingestion service via kafka to ingest this which in turn dumps it in minutely partitioned folders in S3(we can have retention of 1/2 years there for now).
We will have following airflow jobs:
- Minute based job: runs every minute to populate the minutelyTrendingTable with the top K requests from the S3 record for the minute. It processes t-1th minute record.
- Hour based job: runs every minute to populate the hourlyTrendingTable with the top K requests from the minutelyTrendingTable for the hour ending with t-1th minute. It processes t-1th hour record.
- Daily based job: runs every hour to populate the dailyTrendingTable with the top K requests from the hourlyTrendingTable for the day ending with t-1th hour. It processes t-1th day record.

Detailed component design

High-level design

ApiGateway:

This acts as the gateway for us, it will provide:
- AuthZ/ AuthN, act as a reverse-proxy, can throttle the requests as well.

FeedService:

The service is behind the LB as write volume is high, this will provide both the above operations to the customer..
For the getTopKRequestsFor() operations:
- It queries the respective DB, minute-based or hour-based or daily based to get the top-K records for a particular time-period, It also leverages the current_minute_record to finally calculate the top-k requests. As in case of:
  - minuteTable: User could pass at max. 59 minutes, so we could have at max: 59*100(5900 records) + current minute record(would be 1000 keeping tps of 100 in mind), we can use the approach of merge-sort merging to get top K post(complexiy is O(K)) in memory.
  - hourTable: User could pass at max. 23 hours, so we could have at max: 23 * 100(2300 records) + minuteTables(at max. 58 minute, 58 * 100 = 5800 records) + curent_minute record(1000, tps is 100) ==> 9100 records.
  - Daily table: user could pass at max. 30 days, so it could have data like 30*100(3000 records) + hourTable( 23 * 100 = 2300) + minuteTables(at max. 58 minute, 58 * 100 = 5800 records) + curent_minute record(1000, tps is 100) ==> 12100 records. We could easily, calculate the top K using the merging logic of sorted list here.. This could be acheived in memory and
- we can cache this response records to avoid future recalculation(cache-record: <start-timestamp_till_minute, end-timestamp_till_minute: List<top_request_operations>), in case of a cache. We can enhance this information getting whatever is saved in the cache for a start_time_stamp and then only query the DDB with the timestamps >= end_timestamp(but ofcourse, this would be a bit complex and can be kept out of the scope for now).
For the captureTheRequest(): It will populate the current_minute_record table and calls ingestion service via kafka to ingest this which in turn dumps it in minutely partitioned folders in S3(we can have retention of 1/2 years there for now).
We will have following airflow jobs:
- Minute based job: runs every minute to populate the minutelyTrendingTable with the top K requests from the S3 record for the minute. It processes t-1th minute record.
- Hour based job: runs every hour to populate the hourlyTrendingTable with the top K requests from the minutelyTrendingTable for the hour. It processes t-1th hour record.
- Daily based job: runs every hour to populate the dailyTrendingTable with the top K requests from the hourlyTrendingTable for the day. It processes t-1th day record.

Trade offs/Tech choices

We have leveraged DDB, we could have used the sql here as well to provide the better latency. But as our queries are going to be simpler and we are fine with eventual consistency, I went with DDB.
Our services interacts interally via means of Grpc clients with exponential back-off retry strategy in place..
We will have multiple replicas of the feed/ingestion services to avoid any single point of failure
The airflow jobs will run in self-healing mode: means if for a particular timestamp, it will retry, an airflow jobs fails, the next minute will try to back-fill the data.

Failure scenarios/bottlenecks

Lets say for a particular minute, one of the spark-job didn't run: We can write our move-jobs in such a fashion that it takes input the last-successful run and act accordingly to do parallel dumping of data to their respective minute folder

For other jobs, as airflow: implicitly does 3-4(configurable) retries, this gives some level of fault-tolerance. But lets assume that if that also fails, to support availability, we can have following guard rail:
- we could trigger another monitoring job, which will just copy the data from n-1th timestamped partition to the current one. This seems scalable.

Future improvements

We will have some canaries running as live customers, to fire showMeTopFeed() api, in case of any faulty response, they would trigger auto-alarms.
We will have to emit faults/error/P90 latency spike metrics and have graphs built over these with alarms thresholds pre-set to catch any failure/fault/spike.

System requirements

Functional:

A service which returns the top K requests' information in last 'T' time
The K and T values are configurable(would be part of the request params)

Non-Functional:

The ingestion volume is large
The system is of desired high availability, we are fine with eventual-consistency
The system is required to be of low-latency.

Capacity estimation

Assumption:
- Min value of T is 1 minute and max value is in days(let say 30 days at max) for the scope of the problem(we can discuss later if we want to have broader scope).
- Max. value of K would be 100 as I am assuming that we want this information to be shown to the customer, this doesn't make sense to show like top 10k, 1M requests.
- The write/update TPS is higher as compared to the read one.
- We can assume that every second let say we are serving 100 requests, so daily it would be somewhere around: 10 Million requests per day.
- The number of read calls per second could be assumed to be of the rate of 1 TPS.

API design

POST: captureTheRequest(requestObject, epoch_timestamp)
- requestObject: It contains details like: {requestId, requestTargetEndOperation, response_status_code, caller_details)
  - It is requestTargetEndOperation that we want to get the top-k trending informations
  - requestId: this is unique for each request
  - caller_details: It contains informations about the caller of the operations
  - response_status_code: It involves the response_status_code(it is out of the scope of the current problem to filter out the requests with only a particular response_status, but we can keep it for the later purpose)
- Response: 201, created
Get: getTopKRequestsFor(K_number, T, time_units)
- K_number: we want top-k requests
- T: we want to find top requests per T time.
- time_units: it could minutes, hours, days.
  - If it is minutes: the T would be in between: {1, 59}
  - If it is in hours: the T would be in between: {1, 23}
  - If it is in days: the T would be in between: {1,30}
- Response: code: 200, response-body: JSON: List<requestTargetEndOperation>
  - requestTargetEndOperation is a string indicating the Top-k operations name.
  - every element in the list is unique(as it is picking one operation only once)

Database design

We will have following tables primarly:

MinutelyTrendingTable
HourlyTrendingTable
DailyTrendingTable

The table details would be something like: {requestTargetEndOperation, frequency}

For MinutelyTrendingTable: It will have the information like: {current_time_stamp_till_minute, requestTargetEndOperation, count} -- >retention-policy: 61 minutes.

For HourlyTrendingTable: It will have the information like: {current_time_stamp_till_minute, requestTargetEndOperation, count}-- >retention-policy: 25 hours, it could have max. records: 100

For DailyTrendingTable: It will have the information like: {current_time_stamp_till_hour, requestTargetEndOperation, count}-- >retention-policy: 31 days for now, , it could have max. records: 100

where timestamp="" and operation=""

There is one more table: current_minute_record

, we could always have one re-consillation job.

High-level design

ApiGateway:

This acts as the gateway for us, it will provide:
- AuthZ/ AuthN, act as a reverse-proxy, can throttle the requests as well.

FeedService:

The service is behind the LB as write volume is high, this will provide both the above operations to the customer..
For the getTopKRequestsFor() operations:
- It queries the respective DB, minute-based or hour-based or daily based to get the top-K records for a particular time-period, It also leverages the current_minute_record to finally calculate the top-k requests. As in case of:
  - minuteTable: pre-computed[User could pass at max. 59 minutes, so we could have at max: 59*100(5900 records)] + current minute record(would be 1000 keeping tps of 100 in mind), we can use the approach of merge-sort merging to get top K post(complexiy is O(K)) in memory.
    - If the users passed the input as last
  - hourTable: User could pass at max. 23 hours, so we could have at max: pre-computed[23 * 100(2300 records)] + curent_minute record(1000, tps is 100) ==> 3300 records.
  - Daily table: user could pass at max. 30 days, so it could have data like 30*100(3000 records) + minuteTables(at max. 58 minute, 58 * 100 = 5800 records) + curent_minute record(1000, tps is 100) ==> 9800 records. We could easily, calculate the top K using the merging logic of sorted list here.. This could be acheived in memory and
- We can cache this response records to avoid future recalculation(cache-record: <start-timestamp_till_minute, end-timestamp_till_minute: List<top_request_operations>), in case of a cache. We can enhance this information getting whatever is saved in the cache for a start_time_stamp and then only query the DDB with the timestamps >= end_timestamp(but ofcourse, this would be a bit complex and can be kept out of the scope for now).
For the captureTheRequest(): It will populate the current_minute_record table and calls ingestion service via kafka to ingest this which in turn dumps it in minutely partitioned folders in S3(we can have retention of 1/2 years there for now).
We will have following airflow jobs:
- Minute based job: runs every minute to populate the minutelyTrendingTable with the top K requests from the S3 record for the minute. It processes t-1th minute record.
- Hour based job: runs every minute to populate the hourlyTrendingTable with the top K requests from the minutelyTrendingTable for the hour ending with t-1th minute. It processes t-1th hour record.
- Daily based job: runs every hour to populate the dailyTrendingTable with the top K requests from the hourlyTrendingTable for the day ending with t-1th hour. It processes t-1th day record.

Request flows

For the getTopKRequestsFor() operations:
- It queries the respective DB, minute-based or hour-based or daily based to get the top-K records for a particular time-period, It also leverages the current_minute_record to finally calculate the top-k requests. As in case of:
  - minuteTable: pre-computed[User could pass at max. 59 minutes, so we could have at max: 59*100(5900 records)] + current minute record(would be 1000 keeping tps of 100 in mind), we can use the approach of merge-sort merging to get top K post(complexiy is O(K)) in memory.
    - If the users passed the input as last
  - hourTable: User could pass at max. 23 hours, so we could have at max: pre-computed[23 * 100(2300 records)] + curent_minute record(1000, tps is 100) ==> 3300 records.
  - Daily table: user could pass at max. 30 days, so it could have data like 30*100(3000 records) + minuteTables(at max. 58 minute, 58 * 100 = 5800 records) + curent_minute record(1000, tps is 100) ==> 9800 records. We could easily, calculate the top K using the merging logic of sorted list here.. This could be acheived in memory and
- We can cache this response records to avoid future recalculation(cache-record: <start-timestamp_till_minute, end-timestamp_till_minute: List<top_request_operations>), in case of a cache. We can enhance this information getting whatever is saved in the cache for a start_time_stamp and then only query the DDB with the timestamps >= end_timestamp(but ofcourse, this would be a bit complex and can be kept out of the scope for now).
For the captureTheRequest(): It will populate the current_minute_record table and calls ingestion service via kafka to ingest this which in turn dumps it in minutely partitioned folders in S3(we can have retention of 1/2 years there for now).
We will have following airflow jobs:
- Minute based job: runs every minute to populate the minutelyTrendingTable with the top K requests from the S3 record for the minute. It processes t-1th minute record.
- Hour based job: runs every minute to populate the hourlyTrendingTable with the top K requests from the minutelyTrendingTable for the hour ending with t-1th minute. It processes t-1th hour record.
- Daily based job: runs every hour to populate the dailyTrendingTable with the top K requests from the hourlyTrendingTable for the day ending with t-1th hour. It processes t-1th day record.

Detailed component design

High-level design

ApiGateway:

This acts as the gateway for us, it will provide:
- AuthZ/ AuthN, act as a reverse-proxy, can throttle the requests as well.

FeedService:

The service is behind the LB as write volume is high, this will provide both the above operations to the customer..
For the getTopKRequestsFor() operations:
- It queries the respective DB, minute-based or hour-based or daily based to get the top-K records for a particular time-period, It also leverages the current_minute_record to finally calculate the top-k requests. As in case of:
  - minuteTable: User could pass at max. 59 minutes, so we could have at max: 59*100(5900 records) + current minute record(would be 1000 keeping tps of 100 in mind), we can use the approach of merge-sort merging to get top K post(complexiy is O(K)) in memory.
  - hourTable: User could pass at max. 23 hours, so we could have at max: 23 * 100(2300 records) + minuteTables(at max. 58 minute, 58 * 100 = 5800 records) + curent_minute record(1000, tps is 100) ==> 9100 records.
  - Daily table: user could pass at max. 30 days, so it could have data like 30*100(3000 records) + hourTable( 23 * 100 = 2300) + minuteTables(at max. 58 minute, 58 * 100 = 5800 records) + curent_minute record(1000, tps is 100) ==> 12100 records. We could easily, calculate the top K using the merging logic of sorted list here.. This could be acheived in memory and
- we can cache this response records to avoid future recalculation(cache-record: <start-timestamp_till_minute, end-timestamp_till_minute: List<top_request_operations>), in case of a cache. We can enhance this information getting whatever is saved in the cache for a start_time_stamp and then only query the DDB with the timestamps >= end_timestamp(but ofcourse, this would be a bit complex and can be kept out of the scope for now).
For the captureTheRequest(): It will populate the current_minute_record table and calls ingestion service via kafka to ingest this which in turn dumps it in minutely partitioned folders in S3(we can have retention of 1/2 years there for now).
We will have following airflow jobs:
- Minute based job: runs every minute to populate the minutelyTrendingTable with the top K requests from the S3 record for the minute. It processes t-1th minute record.
- Hour based job: runs every hour to populate the hourlyTrendingTable with the top K requests from the minutelyTrendingTable for the hour. It processes t-1th hour record.
- Daily based job: runs every hour to populate the dailyTrendingTable with the top K requests from the hourlyTrendingTable for the day. It processes t-1th day record.

Trade offs/Tech choices

We have leveraged DDB, we could have used the sql here as well to provide the better latency. But as our queries are going to be simpler and we are fine with eventual consistency, I went with DDB.
Our services interacts interally via means of Grpc clients with exponential back-off retry strategy in place..
We will have multiple replicas of the feed/ingestion services to avoid any single point of failure
The airflow jobs will run in self-healing mode: means if for a particular timestamp, it will retry, an airflow jobs fails, the next minute will try to back-fill the data.

Failure scenarios/bottlenecks

For other jobs, as airflow: implicitly does 3-4(configurable) retries, this gives some level of fault-tolerance. But lets assume that if that also fails, to support availability, we can have following guard rail:
- we could trigger another monitoring job, which will just copy the data from n-1th timestamped partition to the current one. This seems scalable.

Future improvements

We will have some canaries running as live customers, to fire showMeTopFeed() api, in case of any faulty response, they would trigger auto-alarms.
We will have to emit faults/error/P90 latency spike metrics and have graphs built over these with alarms thresholds pre-set to catch any failure/fault/spike.