azure_kusto:fixed code comments

tanmaya-panda1 · tanmaya-panda1 · commit 4fd3fd4cbd92 · 2025-10-28T12:00:01.000+05:30
Signed-off-by: Tanmaya Panda &lt;tanmayapanda@microsoft.com&gt;
diff --git a/plugins/out_azure_kusto/azure_kusto.c b/plugins/out_azure_kusto/azure_kusto.c
@@ -912,7 +912,7 @@ static int cb_azure_kusto_init(struct flb_output_instance *ins, struct flb_confi
 
         ctx->timer_created = FLB_FALSE;
         ctx->timer_ms = (int) (ctx->upload_timeout / 6) * 1000;
-        flb_plg_info(ctx->ins, "Using upload size %lu bytes", ctx->file_size);
+        flb_plg_debug(ctx->ins, "Using upload size %lu bytes", ctx->file_size);
     }
 
     flb_output_set_context(ins, ctx);
@@ -949,46 +949,55 @@ static int cb_azure_kusto_init(struct flb_output_instance *ins, struct flb_confi
      */
     if (ctx->streaming_ingestion_enabled == FLB_TRUE) {
         flb_sds_t cluster_endpoint = NULL;
-
-        /* Check if ingestion endpoint contains "ingest-" prefix */
-        if (strstr(ctx->ingestion_endpoint, "ingest-") != NULL) {
-            /* Create cluster endpoint by removing "ingest-" prefix */
+        const char *prefix = "ingest-";
+        const char *schema_end = strstr(ctx->ingestion_endpoint, "://");
+        const char *hostname_start = schema_end ? schema_end + 3 : ctx->ingestion_endpoint;
+        
+        /* Check if hostname starts with "ingest-" prefix */
+        if (strncmp(hostname_start, prefix, strlen(prefix)) == 0) {
+            /* Create cluster endpoint by removing "ingest-" prefix from hostname */
             cluster_endpoint = flb_sds_create(ctx->ingestion_endpoint);
             if (!cluster_endpoint) {
                 flb_plg_error(ctx->ins, "failed to create cluster endpoint string");
+                flb_upstream_destroy(ctx->u);
+                ctx->u = NULL;
                 return -1;
             }
 
-            /* Replace "ingest-" with empty string to get cluster endpoint */
-            char *ingest_pos = strstr(cluster_endpoint, "ingest-");
-            if (ingest_pos) {
-                /* Move the rest of the string to remove "ingest-" */
-                memmove(ingest_pos, ingest_pos + 7, strlen(ingest_pos + 7) + 1);
-                flb_sds_len_set(cluster_endpoint, flb_sds_len(cluster_endpoint) - 7);
+            /* Find the position in our copy and remove the prefix */
+            char *copy_hostname = strstr(cluster_endpoint, "://");
+            if (copy_hostname) {
+                copy_hostname += 3;
+                /* Verify the prefix is still at the expected position */
+                if (strncmp(copy_hostname, prefix, strlen(prefix)) == 0) {
+                    memmove(copy_hostname, copy_hostname + strlen(prefix), 
+                            strlen(copy_hostname + strlen(prefix)) + 1);
+                    flb_sds_len_set(cluster_endpoint, flb_sds_len(cluster_endpoint) - strlen(prefix));
+                }
             }
 
-            flb_plg_info(ctx->ins, "Creating cluster upstream connection to: %s", cluster_endpoint);
-
             /* Create upstream connection to cluster endpoint */
             ctx->u_cluster = flb_upstream_create_url(config, cluster_endpoint, io_flags, ins->tls);
             if (!ctx->u_cluster) {
                 flb_plg_error(ctx->ins, "cluster upstream creation failed for endpoint: %s", cluster_endpoint);
                 flb_sds_destroy(cluster_endpoint);
+                flb_upstream_destroy(ctx->u);
+                ctx->u = NULL;
                 return -1;
             }
 
             flb_sds_destroy(cluster_endpoint);
         } else {
-            flb_plg_warn(ctx->ins, "ingestion endpoint does not contain 'ingest-' prefix, using as cluster endpoint");
+            flb_plg_warn(ctx->ins, "ingestion endpoint hostname does not start with 'ingest-' prefix, using as cluster endpoint");
             /* Use ingestion endpoint directly as cluster endpoint */
             ctx->u_cluster = flb_upstream_create_url(config, ctx->ingestion_endpoint, io_flags, ins->tls);
             if (!ctx->u_cluster) {
                 flb_plg_error(ctx->ins, "cluster upstream creation failed");
+                flb_upstream_destroy(ctx->u);
+                ctx->u = NULL;
                 return -1;
             }
         }
-
-        flb_plg_info(ctx->ins, "Cluster upstream connection created successfully for streaming ingestion");
     }
 
     flb_plg_debug(ctx->ins, "async flag is %d", flb_stream_is_async(&ctx->u->base));
@@ -1446,21 +1455,35 @@ static void cb_azure_kusto_flush(struct flb_event_chunk *event_chunk,
 
         /* Check if streaming ingestion is enabled */
         if (ctx->streaming_ingestion_enabled == FLB_TRUE) {
-            flb_plg_info(ctx->ins, "[FLUSH_STREAMING] Streaming ingestion mode enabled for tag: %s", event_chunk->tag);
-
-            /* Check payload size limit for streaming ingestion (4MB) */
-            flb_plg_info(ctx->ins, "[FLUSH_STREAMING] Checking payload size: %zu bytes against 4MB limit", final_payload_size);
-            if (final_payload_size > 4194304) { /* 4MB = 4 * 1024 * 1024 */
-                flb_plg_error(ctx->ins, "[FLUSH_STREAMING] ERROR: Payload size %zu bytes exceeds 4MB limit for streaming ingestion", final_payload_size);
+            flb_plg_debug(ctx->ins, "[FLUSH_STREAMING] Streaming ingestion mode enabled for tag: %s", event_chunk->tag);
+
+            /* 
+             * Azure Kusto streaming ingestion has TWO limits:
+             * 1. Uncompressed data: 4MB limit
+             * 2. Compressed data: 1MB limit
+             * We need to check both limits
+             */
+            
+            /* Check uncompressed data limit (4MB) */
+            if (json_size > 4194304) { /* 4MB = 4 * 1024 * 1024 */
+                flb_plg_error(ctx->ins, "[FLUSH_STREAMING] ERROR: Uncompressed payload size %zu bytes exceeds 4MB limit for streaming ingestion", json_size);
+                ret = FLB_ERROR;
+                goto error;
+            }
+            
+            /* Check compressed data limit (1MB) if compression is enabled */
+            if (ctx->compression_enabled == FLB_TRUE && final_payload_size > 1048576) { /* 1MB = 1024 * 1024 */
+                flb_plg_error(ctx->ins, "[FLUSH_STREAMING] ERROR: Compressed payload size %zu bytes exceeds 1MB limit for streaming ingestion (uncompressed: %zu bytes)", final_payload_size, json_size);
                 ret = FLB_ERROR;
                 goto error;
             }
-            flb_plg_info(ctx->ins, "[FLUSH_STREAMING] Payload size check passed (%zu bytes < 4MB)", final_payload_size);
+            
+            flb_plg_debug(ctx->ins, "[FLUSH_STREAMING] Payload size checks passed - uncompressed: %zu bytes, compressed: %zu bytes", 
+                         json_size, final_payload_size);
 
             /* Perform streaming ingestion to Kusto */
-            flb_plg_info(ctx->ins, "[FLUSH_STREAMING] Initiating streaming ingestion to Kusto");
+            flb_plg_debug(ctx->ins, "[FLUSH_STREAMING] Initiating streaming ingestion to Kusto");
             ret = azure_kusto_streaming_ingestion(ctx, event_chunk->tag, tag_len, final_payload, final_payload_size);
-            flb_plg_info(ctx->ins, "[FLUSH_STREAMING] Streaming ingestion completed with result: %d", ret);
 
             if (ret != 0) {
                 flb_plg_error(ctx->ins, "[FLUSH_STREAMING] ERROR: Streaming ingestion failed, will retry");
diff --git a/plugins/out_azure_kusto/azure_kusto_conf.c b/plugins/out_azure_kusto/azure_kusto_conf.c
@@ -796,11 +796,11 @@ struct flb_azure_kusto *flb_azure_kusto_conf_create(struct flb_output_instance *
         return NULL;
     }
 
-    /* Validate mutual exclusivity between buffering and streaming ingestion */
+    /* Validate mutual exclusivity between buffering and streaming ingestion
+     * Prefer queued ingestion when buffering is explicitly enabled */
     if (ctx->buffering_enabled && ctx->streaming_ingestion_enabled) {
-        flb_plg_error(ctx->ins, "buffering_enabled and streaming_ingestion_enabled cannot both be true. When buffering is enabled, streaming ingestion is automatically disabled");
-        flb_azure_kusto_conf_destroy(ctx);
-        return NULL;
+        ctx->streaming_ingestion_enabled = FLB_FALSE;
+        flb_plg_warn(ctx->ins, "buffering_enabled=true overrides streaming_ingestion_enabled; streaming disabled");
     }
 
     /* Log ingestion mode selection */
diff --git a/plugins/out_azure_kusto/azure_kusto_ingest.c b/plugins/out_azure_kusto/azure_kusto_ingest.c
@@ -530,38 +530,58 @@ int azure_kusto_streaming_ingestion(struct flb_azure_kusto *ctx, flb_sds_t tag,
     struct tm tm;
     char tmp[64];
     int len;
+    size_t size_limit;
 
     flb_plg_info(ctx->ins, "[STREAMING_INGESTION] Starting for tag: %.*s, payload: %zu bytes, db: %s, table: %s, compression: %s",
                  (int)tag_len, tag, payload_size, ctx->database_name, ctx->table_name, ctx->compression_enabled ? "enabled" : "disabled");
 
+    /* 
+     * Size validation (matching ManagedStreamingIngestClient behavior)
+     * Check payload size against limits before attempting streaming ingestion
+     */
+    if (ctx->compression_enabled) {
+        size_limit = KUSTO_STREAMING_MAX_COMPRESSED_SIZE;
+    } else {
+        size_limit = KUSTO_STREAMING_MAX_UNCOMPRESSED_SIZE;
+    }
+
+    if (payload_size > size_limit) {
+        flb_plg_warn(ctx->ins, "[STREAMING_INGESTION] Payload size %zu bytes exceeds %s limit of %zu bytes - falling back to queued ingestion",
+                     payload_size, ctx->compression_enabled ? "compressed" : "uncompressed", size_limit);
+        
+        /* Fallback to queued ingestion */
+        ret = azure_kusto_queued_ingestion(ctx, tag, tag_len, payload, payload_size, NULL);
+        if (ret == 0) {
+            flb_plg_info(ctx->ins, "[STREAMING_INGESTION] Successfully fell back to queued ingestion");
+        } else {
+            flb_plg_error(ctx->ins, "[STREAMING_INGESTION] Fallback to queued ingestion failed");
+        }
+        return ret;
+    }
+
     now = time(NULL);
     gmtime_r(&now, &tm);
     len = strftime(tmp, sizeof(tmp) - 1, "%a, %d %b %Y %H:%M:%S GMT", &tm);
-    flb_plg_debug(ctx->ins, "[STREAMING_INGESTION] Request timestamp: %s", tmp);
 
     /* Get upstream connection to the main Kusto cluster endpoint (for streaming ingestion) */
     if (!ctx->u_cluster) {
         flb_plg_error(ctx->ins, "[STREAMING_INGESTION] ERROR: cluster upstream not available - streaming ingestion requires cluster endpoint");
         return -1;
     }
 
-    flb_plg_debug(ctx->ins, "[STREAMING_INGESTION] Getting upstream connection to cluster endpoint");
     u_conn = flb_upstream_conn_get(ctx->u_cluster);
     if (!u_conn) {
         flb_plg_error(ctx->ins, "[STREAMING_INGESTION] ERROR: failed to get cluster upstream connection");
         return -1;
     }
-    flb_plg_debug(ctx->ins, "[STREAMING_INGESTION] Successfully obtained upstream connection");
 
     /* Get authentication token */
-    flb_plg_debug(ctx->ins, "[STREAMING_INGESTION] Retrieving OAuth2 authentication token");
     token = get_azure_kusto_token(ctx);
     if (!token) {
         flb_plg_error(ctx->ins, "[STREAMING_INGESTION] ERROR: failed to retrieve OAuth2 token");
         flb_upstream_conn_release(u_conn);
         return -1;
     }
-    flb_plg_debug(ctx->ins, "[STREAMING_INGESTION] Successfully obtained OAuth2 token (length: %zu)", flb_sds_len(token));
 
     /* Build the streaming ingestion URI */
     uri = flb_sds_create_size(256);
@@ -577,17 +597,14 @@ int azure_kusto_streaming_ingestion(struct flb_azure_kusto *ctx, flb_sds_t tag,
         flb_sds_snprintf(&uri, flb_sds_alloc(uri),
                          "/v1/rest/ingest/%s/%s?streamFormat=MultiJSON&mappingName=%s",
                          ctx->database_name, ctx->table_name, ctx->ingestion_mapping_reference);
-        flb_plg_debug(ctx->ins, "[STREAMING_INGESTION] Using ingestion mapping: %s", ctx->ingestion_mapping_reference);
     } else {
         flb_sds_snprintf(&uri, flb_sds_alloc(uri),
                          "/v1/rest/ingest/%s/%s?streamFormat=MultiJSON",
                          ctx->database_name, ctx->table_name);
         flb_plg_debug(ctx->ins, "[STREAMING_INGESTION] No ingestion mapping specified");
     }
-    flb_plg_info(ctx->ins, "[STREAMING_INGESTION] Request URI: %s", uri);
 
     /* Create HTTP client for streaming ingestion */
-    flb_plg_debug(ctx->ins, "[STREAMING_INGESTION] Creating HTTP client for POST request");
     c = flb_http_client(u_conn, FLB_HTTP_POST, uri, payload, payload_size,
                         NULL, 0, NULL, 0);
 
@@ -605,15 +622,10 @@ int azure_kusto_streaming_ingestion(struct flb_azure_kusto *ctx, flb_sds_t tag,
         if (ctx->compression_enabled) {
             flb_http_add_header(c, "Content-Type", 12, "application/json", 16);
             flb_http_add_header(c, "Content-Encoding", 16, "gzip", 4);
-            flb_plg_debug(ctx->ins, "[STREAMING_INGESTION] Headers set for compressed payload");
         } else {
             flb_http_add_header(c, "Content-Type", 12, "application/json; charset=utf-8", 31);
-            flb_plg_debug(ctx->ins, "[STREAMING_INGESTION] Headers set for uncompressed payload");
         }
 
-        flb_plg_debug(ctx->ins, "[STREAMING_INGESTION] Payload sample (first 200 chars): %.200s", (char*)payload);
-        flb_plg_info(ctx->ins, "[STREAMING_INGESTION] Sending HTTP POST request to Kusto engine");
-
         /* Send the HTTP request */
         ret = flb_http_do(c, &resp_size);
 
@@ -623,7 +635,6 @@ int azure_kusto_streaming_ingestion(struct flb_azure_kusto *ctx, flb_sds_t tag,
             /* Check for successful HTTP status codes */
             if (c->resp.status == 200 || c->resp.status == 204) {
                 ret = 0;
-                flb_plg_info(ctx->ins, "[STREAMING_INGESTION] SUCCESS: Data successfully ingested to Kusto (HTTP %d)", c->resp.status);
             } else {
                 ret = -1;
                 flb_plg_error(ctx->ins, "[STREAMING_INGESTION] ERROR: HTTP request failed with status: %i", c->resp.status);
@@ -637,14 +648,12 @@ int azure_kusto_streaming_ingestion(struct flb_azure_kusto *ctx, flb_sds_t tag,
             flb_plg_error(ctx->ins, "[STREAMING_INGESTION] ERROR: HTTP request failed at transport level (ret=%d)", ret);
         }
 
-        flb_plg_debug(ctx->ins, "[STREAMING_INGESTION] Destroying HTTP client");
         flb_http_client_destroy(c);
     } else {
         flb_plg_error(ctx->ins, "[STREAMING_INGESTION] ERROR: failed to create HTTP client context");
     }
 
     /* Cleanup */
-    flb_plg_debug(ctx->ins, "[STREAMING_INGESTION] Cleaning up resources");
     if (uri) {
         flb_sds_destroy(uri);
     }
@@ -653,7 +662,6 @@ int azure_kusto_streaming_ingestion(struct flb_azure_kusto *ctx, flb_sds_t tag,
     }
     flb_upstream_conn_release(u_conn);
 
-    flb_plg_info(ctx->ins, "[STREAMING_INGESTION] Streaming ingestion completed with result: %d", ret);
     return ret;
 }
 
diff --git a/plugins/out_azure_kusto/azure_kusto_ingest.h b/plugins/out_azure_kusto/azure_kusto_ingest.h
@@ -23,6 +23,10 @@
 #include "azure_kusto.h"
 #include "azure_kusto_store.h"
 
+/* Kusto Streaming Ingestion Limits (matching ManagedStreamingIngestClient behavior) */
+#define KUSTO_STREAMING_MAX_UNCOMPRESSED_SIZE (4 * 1024 * 1024)  /* 4 MB */
+#define KUSTO_STREAMING_MAX_COMPRESSED_SIZE   (1 * 1024 * 1024)  /* 1 MB */
+
 int azure_kusto_queued_ingestion(struct flb_azure_kusto *ctx, flb_sds_t tag,
                                  size_t tag_len, flb_sds_t payload, size_t payload_size, struct azure_kusto_file *upload_file);
 

Original file line number	Diff line number	Diff line change
`@@ -796,11 +796,11 @@ struct flb_azure_kusto flb_azure_kusto_conf_create(struct flb_output_instance `
`796`	`796`	`return NULL;`
`797`	`797`	`}`
`798`	`798`
`799`		`- /* Validate mutual exclusivity between buffering and streaming ingestion */`
	`799`	`+ /* Validate mutual exclusivity between buffering and streaming ingestion`
	`800`	`+ * Prefer queued ingestion when buffering is explicitly enabled */`
`800`	`801`	`if (ctx->buffering_enabled && ctx->streaming_ingestion_enabled) {`
`801`		`- flb_plg_error(ctx->ins, "buffering_enabled and streaming_ingestion_enabled cannot both be true. When buffering is enabled, streaming ingestion is automatically disabled");`
`802`		`- flb_azure_kusto_conf_destroy(ctx);`
`803`		`- return NULL;`
	`802`	`+ ctx->streaming_ingestion_enabled = FLB_FALSE;`
	`803`	`+ flb_plg_warn(ctx->ins, "buffering_enabled=true overrides streaming_ingestion_enabled; streaming disabled");`
`804`	`804`	`}`
`805`	`805`
`806`	`806`	`/* Log ingestion mode selection */`