Service developed using Spring Boot. Due to nature of this traffic I used memory and not any database as explained Scalability. The service store aggregated Flow log into structure like
Map<Integer,Map<String,Flow>> aggMap = ConcurrentHashMap<>();
Which the Integer key is Hour and String key is combination of "src_app + dest_app + vpc_id" seperated by "|".
The ConcurrentHashMap is a threadsafe and efficient for concurrent access.
AtomicInteger used to make aggregation calculation Atomic
Since hours increasing it required a separate process to clean up old hours if no longer required.
Service relay on ConcurrentHashMap and performance bottleneck would be concurrency limit of ConcurrentHashMap, specialty when new Aggregation Flow added to Map, this process it required lock on part of map.
Solution in not horizontally accessible, a better solution is to use a time-series database or writing to streaming service like Kafka or AWS Kinesis Stream and process it into a distributed cache like Redis, memcashed or Couchbase.
Code has a test that call controller in a multi-thread load test, 30 threads and each thread send 10,000 request. The send operation test result over 300,000 TPS on i7 laptop
This is a Spring Boot application, to start run this command:
mvn spring-boot:run
curl -X POST "http://localhost:8080/flows" \
-H 'Content-Type: application/json' \
-d '[
{"src_app": "foo", "dest_app": "bar", "vpc_id": "vpc-0", "bytes_tx": 100, "bytes_rx": 300, "hour": 1},
{"src_app": "foo", "dest_app": "bar", "vpc_id": "vpc-0", "bytes_tx": 200, "bytes_rx": 600, "hour": 1},
{"src_app": "baz", "dest_app": "qux", "vpc_id": "vpc-0", "bytes_tx": 100, "bytes_rx": 500, "hour": 1},
{"src_app": "baz", "dest_app": "qux", "vpc_id": "vpc-0", "bytes_tx": 100, "bytes_rx": 500, "hour": 2},
{"src_app": "baz", "dest_app": "qux", "vpc_id": "vpc-1", "bytes_tx": 100, "bytes_rx": 500, "hour": 2}
]'
curl "http://localhost:8080/flows?hour=1"