out_stackdriver: fix batch drop on invalid labels

plugins/out_stackdriver/stackdriver.c

@@ -1805,6 +1805,67 @@ static int pack_payload(int insert_id_extracted,
         return ret;
 }
 
+/*
+ * should_skip_record
+ * Check if a record should be skipped due to invalid fields.
+ * Returns FLB_TRUE if the record should be dropped.
+ *
+ * log_errors: if FLB_TRUE, log why the record is being dropped.
+ *             Set to FLB_FALSE during prescan to avoid duplicate logging.
+ *
+ * The insertId and payload labels are validated by traversing the record;
+ * the results are returned through in_status_out, insert_id_obj_out and
+ * payload_labels_ptr_out (any of which may be NULL) so the caller can reuse
+ * them instead of traversing the record again. They are only written when the
+ * record is not skipped.
+ */
+static int should_skip_record(struct flb_stackdriver *ctx,
+                              msgpack_object *obj,
+                              int log_errors,
+                              insert_id_status *in_status_out,
+                              msgpack_object *insert_id_obj_out,
+                              msgpack_object **payload_labels_ptr_out)
+{
+    insert_id_status in_status;
+    msgpack_object insert_id_obj;
+    msgpack_object *payload_labels_ptr;
+
+    /* Check insertId */
+    in_status = validate_insert_id(&insert_id_obj, obj);
+    if (in_status == INSERTID_INVALID) {
+        if (log_errors == FLB_TRUE) {
+            flb_plg_error(ctx->ins,
+                          "Incorrect insertId received. "
+                          "InsertId should be non-empty string.");
+        }
+        return FLB_TRUE;
+    }
+
+    /* Check labels type */
+    payload_labels_ptr = get_payload_labels(ctx, obj);
+    if (payload_labels_ptr != NULL &&
+        payload_labels_ptr->type != MSGPACK_OBJECT_MAP) {
+        if (log_errors == FLB_TRUE) {
+            flb_plg_error(ctx->ins,
+                          "the type of payload labels should be map, "
+                          "dropping record");
+        }
+        return FLB_TRUE;
+    }
+
+    if (in_status_out != NULL) {
+        *in_status_out = in_status;
+    }
+    if (insert_id_obj_out != NULL) {
+        *insert_id_obj_out = insert_id_obj;
+    }
+    if (payload_labels_ptr_out != NULL) {
+        *payload_labels_ptr_out = payload_labels_ptr;
+    }
+
+    return FLB_FALSE;
+}
+
 static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx,
                                     int total_records,
                                     const char *tag, int tag_len,
@@ -1889,17 +1950,17 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx,
     /* Count number of records */
     array_size = total_records;
 
-    if (formatted_records != NULL) {
-        *formatted_records = 0;
-    }
-
     /* Parameters for labels */
     msgpack_object *payload_labels_ptr;
     int labels_size = 0;
 
     struct flb_log_event_decoder log_decoder;
     struct flb_log_event log_event;
 
+    if (formatted_records != NULL) {
+        *formatted_records = -1;
+    }
+
     ret = flb_log_event_decoder_init(&log_decoder, (char *) data, bytes);
 
     if (ret != FLB_EVENT_DECODER_SUCCESS) {
@@ -1910,20 +1971,15 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx,
     }
 
     /*
-     * Search each entry and validate insertId.
-     * Reject the entry if insertId is invalid.
+     * Search each entry and validate record fields.
+     * Reject entries with invalid insertId or non-map labels.
      * If all the entries are rejected, stop formatting.
-     *
      */
     while ((ret = flb_log_event_decoder_next(
                     &log_decoder,
                     &log_event)) == FLB_EVENT_DECODER_SUCCESS) {
-        /* Extract insertId */
-        in_status = validate_insert_id(&insert_id_obj, log_event.body);
-
-        if (in_status == INSERTID_INVALID) {
-            flb_plg_error(ctx->ins,
-                          "Incorrect insertId received. InsertId should be non-empty string.");
+        if (should_skip_record(ctx, log_event.body, FLB_FALSE,
+                               NULL, NULL, NULL) == FLB_TRUE) {
             array_size -= 1;
         }
     }
@@ -1932,13 +1988,15 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx,
 
     /* Sounds like this should compare to -1 instead of zero */
     if (array_size == 0) {
+        flb_plg_warn(ctx->ins,
+                     "all %d entries skipped due to invalid insertId "
+                     "or labels, dropping batch", total_records);
+        if (formatted_records != NULL) {
+            *formatted_records = 0;
+        }
         return NULL;
     }
 
-    if (formatted_records != NULL) {
-        *formatted_records = array_size;
-    }
-
     /* Create temporal msgpack buffer */
     msgpack_sbuffer_init(&mp_sbuf);
     msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write);
