prometheus_remote_write: Fix cutoff logic.

A single stale metric can suppress encoding of all following non-stale metrics if the return code CMT_ENCODE_PROMETHEUS_REMOTE_WRITE_CUTOFF_ERROR is not treated as success.
fluent · Oct 1, 2024 · 66ced89 · 66ced89
1 parent 4651386
commit 66ced89
Showing 1 changed file with 15 additions and 6 deletions.
diff --git a/src/cmt_encode_prometheus_remote_write.c b/src/cmt_encode_prometheus_remote_write.c
@@ -663,9 +663,11 @@ int pack_basic_type(struct cmt_prometheus_remote_write_context *context,
     int                result;
     struct cfl_list    *head;
     uint64_t            now;
+    int                 cutoff;
 
     context->sequence_number++;
     add_metadata = CMT_ENCODE_PROMETHEUS_REMOTE_WRITE_ADD_METADATA;
+    cutoff = CMT_FALSE;
 
     now = cfl_time_now();
 
@@ -689,7 +691,8 @@ int pack_basic_type(struct cmt_prometheus_remote_write_context *context,
         if (check_staled_timestamp(metric, now,
                                    CMT_ENCODE_PROMETHEUS_REMOTE_WRITE_CUTOFF_THRESHOLD)) {
             /* Skip processing metrics which are staled over over the threshold */
-            return CMT_ENCODE_PROMETHEUS_REMOTE_WRITE_CUTOFF_ERROR;
+            cutoff = CMT_TRUE;
+            continue;
         }
 
         result = pack_basic_metric_sample(context, map, metric, add_metadata);
@@ -703,7 +706,9 @@ int pack_basic_type(struct cmt_prometheus_remote_write_context *context,
         }
     }
 
-    return CMT_ENCODE_PROMETHEUS_REMOTE_WRITE_SUCCESS;
+    return cutoff == CMT_TRUE ?
+           CMT_ENCODE_PROMETHEUS_REMOTE_WRITE_CUTOFF_ERROR :
+           CMT_ENCODE_PROMETHEUS_REMOTE_WRITE_SUCCESS;
 }
 
 int pack_complex_metric_sample(struct cmt_prometheus_remote_write_context *context,
@@ -1111,7 +1116,8 @@ cfl_sds_t cmt_encode_prometheus_remote_write_create(struct cmt *cmt)
         }
     }
 
-    if (result == CMT_ENCODE_PROMETHEUS_REMOTE_WRITE_SUCCESS) {
+    if (result == CMT_ENCODE_PROMETHEUS_REMOTE_WRITE_SUCCESS ||
+        result == CMT_ENCODE_PROMETHEUS_REMOTE_WRITE_CUTOFF_ERROR) {
         /* Gauges */
         cfl_list_foreach(head, &cmt->gauges) {
             gauge = cfl_list_entry(head, struct cmt_gauge, _head);
@@ -1127,7 +1133,8 @@ cfl_sds_t cmt_encode_prometheus_remote_write_create(struct cmt *cmt)
         }
     }
 
-    if (result == CMT_ENCODE_PROMETHEUS_REMOTE_WRITE_SUCCESS) {
+    if (result == CMT_ENCODE_PROMETHEUS_REMOTE_WRITE_SUCCESS ||
+        result == CMT_ENCODE_PROMETHEUS_REMOTE_WRITE_CUTOFF_ERROR) {
         /* Untyped */
         cfl_list_foreach(head, &cmt->untypeds) {
             untyped = cfl_list_entry(head, struct cmt_untyped, _head);
@@ -1139,7 +1146,8 @@ cfl_sds_t cmt_encode_prometheus_remote_write_create(struct cmt *cmt)
         }
     }
 
-    if (result == CMT_ENCODE_PROMETHEUS_REMOTE_WRITE_SUCCESS) {
+    if (result == CMT_ENCODE_PROMETHEUS_REMOTE_WRITE_SUCCESS ||
+        result == CMT_ENCODE_PROMETHEUS_REMOTE_WRITE_CUTOFF_ERROR) {
         /* Summaries */
         cfl_list_foreach(head, &cmt->summaries) {
             summary = cfl_list_entry(head, struct cmt_summary, _head);
@@ -1155,7 +1163,8 @@ cfl_sds_t cmt_encode_prometheus_remote_write_create(struct cmt *cmt)
         }
     }
 
-    if (result == CMT_ENCODE_PROMETHEUS_REMOTE_WRITE_SUCCESS) {
+    if (result == CMT_ENCODE_PROMETHEUS_REMOTE_WRITE_SUCCESS ||
+        result == CMT_ENCODE_PROMETHEUS_REMOTE_WRITE_CUTOFF_ERROR) {
         /* Histograms */
         cfl_list_foreach(head, &cmt->histograms) {
             histogram = cfl_list_entry(head, struct cmt_histogram, _head);