ISE<\/code> team at Microsoft<\/code> recently completed an engagement with a large industrial customer. The system we developed was distributed between the edge and the cloud and made use of technologies including Python<\/code>, Azure IoT<\/code>, Kubernetes<\/code>, Redis<\/code> and Dapr<\/code>. The system was built prior to this engagement and our goal for this phase was to make the system production ready and handle a greater scale of messages.<\/p>\nIn this system, messages originating from the edge would be sent to the cloud where they would be augmented with additional data from the company\u2019s various internal APIs and document storage. Augmented messages would then be processed further in the system; however, this blog post will focus on features surrounding the augmenting process.<\/p>\n
Problems<\/h2>\n
From initial investigations a couple of problems were observed within the message augmentation process.<\/p>\n
Speed of processing<\/h3>\n
A single message could take up to as much as a minute to go through the full augmentation process. With our goal in mind this was deemed insufficient.<\/p>\n
Influencing factors:<\/p>\n
\n- In some cases the data returned from the APIs was very large.<\/li>\n
- Only basic caching was in place and many messages required the same data resulting in repeated retrieval and processing of the same data.<\/li>\n
- Some of the company\u2019s internal APIs had \u2018rate limiting\u2019 in place. Meaning, if the rate limit was hit, we would have to wait a determined length of time before attempting to call again.<\/li>\n
- Each message was processed sequentially, and each step of the augmentation was performed sequentially as well.<\/li>\n<\/ul>\n
Out of memory exceptions<\/h3>\n
On the container performing the augmentation process exit code 137<\/code> \/ OOMKilled<\/code> was observed frequently. This was a significant problem as this would cause the message being processed at the time of error to be dropped.<\/p>\nThe major contributor to this was the way external documents were processed from the Storage Accounts by using the .readall()<\/code> method of the Azure storage client library<\/code>. Reading the entire document into memory with large documents was easy to encounter the container memory limit and cause the container to be shutdown.<\/p>\nExample below, taken from Download a blob with Python – Azure Storage | Microsoft Learn<\/a>, shows downloading the complete contents of the sample-blob.txt<\/code> file using this method and prints the contents.<\/p>\ndef download_blob_to_string(self, blob_service_client: BlobServiceClient, container_name):\r\n blob_client = blob_service_client.get_blob_client(container=container_name, blob=\"sample-blob.txt\")\r\n # encoding param is necessary for readall() to return str, otherwise it returns bytes\r\n downloader = blob_client.download_blob(max_concurrency=1, encoding='UTF-8')\r\n blob_text = downloader.readall()\r\n print(f\"Blob contents: {blob_text}\")<\/code><\/pre>\nNo queuing<\/h3>\n
Any message sent to the cloud was attempted to be processed immediately. No queuing system was in place. Being cautious of future growth in the system and the number of messages it may process, this was deemed a problem.<\/p>\n
Solutions<\/h2>\n
This next section will describe a number of actions taken to resolve the problems previously identified.<\/p>\n
Dedicated Caching Process<\/h3>\n
We introduced a separate process with the sole purpose of building a cache of frequently used data between messages. This ran in its own container on a CRON<\/code><\/a> schedule, allowing easy control of how frequently we update the cache.<\/p>\nThis separation of concerns gave many advantages:<\/p>\n
\n- This reduced the need to lookup \/ filter any commonly used data from APIs or data storage, resulting in faster message processing.<\/li>\n
- Avoids situations where a failure in accessing data sources does not result in a failure of processing a message.<\/li>\n
- Significant reduction in memory usage during message processing. Higher memory usage is only observed during building of cache, but this is known and manageable. Memory and CPU can be added to the container to speed up the cache building process, but having lower limits no longer affects the message processing.<\/li>\n
- When we scale to have multiple instances of the augmentation process, they can all share the same cached data.<\/li>\n<\/ul>\n
Redis<\/code> Queue \/ Dapr<\/code><\/h3>\nA cloud side queue was implemented to ingest edge messages and have the augmentation process consume messages from the queue for processing.<\/p>\n
This was done with a Redis<\/code> container<\/a> for the queue which was then accessed via Dapr<\/code><\/a>.<\/p>\nDapr<\/code> was chosen as it provides an interface to a variety of backend implementations of queueing \/ storage etc. The team had considerations of other queuing implementations, such as Azure Event Grid<\/code>. With Dapr<\/code> in place if the customer development team wish to rework their architecture this will be easy to do.<\/p>\nThe code sample below demonstrates publishing events to queue with the DaprClient<\/code> class. As can be seen, there is no reference to the underlying implementation. Example source How to: Publish a message and subscribe to a topic | Dapr Docs<\/a>.<\/p>\nwith DaprClient() as client:\r\n result = client.publish_event(\r\n pubsub_name=PUBSUB_NAME,\r\n topic_name=TOPIC_NAME,\r\n data=json.dumps(my_data),\r\n data_content_type='application\/json',\r\n )<\/code><\/pre>\nThe next sample shows the subscription to a topic. Sample taken from Declarative, streaming, and programmatic subscription types | Dapr Docs<\/a>.<\/p>\nfrom cloudevents.sdk.event import v1\r\n\r\n@app.route('\/orders', methods=['POST'])\r\ndef process_order(event: v1.Event) -> None:\r\n data = json.loads(event.Data())\r\n logging.info('Subscriber received: ' + str(data))<\/code><\/pre>\nThe reference to the Redis<\/code> implementation is defined within a Dapr<\/code> Component<\/code> configuration file.<\/p>\nAn example Redis<\/code> pub\/sub component definition follows Redis Streams | Dapr Docs<\/a>.<\/p>\napiVersion: dapr.io\/v1alpha1\r\nkind: Component\r\nmetadata:\r\n name: redis-pubsub\r\nspec:\r\n type: pubsub.redis\r\n version: v1\r\n metadata:\r\n - name: redisHost\r\n value: localhost:6379\r\n - name: redisPassword\r\n value: \"<Password>\"\r\n - name: consumerID\r\n value: \"channel1\"\r\n - name: useEntraID\r\n value: \"true\"\r\n - name: enableTLS\r\n value: \"false\"<\/code><\/pre>\nThe complimenting subscription definition is as follows How to: Publish a message and subscribe to a topic | Dapr Docs<\/a>.<\/p>\napiVersion: dapr.io\/v2alpha1\r\nkind: Subscription\r\nmetadata:\r\nname: order-pub-sub\r\nspec:\r\ntopic: orders\r\nroutes: \r\n default: \/orders\r\npubsubname: redis-pubsub\r\nscopes:\r\n- orderprocessing <\/code><\/pre>\nRead docs as a stream<\/h3>\n
As previously mentioned, when reading documents from blob storage the system used the .readall()<\/code> method loading the complete document into memory. It was not foreseen that we would read such large documents for this to be a problem.<\/p>\nThe method of reading was changed to read documents by chunks as a stream.<\/p>\n
Note:\nThis approach is best suited to data that can be consumed in chunks or lines\nE.g. .csv<\/code> or textual data.<\/p><\/blockquote>\nExample below shows the equivalent from the previous sample<\/a> now iterating on each chunk of the document.<\/p>\ndef download_blob_chunks(self, blob_service_client: BlobServiceClient, container_name):\r\n blob_client = blob_service_client.get_blob_client(container=container_name, blob=\"sample-blob.txt\")\r\n # This returns a StorageStreamDownloader\r\n stream = blob_client.download_blob()\r\n chunk_list = []\r\n\r\n # Read data in chunks to avoid loading all into memory at once\r\n for chunk in stream.chunks():\r\n # Process your data (anything can be done here - 'chunk' is a byte array)\r\n chunk_list.append(chunk)<\/code><\/pre>\nSource Download a blob with Python – Azure Storage | Microsoft Learn<\/a>.<\/p>\nLoad testing<\/h2>\n
Load testing was performed at the beginning of the project and after the discussed changes were implemented.<\/p>\n
The graph below demonstrates the change in performance. Here we measure the time taken to process a number of messages.<\/p>\n
![\"load-testing-graph\"](\"https:\/\/devblogs.microsoft.com\/ise\/wp-content\/uploads\/sites\/55\/2025\/08\/load-testing-graph.png\")
<\/p>\n
As you can see, a significant performance improvement has been made. These changes have been successful in contributing to the goal of having the system production ready and to be able to handle a greater scale of messages.<\/p>\n
Future Improvements<\/h2>\nStorage with Dapr<\/code><\/h3>\nDapr<\/code> has the ability to abstract state storage in a similar fashion to the queue mechanism previously described. Quickstart demo available here<\/a>.<\/p>\nwith DaprClient() as client:\r\n\r\n # Save state into the state store\r\n client.save_state(DAPR_STORE_NAME, orderId, str(order))\r\n logging.info('Saving Order: %s', order)\r\n\r\n # Get state from the state store\r\n result = client.get_state(DAPR_STORE_NAME, orderId)\r\n logging.info('Result after get: ' + str(result.data))\r\n\r\n # Delete state from the state store\r\n client.delete_state(store_name=DAPR_STORE_NAME, key=orderId)\r\n logging.info('Deleting Order: %s', order)<\/code><\/pre>\nHere we use the DaprClient<\/code> class to interface with the underlying state storage implementation.<\/p>\nWith an appropriate state storage component file defined. Such as the following for Azure Blob Storage<\/code><\/a>.<\/p>\napiVersion: dapr.io\/v1alpha1\r\nkind: Component\r\nmetadata:\r\n name: <NAME>\r\nspec:\r\n type: state.azure.blobstorage\r\n version: v2\r\n metadata:\r\n - name: accountName\r\n value: \"[your_account_name]\"\r\n - name: accountKey\r\n value: \"[your_account_key]\"\r\n - name: containerName\r\n value: \"[your_container_name]\"<\/code><\/pre>\nSimilarly to the queuing system, this provides the flexibility to change the implemented storage backend via configuration and not code.<\/p>\n
Scaling<\/h3>\nCurrent Scaling<\/h4>\n
The augmentation process container is currently scaled using the Kubernetes<\/code> built-in Horizontal Pod Autoscaler<\/code> (HPA<\/code>) to increase the number of instances. Scaling is based on CPU %, with the assumption that when we have high CPU usage we are processing a high number of messages.<\/p>\nAn example Helm<\/code> chart for HPA<\/code> is as follows:<\/p>\napiVersion: autoscaling\/v1\r\nkind: HorizontalPodAutoscaler\r\nmetadata:\r\n name: contextualizer\r\n namespace: default\r\nspec:\r\n scaleTargetRef:\r\n apiVersion: apps\/v1\r\n kind: Deployment\r\n name: contextualizer\r\n minReplicas: 1\r\n maxReplicas: 5\r\n metrics:\r\n - type: Resource\r\n resource:\r\n name: cpu\r\n target:\r\n type: Utilization\r\n averageUtilization: 75<\/code><\/pre>\nThe value averageUtilization: 75<\/code> means that we calculate the average cpu utilization across the current set of pods and if we exceed 75 then the system will scale.\nFurther details on HPA<\/code> can be found here<\/a>.<\/p>\nKEDA<\/code><\/h3>\n