@@ -2354,6 +2412,17 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx,
                     &log_decoder,
                     &log_event)) == FLB_EVENT_DECODER_SUCCESS) {
         obj = log_event.body;
+
+        /*
+         * Skip records with invalid fields. The insertId and payload labels
+         * validated here are returned so they are not re-extracted below.
+         */
+        if (should_skip_record(ctx, obj, FLB_TRUE,
+                               &in_status, &insert_id_obj,
+                               &payload_labels_ptr) == FLB_TRUE) {
+            continue;
+        }
+
         tms_status = extract_timestamp(obj, &log_event.timestamp);
 
         /*
@@ -2418,33 +2487,16 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx,
             log_name_extracted = FLB_TRUE;
         }
 
-        /* Extract insertId */
-        in_status = validate_insert_id(&insert_id_obj, obj);
+        /*
+         * insertId was already validated and extracted by should_skip_record
+         * (the INVALID case is handled there); reuse that result.
+         */
         if (in_status == INSERTID_VALID) {
             insert_id_extracted = FLB_TRUE;
             entry_size += 1;
         }
-        else if (in_status == INSERTID_NOT_PRESENT) {
-            insert_id_extracted = FLB_FALSE;
-        }
         else {
-            if (trace_extracted == FLB_TRUE) {
-                flb_sds_destroy(trace);
-            }
-
-            if (span_id_extracted == FLB_TRUE) {
-                flb_sds_destroy(span_id);
-            }
-
-            if (project_id_extracted == FLB_TRUE) {
-                flb_sds_destroy(project_id_key);
-            }
-
-            if (log_name_extracted == FLB_TRUE) {
-                flb_sds_destroy(log_name);
-            }
-
-            continue;
+            insert_id_extracted = FLB_FALSE;
         }
 
         /* Extract operation */
@@ -2487,39 +2539,10 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx,
             entry_size += 1;
         }
 
-        /* Extract payload labels */
-        payload_labels_ptr = get_payload_labels(ctx, obj);
-        if (payload_labels_ptr != NULL &&
-            payload_labels_ptr->type != MSGPACK_OBJECT_MAP) {
-            flb_plg_error(ctx->ins, "the type of payload labels should be map");
-            flb_sds_destroy(operation_id);
-            flb_sds_destroy(operation_producer);
-            flb_sds_destroy(source_location_file);
-            flb_sds_destroy(source_location_function);
-
-            if (trace_extracted == FLB_TRUE) {
-                flb_sds_destroy(trace);
-            }
-
-            if (span_id_extracted == FLB_TRUE) {
-                flb_sds_destroy(span_id);
-            }
-
-            if (project_id_extracted == FLB_TRUE) {
-                flb_sds_destroy(project_id_key);
-            }
-
-            if (log_name_extracted == FLB_TRUE) {
-                flb_sds_destroy(log_name);
-            }
-
-            destroy_http_request(&http_request);
-
-            flb_log_event_decoder_destroy(&log_decoder);
-            msgpack_sbuffer_destroy(&mp_sbuf);
-
-            return NULL;
-        }
+        /*
+         * payload labels were already extracted by should_skip_record (the
+         * non-map case is handled there); reuse that result.
+         */
 
         /* Number of parsed labels */
         labels_size = mk_list_size(&ctx->config_labels);
@@ -2708,6 +2731,10 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx,
         return NULL;
     }
 
+    if (formatted_records != NULL) {
+        *formatted_records = array_size;
+    }
+
     return out_buf;
 }
 
@@ -2728,7 +2755,7 @@ static int stackdriver_format_test(struct flb_config *config,
     total_records = flb_mp_count_log_records(data, bytes);
 
     payload = stackdriver_format(ctx, total_records,
-                                (char *) tag, tag_len, data, bytes, config, NULL);
+                                 (char *) tag, tag_len, data, bytes, config, NULL);
     if (payload == NULL) {
         return -1;
     }
