in_kubernetes_events: buffer incomplete JSON across HTTP chunks

Fixes #11252

Problem:
The kubernetes_events plugin was failing with 'bad formed JSON' errors
when HTTP chunked transfer encoding split JSON event objects across
chunk boundaries.

Root Cause Analysis:
- Kubernetes watch API sends newline-delimited JSON over HTTP/1.1
  chunked transfer encoding
- HTTP chunks are arbitrary-sized (commonly 1000-4000 bytes)
- A single JSON object can be split mid-object across chunks:

  Chunk 1 (1000 bytes): {"type":"ADDED","object":{"name":"po
  Chunk 2 (176 bytes):  d-123","spec":{...}}}

- The HTTP client correctly decodes chunks and returns
  FLB_HTTP_CHUNK_AVAILABLE after each chunk
- However, the plugin was trying to parse incomplete JSON from
  each chunk, causing parse errors

Solution:
Implement application-layer buffering in the kubernetes_events plugin:

1. Added chunk_buffer field to k8s_events context to buffer incomplete
   data across HTTP chunks

2. Modified process_http_chunk() to:
   - Prepend any buffered data from previous chunks
   - Parse only complete JSON objects (delimited by newlines)
   - Buffer remaining incomplete data for the next chunk
   - This follows standard network programming practice: the
     application layer handles message boundaries

3. Clear buffer when stream closes or on connection errors

Why not fix in HTTP client:
- HTTP client returns FLB_HTTP_CHUNK_AVAILABLE by design after each
  chunk is decoded
- HTTP layer shouldn't know about JSON or application protocols
- Similar to how TCP delivers packets but HTTP buffers for messages
- This maintains separation of concerns and doesn't break other plugins

Testing:
- Re-enabled events_v1_with_chunkedrecv test which simulates exactly
  this scenario (1176-byte JSON split into 1000+176 byte chunks)

Signed-off-by: Jesse Awan <[email protected]>

Author: I501307

plugins/in_kubernetes_events/kubernetes_events.c

@@ -746,32 +746,95 @@ static int process_http_chunk(struct k8s_events* ctx, struct flb_http_client *c,
     size_t token_size = 0;
     char *token_start = 0;
     char *token_end = NULL;
+    char *search_start;
+    size_t remaining;
+    flb_sds_t working_buffer = NULL;
+
+    /* 
+     * Prepend any buffered incomplete data from previous chunks.
+     * HTTP chunked encoding can split JSON objects across chunk boundaries,
+     * so we need to buffer incomplete data until we find a complete JSON line.
+     */
+    if (ctx->chunk_buffer != NULL) {
+        working_buffer = flb_sds_cat(ctx->chunk_buffer, c->resp.payload, c->resp.payload_size);
+        if (!working_buffer) {
+            flb_plg_error(ctx->ins, "failed to concatenate chunk buffer");
+            return -1;
+        }
+        /* Clear the old buffer since we've concatenated it */
+        ctx->chunk_buffer = NULL;
+        token_start = working_buffer;
+    }
+    else {
+        token_start = c->resp.payload;
+    }
 
-    token_start = c->resp.payload;
-    token_end = strpbrk(token_start, JSON_ARRAY_DELIM);
-    while ( token_end != NULL && ret == 0 ) {
+    search_start = token_start;
+    token_end = strpbrk(search_start, JSON_ARRAY_DELIM);
+    
+    while (token_end != NULL && ret == 0) {
         token_size = token_end - token_start;
+
+        /* Skip empty lines */
+        if (token_size == 0) {
+            token_start = token_end + 1;
+            search_start = token_start;
+            token_end = strpbrk(search_start, JSON_ARRAY_DELIM);
+            continue;
+        }
+
         ret = flb_pack_json(token_start, token_size, &buf_data, &buf_size, &root_type, &consumed);
         if (ret == -1) {
-            flb_plg_debug(ctx->ins, "could not process payload, incomplete or bad formed JSON: %s",
-                          c->resp.payload);
+            flb_plg_debug(ctx->ins, "could not process payload, incomplete or bad formed JSON");
         }
         else {
-            *bytes_consumed += token_size + 1;
+            if (working_buffer) {
+                *bytes_consumed += token_size + 1 - flb_sds_len(ctx->chunk_buffer ? ctx->chunk_buffer : "");
+            }
+            else {
+                *bytes_consumed += token_size + 1;
+            }
             ret = process_watched_event(ctx, buf_data, buf_size);
         }
 
         flb_free(buf_data);
-        if (buf_data) {
-            buf_data = NULL;
+        buf_data = NULL;
+        
+        token_start = token_end + 1;
+        search_start = token_start;
+        token_end = strpbrk(search_start, JSON_ARRAY_DELIM);
+    }
+
+    /* 
+     * If there's remaining data without a newline delimiter, it means the JSON
+     * object is incomplete (split across chunk boundaries). Buffer it for next chunk.
+     */
+    if (working_buffer) {
+        remaining = flb_sds_len(working_buffer) - (token_start - working_buffer);
+    }
+    else {
+        remaining = c->resp.payload_size - (token_start - c->resp.payload);
+    }
+
+    if (remaining > 0 && ret == 0) {
+        ctx->chunk_buffer = flb_sds_create_len(token_start, remaining);
+        if (!ctx->chunk_buffer) {
+            flb_plg_error(ctx->ins, "failed to create chunk buffer");
+            ret = -1;
+        }
+        else {
+            flb_plg_trace(ctx->ins, "buffering %zu bytes of incomplete JSON data", remaining);
         }
-        token_start = token_end+1;
-        token_end = strpbrk(token_start, JSON_ARRAY_DELIM);
     }
 
+    if (working_buffer) {
+        flb_sds_destroy(working_buffer);
+    }
+    
     if (buf_data) {
         flb_free(buf_data);
     }
+    
     return ret;
 }
 
@@ -889,6 +952,13 @@ static int check_and_init_stream(struct k8s_events *ctx)
         flb_upstream_conn_release(ctx->current_connection);
         ctx->current_connection = NULL;
     }
+    
+    /* Clear any buffered incomplete data on failure */
+    if (ctx->chunk_buffer) {
+        flb_sds_destroy(ctx->chunk_buffer);
+        ctx->chunk_buffer = NULL;
+    }
+    
     return FLB_FALSE;
 }
 
@@ -938,6 +1008,12 @@ static int k8s_events_collect(struct flb_input_instance *ins,
         flb_upstream_conn_release(ctx->current_connection);
         ctx->streaming_client = NULL;
         ctx->current_connection = NULL;
+        
+        /* Clear any buffered incomplete data when stream closes */
+        if (ctx->chunk_buffer) {
+            flb_sds_destroy(ctx->chunk_buffer);
+            ctx->chunk_buffer = NULL;
+        }
     }
 
     pthread_mutex_unlock(&ctx->lock);

plugins/in_kubernetes_events/kubernetes_events.h

@@ -85,6 +85,9 @@ struct k8s_events {
     struct flb_connection *current_connection;
     struct flb_http_client *streaming_client;
 
+    /* Buffer for incomplete JSON data from chunked responses */
+    flb_sds_t chunk_buffer;
+
     /* limit for event queries */
     int limit_request;
     /* last highest seen resource_version */

plugins/in_kubernetes_events/kubernetes_events_conf.c

@@ -158,6 +158,9 @@ struct k8s_events *k8s_events_conf_create(struct flb_input_instance *ins)
     pthread_mutexattr_init(&attr);
     pthread_mutex_init(&ctx->lock, &attr);
 
+    /* Initialize buffer for incomplete chunk data */
+    ctx->chunk_buffer = NULL;
+
     /* Load the config map */
     ret = flb_input_config_map_set(ins, (void *) ctx);
     if (ret == -1) {
@@ -289,6 +292,10 @@ void k8s_events_conf_destroy(struct k8s_events *ctx)
         flb_ra_destroy(ctx->ra_resource_version);
     }
 
+    if (ctx->chunk_buffer) {
+        flb_sds_destroy(ctx->chunk_buffer);
+    }
+
     if(ctx->streaming_client) {
         flb_http_client_destroy(ctx->streaming_client);
     }

tests/runtime/in_kubernetes_events.c

@@ -447,7 +447,7 @@ void flb_test_events_with_chunkedrecv()
 TEST_LIST = {
     {"events_v1_with_lastTimestamp", flb_test_events_v1_with_lastTimestamp},
     {"events_v1_with_creationTimestamp", flb_test_events_v1_with_creationTimestamp},
-    //{"events_v1_with_chunkedrecv", flb_test_events_with_chunkedrecv},
+    {"events_v1_with_chunkedrecv", flb_test_events_with_chunkedrecv},
     {NULL, NULL}
 };