in_prometheus_remote_write: defer worker ingestion

Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>

Author: edsiper

plugins/in_prometheus_remote_write/prom_rw.h

@@ -52,10 +52,12 @@ static inline int prom_rw_uses_worker_ingress_queue(
 static inline int prom_rw_ingest_metrics(struct flb_prom_remote_write *ctx,
                                          const char *tag,
                                          size_t tag_len,
-                                         struct cmt *cmt)
+                                         struct cmt *cmt,
+                                         size_t payload_size)
 {
     if (prom_rw_uses_worker_ingress_queue(ctx)) {
-        return flb_input_ingress_queue_metrics(ctx->ins, tag, tag_len, cmt);
+        return flb_input_ingress_queue_metrics(ctx->ins, tag, tag_len,
+                                               cmt, payload_size);
     }
 
     return flb_input_metrics_append(ctx->ins, tag, tag_len, cmt);

plugins/in_prometheus_remote_write/prom_rw_prot.c

@@ -74,14 +74,10 @@ static int process_payload_metrics_ng(struct flb_prom_remote_write *ctx,
         return 400;
     }
 
-    result = prom_rw_ingest_metrics(ctx, NULL, 0, context);
+    result = prom_rw_ingest_metrics(ctx, NULL, 0, context,
+                                    cfl_sds_len(request->body));
 
-    if (prom_rw_uses_worker_ingress_queue(ctx)) {
-        if (result != 0 && result != FLB_INPUT_INGRESS_BUSY) {
-            cmt_decode_prometheus_remote_write_destroy(context);
-        }
-    }
-    else {
+    if (!prom_rw_uses_worker_ingress_queue(ctx)) {
         cmt_decode_prometheus_remote_write_destroy(context);
     }