Fix a spike in monotonic counter values after a failed openmetrics scrape (#20063)

* AGTMETRICS-177 skip flush first value on error

flush_first_value indicates that new monotonic counters values should
assume previous value of zero and be included in the very next flush,
instead of waiting for another value to compute a diff.

This breaks when scrape fails. We can no longer assume that the
counter just appeared, and reporting it may cause an unwanted spike.
Aggregator retention for previous values is also short lived, so if
few scrapes are missed, even old counters will appear as new to the
aggregator, causing even bigger spikes.

* Changelog

* Preserve use process_start_time on first successful scrape

If the target process starts after the agent, we may have a few failed
scrapes, but the values on first scrape can be fresh enough to use as
is.

Author: vickenty

datadog_checks_base/changelog.d/20063.fixed

@@ -0,0 +1 @@
+Fix a spike in monotonic counter values after a failed openmetrics scrape

datadog_checks_base/datadog_checks/base/checks/openmetrics/v2/scraper.py

@@ -230,13 +230,13 @@ def __init__(self, check, config):
         self.use_process_start_time = is_affirmative(config.get('use_process_start_time'))
 
         # Used for monotonic counts
-        self.flush_first_value = False
+        self.flush_first_value = None
 
-    def scrape(self):
+    def _scrape(self):
         """
         Execute a scrape, and for each metric collected, transform the metric.
         """
-        runtime_data = {'flush_first_value': self.flush_first_value, 'static_tags': self.static_tags}
+        runtime_data = {'flush_first_value': bool(self.flush_first_value), 'static_tags': self.static_tags}
 
         # Determine which consume method to use based on target_info config
         if self.target_info:
@@ -251,7 +251,18 @@ def scrape(self):
 
             transformer(metric, self.generate_sample_data(metric), runtime_data)
 
-        self.flush_first_value = True
+    def scrape(self):
+        try:
+            self._scrape()
+            self.flush_first_value = True
+        except:
+            # Don't flush new monotonic counts on next scrape:
+            # 1. Previous value may have expired in the aggregator, causing a spike
+            # 2. New counter itself may be too old and large when we discover it next time.
+            # If we didn't have a successful scrape yet, keep the initial value (use process_start_time to decide).
+            if self.flush_first_value:
+                self.flush_first_value = False
+            raise
 
     def consume_metrics(self, runtime_data):
         """
@@ -260,7 +271,7 @@ def consume_metrics(self, runtime_data):
 
         metric_parser = self.parse_metrics()
 
-        if not self.flush_first_value and self.use_process_start_time:
+        if self.flush_first_value is None and self.use_process_start_time:
             metric_parser = first_scrape_handler(metric_parser, runtime_data, datadog_agent.get_process_start_time())
         if self.label_aggregator.configured:
             metric_parser = self.label_aggregator(metric_parser)
@@ -283,7 +294,7 @@ def consume_metrics_w_target_info(self, runtime_data):
 
         metric_parser = self.parse_metrics()
 
-        if not self.flush_first_value and self.use_process_start_time:
+        if self.flush_first_value is None and self.use_process_start_time:
             metric_parser = first_scrape_handler(metric_parser, runtime_data, datadog_agent.get_process_start_time())
         if self.label_aggregator.configured:
             metric_parser = self.label_aggregator(metric_parser)

datadog_checks_base/tests/base/checks/openmetrics/test_v2/test_first_scrape_handler.py

@@ -86,8 +86,6 @@ def test_first_scrape_handler(
     agent_start_time,
     process_start_time,
 ):
-    mock_http_response(_make_test_data(process_start_time))
-
     check = get_check(
         {
             'metrics': ['.*'],
@@ -98,53 +96,62 @@ def test_first_scrape_handler(
 
     datadog_agent.set_process_start_time(agent_start_time)
 
-    for _ in range(0, 5):
+    for idx in range(0, 10):
         aggregator.reset()
-        dd_run_check(check)
 
-        aggregator.assert_metric(
-            'test.go_memstats_alloc_bytes.count',
-            metric_type=aggregator.MONOTONIC_COUNT,
-            count=1,
-            flush_first_value=expect_first_flush,
-        )
-        aggregator.assert_metric(
-            'test.go_gc_duration_seconds.count',
-            metric_type=aggregator.MONOTONIC_COUNT,
-            count=1,
-            flush_first_value=expect_first_flush,
-        )
+        if idx != 5:
+            mock_http_response(_make_test_data(process_start_time))
+            dd_run_check(check)
 
-        if with_buckets:
-            aggregator.assert_histogram_bucket(
-                'test.skydns_skydns_dns_request_duration_seconds',
-                None,
-                None,
-                None,
-                True,
-                None,
-                None,
-                count=2,
-                flush_first_value=expect_first_flush,
-            )
-        else:
-            aggregator.assert_metric(
-                'test.skydns_skydns_dns_request_duration_seconds.count',
-                metric_type=aggregator.MONOTONIC_COUNT,
-                count=1,
-                flush_first_value=expect_first_flush,
-            )
             aggregator.assert_metric(
-                'test.skydns_skydns_dns_request_duration_seconds.sum',
+                'test.go_memstats_alloc_bytes.count',
                 metric_type=aggregator.MONOTONIC_COUNT,
                 count=1,
                 flush_first_value=expect_first_flush,
             )
             aggregator.assert_metric(
-                'test.skydns_skydns_dns_request_duration_seconds.bucket',
+                'test.go_gc_duration_seconds.count',
                 metric_type=aggregator.MONOTONIC_COUNT,
                 count=1,
                 flush_first_value=expect_first_flush,
             )
 
-        expect_first_flush = True
+            if with_buckets:
+                aggregator.assert_histogram_bucket(
+                    'test.skydns_skydns_dns_request_duration_seconds',
+                    None,
+                    None,
+                    None,
+                    True,
+                    None,
+                    None,
+                    count=2,
+                    flush_first_value=expect_first_flush,
+                )
+            else:
+                aggregator.assert_metric(
+                    'test.skydns_skydns_dns_request_duration_seconds.count',
+                    metric_type=aggregator.MONOTONIC_COUNT,
+                    count=1,
+                    flush_first_value=expect_first_flush,
+                )
+                aggregator.assert_metric(
+                    'test.skydns_skydns_dns_request_duration_seconds.sum',
+                    metric_type=aggregator.MONOTONIC_COUNT,
+                    count=1,
+                    flush_first_value=expect_first_flush,
+                )
+                aggregator.assert_metric(
+                    'test.skydns_skydns_dns_request_duration_seconds.bucket',
+                    metric_type=aggregator.MONOTONIC_COUNT,
+                    count=1,
+                    flush_first_value=expect_first_flush,
+                )
+
+            expect_first_flush = True
+        else:
+            mock_http_response("", status_code=500)
+            with pytest.raises(Exception):
+                dd_run_check(check)
+
+            expect_first_flush = False