refactor: add 3 chunks test, modify buffer logic

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

Author: I501307

plugins/in_kubernetes_events/kubernetes_events.c

@@ -759,9 +759,14 @@ static int process_http_chunk(struct k8s_events* ctx, struct flb_http_client *c,
         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");
+            flb_sds_destroy(ctx->chunk_buffer);
+            ctx->chunk_buffer = NULL;
             return -1;
         }
-        /* Clear the old buffer since we've concatenated it */
+        /* 
+         * flb_sds_cat modifies and returns the first argument, so working_buffer
+         * IS ctx->chunk_buffer (reallocated). Clear our reference to it.
+         */
         ctx->chunk_buffer = NULL;
         token_start = working_buffer;
     }
@@ -788,10 +793,11 @@ static int process_http_chunk(struct k8s_events* ctx, struct flb_http_client *c,
             flb_plg_debug(ctx->ins, "could not process payload, incomplete or bad formed JSON");
         }
         else {
-            if (working_buffer) {
-                *bytes_consumed += token_size + 1 - flb_sds_len(ctx->chunk_buffer ? ctx->chunk_buffer : "");
-            }
-            else {
+            /* 
+             * For non-buffered data, track consumed bytes.
+             * For buffered data, we'll mark everything consumed after the loop.
+             */
+            if (!working_buffer) {
                 *bytes_consumed += token_size + 1;
             }
             ret = process_watched_event(ctx, buf_data, buf_size);
@@ -804,6 +810,12 @@ static int process_http_chunk(struct k8s_events* ctx, struct flb_http_client *c,
         search_start = token_start;
         token_end = strpbrk(search_start, JSON_ARRAY_DELIM);
     }
+    
+    /* 
+     * Always consume all bytes from the current chunk since we've examined them all.
+     * Even if we buffer the data, we've still "consumed" it from the HTTP payload.
+     */
+    *bytes_consumed = c->resp.payload_size;
 
     /* 
      * If there's remaining data without a newline delimiter, it means the JSON
@@ -997,6 +1009,28 @@ static int k8s_events_collect(struct flb_input_instance *ins,
         }
         else if (ret == FLB_HTTP_OK) {
             flb_plg_info(ins, "kubernetes stream closed by api server. Reconnect will happen on next interval.");
+            
+            /* 
+             * If there's buffered data when stream closes, try to process it.
+             * This handles the case where the last chunk doesn't end with a newline.
+             */
+            if (ctx->chunk_buffer && flb_sds_len(ctx->chunk_buffer) > 0) {
+                int buf_ret;
+                int root_type;
+                size_t consumed = 0;
+                char *buf_data = NULL;
+                size_t buf_size;
+                
+                buf_ret = flb_pack_json(ctx->chunk_buffer, flb_sds_len(ctx->chunk_buffer), 
+                                       &buf_data, &buf_size, &root_type, &consumed);
+                if (buf_ret == 0) {
+                    process_watched_event(ctx, buf_data, buf_size);
+                }
+                
+                if (buf_data) {
+                    flb_free(buf_data);
+                }
+            }
         }
         else {
             flb_plg_warn(ins, "events watch failure, http_status=%d payload=%s",

tests/runtime/in_kubernetes_events.c

@@ -444,10 +444,56 @@ void flb_test_events_with_chunkedrecv()
     test_ctx_destroy(ctx);
 }
 
+/* Test with smaller chunks - splits single event into 3 chunks */
+void flb_test_events_with_3chunks()
+{
+    struct flb_lib_out_cb cb_data;
+    struct test_ctx *ctx;
+    int trys;
+
+    int ret;
+    int num;
+    const char *filename = "eventlist_v1_with_lastTimestamp";
+    const char *stream_filename = "watch_v1_with_lastTimestamp";
+
+    clear_output_num();
+
+    /* Use 400 byte chunks to split 1176-byte JSON into 3 chunks */
+    struct test_k8s_server_ctx* k8s_server = initialize_mock_k8s_api(
+        filename, stream_filename, 400
+    );
+
+    cb_data.cb = cb_check_result_json;
+    cb_data.data = (void *)k8s_server;
+
+    ctx = test_ctx_create(&cb_data);
+    if (!TEST_CHECK(ctx != NULL)) {
+        TEST_MSG("test_ctx_create failed");
+        exit(EXIT_FAILURE);
+    }
+
+    ret = flb_start(ctx->flb);
+    TEST_CHECK(ret == 0);
+
+    /* waiting to flush */
+    for (trys = 0; trys < 5 && get_output_num() <= 1; trys++) {
+        flb_time_msleep(1000);
+    }
+
+    num = get_output_num();
+    if (!TEST_CHECK(num >= 2))  {
+        TEST_MSG("2 output records are expected found %d", num);
+    }
+
+    mock_k8s_api_destroy(k8s_server);
+    test_ctx_destroy(ctx);
+}
+
 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_3chunks", flb_test_events_with_3chunks},
     {NULL, NULL}
 };