fix: handle worker messages larger than 64 KB on Azure ACI (#3662)

Author: hassy

packages/artillery/lib/platform/aws-ecs/legacy/plugins/artillery-plugin-sqs-reporter/azure-aqs.js

@@ -4,7 +4,9 @@
 // Non-evaluation use of Artillery on Azure requires a commercial license
 
 const { QueueClient } = require('@azure/storage-queue');
+const { BlobServiceClient } = require('@azure/storage-blob');
 const { DefaultAzureCredential } = require('@azure/identity');
+const { randomUUID } = require('node:crypto');
 
 function getAQS() {
   return new QueueClient(
@@ -13,16 +15,65 @@ function getAQS() {
   );
 }
 
-function sendMessage(queue, body, tags) {
+// Azure Queue Storage has a 64KB message limit
+// Use 60KB threshold to leave margin for encoding overhead
+const AQS_SIZE_LIMIT = 60 * 1024;
+
+let blobContainerClient = null;
+
+function getBlobClient() {
+  if (!blobContainerClient) {
+    const storageAccount = process.env.AZURE_STORAGE_ACCOUNT;
+    const containerName = process.env.AZURE_STORAGE_BLOB_CONTAINER;
+    if (!storageAccount || !containerName) {
+      throw new Error(
+        'AZURE_STORAGE_ACCOUNT and AZURE_STORAGE_BLOB_CONTAINER must be set'
+      );
+    }
+    const blobServiceClient = new BlobServiceClient(
+      `https://${storageAccount}.blob.core.windows.net`,
+      new DefaultAzureCredential()
+    );
+    blobContainerClient = blobServiceClient.getContainerClient(containerName);
+  }
+  return blobContainerClient;
+}
+
+async function sendMessage(queue, body, tags) {
   const payload = JSON.stringify({
     payload: body,
-    // attributes: this.tags
     attributes: tags.reduce((acc, tag) => {
       acc[tag.key] = tag.value;
       return acc;
     }, {})
   });
 
+  // Check if payload exceeds Azure Queue Storage limit
+  if (Buffer.byteLength(payload, 'utf8') > AQS_SIZE_LIMIT) {
+    // Upload to blob storage and send reference
+    const testId = tags.find((t) => t.key === 'testId')?.value;
+    const workerId = tags.find((t) => t.key === 'workerId')?.value;
+    const messageId = randomUUID();
+    const blobName = `tests/${testId}/overflow/${workerId}/${messageId}.json`;
+
+    const blobClient = getBlobClient().getBlockBlobClient(blobName);
+    await blobClient.upload(payload, Buffer.byteLength(payload, 'utf8'));
+
+    // Send reference message
+    const refPayload = JSON.stringify({
+      payload: {
+        _overflowRef: blobName,
+        event: body.event
+      },
+      attributes: tags.reduce((acc, tag) => {
+        acc[tag.key] = tag.value;
+        return acc;
+      }, {})
+    });
+
+    return queue.sendMessage(refPayload);
+  }
+
   return queue.sendMessage(payload);
 }
 

packages/artillery/lib/platform/az/aci.js

@@ -24,6 +24,20 @@ const dotenv = require('dotenv');
 const fs = require('node:fs');
 const request = require('got');
 
+// Helper to convert readable stream to string
+async function streamToString(readableStream) {
+  return new Promise((resolve, reject) => {
+    const chunks = [];
+    readableStream.on('data', (data) => {
+      chunks.push(data.toString());
+    });
+    readableStream.on('end', () => {
+      resolve(chunks.join(''));
+    });
+    readableStream.on('error', reject);
+  });
+}
+
 class PlatformAzureACI {
   constructor(script, variablePayload, opts, platformOpts) {
     this.script = script;
@@ -257,6 +271,28 @@ class PlatformAzureACI {
             throw new Error('AQS message with an empty body');
           }
 
+          // Handle overflow messages stored in blob storage
+          if (payload._overflowRef) {
+            try {
+              const blobClient =
+                this.blobContainerClient.getBlockBlobClient(
+                  payload._overflowRef
+                );
+              const downloadResponse = await blobClient.download(0);
+              const downloaded = await streamToString(
+                downloadResponse.readableStreamBody
+              );
+              const fullMessage = JSON.parse(downloaded);
+              payload = fullMessage.payload;
+              attributes = fullMessage.attributes;
+            } catch (blobErr) {
+              console.error('Failed to fetch worker message:', blobErr);
+              throw new Error(
+                `Failed to fetch worker message: ${payload._overflowRef}`
+              );
+            }
+          }
+
           if (!attributes || !attributes.testId || !attributes.workerId) {
             throw new Error('AQS message with no testId or workerId');
           }
@@ -507,6 +543,10 @@ class PlatformAzureACI {
         name: 'AZURE_STORAGE_ACCOUNT',
         value: this.storageAccount
       },
+      {
+        name: 'AZURE_STORAGE_BLOB_CONTAINER',
+        value: this.blobContainerName
+      },
       {
         name: 'AZURE_SUBSCRIPTION_ID',
         secureValue: this.azureSubscriptionId