integration test

relay-server/src/services/metrics/aggregator.rs

@@ -113,13 +113,16 @@ impl AggregatorService {
  config: AggregatorServiceConfig,
  receiver: Option<Recipient<FlushBuckets, NoResponse>>,
  ) -> Self {
+ let aggregator = aggregator::Aggregator::named(name, config.aggregator);
  Self {
  receiver,
  state: AggregatorState::Running,
  max_total_bucket_bytes: config.max_total_bucket_bytes,
- aggregator: aggregator::Aggregator::named(name, config.aggregator),
  flush_interval_ms: config.flush_interval_ms,
- can_accept_metrics: Arc::new(AtomicBool::new(true)),
+ can_accept_metrics: Arc::new(AtomicBool::new(
+ !aggregator.totals_cost_exceeded(config.max_total_bucket_bytes),
+ )),
+ aggregator,
  }
  }
 

tests/integration/test_healthchecks.py

@@ -123,6 +123,29 @@ def test_readiness_depends_on_aggregator_being_full(mini_sentry, relay):
  assert response.status_code == 503
 
 
+def test_readiness_depends_on_aggregator_being_full_after_metrics(mini_sentry, relay):
+ relay = relay(
+ mini_sentry,
+ {"aggregator": {"max_total_bucket_bytes": 1}},
+ )
+
+ metrics_payload = f"transactions/foo:42|c\ntransactions/bar:17|c"
+ relay.send_metrics(42, metrics_payload)
+
+ for _ in range(100):
+ response = wait_get(relay, "/api/relay/healthcheck/ready/")
+ print(response, response.status_code)
+ if response.status_code == 503:
+ error = str(mini_sentry.test_failures.pop())
+ assert "Health check probe 'aggregator'" in error
+ error = str(mini_sentry.test_failures.pop())
+ assert "aggregator limit exceeded" in error
+ return
+ time.sleep(0.1)
+
+ assert False, "health check never failed"
+
+
 def test_readiness_disk_spool(mini_sentry, relay):
  try:
  temp = tempfile.mkdtemp()