mirror of
https://github.com/hanzoai/o11y.git
synced 2026-10-15 12:58:33 +00:00
PandeyandGitHub
f21ce5b010
* chore(frontend): remove stale e2e scaffold frontend/e2e/ held an unused settings-only test-plan scaffold from Oct 2025. Active Playwright specs live at tests/e2e/. Drop the directory, the orphan playwright.config.ts, the @playwright/test dependency, and the tsconfig references that pinned them. * chore(e2e): migrate formatter from prettier to oxfmt Swap tests/e2e/ onto oxfmt — same tool the frontend adopted in #11057. Style matches frontend/.oxfmtrc.json (tabs, tabWidth:1) so the two TS trees stay visually consistent. Drops .prettierrc.json and .prettierignore, adds the fmt/fmt:check yarn scripts, and reformats the existing specs. * chore(e2e): migrate linter from eslint to oxlint Drop eslint + @typescript-eslint plugins in favour of oxlint 1.59 + tsgolint — same toolchain the frontend adopted in #10176. The .oxlintrc.json mirrors frontend/.oxlintrc.json with plugins scoped to a Playwright TS codebase (eslint, typescript, unicorn, import, promise). Divergence: eslint-plugin-playwright is not ported. Its rules depend on ESLint APIs (context.getAncestors) that oxlint's JS plugin shim does not implement, so the five playwright/* rules are dropped in this migration. * ci(e2e): add fmtlint job Mirror integrationci.yaml's fmtlint job for e2e. Runs oxfmt --check and oxlint on tests/e2e/ under the same safe-to-e2e label gating as the existing test job. * chore(integration): migrate python tooling from black/pylint/isort/autoflake to ruff Replace the four-tool stack with ruff — same motivation as the oxfmt/oxlint swap on the TS side. One tool covers formatting (ruff format), import sorting (I), unused-import/variable cleanup (F401/F841), and the pylint rules we actually care about (E/W/F/UP/B/PL). Rule set mirrors the intent of the prior pylint config: too-many-* checks and magic-value-comparison stay disabled, dangerous-default-value (now B006) stays muted. A handful of newly-surfaced codes (B011/B024/B905/E741/ UP047/PLC0206/PLW2901) are also muted to keep this a pure tool swap — each deserves its own review before enabling. Divergence: ruff caps line-length at 320, so the prior pylint value of 400 drops to 320. Nothing in tree exceeds 320, so no lines wrap. No changes to integrationci.yaml — both fmt/lint steps still call make py-fmt / make py-lint, which now dispatch to ruff. * chore(e2e): restore playwright lint rules via oxlint jsPlugin [email protected] was rewritten against ESLint 8's context.sourceCode.getAncestors() API, which oxlint's JS plugin shim exposes. The 0.16.x version previously ruled out by context.getAncestors() missing is no longer a blocker. Bump to 2.10.2, re-add it as a jsPlugin, and restore the five rules dropped in the initial oxlint migration: expect-expect, no-conditional-in-test, no-page-pause, no-wait-for-timeout, prefer-web-first-assertions. Rule count: 104 → 109. * chore(frontend): remove stale e2e prompt frontend/prompts/generate-e2e-test.md is leftover from the same Oct 2025 scaffold removed in ebf735dcc. It references frontend/e2e/utils/login.util.ts, which no longer exists, and is not wired into anything. * chore(e2e): make .env.local write layout explicit The single f-string with inline \n escapes read as a wall of text after ruff's line-length allowance collapsed it onto one line. Switch to a triple-quoted f-string so the generated .env.local structure is visible in source. Byte-for-byte identical output. * chore(e2e): write .env.local one key per line Open the file with a context manager and emit each key with its own f.write call. Same output as before, but each key-value pair is a discrete statement.
602 lines
20 KiB
Python
602 lines
20 KiB
Python
import math
|
|
import random
|
|
from collections.abc import Generator
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from fixtures.metrics import Metrics
|
|
|
|
|
|
@dataclass
|
|
class DeterministicAnomalies:
|
|
# resets: (offset)
|
|
COUNTER_RESET_OFFSETS: list[timedelta] = field(
|
|
default_factory=lambda: [
|
|
timedelta(hours=6),
|
|
timedelta(hours=24),
|
|
]
|
|
)
|
|
|
|
# gaps: (offset, duration)
|
|
DATA_GAP_OFFSETS: list[tuple[timedelta, timedelta]] = field(
|
|
default_factory=lambda: [
|
|
(timedelta(hours=12), timedelta(minutes=30)), # 30 min gap
|
|
(timedelta(hours=36), timedelta(minutes=15)), # 15 min gap
|
|
]
|
|
)
|
|
|
|
# spikes: (offset, multiplier)
|
|
SPIKE_OFFSETS: list[tuple[timedelta, float]] = field(
|
|
default_factory=lambda: [
|
|
(timedelta(hours=18), 10.0), # at this point the value spikes 10x
|
|
]
|
|
)
|
|
|
|
# 0 value offsets
|
|
ZERO_VALUE_OFFSETS: list[timedelta] = field(
|
|
default_factory=lambda: [
|
|
timedelta(hours=30), # at this point the value drops to zero
|
|
]
|
|
)
|
|
|
|
def is_in_gap(self, offset: timedelta) -> bool:
|
|
for gap_start, gap_duration in self.DATA_GAP_OFFSETS:
|
|
if gap_start <= offset < gap_start + gap_duration:
|
|
return True
|
|
return False
|
|
|
|
def get_spike_multiplier(self, offset: timedelta) -> float:
|
|
for spike_offset, multiplier in self.SPIKE_OFFSETS:
|
|
# lasts for 1 minute, make it configurable?
|
|
if spike_offset <= offset < spike_offset + timedelta(minutes=1):
|
|
return multiplier
|
|
return 1.0
|
|
|
|
def should_reset_counter(self, offset: timedelta) -> bool:
|
|
for reset_offset in self.COUNTER_RESET_OFFSETS:
|
|
if reset_offset <= offset < reset_offset + timedelta(minutes=1):
|
|
return True
|
|
return False
|
|
|
|
def should_be_zero(self, offset: timedelta) -> bool:
|
|
for zero_offset in self.ZERO_VALUE_OFFSETS:
|
|
# window of 5 minutes
|
|
if zero_offset <= offset < zero_offset + timedelta(minutes=5):
|
|
return True
|
|
return False
|
|
|
|
|
|
class MetricsDataGenerator:
|
|
def __init__(
|
|
self,
|
|
start_time: datetime,
|
|
end_time: datetime,
|
|
step_seconds: int = 60,
|
|
seed: int = 42,
|
|
):
|
|
self.start = start_time
|
|
self.end = end_time
|
|
self.step = step_seconds
|
|
self.anomalies = DeterministicAnomalies()
|
|
self.rng = random.Random(seed)
|
|
|
|
duration = (self.end - self.start).total_seconds()
|
|
self.num_points = int(duration / step_seconds)
|
|
|
|
def _timestamps(self) -> Generator[datetime]:
|
|
current = self.start
|
|
while current < self.end:
|
|
yield current
|
|
current += timedelta(seconds=self.step)
|
|
|
|
def _offset_from_start(self, ts: datetime) -> timedelta:
|
|
return ts - self.start
|
|
|
|
def steady_with_noise(self, base: float, noise_pct: float = 0.1) -> Generator[float]:
|
|
for _ in range(self.num_points):
|
|
noise = self.rng.uniform(-noise_pct, noise_pct) * base
|
|
yield base + noise
|
|
|
|
def diurnal_pattern(self, min_val: float, max_val: float) -> Generator[float]:
|
|
amplitude = (max_val - min_val) / 2
|
|
baseline = min_val + amplitude
|
|
|
|
for i, ts in enumerate[datetime](self._timestamps()):
|
|
# peak at 14:00
|
|
hour = ts.hour + ts.minute / 60.0
|
|
# shift so peak is at 14:00
|
|
phase = (hour - 14) * (2 * math.pi / 24)
|
|
value = baseline + amplitude * math.cos(phase)
|
|
# add some noise
|
|
noise = self.rng.uniform(-0.05, 0.05) * value
|
|
yield max(0, value + noise)
|
|
if i >= self.num_points - 1:
|
|
break
|
|
|
|
def monotonic_increasing(self, rate_per_minute: float, noise_pct: float = 0.1) -> Generator[float]:
|
|
value = 0.0
|
|
for ts in self._timestamps():
|
|
offset = self._offset_from_start(ts)
|
|
|
|
# reset now?
|
|
if self.anomalies.should_reset_counter(offset):
|
|
value = 0.0
|
|
|
|
increment = rate_per_minute * (self.step / 60)
|
|
noise = self.rng.uniform(1 - noise_pct, 1 + noise_pct)
|
|
value += increment * noise
|
|
|
|
yield value
|
|
|
|
def spike_pattern(
|
|
self,
|
|
base: float,
|
|
spike_factor: float = 10.0,
|
|
spike_interval_minutes: int = 60,
|
|
) -> Generator[float]:
|
|
for i, ts in enumerate(self._timestamps()):
|
|
offset = self._offset_from_start(ts)
|
|
|
|
# deterministic spike
|
|
multiplier = self.anomalies.get_spike_multiplier(offset)
|
|
if multiplier > 1.0:
|
|
value = base * multiplier
|
|
# periodic spike
|
|
elif i > 0 and i % spike_interval_minutes == 0:
|
|
value = base * spike_factor
|
|
else:
|
|
noise = self.rng.uniform(-0.1, 0.1) * base
|
|
value = base + noise
|
|
|
|
yield value
|
|
|
|
def gauge_with_zero_drops(
|
|
self,
|
|
base: float,
|
|
noise_pct: float = 0.1,
|
|
) -> Generator[float]:
|
|
for ts in self._timestamps():
|
|
offset = self._offset_from_start(ts)
|
|
|
|
if self.anomalies.should_be_zero(offset):
|
|
yield 0.0
|
|
else:
|
|
noise = self.rng.uniform(-noise_pct, noise_pct) * base
|
|
yield max(0, base + noise)
|
|
|
|
def latency_distribution(self, p50_ms: float, p99_ms: float) -> Generator[list[float]]:
|
|
# see otel defaults (in ms)
|
|
# https://opentelemetry.io/docs/specs/otel/metrics/sdk/#explicit-bucket-histogram-aggregation
|
|
buckets = [5, 10, 25, 50, 75, 100, 250, 500, 750, 1000, 2500, 5000, 7500, 10000]
|
|
|
|
for _ in range(self.num_points):
|
|
# log-normal distribution
|
|
sigma = math.log(p99_ms / p50_ms) / 2.33 # z-score for 99th percentile
|
|
mu = math.log(p50_ms)
|
|
|
|
# observations for each bucket
|
|
counts = []
|
|
total_requests = self.rng.randint(100, 1000)
|
|
|
|
for i, bucket in enumerate(buckets):
|
|
# CDF
|
|
if bucket <= 0:
|
|
prob = 0
|
|
else:
|
|
z = (math.log(bucket) - mu) / sigma
|
|
prob = 0.5 * (1 + math.erf(z / math.sqrt(2)))
|
|
|
|
if i == 0:
|
|
count = int(prob * total_requests)
|
|
else:
|
|
prev_bucket = buckets[i - 1]
|
|
z_prev = (math.log(prev_bucket) - mu) / sigma
|
|
prob_prev = 0.5 * (1 + math.erf(z_prev / math.sqrt(2)))
|
|
count = int((prob - prob_prev) * total_requests)
|
|
|
|
counts.append(max(0, count))
|
|
|
|
yield counts
|
|
|
|
def generate_gauge_metrics(
|
|
self,
|
|
metric_name: str = "system.cpu.utilization",
|
|
services: list[str] = None,
|
|
hosts: list[str] = None,
|
|
) -> list[Metrics]:
|
|
if services is None:
|
|
services = ["frontend", "backend", "worker", "database"]
|
|
if hosts is None:
|
|
hosts = ["host-1", "host-2", "host-3", "host-4", "host-5"]
|
|
|
|
metrics = []
|
|
for service in services:
|
|
for host in hosts:
|
|
# diurnal pattern for CPU utilization (as a pct, we could have done state based counter metric as well)
|
|
values = list[float](self.diurnal_pattern(20.0, 80.0))
|
|
labels = {"service_name": service, "host": host}
|
|
|
|
for i, ts in enumerate[datetime](self._timestamps()):
|
|
offset = self._offset_from_start(ts)
|
|
|
|
# inject gaps
|
|
if self.anomalies.is_in_gap(offset):
|
|
continue
|
|
|
|
if i < len(values):
|
|
metrics.append(
|
|
Metrics(
|
|
metric_name=metric_name,
|
|
labels=labels,
|
|
timestamp=ts,
|
|
value=values[i],
|
|
temporality="Unspecified",
|
|
type_="Gauge",
|
|
is_monotonic=False,
|
|
)
|
|
)
|
|
|
|
return metrics
|
|
|
|
def generate_cumulative_counter_metrics(
|
|
self,
|
|
metric_name: str = "http.server.request.count",
|
|
services: list[str] = None,
|
|
endpoints: list[str] = None,
|
|
status_codes: list[str] = None,
|
|
) -> list[Metrics]:
|
|
if services is None:
|
|
services = ["api", "web", "auth"]
|
|
if endpoints is None:
|
|
endpoints = [
|
|
"/api/v1/users",
|
|
"/api/v1/orders",
|
|
"/api/v1/products",
|
|
"/health",
|
|
"/metrics",
|
|
]
|
|
if status_codes is None:
|
|
status_codes = ["200", "500"]
|
|
|
|
metrics = []
|
|
for service in services:
|
|
for endpoint in endpoints:
|
|
for status_code in status_codes:
|
|
# rate varies by status code
|
|
base_rate = 100.0 if status_code == "200" else 5.0
|
|
values = list(self.monotonic_increasing(base_rate, 0.1))
|
|
labels = {
|
|
"service_name": service,
|
|
"endpoint": endpoint,
|
|
"status_code": status_code,
|
|
}
|
|
|
|
for i, ts in enumerate[datetime](self._timestamps()):
|
|
offset = self._offset_from_start(ts)
|
|
|
|
# inject artificial gaps
|
|
if self.anomalies.is_in_gap(offset):
|
|
continue
|
|
|
|
if i < len(values):
|
|
metrics.append(
|
|
Metrics(
|
|
metric_name=metric_name,
|
|
labels=labels,
|
|
timestamp=ts,
|
|
value=values[i],
|
|
temporality="Cumulative",
|
|
type_="Sum",
|
|
is_monotonic=True,
|
|
)
|
|
)
|
|
|
|
return metrics
|
|
|
|
def generate_delta_counter_metrics(
|
|
self,
|
|
metric_name: str = "http.server.request.delta",
|
|
services: list[str] = None,
|
|
endpoints: list[str] = None,
|
|
status_codes: list[str] = None,
|
|
) -> list[Metrics]:
|
|
if services is None:
|
|
services = ["api", "web", "auth"]
|
|
if endpoints is None:
|
|
endpoints = [
|
|
"/api/v1/users",
|
|
"/api/v1/orders",
|
|
"/api/v1/products",
|
|
"/health",
|
|
"/metrics",
|
|
]
|
|
if status_codes is None:
|
|
status_codes = ["200", "500"]
|
|
|
|
metrics = []
|
|
for service in services:
|
|
for endpoint in endpoints:
|
|
for status_code in status_codes:
|
|
base_count = 100.0 if status_code == "200" else 5.0
|
|
labels = {
|
|
"service_name": service,
|
|
"endpoint": endpoint,
|
|
"status_code": status_code,
|
|
}
|
|
|
|
for ts in self._timestamps():
|
|
offset = self._offset_from_start(ts)
|
|
|
|
if self.anomalies.is_in_gap(offset):
|
|
continue
|
|
|
|
multiplier = self.anomalies.get_spike_multiplier(offset)
|
|
noise = self.rng.uniform(0.8, 1.2)
|
|
value = base_count * multiplier * noise
|
|
|
|
metrics.append(
|
|
Metrics(
|
|
metric_name=metric_name,
|
|
labels=labels,
|
|
timestamp=ts,
|
|
value=value,
|
|
temporality="Delta",
|
|
type_="Sum",
|
|
is_monotonic=True,
|
|
)
|
|
)
|
|
|
|
return metrics
|
|
|
|
def generate_connection_gauge_metrics(
|
|
self,
|
|
metric_name: str = "db.client.connections",
|
|
services: list[str] = None,
|
|
pool_names: list[str] = None,
|
|
) -> list[Metrics]:
|
|
if services is None:
|
|
services = ["backend", "worker"]
|
|
if pool_names is None:
|
|
pool_names = ["primary", "replica"]
|
|
|
|
metrics = []
|
|
for service in services:
|
|
for pool in pool_names:
|
|
base = 20.0 if pool == "primary" else 10.0
|
|
values = list[float](self.gauge_with_zero_drops(base, 0.2))
|
|
labels = {"service_name": service, "pool": pool}
|
|
|
|
for i, ts in enumerate[datetime](self._timestamps()):
|
|
offset = self._offset_from_start(ts)
|
|
|
|
if self.anomalies.is_in_gap(offset):
|
|
continue
|
|
|
|
if i < len(values):
|
|
metrics.append(
|
|
Metrics(
|
|
metric_name=metric_name,
|
|
labels=labels,
|
|
timestamp=ts,
|
|
value=values[i],
|
|
temporality="Unspecified",
|
|
type_="Gauge",
|
|
is_monotonic=False,
|
|
)
|
|
)
|
|
|
|
return metrics
|
|
|
|
def generate_gc_duration_metrics(
|
|
self,
|
|
metric_name: str = "process.runtime.gc.duration",
|
|
services: list[str] = None,
|
|
) -> list[Metrics]:
|
|
if services is None:
|
|
services = ["frontend", "backend", "worker", "api"]
|
|
|
|
metrics = []
|
|
for service in services:
|
|
values = list[float](self.spike_pattern(5.0, 10.0, 30)) # bump every 30 minutes (to similate real pattern)
|
|
labels = {"service_name": service}
|
|
|
|
for i, ts in enumerate[datetime](self._timestamps()):
|
|
offset = self._offset_from_start(ts)
|
|
|
|
if self.anomalies.is_in_gap(offset):
|
|
continue
|
|
|
|
if i < len(values):
|
|
metrics.append(
|
|
Metrics(
|
|
metric_name=metric_name,
|
|
labels=labels,
|
|
timestamp=ts,
|
|
value=values[i],
|
|
temporality="Unspecified",
|
|
type_="Gauge",
|
|
is_monotonic=False,
|
|
)
|
|
)
|
|
|
|
return metrics
|
|
|
|
def generate_all_metrics(self) -> dict[str, list[Metrics]]:
|
|
return {
|
|
"gauge": self.generate_gauge_metrics(),
|
|
"cumulative_counter": self.generate_cumulative_counter_metrics(),
|
|
"delta_counter": self.generate_delta_counter_metrics(),
|
|
"connection_gauge": self.generate_connection_gauge_metrics(),
|
|
"gc_duration": self.generate_gc_duration_metrics(),
|
|
}
|
|
|
|
def generate_flat_metrics(self) -> list[Metrics]:
|
|
all_metrics = []
|
|
for metrics_list in self.generate_all_metrics().values():
|
|
all_metrics.extend(metrics_list)
|
|
return all_metrics
|
|
|
|
|
|
def create_test_data_generator(
|
|
duration_hours: int = 48,
|
|
step_seconds: int = 60,
|
|
seed: int = 42,
|
|
) -> MetricsDataGenerator:
|
|
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
|
start = now - timedelta(hours=duration_hours)
|
|
return MetricsDataGenerator(
|
|
start_time=start,
|
|
end_time=now,
|
|
step_seconds=step_seconds,
|
|
seed=seed,
|
|
)
|
|
|
|
|
|
def generate_simple_gauge_series(
|
|
metric_name: str,
|
|
labels: dict[str, str],
|
|
values: list[float],
|
|
start_time: datetime | None = None,
|
|
step_seconds: int = 60,
|
|
) -> list[Metrics]:
|
|
if start_time is None:
|
|
start_time = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
|
|
|
metrics = []
|
|
for i, value in enumerate(values):
|
|
ts = start_time + timedelta(seconds=i * step_seconds)
|
|
metrics.append(
|
|
Metrics(
|
|
metric_name=metric_name,
|
|
labels=labels,
|
|
timestamp=ts,
|
|
value=value,
|
|
temporality="Unspecified",
|
|
type_="Gauge",
|
|
is_monotonic=False,
|
|
)
|
|
)
|
|
return metrics
|
|
|
|
|
|
def generate_simple_counter_series(
|
|
metric_name: str,
|
|
labels: dict[str, str],
|
|
values: list[float],
|
|
temporality: str = "Cumulative",
|
|
start_time: datetime | None = None,
|
|
step_seconds: int = 60,
|
|
) -> list[Metrics]:
|
|
if start_time is None:
|
|
start_time = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
|
|
|
metrics = []
|
|
for i, value in enumerate(values):
|
|
ts = start_time + timedelta(seconds=i * step_seconds)
|
|
metrics.append(
|
|
Metrics(
|
|
metric_name=metric_name,
|
|
labels=labels,
|
|
timestamp=ts,
|
|
value=value,
|
|
temporality=temporality,
|
|
type_="Sum",
|
|
is_monotonic=True,
|
|
)
|
|
)
|
|
return metrics
|
|
|
|
|
|
def generate_non_monotonic_sum_series(
|
|
metric_name: str,
|
|
labels: dict[str, str],
|
|
values: list[float],
|
|
temporality: str = "Cumulative",
|
|
start_time: datetime | None = None,
|
|
step_seconds: int = 60,
|
|
) -> list[Metrics]:
|
|
if start_time is None:
|
|
start_time = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
|
|
|
metrics = []
|
|
for i, value in enumerate(values):
|
|
ts = start_time + timedelta(seconds=i * step_seconds)
|
|
metrics.append(
|
|
Metrics(
|
|
metric_name=metric_name,
|
|
labels=labels,
|
|
timestamp=ts,
|
|
value=value,
|
|
temporality=temporality,
|
|
type_="Sum",
|
|
is_monotonic=False, # non-monotonic sums are gauges for us
|
|
)
|
|
)
|
|
return metrics
|
|
|
|
|
|
def generate_counter_with_resets(
|
|
metric_name: str,
|
|
labels: dict[str, str],
|
|
num_points: int,
|
|
rate_per_point: float,
|
|
reset_at_indices: list[int],
|
|
start_time: datetime | None = None,
|
|
step_seconds: int = 60,
|
|
) -> list[Metrics]:
|
|
if start_time is None:
|
|
start_time = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
|
|
|
metrics = []
|
|
value = 0.0
|
|
for i in range(num_points):
|
|
if i in reset_at_indices:
|
|
value = 0.0
|
|
else:
|
|
value += rate_per_point
|
|
|
|
ts = start_time + timedelta(seconds=i * step_seconds)
|
|
metrics.append(
|
|
Metrics(
|
|
metric_name=metric_name,
|
|
labels=labels,
|
|
timestamp=ts,
|
|
value=value,
|
|
temporality="Cumulative",
|
|
type_="Sum",
|
|
is_monotonic=True,
|
|
)
|
|
)
|
|
return metrics
|
|
|
|
|
|
def generate_sparse_series(
|
|
metric_name: str,
|
|
labels: dict[str, str],
|
|
values_at_indices: dict[int, float],
|
|
total_points: int,
|
|
temporality: str = "Unspecified",
|
|
type_: str = "Gauge",
|
|
start_time: datetime | None = None,
|
|
step_seconds: int = 60,
|
|
) -> list[Metrics]:
|
|
if start_time is None:
|
|
start_time = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
|
|
|
metrics = []
|
|
for i, value in values_at_indices.items():
|
|
if i < total_points:
|
|
ts = start_time + timedelta(seconds=i * step_seconds)
|
|
metrics.append(
|
|
Metrics(
|
|
metric_name=metric_name,
|
|
labels=labels,
|
|
timestamp=ts,
|
|
value=value,
|
|
temporality=temporality,
|
|
type_=type_,
|
|
is_monotonic=type_ == "Sum",
|
|
)
|
|
)
|
|
return metrics
|