out_stackdriver: fix metric accounting for dropped records and decouple logic

This patch refactors the response handling in the stackdriver output plugin to resolve incorrect metric accounting and inconsistent behavior between builds with and without Prometheus metrics enabled.

1. Defer Metric Accounting to Core Engine: Removed manual increment of cmt_dropped_records on standard 4xx errors. The core engine automatically increments drop counters when the plugin returns FLB_ERROR. Manual increments caused double counting. This aligns with how other plugins (like out_cloudwatch_logs) defer core accounting to the engine.

2. Decouple Core Logic from Metrics: Moved parse_partial_success_response and partial success return code handling outside of #ifdef FLB_HAVE_METRICS. This ensures that partial successes always return FLB_OK (clearing the chunk from queue) regardless of whether metrics telemetry is enabled, preventing unnecessary retries and ensuring consistent behavior across different build configurations.

3. Added Visibility Logging: Added a warning log for partial successes that reports the exact number of failed records, providing operator visibility into silent drops.

4. Style Guide Compliance: Ensured all variables are declared at the beginning of the function scope as required by Fluent Bit style guidelines.

Signed-off-by: pdewilde <pdewilde@google.com>

Author: pdewilde

plugins/out_stackdriver/stackdriver.c

@@ -2937,6 +2937,7 @@ static void cb_stackdriver_flush(struct flb_event_chunk *event_chunk,
     (void) config;
     int ret;
     int code;
+    int failed_records = 0;
     int ret_partial_success;
     int ret_code = FLB_RETRY;
     int formatted_records = 0;
@@ -3069,55 +3070,60 @@ static void cb_stackdriver_flush(struct flb_event_chunk *event_chunk,
             ret_code = FLB_OK;
         }
         else {
-#ifdef FLB_HAVE_METRICS
           /* check partial success */
           ret_partial_success =
                 parse_partial_success_response(c,
                                                ctx,
                                                ts,
                                                grpc_status_counts);
 
-            int failed_records = 0;
-            if (ret_partial_success == 0) {
-              for (code = 0; code < GRPC_STATUS_CODES_SIZE; code++) {
-                if (grpc_status_counts[code] != 0) {
-                  failed_records += grpc_status_counts[code];
-                }
-              }
-              cmt_counter_add(ctx->ins->cmt_dropped_records, ts,
-                              failed_records, 1, (char* []) {name});
-              int successful_records =
-                  formatted_records - failed_records;
-              if (successful_records != 0) {
-                add_record_metrics(ctx, ts, successful_records, 200, 0);
+          if (ret_partial_success == 0) {
+            ret_code = FLB_OK;
+
+            for (code = 0; code < GRPC_STATUS_CODES_SIZE; code++) {
+              if (grpc_status_counts[code] != 0) {
+                failed_records += grpc_status_counts[code];
               }
             }
-            else {
-              add_record_metrics(ctx, ts, formatted_records,
-                                 c->resp.status, -1);
-            }
-#endif
-          if (c->resp.status >= 400 && c->resp.status < 500) {
-            ret_code = FLB_ERROR;
+
+            flb_plg_warn(ctx->ins, "partial success: %d of %d records dropped",
+                         failed_records, formatted_records);
+
 #ifdef FLB_HAVE_METRICS
             cmt_counter_add(ctx->ins->cmt_dropped_records, ts,
-                            formatted_records, 1,
-                            (char* []) {name});
+                            failed_records, 1, (char* []) {name});
+            int successful_records =
+                formatted_records - failed_records;
+            if (successful_records != 0) {
+              add_record_metrics(ctx, ts, successful_records, 200, 0);
+            }
 #endif
-            flb_plg_warn(ctx->ins, "tag=%s error sending to Cloud Logging: %s", event_chunk->tag,
-                         c->resp.payload);
           }
+#ifdef FLB_HAVE_METRICS
           else {
-            if (c->resp.payload_size > 0) {
-              /* we got an error */
+            add_record_metrics(ctx, ts, formatted_records,
+                               c->resp.status, -1);
+          }
+#endif
+          /* Partial success is set FLB_OK, with manual update of failure metrics */
+          if (ret_partial_success != 0) {
+            if (c->resp.status >= 400 && c->resp.status < 500) {
+              ret_code = FLB_ERROR;
               flb_plg_warn(ctx->ins, "tag=%s error sending to Cloud Logging: %s", event_chunk->tag,
                            c->resp.payload);
             }
             else {
-              flb_plg_debug(ctx->ins, "tag=%s response from Cloud Logging: %s", event_chunk->tag,
-                            c->resp.payload);
+              if (c->resp.payload_size > 0) {
+                /* we got an error */
+                flb_plg_warn(ctx->ins, "tag=%s error sending to Cloud Logging: %s", event_chunk->tag,
+                             c->resp.payload);
+              }
+              else {
+                flb_plg_debug(ctx->ins, "tag=%s response from Cloud Logging: %s", event_chunk->tag,
+                              c->resp.payload);
+              }
+              ret_code = FLB_RETRY;
             }
-            ret_code = FLB_RETRY;
           }
         }
     }