Yük Testlerinde Kurumsal Gözlemlenebilirlik: Gevent ile ReportPortal'a Canlı Metrik ve Yüzdelik (P99) Raporlama
LocustPilot'ın Gevent coroutine'leri ile ReportPortal'a canlı telemetri akıtması, hata patlamalarını tekilleştirmesi ve P99 yüzdelik analizi.
TL;DR: High-throughput load tests generating thousands of requests per second can easily overwhelm reporting backends with log spam. In LocustPilot, we solved this by implementing an asynchronous gevent listener for ReportPortal that deduplicates error cascades, streams live endpoint metrics every 30 seconds, and calculates comprehensive response time percentiles (P50 to P99) with automated quality gate verdicts.
The Log Flood Dilemma in Load Testing
In standard functional UI or API tests, every failure is unique and warrants a detailed log entry.
In performance testing, the dynamic is completely inverted:
- When an endpoint begins to degrade under 5,000 RPS, it does not fail once—it fails 50,000 times with the exact same
504 Gateway TimeoutorConnection Refusederror. - If your test framework attempts to log every single failure to a centralized telemetry platform like ReportPortal, you will DDoS your own reporting database, exhaust memory, and freeze the test execution.
To provide enterprise-grade observability without degrading load test performance, we built ReportPortalListener (app/core/rp_listener.py) in LocustPilot.
1. Background Telemetry with Gevent Greenlets
Locust relies on gevent for lightweight, non-blocking coroutines. Rather than blocking worker execution during network logging, LocustPilot spawns a background greenlet (_periodic_stats_logger) that flushes aggregated endpoint statistics every 30 seconds:
# app/core/rp_listener.py
import gevent
from reportportal_client.helpers import timestamp
def on_test_start(self, **kwargs):
# Initialize ReportPortal Launch and Test Item
self.launch_uuid = self.rp_client.start_launch(
name=self.rp_launch_name,
start_time=timestamp(),
description=self.rp_launch_desc,
)
self.test_item_uuid = self.rp_client.start_test_item(
name="Load Test Execution",
start_time=timestamp(),
item_type="STEP",
)
# Spawn background stats publisher
self.running = True
gevent.spawn(self._periodic_stats_logger)
def _periodic_stats_logger(self):
"""Log live endpoint stats every 30 seconds without blocking load workers"""
while self.running:
gevent.sleep(30)
if self.running and self.rp_client and self.test_item_uuid:
lines = ["📊 Live Endpoint Telemetry", "=" * 80]
lines.append(f"{'Method':<8} {'Name':<40} {'Reqs':<8} {'Fails':<8} {'Avg(ms)':<10} {'RPS':<8}")
lines.append("-" * 80)
for stat in self.env.stats.entries.values():
if stat.num_requests > 0:
lines.append(
f"{stat.method:<8} {stat.name[:40]:<40} "
f"{stat.num_requests:<8} {stat.num_failures:<8} "
f"{stat.avg_response_time:<10.0f} {stat.current_rps:<8.1f}"
)
self.rp_client.log(
time=timestamp(),
message="\n".join(lines),
level="INFO",
item_id=self.test_item_uuid,
)This ensures team leads and developers watching the ReportPortal dashboard receive live health updates without injecting latency into the simulated users.
2. Smart Error Deduplication (Unique Error Logging)
When a microservice falters, LocustPilot captures the first instance of an error to preserve diagnostic context, while silently aggregating subsequent identical failures into a counter:
def on_request(self, request_type, name, response_time, response_length, exception=None, response=None, **kwargs):
has_exception = exception is not None
has_manual_failure = response is not None and getattr(response, "_manual_result", None) is False
if has_exception or has_manual_failure:
error_key = f"{request_type}:{name}"
# Initialize counter on first occurrence
if error_key not in self.last_stats:
self.last_stats[error_key] = {"count": 0, "logged": False}
self.last_stats[error_key]["count"] += 1
# Only send the first failure payload to ReportPortal
if not self.last_stats[error_key]["logged"] and self.rp_client:
error_msg = str(exception) if has_exception else getattr(response, "_manual_result_msg", "Manual failure")
self.rp_client.log(
time=timestamp(),
message=f"❌ First Failure Detected: {request_type} {name}\nDetails: {error_msg}",
level="ERROR",
item_id=self.test_item_uuid,
)
self.last_stats[error_key]["logged"] = True- Result: Zero log spam. If 100,000 requests fail with the same connection error, ReportPortal receives exactly 1 clean diagnostic entry with a summary count at the end of the test.
3. Comprehensive Percentiles and Automated Quality Gates
Average response time is a dangerous vanity metric in performance engineering; a 200ms average can hide a 99th percentile spike of 12 seconds.
When a test completes (on_test_stop), LocustPilot computes full percentile distributions and enforces automated quality gates:
def on_test_stop(self, **kwargs):
total = self.env.stats.total
# 1. Percentile Distribution Table (50% to 100%)
for stat in self.env.stats.entries.values():
if stat.num_requests > 0:
p50 = stat.get_response_time_percentile(0.50)
p90 = stat.get_response_time_percentile(0.90)
p95 = stat.get_response_time_percentile(0.95)
p99 = stat.get_response_time_percentile(0.99)
max_rt = stat.max_response_time
# 2. Automated Quality Gate Rules
has_issue = False
if total.avg_response_time > 200:
log_warning(f"⚠️ Average response time exceeded 200ms! (Actual: {total.avg_response_time:.2f} ms)")
has_issue = True
if total.fail_ratio > 0.05:
log_warning(f"⚠️ Failed request ratio exceeded 5%! (Actual: {total.fail_ratio:.2%})")
has_issue = True
# 3. Final Build Verdict
success_rate = (1 - total.fail_ratio) * 100
status = "PASSED" if success_rate >= 75 and not has_issue else "FAILED"
self.rp_client.finish_test_item(item_id=self.test_item_uuid, end_time=timestamp(), status=status)
self.rp_client.finish_launch(end_time=timestamp())What Is Next?
In Part 4, we move from single-instance tests to large-scale distributed load: deploying LocustPilot on Kubernetes using Helm charts and Docker worker nodes to simulate over 100,000 requests per second.
👉 Check out the ReportPortal Listener on GitHub
FAQ
Why is ReportPortal preferred over standard HTML reports?
HTML reports are static and isolated on developer machines. ReportPortal centralizes performance history across sprints, enables historical trend comparisons, and automatically flags regression anomalies.
Can custom quality gate thresholds be configured?
Yes. Quality thresholds (such as max average latency or acceptable failure ratio) can be configured via .env variables or custom BaseLocustUser validators.