modified endpoint

Author: tanmaya-panda1

plugins/out_azure_kusto/azure_kusto.c

@@ -943,6 +943,54 @@ static int cb_azure_kusto_init(struct flb_output_instance *ins, struct flb_confi
         return -1;
     }
 
+    /*
+     * Create upstream context for Kusto Cluster endpoint (for streaming ingestion)
+     * Convert ingestion endpoint to cluster endpoint by removing "ingest-" prefix
+     */
+    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 */
+            cluster_endpoint = flb_sds_create(ctx->ingestion_endpoint);
+            if (!cluster_endpoint) {
+                flb_plg_error(ctx->ins, "failed to create cluster endpoint string");
+                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);
+            }
+            
+            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);
+                return -1;
+            }
+            
+            flb_sds_destroy(cluster_endpoint);
+        } else {
+            flb_plg_warn(ctx->ins, "ingestion endpoint does not contain '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");
+                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));
 
     /* Create oauth2 context */
@@ -1529,6 +1577,11 @@ static int cb_azure_kusto_exit(void *data, struct flb_config *config)
         ctx->u = NULL;
     }
 
+    if (ctx->u_cluster) {
+        flb_upstream_destroy(ctx->u_cluster);
+        ctx->u_cluster = NULL;
+    }
+
     pthread_mutex_destroy(&ctx->resources_mutex);
     pthread_mutex_destroy(&ctx->token_mutex);
     pthread_mutex_destroy(&ctx->blob_mutex);

plugins/out_azure_kusto/azure_kusto.h

@@ -170,6 +170,9 @@ struct flb_azure_kusto {
     /* Upstream connection to the backend server */
     struct flb_upstream *u;
 
+    /* Upstream connection to the main Kusto cluster for streaming ingestion */
+    struct flb_upstream *u_cluster;
+
     struct flb_upstream *imds_upstream;
 
     /* Fluent Bit context */

plugins/out_azure_kusto/azure_kusto_ingest.c

@@ -544,12 +544,16 @@ int azure_kusto_streaming_ingestion(struct flb_azure_kusto *ctx, flb_sds_t tag,
     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 engine (not ingestion endpoint) */
-    flb_plg_debug(ctx->ins, "[STREAMING_INGESTION] Getting upstream connection to endpoint: %s", 
-                  ctx->ingestion_endpoint ? ctx->ingestion_endpoint : "NULL");
-    u_conn = flb_upstream_conn_get(ctx->u);
+    /* Get upstream connection to the main Kusto cluster endpoint (for streaming ingestion) */
+    flb_plg_debug(ctx->ins, "[STREAMING_INGESTION] Getting upstream connection to cluster endpoint");
+    if (!ctx->u_cluster) {
+        flb_plg_error(ctx->ins, "[STREAMING_INGESTION] ERROR: Cluster upstream not available - streaming ingestion requires cluster endpoint");
+        return -1;
+    }
+    
+    u_conn = flb_upstream_conn_get(ctx->u_cluster);
     if (!u_conn) {
-        flb_plg_error(ctx->ins, "[STREAMING_INGESTION] ERROR: Failed to get upstream connection");
+        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");