@@ -2777,9 +2804,9 @@ static void update_http_metrics(struct flb_stackdriver* ctx,
 }
 
 static void update_retry_metric(struct flb_stackdriver *ctx,
-                                 int retried_records,
-                                 uint64_t ts,
-                                 int http_status)
+                                int retried_records,
+                                uint64_t ts,
+                                int http_status)
 {
     char tmp[32];
     char *name = (char *) flb_output_name(ctx->ins);
@@ -2978,7 +3005,10 @@ static void cb_stackdriver_flush(struct flb_event_chunk *event_chunk,
     int code;
     int ret_partial_success;
     int ret_code = FLB_RETRY;
-    int formatted_records = 0;
+    int formatted_records = -1;
+    int skipped_records = 0;
+    int failed_records = 0;
+    int successful_records = 0;
     int grpc_status_counts[GRPC_STATUS_CODES_SIZE] = {0};
     size_t b_sent;
     flb_sds_t token;
@@ -3005,6 +3035,15 @@ static void cb_stackdriver_flush(struct flb_event_chunk *event_chunk,
                                      config,
                                      &formatted_records);
     if (!payload_buf) {
+        if (formatted_records == 0) {
+#ifdef FLB_HAVE_METRICS
+            cmt_counter_add(ctx->ins->cmt_dropped_records, ts,
+                            (int) event_chunk->total_events,
+                            1, (char *[]) {name});
+#endif
+            FLB_OUTPUT_RETURN(FLB_OK);
+        }
+
 #ifdef FLB_HAVE_METRICS
         cmt_counter_inc(ctx->cmt_failed_requests,
                         ts, 1, (char *[]) {name});
@@ -3015,6 +3054,11 @@ static void cb_stackdriver_flush(struct flb_event_chunk *event_chunk,
         FLB_OUTPUT_RETURN(FLB_RETRY);
     }
 
+    skipped_records = (int) event_chunk->total_events - formatted_records;
+    if (skipped_records < 0) {
+        skipped_records = 0;
+    }
+
     if (ctx->test_log_entry_format) {
         printf("%s\n", payload_buf);
         flb_sds_destroy(payload_buf);
@@ -3116,7 +3160,6 @@ static void cb_stackdriver_flush(struct flb_event_chunk *event_chunk,
                                                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) {
@@ -3125,8 +3168,10 @@ static void cb_stackdriver_flush(struct flb_event_chunk *event_chunk,
               }
               cmt_counter_add(ctx->ins->cmt_dropped_records, ts,
                               failed_records, 1, (char* []) {name});
-              int successful_records =
-                  formatted_records - failed_records;
+              successful_records = formatted_records - failed_records;
+              if (successful_records < 0) {
+                successful_records = 0;
+              }
               if (successful_records != 0) {
                 add_record_metrics(ctx, ts, successful_records, 200, 0);
               }
@@ -3178,6 +3223,11 @@ static void cb_stackdriver_flush(struct flb_event_chunk *event_chunk,
         flb_metrics_sum(FLB_STACKDRIVER_FAILED_REQUESTS, 1, ctx->ins->metrics);
     }
 
+    if (ret_code != FLB_RETRY && skipped_records > 0) {
+        cmt_counter_add(ctx->ins->cmt_dropped_records, ts,
+                        skipped_records, 1, (char *[]) {name});
+    }
+
     if (ret_code == FLB_RETRY) {
         update_retry_metric(ctx, formatted_records, ts, c->resp.status);
     }

tests/runtime/data/stackdriver/stackdriver_batch_all_labels_not_a_map.log

@@ -0,0 +1,2 @@
+{"message":"bad record 1", "logging.googleapis.com/labels": "not_a_map"}
+{"message":"bad record 2", "logging.googleapis.com/labels": 12345}

tests/runtime/data/stackdriver/stackdriver_batch_first_record_labels_not_a_map.log

@@ -0,0 +1,2 @@
+{"message":"bad first record", "logging.googleapis.com/labels": "not_a_map"}
+{"message":"valid second record"}

tests/runtime/data/stackdriver/stackdriver_batch_labels_not_a_map.log

@@ -0,0 +1,3 @@
+{"message":"valid record with labels", "logging.googleapis.com/labels": {"testA": "valA", "testB": "valB"}}
+{"message":"bad labels - not a map", "logging.googleapis.com/labels": "not_a_map"}
+{"message":"valid record without labels"}

tests/runtime/data/stackdriver/stackdriver_batch_mixed_errors.log

@@ -0,0 +1,4 @@
+{"message":"valid record no special fields"}
+{"message":"bad insertId", "logging.googleapis.com/insertId": 123}
+{"message":"bad labels", "logging.googleapis.com/labels": "not_a_map"}
+{"message":"valid record with labels", "logging.googleapis.com/labels": {"testA": "valA"}}