Elasticsearch Retention on ILM — Implementation Plan
For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (
- [ ]) syntax for tracking.
Goal: Replace the emergency/critical/flood retention ladder with Elasticsearch ILM rollover (age rule, zero Logster code) plus a small disk-cap worker that deletes the oldest whole Logster index while disk is above target, on Docker Compose and the air-gapped OpenShift chart.
Architecture: One stdlib-only module, layout.py, declaratively converges ILM policies, index templates, legacy-index adoption and rollover write aliases. It runs from init.py (at deploy time) and from the worker (every check, as self-heal). diskcap.py plans whole-index deletions from exact index sizes, never touching a write index or data newer than the floor. Writers move to write aliases (logster-<family>-write), readers to wildcard patterns (logster-<family>*). Vector retries flood-stage rejections (request_retry_partial).
Tech Stack: Python 3.12 stdlib (alpine logster-elasticsearch-init image), Elasticsearch 9.5.4 basic license (ILM), Vector 0.52, Node 24 Express dashboard server + React client, FastAPI api, Helm 3 chart for OpenShift, Docker Compose.
Spec: docs/superpowers/specs/2026-09-25-es-ilm-retention-design.md (approved, revision 2). Executors read the spec and this plan together. Section references (§) point into the spec.
Global Constraints
deploy/elasticsearch/*.pystays Python standard library only (runs onalpine:3.22+python3, no pip). Target Python 3.12.- Air-gapped: no new images, no network calls beyond
http://elasticsearch:9200(and Kafka for Vector). The Vector initContainer reuses thelogster-elasticsearch-initimage. - Names (verbatim): write alias
logster-<family>-write; rollover indexlogster-<family>-NNNNNN(6+ digits, first is000001); read patternlogster-<family>*; legacy indexlogster-<family>; ILM policieslogster-<family>andlogster-<family>-legacy; familiesevents,inferences,graphs. - The worker may delete only names matching
^logster-(events|inferences|graphs)(-\d{6,})?$, only by exact name, never a write index, and never an index holding data newer thanRETENTION_MIN_AGE_HOURS(checked with Elasticsearch's clock). - Knobs and defaults:
RETENTION_EVENTS_DAYS=30,RETENTION_INFERENCES_DAYS=90,RETENTION_GRAPHS_DAYS=14(0–3650, 0 = keep forever),RETENTION_TARGET_PERCENT=75(1–89),RETENTION_MIN_AGE_HOURS=24(1–720),RETENTION_CHECK_INTERVAL_MINUTES=5(1–1440). An invalid value logs a WARNING and uses the default. It never exits. - Removed knobs, warned about when set:
RETENTION_STRIP_RAW_EVENT_DAYS,RETENTION_INTERVAL_HOURS,RETENTION_EMERGENCY_THRESHOLD_PERCENT,RETENTION_CRITICAL_THRESHOLD_PERCENT. - Rollover:
max_age: 1dandmax_primary_shard_size = clamp(floor(total_fs_bytes/100) in whole GiB, 1, 50)gb. If the disk size is unknown, keep the existing policy, or use5gbon first bootstrap. - Change detection for policies and templates uses
_meta.logster_hash= SHA-256 of canonical JSON (sort_keys=True, separators=(",", ":")) of the body without_meta. init.pyexits non-zero only when Elasticsearch is unreachable or rejects credentials, or when a PUT of cluster settings, a policy, a template or a bootstrap index is rejected (§5.1).converge()gets up to 5 attempts, 10 s apart, before init fails. Fatal (the "unreachable / misconfiguration" class) are: a failed policy/template/alias GET, a failed_cat/indiceslisting in_bootstrap, a missing template file, and a crash in_converge_policies,_converge_templatesor_ensure_write_aliases. A rejected/_cluster/rerouteis only a WARNING (exit 0).- The Elasticsearch
-Eargument lists indeploy/docker-compose.ymlanddeploy/openshift/logster/templates/elasticsearch/statefulset.yamlmust be identical (CLAUDE.md invariant, enforced bytest_chart_drift.py). deploy/openshift/logster/files/elasticsearch/*must be byte-identical todeploy/elasticsearch/*for every shipped file.files/vector/vector.yamlmust equaldeploy/vector/vector.yamlexcept for theES_PASSWORDdefault.- Commits (user rule): never commit unless the user explicitly asks. Each task ends by staging its files and proposing a commit message. If the user has asked for per-task commits, commit as
Abdullah Chaudhrywith noCo-Authored-Bytrailer. - Keep existing code idioms:
service.pyorchestrates,metrics.pyholds Prometheus metrics, and the dashboard server queries ES directly viaelastic.js.
Review Focus
These five input classes aren't covered by any spec scenario but are likely to hit real users. Each has a pinning test in the task named.
- Index names that look like Logster's but aren't managed (
logster-events-backup,logster-events-00001,logster-eventsx, an operator's copy). Expected: never deleted. Test: Task 5test_look_alike_names_are_never_deleted. - Future-dated
@timestampfrom a shipper with a wrong clock. Expected: the index counts as recent, so the disk rule never deletes it early and never crashes. Test: Task 5test_future_dated_data_counts_as_recent. - The legacy index is already flood-blocked when adoption runs (settings PUT → 429). Expected: logged as a problem, not fatal, retried on the next check. Test: Task 4
test_legacy_under_flood_block_is_a_problem_not_fatal. - Credentials rejected mid-life (password rotated, HTTP 401). Expected: the worker logs a hint and keeps running; init exits 1 with the hint. Tests: Task 7
test_401_logs_credential_hint_and_keeps_running, Task 8test_401_exits_one_with_hint. - Messy env values (
" 30 ","1.5","-1","99999",""). Expected: trimmed or defaulted with a WARNING, never a crash. Test: Task 1test_messy_values_fall_back_with_warning.
File Map
Create
| File | Responsibility |
|---|---|
| deploy/elasticsearch/eshttp.py | stdlib HTTP request, JSON call, make_logger, es_status, fs_usage, iso_ms |
| deploy/elasticsearch/retention_config.py | RetentionConfig, load_config, rollover_size_gb, removed-var warnings |
| deploy/elasticsearch/layout.py | converge(), wait_for_aliases(), names/regexes, CLI layout.py wait |
| deploy/elasticsearch/diskcap.py | enforce() — the disk rule |
| deploy/elasticsearch/fake_es.py | In-memory ES double for unit tests (not shipped) |
| deploy/elasticsearch/test_eshttp.py, test_retention_config.py, test_fake_es.py, test_layout.py, test_templates.py, test_diskcap.py, test_init.py, test_chart_drift.py, test_vector_config.py | Unit tests |
| deploy/elasticsearch/integration/run.sh, integration/test_es_integration.py, integration/test_vector_integration.py, integration/vector-it.yaml | Real-image integration suite |
| services/inference/tests/test_graph_sink.py | Bulk per-item error accounting |
| services/api/tests/test_whitelist_store.py | Graph/inference lookups against a stub ES |
| services/dashboard/server/src/lib/indexFamilies.js | Group _cat/indices rows by family |
| services/dashboard/server/test/indexFamilies.test.js, test/inferencesRoute.test.js, test/graphDoc.test.js | Dashboard tests |
| .github/workflows/es-retention-ci.yml | Unit tests, drift test, helm lint/template |
Modify
| File | Change |
|---|---|
| deploy/elasticsearch/retention.py | Rewrite: loop of converge → ILM report → disk rule → forecast |
| deploy/elasticsearch/init.py | Rewrite on top of layout.converge(); new exit semantics |
| deploy/elasticsearch/capacity.py | Patterns, no strip model, target_pct only |
| deploy/elasticsearch/logster-{events,inferences,graphs}.template.json | Lifecycle settings, best_compression, sort mappings |
| deploy/elasticsearch/test_retention.py, test_capacity.py | Rewritten / updated |
| deploy/vector/vector.yaml + chart copy | Write aliases, request_retry_partial: true |
| deploy/docker-compose.yml | ES args (guard + ml), init env, inference gating, retention env |
| deploy/openshift/logster/** | ConfigMap, init Job env, retention Deployment, Vector initContainer + checksum, ES arg, values, NOTES, README, file copies |
| libs/logster-common/logster_common/config.py, deploy/service-config.yaml, libs/logster-ml/logster_ml/graph/hydration.py | Index defaults |
| services/inference/src/logster_inference/sinks/graph_sink.py | Per-item bulk errors |
| services/api/src/logster_api/whitelist_store.py, evidence.py, services/api/tests/test_evidence_endpoint.py | Pattern reads, _index-targeted update |
| services/dashboard/server/src/config.js, routes/{graph,inferenceDetail,inferences,recentAttacks,search,storage}.js, lib/capacity.js, tests | Patterns, sort allowlist, unmapped_type on @timestamp sorts, family grouping, no strip |
| services/dashboard/client/src/components/health/StorageSection.jsx, lib/panelInfo.js | "Days until early deletion" |
| tests/e2e-openshift/lib.sh, scripts/stress_test/producer.py | Patterns |
| Docs + CLAUDE.md | §14 of the spec |
How to run the tests (used by every task)
# ES scripts (stdlib unittest; discovery puts deploy/elasticsearch on sys.path)
venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_*.py' -v
# one file:
venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_layout.py' -v
# Python services
venv/bin/python -m pytest services/api/tests services/inference/tests/test_graph_sink.py -q
# Dashboard server
(cd services/dashboard/server && npm test)
Task 0: Branch and baseline
Files: none changed.
- [ ] Step 1: Check the working tree and coordinate with in-flight work
Run: git status --short
Expected: deploy/elasticsearch/retention.py and services/inference/src/logster_inference/engine.py are modified (user's uncommitted work), plus untracked files.
docs/superpowers/plans/2026-09-25-pipeline-overload-hardening.md is a separate plan written the same day, and it overlaps with this one:
- Its Task 7 (an ES-share guard in the old emergency_sweep, plus syncing the chart copy of retention.py) is superseded by this plan's Tasks 7 and 16. Skip it if this plan ships.
- Both plans edit deploy/docker-compose.yml, deploy/openshift/logster/values.yaml and the chart. Run the two plans one after the other, never in parallel, and rebase the second onto the first.
- The uncommitted engine.py edit (wl_dropped snapshot) belongs to that plan. Keep it. It currently breaks services/inference/tests/test_script_graph_integration.py, which is that plan's to resolve. Task 11 counts that test among the 8 recorded baseline failures.
- The uncommitted retention.py edit (_env_int warns instead of exiting) is superseded by Task 7's rewrite, whose retention_config.py has the same warn-and-default behaviour.
Ask the user which plan runs first before creating the branch. Do not stash or discard anything without the user's answer.
- [ ] Step 2: Create the branch (or worktree per superpowers:using-git-worktrees)
- [ ] Step 3: Record the baseline test results
venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_*.py' 2>&1 | tail -3
venv/bin/python -m pytest services/api/tests -q 2>&1 | tail -2
(cd services/dashboard/server && npm test 2>&1 | tail -5)
Task 1: eshttp.py and retention_config.py
Files:
- Create: deploy/elasticsearch/eshttp.py, deploy/elasticsearch/retention_config.py
- Test: deploy/elasticsearch/test_eshttp.py, deploy/elasticsearch/test_retention_config.py
Interfaces:
- Produces:
- eshttp.request(method, path, data=None, timeout=60) -> (status|None, body:str)
- eshttp.call(request_fn, method, path, body=None, timeout=30) -> (status|None, doc|None)
- eshttp.make_logger(prefix) -> log(*parts)
- eshttp.es_status(request_fn) -> status|None
- eshttp.fs_usage(request_fn) -> (used, total, pct)|None
- eshttp.iso_ms(ms) -> str
- eshttp.main(argv) -> int (operator CLI: python3 eshttp.py METHOD PATH [JSON])
- retention_config.FAMILIES, RetentionConfig (fields events_days, inferences_days, graphs_days, target_percent, min_age_hours, check_interval_minutes; methods days(family), describe())
- load_config(environ, log) -> RetentionConfig
- rollover_size_gb(total_fs_bytes) -> int|None
- DEFAULT_ROLLOVER_GB = 5, REMOVED_VARS, GIB
- [ ] Step 1: Write the failing tests
deploy/elasticsearch/test_eshttp.py:
"""eshttp helper tests; run with stdlib unittest discovery."""
import json
import unittest
import eshttp
def fixed(status, doc):
body = doc if isinstance(doc, str) else json.dumps(doc)
return lambda method, path, data=None, timeout=60: (status, body)
class TestCall(unittest.TestCase):
def test_parses_json_and_tolerates_garbage(self):
self.assertEqual(eshttp.call(fixed(200, {"a": 1}), "GET", "/x"), (200, {"a": 1}))
self.assertEqual(eshttp.call(fixed(500, "not json"), "GET", "/x"), (500, None))
self.assertEqual(eshttp.call(fixed(None, "refused"), "GET", "/x"), (None, None))
def test_serializes_body(self):
seen = {}
def req(method, path, data=None, timeout=60):
seen.update(method=method, data=data, timeout=timeout)
return 200, "{}"
eshttp.call(req, "PUT", "/x", {"k": [1]}, timeout=7)
self.assertEqual(seen, {"method": "PUT", "data": '{"k": [1]}', "timeout": 7})
class TestFsUsage(unittest.TestCase):
def test_used_is_total_minus_available(self):
reply = fixed(200, {"_nodes": {"failed": 0}, "nodes": {"fs": {
"total_in_bytes": 1000, "available_in_bytes": 300}}})
self.assertEqual(eshttp.fs_usage(reply), (700, 1000, 70.0))
def test_unknown_or_impossible_readings_are_none(self):
for reply in [fixed(None, "timeout"), fixed(500, {}), fixed(200, "invalid"),
fixed(200, {}),
fixed(200, {"_nodes": {"failed": 1}, "nodes": {"fs": {
"total_in_bytes": 1000, "available_in_bytes": 300}}}),
fixed(200, {"nodes": {"fs": {"total_in_bytes": 1000}}}),
fixed(200, {"nodes": {"fs": {"total_in_bytes": 1000,
"available_in_bytes": -1}}}),
fixed(200, {"nodes": {"fs": {"total_in_bytes": 1000,
"available_in_bytes": 1001}}}),
fixed(200, {"nodes": {"fs": {"total_in_bytes": True,
"available_in_bytes": 0}}})]:
with self.subTest(reply=reply):
self.assertIsNone(eshttp.fs_usage(reply))
class TestCli(unittest.TestCase):
def test_usage_error(self):
self.assertEqual(eshttp.main(["eshttp.py", "GET"]), 2)
def test_prints_body_and_maps_status_to_exit_code(self):
from unittest.mock import patch
with patch.object(eshttp, "request", return_value=(200, '{"ok":1}')) as req, \
patch("builtins.print") as out:
self.assertEqual(eshttp.main(["eshttp.py", "put", "/x", '{"a":1}']), 0)
req.assert_called_once_with("PUT", "/x", '{"a":1}')
out.assert_called_once_with('{"ok":1}')
with patch.object(eshttp, "request", return_value=(404, "{}")), patch("builtins.print"):
self.assertEqual(eshttp.main(["eshttp.py", "GET", "/missing"]), 1)
with patch.object(eshttp, "request", return_value=(None, "refused")), \
patch("builtins.print") as out:
self.assertEqual(eshttp.main(["eshttp.py", "GET", "/"]), 1)
out.assert_called_once_with("Elasticsearch unreachable: refused")
class TestMisc(unittest.TestCase):
def test_es_status(self):
self.assertEqual(eshttp.es_status(fixed(401, {})), 401)
self.assertIsNone(eshttp.es_status(fixed(None, "x")))
def test_iso_ms(self):
self.assertEqual(eshttp.iso_ms(0), "1970-01-01T00:00:00Z")
if __name__ == "__main__":
unittest.main()
deploy/elasticsearch/test_retention_config.py:
"""Retention config tests; run with stdlib unittest discovery."""
import unittest
import retention_config as rc
class TestLoadConfig(unittest.TestCase):
def load(self, env):
logs = []
cfg = rc.load_config(env, lambda *p: logs.append(" ".join(map(str, p))))
return cfg, logs
def test_defaults(self):
cfg, logs = self.load({})
self.assertEqual(cfg, rc.RetentionConfig(30, 90, 14, 75, 24, 5))
self.assertEqual(logs, [])
def test_valid_overrides(self):
cfg, _ = self.load({"RETENTION_EVENTS_DAYS": "7", "RETENTION_INFERENCES_DAYS": "0",
"RETENTION_GRAPHS_DAYS": "3", "RETENTION_TARGET_PERCENT": "60",
"RETENTION_MIN_AGE_HOURS": "6",
"RETENTION_CHECK_INTERVAL_MINUTES": "1"})
self.assertEqual(cfg, rc.RetentionConfig(7, 0, 3, 60, 6, 1))
self.assertEqual(cfg.days("inferences"), 0)
def test_messy_values_fall_back_with_warning(self):
cfg, logs = self.load({"RETENTION_EVENTS_DAYS": " 30 ",
"RETENTION_GRAPHS_DAYS": "1.5",
"RETENTION_INFERENCES_DAYS": "-1",
"RETENTION_TARGET_PERCENT": "99999",
"RETENTION_MIN_AGE_HOURS": "",
"RETENTION_CHECK_INTERVAL_MINUTES": "abc"})
self.assertEqual(cfg, rc.RetentionConfig(30, 90, 14, 75, 24, 5))
self.assertEqual(len([l for l in logs if l.startswith("WARNING")]), 4)
def test_target_must_stay_below_high_watermark(self):
cfg, logs = self.load({"RETENTION_TARGET_PERCENT": "90"})
self.assertEqual(cfg.target_percent, 75)
self.assertIn("outside 1..89", logs[0])
def test_removed_vars_warn(self):
_, logs = self.load({"RETENTION_EMERGENCY_THRESHOLD_PERCENT": "85",
"RETENTION_STRIP_RAW_EVENT_DAYS": "7",
"RETENTION_INTERVAL_HOURS": ""})
self.assertEqual(sorted(logs), [
"WARNING: RETENTION_EMERGENCY_THRESHOLD_PERCENT is no longer used (ILM retention); ignoring",
"WARNING: RETENTION_STRIP_RAW_EVENT_DAYS is no longer used (ILM retention); ignoring",
])
def test_describe(self):
self.assertEqual(rc.RetentionConfig().describe(),
"events=30d inferences=90d graphs=14d target=75% "
"min_age=24h check_interval=5m")
class TestRolloverSize(unittest.TestCase):
def test_one_percent_of_disk_clamped(self):
GIB = rc.GIB
self.assertEqual(rc.rollover_size_gb(15 * GIB), 1)
self.assertEqual(rc.rollover_size_gb(100 * GIB), 1)
self.assertEqual(rc.rollover_size_gb(403 * GIB), 4)
self.assertEqual(rc.rollover_size_gb(10 * 1024 * GIB), 50)
def test_unknown_size(self):
for value in (None, 0, -5, "100"):
self.assertIsNone(rc.rollover_size_gb(value))
if __name__ == "__main__":
unittest.main()
- [ ] Step 2: Run the tests to confirm they fail
Run: venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_eshttp.py' -v; venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_retention_config.py' -v
Expected: ModuleNotFoundError: No module named 'eshttp' / 'retention_config'.
- [ ] Step 3: Implement
deploy/elasticsearch/eshttp.py
"""Shared stdlib helpers for the Elasticsearch init / retention scripts.
These scripts run on logster-elasticsearch-init (deploy/elasticsearch/
init.Dockerfile: alpine + python3, no pip), so everything here is standard
library only. `request` never raises: it returns (status, body), and status
is None when Elasticsearch is unreachable. An HTTP error status is not a
transport failure — the body then carries Elasticsearch's error document.
"""
import base64
import datetime
import json
import math
import os
import sys
import urllib.error
import urllib.request
ES = os.environ.get("ES_HOST") or "http://elasticsearch:9200"
ES_USER = os.environ.get("ES_USERNAME") or "elastic"
ES_PASS = os.environ.get("ES_PASSWORD") or ""
def _headers():
h = {"Content-Type": "application/json"}
if ES_PASS:
token = base64.b64encode(f"{ES_USER}:{ES_PASS}".encode()).decode()
h["Authorization"] = "Basic " + token
return h
def request(method, path, data=None, timeout=60):
"""Return (status, body). status is None when ES is unreachable."""
if isinstance(data, str):
data = data.encode()
req = urllib.request.Request(ES + path, data=data, method=method,
headers=_headers())
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
return resp.status, resp.read().decode(errors="replace")
except urllib.error.HTTPError as exc:
return exc.code, exc.read().decode(errors="replace")
except Exception as exc: # URLError, socket timeout, connection refused
return None, str(exc)
def call(request_fn, method, path, body=None, timeout=30):
"""JSON convenience over a request function: returns (status, doc).
doc is the parsed JSON body, or None when the body is empty or not JSON.
"""
data = None if body is None else json.dumps(body)
status, raw = request_fn(method, path, data, timeout)
try:
doc = json.loads(raw) if raw else None
except ValueError:
doc = None
return status, doc
def make_logger(prefix):
def log(*parts):
stamp = datetime.datetime.now(datetime.timezone.utc).strftime(
"%Y-%m-%dT%H:%M:%SZ")
print(f"[{prefix}] {stamp} {' '.join(str(p) for p in parts)}",
flush=True)
return log
def es_status(request_fn):
"""HTTP status of GET /_cluster/health, or None when unreachable."""
status, _ = request_fn("GET", "/_cluster/health", None, 10)
return status
def fs_usage(request_fn):
"""Return (used, total, percent) of the ES data filesystem, or None.
used = total - available ("available", not "free": they differ by
reserved blocks, and ES's watermarks and the dashboard use available).
A failed node, missing field or impossible value returns None, so an
unknown reading can never authorize a deletion.
"""
status, doc = call(request_fn, "GET", "/_cluster/stats", timeout=10)
if status != 200 or not isinstance(doc, dict):
return None
try:
if doc.get("_nodes", {}).get("failed", 0):
return None
fs = doc["nodes"]["fs"]
total, available = fs["total_in_bytes"], fs["available_in_bytes"]
except (KeyError, TypeError, AttributeError):
return None
for value in (total, available):
if (not isinstance(value, (int, float)) or isinstance(value, bool)
or not math.isfinite(value)):
return None
if total <= 0 or not 0 <= available <= total:
return None
used = total - available
return used, total, used / total * 100
def iso_ms(ms):
return datetime.datetime.fromtimestamp(
ms / 1000, datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def main(argv):
"""Operator helper — the Elasticsearch image ships no curl:
oc exec deploy/elasticsearch-retention -- \\
python3 /templates/eshttp.py GET '/_cat/aliases/logster-*-write?v'
Prints the response body; exit 0 on HTTP < 400, 1 otherwise, 2 on usage.
"""
if len(argv) not in (3, 4):
print("usage: eshttp.py METHOD PATH [JSON_BODY]", file=sys.stderr)
return 2
status, body = request(argv[1].upper(), argv[2], argv[3] if len(argv) == 4 else None)
print(body if status is not None else f"Elasticsearch unreachable: {body}")
return 0 if status is not None and status < 400 else 1
if __name__ == "__main__":
sys.exit(main(sys.argv))
- [ ] Step 4: Implement
deploy/elasticsearch/retention_config.py
"""Retention knobs shared by init.py (ILM policies) and retention.py.
Every knob is an env var, so Compose and the Helm chart set them the same
way. A bad value never stops a worker: it logs a WARNING and uses the
default (spec §10).
"""
from dataclasses import dataclass
GIB = 2 ** 30
DEFAULT_ROLLOVER_GB = 5
FAMILIES = ("events", "inferences", "graphs")
REMOVED_VARS = (
"RETENTION_STRIP_RAW_EVENT_DAYS",
"RETENTION_INTERVAL_HOURS",
"RETENTION_EMERGENCY_THRESHOLD_PERCENT",
"RETENTION_CRITICAL_THRESHOLD_PERCENT",
)
# field, env var, default, minimum, maximum
_SPECS = (
("events_days", "RETENTION_EVENTS_DAYS", 30, 0, 3650),
("inferences_days", "RETENTION_INFERENCES_DAYS", 90, 0, 3650),
("graphs_days", "RETENTION_GRAPHS_DAYS", 14, 0, 3650),
("target_percent", "RETENTION_TARGET_PERCENT", 75, 1, 89),
("min_age_hours", "RETENTION_MIN_AGE_HOURS", 24, 1, 720),
("check_interval_minutes", "RETENTION_CHECK_INTERVAL_MINUTES", 5, 1, 1440),
)
@dataclass(frozen=True)
class RetentionConfig:
events_days: int = 30
inferences_days: int = 90
graphs_days: int = 14
target_percent: int = 75
min_age_hours: int = 24
check_interval_minutes: int = 5
def days(self, family):
return getattr(self, f"{family}_days")
def describe(self):
return (f"events={self.events_days}d inferences={self.inferences_days}d "
f"graphs={self.graphs_days}d target={self.target_percent}% "
f"min_age={self.min_age_hours}h "
f"check_interval={self.check_interval_minutes}m")
def load_config(environ, log):
values = {}
for field_name, var, default, low, high in _SPECS:
raw = (environ.get(var) or "").strip()
if not raw:
values[field_name] = default
continue
try:
value = int(raw)
except ValueError:
log(f"WARNING: {var}={raw!r} is not an integer; using default {default}")
values[field_name] = default
continue
if not low <= value <= high:
log(f"WARNING: {var}={value} is outside {low}..{high}; "
f"using default {default}")
value = default
values[field_name] = value
for var in REMOVED_VARS:
if (environ.get(var) or "").strip():
log(f"WARNING: {var} is no longer used (ILM retention); ignoring")
return RetentionConfig(**values)
def rollover_size_gb(total_fs_bytes):
"""Max primary shard size per rollover index: ~1% of the filesystem in
whole GiB, clamped to 1..50 (spec D12). None when the size is unknown."""
if (not isinstance(total_fs_bytes, (int, float))
or isinstance(total_fs_bytes, bool) or total_fs_bytes <= 0):
return None
return max(1, min(50, int(total_fs_bytes // 100 // GIB)))
- [ ] Step 5: Run the tests to confirm they pass
Run: venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_eshttp.py' -v && venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_retention_config.py' -v
Expected: all OK.
- [ ] Step 6: Stage
git add deploy/elasticsearch/eshttp.py deploy/elasticsearch/retention_config.py deploy/elasticsearch/test_eshttp.py deploy/elasticsearch/test_retention_config.py
feat(retention): shared ES helpers and validated retention config
Task 2: In-memory Elasticsearch double for unit tests
Files:
- Create: deploy/elasticsearch/fake_es.py
- Test: deploy/elasticsearch/test_fake_es.py
Interfaces:
- Produces:
- FakeES(now_ms=1_800_000_000_000, total=100*GIB, available=60*GIB), callable as a request function.
- Attributes: indices, policies, templates, cluster_settings, calls, fail (list of (method, path_prefix, (status, body))), reachable, status_override, nodes_failed, shards_failed, on_create.
- Methods: add_index(name, docs=(), store=0, settings=None, aliases=None, created_ms=now), mutations(), writes_to(prefix).
- FakeIndex fields: name, docs, store, settings (flat str values), aliases, ilm.
- Constants: HOUR_MS, DAY_MS, GIB.
- [ ] Step 1: Write the failing test
deploy/elasticsearch/test_fake_es.py:
"""Sanity tests for the ES double itself, so layout/diskcap tests can trust it."""
import json
import unittest
from fake_es import FakeES, GIB
def body(reply):
return json.loads(reply[1])
class TestFakeES(unittest.TestCase):
def setUp(self):
self.es = FakeES()
def test_missing_concrete_is_404_but_missing_wildcard_is_empty(self):
self.assertEqual(self.es("GET", "/nope/_settings")[0], 404)
self.assertEqual(body(self.es("GET", "/_cat/indices/logster-*?format=json")), [])
self.assertEqual(self.es("GET", "/_alias/logster-events-write")[0], 404)
self.assertEqual(self.es("GET", "/_alias/logster-*-write"), (200, "{}"))
def test_create_enforces_single_write_index_and_existence(self):
alias = {"aliases": {"a": {"is_write_index": True}}}
self.assertEqual(self.es("PUT", "/i-1", json.dumps(alias))[0], 200)
self.assertEqual(self.es("PUT", "/i-1", "{}")[0], 400)
reply = self.es("PUT", "/i-2", json.dumps(alias))
self.assertEqual(body(reply)["error"]["type"], "illegal_argument_exception")
def test_template_settings_apply_on_create(self):
self.es("PUT", "/_index_template/t", json.dumps({
"index_patterns": ["logster-x*"], "priority": 200,
"template": {"settings": {"index.codec": "best_compression"}}}))
self.es("PUT", "/logster-x-000001", "{}")
self.assertEqual(self.es.indices["logster-x-000001"].settings["index.codec"],
"best_compression")
def test_policy_is_normalized_but_meta_kept(self):
self.es("PUT", "/_ilm/policy/p", json.dumps({"policy": {
"_meta": {"logster_hash": "h"},
"phases": {"delete": {"min_age": "1d", "actions": {"delete": {}}}}}}))
stored = body(self.es("GET", "/_ilm/policy/p"))["p"]["policy"]
self.assertEqual(stored["_meta"], {"logster_hash": "h"})
self.assertTrue(stored["phases"]["delete"]["actions"]["delete"]["delete_searchable_snapshot"])
def test_delete_requires_exact_name(self):
self.es.add_index("logster-events-000001")
self.assertEqual(self.es("DELETE", "/logster-events-*")[0], 400)
self.assertEqual(self.es("DELETE", "/logster-events-000001")[0], 200)
self.assertEqual(self.es("DELETE", "/logster-events-000001")[0], 404)
def test_flood_block_rejects_settings_writes(self):
self.es.add_index("i", settings={"index.blocks.read_only_allow_delete": True})
self.assertEqual(self.es("PUT", "/i/_settings", '{"x": 1}')[0], 429)
def test_terms_agg_truncates_at_size(self):
for n in range(12):
self.es.add_index(f"logster-events-{n:06d}", docs=[self.es.now_ms])
reply = body(self.es("POST", "/logster-events*/_search", json.dumps({
"size": 0, "aggs": {"t": {"terms": {"field": "_index"}}}})))
self.assertEqual(len(reply["aggregations"]["t"]["buckets"]), 10)
self.assertEqual(reply["aggregations"]["t"]["sum_other_doc_count"], 2)
def test_cluster_stats(self):
fs = body(self.es("GET", "/_cluster/stats"))["nodes"]["fs"]
self.assertEqual(fs, {"total_in_bytes": 100 * GIB, "available_in_bytes": 60 * GIB})
if __name__ == "__main__":
unittest.main()
- [ ] Step 2: Run it to confirm it fails
Run: venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_fake_es.py' -v
Expected: ModuleNotFoundError: No module named 'fake_es'.
- [ ] Step 3: Implement
deploy/elasticsearch/fake_es.py
"""In-memory double for the slice of the Elasticsearch REST API that
layout.py, diskcap.py, retention.py and init.py use.
Test helper only — never shipped (test_chart_drift.SHIPPED_SCRIPTS lists
what the chart ships). It mirrors ES 9.5 where the scripts depend on it:
404 for a missing concrete name but 200-empty for an unmatched wildcard,
one write index per alias, resource_already_exists on create,
destructive_requires_name on DELETE, read_only_allow_delete rejecting
settings writes, and _meta kept verbatim while policy bodies are normalized.
"""
import copy
import fnmatch
import json
import re
from urllib.parse import parse_qs, urlsplit
HOUR_MS = 3_600_000
DAY_MS = 24 * HOUR_MS
GIB = 2 ** 30
def _err(status, etype, reason=""):
return status, json.dumps({"error": {"type": etype, "reason": reason},
"status": status})
def _ok(doc=None):
return 200, json.dumps({"acknowledged": True} if doc is None else doc)
def _flat(value):
if isinstance(value, bool):
return "true" if value else "false"
return str(value)
def _wild(expr):
return any(c in expr for c in "*?")
class FakeIndex:
def __init__(self, name, docs=(), store=0, settings=None, aliases=None,
created_ms=0):
self.name = name
self.docs = list(docs) # @timestamp values, epoch ms
self.store = store # bytes reported as store.size
self.settings = {"index.creation_date": str(created_ms)}
self.settings.update({k: _flat(v) for k, v in (settings or {}).items()})
self.aliases = {a: dict(m or {}) for a, m in (aliases or {}).items()}
self.ilm = {"phase": "hot", "action": "complete", "step": "complete"}
class FakeES:
def __init__(self, now_ms=1_800_000_000_000, total=100 * GIB,
available=60 * GIB):
self.now_ms = now_ms
self.total, self.available = total, available
self.nodes_failed = 0
self.shards_failed = 0
self.reachable = True
self.status_override = None
self.indices = {}
self.policies = {}
self.templates = {}
self.cluster_settings = {}
self.calls = []
self.fail = []
self.on_create = None
# ── test helpers ─────────────────────────────────────────────────
def add_index(self, name, **kw):
kw.setdefault("created_ms", self.now_ms)
self.indices[name] = FakeIndex(name, **kw)
return self.indices[name]
def mutations(self):
return [(m, p) for m, p, _ in self.calls
if m in ("PUT", "DELETE") or (m == "POST" and "/_search" not in p)]
def writes_to(self, prefix):
return [(m, p) for m, p in self.mutations() if p.startswith(prefix)]
# ── transport ────────────────────────────────────────────────────
def __call__(self, method, path, data=None, timeout=60):
body = json.loads(data) if data else None
self.calls.append((method, path, body))
if not self.reachable:
return None, "connection refused"
if self.status_override:
return _err(self.status_override, "security_exception")
for m, prefix, reply in self.fail:
if m == method and path.startswith(prefix):
return reply
url = urlsplit(path)
parts = [p for p in url.path.split("/") if p]
query = {k: v[-1] for k, v in parse_qs(url.query).items()}
try:
return self._route(method, parts, query, body)
except KeyError as exc:
return _err(404, "index_not_found_exception",
f"no such index [{exc.args[0]}]")
def _route(self, method, parts, q, body):
head = parts[0] if parts else ""
if head == "_cluster":
return self._cluster(method, parts[1:], body)
if head == "_ilm" and parts[1] == "policy":
return self._policy(method, parts[2], body)
if head == "_index_template":
return self._template(method, parts[1], body)
if head == "_alias":
return self._alias(parts[1])
if head == "_cat" and parts[1] == "indices":
return self._cat(parts[2] if len(parts) > 2 else "*")
if len(parts) == 1:
if method == "PUT":
return self._create(parts[0], body or {})
if method == "DELETE":
return self._delete(parts[0])
return _err(400, "unsupported", parts[0])
target, action = parts[0], parts[1]
if action == "_settings":
return self._settings(method, target, body or {})
if action == "_search":
return self._search(target, q, body or {})
if action == "_ilm" and parts[2] == "explain":
return self._explain(target, q)
if action == "_ilm" and parts[2] == "retry":
return self._retry(target)
return _err(400, "unsupported", "/".join(parts))
def _resolve(self, expr, ignore_unavailable=False):
out = []
for part in expr.split(","):
if _wild(part):
out += [n for n in sorted(self.indices) if fnmatch.fnmatchcase(n, part)]
elif part in self.indices:
out.append(part)
else:
members = [n for n, i in sorted(self.indices.items()) if part in i.aliases]
if not members and not ignore_unavailable:
raise KeyError(part)
out += members
return list(dict.fromkeys(out))
def _cluster(self, method, parts, body):
what = parts[0] if parts else ""
if what == "health":
if len(parts) > 1:
self._resolve(parts[1])
return 200, json.dumps({"status": "green", "timed_out": False})
if what == "stats":
return 200, json.dumps({
"_nodes": {"failed": self.nodes_failed},
"nodes": {"fs": {"total_in_bytes": self.total,
"available_in_bytes": self.available}}})
if what == "settings" and method == "PUT":
self.cluster_settings.update((body or {}).get("persistent", {}))
return _ok()
if what == "reroute":
return _ok()
return _err(400, "unsupported")
def _policy(self, method, name, body):
if method == "GET":
if name not in self.policies:
return _err(404, "resource_not_found_exception")
return 200, json.dumps({name: {"version": 1, "policy": self.policies[name]}})
stored = copy.deepcopy(body["policy"])
for phase in stored.get("phases", {}).values(): # ES normalizes on write
phase.setdefault("min_age", "0ms")
if "delete" in phase.get("actions", {}):
phase["actions"]["delete"].setdefault("delete_searchable_snapshot", True)
self.policies[name] = stored
return _ok()
def _template(self, method, name, body):
if method == "GET":
if name not in self.templates:
return _err(404, "resource_not_found_exception")
return 200, json.dumps({"index_templates": [
{"name": name, "index_template": self.templates[name]}]})
self.templates[name] = copy.deepcopy(body)
return _ok()
def _alias(self, expr):
patterns = expr.split(",")
out = {}
for name, idx in sorted(self.indices.items()):
hits = {a: m for a, m in idx.aliases.items()
if any(fnmatch.fnmatchcase(a, p) for p in patterns)}
if hits:
out[name] = {"aliases": hits}
if not out and not _wild(expr):
return 404, json.dumps({"error": f"alias [{expr}] missing", "status": 404})
return 200, json.dumps(out)
def _cat(self, expr):
rows = [{"index": n, "health": "green",
"docs.count": str(len(self.indices[n].docs)),
"store.size": str(self.indices[n].store)}
for n in self._resolve(expr)]
return 200, json.dumps(rows)
def _create(self, name, body):
if self.on_create:
self.on_create(name)
if name in self.indices:
return _err(400, "resource_already_exists_exception",
f"index [{name}] already exists")
if any(name in i.aliases for i in self.indices.values()):
return _err(400, "invalid_index_name_exception", "an alias has that name")
for alias, meta in (body.get("aliases") or {}).items():
if (meta or {}).get("is_write_index") and any(
(i.aliases.get(alias) or {}).get("is_write_index")
for i in self.indices.values()):
return _err(400, "illegal_argument_exception",
f"alias [{alias}] has more than one write index")
matches = [t for t in self.templates.values()
if any(fnmatch.fnmatchcase(name, p) for p in t.get("index_patterns", []))]
settings = {}
if matches:
best = max(matches, key=lambda t: t.get("priority", 0))
settings.update((best.get("template") or {}).get("settings") or {})
settings.update(body.get("settings") or {})
self.add_index(name, settings=settings, aliases=body.get("aliases"))
return _ok({"acknowledged": True, "shards_acknowledged": True, "index": name})
def _delete(self, expr):
if _wild(expr) or "," in expr or expr == "_all":
return _err(400, "illegal_argument_exception",
"Wildcard expressions or all indices are not allowed")
if expr not in self.indices:
return _err(404, "index_not_found_exception")
del self.indices[expr]
return _ok()
def _settings(self, method, target, body):
names = self._resolve(target)
if method == "GET":
return 200, json.dumps({n: {"settings": dict(self.indices[n].settings)}
for n in names})
blocked = [n for n in names if self.indices[n].settings.get(
"index.blocks.read_only_allow_delete") == "true"]
if blocked:
return _err(429, "cluster_block_exception",
f"index [{blocked[0]}] blocked by: [TOO_MANY_REQUESTS/12/"
"disk usage exceeded flood-stage watermark]")
flat = ({f"index.{k}": v for k, v in body["index"].items()}
if isinstance(body.get("index"), dict) else body)
for n in names:
for key, value in flat.items():
self.indices[n].settings[key] = _flat(value)
return _ok()
def _search(self, target, q, body):
names = self._resolve(target, ignore_unavailable=q.get("ignore_unavailable") == "true")
out = {"timed_out": False,
"_shards": {"total": len(names), "failed": self.shards_failed},
"hits": {"hits": []}, "aggregations": {}}
for key, spec in (body.get("aggs") or {}).items():
out["aggregations"][key] = self._agg(spec, names)
return 200, json.dumps(out)
def _agg(self, spec, names, keep=None):
pairs = [(n, ts) for n in names for ts in self.indices[n].docs
if keep is None or keep(ts)]
if "max" in spec:
return {"value": max((ts for _, ts in pairs), default=None)}
if "filter" in spec:
gte = spec["filter"]["range"]["@timestamp"]["gte"]
low = self.now_ms - int(re.fullmatch(r"now-(\d+)h", gte).group(1)) * HOUR_MS
inner = {k: self._agg(v, names, lambda ts: ts >= low)
for k, v in (spec.get("aggs") or {}).items()}
return {"doc_count": sum(1 for _, ts in pairs if ts >= low), **inner}
if "terms" in spec:
counts = {}
for n, _ in pairs:
counts[n] = counts.get(n, 0) + 1
ordered = sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))
size = spec["terms"].get("size", 10)
buckets = []
for n, c in ordered[:size]:
bucket = {"key": n, "doc_count": c}
for k, v in (spec.get("aggs") or {}).items():
bucket[k] = self._agg(v, [n], keep)
buckets.append(bucket)
return {"buckets": buckets,
"sum_other_doc_count": sum(c for _, c in ordered[size:])}
raise ValueError(f"unsupported aggregation {spec}")
def _explain(self, target, q):
out = {}
for n in self._resolve(target):
idx = self.indices[n]
managed = "index.lifecycle.name" in idx.settings
info = {"index": n, "managed": managed}
if managed:
info.update(idx.ilm, policy=idx.settings["index.lifecycle.name"])
if q.get("only_errors") == "true" and info.get("step") != "ERROR":
continue
out[n] = info
return 200, json.dumps({"indices": out})
def _retry(self, name):
idx = self.indices.get(name)
if idx is None:
raise KeyError(name)
if idx.ilm.get("step") != "ERROR":
return _err(400, "illegal_argument_exception", "not in an error step")
idx.ilm = {"phase": idx.ilm.get("phase"), "action": idx.ilm.get("action"),
"step": idx.ilm.get("failed_step", "check-rollover-ready")}
return _ok()
- [ ] Step 4: Run it to confirm it passes
Run: venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_fake_es.py' -v
Expected: 8 tests OK.
- [ ] Step 5: Stage
test(retention): in-memory ES double for layout and disk-rule tests
Task 3: layout.py, part 1: ILM policies, index templates, template files
Files:
- Create: deploy/elasticsearch/layout.py
- Modify: deploy/elasticsearch/logster-events.template.json, logster-inferences.template.json, logster-graphs.template.json
- Test: deploy/elasticsearch/test_layout.py, deploy/elasticsearch/test_templates.py
Interfaces:
- Consumes (Task 1): eshttp.call, eshttp.fs_usage, retention_config.rollover_size_gb, DEFAULT_ROLLOVER_GB, FAMILIES
- Produces:
- Constants: FAMILIES, ROLLOVER_RE, MANAGED_RE, READ_PATTERNS ("logster-events*,logster-inferences*,logster-graphs*"), ROLLOVER_PATTERNS
- Name helpers: write_alias(f), legacy_index(f), policy_name(f), legacy_policy_name(f)
- Pure helpers: canonical_hash(body) -> str, family_policy(days, size_gb) -> dict, legacy_policy(days) -> dict, load_templates(dir) -> {name: body}
- ConvergeResult(fatal: list, problems: list, changes: list)
- converge(request, log, cfg, template_dir) -> ConvergeResult (never raises)
- [ ] Step 1: Write the failing tests
deploy/elasticsearch/test_templates.py:
"""The managed index templates carry the ILM wiring and sort mappings (spec §6.2)."""
import json
import unittest
from pathlib import Path
HERE = Path(__file__).resolve().parent
def load(name):
return json.loads((HERE / f"{name}.template.json").read_text())
class TestManagedTemplates(unittest.TestCase):
def test_lifecycle_wiring_and_codec(self):
for family in ("events", "inferences", "graphs"):
with self.subTest(family=family):
t = load(f"logster-{family}")
self.assertEqual(t["index_patterns"], [f"logster-{family}*"])
s = t["template"]["settings"]
self.assertEqual(s["index.lifecycle.name"], f"logster-{family}")
self.assertEqual(s["index.lifecycle.rollover_alias"], f"logster-{family}-write")
self.assertEqual(s["index.codec"], "best_compression")
self.assertEqual(s["index.number_of_replicas"], 0)
self.assertEqual(t["template"]["mappings"]["properties"]["@timestamp"],
{"type": "date"})
def test_events_timestamp_sort_field_is_mapped(self):
props = load("logster-events")["template"]["mappings"]["properties"]
self.assertEqual(props["timestamp"], {"type": "float"})
def test_graphs_keep_their_strict_mapping(self):
mappings = load("logster-graphs")["template"]["mappings"]
self.assertEqual(mappings["dynamic"], "false")
self.assertEqual(mappings["properties"]["inference_id"], {"type": "keyword"})
def test_no_template_file_carries_meta(self):
for path in HERE.glob("*.template.json"):
self.assertNotIn("_meta", json.loads(path.read_text()), path.name)
if __name__ == "__main__":
unittest.main()
deploy/elasticsearch/test_layout.py:
"""layout.converge tests against the in-memory ES double (spec §6-§7.1)."""
import json
import tempfile
import unittest
from pathlib import Path
import layout
from fake_es import DAY_MS, GIB, FakeES
from retention_config import RetentionConfig
HERE = Path(__file__).resolve().parent
ALL_POLICIES = {f"logster-{f}{s}" for f in ("events", "inferences", "graphs")
for s in ("", "-legacy")}
ALL_TEMPLATES = {p.name[: -len(".template.json")] for p in HERE.glob("*.template.json")}
class LayoutCase(unittest.TestCase):
def setUp(self):
self.es = FakeES()
self.logs = []
def log(self, *parts):
self.logs.append(" ".join(map(str, parts)))
def converge(self, cfg=None):
return layout.converge(self.es, self.log, cfg or RetentionConfig(), str(HERE))
class TestPolicyBodies(unittest.TestCase):
def test_family_policy_rolls_over_daily_and_deletes_after_days(self):
self.assertEqual(layout.family_policy(30, 7), {"phases": {
"hot": {"min_age": "0ms", "actions": {"rollover": {
"max_age": "1d", "max_primary_shard_size": "7gb"}}},
"delete": {"min_age": "30d", "actions": {"delete": {}}}}})
def test_zero_days_keeps_forever(self):
self.assertNotIn("delete", layout.family_policy(0, 1)["phases"])
self.assertEqual(layout.legacy_policy(0),
{"phases": {"hot": {"min_age": "0ms", "actions": {}}}})
self.assertEqual(layout.legacy_policy(14), {"phases": {"delete": {
"min_age": "14d", "actions": {"delete": {}}}}})
def test_canonical_hash_ignores_key_order(self):
self.assertEqual(layout.canonical_hash({"a": 1, "b": [1, {"c": 2, "d": 3}]}),
layout.canonical_hash({"b": [1, {"d": 3, "c": 2}], "a": 1}))
def test_load_templates_strips_meta(self):
with tempfile.TemporaryDirectory() as tmp:
Path(tmp, "x.template.json").write_text(json.dumps(
{"index_patterns": ["x*"], "_meta": {"logster_hash": "old"}}))
Path(tmp, "ignored.json").write_text("{}")
self.assertEqual(layout.load_templates(tmp), {"x": {"index_patterns": ["x*"]}})
class TestPoliciesAndTemplates(LayoutCase):
def test_fresh_cluster_gets_all_policies_and_templates(self):
result = self.converge()
self.assertEqual(result.fatal, [])
self.assertEqual(set(self.es.policies), ALL_POLICIES)
self.assertEqual(set(self.es.templates), ALL_TEMPLATES)
self.assertEqual(self.es.policies["logster-events"]["_meta"]["logster_hash"],
layout.canonical_hash(layout.family_policy(30, 1)))
self.assertEqual(self.es.templates["logster-events"]["_meta"]["logster_hash"],
layout.canonical_hash(layout.load_templates(str(HERE))["logster-events"]))
def test_second_run_is_a_no_op(self):
self.converge()
before = len(self.es.mutations())
self.converge()
self.assertEqual([m for m in self.es.mutations()[before:]
if "/_ilm/policy/" in m[1] or "/_index_template/" in m[1]], [])
def test_changed_days_rewrites_only_that_familys_policies(self):
self.converge()
before = len(self.es.mutations())
self.converge(RetentionConfig(events_days=7))
puts = sorted(p for _, p in self.es.mutations()[before:] if p.startswith("/_ilm/policy/"))
self.assertEqual(puts, ["/_ilm/policy/logster-events",
"/_ilm/policy/logster-events-legacy"])
self.assertEqual(self.es.policies["logster-events"]["phases"]["delete"]["min_age"], "7d")
def test_size_cap_follows_disk_size(self):
for total, expect in ((400 * GIB, "4gb"), (15 * GIB, "1gb"), (10 * 1024 * GIB, "50gb")):
with self.subTest(total=total):
self.es = FakeES(total=total, available=total // 2)
self.converge()
rollover = self.es.policies["logster-graphs"]["phases"]["hot"]["actions"]["rollover"]
self.assertEqual(rollover["max_primary_shard_size"], expect)
def test_unknown_disk_size_uses_default_then_keeps_existing(self):
self.es.nodes_failed = 1
self.converge()
rollover = self.es.policies["logster-events"]["phases"]["hot"]["actions"]["rollover"]
self.assertEqual(rollover["max_primary_shard_size"], "5gb")
before = len(self.es.mutations())
self.converge()
self.assertNotIn(("PUT", "/_ilm/policy/logster-events"), self.es.mutations()[before:])
def test_rejected_policy_put_is_fatal(self):
self.es.fail.append(("PUT", "/_ilm/policy/logster-events", (400, json.dumps(
{"error": {"type": "x_content_parse_exception", "reason": "bad"}}))))
result = self.converge()
self.assertTrue(any("policy logster-events: PUT rejected (HTTP 400) "
"x_content_parse_exception" in f for f in result.fatal))
def test_unreachable_es_is_fatal_not_an_exception(self):
self.es.reachable = False
self.assertTrue(self.converge().fatal)
if __name__ == "__main__":
unittest.main()
- [ ] Step 2: Run them to confirm they fail
Run: venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_templates.py' -v; venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_layout.py' -v
Expected: template assertions fail (KeyError: 'index.lifecycle.name'), and ModuleNotFoundError: No module named 'layout'.
- [ ] Step 3: Update the three managed templates
deploy/elasticsearch/logster-events.template.json:
{
"index_patterns": ["logster-events*"],
"priority": 200,
"template": {
"settings": {
"index.number_of_shards": 1,
"index.number_of_replicas": 0,
"index.refresh_interval": "5s",
"index.codec": "best_compression",
"index.lifecycle.name": "logster-events",
"index.lifecycle.rollover_alias": "logster-events-write"
},
"mappings": {
"properties": {
"@timestamp": { "type": "date" },
"timestamp": { "type": "float" }
}
}
}
}
deploy/elasticsearch/logster-inferences.template.json:
{
"index_patterns": ["logster-inferences*"],
"priority": 200,
"template": {
"settings": {
"index.number_of_shards": 1,
"index.number_of_replicas": 0,
"index.refresh_interval": "5s",
"index.codec": "best_compression",
"index.lifecycle.name": "logster-inferences",
"index.lifecycle.rollover_alias": "logster-inferences-write"
},
"mappings": {
"properties": {
"@timestamp": { "type": "date" }
}
}
}
}
deploy/elasticsearch/logster-graphs.template.json: in the settings object only, add these three keys after "index.mapping.total_fields.limit": 3000. Leave mappings unchanged; it already maps @timestamp as date.
"index.mapping.total_fields.limit": 3000,
"index.codec": "best_compression",
"index.lifecycle.name": "logster-graphs",
"index.lifecycle.rollover_alias": "logster-graphs-write"
- [ ] Step 4: Implement
deploy/elasticsearch/layout.py
#!/usr/bin/env python3
"""Declarative Elasticsearch layout for Logster (spec §6-§7.1): ILM
policies, index templates, adoption of pre-ILM ("legacy") indices, and the
rollover write aliases.
converge() is idempotent, safe to run from init.py and the retention worker
at the same time, and never deletes anything. Stdlib only (see eshttp.py).
"""
import glob
import hashlib
import json
import os
import re
import sys
import time
from dataclasses import dataclass, field
import eshttp
import retention_config
FAMILIES = retention_config.FAMILIES
ROLLOVER_RE = re.compile(r"^logster-(events|inferences|graphs)-(\d{6,})$")
# The only index names the retention worker may ever delete.
MANAGED_RE = re.compile(r"^logster-(events|inferences|graphs)(-\d{6,})?$")
READ_PATTERNS = ",".join(f"logster-{f}*" for f in FAMILIES)
ROLLOVER_PATTERNS = ",".join(f"logster-{f}-*" for f in FAMILIES)
def write_alias(family):
return f"logster-{family}-write"
def legacy_index(family):
return f"logster-{family}"
def policy_name(family):
return f"logster-{family}"
def legacy_policy_name(family):
return f"logster-{family}-legacy"
def canonical_hash(body):
blob = json.dumps(body, sort_keys=True, separators=(",", ":")).encode()
return hashlib.sha256(blob).hexdigest()
def family_policy(days, size_gb):
phases = {"hot": {"min_age": "0ms", "actions": {"rollover": {
"max_age": "1d", "max_primary_shard_size": f"{size_gb}gb"}}}}
if days > 0:
phases["delete"] = {"min_age": f"{days}d", "actions": {"delete": {}}}
return {"phases": phases}
def legacy_policy(days):
if days > 0:
return {"phases": {"delete": {"min_age": f"{days}d",
"actions": {"delete": {}}}}}
# 0 = keep forever: a do-nothing policy keeps adoption uniform.
return {"phases": {"hot": {"min_age": "0ms", "actions": {}}}}
def load_templates(template_dir):
"""{name: body} for every *.template.json, with any `_meta` removed."""
out = {}
for path in sorted(glob.glob(os.path.join(template_dir, "*.template.json"))):
name = os.path.basename(path)[: -len(".template.json")]
with open(path, encoding="utf-8") as fh:
body = json.load(fh)
body.pop("_meta", None)
out[name] = body
return out
@dataclass
class ConvergeResult:
fatal: list = field(default_factory=list) # layout not established
problems: list = field(default_factory=list) # degraded; retried next run
changes: list = field(default_factory=list) # what this run changed
@dataclass
class _Ctx:
request: object
log: object
cfg: object
template_dir: str
size_gb: object
result: ConvergeResult
def converge(request, log, cfg, template_dir):
"""Bring the cluster to the desired layout. Never raises, never deletes."""
result = ConvergeResult()
usage = eshttp.fs_usage(request)
ctx = _Ctx(request, log, cfg, template_dir,
retention_config.rollover_size_gb(usage[1]) if usage else None,
result)
for step in (_converge_policies, _converge_templates):
try:
step(ctx)
except Exception as exc: # noqa: BLE001 — one broken step must not hide the rest
result.fatal.append(f"{step.__name__} crashed: {exc!r}")
for change in result.changes:
log(f"layout: {change}")
return result
def _reason(doc):
err = doc.get("error") if isinstance(doc, dict) else None
if isinstance(err, dict):
return f"{err.get('type', '')}: {str(err.get('reason', ''))[:200]}"
return str(err or "")[:200]
def _stored_hash(request, get_path, extract):
"""Look up the logster_hash in an object's _meta.
Returns ("missing", None), ("present", hash-or-None) or
("error", http_status).
"""
status, doc = eshttp.call(request, "GET", get_path)
if status == 404:
return "missing", None
if status != 200 or not isinstance(doc, dict):
return "error", status
try:
meta = extract(doc) or {}
except (KeyError, IndexError, TypeError, AttributeError):
return "present", None
return "present", meta.get("logster_hash")
def _converge_policies(ctx):
for family in FAMILIES:
days = ctx.cfg.days(family)
for name, kind in ((policy_name(family), "family"),
(legacy_policy_name(family), "legacy")):
state, stored = _stored_hash(
ctx.request, f"/_ilm/policy/{name}",
lambda doc, n=name: doc[n]["policy"].get("_meta"))
if state == "error":
ctx.result.fatal.append(f"policy {name}: GET failed (HTTP {stored})")
continue
if kind == "family":
size = ctx.size_gb
if size is None:
if state == "present":
continue # disk size unknown: keep the existing policy
size = retention_config.DEFAULT_ROLLOVER_GB
body = family_policy(days, size)
else:
body = legacy_policy(days)
digest = canonical_hash(body)
if stored == digest:
continue
payload = {"policy": {**body, "_meta": {"managed_by": "logster",
"logster_hash": digest}}}
status, doc = eshttp.call(ctx.request, "PUT", f"/_ilm/policy/{name}", payload)
if status != 200:
ctx.result.fatal.append(
f"policy {name}: PUT rejected (HTTP {status}) {_reason(doc)}")
else:
ctx.result.changes.append(
f"policy {name} {'created' if state == 'missing' else 'updated'}")
def _converge_templates(ctx):
templates = load_templates(ctx.template_dir)
if not templates:
ctx.result.fatal.append(f"no *.template.json found in {ctx.template_dir}")
return
for name, body in templates.items():
state, stored = _stored_hash(
ctx.request, f"/_index_template/{name}",
lambda doc: doc["index_templates"][0]["index_template"].get("_meta"))
if state == "error":
ctx.result.fatal.append(f"template {name}: GET failed (HTTP {stored})")
continue
digest = canonical_hash(body)
if stored == digest:
continue
payload = {**body, "_meta": {"managed_by": "logster", "logster_hash": digest}}
status, doc = eshttp.call(ctx.request, "PUT", f"/_index_template/{name}", payload)
if status != 200:
ctx.result.fatal.append(
f"template {name}: PUT rejected (HTTP {status}) {_reason(doc)}")
else:
ctx.result.changes.append(
f"template {name} {'created' if state == 'missing' else 'updated'}")
sys and time are used by Task 4. Leave them in so Task 4 only adds functions. Dates are formatted with eshttp.iso_ms, so layout.py does not import datetime.)
- [ ] Step 5: Run the tests to confirm they pass
Run: venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_templates.py' -v && venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_layout.py' -v
Expected: all OK.
- [ ] Step 6: Stage
git add deploy/elasticsearch/layout.py deploy/elasticsearch/test_layout.py deploy/elasticsearch/test_templates.py deploy/elasticsearch/logster-events.template.json deploy/elasticsearch/logster-inferences.template.json deploy/elasticsearch/logster-graphs.template.json
feat(retention): converge ILM policies and index templates by content hash
Task 4: layout.py, part 2: legacy adoption, write aliases, orphans, wait
Files:
- Modify: deploy/elasticsearch/layout.py
- Test: deploy/elasticsearch/test_layout.py (append classes)
Interfaces:
- Consumes (Task 3): _Ctx, _reason, the name helpers, ROLLOVER_RE, ROLLOVER_PATTERNS
- Produces:
- alias_members(request, alias) -> ("ok", {index: is_write_flag}) | ("missing", {}) | ("error", status)
- writer_of(members) -> index|None (explicit is_write_index: true, else a lone member with no flag)
- _alias_state(request, family) -> "ready" | "squatted" | "waiting": "ready" by the same rule as converge() (alias_members returns "ok" and writer_of(members) is not None, so a lone member with no is_write_index flag counts, exactly as ES writes to it); "squatted" when the alias is missing and a concrete index holds its name (writes land there)
- write_aliases_ready(request) -> bool: no family is "waiting" (a squatted family counts as ready), so the Vector initContainer and converge() can never disagree
- squatter_message(alias) -> str: the SQUATTER text, shared by converge() (a problem every check) and wait_for_aliases (logged as ERROR: … once per alias)
- Legacy adoption writes index.lifecycle.name: logster-<family>-legacy, index.lifecycle.origination_date and index.lifecycle.indexing_complete: true. The last is needed because a legacy-named index created while rolled back to v1.2.1 picks up the new family template, and ILM's hot/rollover step then errors on it forever (real ES 9.5.4 probe); on a normal legacy index it is a no-op
- _release_orphans treats HTTP 400 from _ilm/retry as success: the index is no longer in an ERROR step because ILM's own auto-retry got there first
- wait_for_aliases(request, log, poll_seconds=5, log_every_seconds=60, sleep=time.sleep, max_wait_seconds=None) -> bool
- CLI python3 layout.py wait
- converge() now also runs _adopt_legacy, _ensure_write_aliases, _release_orphans
- Crash classification in converge() (spec §5.1, consumed by Task 8's init exit code): an exception inside a self-heal step (_adopt_legacy, _release_orphans) is appended to result.problems as "<step> crashed: <repr>"; an exception inside _converge_policies, _converge_templates or _ensure_write_aliases stays in result.fatal, same text
- [ ] Step 1: Append the failing tests to
deploy/elasticsearch/test_layout.py(above theif __name__block)
Also add from unittest import mock directly below import unittest at the top of the file (the crash tests patch layout's step functions).
class TestLegacyAdoption(LayoutCase):
def test_legacy_index_is_adopted_with_origination_from_newest_doc(self):
newest = self.es.now_ms - 3 * DAY_MS
self.es.add_index("logster-events", docs=[newest - DAY_MS, newest])
self.assertEqual(self.converge().fatal, [])
s = self.es.indices["logster-events"].settings
self.assertEqual(s["index.lifecycle.name"], "logster-events-legacy")
self.assertEqual(s["index.lifecycle.origination_date"], str(newest))
# Releases a legacy-named index stuck in the family policy's rollover step.
self.assertEqual(s["index.lifecycle.indexing_complete"], "true")
def test_empty_legacy_index_ages_from_creation_date(self):
self.es.add_index("logster-graphs", created_ms=1_700_000_000_000)
self.converge()
self.assertEqual(self.es.indices["logster-graphs"].settings[
"index.lifecycle.origination_date"], "1700000000000")
def test_legacy_origination_follows_new_documents(self):
idx = self.es.add_index("logster-inferences", docs=[self.es.now_ms - DAY_MS])
self.converge()
idx.docs.append(self.es.now_ms)
before = len(self.es.mutations())
self.converge()
self.assertIn(("PUT", "/logster-inferences/_settings"), self.es.mutations()[before:])
self.assertEqual(idx.settings["index.lifecycle.origination_date"], str(self.es.now_ms))
def test_adopted_legacy_index_is_not_rewritten(self):
self.es.add_index("logster-events", docs=[self.es.now_ms])
self.converge()
n = len(self.es.writes_to("/logster-events/_settings"))
self.converge()
self.assertEqual(len(self.es.writes_to("/logster-events/_settings")), n)
def test_legacy_without_docs_or_creation_date_is_left_alone(self):
idx = self.es.add_index("logster-events")
idx.settings.pop("index.creation_date")
result = self.converge()
self.assertTrue(any(p.startswith("legacy logster-events: no documents and no "
"creation date") for p in result.problems))
self.assertNotIn("index.lifecycle.name", idx.settings)
def test_legacy_under_flood_block_is_a_problem_not_fatal(self):
self.es.add_index("logster-events", docs=[self.es.now_ms],
settings={"index.blocks.read_only_allow_delete": True})
result = self.converge()
self.assertEqual(result.fatal, [])
self.assertTrue(any("legacy logster-events: adoption PUT rejected (HTTP 429)" in p
for p in result.problems))
class TestWriteAliases(LayoutCase):
def test_fresh_install_bootstraps_one_write_index_per_family(self):
self.assertEqual(self.converge().fatal, [])
for family in ("events", "inferences", "graphs"):
idx = self.es.indices[f"logster-{family}-000001"]
self.assertEqual(idx.aliases, {f"logster-{family}-write": {"is_write_index": True}})
self.assertEqual(idx.settings["index.lifecycle.name"], f"logster-{family}")
self.assertEqual(idx.settings["index.lifecycle.rollover_alias"],
f"logster-{family}-write")
self.assertEqual(idx.settings["index.codec"], "best_compression")
def test_second_full_run_changes_nothing(self):
self.converge()
result = self.converge()
self.assertEqual((result.fatal, result.problems, result.changes), ([], [], []))
def test_write_index_missing_lifecycle_settings_is_repaired(self):
self.converge()
idx = self.es.indices["logster-events-000001"]
del idx.settings["index.lifecycle.rollover_alias"]
result = self.converge()
self.assertEqual(idx.settings["index.lifecycle.rollover_alias"], "logster-events-write")
self.assertTrue(any("write index logster-events-000001" in c for c in result.changes))
def test_deleted_write_index_is_replaced_with_the_next_number(self):
self.converge()
self.es.indices["logster-events-000001"].aliases["logster-events-write"] = {
"is_write_index": False}
self.es.add_index("logster-events-000002",
aliases={"logster-events-write": {"is_write_index": False}})
self.assertEqual(self.converge().fatal, [])
self.assertEqual(self.es.indices["logster-events-000003"].aliases["logster-events-write"],
{"is_write_index": True})
def test_lone_member_without_flag_counts_as_writer(self):
self.es.add_index("logster-graphs-000007", aliases={"logster-graphs-write": {}})
self.converge()
self.assertNotIn("logster-graphs-000008", self.es.indices)
def test_squatter_is_reported_not_fatal_and_not_bootstrapped(self):
self.es.add_index("logster-graphs-write", docs=[self.es.now_ms])
result = self.converge()
self.assertEqual(result.fatal, [])
self.assertTrue(any(p.startswith("SQUATTER: a concrete index named logster-graphs-write")
for p in result.problems))
self.assertFalse([n for n in self.es.indices if n.startswith("logster-graphs-0")])
def test_bootstrap_race_lost_to_another_converge_is_success(self):
def rival(name):
if name == "logster-events-000001" and name not in self.es.indices:
self.es.add_index(name, aliases={"logster-events-write": {"is_write_index": True}})
self.es.on_create = rival
result = self.converge()
self.assertEqual(result.fatal, [])
self.assertTrue(any("created concurrently" in c for c in result.changes))
def test_bootstrap_rejected_without_a_writer_is_fatal(self):
self.es.fail.append(("PUT", "/logster-events-000001", (400, json.dumps(
{"error": {"type": "validation_exception", "reason": "too many shards"}}))))
result = self.converge()
self.assertTrue(any(f.startswith("bootstrap logster-events-000001: rejected (HTTP 400) "
"validation_exception") for f in result.fatal))
class TestOrphans(LayoutCase):
def setUp(self):
super().setUp()
self.converge() # 000001 is now the writer of every family
def add_former_writer(self, step, failed_step=None, complete=None):
settings = {"index.lifecycle.name": "logster-events",
"index.lifecycle.rollover_alias": "logster-events-write"}
if complete is not None:
settings["index.lifecycle.indexing_complete"] = complete
idx = self.es.add_index("logster-events-000000",
docs=[self.es.now_ms - 5 * DAY_MS], settings=settings)
idx.ilm = {"phase": "hot", "action": "rollover", "step": step}
if failed_step:
idx.ilm["failed_step"] = failed_step
return idx
def test_orphan_in_error_is_released_and_retried(self):
idx = self.add_former_writer("ERROR", failed_step="check-rollover-ready")
result = self.converge()
self.assertEqual(idx.settings["index.lifecycle.indexing_complete"], "true")
self.assertIn(("POST", "/logster-events-000000/_ilm/retry"), self.es.mutations())
self.assertEqual(idx.ilm["step"], "check-rollover-ready")
self.assertTrue(any("released former write index logster-events-000000" in c
for c in result.changes))
def test_retry_400_means_ilm_already_retried(self):
# ILM's auto-retry flips an index between ERROR and the failed step;
# _ilm/retry then answers 400 (not in an error step). That is success.
idx = self.add_former_writer("ERROR", failed_step="check-rollover-ready")
self.es.fail.append(("POST", "/logster-events-000000/_ilm/retry", (400, json.dumps(
{"error": {"type": "illegal_argument_exception",
"reason": "cannot retry an action for an index that has not "
"encountered an error"}}))))
result = self.converge()
self.assertEqual(idx.settings["index.lifecycle.indexing_complete"], "true")
self.assertFalse(any("ILM retry rejected" in p for p in result.problems))
self.assertTrue(any("released former write index logster-events-000000" in c
for c in result.changes))
def test_normal_post_rollover_steps_are_left_alone(self):
self.add_former_writer("set-indexing-complete", complete=True)
before = len(self.es.mutations())
result = self.converge()
self.assertEqual(self.es.mutations()[before:], [])
self.assertEqual(result.changes, [])
def test_current_writer_is_never_released(self):
writer = self.es.indices["logster-events-000001"]
writer.ilm = {"phase": "hot", "action": "rollover", "step": "check-rollover-ready"}
self.converge()
self.assertNotIn("index.lifecycle.indexing_complete", writer.settings)
class TestWaitForAliases(LayoutCase):
def test_returns_true_once_every_alias_has_a_writer(self):
self.converge()
self.assertTrue(layout.wait_for_aliases(self.es, self.log, sleep=lambda s: None,
max_wait_seconds=0))
def test_gives_up_after_max_wait_when_not_ready(self):
sleeps = []
self.assertFalse(layout.wait_for_aliases(self.es, self.log, poll_seconds=5,
sleep=sleeps.append, max_wait_seconds=20))
self.assertEqual(sleeps, [5, 5, 5, 5])
self.assertTrue(any("waiting for write aliases" in line for line in self.logs))
def test_wait_ready_accepts_lone_unflagged_member(self):
# ES writes to a single-index alias that has no is_write_index flag,
# and converge() (writer_of) accepts it without bootstrapping, so the
# initContainer must not wait forever on it.
self.converge()
for family in ("events", "inferences", "graphs"):
self.es.indices[f"logster-{family}-000001"].aliases[
f"logster-{family}-write"] = {}
self.assertTrue(layout.write_aliases_ready(self.es))
def test_wait_not_ready_when_every_member_is_flagged_false(self):
self.converge()
self.es.indices["logster-events-000001"].aliases["logster-events-write"] = {
"is_write_index": False}
self.es.add_index("logster-events-000002",
aliases={"logster-events-write": {"is_write_index": False}})
self.assertFalse(layout.write_aliases_ready(self.es))
def test_wait_treats_a_squatted_alias_name_as_ready_and_logs_it(self):
# Writes to the name land in the squatter, so Vector must not block;
# converge() keeps reporting the fix every check.
self.es.add_index("logster-graphs-write", docs=[self.es.now_ms])
self.converge() # bootstraps events + inferences, reports the squatter
self.logs.clear()
self.assertTrue(layout.wait_for_aliases(self.es, self.log, sleep=lambda s: None,
max_wait_seconds=0))
self.assertTrue(any("SQUATTER" in line and "logster-graphs-write" in line
for line in self.logs))
class TestStepCrashes(LayoutCase):
"""A crash in a step that establishes the layout is fatal (init exits 1);
a crash in a self-heal step is a problem, retried on the next check."""
def crash(self, name):
def step(ctx):
raise RuntimeError("boom")
step.__name__ = name
return mock.patch.object(layout, name, step)
def test_self_heal_step_crash_is_a_problem_not_fatal(self):
for name in ("_adopt_legacy", "_release_orphans"):
with self.subTest(step=name):
self.es = FakeES()
with self.crash(name):
result = self.converge()
self.assertEqual(result.fatal, [])
self.assertIn(f"{name} crashed: RuntimeError('boom')", result.problems)
# The other steps still ran: the write index was bootstrapped.
self.assertIn("logster-events-000001", self.es.indices)
def test_layout_step_crash_is_fatal(self):
for name in ("_converge_policies", "_converge_templates", "_ensure_write_aliases"):
with self.subTest(step=name):
self.es = FakeES()
with self.crash(name):
result = self.converge()
self.assertIn(f"{name} crashed: RuntimeError('boom')", result.fatal)
self.assertNotIn(f"{name} crashed: RuntimeError('boom')", result.problems)
- [ ] Step 2: Run them to confirm they fail
Run: venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_layout.py' -v
Expected: the new tests fail (no aliases are bootstrapped; AttributeError: module 'layout' has no attribute 'wait_for_aliases' / 'write_aliases_ready'; the TestStepCrashes subtests error because layout has no _adopt_legacy / _release_orphans / _ensure_write_aliases to patch). The Task 3 tests still pass.
- [ ] Step 3: Add the functions to
deploy/elasticsearch/layout.py(append at the end of the file)
Dates in log text are formatted with eshttp.iso_ms (Task 1); do not define a local copy. If layout.py still has an import datetime line, delete it: nothing in the module uses it.
def _index_settings(request, index):
"""("absent", None) | ("ok", flat_settings) | ("error", http_status)."""
status, doc = eshttp.call(request, "GET", f"/{index}/_settings?flat_settings=true")
if status == 404:
return "absent", None
if status != 200 or not isinstance(doc, dict):
return "error", status
entry = doc.get(index)
if not isinstance(entry, dict):
return "absent", None # the name resolved to something else (an alias)
return "ok", entry.get("settings") or {}
def _newest_timestamp(request, index):
"""(ok, newest epoch-ms or None)."""
status, doc = eshttp.call(request, "POST", f"/{index}/_search", {
"size": 0, "track_total_hits": False,
"aggs": {"newest": {"max": {"field": "@timestamp"}}}})
if (status != 200 or not isinstance(doc, dict) or doc.get("timed_out")
or (doc.get("_shards") or {}).get("failed", 0)):
return False, None
value = ((doc.get("aggregations") or {}).get("newest") or {}).get("value")
return True, (int(value) if isinstance(value, (int, float)) else None)
def _adopt_legacy(ctx):
"""Put pre-ILM concrete indices under a delete-only policy aged from
their newest document (spec §7.1 step 3, D7)."""
for family in FAMILIES:
name = legacy_index(family)
state, settings = _index_settings(ctx.request, name)
if state == "absent":
continue
if state == "error":
ctx.result.problems.append(f"legacy {name}: settings unreadable (HTTP {settings})")
continue
ok, newest = _newest_timestamp(ctx.request, name)
if not ok:
ctx.result.problems.append(f"legacy {name}: newest @timestamp unreadable")
continue
if newest is None:
created = str(settings.get("index.creation_date") or "")
if not created.isdigit():
ctx.result.problems.append(
f"legacy {name}: no documents and no creation date — left unmanaged")
continue
newest = int(created)
# indexing_complete: a bare name auto-created while rolled back
# (spec §8.3) picked up the family template's policy and sits in its
# rollover step. After a policy switch ES keeps running the cached hot
# phase, so without it the index never reaches the legacy delete
# phase (verified on ES 9.5.4). Harmless on a never-managed index.
want = {"index.lifecycle.name": legacy_policy_name(family),
"index.lifecycle.origination_date": str(newest),
"index.lifecycle.indexing_complete": "true"}
if all(settings.get(k) == v for k, v in want.items()):
continue
status, doc = eshttp.call(ctx.request, "PUT", f"/{name}/_settings", {
"index.lifecycle.name": legacy_policy_name(family),
"index.lifecycle.origination_date": newest,
"index.lifecycle.indexing_complete": True})
if status != 200:
ctx.result.problems.append(
f"legacy {name}: adoption PUT rejected (HTTP {status}) {_reason(doc)}")
else:
ctx.result.changes.append(
f"legacy index {name} managed by {legacy_policy_name(family)}, "
f"aged from its newest document ({eshttp.iso_ms(newest)})")
def alias_members(request, alias):
"""("ok", {index: is_write_flag}) | ("missing", {}) | ("error", status)."""
status, doc = eshttp.call(request, "GET", f"/_alias/{alias}")
if status == 404:
return "missing", {}
if status != 200 or not isinstance(doc, dict):
return "error", status
members = {}
for index, entry in doc.items():
aliases = (entry or {}).get("aliases") or {}
if alias in aliases:
members[index] = (aliases[alias] or {}).get("is_write_index")
return ("ok", members) if members else ("missing", {})
def writer_of(members):
"""The index writes go to: explicit is_write_index, else a lone member
without a flag (ES treats a single-index alias as its write target)."""
explicit = [index for index, flag in members.items() if flag is True]
if explicit:
return explicit[0]
if len(members) == 1 and next(iter(members.values())) is None:
return next(iter(members))
return None
def _ensure_write_aliases(ctx):
for family in FAMILIES:
alias = write_alias(family)
state, members = alias_members(ctx.request, alias)
if state == "error":
ctx.result.fatal.append(f"alias {alias}: GET failed (HTTP {members})")
continue
writer = writer_of(members) if state == "ok" else None
if writer:
_ensure_write_index_settings(ctx, family, writer)
continue
if state == "missing" and _squatted(ctx.request, alias):
ctx.result.problems.append(squatter_message(alias))
continue
_bootstrap(ctx, family)
def _squatted(request, alias):
"""True when a concrete index holds the write-alias name (spec §7.1):
writes land in it, and no alias can be created until it is moved."""
state, _ = _index_settings(request, alias)
return state == "ok"
def squatter_message(alias):
return (f"SQUATTER: a concrete index named {alias} holds the write-alias "
"name, so writes land in an unmanaged index. See 'Write-alias "
"squatter' in docs/technical-guides/openshift-deployment/"
"data-retention.md for the fix.")
def _bootstrap(ctx, family):
alias = write_alias(family)
status, rows = eshttp.call(
ctx.request, "GET",
f"/_cat/indices/logster-{family}-*?format=json&h=index&expand_wildcards=all")
if status != 200 or not isinstance(rows, list):
ctx.result.fatal.append(f"alias {alias}: cannot list rollover indices (HTTP {status})")
return
suffixes = [int(m.group(2)) for row in rows
if (m := ROLLOVER_RE.match(str((row or {}).get("index", ""))))
and m.group(1) == family]
name = f"logster-{family}-{max(suffixes, default=0) + 1:06d}"
status, doc = eshttp.call(ctx.request, "PUT", f"/{name}", {
"aliases": {alias: {"is_write_index": True}},
"settings": {"index.lifecycle.name": policy_name(family),
"index.lifecycle.rollover_alias": alias}})
if status == 200:
ctx.result.changes.append(f"bootstrapped {name} as the write index of {alias}")
return
# init and the worker may bootstrap at the same moment. The loser gets
# resource_already_exists or "more than one write index": success if a
# writer now exists.
state, members = alias_members(ctx.request, alias)
if state == "ok" and writer_of(members):
ctx.result.changes.append(
f"{alias} write index created concurrently ({writer_of(members)})")
return
ctx.result.fatal.append(f"bootstrap {name}: rejected (HTTP {status}) {_reason(doc)}")
def _ensure_write_index_settings(ctx, family, index):
state, settings = _index_settings(ctx.request, index)
if state != "ok":
ctx.result.problems.append(f"write index {index}: settings unreadable")
return
want = {"index.lifecycle.name": policy_name(family),
"index.lifecycle.rollover_alias": write_alias(family)}
missing = {k: v for k, v in want.items() if settings.get(k) != v}
if not missing:
return
status, doc = eshttp.call(ctx.request, "PUT", f"/{index}/_settings", missing)
if status != 200:
ctx.result.problems.append(
f"write index {index}: lifecycle settings PUT rejected (HTTP {status}) {_reason(doc)}")
else:
ctx.result.changes.append(f"write index {index}: set {', '.join(sorted(missing))}")
def _release_orphans(ctx):
"""A former write index whose alias was removed by hand never finishes
ILM's rollover step (spec §7.1 step 5). Tell ILM to skip it."""
status, doc = eshttp.call(ctx.request, "GET", f"/{ROLLOVER_PATTERNS}/_ilm/explain")
if status == 404:
return
if status != 200 or not isinstance(doc, dict):
ctx.result.problems.append(f"orphan check skipped: ILM explain unavailable (HTTP {status})")
return
writers = set()
for family in FAMILIES:
state, members = alias_members(ctx.request, write_alias(family))
if state == "error":
ctx.result.problems.append("orphan check skipped: write aliases unreadable")
return
if state == "ok" and writer_of(members):
writers.add(writer_of(members))
for name, info in sorted((doc.get("indices") or {}).items()):
if (not ROLLOVER_RE.match(name) or name in writers or not isinstance(info, dict)
or info.get("phase") != "hot" or info.get("action") != "rollover"):
continue
acted = False
state, settings = _index_settings(ctx.request, name)
if state == "ok" and settings.get("index.lifecycle.indexing_complete") != "true":
put_status, _ = eshttp.call(ctx.request, "PUT", f"/{name}/_settings",
{"index.lifecycle.indexing_complete": True})
if put_status != 200:
ctx.result.problems.append(
f"orphan {name}: indexing_complete PUT rejected (HTTP {put_status})")
continue
acted = True
if info.get("step") == "ERROR":
retry_status, _ = eshttp.call(ctx.request, "POST", f"/{name}/_ilm/retry")
if retry_status == 200:
acted = True
elif retry_status != 400:
ctx.result.problems.append(
f"orphan {name}: ILM retry rejected (HTTP {retry_status})")
continue
# 400: the index left ERROR between explain and this retry (ES
# auto-retries check-rollover-ready). indexing_complete is set,
# so ILM moves on by itself.
if acted:
ctx.result.changes.append(f"released former write index {name} from rollover")
def _alias_state(request, family):
""""ready": the write alias has a writer, by the rule converge() uses
(writer_of: an explicit is_write_index, or a lone member without a flag,
which ES itself writes to). "squatted": a concrete index holds the alias
name; writes land in it, and converge() never creates that alias.
"waiting": converge() has yet to create or repair the alias."""
alias = write_alias(family)
state, members = alias_members(request, alias)
if state == "ok" and writer_of(members) is not None:
return "ready"
if state == "missing" and _squatted(request, alias):
return "squatted"
return "waiting"
def write_aliases_ready(request):
"""True when no family is still "waiting" (see _alias_state). A squatted
family counts as ready: writes to its name succeed, and waiting on an
alias converge() will never create would only block ingestion."""
return all(_alias_state(request, family) != "waiting" for family in FAMILIES)
def wait_for_aliases(request, log, poll_seconds=5, log_every_seconds=60,
sleep=time.sleep, max_wait_seconds=None):
"""Block until no write alias is still "waiting" (used by the Vector
initContainer). A squatted alias is logged as an ERROR once and does not
block. Returns False only if max_wait_seconds elapses."""
waited = 0
next_log = 0
reported = set()
while True:
states = {write_alias(family): _alias_state(request, family) for family in FAMILIES}
for alias, state in states.items():
if state == "squatted" and alias not in reported:
log(f"ERROR: {squatter_message(alias)}")
reported.add(alias)
waiting = [alias for alias, state in states.items() if state == "waiting"]
if not waiting:
log("write aliases ready")
return True
if max_wait_seconds is not None and waited >= max_wait_seconds:
return False
if waited >= next_log:
log(f"waiting for write aliases ({waited}s): {', '.join(waiting)} — the "
"retention worker or elasticsearch-init creates them")
next_log += log_every_seconds
sleep(poll_seconds)
waited += poll_seconds
if __name__ == "__main__":
if sys.argv[1:] == ["wait"]:
wait_for_aliases(eshttp.request, eshttp.make_logger("wait-es-layout"))
sys.exit(0)
print("usage: layout.py wait", file=sys.stderr)
sys.exit(2)
Then in converge(), replace the whole step loop (the for step in (_converge_policies, _converge_templates): statement and its try/except body) with:
# A crash in a step that establishes the layout (policies, templates,
# write aliases) is fatal: init exits 1 on it. A crash in a self-heal
# step (legacy adoption, orphan release) is a problem: logged, and
# retried by the worker's next check (spec §5.1).
for step, sink in ((_converge_policies, result.fatal),
(_converge_templates, result.fatal),
(_adopt_legacy, result.problems),
(_ensure_write_aliases, result.fatal),
(_release_orphans, result.problems)):
try:
step(ctx)
except Exception as exc: # noqa: BLE001 — one broken step must not hide the rest
sink.append(f"{step.__name__} crashed: {exc!r}")
converge(): the steps are looked up when converge() runs, which is what TestStepCrashes patches.
- [ ] Step 4: Run all ES unit tests
Run: venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_layout.py' -v
Expected: all OK, including the Task 3 tests.
- [ ] Step 5: Stage
feat(retention): adopt legacy indices, bootstrap and self-heal write aliases
Task 5: diskcap.py, the disk rule
Files:
- Create: deploy/elasticsearch/diskcap.py
- Test: deploy/elasticsearch/test_diskcap.py
Interfaces:
- Consumes: eshttp.call, eshttp.fs_usage, eshttp.iso_ms (Task 1); layout.MANAGED_RE, layout.READ_PATTERNS, layout.writer_of (Tasks 3–4); RetentionConfig.target_percent, .min_age_hours
- Produces:
- Outcome(action, percent=None, deleted=()), where action is "skipped" | "under_target" | "floor_reached" | "deleted"
- enforce(request, log, cfg) -> Outcome
- TERMS_SIZE = 10000 (module global, read at call time)
- [ ] Step 1: Write the failing test
deploy/elasticsearch/test_diskcap.py
"""Disk rule tests (spec §7.3) against the in-memory ES double."""
import unittest
from unittest.mock import patch
import diskcap
from fake_es import GIB, HOUR_MS, FakeES
from retention_config import RetentionConfig
CFG = RetentionConfig() # target 75 %, floor 24 h
class DiskCase(unittest.TestCase):
def setUp(self):
self.es = FakeES(total=100 * GIB, available=20 * GIB) # 80 % used → 5 GiB excess
self.logs = []
def log(self, *parts):
self.logs.append(" ".join(map(str, parts)))
def index(self, name, age_hours, store_gib, writer=None, docs=1):
aliases = {}
if writer is not None:
aliases = {f"logster-{name.split('-')[1]}-write": {"is_write_index": writer}}
ts = self.es.now_ms - int(age_hours * HOUR_MS)
return self.es.add_index(name, docs=[ts] * docs, store=int(store_gib * GIB),
aliases=aliases)
def enforce(self, cfg=CFG):
return diskcap.enforce(self.es, self.log, cfg)
def deleted(self):
return [p[1:] for m, p in self.es.mutations() if m == "DELETE"]
class TestDiskRule(DiskCase):
def test_under_target_does_nothing(self):
self.es.available = 30 * GIB # 70 %
self.index("logster-events-000001", 240, 3, writer=False)
self.assertEqual(self.enforce().action, "under_target")
self.assertEqual(self.deleted(), [])
def test_deletes_oldest_first_and_stops_when_excess_is_covered(self):
self.index("logster-events-000001", 240, 3, writer=False)
self.index("logster-inferences-000001", 216, 1, writer=False)
self.index("logster-events-000002", 120, 3, writer=False)
self.index("logster-events-000003", 48, 3, writer=False)
self.index("logster-events-000004", 0.1, 1, writer=True)
self.assertEqual(self.enforce().action, "deleted")
self.assertEqual(self.deleted(), ["logster-events-000001", "logster-inferences-000001",
"logster-events-000002"])
self.assertIn("logster-events-000003", self.es.indices)
def test_write_index_is_never_deleted_even_when_oldest(self):
self.index("logster-graphs-000001", 500, 10, writer=True)
self.assertEqual(self.enforce().action, "floor_reached")
self.assertEqual(self.deleted(), [])
def test_floor_protects_recent_data(self):
self.index("logster-events-000001", 2, 10, writer=False)
self.assertEqual(self.enforce().action, "floor_reached")
self.assertTrue(any("older than the 24h floor" in line for line in self.logs))
def test_not_enough_eligible_data_deletes_all_of_it_and_warns(self):
self.index("logster-events-000001", 100, 1, writer=False)
self.index("logster-events", 200, 1) # legacy index, no alias
self.index("logster-events-000002", 1, 1, writer=True)
self.enforce()
self.assertEqual(self.deleted(), ["logster-events", "logster-events-000001"])
self.assertTrue(any("something other than these indices" in line for line in self.logs))
def test_empty_indices_are_skipped(self):
self.index("logster-events-000001", 240, 5, writer=False, docs=0)
self.assertEqual(self.enforce().action, "floor_reached")
def test_look_alike_names_are_never_deleted(self):
for name in ("logster-events-backup", "logster-events-00001", "logster-eventsx",
"logster-events-write-old", "logster-capacity"):
self.index(name, 1000, 5)
self.assertEqual(self.enforce().action, "floor_reached")
self.assertEqual(self.deleted(), [])
def test_future_dated_data_counts_as_recent(self):
self.index("logster-events-000001", -72, 10, writer=False) # 3 days ahead
self.assertEqual(self.enforce().action, "floor_reached")
def test_more_than_ten_indices_are_all_considered(self):
for n in range(15):
self.index(f"logster-events-{n + 1:06d}", 1000 - n, 0.5, writer=False)
self.enforce()
self.assertEqual(self.deleted(), [f"logster-events-{n + 1:06d}" for n in range(10)])
def test_truncated_terms_fail_closed(self):
for n in range(8):
self.index(f"logster-events-{n + 1:06d}", 1000, 1, writer=False)
with patch.object(diskcap, "TERMS_SIZE", 5):
self.assertEqual(self.enforce().action, "skipped")
self.assertEqual(self.deleted(), [])
def test_partial_shard_failure_fails_closed(self):
self.index("logster-events-000001", 1000, 10, writer=False)
self.es.shards_failed = 1
self.assertEqual(self.enforce().action, "skipped")
def test_unknown_disk_usage_skips(self):
self.es.nodes_failed = 1
self.assertEqual(self.enforce().action, "skipped")
def test_unreadable_aliases_skip(self):
self.index("logster-events-000001", 1000, 10, writer=False)
self.es.fail.append(("GET", "/_alias/", (500, "{}")))
self.assertEqual(self.enforce().action, "skipped")
self.assertEqual(self.deleted(), [])
def test_already_gone_counts_and_a_failed_delete_stops(self):
self.index("logster-events-000001", 300, 1, writer=False)
self.index("logster-events-000002", 200, 1, writer=False)
self.index("logster-events-000003", 100, 5, writer=False)
self.es.fail.append(("DELETE", "/logster-events-000001", (404, "{}")))
self.es.fail.append(("DELETE", "/logster-events-000002", (500, "{}")))
out = self.enforce()
self.assertEqual(out.deleted, ("logster-events-000001",))
self.assertIn("logster-events-000003", self.es.indices)
self.assertTrue(any("DELETE logster-events-000002 failed (HTTP 500)" in line
for line in self.logs))
def test_writer_change_between_plan_and_delete_is_respected(self):
self.index("logster-events-000001", 300, 10, writer=False)
with patch.object(diskcap, "_write_indices",
side_effect=[set(), {"logster-events-000001"}]):
out = self.enforce()
self.assertEqual(out.deleted, ())
self.assertIn("logster-events-000001", self.es.indices)
def test_deletion_is_logged_with_size_and_age(self):
self.index("logster-events-000001", 240, 6, writer=False)
self.enforce()
self.assertTrue(any(line.startswith("retention: deleted logster-events-000001 "
"(6144 MiB, newest ") for line in self.logs))
if __name__ == "__main__":
unittest.main()
- [ ] Step 2: Run it to confirm it fails
Run: venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_diskcap.py' -v
Expected: ModuleNotFoundError: No module named 'diskcap'.
- [ ] Step 3: Implement
deploy/elasticsearch/diskcap.py
"""Disk rule for the retention worker (spec §7.3).
While the Elasticsearch data filesystem is above the target, delete the
oldest whole Logster indices. Never a write index, never one holding data
newer than the floor (evaluated with Elasticsearch's clock), never anything
outside layout.MANAGED_RE, and only by exact name.
The plan is computed once per check from exact index sizes; there is no
read-disk / delete / read-disk-again loop, so a stale disk reading cannot
cause over-deletion. The next check corrects any estimate error.
"""
from dataclasses import dataclass
import eshttp
import layout
TERMS_SIZE = 10000
@dataclass(frozen=True)
class Outcome:
action: str # skipped | under_target | floor_reached | deleted
percent: float = None
deleted: tuple = ()
def _open_indices(request):
"""[(name, docs, store_bytes)] for open indices under the read patterns."""
status, rows = eshttp.call(
request, "GET",
f"/_cat/indices/{layout.READ_PATTERNS}?format=json&bytes=b"
"&h=index,docs.count,store.size&expand_wildcards=open")
if status != 200 or not isinstance(rows, list):
return None
out = []
for row in rows:
if not isinstance(row, dict):
return None
try:
out.append((str(row.get("index", "")), int(row.get("docs.count") or 0),
int(row.get("store.size") or 0)))
except (TypeError, ValueError):
return None
return out
def _write_indices(request):
"""Names of every current write index, or None when unreadable."""
status, doc = eshttp.call(request, "GET", "/_alias/logster-*-write")
if status == 404:
return set()
if status != 200 or not isinstance(doc, dict):
return None
by_alias = {}
for index, entry in doc.items():
for alias, meta in ((entry or {}).get("aliases") or {}).items():
by_alias.setdefault(alias, {})[index] = (meta or {}).get("is_write_index")
return {w for members in by_alias.values() if (w := layout.writer_of(members))}
def _index_ages(request, floor_hours):
"""(newest @timestamp by index, set of indices with data newer than the
floor), or None — which fails the check closed."""
status, doc = eshttp.call(
request, "POST",
f"/{layout.READ_PATTERNS}/_search?ignore_unavailable=true&allow_no_indices=true", {
"size": 0, "track_total_hits": False,
"aggs": {
"by_index": {"terms": {"field": "_index", "size": TERMS_SIZE},
"aggs": {"newest": {"max": {"field": "@timestamp"}}}},
"recent": {"filter": {"range": {"@timestamp": {"gte": f"now-{floor_hours}h"}}},
"aggs": {"idx": {"terms": {"field": "_index", "size": TERMS_SIZE}}}},
}}, timeout=60)
if (status != 200 or not isinstance(doc, dict) or doc.get("timed_out")
or (doc.get("_shards") or {}).get("failed", 0)):
return None
try:
by_index = doc["aggregations"]["by_index"]
recent = doc["aggregations"]["recent"]["idx"]
# A truncated bucket list would make recent indices look eligible.
if by_index.get("sum_other_doc_count", 0) or recent.get("sum_other_doc_count", 0):
return None
newest = {b["key"]: b["newest"]["value"] for b in by_index["buckets"]
if isinstance(b["newest"].get("value"), (int, float))}
return newest, {b["key"] for b in recent["buckets"]}
except (KeyError, TypeError, AttributeError):
return None
def enforce(request, log, cfg):
usage = eshttp.fs_usage(request)
if usage is None:
log("disk rule: disk usage unavailable — skipping this check")
return Outcome("skipped")
used, total, pct = usage
if pct <= cfg.target_percent:
return Outcome("under_target", pct)
excess = used - total * cfg.target_percent / 100
indices = _open_indices(request)
writers = _write_indices(request)
ages = _index_ages(request, cfg.min_age_hours)
if indices is None or writers is None or ages is None:
log(f"disk rule: disk {pct:.1f}% > target {cfg.target_percent}% but index state "
"is incomplete (failed or partial read) — skipping this check")
return Outcome("skipped", pct)
newest, recent = ages
candidates = sorted(
(newest[name], name, store) for name, docs, store in indices
if layout.MANAGED_RE.match(name) and name not in writers and docs > 0
and name in newest and name not in recent)
plan, planned = [], 0
for item in candidates:
if planned >= excess:
break
plan.append(item)
planned += item[2]
if not plan:
log(f"WARNING: disk {pct:.1f}% > target {cfg.target_percent}% and no Logster "
f"index is older than the {cfg.min_age_hours}h floor — nothing deleted")
return Outcome("floor_reached", pct)
writers_now = _write_indices(request)
if writers_now is None:
log("disk rule: write aliases unreadable right before deletion — skipping this check")
return Outcome("skipped", pct)
deleted = []
for ts, name, store in plan:
if name in writers_now:
continue
status, _ = eshttp.call(request, "DELETE", f"/{name}", timeout=60)
if status not in (200, 404): # 404: ILM deleted it first
log(f"disk rule: DELETE {name} failed (HTTP {status}) — stopping this check")
break
deleted.append(name)
log(f"retention: deleted {name} ({store / 2**20:.0f} MiB, newest "
f"{eshttp.iso_ms(ts)}) — disk {pct:.1f}% > target {cfg.target_percent}%")
if planned < excess:
log(f"WARNING: deleting every Logster index older than the {cfg.min_age_hours}h "
f"floor frees ~{planned / 2**30:.1f} GiB of the {excess / 2**30:.1f} GiB "
"needed — the rest of the disk is used by something other than these indices")
return Outcome("deleted", pct, tuple(deleted))
- [ ] Step 4: Run it to confirm it passes
Run: venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_diskcap.py' -v
Expected: 16 tests OK.
- [ ] Step 5: Stage
feat(retention): disk rule deletes the oldest whole index above target
Task 6: capacity.py: family patterns, no strip model, target only
Files:
- Modify: deploy/elasticsearch/capacity.py (full replacement below)
- Test: deploy/elasticsearch/test_capacity.py (full replacement below)
Interfaces:
- Produces:
- compute_forecast(request, log, *, events_days, inferences_days, graphs_days, target_pct, usage, es_store_bytes, now_ms=None) -> dict|None
- managed_store_bytes(request) -> int|None
- publish(request, log, forecast) -> bool (unchanged)
- Constants FAMILIES, READ_PATTERNS, MERGE_OVERHEAD, MAX_PROJECTION_DAYS, DAY_MS
- Forecast doc contract, consumed by the dashboard in Task 14:
- Removed: emergency_percent, strip_ratio, indices[].strip_raw_event_days.
- Kept: target_percent, avg_event_json_bytes, avg_raw_event_json_bytes, and every other field.
- indices[].name stays the family name (logster-events).
- [ ] Step 1: Replace
deploy/elasticsearch/test_capacity.py
"""Capacity forecast tests; run with stdlib unittest discovery."""
import importlib.util
import json
from pathlib import Path
import unittest
_spec = importlib.util.spec_from_file_location(
"capacity", Path(__file__).with_name("capacity.py"))
capacity = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(capacity)
NOW = 1_800_000_000_000
DAY = capacity.DAY_MS
GIB = 2 ** 30
SAMPLE_DOC = {"event_id": "e", "data": {"x": "a" * 200},
"raw_event": {"payload": "b" * 600}}
_FULL = len(json.dumps(SAMPLE_DOC, separators=(",", ":")))
_RAW = len(json.dumps(SAMPLE_DOC["raw_event"], separators=(",", ":")))
def index(total, last_24h, oldest_days, store):
return {"total": total, "last_24h": last_24h,
"oldest": NOW - oldest_days * DAY if total else None, "store": store}
class FakeES:
"""Routes the forecast's reads by path; `fail` names path fragments that 500."""
def __init__(self, indices, sample=(SAMPLE_DOC,) * 3, fail=()):
self.indices, self.sample, self.fail = indices, sample, fail
self.calls = []
def __call__(self, method, path, data=None, timeout=60):
self.calls.append((method, path))
target = path.split("/")[1].split("?")[0]
if any(f in path for f in self.fail):
return 500, "{}"
if path.endswith("_doc/current"):
return 201, "{}"
if "_stats/store" in path:
if "," in target: # managed_store_bytes over all three patterns
total = sum(i["store"] for i in self.indices.values())
return 200, json.dumps({"_all": {"total": {"store": {"size_in_bytes": total}}}})
return 200, json.dumps({"_all": {"primaries": {"store": {
"size_in_bytes": self.indices[target]["store"]}}}})
body = json.loads(data)
if body.get("size"): # event sample
return 200, json.dumps({"hits": {"hits": [{"_source": d} for d in self.sample]}})
i = self.indices[target]
return 200, json.dumps({"hits": {"total": {"value": i["total"]}},
"aggregations": {"oldest": {"value": i["oldest"]},
"last_24h": {"doc_count": i["last_24h"]}}})
def forecast(es, usage=(40 * GIB, 100 * GIB, 40), es_store=10 * GIB, **kw):
args = dict(events_days=30, inferences_days=90, graphs_days=14, target_pct=75)
args.update(kw)
return capacity.compute_forecast(es, lambda *a: None, usage=usage,
es_store_bytes=es_store, now_ms=NOW, **args)
def steady_es():
# Events: 1M/day at 1000 B/doc, 20 days in. Inferences: 100/day at 2000 B.
return FakeES({
"logster-events*": index(10_000_000, 1_000_000, 20, 1000 * 10_000_000),
"logster-inferences*": index(1000, 100, 10, 1000 * 2000),
"logster-graphs*": index(0, 0, 0, 0),
})
class TestForecast(unittest.TestCase):
def test_steady_state_is_rate_times_size_times_days(self):
f = forecast(steady_es())
ev, inf = f["indices"][0], f["indices"][1]
self.assertEqual(ev["name"], "logster-events")
self.assertAlmostEqual(ev["bytes_per_doc"], 1000, delta=0.5)
self.assertEqual(ev["docs_per_day"], 1_000_000)
self.assertAlmostEqual(ev["steady_state_bytes"],
1_000_000 * 1000 * 30 * capacity.MERGE_OVERHEAD, delta=1)
self.assertAlmostEqual(inf["steady_state_bytes"],
100 * 2000 * 90 * capacity.MERGE_OVERHEAD, delta=1)
self.assertAlmostEqual(f["es_need_bytes"],
ev["steady_state_bytes"] + inf["steady_state_bytes"], delta=2)
self.assertFalse(f["calibrating"])
self.assertEqual(f["avg_event_json_bytes"], _FULL)
self.assertEqual(f["avg_raw_event_json_bytes"], _RAW)
def test_doc_carries_no_strip_or_emergency_fields(self):
f = forecast(steady_es())
for key in ("strip_ratio", "emergency_percent"):
self.assertNotIn(key, f)
self.assertEqual(f["target_percent"], 75)
self.assertNotIn("strip_raw_event_days", f["indices"][0])
def test_reads_go_through_the_family_patterns(self):
es = steady_es()
forecast(es)
self.assertEqual({p.split("/")[1].split("?")[0] for _, p in es.calls},
{"logster-events*", "logster-inferences*", "logster-graphs*"})
def test_growth_curve_flattens_at_steady_state(self):
curve = forecast(steady_es())["growth_curve"]
per_day = (1_000_000 * 1000 + 100 * 2000) * capacity.MERGE_OVERHEAD
self.assertEqual(curve[0], 0)
self.assertAlmostEqual(curve[1], per_day, delta=2)
self.assertEqual(len(curve), 81) # inferences stop growing on day 80
self.assertGreater(curve[-1], curve[-2])
def test_young_install_is_calibrating_and_rate_is_scaled(self):
es = FakeES({"logster-events*": index(250, 250, 0.25, 250 * 1000),
"logster-inferences*": index(0, 0, 0, 0),
"logster-graphs*": index(0, 0, 0, 0)})
f = forecast(es)
self.assertTrue(f["calibrating"])
self.assertEqual(f["history_hours"], 6)
self.assertEqual(f["indices"][0]["docs_per_day"], 1000)
def test_disabled_delete_rule_is_unbounded(self):
f = forecast(steady_es(), events_days=0)
self.assertIsNone(f["es_need_bytes"])
self.assertEqual(f["unbounded_indices"], ["logster-events"])
self.assertEqual(len(f["growth_curve"]), capacity.MAX_PROJECTION_DAYS + 1)
def test_any_failed_read_yields_no_forecast(self):
for fail in ("logster-inferences*/_search", "logster-events*/_stats"):
with self.subTest(fail=fail):
self.assertIsNone(forecast(FakeES(steady_es().indices, fail=(fail,))))
self.assertIsNone(forecast(steady_es(), usage=None))
self.assertIsNone(forecast(steady_es(), es_store=None))
def test_missing_sample_leaves_message_sizes_unknown(self):
es = steady_es()
es.sample = ()
f = forecast(es)
self.assertIsNone(f["avg_event_json_bytes"])
self.assertIsNotNone(f["es_need_bytes"])
def test_managed_store_bytes_sums_all_families(self):
self.assertEqual(capacity.managed_store_bytes(steady_es()),
1000 * 10_000_000 + 1000 * 2000)
self.assertIsNone(capacity.managed_store_bytes(FakeES({}, fail=("_stats",))))
def test_publish_writes_current_doc(self):
es = steady_es()
self.assertTrue(capacity.publish(es, lambda *a: None, forecast(es)))
self.assertEqual(es.calls[-1], ("PUT", "/logster-capacity/_doc/current"))
es.fail = ("_doc/current",)
self.assertFalse(capacity.publish(es, lambda *a: None, forecast(es)))
if __name__ == "__main__":
unittest.main()
- [ ] Step 2: Run it to confirm it fails
Run: venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_capacity.py' -v
Expected: TypeError: compute_forecast() got an unexpected keyword argument 'target_pct'… or a missing strip_raw_days error, and AttributeError: … 'managed_store_bytes'.
- [ ] Step 3: Replace
deploy/elasticsearch/capacity.py
"""Storage capacity forecast for the elasticsearch-retention worker.
Answers "at today's ingest rate, will the configured retention fit under the
disk target?" and writes the answer to one ES doc
(logster-capacity/_doc/current) that the dashboard's /api/storage route
reads. The dashboard never talks to this worker directly; ES is the
hand-off, like every other dashboard input.
Why a forecast and not just "disk is at X%": usage % only turns red once the
disk is nearly full. An under-provisioned install sits at a healthy 40% for
its first week; after that the disk rule (diskcap.py) starts deleting the
oldest data before its retention age. The forecast flags that on day one.
Model (all measured, nothing hard-coded per event):
* r docs/day per family, counted on @timestamp (Vector's ingest time)
over the trailing 24 h, or the observed span if younger.
* b bytes/doc = primary store bytes / doc count over the family's legacy
and rollover indices, read through the logster-<family>* pattern.
* Footprint `a` days after a family's first doc is r * b * min(a, D).
Steady state is a >= D. A curve of added bytes per future day lets the
dashboard find when the disk crosses the target, after adding Kafka.
The Kafka reservation, the ratio against the budget, the effective
retention and the banner level are computed by the dashboard
(services/dashboard/server/src/lib/capacity.js): only it can read the Kafka
topic configs, and on a shared filesystem Kafka's space changes every one of
those answers.
Stdlib only, like the rest of the worker. `request` and `log` are passed in
so this module has no import-time dependency on retention.py.
"""
import datetime
import json
CAPACITY_INDEX = "logster-capacity"
# (name the dashboard joins on, read pattern covering legacy + rollover indices)
FAMILIES = (
("logster-events", "logster-events*"),
("logster-inferences", "logster-inferences*"),
("logster-graphs", "logster-graphs*"),
)
READ_PATTERNS = ",".join(pattern for _, pattern in FAMILIES)
EVENTS_PATTERN = FAMILIES[0][1]
# A forecast from less than a day of data swings with the diurnal cycle; the
# dashboard suppresses the warning banner until this much history exists.
CALIBRATION_HOURS = 24
# Recent events sampled for average JSON sizes (Kafka message-size estimate).
EVENT_SAMPLE_SIZE = 200
# Margin for segment merges and translog on top of the primary store
# measured now.
MERGE_OVERHEAD = 1.15
# Horizon for the projected growth curve. Longer than any shipped retention;
# a disabled rule (days <= 0, keep forever) grows linearly to this cap.
MAX_PROJECTION_DAYS = 365
DAY_MS = 86_400_000
def _index_stats(request, pattern):
"""Return {total, last_24h, oldest_ms, store_bytes} or None.
A pattern that matches no index is a valid empty result, not a failure.
"""
aggs = {
"oldest": {"min": {"field": "@timestamp"}},
"last_24h": {"filter": {"range": {"@timestamp": {"gte": "now-24h"}}}},
}
status, body = request(
"POST", f"/{pattern}/_search?ignore_unavailable=true&allow_no_indices=true",
json.dumps({"size": 0, "track_total_hits": True, "aggs": aggs}),
timeout=30)
if status is None or status >= 400:
return None
try:
result = json.loads(body)
if result.get("timed_out") or result.get("_shards", {}).get("failed", 0):
return None
total = result["hits"]["total"]["value"]
stats = {
"total": total,
"last_24h": result["aggregations"]["last_24h"]["doc_count"],
"oldest_ms": result["aggregations"]["oldest"]["value"],
}
except (json.JSONDecodeError, KeyError, TypeError, AttributeError):
return None
if not total:
stats["store_bytes"] = 0
return stats
status, body = request(
"GET", f"/{pattern}/_stats/store?ignore_unavailable=true&allow_no_indices=true",
timeout=10)
if status is None or status >= 400:
return None
try:
stats["store_bytes"] = (json.loads(body)["_all"]["primaries"]
["store"]["size_in_bytes"])
except (json.JSONDecodeError, KeyError, TypeError):
return None
return stats
def managed_store_bytes(request):
"""Total store (primaries + replicas) of every Logster data index, or None."""
status, body = request(
"GET", f"/{READ_PATTERNS}/_stats/store"
"?ignore_unavailable=true&allow_no_indices=true", timeout=10)
if status is None or status >= 400:
return None
try:
return json.loads(body)["_all"]["total"]["store"]["size_in_bytes"]
except (json.JSONDecodeError, KeyError, TypeError):
return None
def _event_sample(request):
"""(avg full JSON bytes, avg raw_event JSON bytes) over recent events
that carry raw_event, or None. Producers do not compress, so JSON length
approximates the Kafka message size the dashboard needs."""
status, body = request(
"POST", f"/{EVENTS_PATTERN}/_search?ignore_unavailable=true&allow_no_indices=true",
json.dumps({
"size": EVENT_SAMPLE_SIZE,
"sort": [{"@timestamp": "desc"}],
"query": {"exists": {"field": "raw_event"}},
}), timeout=30)
if status is None or status >= 400:
return None
try:
hits = json.loads(body)["hits"]["hits"]
except (json.JSONDecodeError, KeyError, TypeError):
return None
full = raw = n = 0
for hit in hits:
src = hit.get("_source")
if not isinstance(src, dict) or "raw_event" not in src:
continue
n += 1
full += len(json.dumps(src, separators=(",", ":")))
raw += len(json.dumps(src["raw_event"], separators=(",", ":")))
if not n:
return None
return full / n, raw / n
def _footprint(rate, bytes_per_doc, age_days, keep_days):
"""Bytes held `age_days` after a family's first doc (no overhead).
keep_days <= 0 means keep forever: data is never aged out.
"""
held = age_days if keep_days <= 0 else min(age_days, keep_days)
return rate * bytes_per_doc * held
def compute_forecast(request, log, *, events_days, inferences_days,
graphs_days, target_pct, usage, es_store_bytes, now_ms=None):
"""Build the forecast doc, or return None if a required read failed.
usage is eshttp.fs_usage()'s (used, total, pct) tuple and es_store_bytes
managed_store_bytes(), both measured by the caller in the same check.
"""
if usage is None or es_store_bytes is None:
return None
if now_ms is None:
now_ms = int(datetime.datetime.now(datetime.timezone.utc).timestamp()
* 1000)
policies = tuple(zip(FAMILIES, (events_days, inferences_days, graphs_days)))
stats = {}
for (name, pattern), _ in policies:
s = _index_stats(request, pattern)
if s is None:
log(f"capacity forecast: stats for {name} unavailable — skipping")
return None
stats[name] = s
avg_event_bytes = avg_raw_bytes = None
if stats[FAMILIES[0][0]]["total"]:
sample = _event_sample(request)
if sample is not None:
avg_event_bytes, avg_raw_bytes = sample
oldest = [s["oldest_ms"] for s in stats.values() if s["oldest_ms"] is not None]
history_hours = (now_ms - min(oldest)) / 3_600_000 if oldest else 0.0
indices = []
need = 0.0
unbounded = []
curve = [0.0] * (MAX_PROJECTION_DAYS + 1)
for (name, _), keep_days in policies:
s = stats[name]
age_days = (max(0.0, (now_ms - s["oldest_ms"]) / DAY_MS)
if s["oldest_ms"] is not None else 0.0)
# Rate over the trailing 24 h, scaled up when the family is younger
# (a 6 h old install must not report a quarter of its real rate).
window_h = min(24.0, max(1.0, age_days * 24))
rate = s["last_24h"] * 24.0 / window_h
bytes_per_doc = s["store_bytes"] / s["total"] if s["total"] else 0.0
now_fp = _footprint(rate, bytes_per_doc, age_days, keep_days)
for day in range(MAX_PROJECTION_DAYS + 1):
curve[day] += (_footprint(rate, bytes_per_doc, age_days + day, keep_days)
- now_fp) * MERGE_OVERHEAD
if keep_days <= 0:
unbounded.append(name)
steady = None
else:
steady = _footprint(rate, bytes_per_doc, keep_days, keep_days) * MERGE_OVERHEAD
need += steady
indices.append({
"name": name,
"retention_days": keep_days,
"docs_per_day": round(rate, 1),
"bytes_per_doc": round(bytes_per_doc, 1),
"daily_bytes": round(rate * bytes_per_doc),
"store_bytes": s["store_bytes"],
"steady_state_bytes": round(steady) if steady is not None else None,
"age_days": round(age_days, 2),
})
used, total, _ = usage
return {
"@timestamp": now_ms,
"computed_at": datetime.datetime.fromtimestamp(
now_ms / 1000, datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
"calibrating": history_hours < CALIBRATION_HOURS,
"history_hours": round(history_hours, 2),
"disk_used_bytes": used,
"disk_total_bytes": total,
"es_store_bytes": es_store_bytes,
"target_percent": target_pct,
"merge_overhead": MERGE_OVERHEAD,
"avg_event_json_bytes": round(avg_event_bytes) if avg_event_bytes else None,
"avg_raw_event_json_bytes": round(avg_raw_bytes) if avg_raw_bytes else None,
"es_need_bytes": None if unbounded else round(need),
"unbounded_indices": unbounded,
# growth_curve[d] = bytes the managed families add over the next d
# days (0 at steady state). Daily points; trailing flat tail trimmed.
"growth_curve": _trim_curve(curve),
"indices": indices,
}
def _trim_curve(curve):
"""Round to whole bytes and drop the flat steady-state tail; readers
treat any day past the end as the last value."""
points = [round(v) for v in curve]
while len(points) > 2 and points[-1] == points[-2]:
points.pop()
return points
def publish(request, log, forecast):
"""Write the forecast as logster-capacity/_doc/current. Returns bool."""
status, body = request(
"PUT", f"/{CAPACITY_INDEX}/_doc/current", json.dumps(forecast),
timeout=30)
if status is None or status >= 400:
log(f"capacity forecast publish failed: status={status} {body[:200]}")
return False
need = forecast["es_need_bytes"]
log("capacity forecast: "
f"events {forecast['indices'][0]['docs_per_day']:.0f}/day, "
f"ES steady-state "
f"{'unbounded' if need is None else f'{need / 2**30:.1f} GiB'} vs "
f"budget {forecast['disk_total_bytes'] * forecast['target_percent'] / 100 / 2**30:.1f} GiB"
f"{' (calibrating)' if forecast['calibrating'] else ''}")
return True
- [ ] Step 4: Run it to confirm it passes
Run: venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_capacity.py' -v
Expected: 10 tests OK.
- [ ] Step 5: Stage
feat(retention): capacity forecast reads family patterns, drops the strip model
Task 7: Rewrite retention.py, the worker loop
Files:
- Modify: deploy/elasticsearch/retention.py (full replacement; this supersedes the user's uncommitted _env_int edit)
- Test: deploy/elasticsearch/test_retention.py (full replacement)
Interfaces:
- Consumes: eshttp.request/call/es_status/fs_usage/make_logger, layout.converge/ConvergeResult/READ_PATTERNS, diskcap.enforce, capacity.compute_forecast/managed_store_bytes/publish, retention_config.load_config
- Produces:
- retention.log, retention.TEMPLATE_DIR
- ilm_error_report(request, log), publish_capacity_forecast(request, log, cfg), run_check(request, log, cfg, template_dir), main()
- Env RETENTION_RUN_ONCE=true runs a single check (tests and manual runs)
- [ ] Step 1: Replace
deploy/elasticsearch/test_retention.py
"""Retention worker loop tests; run with stdlib unittest discovery."""
import os
import unittest
from unittest.mock import MagicMock, call, patch
import diskcap
import eshttp
import layout
import retention
from fake_es import FakeES
from retention_config import RetentionConfig
class WorkerCase(unittest.TestCase):
def setUp(self):
self.es = FakeES()
self.logs = []
patch.object(eshttp, "request", self.es).start()
patch.object(retention, "log",
lambda *p: self.logs.append(" ".join(map(str, p)))).start()
patch.dict(os.environ, {"RETENTION_RUN_ONCE": "true"}, clear=True).start()
self.addCleanup(patch.stopall)
class TestRunCheck(WorkerCase):
def test_steps_run_in_order(self):
order = MagicMock()
order.converge.return_value = layout.ConvergeResult()
with patch.object(layout, "converge", order.converge), \
patch.object(retention, "ilm_error_report", order.ilm), \
patch.object(diskcap, "enforce", order.disk), \
patch.object(retention, "publish_capacity_forecast", order.forecast):
retention.run_check(self.es, retention.log, RetentionConfig(), "/t")
self.assertEqual([c[0] for c in order.mock_calls],
["converge", "ilm", "disk", "forecast"])
def test_a_failing_step_does_not_skip_the_rest(self):
with patch.object(layout, "converge", side_effect=RuntimeError("boom")), \
patch.object(diskcap, "enforce") as disk, \
patch.object(retention, "publish_capacity_forecast") as forecast:
retention.run_check(self.es, retention.log, RetentionConfig(), "/t")
disk.assert_called_once()
forecast.assert_called_once()
self.assertIn("ERROR: layout converge failed: RuntimeError('boom')", self.logs)
def test_layout_problems_are_logged_as_warnings(self):
result = layout.ConvergeResult(fatal=["f"], problems=["p"])
with patch.object(layout, "converge", return_value=result), \
patch.object(diskcap, "enforce"), \
patch.object(retention, "publish_capacity_forecast"):
retention.run_check(self.es, retention.log, RetentionConfig(), "/t")
self.assertIn("WARNING: layout: f", self.logs)
self.assertIn("WARNING: layout: p", self.logs)
class TestMain(WorkerCase):
def test_reachable_es_runs_one_check(self):
with patch.object(retention, "run_check") as run:
retention.main()
run.assert_called_once()
self.assertTrue(self.logs[0].startswith("starting — events=30d"))
def test_unreachable_es_skips_the_check(self):
self.es.reachable = False
with patch.object(retention, "run_check") as run:
retention.main()
run.assert_not_called()
self.assertTrue(any("Elasticsearch unavailable (HTTP None)" in l for l in self.logs))
def test_retries_sooner_while_es_is_unavailable(self):
os.environ.pop("RETENTION_RUN_ONCE")
self.es.reachable = False
with patch.object(retention.time, "sleep", side_effect=[None, SystemExit]) as sleep:
with self.assertRaises(SystemExit):
retention.main()
self.assertEqual(sleep.call_args_list, [call(30), call(30)])
def test_401_logs_credential_hint_and_keeps_running(self):
self.es.status_override = 401
with patch.object(retention, "run_check") as run:
retention.main()
run.assert_not_called()
self.assertTrue(any("401: check ES_PASSWORD" in l for l in self.logs))
def test_an_exception_in_a_check_never_ends_the_process(self):
with patch.object(retention, "run_check", side_effect=RuntimeError("x")):
retention.main() # returns normally in run-once mode
self.assertTrue(any(l.startswith("ERROR: check failed") for l in self.logs))
def test_sleeps_for_the_check_interval_between_checks(self):
os.environ.pop("RETENTION_RUN_ONCE")
os.environ["RETENTION_CHECK_INTERVAL_MINUTES"] = "2"
with patch.object(retention, "run_check"), \
patch.object(retention.time, "sleep", side_effect=[None, SystemExit]) as sleep:
with self.assertRaises(SystemExit):
retention.main()
self.assertEqual(sleep.call_args_list, [call(120), call(120)])
def test_sigterm_handler_exits_zero(self):
with patch.object(retention.signal, "signal") as register, \
patch.object(retention, "run_check"):
retention.main()
signum, handler = register.call_args[0]
self.assertEqual(signum, retention.signal.SIGTERM)
with self.assertRaises(SystemExit) as stop:
handler(signum, None)
self.assertEqual(stop.exception.code, 0)
def test_removed_vars_are_warned_at_startup(self):
os.environ["RETENTION_EMERGENCY_THRESHOLD_PERCENT"] = "85"
with patch.object(retention, "run_check"):
retention.main()
self.assertTrue(any("RETENTION_EMERGENCY_THRESHOLD_PERCENT is no longer used" in l
for l in self.logs))
class TestIlmErrorReport(WorkerCase):
def test_warns_for_each_stuck_index(self):
idx = self.es.add_index("logster-events-000001",
settings={"index.lifecycle.name": "logster-events"})
idx.ilm = {"phase": "hot", "action": "rollover", "step": "ERROR",
"failed_step": "check-rollover-ready",
"step_info": {"reason": "rollover alias does not point to index"}}
retention.ilm_error_report(self.es, retention.log)
self.assertEqual(self.logs, [
"WARNING: ILM is stuck on logster-events-000001: hot/rollover failed at step "
"check-rollover-ready: rollover alias does not point to index"])
def test_unavailable_report_is_logged_not_raised(self):
self.es.fail.append(("GET", "/logster-events*", (503, "{}")))
retention.ilm_error_report(self.es, retention.log)
self.assertEqual(self.logs, ["ILM error report unavailable (HTTP 503)"])
class TestForecast(WorkerCase):
def test_forecast_uses_config_values(self):
with patch.object(retention.capacity, "compute_forecast", return_value=None) as compute:
retention.publish_capacity_forecast(
self.es, retention.log, RetentionConfig(events_days=9, target_percent=60))
kwargs = compute.call_args.kwargs
self.assertEqual((kwargs["events_days"], kwargs["target_pct"]), (9, 60))
if __name__ == "__main__":
unittest.main()
- [ ] Step 2: Run it to confirm it fails
Run: venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_retention.py' -v
Expected: failures such as AttributeError: module 'retention' has no attribute 'run_check'.
- [ ] Step 3: Replace
deploy/elasticsearch/retention.py
#!/usr/bin/env python3
"""elasticsearch-retention: the Logster retention worker.
Spec: docs/superpowers/specs/2026-09-25-es-ilm-retention-design.md.
Elasticsearch ILM deletes data by age (policies written by layout.py). This
worker does the rest, every RETENTION_CHECK_INTERVAL_MINUTES:
1. layout.converge() repair policies, templates, write aliases (self-heal)
2. ILM error report WARN for every index ILM is stuck on
3. diskcap.enforce() while disk > target, delete the oldest whole index
4. capacity forecast best effort, for the dashboard
Each step is isolated: one failing never skips the others, and nothing in
the loop can end the process. Stdlib only; runs on the
logster-elasticsearch-init image next to init.py.
"""
import os
import signal
import sys
import time
import capacity
import diskcap
import eshttp
import layout
from retention_config import load_config
log = eshttp.make_logger("retention")
TEMPLATE_DIR = (os.environ.get("ES_TEMPLATE_DIR")
or os.path.dirname(os.path.abspath(__file__)))
UNAVAILABLE_RETRY_SECONDS = 30
def ilm_error_report(request, log):
status, doc = eshttp.call(
request, "GET", f"/{layout.READ_PATTERNS}/_ilm/explain?only_errors=true", timeout=30)
if status == 404:
return
if status != 200 or not isinstance(doc, dict):
log(f"ILM error report unavailable (HTTP {status})")
return
for name, info in sorted((doc.get("indices") or {}).items()):
info = info or {}
reason = str((info.get("step_info") or {}).get("reason") or "")[:200]
log(f"WARNING: ILM is stuck on {name}: {info.get('phase')}/{info.get('action')} "
f"failed at step {info.get('failed_step')}: {reason}")
def publish_capacity_forecast(request, log, cfg):
forecast = capacity.compute_forecast(
request, log,
events_days=cfg.events_days, inferences_days=cfg.inferences_days,
graphs_days=cfg.graphs_days, target_pct=cfg.target_percent,
usage=eshttp.fs_usage(request),
es_store_bytes=capacity.managed_store_bytes(request))
if forecast is not None:
capacity.publish(request, log, forecast)
def run_check(request, log, cfg, template_dir):
def converge_step():
result = layout.converge(request, log, cfg, template_dir)
for message in result.fatal + result.problems:
log(f"WARNING: layout: {message}")
steps = (
("layout converge", converge_step),
("ILM error report", lambda: ilm_error_report(request, log)),
("disk rule", lambda: diskcap.enforce(request, log, cfg)),
("capacity forecast", lambda: publish_capacity_forecast(request, log, cfg)),
)
for name, step in steps:
try:
step()
except Exception as exc: # noqa: BLE001 — isolation is the point
log(f"ERROR: {name} failed: {exc!r}")
def _exit_on_sigterm(signum, frame):
# The worker is PID 1 in its container, and the kernel ignores signals
# with no handler for PID 1 — without this, a rollout leaves the old
# worker running for the whole 30 s grace period next to the new one.
log("SIGTERM received — exiting")
sys.exit(0)
def main():
signal.signal(signal.SIGTERM, _exit_on_sigterm)
cfg = load_config(os.environ, log)
log(f"starting — {cfg.describe()}")
run_once = (os.environ.get("RETENTION_RUN_ONCE") or "").lower() in ("1", "true", "yes")
while True:
reachable = False
try:
status = eshttp.es_status(eshttp.request)
reachable = status is not None and status < 400
if reachable:
run_check(eshttp.request, log, cfg, TEMPLATE_DIR)
else:
hint = " (401: check ES_PASSWORD)" if status == 401 else ""
log(f"Elasticsearch unavailable (HTTP {status}){hint} — retrying in "
f"{UNAVAILABLE_RETRY_SECONDS}s")
except Exception as exc: # noqa: BLE001 — the worker must never exit
log(f"ERROR: check failed: {exc!r}")
if run_once:
return
# While ES is starting (every install/upgrade restarts it), retry
# quickly: the Vector initContainer waits for the aliases this creates.
time.sleep(cfg.check_interval_minutes * 60 if reachable else UNAVAILABLE_RETRY_SECONDS)
if __name__ == "__main__":
main()
- [ ] Step 4: Run the whole ES suite
Run: venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_*.py' -v
Expected: all OK. test_init.py doesn't exist yet. The old init.py still imports and runs as before.
- [ ] Step 5: Stage
feat(retention): isolated converge / ILM report / disk rule / forecast loop
Task 8: Rewrite init.py on top of layout.converge()
Files:
- Modify: deploy/elasticsearch/init.py (full replacement)
- Test: deploy/elasticsearch/test_init.py (new)
Interfaces:
- Consumes: eshttp.*, layout.converge/alias_members/writer_of/write_alias/FAMILIES, retention_config.load_config
- Produces: init.main(). It exits via sys.exit(1) only in the §5.1 cases; otherwise it returns. layout.converge() gets up to CONVERGE_ATTEMPTS = 5 attempts, CONVERGE_RETRY_SECONDS = 10 apart (via time.sleep, which the tests patch), while result.fatal is non-empty; init exits 1 only if the fifth attempt is still fatal. A rejected /_cluster/reroute logs a WARNING and init continues to done. (exit 0).
- [ ] Step 1: Write the failing test
deploy/elasticsearch/test_init.py
"""init.py bootstrap tests (spec §5.1 exit semantics)."""
import json
import os
import unittest
from unittest.mock import call, patch
import eshttp
import init
from fake_es import FakeES
class InitCase(unittest.TestCase):
def setUp(self):
self.es = FakeES()
self.logs = []
patch.object(eshttp, "request", self.es).start()
patch.object(init, "log", lambda *p: self.logs.append(" ".join(map(str, p)))).start()
self.sleep = patch.object(init.time, "sleep").start()
patch.dict(os.environ, {}, clear=True).start()
self.addCleanup(patch.stopall)
def run_init(self):
try:
init.main()
return 0
except SystemExit as exc:
return exc.code
class TestInit(InitCase):
def test_happy_path_converges_and_exits_zero(self):
self.assertEqual(self.run_init(), 0)
self.assertEqual(self.es.cluster_settings[
"cluster.routing.allocation.disk.watermark.flood_stage"], "95%")
for family in ("events", "inferences", "graphs"):
self.assertIn(f"logster-{family}-write",
self.es.indices[f"logster-{family}-000001"].aliases)
self.assertIn("logster-capacity", self.es.indices)
health = [p for _, p, _ in self.es.calls if p.startswith("/_cluster/health/")]
self.assertEqual(health, ["/_cluster/health/logster-events-000001,logster-inferences-"
"000001,logster-graphs-000001?wait_for_status=green&timeout=60s"])
self.assertEqual(self.logs[-1], "done.")
def test_es_never_reachable_exits_one(self):
self.es.reachable = False
self.assertEqual(self.run_init(), 1)
self.assertEqual(self.sleep.call_count, 150)
def test_401_exits_one_with_hint(self):
self.es.status_override = 401
self.assertEqual(self.run_init(), 1)
self.assertTrue(any("check ES_PASSWORD" in l for l in self.logs))
def test_rejected_watermarks_exit_one(self):
self.es.fail.append(("PUT", "/_cluster/settings", (400, "{}")))
self.assertEqual(self.run_init(), 1)
def test_fatal_converge_exits_one(self):
self.es.fail.append(("PUT", "/_index_template/logster-events", (400, json.dumps(
{"error": {"type": "x", "reason": "y"}}))))
self.assertEqual(self.run_init(), 1)
self.assertTrue(any(l.startswith("ERROR: template logster-events: PUT rejected")
for l in self.logs))
def test_squatter_is_logged_and_exits_zero(self):
self.es.add_index("logster-graphs-write")
self.assertEqual(self.run_init(), 0)
self.assertTrue(any("SQUATTER" in l for l in self.logs))
self.assertTrue(any("not every write alias has a write index" in l for l in self.logs))
def test_allocation_timeout_is_a_warning_not_a_failure(self):
self.es.fail.append(("GET", "/_cluster/health/", (408, json.dumps({"timed_out": True}))))
self.assertEqual(self.run_init(), 0)
self.assertTrue(any("not allocated within 60 s (HTTP 408)" in l for l in self.logs))
def test_days_from_env_reach_the_policies(self):
os.environ["RETENTION_GRAPHS_DAYS"] = "3"
self.run_init()
self.assertEqual(self.es.policies["logster-graphs"]["phases"]["delete"]["min_age"], "3d")
def test_rejected_reroute_is_a_warning_and_exits_zero(self):
self.es.fail.append(("POST", "/_cluster/reroute", (403, json.dumps(
{"error": {"type": "security_exception", "reason": "no"}}))))
self.assertEqual(self.run_init(), 0)
self.assertTrue(any(l.startswith("WARNING: cluster reroute rejected (HTTP 403)")
for l in self.logs))
self.assertEqual(self.logs[-1], "done.")
def test_transient_fatal_converge_is_retried_then_exits_zero(self):
# A node that has just come up fails the policy GET once; the retry
# 10 s later succeeds. The patched sleep clears the failure instead
# of sleeping for real.
self.es.fail.append(("GET", "/_ilm/policy/logster-events", (503, "{}")))
self.sleep.side_effect = lambda seconds: self.es.fail.clear()
self.assertEqual(self.run_init(), 0)
self.assertEqual(self.sleep.call_args_list, [call(10)])
self.assertTrue(any("layout not established (attempt 1/5)" in l for l in self.logs))
self.assertIn("logster-events", self.es.policies)
self.assertEqual(self.logs[-1], "done.")
def test_persistent_fatal_converge_exits_one_after_five_attempts(self):
self.es.fail.append(("GET", "/_ilm/policy/logster-events", (503, "{}")))
self.assertEqual(self.run_init(), 1)
self.assertEqual(self.sleep.call_args_list, [call(10)] * 4)
gets = [p for m, p, _ in self.es.calls
if (m, p) == ("GET", "/_ilm/policy/logster-events")]
self.assertEqual(len(gets), 5)
self.assertTrue(any(l.startswith("ERROR: policy logster-events: GET failed (HTTP 503)")
for l in self.logs))
self.assertNotIn("done.", self.logs)
if __name__ == "__main__":
unittest.main()
The existing expectations still hold with the retry: test_es_never_reachable_exits_one fails in wait_for_es before any converge (150 sleeps, none of 10 s), and test_fatal_converge_exits_one (a template PUT rejected on every attempt) exits 1 after five attempts with the same ERROR: line — the patched sleep makes the four 10 s waits instant.
- [ ] Step 2: Run it to confirm it fails
Run: venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_init.py' -v
Expected: failures. The old init.main() has no log, and it tries real HTTP to elasticsearch:9200.
- [ ] Step 3: Replace
deploy/elasticsearch/init.py
#!/usr/bin/env python3
"""elasticsearch-init: one-shot bootstrap for the Logster ES cluster.
Stdlib only; runs on logster-elasticsearch-init (alpine + python3). Runs on
every Compose boot and every Helm install/upgrade, so it is idempotent:
1. Wait for Elasticsearch (up to 5 minutes).
2. Persistent disk watermarks (85/90/95 %).
3. layout.converge(): ILM policies, index templates, legacy adoption,
write aliases (spec §7.1) — up to 5 attempts, 10 s apart, while
the layout is not established.
4. Pre-create logster-capacity (while the disk is healthy — above the high
watermark a new index cannot allocate); force replicas=0 on
logster-* / .kibana* (single node).
5. Reroute with retry_failed (cold first runs can exhaust allocation
retries while ES warms up). A rejection is only a warning.
6. Wait up to 60 s for the three write indices to be allocated.
Exit codes (spec §5.1): non-zero ONLY when Elasticsearch is unreachable or
rejects credentials, or a PUT of cluster settings, a policy, a template or a
bootstrap index is rejected — misconfigurations that break the pipeline
anyway. converge() gets up to five attempts (~50 s) first. A failed policy,
template or alias GET or rollover-index listing counts as "unreachable"; a
missing template file or a crash in the policy, template or write-alias
step as "misconfiguration": exiting 0 without a write alias would let Compose's
Vector write into a 404 that it never retries. A rejected reroute, an
allocation timeout, a squatting index, or a legacy-adoption or orphan-release
problem is logged and exits 0: on Compose/OVA init gates Vector and does not
re-run until the next boot, so failing would block ingestion even after the
retention worker has freed space. The worker re-runs converge() every check.
"""
import json
import os
import sys
import time
import eshttp
import layout
from retention_config import load_config
log = eshttp.make_logger("init")
TEMPLATE_DIR = (os.environ.get("ES_TEMPLATE_DIR")
or os.path.dirname(os.path.abspath(__file__)))
# Flood stage (95 %) is the single-node safety net: ES blocks writes there
# instead of corrupting data. The retention worker keeps disk under its
# target (75 % by default) well before that.
WATERMARKS = {
"cluster.routing.allocation.disk.threshold_enabled": True,
"cluster.routing.allocation.disk.watermark.low": "85%",
"cluster.routing.allocation.disk.watermark.high": "90%",
"cluster.routing.allocation.disk.watermark.flood_stage": "95%",
}
# A node that has only just come up can still fail GETs/PUTs while its
# shards recover: converge() gets this many attempts, this far apart.
CONVERGE_ATTEMPTS = 5
CONVERGE_RETRY_SECONDS = 10
def fail(message):
log(f"ERROR: {message}")
sys.exit(1)
def wait_for_es():
status = None
for _ in range(150): # 150 × 2 s = 5 minutes
status = eshttp.es_status(eshttp.request)
if status is not None and status < 400:
return
time.sleep(2)
hint = " (401: wrong credentials — check ES_PASSWORD)" if status == 401 else ""
fail(f"Elasticsearch did not become available within 5 minutes "
f"(last HTTP {status}){hint}")
def wait_for_write_indices():
names = []
for family in layout.FAMILIES:
state, members = layout.alias_members(eshttp.request, layout.write_alias(family))
writer = layout.writer_of(members) if state == "ok" else None
if writer:
names.append(writer)
if len(names) != len(layout.FAMILIES):
log("WARNING: not every write alias has a write index yet — "
"the retention worker repairs it on its next check")
return
status, doc = eshttp.call(
eshttp.request, "GET",
f"/_cluster/health/{','.join(names)}?wait_for_status=green&timeout=60s",
timeout=90)
if status == 200 and isinstance(doc, dict) and not doc.get("timed_out"):
log(f"write indices allocated: {', '.join(names)}")
else:
log(f"WARNING: write indices not allocated within 60 s (HTTP {status}) — "
"usually disk above the 90% high watermark; Vector retries its writes "
"and the retention worker frees space")
def converge_layout(cfg):
"""layout.converge(), retried while the layout is not established."""
for attempt in range(1, CONVERGE_ATTEMPTS + 1):
result = layout.converge(eshttp.request, log, cfg, TEMPLATE_DIR)
if not result.fatal or attempt == CONVERGE_ATTEMPTS:
return result
log(f"WARNING: layout not established (attempt {attempt}/{CONVERGE_ATTEMPTS}): "
f"{'; '.join(result.fatal)} — retrying in {CONVERGE_RETRY_SECONDS} s")
time.sleep(CONVERGE_RETRY_SECONDS)
def main():
log(f"waiting for Elasticsearch at {eshttp.ES}")
wait_for_es()
status, _ = eshttp.call(eshttp.request, "PUT", "/_cluster/settings",
{"persistent": WATERMARKS})
if status != 200:
fail(f"disk-watermark settings rejected (HTTP {status})")
cfg = load_config(os.environ, log)
log(f"converging layout — {cfg.describe()}")
result = converge_layout(cfg)
for problem in result.problems:
log(f"WARNING: {problem}")
if result.fatal:
for message in result.fatal:
log(f"ERROR: {message}")
sys.exit(1)
# 400 (resource_already_exists) on later runs is fine.
eshttp.request("PUT", "/logster-capacity", "{}")
# Tolerated on failure (e.g. a flood-stage block on an old index).
eshttp.request("PUT", "/logster-*,.kibana*/_settings"
"?ignore_unavailable=true&allow_no_indices=true",
json.dumps({"index": {"number_of_replicas": 0}}))
# Only a nudge: ES retries allocation on its own, so a rejected reroute
# must not block ingestion (spec §5.1).
status, _ = eshttp.call(eshttp.request, "POST", "/_cluster/reroute?retry_failed=true")
if status != 200:
log(f"WARNING: cluster reroute rejected (HTTP {status}) — continuing")
wait_for_write_indices()
log("done.")
if __name__ == "__main__":
main()
- [ ] Step 4: Run the whole ES suite
Run: venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_*.py' -v
Expected: all OK.
- [ ] Step 5: Stage
feat(retention): init converges the ILM layout; fails only on misconfiguration
Task 9: Real-image integration suite
Files:
- Create: deploy/elasticsearch/integration/run.sh, deploy/elasticsearch/integration/test_es_integration.py
Interfaces:
- Consumes: the scripts exactly as shipped (eshttp, layout, diskcap, retention_config)
- Produces: helpers used by Task 10: LIVE, TEMPLATES, log, call, now_ms, bulk, seed, wait_until, reset_cluster, writer_of, watermarks_below, index_setting
This task runs a real ES from the image we ship (deploy-elasticsearch locally, or logster-elasticsearch:<tag> from a release), with the data directory on a 1 GiB tmpfs, so disk % is real. It proves the spec's unverified assumptions: the -E auto-create guard, the keep-forever policy body, origination_date re-PUT (its effect, read back through _ilm/explain), rollover driven by the ILM policy itself, indexing_complete releasing an orphan, adoption releasing a legacy-named index created during a rollback to v1.2.1 (test_14), GET /_alias/<name> answering 404 for a concrete index squatting on a write-alias name, which the squatter check and layout.py wait rely on (test_15), flood-stage deletion and release, and the best_compression saving.
- [ ] Step 1: Create
deploy/elasticsearch/integration/run.shandchmod +xit
#!/usr/bin/env bash
# Real-image integration suite for the ES layout / retention scripts
# (spec §12.2). Starts a throwaway Elasticsearch from the SAME image we ship,
# data dir on a size-limited tmpfs so disk % is real and ES-dominated, runs
# the scenarios, and removes everything. Never uses docker compose: a compose
# project named "deploy" would join the live stack.
#
# deploy/elasticsearch/integration/run.sh
# ES_IT_IMAGE=docker.io/eunomatix/logster-elasticsearch:v1.3.0 deploy/elasticsearch/integration/run.sh
set -euo pipefail
HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
IMAGE="${ES_IT_IMAGE:-deploy-elasticsearch}"
VECTOR_IMAGE="${ES_IT_VECTOR_IMAGE:-timberio/vector:0.52.0-distroless-static}"
RUN_ID="logster-it-$$"
NET="$RUN_ID-net"
ES_NAME="$RUN_ID-es"
PORT="${ES_IT_PORT:-19200}"
PASS="it-password-$$"
# Must match deploy/docker-compose.yml and the chart's statefulset.yaml.
GUARD='-Eaction.auto_create_index=-logster-events,-logster-inferences,-logster-graphs,-logster-*-write,+*'
cleanup() {
docker ps -aq --filter "name=$RUN_ID" | xargs -r docker rm -f >/dev/null 2>&1 || true
docker network rm "$NET" >/dev/null 2>&1 || true
}
trap cleanup EXIT
docker image inspect "$IMAGE" >/dev/null 2>&1 \
|| { echo "image $IMAGE not found — build it: (cd deploy && docker compose build elasticsearch)" >&2; exit 2; }
docker network create "$NET" >/dev/null
docker run -d --name "$ES_NAME" --network "$NET" -p "127.0.0.1:$PORT:9200" \
--tmpfs /usr/share/elasticsearch/data:rw,size=1g,uid=1000,gid=0 \
-e ELASTIC_PASSWORD="$PASS" -e ES_JAVA_OPTS="-Xms512m -Xmx512m" \
"$IMAGE" elasticsearch \
-Ediscovery.type=single-node \
-Expack.security.enabled=true \
-Expack.security.autoconfiguration.enabled=false \
-Expack.security.http.ssl.enabled=false \
-Expack.security.transport.ssl.enabled=false \
-Expack.ml.enabled=false \
"$GUARD" \
-Eindices.lifecycle.poll_interval=5s \
-Ecluster.info.update.interval=10s >/dev/null
echo "waiting for Elasticsearch ($IMAGE) on 127.0.0.1:$PORT"
for _ in $(seq 1 90); do
curl -fsS -u "elastic:$PASS" "http://127.0.0.1:$PORT/_cluster/health" >/dev/null 2>&1 && break
sleep 2
done
curl -fsS -u "elastic:$PASS" \
"http://127.0.0.1:$PORT/_cluster/health?wait_for_status=green&timeout=60s" >/dev/null
export LOGSTER_ES_IT=1 ES_HOST="http://127.0.0.1:$PORT" ES_PASSWORD="$PASS" \
LOGSTER_ES_IT_NET="$NET" LOGSTER_ES_IT_ES_NAME="$ES_NAME" \
LOGSTER_ES_IT_VECTOR_IMAGE="$VECTOR_IMAGE" LOGSTER_ES_IT_RUN_ID="$RUN_ID"
python3 -m unittest discover -s "$HERE" -p 'test_*integration.py' -v
- [ ] Step 2: Create
deploy/elasticsearch/integration/test_es_integration.py
"""Real-Elasticsearch integration scenarios (spec §12.2).
Run through run.sh; skipped unless LOGSTER_ES_IT=1. Exercises the scripts
exactly as shipped against the image we ship, on a 1 GiB tmpfs data dir.
"""
import base64
import json
import os
import sys
import threading
import time
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import diskcap # noqa: E402
import eshttp # noqa: E402
import layout # noqa: E402
from retention_config import RetentionConfig # noqa: E402
LIVE = os.environ.get("LOGSTER_ES_IT") == "1"
TEMPLATES = str(Path(__file__).resolve().parent.parent)
HOUR_MS = 3_600_000
DAY_MS = 24 * HOUR_MS
LOGS = []
def log(*parts):
line = " ".join(str(p) for p in parts)
LOGS.append(line)
print(" ", line)
def call(method, path, body=None, timeout=60):
return eshttp.call(eshttp.request, method, path, body, timeout)
def now_ms():
return int(time.time() * 1000)
def bulk(target, docs):
lines = []
for doc in docs:
lines += [json.dumps({"index": {}}), json.dumps(doc)]
status, raw = eshttp.request("POST", f"/{target}/_bulk?refresh=true",
"\n".join(lines) + "\n", 300)
try:
return status, json.loads(raw)
except ValueError:
return status, {"raw": raw}
def seed(target, n, ts_ms, payload_bytes=0):
"""n docs at @timestamp ts_ms, each with payload_bytes of random data
(incompressible, so store sizes are predictable)."""
for start in range(0, n, 2000):
docs = [{"@timestamp": ts_ms, "timestamp": ts_ms / 1000,
"blob": base64.b64encode(os.urandom(payload_bytes)).decode()}
for _ in range(min(2000, n - start))]
status, res = bulk(target, docs)
assert status == 200 and not res.get("errors"), (status, str(res)[:500])
def index_setting(index, key):
status, doc = call("GET", f"/{index}/_settings?flat_settings=true")
return ((doc or {}).get(index) or {}).get("settings", {}).get(key)
def wait_until(predicate, timeout=120, step=2, what="condition"):
deadline = time.time() + timeout
while time.time() < deadline:
if predicate():
return
time.sleep(step)
raise AssertionError(f"timed out after {timeout}s waiting for {what}\n" + "\n".join(LOGS))
def watermarks_below(pct):
call("PUT", "/_cluster/settings", {"transient": {
"cluster.routing.allocation.disk.watermark.low": f"{pct - 3:.1f}%",
"cluster.routing.allocation.disk.watermark.high": f"{pct - 2:.1f}%",
"cluster.routing.allocation.disk.watermark.flood_stage": f"{pct - 1:.1f}%"}})
def reset_cluster():
call("PUT", "/_cluster/settings", {"transient": {
"cluster.routing.allocation.disk.watermark.low": None,
"cluster.routing.allocation.disk.watermark.high": None,
"cluster.routing.allocation.disk.watermark.flood_stage": None}})
# test_04 speeds ILM up; back to the node's -E value (5 s).
call("PUT", "/_cluster/settings", {"persistent": {
"indices.lifecycle.poll_interval": None}})
status, rows = call("GET", "/_cat/indices/logster-*,zz-*?format=json&h=index"
"&expand_wildcards=all")
for row in rows or []:
call("DELETE", f"/{row['index']}")
for family in ("events", "inferences", "graphs"):
for name in (f"logster-{family}", f"logster-{family}-legacy"):
call("DELETE", f"/_ilm/policy/{name}")
for path in Path(TEMPLATES).glob("*.template.json"):
call("DELETE", f"/_index_template/{path.name[:-len('.template.json')]}")
LOGS.clear()
def writer_of(family):
state, members = layout.alias_members(eshttp.request, layout.write_alias(family))
return layout.writer_of(members) if state == "ok" else None
def realistic_events(n):
return [{
"@timestamp": now_ms(), "event_id": f"{i:032x}", "tenant_id": "default",
"endpoint_id": f"WIN-{i % 50:04d}", "platform": "windows", "event_type": "process",
"data": {"image": f"C:\\Windows\\System32\\svc{i % 200}.exe",
"process_guid": f"{{{i:08x}-0000-0000-0000-000000000000}}",
"command_line": f"svc{i % 200}.exe -k netsvcs -p -s Svc{i % 37}",
"user": "NT AUTHORITY\\SYSTEM"},
"raw_event": {"winlog": {"event_id": 1, "channel": "Microsoft-Windows-Sysmon/Operational",
"event_data": {"Image": f"C:\\Windows\\System32\\svc{i % 200}.exe",
"Hashes": f"SHA256={i:064x}"}}},
} for i in range(n)]
@unittest.skipUnless(LIVE, "set LOGSTER_ES_IT=1 (run integration/run.sh)")
class LiveCase(unittest.TestCase):
def setUp(self):
reset_cluster()
def converge(self, cfg=None):
result = layout.converge(eshttp.request, log, cfg or RetentionConfig(), TEMPLATES)
self.assertEqual(result.fatal, [], "\n".join(LOGS))
return result
class TestLayoutLive(LiveCase):
def test_01_fresh_converge_then_idempotent(self):
self.converge()
for family in ("events", "inferences", "graphs"):
index = f"logster-{family}-000001"
self.assertEqual(writer_of(family), index)
self.assertEqual(index_setting(index, "index.lifecycle.name"), f"logster-{family}")
self.assertEqual(index_setting(index, "index.codec"), "best_compression")
again = self.converge()
self.assertEqual((again.problems, again.changes), ([], []))
def test_02_auto_create_guard_blocks_alias_and_legacy_names(self):
for name in ("logster-events-write", "logster-events", "logster-graphs-write"):
with self.subTest(name=name):
status, res = bulk(name, [{"@timestamp": now_ms()}])
item = res["items"][0]["index"]
self.assertEqual(item["status"], 404, item)
self.assertEqual(item["error"]["type"], "index_not_found_exception")
self.assertEqual(call("GET", f"/{name}")[0], 404)
def test_03_legacy_adoption_origination_update_and_keep_forever(self):
old = now_ms() - 3 * DAY_MS
self.assertEqual(call("PUT", "/logster-events", {})[0], 200) # explicit create is allowed
seed("logster-events", 10, old)
self.converge()
self.assertEqual(index_setting("logster-events", "index.lifecycle.name"),
"logster-events-legacy")
self.assertEqual(index_setting("logster-events", "index.lifecycle.origination_date"),
str(old))
# ES accepts indexing_complete on a legacy index outside any rollover.
self.assertEqual(index_setting("logster-events", "index.lifecycle.indexing_complete"),
"true")
newer = now_ms() - DAY_MS
seed("logster-events", 1, newer)
self.converge()
self.assertEqual(index_setting("logster-events", "index.lifecycle.origination_date"),
str(newer))
# The re-PUT takes effect, not just the stored setting: ILM now ages the
# index from `newer` (spec §12.2 scenario 8; §7.1 step 3 relies on it).
lifecycle_date = lambda: call("GET", "/logster-events/_ilm/explain")[1][
"indices"]["logster-events"].get("lifecycle_date_millis")
wait_until(lambda: lifecycle_date() == newer, timeout=30,
what="ILM to age logster-events from the re-PUT origination_date")
self.converge(RetentionConfig(events_days=0)) # ES must accept keep-forever bodies
policy = call("GET", "/_ilm/policy/logster-events-legacy")[1][
"logster-events-legacy"]["policy"]
self.assertNotIn("delete", policy["phases"])
def test_04_ilm_rollover_moves_writes_and_reads_span_both(self):
"""Spec §12.2 scenario 4: the ILM policy itself rolls the index over
(no manual _rollover), through the rollover_alias + policy wiring
that converge() set up."""
self.converge()
# Poll ILM every second (reset_cluster() restores the node's 5 s) and
# override the shipped policy so its hot rollover also fires at one
# document. Everything else in the policy is kept. Dropping _meta
# makes the next converge() see a hash mismatch and restore it.
self.assertEqual(call("PUT", "/_cluster/settings", {"persistent": {
"indices.lifecycle.poll_interval": "1s"}})[0], 200)
policy = call("GET", "/_ilm/policy/logster-events")[1]["logster-events"]["policy"]
policy.pop("_meta", None)
policy["phases"]["hot"]["actions"]["rollover"]["max_docs"] = 1
self.assertEqual(call("PUT", "/_ilm/policy/logster-events", {"policy": policy})[0], 200)
seed("logster-events-write", 5, now_ms())
wait_until(lambda: writer_of("events") == "logster-events-000002", timeout=120,
what="ILM to roll logster-events over to 000002 (max_docs: 1)")
# Restore the shipped policy, so 000002 stays the write index for the
# rest of this test and no rollover is in flight when the next test
# resets the cluster.
self.converge()
# The new write index is still empty: a sort across the family must
# not fail a shard (spec §6.2 explicit mappings).
for field in ("@timestamp", "timestamp"):
status, res = call("POST", "/logster-events*/_search",
{"size": 1, "sort": [{field: "desc"}]})
self.assertEqual((status, res["_shards"]["failed"]), (200, 0), res)
seed("logster-events-write", 3, now_ms())
self.assertEqual(call("GET", "/logster-events*/_count")[1]["count"], 8)
self.assertEqual(call("GET", "/logster-events-000002/_count")[1]["count"], 3)
def test_05_concurrent_converge_creates_exactly_one_write_index(self):
results = []
threads = [threading.Thread(target=lambda: results.append(
layout.converge(eshttp.request, log, RetentionConfig(), TEMPLATES)))
for _ in range(2)]
for t in threads:
t.start()
for t in threads:
t.join()
self.assertEqual([r.fatal for r in results], [[], []], "\n".join(LOGS))
for family in ("events", "inferences", "graphs"):
status, rows = call("GET", f"/_cat/indices/logster-{family}-*?format=json&h=index")
self.assertEqual([r["index"] for r in rows], [f"logster-{family}-000001"])
def test_06_orphaned_write_index_is_released_and_reaches_delete(self):
cfg = RetentionConfig(events_days=1)
self.converge(cfg)
seed("logster-events-write", 1, now_ms())
call("POST", "/_aliases", {"actions": [{"remove": {
"index": "logster-events-000001", "alias": "logster-events-write"}}]})
call("PUT", "/logster-events-000001/_settings",
{"index.lifecycle.origination_date": now_ms() - 3 * DAY_MS})
explain = lambda: call("GET", "/logster-events-000001/_ilm/explain")[1][
"indices"]["logster-events-000001"]
wait_until(lambda: explain().get("step") == "ERROR",
what="the orphan to hit ILM's rollover error")
self.converge(cfg)
self.assertEqual(writer_of("events"), "logster-events-000002")
wait_until(lambda: call("GET", "/logster-events-000001")[0] == 404, timeout=180,
what="ILM to delete the released orphan")
def test_09_ilm_deletes_adopted_legacy_by_age(self):
self.assertEqual(call("PUT", "/logster-graphs", {})[0], 200)
seed("logster-graphs", 5, now_ms() - 5 * DAY_MS)
self.converge(RetentionConfig(graphs_days=1))
wait_until(lambda: call("GET", "/logster-graphs")[0] == 404, timeout=180,
what="ILM to delete the adopted legacy index")
def test_10_self_heal_after_write_index_deleted(self):
self.converge()
seed("logster-inferences-write", 2, now_ms())
call("POST", "/logster-inferences-write/_rollover")
self.assertEqual(call("DELETE", "/logster-inferences-000002")[0], 200)
status, res = bulk("logster-inferences-write", [{"@timestamp": now_ms()}])
self.assertTrue(status >= 400 or res.get("errors"), res) # no write index → rejected
self.converge()
self.assertEqual(writer_of("inferences"), "logster-inferences-000002")
status, res = bulk("logster-inferences-write", [{"@timestamp": now_ms()}])
self.assertFalse(res.get("errors"), res)
def test_11_best_compression_saving_on_realistic_events(self):
events = realistic_events(20_000)
for name, codec in (("zz-codec-default", "default"),
("zz-codec-best", "best_compression")):
call("PUT", f"/{name}", {"settings": {"index.codec": codec,
"index.number_of_replicas": 0}})
for start in range(0, len(events), 2000):
bulk(name, events[start:start + 2000])
call("POST", f"/{name}/_forcemerge?max_num_segments=1", timeout=300)
call("POST", "/_flush")
rows = call("GET", "/_cat/indices/zz-codec-*?format=json&bytes=b&h=index,store.size")[1]
sizes = {r["index"]: int(r["store.size"]) for r in rows}
ratio = sizes["zz-codec-best"] / sizes["zz-codec-default"]
print(f"\n best_compression / default store size = {ratio:.3f} (record in the PR)")
self.assertLess(ratio, 1.0)
def test_14_rollback_created_legacy_index_is_released_and_deleted(self):
"""Upgrade → the legacy index ages out → helm rollback to v1.2.1 (its
init hook does not run on rollback, so the new templates stay) → its
writers create the bare logster-events again, which picks up the
family policy + rollover_alias from the new template, so ILM's
hot/rollover step errors on it forever. Roll-forward adoption
(legacy policy + indexing_complete) must release it; ILM then ages it
out."""
self.converge()
self.assertEqual(call("PUT", "/_cluster/settings", {"persistent": {
"indices.lifecycle.poll_interval": "1s"}})[0], 200) # reset_cluster() restores 5 s
# v1.2.1 auto-creates it; the guard refuses auto-create here, so create
# it explicitly — the template applies at creation either way.
self.assertEqual(call("PUT", "/logster-events", {})[0], 200)
self.assertEqual(index_setting("logster-events", "index.lifecycle.name"), "logster-events")
self.assertEqual(index_setting("logster-events", "index.lifecycle.rollover_alias"),
"logster-events-write")
seed("logster-events", 1, now_ms() - 60 * DAY_MS) # older than events_days=30
explain = lambda: call("GET", "/logster-events/_ilm/explain")[1][
"indices"]["logster-events"]
errors = []
def rollover_error():
# ILM auto-retries, so ERROR and check-rollover-ready alternate.
info = explain()
if info.get("action") == "rollover" and info.get("step") == "ERROR":
errors.append(info)
return bool(errors)
wait_until(rollover_error, timeout=60, step=0.5,
what="the rolled-back logster-events to hit ILM's rollover error")
self.assertIn("rollover_alias", json.dumps(errors[0].get("step_info")))
self.converge() # roll forward
self.assertEqual(index_setting("logster-events", "index.lifecycle.name"),
"logster-events-legacy")
self.assertEqual(index_setting("logster-events", "index.lifecycle.indexing_complete"),
"true")
# Released from hot/rollover, through the legacy delete phase (60 d > 30 d).
wait_until(lambda: call("GET", "/logster-events")[0] == 404, timeout=60,
what="ILM to release the adopted index from rollover and delete it")
self.assertEqual(writer_of("events"), "logster-events-000001")
def test_15_squatted_alias_name_is_reported_and_counts_as_ready(self):
"""The squatter check relies on GET /_alias/<name> answering 404 when
<name> is a concrete index. The guard blocks only auto-create, so an
explicit create still squats."""
self.assertEqual(call("PUT", "/logster-graphs-write", {})[0], 200)
self.addCleanup(call, "DELETE", "/logster-graphs-write")
self.assertEqual(call("GET", "/_alias/logster-graphs-write")[0], 404)
result = self.converge()
self.assertTrue(any(p.startswith("SQUATTER: a concrete index named logster-graphs-write")
for p in result.problems), result.problems)
# No bootstrap for the squatted family; the others are bootstrapped.
self.assertEqual(call("GET", "/_cat/indices/logster-graphs-0*?format=json&h=index")[1], [])
self.assertEqual(writer_of("events"), "logster-events-000001")
self.assertTrue(layout.write_aliases_ready(eshttp.request))
lines = []
self.assertTrue(layout.wait_for_aliases(
eshttp.request, lambda *p: lines.append(" ".join(map(str, p))), max_wait_seconds=0))
self.assertEqual([l for l in lines if "SQUATTER" in l],
["ERROR: " + layout.squatter_message("logster-graphs-write")])
class TestDiskRuleLive(LiveCase):
def build_generations(self):
"""events 000001..000003 hold data 10 d / 8 d / 1 h old (~50 MB
each); 000004 is the empty write index."""
self.converge()
for age in (10 * DAY_MS, 8 * DAY_MS, HOUR_MS):
seed("logster-events-write", 40_000, now_ms() - age, payload_bytes=1000)
self.assertEqual(call("POST", "/logster-events-write/_rollover")[0], 200)
call("POST", "/_flush")
rows = call("GET", "/_cat/indices/logster-events-*?format=json&bytes=b&h=index,store.size")[1]
return {r["index"]: int(r["store.size"]) for r in rows}
def target_for(self, free_bytes):
used, total, _ = eshttp.fs_usage(eshttp.request)
return int((used - free_bytes) / total * 100)
def test_07_deletes_oldest_non_write_and_honors_the_floor(self):
sizes = self.build_generations()
target = self.target_for(sizes["logster-events-000001"]
+ sizes["logster-events-000002"] // 2)
out = diskcap.enforce(eshttp.request, log, RetentionConfig(target_percent=target))
self.assertEqual(out.deleted, ("logster-events-000001", "logster-events-000002"),
"\n".join(LOGS))
for kept in ("logster-events-000003", "logster-events-000004"):
self.assertEqual(call("GET", f"/{kept}")[0], 200)
def test_08_real_flood_stage_block_deletion_and_release(self):
sizes = self.build_generations()
_, _, pct = eshttp.fs_usage(eshttp.request)
watermarks_below(pct)
blocked = lambda: index_setting("logster-events-000004",
"index.blocks.read_only_allow_delete") == "true"
wait_until(blocked, what="the flood-stage block")
status, res = bulk("logster-events-write", [{"@timestamp": now_ms()}])
self.assertEqual(res["items"][0]["index"]["status"], 429, res)
target = self.target_for(sizes["logster-events-000001"]
+ sizes["logster-events-000002"] // 2)
out = diskcap.enforce(eshttp.request, log, RetentionConfig(target_percent=target))
self.assertEqual(len(out.deleted), 2, "\n".join(LOGS))
wait_until(lambda: not blocked(), what="ES to release the block below the high watermark")
status, res = bulk("logster-events-write", [{"@timestamp": now_ms()}])
self.assertFalse(res.get("errors"), res)
if __name__ == "__main__":
unittest.main()
- [ ] Step 3: Confirm the suite is skipped by the normal unit run
Run: venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_*.py' 2>&1 | tail -3
Expected: OK. The integration/ directory has no __init__.py, so plain discovery doesn't recurse into it.
- [ ] Step 4: Run the suite against the real image
Run: deploy/elasticsearch/integration/run.sh 2>&1 | tail -40
Expected: every test_0x/test_1x passes, and the best_compression ratio is printed. The Vector test file doesn't exist yet. Record the ratio for the PR description.
Contingencies. Each one changes code, so report it to the user before continuing:
- test_02 fails because ES 9.5.4 rejects action.auto_create_index as a -E startup arg (the container won't start, or the setting is ignored). Remove the arg from run.sh. Add a step to init.py before layout.converge(): PUT /_cluster/settings {"persistent": {"action.auto_create_index": "<same pattern>"}}, fatal on rejection. Add a test_init.py assertion for it. Drop the arg from Task 15/16's arg lists and update spec §6.3.
- test_06 times out because ILM re-errors after indexing_complete (ES checks the alias before the flag). Change _release_orphans to move orphans onto the delete-only legacy policy instead:
1. POST /<orphan>/_ilm/remove
2. PUT /<orphan>/_settings {"index.lifecycle.name": legacy_policy_name(family), "index.lifecycle.origination_date": <newest doc ms>}
Update TestOrphans accordingly (assert those two calls instead of indexing_complete + retry) and rerun.
- test_04 times out waiting for logster-events-000002 because ES kept 000001's cached hot-phase definition after the policy PUT (ILM normally refreshes it when only the rollover conditions change). Re-attach the index so ILM starts over from the override: right after the override PUT, add call("POST", "/logster-events-000001/_ilm/remove") and call("PUT", "/logster-events-000001/_settings", {"index.lifecycle.name": "logster-events", "index.lifecycle.rollover_alias": "logster-events-write"}), then rerun.
- test_03 fails on events_days=0 (ES rejects a phase with empty actions). Change legacy_policy(0) to {"phases": {"hot": {"min_age": "0ms", "actions": {"set_priority": {"priority": 0}}}}} and update TestPolicyBodies.test_zero_days_keeps_forever.
- [ ] Step 5: Stage
git add deploy/elasticsearch/integration/run.sh deploy/elasticsearch/integration/test_es_integration.py
test(retention): real-image integration suite on a size-limited tmpfs
Task 10: Vector write aliases + request_retry_partial, proven through a real flood block
Files:
- Modify: deploy/vector/vector.yaml, deploy/elasticsearch/integration/run.sh (validate the shipped Vector config)
- Create: deploy/elasticsearch/integration/vector-it.yaml, deploy/elasticsearch/integration/test_vector_integration.py
- Test: deploy/elasticsearch/test_vector_config.py (new; stdlib unit test, found by the normal unit run and by CI)
Interfaces:
- Consumes (Task 9): LIVE, TEMPLATES, call, log, now_ms, seed, wait_until, reset_cluster, watermarks_below, index_setting
- Produces: Vector sinks writing to logster-events-write / logster-inferences-write with request_retry_partial: true
- [ ] Step 1: Write the flood-block scenario
deploy/elasticsearch/integration/vector-it.yaml
# Vector config for the flood-stage retry scenario (integration only).
data_dir: /var/lib/vector
sources:
gen:
type: demo_logs
format: shuffle
lines: ["retry-probe"]
count: ${EVENT_COUNT}
interval: 0.0
# Never ends, so Vector keeps running (and retrying) after demo_logs has
# emitted its last event, as the Kafka source does in production. Without
# it Vector starts shutting down at once, and its 60 s graceful-shutdown
# limit can cut in-flight retries short.
keepalive:
type: internal_metrics
transforms:
shape:
type: remap
inputs: [gen]
source: |
. = {"message": "retry-probe", "@timestamp": now(), "_doc_id": uuid_v4()}
sinks:
es:
type: elasticsearch
inputs: [shape]
endpoints: ["${ES_URL}"]
bulk:
index: logster-events-write
id_key: _doc_id
request_retry_partial: ${RETRY_PARTIAL}
auth:
strategy: basic
user: elastic
password: "${ES_PASSWORD}"
batch:
max_events: 50
timeout_secs: 1
keepalive_sink:
type: blackhole
inputs: [keepalive]
print_interval_secs: 0
deploy/elasticsearch/integration/test_vector_integration.py:
"""Vector 0.52 + request_retry_partial through a real flood-stage block
(spec D14, §12.2 scenario 12). Run through run.sh; skipped unless
LOGSTER_ES_IT=1."""
import os
import subprocess
import sys
import time
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
from test_es_integration import (LIVE, TEMPLATES, call, eshttp, index_setting, # noqa: E402
layout, log, now_ms, reset_cluster, seed,
wait_until, watermarks_below, RetentionConfig)
HERE = Path(__file__).resolve().parent
RUN_ID = os.environ.get("LOGSTER_ES_IT_RUN_ID", "logster-it")
def run_vector(name, retry_partial, count):
subprocess.run([
"docker", "run", "-d", "--name", f"{RUN_ID}-{name}",
"--network", os.environ["LOGSTER_ES_IT_NET"], "--tmpfs", "/var/lib/vector",
"-e", f"ES_URL=http://{os.environ['LOGSTER_ES_IT_ES_NAME']}:9200",
"-e", f"ES_PASSWORD={os.environ['ES_PASSWORD']}",
"-e", f"EVENT_COUNT={count}",
"-e", f"RETRY_PARTIAL={'true' if retry_partial else 'false'}",
"-v", f"{HERE / 'vector-it.yaml'}:/etc/vector/vector.yaml:ro",
os.environ["LOGSTER_ES_IT_VECTOR_IMAGE"]], check=True, capture_output=True)
def stop_vector(name):
subprocess.run(["docker", "rm", "-f", f"{RUN_ID}-{name}"], capture_output=True)
def probe_count():
call("POST", "/logster-events*/_refresh")
status, doc = call("POST", "/logster-events*/_count",
{"query": {"match_phrase": {"message": "retry-probe"}}})
return (doc or {}).get("count", 0)
@unittest.skipUnless(LIVE, "set LOGSTER_ES_IT=1 (run integration/run.sh)")
class TestVectorRetriesFloodStage(unittest.TestCase):
def setUp(self):
reset_cluster()
result = layout.converge(eshttp.request, log, RetentionConfig(), TEMPLATES)
self.assertEqual(result.fatal, [])
self.addCleanup(stop_vector, "drop")
self.addCleanup(stop_vector, "retry")
def test_12_flood_block_rejections_are_retried_not_dropped(self):
call("PUT", "/zz-filler", {"settings": {"index.number_of_replicas": 0}})
seed("zz-filler", 150_000, now_ms(), payload_bytes=1000)
call("POST", "/_flush")
_, _, pct = eshttp.fs_usage(eshttp.request)
watermarks_below(pct)
blocked = lambda: index_setting("logster-events-000001",
"index.blocks.read_only_allow_delete") == "true"
wait_until(blocked, what="the flood-stage block")
run_vector("drop", retry_partial=False, count=200) # control: today's config
time.sleep(20)
stop_vector("drop")
run_vector("retry", retry_partial=True, count=200) # the D14 fix
time.sleep(20)
self.assertEqual(probe_count(), 0) # still blocked: nothing has landed
self.assertEqual(call("DELETE", "/zz-filler")[0], 200) # allowed under the block
wait_until(lambda: not blocked(), timeout=120, what="the block to be released")
wait_until(lambda: probe_count() >= 200, timeout=180, what="retried events to land")
time.sleep(10)
# Exactly the retried run's 200: the control run's events were dropped.
self.assertEqual(probe_count(), 200)
if __name__ == "__main__":
unittest.main()
In deploy/elasticsearch/integration/run.sh, insert directly after the docker image inspect "$IMAGE" … statement (before docker network create), so every integration run also proves the shipped Vector config loads in the shipped Vector image (timberio/vector:0.52.0-distroless-static, the same tag as deploy/docker-compose.yml's vector service):
# The shipped Vector config must load in the shipped Vector image.
VECTOR_YAML="$(cd "$HERE/../.." && pwd)/vector/vector.yaml"
docker run --rm -v "$VECTOR_YAML:/etc/vector/vector.yaml:ro" "$VECTOR_IMAGE" \
validate --no-environment /etc/vector/vector.yaml
- [ ] Step 2: Write the failing config test
deploy/elasticsearch/test_vector_config.py
The integration scenario runs vector-it.yaml, never the shipped file, so this unit test pins the Step 4 edits in deploy/vector/vector.yaml and pins that vector-it.yaml tests the same retry configuration.
"""The shipped Vector config carries the sink settings the ILM layout needs
(spec D9, D14): writes go through the write aliases, and per-item 429/5xx
rejections are retried. Stdlib only (no PyYAML): each sink's block is read
as text.
integration/test_vector_integration.py proves the retry behaviour with
integration/vector-it.yaml. This test pins the same settings in the shipped
deploy/vector/vector.yaml, and pins that the integration config exercises
the shipped retry configuration.
"""
import re
import unittest
from pathlib import Path
HERE = Path(__file__).resolve().parent # deploy/elasticsearch
SHIPPED = HERE.parent / "vector" / "vector.yaml"
IT_CONFIG = HERE / "integration" / "vector-it.yaml"
def sink_block(path, sink):
"""The indented body of `sinks.<sink>`, as a list of lines."""
lines = path.read_text(encoding="utf-8").splitlines()
body, inside = [], False
for line in lines[lines.index("sinks:") + 1:]:
if re.match(r"^[^\s#]", line): # the next top-level key
break
if re.match(r"^ [^\s#]", line): # a sink name
inside = line.strip() == f"{sink}:"
continue
if inside:
body.append(line)
if not body:
raise AssertionError(f"sink {sink!r} not found in {path}")
return body
def setting(body, key):
"""The value of `key:` in a sink body (first match, comments ignored)."""
for line in body:
m = re.match(rf"^\s+{re.escape(key)}:\s*(.*?)\s*$", line)
if m:
return m.group(1)
return None
def retry_settings(body):
"""Every retry-related setting line except request_retry_partial."""
return sorted(line.strip() for line in body
if "retry" in line and not line.strip().startswith("#")
and "request_retry_partial" not in line)
class TestShippedVectorSinks(unittest.TestCase):
def test_sinks_write_through_the_write_aliases(self):
self.assertEqual(setting(sink_block(SHIPPED, "es_events"), "index"),
'"${EVENTS_INDEX:-logster-events-write}"')
self.assertEqual(setting(sink_block(SHIPPED, "es_inferences"), "index"),
'"${INFERENCES_INDEX:-logster-inferences-write}"')
def test_sinks_retry_partial_rejections_idempotently(self):
for sink in ("es_events", "es_inferences"):
with self.subTest(sink=sink):
body = sink_block(SHIPPED, sink)
self.assertEqual(setting(body, "request_retry_partial"), "true")
# id_key makes a retried item an overwrite, never a duplicate.
self.assertEqual(setting(body, "id_key"), "_doc_id")
def test_integration_config_exercises_the_shipped_retry_settings(self):
shipped = sink_block(SHIPPED, "es_events")
it = sink_block(IT_CONFIG, "es")
# The scenario toggles the flag: false for the control run, true for the fix.
self.assertEqual(setting(it, "request_retry_partial"), "${RETRY_PARTIAL}")
self.assertEqual(setting(it, "id_key"), setting(shipped, "id_key"))
self.assertEqual(setting(it, "index"), "logster-events-write")
self.assertEqual(retry_settings(it), retry_settings(shipped))
if __name__ == "__main__":
unittest.main()
- [ ] Step 3: Run it to confirm it fails
Run: venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_vector_config.py' -v
Expected: test_sinks_write_through_the_write_aliases fails ('"${EVENTS_INDEX:-logster-events}"' != '"${EVENTS_INDEX:-logster-events-write}"') and both subtests of test_sinks_retry_partial_rejections_idempotently fail (None != 'true'). test_integration_config_exercises_the_shipped_retry_settings passes.
- [ ] Step 4: Update
deploy/vector/vector.yaml
In es_events, change the index default and add the retry option directly under id_key: _doc_id:
bulk:
index: "${EVENTS_INDEX:-logster-events-write}"
id_key: _doc_id
# Retry per-item 429s (flood-stage block) and 5xx instead of dropping
# them (spec D14). id_key makes a retry an idempotent overwrite. A 404
# (missing write alias) is not retriable by Vector's design.
request_retry_partial: true
es_inferences:
bulk:
index: "${INFERENCES_INDEX:-logster-inferences-write}"
id_key: _doc_id
request_retry_partial: true
logster-events and logster-inferences, so say "the logster-events-write / logster-inferences-write aliases".
- [ ] Step 5: Run the unit suite
Run: venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_*.py' 2>&1 | tail -3
Expected: OK, including the three test_vector_config.py tests.
- [ ] Step 6: Run the integration suite three consecutive times
Each run takes several minutes, so three runs exceed the 10-minute foreground limit: start the loop with the Bash tool's run_in_background and wait for it to finish.
for i in 1 2 3; do
echo "=== integration run $i/3"
deploy/elasticsearch/integration/run.sh > "${TMPDIR:-/tmp}/logster-it-run-$i.log" 2>&1 \
|| { tail -60 "${TMPDIR:-/tmp}/logster-it-run-$i.log"; echo "run $i FAILED"; break; }
grep -E 'Validated|test_12|^Ran |^OK' "${TMPDIR:-/tmp}/logster-it-run-$i.log"
done
Validated (the updated deploy/vector/vector.yaml loads in the shipped Vector image), test_12_flood_block_rejections_are_retried_not_dropped ... ok and OK; no FAILED. test_12 proves both halves: without the flag events are dropped, with it they land exactly once. A failure in any run means the scenario is flaky: find and fix the cause, then restart the count at run 1.
If the count ends at 400, ES rejected the whole bulk request with a request-level 429, which Vector retries even without the flag. Stop and report to the user: D14 is still correct, but the spec's §13 flood row is then understated.
- [ ] Step 7: Stage
git add deploy/vector/vector.yaml deploy/elasticsearch/integration/run.sh deploy/elasticsearch/integration/vector-it.yaml deploy/elasticsearch/integration/test_vector_integration.py deploy/elasticsearch/test_vector_config.py
fix(vector): write through ILM aliases and retry flood-stage rejections
Task 11: Inference: write alias, read pattern, per-item bulk errors
Files:
- Modify: libs/logster-common/logster_common/config.py:126,150 (+ docstring line 142), deploy/service-config.yaml:65-71, libs/logster-ml/logster_ml/graph/hydration.py:116, services/inference/src/logster_inference/sinks/graph_sink.py
- Test: services/inference/tests/test_graph_sink.py (new)
Interfaces:
- Produces:
- graph_sink.summarize_bulk(payload: bytes, n_docs: int) -> tuple[int, int, str | None]
- GraphPersistConfig.es_index == "logster-graphs-write"
- PhantomHydrationConfig.es_index == "logster-events*"
- [ ] Step 1: Write the failing test
services/inference/tests/test_graph_sink.py
"""GraphSink bulk accounting and the index names the inference service uses."""
import json
from logster_common.config import GraphPersistConfig, PhantomHydrationConfig
from logster_inference.sinks.graph_sink import summarize_bulk
def item(status, etype=None):
body = {"_index": "logster-graphs-000001", "status": status}
if etype:
body["error"] = {"type": etype, "reason": f"{etype} reason"}
return {"index": body}
def test_all_indexed():
payload = json.dumps({"errors": False, "items": [item(201), item(200)]}).encode()
assert summarize_bulk(payload, 2) == (2, 0, None)
def test_per_item_failures_are_counted_with_the_first_reason():
payload = json.dumps({"errors": True, "items": [
item(201), item(404, "index_not_found_exception"),
item(429, "cluster_block_exception")]}).encode()
assert summarize_bulk(payload, 3) == (
1, 2, "404 index_not_found_exception: index_not_found_exception reason")
def test_unparseable_response_counts_every_doc_as_failed():
assert summarize_bulk(b"<html>", 4) == (0, 4, "unparseable bulk response")
def test_writes_go_to_the_alias_and_hydration_reads_the_pattern():
assert GraphPersistConfig().es_index == "logster-graphs-write"
assert PhantomHydrationConfig().es_index == "logster-events*"
- [ ] Step 2: Run it to confirm it fails
Run: venv/bin/python -m pytest services/inference/tests/test_graph_sink.py -q
Expected: ImportError: cannot import name 'summarize_bulk'.
- [ ] Step 3: Implement
services/inference/src/logster_inference/sinks/graph_sink.py:
1. Change the constructor default es_index: str = "logster-graphs" to es_index: str = "logster-graphs-write".
2. Add this function above class GraphSink:
def summarize_bulk(payload: bytes, n_docs: int) -> tuple[int, int, str | None]:
"""(indexed, failed, first_error) from a 2xx ``_bulk`` response.
A bulk request returns 200 even when individual items fail (a missing
write alias → 404, a flood-stage block → 429), so the per-item status is
what counts. An unparseable body counts every doc as failed: a broken
response must never be reported as success.
"""
try:
doc = json.loads(payload)
items = doc.get("items") or []
except (ValueError, AttributeError):
return 0, n_docs, "unparseable bulk response"
if not doc.get("errors"):
return n_docs, 0, None
failed, first = 0, None
for entry in items:
result = next(iter(entry.values()), {}) if isinstance(entry, dict) else {}
status = int(result.get("status", 500))
if status >= 300:
failed += 1
if first is None:
err = result.get("error") or {}
first = f"{status} {err.get('type', '')}: {str(err.get('reason', ''))[:200]}"
return n_docs - failed, failed, first
_flush, replace the if 200 <= resp.status < 300: block (the one that does _indexed_total.inc(len(docs))) with:
if 200 <= resp.status < 300:
indexed, failed, first = summarize_bulk(payload, len(docs))
if _indexed_total is not None and indexed:
_indexed_total.inc(indexed)
if failed:
self._record_error("item", failed)
logger.warning(
"Graph sink: %d of %d docs rejected by Elasticsearch: %s",
failed, len(docs), first,
)
# A 2xx bulk response can still carry per-item failures.
4. Change _record_error to take a count:
def _record_error(self, reason: str, count: int = 1) -> None:
if _index_errors_total is not None:
_index_errors_total.labels(reason=reason).inc(count)
libs/logster-common/logster_common/config.py:
- Line 126: es_index: str = "logster-graphs-write", with a comment above it: # ILM write alias (spec 2026-09-25-es-ilm-retention); never a concrete index.
- Line 150: es_index: str = "logster-events*", with a comment above it: # Read pattern: pre-ILM logster-events plus its rollover indices.
- In the PhantomHydrationConfig docstring, change "it queries logster-events" to "it queries the logster-events* indices".
libs/logster-ml/logster_ml/graph/hydration.py:116: index: str = "logster-events*",
deploy/service-config.yaml (phantom_hydration block): set es_index: "logster-events*", and in the comment above it change "look it up in the logster-events ES index" to "look it up in the logster-events* indices".
- [ ] Step 4: Run the tests
Run: venv/bin/python -m pytest services/inference/tests/test_graph_sink.py -q && venv/bin/python -m pytest services/inference/tests -q 2>&1 | grep -E '^FAILED|passed|failed'
Expected: the 4 new tests pass. The full inference suite ends 8 failed, 373 passed (the recorded pre-change baseline was 8 failed, 369 passed; this task adds 4 passing tests), and the FAILED lines name exactly these 8 known, pre-existing failures:
- services/inference/tests/test_script_graph_integration.py::test_v1_inference_persists_the_graph_passed_to_classifier (the user's uncommitted engine.py edit; the hardening plan's to fix)
- services/inference/tests/test_script_pinning.py::test_all_script_nodes_survive_budget_pruning
- services/inference/tests/test_service_license.py::test_process_event_consumes_license_and_buffers
- services/inference/tests/test_service_license.py::test_network_event_buffers_without_calling_check
- services/inference/tests/test_service_license.py::test_only_process_and_file_consume_license
- services/inference/tests/test_service_license.py::test_over_cap_counted_event_still_buffered
- services/inference/tests/test_service_license.py::test_expired_license_terminates_on_counted_event
- services/inference/tests/test_service_license.py::test_expired_license_does_not_terminate_on_uncounted_event
Any other FAILED line is a new failure: fix it in this task. If one of the known 8 has meanwhile been fixed elsewhere, the counts shift by that much; the check is that every FAILED line is in this list. Don't fix the 8 known ones here.
- [ ] Step 5: Stage
git add libs/logster-common/logster_common/config.py deploy/service-config.yaml libs/logster-ml/logster_ml/graph/hydration.py services/inference/src/logster_inference/sinks/graph_sink.py services/inference/tests/test_graph_sink.py
fix(inference): write graphs through the ILM alias, count per-item bulk failures
Task 12: API: graph and inference lookups over patterns, _index-targeted updates
Files:
- Modify: services/api/src/logster_api/whitelist_store.py:28-120, services/api/src/logster_api/evidence.py:365-366,450-458, services/api/tests/test_evidence_endpoint.py
- Test: services/api/tests/test_whitelist_store.py (new)
Interfaces:
- Produces:
- WhitelistStore.get_graph_doc(inference_id) -> dict|None. The dict includes _id and _index.
- WhitelistStore.update_graph_doc(doc_id, partial, *, index). Raises WhitelistStoreError for a missing or wildcard index.
- WhitelistStore.get_inference_doc(inference_id): newest copy first.
- [ ] Step 1: Write the failing test
services/api/tests/test_whitelist_store.py
"""WhitelistStore against a stub Elasticsearch: pattern reads, _index-targeted updates."""
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import pytest
from logster_api.whitelist_store import WhitelistStore, WhitelistStoreError
NEWEST_FIRST = [{"@timestamp": {"order": "desc", "unmapped_type": "date"}}]
class StubES:
def __init__(self):
self.requests = []
self.replies = {} # (method, path prefix) -> (status, body)
stub = self
class Handler(BaseHTTPRequestHandler):
def _serve(self):
length = int(self.headers.get("Content-Length") or 0)
body = json.loads(self.rfile.read(length)) if length else None
stub.requests.append((self.command, self.path, body))
status, reply = 200, {}
for (method, prefix), value in stub.replies.items():
if method == self.command and self.path.startswith(prefix):
status, reply = value
data = json.dumps(reply).encode()
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
do_GET = do_POST = do_PUT = _serve
def log_message(self, *args):
pass
self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
threading.Thread(target=self.server.serve_forever, daemon=True).start()
self.url = f"http://127.0.0.1:{self.server.server_address[1]}"
@pytest.fixture
def es():
stub = StubES()
yield stub
stub.server.shutdown()
@pytest.fixture
def store(es):
return WhitelistStore(es_host=es.url)
def hit(index, doc_id, source):
return {"hits": {"hits": [{"_index": index, "_id": doc_id, "_source": source}]}}
def test_get_graph_doc_searches_the_pattern_and_returns_the_concrete_index(es, store):
es.replies[("POST", "/logster-graphs*/_search")] = (
200, hit("logster-graphs-000003", "inf-1", {"inference_id": "inf-1", "nodes": []}))
doc = store.get_graph_doc("inf-1")
assert (doc["_index"], doc["_id"], doc["nodes"]) == ("logster-graphs-000003", "inf-1", [])
method, path, body = es.requests[-1]
assert (method, path) == ("POST", "/logster-graphs*/_search")
assert body["size"] == 1 and body["sort"] == NEWEST_FIRST
should = body["query"]["bool"]["should"]
assert {"ids": {"values": ["inf-1"]}} in should
assert {"term": {"inference_id": "inf-1"}} in should
def test_get_graph_doc_returns_none_without_hits(es, store):
es.replies[("POST", "/logster-graphs*/_search")] = (200, {"hits": {"hits": []}})
assert store.get_graph_doc("missing") is None
def test_update_graph_doc_targets_the_concrete_index(es, store):
store.update_graph_doc("inf-1", {"evidence": {"status": "success"}},
index="logster-graphs-000003")
method, path, body = es.requests[-1]
assert (method, path) == ("POST", "/logster-graphs-000003/_update/inf-1?refresh=true")
assert body == {"doc": {"evidence": {"status": "success"}}}
@pytest.mark.parametrize("index", ["", "logster-graphs*", "a,b"])
def test_update_graph_doc_refuses_patterns(es, store, index):
with pytest.raises(WhitelistStoreError):
store.update_graph_doc("inf-1", {}, index=index)
assert es.requests == []
def test_get_inference_doc_prefers_the_newest_copy(es, store):
es.replies[("POST", "/logster-inferences*/_search")] = (
200, hit("logster-inferences-000002", "x", {"inference_id": "inf-1"}))
assert store.get_inference_doc("inf-1") == {"inference_id": "inf-1"}
assert es.requests[-1][2]["sort"] == NEWEST_FIRST
- [ ] Step 2: Run it to confirm it fails
Run: venv/bin/python -m pytest services/api/tests/test_whitelist_store.py -q
Expected: failures. The GET goes to /logster-graphs/_doc/..., and update_graph_doc doesn't accept index=.
- [ ] Step 3: Implement in
whitelist_store.py
Constructor defaults:
Replaceget_graph_doc, get_inference_doc and update_graph_doc:
def get_graph_doc(self, inference_id: str) -> Optional[Dict[str, Any]]:
"""Fetch a persisted graph document.
Graphs live in ILM rollover indices read through ``logster-graphs*``
(plus the pre-ILM ``logster-graphs`` index on upgraded installs), so a
single-document GET by id is impossible. Search by ``_id`` (the graph
sink indexes with ``_id = inference_id``) or by the ``inference_id``
field, newest first in case a rollover left a duplicate. The result
carries ``_id`` and the concrete ``_index`` so partial updates hit the
right document.
"""
resp = self._request("POST", f"{self._graphs_index}/_search", {
"size": 1,
"sort": [{"@timestamp": {"order": "desc", "unmapped_type": "date"}}],
"query": {"bool": {"minimum_should_match": 1, "should": [
{"ids": {"values": [inference_id]}},
{"term": {"inference_id": inference_id}},
]}},
})
hits = (resp or {}).get("hits", {}).get("hits", [])
if not hits:
return None
hit = hits[0]
return {**(hit.get("_source") or {}),
"_id": hit.get("_id", inference_id), "_index": hit.get("_index")}
def get_inference_doc(self, inference_id: str) -> Optional[Dict[str, Any]]:
"""Fetch a persisted inference result from ``logster-inferences*``.
``inference_id`` is dynamically mapped as ``text`` + ``.keyword`` (the
index template maps only ``@timestamp``), so query the keyword
sub-field. Newest first: a re-published result (TTP enrichment, Kafka
redelivery) written after an ILM rollover exists twice. ES failures
propagate as ``WhitelistStoreError`` so the caller can report a
retryable error instead of a silent "not found".
"""
body = {
"size": 1,
"sort": [{"@timestamp": {"order": "desc", "unmapped_type": "date"}}],
"query": {"term": {"inference_id.keyword": inference_id}},
}
resp = self._request("POST", f"{self._inferences_index}/_search", body)
hits = (resp or {}).get("hits", {}).get("hits", [])
return hits[0]["_source"] if hits else None
def update_graph_doc(self, doc_id: str, partial: Dict[str, Any], *, index: str) -> None:
"""Partial-update a persisted graph document (e.g. to cache evidence).
``index`` must be the concrete index the document was read from
(``get_graph_doc()["_index"]``): ``_update`` cannot target a pattern
or a multi-index alias.
"""
if not index or any(c in index for c in "*?,"):
raise WhitelistStoreError(f"update_graph_doc needs a concrete index, got {index!r}")
self._request(
"POST",
f"{urlparse.quote(index, safe='')}/_update/"
f"{urlparse.quote(doc_id, safe='')}?refresh=true",
{"doc": partial},
)
In evidence.py, replace the doc_id line (365-366):
# The store attaches the real ES ``_id`` and the concrete ``_index`` it
# lives in (graphs are spread over ILM rollover indices).
doc_id = graph_doc.get("_id") or inference_id
doc_index = graph_doc.get("_index")
if result["status"] == "success":
cache_doc = {
**result,
"extracted_at": datetime.now(timezone.utc).isoformat(),
"model": lc_config.model,
}
if not doc_index:
logger.warning("Not caching evidence for %s: graph doc has no _index", inference_id)
else:
try:
store.update_graph_doc(doc_id, {"evidence": cache_doc}, index=doc_index)
except Exception as e:
logger.warning("Failed to cache evidence for %s: %s", inference_id, e)
In services/api/tests/test_evidence_endpoint.py:
- Add "_index": "logster-graphs-000001", to the get_graph_doc.return_value dict in each of test_evidence_full_extraction, test_evidence_empty_result_is_not_cached_and_can_retry, test_evidence_replaces_empty_legacy_cache and test_evidence_cache_targets_store_doc_id.
- In test_evidence_cache_targets_store_doc_id, add after the existing call_args[0][0] assertion:
def test_evidence_without_index_is_returned_but_not_cached(client, mock_store):
mock_store.get_graph_doc.return_value = {
"_id": "doc-1",
"inference_id": "inf-1",
"nodes": [{"id": "{AAA}", "type": "process", "originating_event_id": "evt-1"}],
"edges": [],
}
mock_store.get_inference_doc.return_value = {"lc_verdict": "attack"}
response = json.dumps({"evidence": [{"step": 1, "node_ids": ["{AAA}"]}]})
with patch("logster_api.evidence.call_llm", return_value=response):
data = client.post("/api/evidence/inf-1").json()
assert data["all_evidence_node_ids"] == ["{AAA}"]
mock_store.update_graph_doc.assert_not_called()
- [ ] Step 4: Run the API suite
Run: venv/bin/python -m pytest services/api/tests -q
Expected: all pass.
- [ ] Step 5: Stage
git add services/api/src/logster_api/whitelist_store.py services/api/src/logster_api/evidence.py services/api/tests/test_whitelist_store.py services/api/tests/test_evidence_endpoint.py
fix(api): read graphs/inferences through ILM patterns, update by concrete index
Task 13: Dashboard reads: patterns, graph lookup by search, newest-first, sort allowlist
Files:
- Modify: services/dashboard/server/src/config.js:6-8, src/routes/graph.js:80-113, src/routes/inferenceDetail.js:201-218, src/routes/inferences.js:15-26, src/routes/recentAttacks.js:26, src/routes/search.js:99,123
- Test: services/dashboard/server/test/graphDoc.test.js, test/inferencesRoute.test.js (new), test/inferenceDetail.test.js (append)
Interfaces:
- Produces: findInference(esClient, inferenceId) -> source|null, exported from routes/inferenceDetail.js
- [ ] Step 1: Write the failing tests
services/dashboard/server/test/graphDoc.test.js:
import { describe, it, before, after, mock } from 'node:test';
import assert from 'node:assert/strict';
import http from 'node:http';
import express from 'express';
process.env.ENABLE_GRAPH_VIEW = 'true';
const { es } = await import('../src/elastic.js');
const { default: router } = await import('../src/routes/graph.js');
let server;
let base;
describe('graph document route', () => {
before(async () => {
const app = express();
app.use(router);
server = http.createServer(app);
await new Promise((r) => server.listen(0, '127.0.0.1', r));
base = `http://127.0.0.1:${server.address().port}`;
});
after(async () => {
mock.restoreAll();
await new Promise((r) => server.close(r));
});
it('searches the graphs pattern by id, newest first', async () => {
const search = mock.method(es, 'search', async () => ({ hits: { hits: [
{ _index: 'logster-graphs-000002', _id: 'inf-1', _source: { inference_id: 'inf-1', nodes: [] } },
] } }));
const res = await fetch(`${base}/graph/inf-1`);
assert.equal(res.status, 200);
assert.deepEqual(await res.json(), { inference_id: 'inf-1', nodes: [] });
const { index, body } = search.mock.calls[0].arguments[0];
assert.equal(index, 'logster-graphs*');
assert.deepEqual(body.sort, [{ '@timestamp': { order: 'desc', unmapped_type: 'date' } }]);
assert.deepEqual(body.query.bool.should, [
{ ids: { values: ['inf-1'] } }, { term: { inference_id: 'inf-1' } },
]);
search.mock.restore();
});
it('404s when no graph was persisted', async () => {
const search = mock.method(es, 'search', async () => ({ hits: { hits: [] } }));
assert.equal((await fetch(`${base}/graph/inf-2`)).status, 404);
search.mock.restore();
});
});
services/dashboard/server/test/inferencesRoute.test.js:
import { describe, it, before, after, mock } from 'node:test';
import assert from 'node:assert/strict';
import http from 'node:http';
import express from 'express';
const { es } = await import('../src/elastic.js');
const { default: router } = await import('../src/routes/inferences.js');
let server;
let base;
let search;
describe('inferences list route', () => {
before(async () => {
search = mock.method(es, 'search', async () => ({ hits: { total: { value: 0 }, hits: [] } }));
const app = express();
app.use(router);
server = http.createServer(app);
await new Promise((r) => server.listen(0, '127.0.0.1', r));
base = `http://127.0.0.1:${server.address().port}`;
});
after(async () => {
mock.restoreAll();
await new Promise((r) => server.close(r));
});
it('reads the pattern and sorts by @timestamp with an unmapped_type', async () => {
assert.equal((await fetch(`${base}/inferences`)).status, 200);
const { index, body } = search.mock.calls.at(-1).arguments[0];
assert.equal(index, 'logster-inferences*');
assert.deepEqual(body.sort, [{ '@timestamp': { order: 'desc', unmapped_type: 'date' } }]);
});
it('accepts allowlisted numeric sorts', async () => {
await fetch(`${base}/inferences?sort=attack_prob&order=asc`);
assert.deepEqual(search.mock.calls.at(-1).arguments[0].body.sort,
[{ attack_prob: { order: 'asc', unmapped_type: 'float' } }]);
});
it('rejects unknown sort fields before querying ES', async () => {
const count = search.mock.callCount();
assert.equal((await fetch(`${base}/inferences?sort=endpoint_id`)).status, 400);
assert.equal(search.mock.callCount(), count);
});
});
Append to services/dashboard/server/test/inferenceDetail.test.js. Also add findInference to its import line: import { fetchInferenceEvents, findInference, mapEventHit, parseLinuxAudit } from '../src/routes/inferenceDetail.js';
test('findInference asks for the newest copy across the pattern', async () => {
let seen;
const esClient = { search: async (req) => {
seen = req;
return { hits: { hits: [{ _source: { inference_id: 'i' } }] } };
} };
assert.deepEqual(await findInference(esClient, 'i'), { inference_id: 'i' });
assert.equal(seen.index, 'logster-inferences*');
assert.deepEqual(seen.body.sort, [{ '@timestamp': { order: 'desc', unmapped_type: 'date' } }]);
assert.deepEqual(seen.body.query, { term: { 'inference_id.keyword': 'i' } });
});
test('findInference returns null when absent', async () => {
const esClient = { search: async () => ({ hits: { hits: [] } }) };
assert.equal(await findInference(esClient, 'nope'), null);
});
- [ ] Step 2: Run them to confirm they fail
Run: (cd services/dashboard/server && node --test test/graphDoc.test.js test/inferencesRoute.test.js test/inferenceDetail.test.js)
Expected: failures. es.get is called instead of es.search, the index is logster-inferences, and findInference is not exported.
- [ ] Step 3: Implement
src/config.js lines 6-8:
// Read patterns: the pre-ILM index (upgraded installs) plus its daily ILM
// rollover indices. Writers use the logster-*-write aliases instead.
export const INFERENCES_INDEX = 'logster-inferences*';
export const EVENTS_INDEX = 'logster-events*';
export const GRAPHS_INDEX = process.env.GRAPHS_INDEX || 'logster-graphs*';
src/routes/graph.js: replace the body of the try { in router.get('/graph/:inference_id'…) up to the if (!source) check with:
// Graphs are spread over ILM rollover indices, so a GET by id cannot
// work; search by _id (GraphSink indexes with _id = inference_id) or the
// inference_id keyword field, newest first.
const search = await es.search({
index: GRAPHS_INDEX,
body: {
size: 1,
sort: [{ '@timestamp': { order: 'desc', unmapped_type: 'date' } }],
query: { bool: { minimum_should_match: 1, should: [
{ ids: { values: [inference_id] } },
{ term: { inference_id } },
] } },
},
});
const source = search.hits.hits[0]?._source ?? null;
if (!source) return 404 and the catch block as they are. Also update the JSDoc line "sourced from the logster-graphs ES index" to "sourced from the logster-graphs* indices".
src/routes/inferenceDetail.js: add above the router:
// Newest first: a re-published inference (TTP enrichment, Kafka redelivery)
// written after an ILM rollover exists in two indices.
export async function findInference(esClient, inferenceId) {
const resp = await esClient.search({
index: INFERENCES_INDEX,
body: {
size: 1,
sort: [{ '@timestamp': { order: 'desc', unmapped_type: 'date' } }],
query: { term: { 'inference_id.keyword': inferenceId } },
},
});
return resp.hits.hits[0]?._source ?? null;
}
infResp block and the total.value === 0 check with:
const inference = await findInference(es, inference_id);
if (!inference) {
return res.status(404).json({ error: 'Not found' });
}
const inference = infResp.hits.hits[0]._source; line.
src/routes/recentAttacks.js:26 and src/routes/search.js:99,123: in each sort entry on @timestamp, add unmapped_type: 'date' (e.g. { '@timestamp': { order: 'desc', unmapped_type: 'date' } }). The templates already map @timestamp, so this is insurance: without it, a sort across the pattern returns HTTP 400 if every matching index lacks the mapping (verified on 9.5.4).
src/routes/inferences.js: add above the router:
// Sortable fields and their types. unmapped_type keeps a sort across
// logster-inferences* from failing a shard on a freshly rolled-over,
// still-empty index.
const SORT_FIELDS = {
'@timestamp': 'date',
attack_prob: 'float',
confidence: 'float',
lc_attack_prob: 'float',
lc_confidence: 'float',
};
sortField/sortOrder lines with:
const sortField = req.query.sort || '@timestamp';
if (!Object.hasOwn(SORT_FIELDS, sortField)) {
return res.status(400).json({ error: `unsupported sort field: ${sortField}` });
}
const sortOrder = req.query.order === 'asc' ? 'asc' : 'desc';
- [ ] Step 4: Run the dashboard server suite
Run: (cd services/dashboard/server && npm test)
Expected: all pass: the original 43 plus the new tests.
- [ ] Step 5: Stage
git add services/dashboard/server/src/config.js services/dashboard/server/src/routes/graph.js services/dashboard/server/src/routes/inferenceDetail.js services/dashboard/server/src/routes/inferences.js services/dashboard/server/src/routes/recentAttacks.js services/dashboard/server/src/routes/search.js services/dashboard/server/test/graphDoc.test.js services/dashboard/server/test/inferencesRoute.test.js services/dashboard/server/test/inferenceDetail.test.js
fix(dashboard): read ILM index patterns; graph lookup by search; sort allowlist
Task 14: Dashboard storage: family grouping, forecast without strip or emergency
Files:
- Create: services/dashboard/server/src/lib/indexFamilies.js, services/dashboard/server/test/indexFamilies.test.js
- Modify: services/dashboard/server/src/routes/storage.js:18-24,94-103, src/lib/capacity.js, test/capacity.test.js, test/storage.test.js, services/dashboard/client/src/components/health/StorageSection.jsx:221-224, services/dashboard/client/src/lib/panelInfo.js:72-74
Interfaces:
- Consumes (Task 6): a forecast doc with target_percent and no emergency_percent / strip_ratio / strip_raw_event_days
- Produces:
- familyOf(indexName) -> string|null, groupIndexFamilies(catRows) -> [{name, health, docs_count, store_size_bytes, index_count}]
- /api/storage capacity.days_to_target replaces days_to_emergency
- [ ] Step 1: Write the failing tests
services/dashboard/server/test/indexFamilies.test.js:
import { describe, it } from 'node:test';
import assert from 'node:assert/strict';
import { familyOf, groupIndexFamilies } from '../src/lib/indexFamilies.js';
describe('familyOf', () => {
it('maps legacy and rollover indices to their family', () => {
assert.equal(familyOf('logster-events'), 'logster-events');
assert.equal(familyOf('logster-events-000012'), 'logster-events');
assert.equal(familyOf('logster-graphs-000001'), 'logster-graphs');
assert.equal(familyOf('logster-alerts'), 'logster-alerts');
});
it('ignores everything else', () => {
for (const name of ['logster-capacity', 'logster-graph-whitelists', 'logster-events-backup',
'logster-events-00001', '.kibana_1', undefined]) {
assert.equal(familyOf(name), null);
}
});
});
describe('groupIndexFamilies', () => {
it('sums docs and store per family, keeps the worst health, sorts by size', () => {
const rows = [
{ index: 'logster-graphs-000001', health: 'green', 'docs.count': '1', 'store.size': '7' },
{ index: 'logster-events', health: 'green', 'docs.count': '10', 'store.size': '100' },
{ index: 'logster-events-000001', health: 'yellow', 'docs.count': '5', 'store.size': '50' },
{ index: 'logster-events-000002', health: null, 'docs.count': null, 'store.size': null },
{ index: 'logster-capacity', health: 'red', 'docs.count': '1', 'store.size': '9' },
];
assert.deepEqual(groupIndexFamilies(rows), [
{ name: 'logster-events', health: 'yellow', docs_count: 15, store_size_bytes: 150, index_count: 3 },
{ name: 'logster-graphs', health: 'green', docs_count: 1, store_size_bytes: 7, index_count: 1 },
]);
});
});
In test/capacity.test.js:
- In the forecast() fixture, delete emergency_percent: 85,, strip_ratio: 0.5,, and both strip_raw_event_days: …, keys.
- Replace the four affected tests in describe('assessCapacity', …). Delete each of these existing it(…) blocks entirely and paste its replacement (below, same order) in its place — do not append:
1. it('flags under-provisioning with effective retention and a disk size', …) → replacement keeps the title.
2. it('reports shortened retention when events do not fit', …) → replacement keeps the title.
3. it('goes critical when the emergency threshold is days away', …) → replaced by it('goes critical when the target is days away', …).
4. it('warns about disabled retention once the disk would fill', …) → replaced by it('warns about disabled retention once the disk would reach the target', …).
Afterwards, grep -n "days_to_emergency\|emergency threshold\|disk would fill\|emergency_percent\|strip_" services/dashboard/server/test/capacity.test.js prints nothing, and the file still has 11 it( blocks under describe('assessCapacity'.
it('flags under-provisioning with effective retention and a disk size', () => {
const r = assessCapacity({
forecast: forecast({ es_need_bytes: 80 * GIB }), now: NOW,
});
assert.equal(r.level, 'warning');
assert.equal(r.need_bytes, 90 * GIB);
// Room for events: 75 − 10 non-ES − 9 inferences = 56 GiB at 2 GiB/day.
assert.equal(r.effective_events_days, 28);
assert.match(r.reasons[0], /^Under-provisioned: at ≈2\.0 M events\/day/);
assert.match(r.reasons[0], /Expect ~28 of 30 days of event retention/);
assert.match(r.reasons[0], /Recommended disk: 120 GiB/);
});
it('reports shortened retention when events do not fit', () => {
const r = assessCapacity({
forecast: forecast({ es_need_bytes: 80 * GIB, disk_total_bytes: 40 * GIB,
disk_used_bytes: 20 * GIB, es_store_bytes: 15 * GIB }),
now: NOW,
});
// Budget 30, non-ES 5, inferences 9 → 16 GiB at 2 GiB/day.
assert.equal(r.effective_events_days, 8);
assert.match(r.reasons[0], /Expect ~8 of 30 days of event retention/);
});
it('goes critical when the target is days away', () => {
const r = assessCapacity({
forecast: forecast({ es_need_bytes: 80 * GIB, disk_used_bytes: 73 * GIB,
es_store_bytes: 63 * GIB }),
now: NOW,
});
// 63 + 10 = 73 now; +1 GiB on day 1 is 74; +2 on day 2 reaches 75.
assert.equal(r.days_to_target, 2);
assert.equal(r.level, 'critical');
assert.match(r.reasons[0], /deleted before its retention age in ~2 days/);
});
it('warns about disabled retention once the disk would reach the target', () => {
const r = assessCapacity({
forecast: forecast({ es_need_bytes: null, unbounded_indices: ['logster-events'],
growth_curve: Array.from({ length: 366 }, (_, d) => d * GIB) }),
now: NOW,
});
// 40 GiB used, target at 75 → day 35.
assert.equal(r.days_to_target, 35);
assert.equal(r.ratio, null);
assert.equal(r.level, 'warning');
assert.match(r.reasons[0], /Retention is disabled for logster-events.*reaches 75%/);
});
In test/storage.test.js:
- Make the _cat/indices reply return family rows:
return esReply(res, 200, [
{ index: 'logster-events', health: 'green', 'docs.count': '10', 'store.size': String(20 * GIB) },
{ index: 'logster-events-000001', health: 'green', 'docs.count': '5', 'store.size': String(10 * GIB) },
{ index: 'logster-capacity', health: 'green', 'docs.count': '1', 'store.size': '100' },
]);
capacityDoc, delete emergency_percent, strip_ratio and strip_raw_event_days.
- After the existing retention_days assertion, add:
assert.equal(body.elasticsearch.indices.length, 1);
assert.equal(body.elasticsearch.indices[0].name, 'logster-events');
assert.equal(body.elasticsearch.indices[0].docs_count, 15);
assert.equal(body.elasticsearch.indices[0].store_size_bytes, 30 * GIB);
assert.equal(body.elasticsearch.indices[0].index_count, 2);
- [ ] Step 2: Run them to confirm they fail
Run: (cd services/dashboard/server && node --test test/indexFamilies.test.js test/capacity.test.js test/storage.test.js)
Expected: the module is missing, effective_events_days is still 30, and days_to_target is undefined.
- [ ] Step 3: Implement
services/dashboard/server/src/lib/indexFamilies.js:
// Logster data indices come in families: the pre-ILM index on upgraded
// installs (`logster-events`) plus its ILM rollover indices
// (`logster-events-000001`, …). The dashboard shows one row per family.
const FAMILY_RE = /^(logster-(?:events|inferences|graphs|alerts))(?:-\d{6,})?$/;
const HEALTH_RANK = { green: 0, yellow: 1, red: 2 };
export function familyOf(indexName) {
const m = FAMILY_RE.exec(indexName || '');
return m ? m[1] : null;
}
export function groupIndexFamilies(catRows) {
const families = new Map();
for (const row of catRows || []) {
const name = familyOf(row.index);
if (!name) continue;
const f = families.get(name)
|| { name, health: 'green', docs_count: 0, store_size_bytes: 0, index_count: 0 };
f.docs_count += parseInt(row['docs.count'] || '0', 10);
f.store_size_bytes += parseInt(row['store.size'] || '0', 10);
f.index_count += 1;
const rank = HEALTH_RANK[row.health];
if (rank !== undefined && rank > HEALTH_RANK[f.health]) f.health = row.health;
families.set(name, f);
}
return [...families.values()].sort((a, b) => b.store_size_bytes - a.store_size_bytes);
}
src/routes/storage.js:
- Delete LOGSTER_INDEX_PREFIX and LOGSTER_DATA_INDICES (lines 18-24).
- Add import { groupIndexFamilies } from '../lib/indexFamilies.js';
- Replace the const indices = catIndices…sort(…) statement with:
const indices = groupIndexFamilies(catIndices).map((f) => ({
...f,
retention_days: INDEX_RETENTION_DAYS[f.name] ?? null,
}));
src/lib/capacity.js:
1. Replace the CRITICAL_DAYS comment:
// Escalate to critical when the disk reaches the retention target within
// this many days: past that, the disk rule deletes the oldest data before
// its retention age.
STALE_AFTER_MS comment, change (30 by default) to (5 by default).
3. Replace eventDaysInRoom and its comment:
// Days of events that fit in `room` bytes, capped at the configured retention.
function eventDaysInRoom(room, perDay, keepDays) {
if (room <= 0) return 0;
return Math.min(room / perDay, keepDays);
}
assessCapacity:
- Delete const emergencyBytes = …;.
- Rename daysToEmergency to daysToTarget everywhere in the function, and compare against budget in the loop. The loop comment becomes: // First day on which projected usage reaches the target, where the disk rule starts deleting the oldest data before its retention age.
- Change the eventDaysInRoom(...) call to eventDaysInRoom(budget - nonEsSteady - otherIndices, events.daily_bytes * forecast.merge_overhead, events.retention_days).
- In the result object, replace days_to_emergency: daysToEmergency, with days_to_target: daysToTarget,.
- In the unbounded reason, change ${forecast.emergency_percent}% to ${forecast.target_percent}%.
- Replace the critical parts.push(...) with:
parts.push(daysToTarget === 0
? 'The oldest data is already being deleted before its retention age.'
: `The oldest data will be deleted before its retention age in ~${daysToTarget} days.`);
services/dashboard/client/src/components/health/StorageSection.jsx (tile at lines 221-224):
<StatTile
title="Days until early deletion"
value={capacity.days_to_target == null ? 'Never' : formatDays(capacity.days_to_target)}
/>
services/dashboard/client/src/lib/panelInfo.js: replace the two strings.
- 'storage.es': 'Elasticsearch disk usage and per-family breakdown (a family is the pre-upgrade index plus its daily rollover indices). Only Logster data indices are shown. Retention days come from the elasticsearch-retention worker once it has published a capacity forecast; until then (and for logster-alerts) they are the shipped defaults.'
- 'storage.capacity': 'Whether the disk can hold the configured retention at the current ingest rate. The retention worker measures docs/day (last 24 h) and bytes/doc per index family; Kafka\'s configured maximum is reserved when it shares the disk. The budget is the retention target (75% of disk by default): above it, the oldest data is deleted before its retention age.'
- [ ] Step 4: Run the tests and build the client
Run the three commands independently. The client suite has one known failure, so it exits 1 by design: never chain the build behind it with &&.
(cd services/dashboard/server && npm test 2>&1 | grep -E '^ℹ (tests|pass|fail)')
(cd services/dashboard/client && npm test 2>&1 | grep -E '^✖|^ℹ (tests|pass|fail)')
(cd services/dashboard/client && npm run build 2>&1 | tail -3)
ℹ fail 0.
- Client: ℹ tests 12, ℹ pass 11, ℹ fail 1, and the only failing test named on the ✖ lines is eBPF process event uses eBPF label (in test/eventSummary.test.js). That is the recorded baseline (11 pass / 1 fail, also failing on dev); don't fix it here. Any other ✖ test is a new failure: fix it in this task.
- Build: the Vite output ends with ✓ built in ….
- [ ] Step 5: Stage
git add services/dashboard/server/src/lib/indexFamilies.js services/dashboard/server/test/indexFamilies.test.js services/dashboard/server/src/routes/storage.js services/dashboard/server/src/lib/capacity.js services/dashboard/server/test/capacity.test.js services/dashboard/server/test/storage.test.js services/dashboard/client/src/components/health/StorageSection.jsx services/dashboard/client/src/lib/panelInfo.js
feat(dashboard): per-family storage rows; forecast measured against the target
Task 15: Docker Compose (dev + OVA appliance)
Files:
- Modify: deploy/docker-compose.yml (elasticsearch command, elasticsearch-init env, elasticsearch-retention comment + env, inference.depends_on)
Interfaces:
- Produces:
- ES arg list identical to the chart's (Task 16 enforces it).
- RETENTION_* knobs as in Global Constraints.
- [ ] Step 1: Edit
deploy/docker-compose.yml
In elasticsearch.command, after - -Expack.security.transport.ssl.enabled=false, add:
# ML is unused by Logster (matches the chart's statefulset.yaml).
- -Expack.ml.enabled=false
# Refuse to auto-create the ILM write-alias names or the bare
# pre-ILM names: a write that races the layout bootstrap fails
# instead of creating an unmanaged index that squats on the alias
# (spec D9). Keep identical to the chart's statefulset.yaml.
- "-Eaction.auto_create_index=-logster-events,-logster-inferences,-logster-graphs,-logster-*-write,+*"
In elasticsearch-init.environment, below ES_PASSWORD:
# init writes the ILM policies (layout.py), so it gets the same
# retention days as the worker.
RETENTION_EVENTS_DAYS: ${RETENTION_EVENTS_DAYS:-30}
RETENTION_INFERENCES_DAYS: ${RETENTION_INFERENCES_DAYS:-90}
RETENTION_GRAPHS_DAYS: ${RETENTION_GRAPHS_DAYS:-14}
Replace the comment block above elasticsearch-retention: with:
# Retention worker. Elasticsearch ILM deletes whole daily indices older
# than RETENTION_*_DAYS (0 = keep forever); this worker keeps the layout
# healthy and, while disk is above RETENTION_TARGET_PERCENT, deletes the
# oldest whole Logster index — never the one being written, never data
# newer than RETENTION_MIN_AGE_HOURS. See elasticsearch/retention.py.
environment: block with:
environment:
ES_PASSWORD: ${ELASTIC_PASSWORD:-logster-dev-es-password}
RETENTION_EVENTS_DAYS: ${RETENTION_EVENTS_DAYS:-30}
RETENTION_INFERENCES_DAYS: ${RETENTION_INFERENCES_DAYS:-90}
RETENTION_GRAPHS_DAYS: ${RETENTION_GRAPHS_DAYS:-14}
RETENTION_TARGET_PERCENT: ${RETENTION_TARGET_PERCENT:-75}
RETENTION_MIN_AGE_HOURS: ${RETENTION_MIN_AGE_HOURS:-24}
RETENTION_CHECK_INTERVAL_MINUTES: ${RETENTION_CHECK_INTERVAL_MINUTES:-5}
In inference.depends_on, delete these two lines (spec D11; OpenShift never gated inference on init):
vector's elasticsearch-init gate as it is.
- [ ] Step 2: Validate
Run: docker compose -f deploy/docker-compose.yml config -q && docker compose -f deploy/docker-compose.yml config | grep -n 'auto_create_index\|RETENTION_\|elasticsearch-init' | head -30
Expected: no error. The guard arg appears once. RETENTION_STRIP_RAW_EVENT_DAYS, RETENTION_INTERVAL_HOURS, RETENTION_EMERGENCY_THRESHOLD_PERCENT and RETENTION_CRITICAL_THRESHOLD_PERCENT are absent. elasticsearch-init appears under vector.depends_on but not under inference.
Also run: grep -c 'ELASTIC_PASSWORD' deploy/docker-compose.yml
Expected: 8 — the same as before this task (run it on dev first to confirm). The packer 04-configure-logster.sh substitution counts these lines; this task must not change the count.
- [ ] Step 3: Stage
chore(compose): auto-create guard, ILM retention knobs, ungate inference
Task 16: OpenShift Helm chart + drift guard + e2e patterns
Files:
- Copy: deploy/elasticsearch/{capacity,diskcap,eshttp,init,layout,retention,retention_config}.py and deploy/elasticsearch/*.template.json → deploy/openshift/logster/files/elasticsearch/; deploy/vector/vector.yaml → deploy/openshift/logster/files/vector/vector.yaml (password default stripped)
- Modify: deploy/openshift/logster/templates/elasticsearch/{scripts-configmap,init-job,retention-deployment,statefulset}.yaml, templates/vector/deployment.yaml, values.yaml (retention block), templates/NOTES.txt:22-24, tests/e2e-openshift/lib.sh:30-31
- Test: deploy/elasticsearch/test_chart_drift.py (new)
Interfaces:
- Produces:
- SHIPPED_SCRIPTS (capacity.py, diskcap.py, eshttp.py, init.py, layout.py, retention.py, retention_config.py).
- Values retention.{eventsDays,inferencesDays,graphsDays,targetPercent,minAgeHours,checkIntervalMinutes,resources}.
- [ ] Step 1: Write the failing drift test
deploy/elasticsearch/test_chart_drift.py
"""The Helm chart ships copies of deploy/ files; they must not drift (spec §12.1)."""
import re
import unittest
from pathlib import Path
HERE = Path(__file__).resolve().parent # deploy/elasticsearch
DEPLOY = HERE.parent
CHART = DEPLOY / "openshift" / "logster"
CHART_ES = CHART / "files" / "elasticsearch"
SHIPPED_SCRIPTS = ("capacity.py", "diskcap.py", "eshttp.py", "init.py", "layout.py",
"retention.py", "retention_config.py")
ES_ARG = re.compile(r'^\s*-\s+"?(-E[^"\s]+)"?\s*$', re.M)
GUARD = ("-Eaction.auto_create_index=-logster-events,-logster-inferences,"
"-logster-graphs,-logster-*-write,+*")
class TestChartDrift(unittest.TestCase):
def test_shipped_scripts_are_identical(self):
for name in SHIPPED_SCRIPTS:
with self.subTest(name=name):
self.assertEqual((CHART_ES / name).read_bytes(), (HERE / name).read_bytes(),
f"run: cp deploy/elasticsearch/{name} "
"deploy/openshift/logster/files/elasticsearch/")
def test_chart_ships_no_other_python(self):
self.assertEqual(sorted(p.name for p in CHART_ES.glob("*.py")), sorted(SHIPPED_SCRIPTS))
def test_templates_are_identical(self):
canonical = {p.name: p.read_bytes() for p in HERE.glob("*.template.json")}
chart = {p.name: p.read_bytes() for p in CHART_ES.glob("*.template.json")}
self.assertEqual(chart, canonical)
def test_configmap_ships_every_script(self):
text = (CHART / "templates" / "elasticsearch" / "scripts-configmap.yaml").read_text()
for name in SHIPPED_SCRIPTS:
self.assertIn(f'"{name}"', text)
def test_vector_config_matches_except_password_default(self):
canonical = (DEPLOY / "vector" / "vector.yaml").read_text().replace(
"${ES_PASSWORD:-logster-dev-es-password}", "${ES_PASSWORD}")
self.assertEqual((CHART / "files" / "vector" / "vector.yaml").read_text(), canonical)
def test_es_args_match_between_compose_and_chart(self):
compose = ES_ARG.findall((DEPLOY / "docker-compose.yml").read_text())
chart = ES_ARG.findall(
(CHART / "templates" / "elasticsearch" / "statefulset.yaml").read_text())
self.assertTrue(compose)
self.assertEqual(compose, chart)
self.assertIn(GUARD, chart)
def test_integration_suite_starts_es_with_the_shipped_guard(self):
# integration/run.sh copies the guard; scenarios 2 and 13 must test
# the one Compose ships.
run_sh = (HERE / "integration" / "run.sh").read_text()
m = re.search(r"^GUARD='([^']*)'$", run_sh, re.M)
self.assertIsNotNone(m, "GUARD='...' line missing from integration/run.sh")
compose = ES_ARG.findall((DEPLOY / "docker-compose.yml").read_text())
self.assertEqual([a for a in compose if a.startswith("-Eaction.auto_create_index=")],
[m.group(1)])
if __name__ == "__main__":
unittest.main()
- [ ] Step 2: Run it to confirm it fails
Run: venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_chart_drift.py' -v
Expected: failures. New scripts are missing from the chart, the old retention.py differs, and the ES args differ.
- [ ] Step 3: Sync the chart copies
cp deploy/elasticsearch/{capacity,diskcap,eshttp,init,layout,retention,retention_config}.py deploy/openshift/logster/files/elasticsearch/
cp deploy/elasticsearch/*.template.json deploy/openshift/logster/files/elasticsearch/
sed 's/\${ES_PASSWORD:-logster-dev-es-password}/${ES_PASSWORD}/' deploy/vector/vector.yaml > deploy/openshift/logster/files/vector/vector.yaml
- [ ] Step 4: Edit the chart templates
templates/elasticsearch/scripts-configmap.yaml: replace the three explicit init.py/retention.py/capacity.py entries with:
data:
{{- range $name := list "capacity.py" "diskcap.py" "eshttp.py" "init.py" "layout.py" "retention.py" "retention_config.py" }}
{{ $name }}: |
{{ $.Files.Get (printf "files/elasticsearch/%s" $name) | indent 4 }}
{{- end }}
.Files.Glob "files/elasticsearch/*.template.json" range below it.
templates/elasticsearch/init-job.yaml:
- Change the header comment to # One-shot ES bootstrap (watermarks, ILM layout via layout.converge(), replicas=0, reroute) — the compose elasticsearch-init service. Idempotent; re-run on every install/upgrade. Nothing waits on this hook: the retention worker converges the same layout, so helm --wait cannot deadlock.
- Under env: after the esPasswordEnv include:
# init writes the ILM policies (layout.py), so it needs the days.
- name: RETENTION_EVENTS_DAYS
value: {{ .Values.retention.eventsDays | quote }}
- name: RETENTION_INFERENCES_DAYS
value: {{ .Values.retention.inferencesDays | quote }}
- name: RETENTION_GRAPHS_DAYS
value: {{ .Values.retention.graphsDays | quote }}
templates/elasticsearch/retention-deployment.yaml:
- Header comment: # Retention worker (the compose elasticsearch-retention service): converges the ILM layout, reports ILM errors, and enforces the disk target by deleting the oldest whole index. Loops internally every retention.checkIntervalMinutes, so it runs as a Deployment rather than a CronJob.
- Under spec: after replicas: 1:
# One worker at a time (spec D13): the old pod is taken down before the new
# one is created, and retention.py exits on SIGTERM, so two disk rules never
# overlap. Not `type: Recreate`: switching an existing Deployment to
# Recreate fails `helm upgrade` under Helm 4's server-side apply
# ("spec.strategy.rollingUpdate: Forbidden"), with or without
# `rollingUpdate: null` — verified on CRC (OCP 4.22) with Helm 4.2.4, while
# this RollingUpdate form upgrades cleanly with Helm 3.8.2, 3.22.0 and 4.2.4.
strategy:
type: RollingUpdate
rollingUpdate:
maxSurge: 0
maxUnavailable: 1
RETENTION_* env entries with:
- name: RETENTION_EVENTS_DAYS
value: {{ .Values.retention.eventsDays | quote }}
- name: RETENTION_INFERENCES_DAYS
value: {{ .Values.retention.inferencesDays | quote }}
- name: RETENTION_GRAPHS_DAYS
value: {{ .Values.retention.graphsDays | quote }}
# `default` on the three knobs the customer's v1.2.1 chart never
# had (targetPercent, minAgeHours, checkIntervalMinutes): an
# upgrade with --reuse-values from v1.2.1 carries no such keys and
# would otherwise render `value:` null (verified with Helm 3.22
# against chart 1.2.1+splunk-extip.1). Do NOT add `default` to the
# *Days keys: sprig treats 0 (keep forever) as empty and would
# silently replace it.
- name: RETENTION_TARGET_PERCENT
value: {{ .Values.retention.targetPercent | default 75 | quote }}
- name: RETENTION_MIN_AGE_HOURS
value: {{ .Values.retention.minAgeHours | default 24 | quote }}
- name: RETENTION_CHECK_INTERVAL_MINUTES
value: {{ .Values.retention.checkIntervalMinutes | default 5 | quote }}
templates/elasticsearch/statefulset.yaml: after - -Expack.ml.enabled=false add:
# Refuse to auto-create the ILM write-alias names or the bare
# pre-ILM names (spec D9). Keep identical to deploy/docker-compose.yml.
- "-Eaction.auto_create_index=-logster-events,-logster-inferences,-logster-graphs,-logster-*-write,+*"
templates/vector/deployment.yaml:
- Under template.metadata, add annotations, so config changes roll Vector:
annotations:
checksum/vector-config: {{ include (print $.Template.BasePath "/vector/configmap.yaml") . | sha256sum }}
spec: (pod), before containers::
# Vector must not write before the ILM write aliases exist: a write to a
# missing alias is dropped (spec D9/D11). The retention worker and the
# init hook create them; this waits (no timeout) until they do, or until
# a concrete index squats on an alias name (writes land there; it logs
# a SQUATTER error and lets Vector start). It never
# waits on the post-install hook, so `helm --wait` cannot deadlock.
initContainers:
- name: wait-es-layout
image: {{ include "logster.image" (dict "root" . "name" "elasticsearch-init") }}
{{- include "logster.containerSecurityContext" . | nindent 10 }}
command: ["python3", "/templates/layout.py", "wait"]
env:
{{- include "logster.esPasswordEnv" . | nindent 12 }}
resources:
requests: {cpu: 10m, memory: 32Mi}
limits: {cpu: 50m, memory: 64Mi}
volumeMounts:
- name: es-scripts
mountPath: /templates
readOnly: true
volumes:, add:
values.yaml: replace the retention: block with:
retention:
# Age rule — Elasticsearch ILM deletes whole daily indices older than this
# many days (0 = keep forever). Data is kept at least N days, at most N+1.
eventsDays: 30
inferencesDays: 90
graphsDays: 14
# Disk rule — while the Elasticsearch volume is above targetPercent, the
# worker deletes the oldest whole Logster index: never the index being
# written, never one holding data newer than minAgeHours.
targetPercent: 75
minAgeHours: 24
checkIntervalMinutes: 5
resources:
requests: {cpu: 10m, memory: 32Mi}
limits: {cpu: 50m, memory: 64Mi}
templates/NOTES.txt: replace the three retention lines under Elasticsearch PVC: with:
Retention (ILM): events={{ .Values.retention.eventsDays }}d, inferences={{ .Values.retention.inferencesDays }}d, graphs={{ .Values.retention.graphsDays }}d
Disk rule: above {{ .Values.retention.targetPercent }}% the oldest whole index is deleted
(never the index being written, never data newer than {{ .Values.retention.minAgeHours }}h).
tests/e2e-openshift/lib.sh:30-31:
export ES_EVENTS_INDEX="${ES_EVENTS_INDEX:-logster-events*}"
export ES_INFER_INDEX="${ES_INFER_INDEX:-logster-inferences*}"
- [ ] Step 5: Run the drift test and render the chart
Run:
venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_*.py' 2>&1 | tail -3
helm lint deploy/openshift/logster --set elasticsearch.password=ci --set license.existingSecret=ci-license
RENDER=$(mktemp)
helm template logster deploy/openshift/logster --set elasticsearch.password=ci --set license.existingSecret=ci-license > "$RENDER"
helm template logster deploy/openshift/logster --set elasticsearch.password=ci --set license.existingSecret=ci-license -s templates/vector/deployment.yaml | grep -c 'wait-es-layout'
grep -n 'maxSurge: 0\|auto_create_index\|RETENTION_MIN_AGE_HOURS\|layout.py: |' "$RENDER"
rm -f "$RENDER"
helm lint: 1 chart(s) linted, 0 chart(s) failed.
- 1: wait-es-layout appears once inside the rendered Vector Deployment (the initContainer name). Count it there only: the full render also carries it in the logster-es-scripts ConfigMap, as layout.py's make_logger("wait-es-layout"), so a whole-render count is 2.
- The grep shows the maxSurge: 0 strategy, the guard arg, RETENTION_MIN_AGE_HOURS (retention Deployment) and the layout.py ConfigMap key.
- [ ] Step 6: Stage
git add deploy/openshift/logster deploy/elasticsearch/test_chart_drift.py tests/e2e-openshift/lib.sh
feat(chart): ILM retention layout, Vector alias gate, single-worker rollout, drift guard
Task 17: CI workflow for the retention scripts and chart
Files:
- Create: .github/workflows/es-retention-ci.yml
This follows the existing workflow conventions: actions/checkout@v4, azure/setup-helm@v4, actions/setup-python@v5, and no sudo. It runs only the unit tests, the drift test and the chart render. The integration suite needs the DHI base image (docker login dhi.io) and stays a local/pre-release gate (Task 19).
- [ ] Step 1: Create the workflow
name: es-retention-ci
# Unit tests for the stdlib Elasticsearch init/retention scripts, the chart
# drift guard (deploy/ copies vs the Helm chart's files/), and a Helm
# lint + render. The real-image integration suite
# (deploy/elasticsearch/integration/run.sh) needs the DHI base image and runs
# locally before a release.
on:
pull_request:
paths:
- "deploy/elasticsearch/**"
- "deploy/openshift/**"
- "deploy/vector/**"
- "deploy/docker-compose.yml"
- ".github/workflows/es-retention-ci.yml"
workflow_dispatch:
concurrency:
group: es-retention-ci-${{ github.ref }}
cancel-in-progress: true
jobs:
test:
name: Retention scripts + chart
runs-on: ubuntu-latest
timeout-minutes: 15
steps:
- name: Checkout
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Unit tests and chart drift guard (stdlib only)
run: python -m unittest discover -s deploy/elasticsearch -p 'test_*.py' -v
- name: Set up Helm
uses: azure/setup-helm@v4
- name: Helm lint
run: helm lint deploy/openshift/logster --set elasticsearch.password=ci --set license.existingSecret=ci-license
- name: Helm render
run: helm template logster deploy/openshift/logster --set elasticsearch.password=ci --set license.existingSecret=ci-license > /dev/null
- [ ] Step 2: Check the YAML locally
Run: venv/bin/python -c "import yaml,sys; yaml.safe_load(open('.github/workflows/es-retention-ci.yml')); print('ok')"
Expected: ok. The workflow runs for real on the PR (Task 21).
- [ ] Step 3: Stage
ci: run retention unit tests, chart drift guard and helm lint on PRs
Task 18: Documentation and CLAUDE.md
Files:
- Modify (docs): docs/technical-guides/openshift-deployment/data-retention.md (the Elasticsearch retention section, ~lines 173-255), docs/technical-guides/openshift-deployment/install-logster-app-helm.md (Prerequisites helm row; B3 "Watch it converge"), deploy/openshift/logster/README.md (the "Storage pressure and retention" section, ~lines 219-240, and the verify commands at ~286-288), docs/technical-guides/admin-guide/daily-operations.md ("Backup and retention" table), docs/technical-guides/admin-guide/important-considerations.md (ILM checklist item; pre-flight check 4), docs/technical-guides/enterprise-architecture.md ("Data stores" table; the Vector/"Logstash" write-target paragraph), packer/files/env.example, scripts/stress_test/producer.py:17
- Modify (doc queries of concrete index names → read patterns / write aliases): docs/technical-guides/troubleshooting-guide.md (incl. the "Elasticsearch disk full" section), scripts/stress_test/README.md, docs/technical-guides/logster-architecture.md, docs/technical-guides/admin-guide/installation.md, docs/technical-guides/admin-guide/model-deployment.md, docs/technical-guides/admin-guide/log-sources.md (plus enterprise-architecture.md above)
- Modify (comment / log text only, no behaviour change): services/api/src/logster_api/whitelist_store.py, services/api/src/logster_api/app.py, libs/logster-ml/logster_ml/graph/hydration.py, libs/logster-ml/logster_ml/graph/serialize.py, libs/logster-common/logster_common/es.py, services/dashboard/client/src/components/graph/CytoscapeGraph.jsx, services/dashboard/client/src/components/graph/WhitelistPanel.jsx, services/dashboard/client/src/context/DashboardContext.jsx, tests/e2e-openshift/verify-ingest.sh
- Rule for every doc command this task writes or changes: if a URL or path contains * (or ?), single-quote it, as in 'http://localhost:9200/logster-events*/_count'. Customers paste these commands into zsh as well as bash.
- Modify, local only (never staged): CLAUDE.md — it is gitignored (git check-ignore -v CLAUDE.md prints .gitignore:249:CLAUDE.md CLAUDE.md), so git add refuses it
- [ ] Step 1: Rewrite the Elasticsearch retention section of
data-retention.md
Read the file first, then replace the section that currently says "does not use ILM" (and everything about emergency/critical/flood sweeps) with:
## Elasticsearch retention
Logster stores events, inferences and graph snapshots in daily Elasticsearch
indices managed by Index Lifecycle Management (ILM). Two rules keep storage in
check:
| Rule | Who enforces it | What it does |
|---|---|---|
| **Age** | Elasticsearch ILM | Deletes whole daily indices older than `retention.eventsDays` / `inferencesDays` / `graphsDays`. Data is kept at least N days and at most N+1. `0` keeps data forever. |
| **Disk** | `elasticsearch-retention` worker, every `retention.checkIntervalMinutes` | While the Elasticsearch volume is above `retention.targetPercent`, deletes the oldest whole Logster index, and repeats on the next check if needed. |
The disk rule never deletes:
- the index currently being written;
- an index holding data newer than `retention.minAgeHours` (default 24 h). This is the limit of how far the worker will go when trying to meet the target;
- anything that is not a Logster data index.
If the volume stays above the target because nothing old enough is left, the worker logs a warning every check and the dashboard's storage banner stays up. Elasticsearch's own flood-stage protection (95 %) blocks writes before the disk can fill. Ingestion then pauses and Kafka buffers the data. No data is corrupted.
### Settings
| values.yaml | Default | Meaning |
|---|---|---|
| `retention.eventsDays` | 30 | Age rule for events (0 = keep forever) |
| `retention.inferencesDays` | 90 | Age rule for inferences |
| `retention.graphsDays` | 14 | Age rule for graph snapshots |
| `retention.targetPercent` | 75 | Disk rule target, % of the Elasticsearch volume (1–89) |
| `retention.minAgeHours` | 24 | Disk rule never deletes data newer than this (1–720) |
| `retention.checkIntervalMinutes` | 5 | How often the worker checks |
Invalid values are logged as warnings and replaced by the defaults. The worker never stops because of them. `retention.stripRawEventDays`, `intervalHours`, `emergencyThresholdPercent` and `criticalThresholdPercent` are no longer used.
### Upgrading from a release without ILM retention
Nothing is moved or reindexed. Your existing `logster-events`, `logster-inferences` and `logster-graphs` indices stay readable and are deleted by ILM once their newest document is older than the retention period. New data goes to the daily indices.
Upgrade by re-applying the values you installed with. This works on every supported Helm version (3.10+) and picks up the new release's image tags and defaults:
```bash
helm get values logster -n logster -o yaml > logster-values.yaml
helm upgrade logster ./logster-<ver>.tgz -n logster -f logster-values.yaml --timeout 20m
rm logster-values.yaml # it contains the Elasticsearch password and the license
```
Do not upgrade to a new release with `--reuse-values` alone: Helm then keeps the previous release's chart defaults, including its image tags.
Do not add `--wait` if the Splunk receiver is a `LoadBalancer` Service on a cluster without a load-balancer provider: Helm then waits for an external IP that never arrives, and the upgrade times out (verified on CRC with v1.2.1's own values; this predates the retention change).
During the upgrade the `vector` pod shows `Init:0/1` for a few minutes while Elasticsearch restarts and the retention worker creates the new indices (measured on CRC: ~2 minutes). Vector's previous pod keeps shipping until the new one is ready.
Each existing index holds up to N days of data. If the volume is above the target at upgrade time, the disk rule deletes the whole oldest existing index in one step, once its newest document is older than `retention.minAgeHours`.
### Rolling back
`helm rollback` is safe: nothing crashes and no data is corrupted. The previous release reads only the pre-upgrade indices, so data written after the upgrade is hidden until you roll forward again. Its retention worker cannot free space in the daily indices, so roll forward promptly and do not stay rolled back on a nearly full volume.
### Checking status
The Elasticsearch image has no `curl`, so these commands run through the retention worker pod, using the same stdlib client the worker itself uses (`/templates/eshttp.py METHOD PATH [JSON]`).
```bash
oc logs deploy/elasticsearch-retention -n logster --tail=50
oc exec -n logster deploy/elasticsearch-retention -- python3 /templates/eshttp.py GET '/_cat/aliases/logster-*-write?v'
oc exec -n logster deploy/elasticsearch-retention -- python3 /templates/eshttp.py GET '/logster-*/_ilm/explain?only_errors=true'
```
### Write-alias squatter
If the worker logs `SQUATTER: a concrete index named logster-<family>-write`, a plain index holds the name the write alias needs. Writes still succeed, but that index is not managed. Replace `<family>` with `events`, `inferences` or `graphs`, then:
```bash
oc scale deploy/vector -n logster --replicas=0
oc exec -n logster deploy/elasticsearch-retention -- python3 /templates/eshttp.py PUT /logster-<family>-write/_block/write
oc exec -n logster deploy/elasticsearch-retention -- python3 /templates/eshttp.py POST /logster-<family>-write/_clone/logster-<family>-000000
oc exec -n logster deploy/elasticsearch-retention -- python3 /templates/eshttp.py DELETE /logster-<family>-write
oc scale deploy/vector -n logster --replicas=1
```
The retention worker re-creates the write alias within one check. The clone keeps the data under a managed name and ages out normally.
These commands are verified on CRC in Task 20, per the rule that every customer-facing command is tested on CRC first. If any command needs to change there, update this section with the verified form.
- [ ] Step 2: Update the chart README
In deploy/openshift/logster/README.md:
- Replace the "Storage pressure and retention" section body with a short summary: the two rules, what is never deleted, the settings table link, and "See docs/technical-guides/openshift-deployment/data-retention.md for upgrade, rollback and troubleshooting."
- In the verify block (~286-288), switch both reads to the patterns, with every URL single-quoted (an unquoted * is a shell glob; zsh aborts on it with "no matches found"):
oc port-forward svc/elasticsearch 9200 & curl -su "elastic:$PW" localhost:9200/logster-events/_count → oc port-forward svc/elasticsearch 9200 & curl -su "elastic:$PW" 'localhost:9200/logster-events*/_count'
curl -su "elastic:$PW" 'localhost:9200/logster-inferences/_search?size=3&sort=@timestamp:desc' → curl -su "elastic:$PW" 'localhost:9200/logster-inferences*/_search?size=3&sort=@timestamp:desc'
- [ ] Step 3: Update the other docs
These four edits are required (spec §14); make each exactly as written.
docs/technical-guides/admin-guide/daily-operations.md, "Backup and retention" table: replace the row
| Elasticsearch indices | `es_data` volume | **No ILM policy configured by default** — configure one before long-lived deployments. |
| Elasticsearch indices | `es_data` volume | Retention is automatic. Elasticsearch ILM deletes daily indices older than the retention period (`RETENTION_EVENTS_DAYS` 30, `RETENTION_INFERENCES_DAYS` 90, `RETENTION_GRAPHS_DAYS` 14; `0` keeps data forever). Above the disk target (`RETENTION_TARGET_PERCENT`, 75 % by default) the `elasticsearch-retention` worker deletes the oldest whole index, never the one being written and never data newer than 24 hours (`RETENTION_MIN_AGE_HOURS`). On the appliance set these in `/etc/logster/logster.env` (see `/etc/logster/logster.env.example`) and run `sudo systemctl restart logster.service`; elsewhere, export them before `docker compose up`. |
docs/technical-guides/enterprise-architecture.md, "Data stores" table, the Elasticsearch row: replace the last cell Persistent (configure ILM for retention). with
Persistent. Elasticsearch ILM deletes daily indices older than the retention period; above the disk target (75 % by default) the retention worker deletes the oldest whole index.
| unchanged).
docs/technical-guides/admin-guide/important-considerations.md:
- In "Durability and persistence", replace the checklist item
(Both blocks are indented four spaces here only so plan checkbox trackers do not read them as plan steps; in the doc they stay at column 0, with their original continuation indent.)
- [ ] **Configure Elasticsearch index lifecycle management (ILM)** for
`logster-events` and `logster-inferences`. Without ILM, both
indices grow forever.
- [ ] **Review the Elasticsearch retention settings.** ILM is set up
automatically: Elasticsearch ILM deletes daily indices older than
the retention period (`RETENTION_EVENTS_DAYS` 30,
`RETENTION_INFERENCES_DAYS` 90, `RETENTION_GRAPHS_DAYS` 14 by
default; `0` keeps data forever). Above the disk target
(`RETENTION_TARGET_PERCENT`, 75 % by default) the retention worker
deletes the oldest whole index, never the one being written and
never data newer than 24 hours. Size the `es_data` volume so the
configured retention fits under the target: System Health →
Storage in the dashboard shows whether it does.
4. Delete `logster-events` and verify that ILM recreates the index
and Logstash keeps writing without intervention.
logster-events is never auto-created, so the old check no longer applies)
4. Check that retention is in place:
`curl -u elastic:$ELASTIC_PASSWORD 'http://localhost:9200/_cat/aliases/logster-*-write?v'`
lists `logster-events-write`, `logster-inferences-write` and
`logster-graphs-write`;
`curl -u elastic:$ELASTIC_PASSWORD 'http://localhost:9200/logster-*/_ilm/explain?only_errors=true'`
returns `{"indices":{}}`; and `docker compose logs elasticsearch-retention`
shows no `WARNING` lines.
packer/files/env.example: insert after the EXTERNAL_KAFKA_LAN_HOST= line (before the # NOTE — do NOT set ELASTIC_PASSWORD here. block), with one blank line before and after:
# OPTIONAL — Elasticsearch retention. Defaults shown; leave them commented
# to keep the defaults. An invalid value is logged as a warning and replaced
# by its default; the stack never refuses to start because of one.
# Age rule: Elasticsearch ILM deletes daily indices older than N days
# (0–3650; 0 = keep forever).
# RETENTION_EVENTS_DAYS=30
# RETENTION_INFERENCES_DAYS=90
# RETENTION_GRAPHS_DAYS=14
# Disk rule: while the Elasticsearch disk is above this % (1–89), the
# retention worker deletes the oldest whole index — never the index being
# written, never data newer than RETENTION_MIN_AGE_HOURS (1–720).
# RETENTION_TARGET_PERCENT=75
# RETENTION_MIN_AGE_HOURS=24
# How often the retention worker checks, in minutes (1–1440).
# RETENTION_CHECK_INTERVAL_MINUTES=5
Then sweep the remaining docs for old retention wording (zsh: keep the globs as written, all three match files):
Run: grep -n -i 'emergency\|critical sweep\|flood-stage recovery\|strip.*raw_event\|raw_event.*strip\|stripRawEvent\|intervalHours\|delete_by_query\|ILM' docs/technical-guides/admin-guide/*.md docs/technical-guides/enterprise-architecture.md docs/user-guides/*.md
For every hit that still describes the old retention behavior (or says ILM must be configured by hand), rewrite the sentence to match the table in Step 1. Use these canonical phrasings: - "Elasticsearch ILM deletes daily indices older than the retention period." - "Above the disk target (75 % by default) the retention worker deletes the oldest whole index, never the one being written and never data newer than 24 hours."
docs/technical-guides/openshift-deployment/install-logster-app-helm.md B3 "Watch it converge": add one sentence — "On a fresh install the vector pod shows Init:0/1 until Elasticsearch is up and the retention worker has created Logster's indices (a few minutes); this is expected."
docs/technical-guides/openshift-deployment/install-logster-app-helm.md Prerequisites table: change the helm CLI row from v3.8+ to v3.10+. Verified: Helm 3.8.2 and 3.9.4 fail to install/render the chart (templates/config/secrets.yaml:43: … index of untyped nil — the lookup-based secret reuse), 3.10.0 through 4.2.4 succeed; this predates the retention work but is on the customer's install path.
scripts/stress_test/producer.py:17: in the docstring cleanup hint, change logster-events/_delete_by_query to logster-events*/_delete_by_query.
services/api/src/logster_api/whitelist_store.py (Task 12 rewrote the middle of this file, so find these by text, not line number):
- The module docstring's "Reads persisted graphs from logster-graphs": say logster-graphs* instead.
- The comment found by grep -n 'has no mapping template' services/api/src/logster_api/whitelist_store.py: its first line, "logster-inferences has no mapping template, so string fields are", becomes "The logster-inferences* template maps only @timestamp, so string fields are". Leave the rest of that comment as it is.
Other stale comments that name the concrete indices (comment/log-text only — no behaviour change): services/api/src/logster_api/app.py:163 ("retention-rotated (14 days)"), libs/logster-ml/logster_ml/graph/hydration.py:8, libs/logster-ml/logster_ml/graph/serialize.py (index name in module docstring), libs/logster-common/logster_common/es.py:5, services/dashboard/client/src/components/graph/CytoscapeGraph.jsx:70, services/dashboard/client/src/components/graph/WhitelistPanel.jsx:43, services/dashboard/client/src/context/DashboardContext.jsx:48, tests/e2e-openshift/verify-ingest.sh:29 (log text). Update each to name the pattern (logster-events* etc.) or the write alias, whichever the code actually uses.
- [ ] Doc queries: concrete index names → read patterns / write aliases
After the upgrade a fresh install has no logster-events / logster-inferences / logster-graphs index at all (the auto-create guard refuses those names), so a query of the bare name answers 404. Reads switch to the read pattern logster-<family>*, which also covers the legacy index on upgraded installs. Write-target descriptions switch to the logster-<family>-write alias. Text that only describes the family or the legacy index (for example "the dashboard reads from logster-inferences directly", or a legacy-index explanation) stays as it is. Where a URL was unquoted, the new form quotes it: an unquoted * (or ?) is a shell glob, and zsh aborts on it with "no matches found".
docs/technical-guides/troubleshooting-guide.md:
- "Elasticsearch disk full" section: its Cause / Immediate fix / Structural fix describe the old layout. The fix deleted logster-events / logster-inferences and relied on auto-create to bring them back, which the guard now refuses. Replace everything from **Cause.** Elasticsearch ran out of disk space. In the default down to and including the line [Admin Guide: Important Considerations](admin-guide/important-considerations.md#durability-and-persistence). with:
**Cause.** Elasticsearch ran out of disk space. Normally retention
prevents this: Elasticsearch ILM deletes daily indices older than the
retention period, and above the disk target (75 % by default) the
retention worker deletes the oldest whole index, never the one being
written and never data newer than 24 hours. The disk still fills when
something other than Logster's indices takes the space, or when all of
Logster's data is newer than 24 hours.
**Immediate fix.**
```bash
# What is the retention worker doing? It logs every check.
docker compose logs --tail=50 elasticsearch-retention
# Which Logster indices hold the space?
curl -u elastic:$ELASTIC_PASSWORD 'http://localhost:9200/_cat/indices/logster-*?v&s=index&h=index,docs.count,store.size'
```
If the worker logs `no Logster index is older than the 24h floor —
nothing deleted`, Logster's own data is not what fills the disk, or all
of it is newer than 24 hours: free space outside Elasticsearch, or lower
`RETENTION_*_DAYS` / `RETENTION_MIN_AGE_HOURS`. On the appliance, set
them in `/etc/logster/logster.env` and run
`sudo systemctl restart logster.service` (a bare `docker compose up`
there would start the worker without its Elasticsearch password);
elsewhere, export them, then run
`docker compose up -d elasticsearch-retention` from the `deploy`
directory. Never delete Logster indices by wildcard. Elasticsearch lifts its flood-stage write block by
itself once usage drops below the 90 % high watermark, and Vector then
delivers the writes it was retrying.
**Structural fix.** Size the `es_data` volume so the configured
retention fits under the target. System Health → Storage in the
dashboard shows whether it does, and how many days remain before the
oldest data is deleted early — see
[Admin Guide: Important Considerations](admin-guide/important-considerations.md#durability-and-persistence).
_count commands (the plain one and the one followed by -H 'Content-Type: application/json' \):
curl -u elastic:$ELASTIC_PASSWORD 'http://localhost:9200/logster-inferences/_count' → curl -u elastic:$ELASTIC_PASSWORD 'http://localhost:9200/logster-inferences*/_count' (the second keeps its trailing \).
- Two inline commands — "Dashboard tiles are empty" → Fix paths item 1 (followed by "— are raw events reaching ES?") and "High error prediction rate" item 3 (followed by "should be climbing over time"):
`curl -u elastic:$ELASTIC_PASSWORD http://localhost:9200/logster-events/_count` → `curl -u elastic:$ELASTIC_PASSWORD 'http://localhost:9200/logster-events*/_count'`
- "Still stuck?" item 3:
curl -u elastic:$ELASTIC_PASSWORD 'http://localhost:9200/logster-events/_search?pretty&size=5' → curl -u elastic:$ELASTIC_PASSWORD 'http://localhost:9200/logster-events*/_search?pretty&size=5'
curl -u elastic:$ELASTIC_PASSWORD 'http://localhost:9200/logster-inferences/_search?pretty&size=5' → curl -u elastic:$ELASTIC_PASSWORD 'http://localhost:9200/logster-inferences*/_search?pretty&size=5'
scripts/stress_test/README.md (the cleanup deletes documents by query across the family, the same form as the producer.py:17 docstring; it never deletes an index):
- curl -s 'http://localhost:9200/logster-events/_count?q=tenant_id:stress' | jq . → curl -s 'http://localhost:9200/logster-events*/_count?q=tenant_id:stress' | jq .
- curl -XPOST 'http://localhost:9200/logster-events/_delete_by_query' \ → curl -XPOST 'http://localhost:9200/logster-events*/_delete_by_query' \
- curl -XPOST 'http://localhost:9200/logster-inferences/_delete_by_query' \ → curl -XPOST 'http://localhost:9200/logster-inferences*/_delete_by_query' \
docs/technical-guides/logster-architecture.md ("Query Elasticsearch" block):
- curl -u elastic:$ELASTIC_PASSWORD http://localhost:9200/logster-events/_search?pretty → curl -u elastic:$ELASTIC_PASSWORD 'http://localhost:9200/logster-events*/_search?pretty'
- curl -u elastic:$ELASTIC_PASSWORD http://localhost:9200/logster-inferences/_search?pretty → curl -u elastic:$ELASTIC_PASSWORD 'http://localhost:9200/logster-inferences*/_search?pretty'
docs/technical-guides/admin-guide/installation.md ("Confirm events are arriving"):
- curl -u elastic:$ELASTIC_PASSWORD 'http://localhost:9200/logster-events/_count' → curl -u elastic:$ELASTIC_PASSWORD 'http://localhost:9200/logster-events*/_count'
- curl -u elastic:$ELASTIC_PASSWORD 'http://localhost:9200/logster-inferences/_count' → curl -u elastic:$ELASTIC_PASSWORD 'http://localhost:9200/logster-inferences*/_count'
docs/technical-guides/admin-guide/model-deployment.md (Step 4 smoke test):
- curl -u elastic:$ELASTIC_PASSWORD 'http://localhost:9200/logster-inferences/_count' \ → curl -u elastic:$ELASTIC_PASSWORD 'http://localhost:9200/logster-inferences*/_count' \
docs/technical-guides/admin-guide/log-sources.md ("Confirm normalized events are landing in Elasticsearch"):
- curl -u elastic:$ELASTIC_PASSWORD 'http://localhost:9200/logster-events/_count' → curl -u elastic:$ELASTIC_PASSWORD 'http://localhost:9200/logster-events*/_count'
docs/technical-guides/enterprise-architecture.md, the paragraph under the "Data stores" table that says where the Kafka → Elasticsearch pipelines write: replace
`normalized-endpoint-events` to the `logster-events` index and
`logster-inference-results` to the `logster-inferences` index, using the
`normalized-endpoint-events` through the `logster-events-write` alias and
`logster-inference-results` through the `logster-inferences-write` alias
(Elasticsearch ILM rolls each alias over to a new daily index), using the
Verify that no query of a concrete name is left. A query names the index directly followed by / or ?, so plain descriptions of the family or of the legacy index do not match:
grep -rnE "logster-(events|inferences|graphs)(/|\?)" --include='*.md' --include='*.sh' --include='*.txt' --include='*.py' --include='*.yaml' docs scripts deploy/openshift/logster tests/e2e-openshift | grep -v 'docs/superpowers/'
producer.py:17; Task 16 replaced the chart's old retention.py copy, whose /logster-events/_update_by_query would otherwise match.)
- [ ] Step 4: Update
CLAUDE.md
CLAUDE.md is gitignored and untracked, so this edit stays local: Step 6 does not stage it, and the task report must say that the CLAUDE.md update is local-only.
- Architecture block: replace
ES retention worker keeps disk usage in check (emergency eviction)withES ILM ages data out; the retention worker enforces the disk target (oldest whole index first). - Tech stack, Storage bullet: replace "An
elasticsearch-retentionsidecar monitors disk usage and runs emergency eviction sweeps." with "Data indices are ILM rollover indices behind write aliases; theelasticsearch-retentionsidecar converges that layout and deletes the oldest whole index while disk is above target." - Replace the invariant paragraph "Data retention is self-healing, not preventive." with:
**Elasticsearch layout: writers use aliases, readers use patterns.** `logster-events`, `logster-inferences` and `logster-graphs` are ILM rollover families: write through `logster-<family>-write`, read through `logster-<family>*` (which also covers the pre-ILM concrete index on upgraded installs). Never GET/`_update` a document by id through a pattern or alias — search by `_id`, then update the concrete `_index` from the hit. Never delete by wildcard. `deploy/elasticsearch/layout.py` owns policies, templates and aliases (`converge()` runs in init and every worker check); `diskcap.py` is the only code that deletes indices (oldest first, never a write index, never data newer than `RETENTION_MIN_AGE_HOURS`). Age-based deletion is ILM's job — do not add delete_by_query retention. Chart copies under `deploy/openshift/logster/files/` must stay byte-identical (`test_chart_drift.py`). - "Modify data retention" list: replace item 1 with:
1. Layout (policies, templates, aliases) → deploy/elasticsearch/layout.py; disk rule → diskcap.py; loop → retention.py; knobs → retention_config.py (then cp to deploy/openshift/logster/files/elasticsearch/) - Configuration quick reference: replace the "Retention worker env vars" bullet with:
Retention env vars: RETENTION_EVENTS_DAYS (30), RETENTION_INFERENCES_DAYS (90), RETENTION_GRAPHS_DAYS (14), RETENTION_TARGET_PERCENT (75), RETENTION_MIN_AGE_HOURS (24), RETENTION_CHECK_INTERVAL_MINUTES (5) -
The
init.pycomment about "80%/90%" was removed by Task 8. Confirm withgrep -n '80%' deploy/elasticsearch/init.py; expected: no output. -
[ ] Step 5: Verify no stale wording remains
Run (the --include globs are quoted so zsh does not expand them):
grep -rn -i 'emergency sweep\|critical sweep\|emergencyThresholdPercent\|stripRawEventDays\|RETENTION_EMERGENCY\|RETENTION_STRIP_RAW' --include='*.md' --include='*.yaml' --include='*.txt' --include='*.py' --include='*.js' --include='*.jsx' . | grep -v 'docs/superpowers/' | grep -v node_modules | grep -v retention_config.py | grep -v test_ | grep -v 'are no longer used'
retention_config.py and the tests name the removed variables to warn about them, and the data-retention.md Settings line from Step 1 ("… retention.stripRawEventDays, intervalHours, emergencyThresholdPercent and criticalThresholdPercent are no longer used.") names the removed knobs so customers know to drop them.
- [ ] Step 6: Stage
git add docs/technical-guides deploy/openshift/logster/README.md packer/files/env.example scripts/stress_test/producer.py scripts/stress_test/README.md services/api/src/logster_api/whitelist_store.py services/api/src/logster_api/app.py libs/logster-ml/logster_ml/graph/hydration.py libs/logster-ml/logster_ml/graph/serialize.py libs/logster-common/logster_common/es.py services/dashboard/client/src/components/graph/CytoscapeGraph.jsx services/dashboard/client/src/components/graph/WhitelistPanel.jsx services/dashboard/client/src/context/DashboardContext.jsx tests/e2e-openshift/verify-ingest.sh
docs/technical-guides stages every edited guide under it (data-retention, install-logster-app-helm, daily-operations, important-considerations, installation, model-deployment, log-sources, enterprise-architecture, logster-architecture, troubleshooting-guide). CLAUDE.md is deliberately absent (gitignored, local-only; see Step 4). git status --short CLAUDE.md prints nothing both before and after this step.
Proposed message: docs: ILM retention, upgrade/rollback, squatter fix
Task 19: Live verification: Compose upgrade on this dev box
Files: none (runbook). Record the results in the PR description.
This upgrades the running dev stack in place. The stack's current indices are concrete, which makes them the legacy indices, so this exercises the real upgrade path. The prerequisites need the user's approval: Task 0's decision about the uncommitted engine.py, and running from this checkout.
- [ ] Step 1: Baseline
es() { curl -s -u elastic:logster-dev-es-password "localhost:9200$1" "${@:2}"; }
es '/_cat/indices/logster-*?v&h=index,docs.count,store.size'
es '/logster-events/_count'; echo; es '/logster-inferences/_count'; echo
df -h / | tail -1
logster-events, logster-inferences and logster-graphs exist. Write down their counts.
Hydration "before" sample (for the Hydration guard below). It must be taken now, before anything is upgraded. The inference container publishes no port on Compose and is distroless, so read its counters from Prometheus, which scrapes inference:8002:
prom() { curl -s 'http://localhost:9090/api/v1/query' --data-urlencode "query=$1" | python3 -c 'import json,sys; [print(r["metric"].get("result", "ratio"), r["value"][1]) for r in json.load(sys.stdin)["data"]["result"]]'; }
prom 'sum by (result) (logster_phantom_hydration_requests_total)'
prom 'sum(logster_phantom_hydration_latency_seconds_bucket{le="0.5"}) / sum(logster_phantom_hydration_latency_seconds_count)'
result (cached, hit, miss, and error only if any occurred; no error line means zero errors), then ratio <share of calls ≤ 0.5 s>. These are totals since the current (pre-upgrade) inference container started. Write down the error share of the total and the ratio. If both queries print nothing, there was no hydration traffic before the upgrade: record that, and use the reference numbers in the Hydration guard as the baseline.
- [ ] Step 2: Integration suite green on the image we'll run
Run: deploy/elasticsearch/integration/run.sh 2>&1 | tail -30
Expected: all scenarios pass (Tasks 9 and 10).
- [ ] Step 3: Build, check the disk, then roll out
3a. Build first: the new image layers land on this same disk, so the disk check below must see them.
3b. Disk check: pick this box's retention target. The upgraded worker enforces RETENTION_TARGET_PERCENT (75 by default) from its first check, right after up -d, and on every check after that. It never deletes data newer than 24 h (RETENTION_MIN_AGE_HOURS). While ingest is live, each legacy index's newest document is recent, so the legacy indices are protected until about 24 h after the old Vector stops writing to them (the Vector restart in 3c). On an idle box their newest document may already be older than 24 h, and then the very first check deletes them if disk is above the target. Oldest goes first: the legacy indices hold this box's whole pre-upgrade history.
PCT=$(curl -s -u elastic:logster-dev-es-password 'localhost:9200/_cluster/stats?filter_path=nodes.fs' | python3 -c 'import json,sys; f=json.load(sys.stdin)["nodes"]["fs"]; print(int((f["total_in_bytes"]-f["available_in_bytes"])/f["total_in_bytes"]*100))')
echo "disk: ${PCT}% suggested target with 5 points of headroom: $(( PCT + 5 > 89 ? 89 : PCT + 5 ))%"
for i in logster-events logster-inferences logster-graphs; do
curl -s -u elastic:logster-dev-es-password "localhost:9200/$i/_search?size=0" -H 'Content-Type: application/json' -d '{"aggs":{"newest":{"max":{"field":"@timestamp"}}}}' | python3 -c "import json,sys; d=json.load(sys.stdin); print('$i newest @timestamp:', ((d.get('aggregations') or {}).get('newest') or {}).get('value_as_string') or d.get('error', {}).get('type', 'no data'))"
done
echo 75 > "${TMPDIR:-/tmp}/logster-t19-target"
RETENTION_TARGET_PERCENT this dev box should run with. Show them the disk %, each legacy index's newest @timestamp, and the suggested target, and explain:
- With the default 75 %, and with usage above it, the worker deletes the legacy logster-events / logster-inferences / logster-graphs indices, oldest first, once their newest document is older than 24 h. That is right away for any whose newest @timestamp is already older, and about 24 h after 3c for the rest.
- To keep them, pick a target with at least 5 points of headroom above current usage (the suggestion above). Usage keeps growing with ingest, so a target just above today's usage is soon exceeded.
- The worker accepts only 1–89; a larger value logs a warning and falls back to 75. So above 84 % the headroom shrinks, and at 89 % or more no valid target protects them. Say so, and let the user choose between freeing disk first and accepting the deletion.
Write the answer into the file, e.g. for 85: echo 85 > "${TMPDIR:-/tmp}/logster-t19-target".
3c. Roll out:
cd deploy
RETENTION_TARGET_PERCENT=$(cat "${TMPDIR:-/tmp}/logster-t19-target") docker compose -f docker-compose.yml -f docker-compose.ci.yml --profile services up -d
docker compose restart vector # its bind-mounted vector.yaml changed
cd ..
docker compose … up of elasticsearch-retention in this task passes the same RETENTION_TARGET_PERCENT=$(cat "${TMPDIR:-/tmp}/logster-t19-target"), so the box keeps the chosen target.
The docker-compose.ci.yml overlay is required on this box: without it the inference container exits 78 because the license isn't mounted.
Expected: Elasticsearch is recreated (its command changed) and becomes healthy, elasticsearch-init exits 0, and every service is Up.
- [ ] Step 4: Verify the layout and legacy adoption
es() { curl -s -u elastic:logster-dev-es-password "localhost:9200$1" "${@:2}"; }
docker inspect deploy-elasticsearch-init-1 --format '{{.State.ExitCode}}'
docker compose -f deploy/docker-compose.yml logs elasticsearch-init | tail -20
es '/_cat/aliases/logster-*-write?v'
es '/logster-events,logster-inferences,logster-graphs/_settings?flat_settings=true&filter_path=**.index.lifecycle*'
es '/logster-*/_ilm/explain?only_errors=true'; echo
es '/_cluster/settings?include_defaults=true&filter_path=**.auto_create_index'; echo
0, and init logs end with done..
- Three write aliases on logster-*-000001.
- The legacy indices carry logster-<family>-legacy, an origination_date and index.lifecycle.indexing_complete: "true".
- {"indices":{}} (no ILM errors).
- The guard pattern shows under auto_create_index.
- [ ] Step 5: Ingest and read back
es() { curl -s -u elastic:logster-dev-es-password "localhost:9200$1" "${@:2}"; }
es '/_cat/indices/logster-*?v&h=index,docs.count&s=index' # sample A
source venv/bin/activate
python scripts/stress_test/producer.py --endpoints 2 --events-per-second-per-endpoint 5 --duration-seconds 120
es '/_cat/indices/logster-*?v&h=index,docs.count&s=index' # sample B, ≥ 120 s after A
docker compose restart vector in Step 3 (spec §8.2). That is not a failure.
- After the restart they stay flat: in samples A and B, taken ≥ 60 s apart with the producer running in between, logster-events and logster-inferences show identical counts, while logster-events-000001 is higher in B. logster-inferences-000001 (and logster-graphs-000001, graph persistence is on in service-config.yaml) grow as inference windows close; if B does not show it yet, re-read a few minutes later — the legacy counts must still be unchanged then.
-
[ ] Step 6: Exercise every reader
-
Open
http://localhost:5001and check each view: - The Inferences list loads.
- Open a new inference and an old one (from before the upgrade): both show their events.
- Graph view opens for both.
- On an attack inference, the Evidence walkthrough runs, and a second open is served from cache.
- Whitelist preview and create work on a graph.
- System Health → Storage shows one row per family, the "Days until early deletion" tile, and no stale-forecast warning (the worker publishes within 5 min).
docker compose -f deploy/docker-compose.yml logs --since 15m inference | grep -i -E 'hydrat|graph sink' | tail: no errors. There are no "Graph sink: N of M docs rejected" warnings.-
docker compose -f deploy/docker-compose.yml logs --since 15m elasticsearch-retention:starting — events=30d … check_interval=5m, and no WARNING lines. One exception: if this box runs with a target below its current usage (3b), expectWARNING: disk …% > target …% and no Logster index is older than the 24h floor — nothing deletedon each check while all its data is newer than 24 h. Any other WARNING is a failure. -
[ ] Browser verification (Playwright)
Drive the dashboard in a real browser with the Playwright MCP tools; this is the recorded pass over the Step 6 views. Load the tools first with ToolSearch select:mcp__playwright__browser_navigate,mcp__playwright__browser_snapshot,mcp__playwright__browser_click,mcp__playwright__browser_type,mcp__playwright__browser_console_messages,mcp__playwright__browser_network_requests,mcp__playwright__browser_take_screenshot.
mcp__playwright__browser_navigatetohttp://localhost:5001. If the login page appears (AUTH_MODE=local), ask the user for the dashboard credentials. Never reset or change the password.- Storage.
mcp__playwright__browser_clickthe "System Health" page switcher in the top bar, thenmcp__playwright__browser_snapshot. The Storage section must list exactly one row per family (logster-events,logster-inferences,logster-graphs, andlogster-alertsif that index exists), with no…-000001rows. Each row's docs count must be the legacy index plus its rollover indices summed: compare withcurl -s -u elastic:logster-dev-es-password 'localhost:9200/_cat/indices/logster-events*?h=index,docs.count'(the two reads differ only by what was ingested in between). The "Days until early deletion" tile is shown, and "Days to emergency sweep" is gone.mcp__playwright__browser_take_screenshotwith filenamees-ilm-t19-storage.png. - Inference detail. Back on "Dashboard", in the "Inference Log" section, open one inference from before the upgrade and one from after (Step 5). Each opens the investigation modal with its events listed. Screenshot
es-ilm-t19-inference.png. - Graph view (
ENABLE_GRAPH_VIEWis"true"indeploy/docker-compose.yml). Open the graph from the inference's row or its investigation modal: the graph renders with nodes. Screenshotes-ilm-t19-graph.png. - Recent attacks. The sidebar "Alert Feed" loads.
mcp__playwright__browser_network_requestsshowsGET /api/attacks/recent→ 200. An empty feed is fine if the time range has no attacks; the 200 is the check. Screenshotes-ilm-t19-alerts.png. - Search. Click "Search" in the top bar, then
mcp__playwright__browser_typedesktop-stress-0000(a Step 5 endpoint) and submit. Results render, andGET /api/search…→ 200 inmcp__playwright__browser_network_requests. Screenshotes-ilm-t19-search.png. - Console.
mcp__playwright__browser_console_messages: zeroerrorentries across the whole session.
Evidence to keep for Step 8 and the task report: the five screenshots (the tool prints where it saved each file), the browser_console_messages output, and the browser_network_requests lines for /api/storage, /api/inference/…, /api/graph/…, /api/attacks/recent and /api/search with their status codes.
- [ ] Hydration guard (graph-building input must not change — spec review F5)
The "before" sample was taken in Step 1. After at least 30 minutes of ingest on the new layout (counting from Step 3), read the same counters for the last 30 minutes only. increase() covers just the new layout and handles the counter reset from the recreated inference container:
prom() { curl -s 'http://localhost:9090/api/v1/query' --data-urlencode "query=$1" | python3 -c 'import json,sys; [print(r["metric"].get("result", "ratio"), r["value"][1]) for r in json.load(sys.stdin)["data"]["result"]]'; }
prom 'sum by (result) (increase(logster_phantom_hydration_requests_total[30m]))'
prom 'sum(increase(logster_phantom_hydration_latency_seconds_bucket{le="0.5"}[30m])) / sum(increase(logster_phantom_hydration_latency_seconds_count[30m]))'
python scripts/stress_test/producer.py --endpoints 2 --events-per-second-per-endpoint 5 --duration-seconds 1800, started with run_in_background).
Expected: the result="error" share of logster_phantom_hydration_requests_total does not rise (errors include the 0.5 s timeouts), and the le="0.5" latency bucket still holds ≥ 99.9 % of _count. Reference (100k events, chart ES sizing, 4 concurrent callers): 1 index p99 40 ms; 36 daily indices p99 160 ms, worst 174 ms, 0/400 over 0.5 s. If errors rise, stop and report — do not change phantom_hydration.timeout_seconds without the user's decision (it changes graph-building behaviour).
- [ ] Step 7: Exercise the disk rule on live data (floor path)
The destructive variant below can delete this box's data, including the legacy indices (its pre-upgrade history). The worker's first check runs the moment its container starts, so nothing in 7b runs before the user has said yes.
7a. Read-only: current usage and the indices that exist.
PCT=$(curl -s -u elastic:logster-dev-es-password 'localhost:9200/_cluster/stats?filter_path=nodes.fs' | python3 -c 'import json,sys; f=json.load(sys.stdin)["nodes"]["fs"]; print(int((f["total_in_bytes"]-f["available_in_bytes"])/f["total_in_bytes"]*100))')
echo "disk: ${PCT}%"
curl -s -u elastic:logster-dev-es-password 'localhost:9200/_cat/indices/logster-*?v&h=index,docs.count,store.size&s=index'
7b. Destructive — only after an explicit yes. The target is PCT−1, clamped to 1–89 (an out-of-range value would be rejected and replaced by 75):
PCT=$(curl -s -u elastic:logster-dev-es-password 'localhost:9200/_cluster/stats?filter_path=nodes.fs' | python3 -c 'import json,sys; f=json.load(sys.stdin)["nodes"]["fs"]; print(int((f["total_in_bytes"]-f["available_in_bytes"])/f["total_in_bytes"]*100))')
T=$(( PCT - 1 )); T=$(( T < 1 ? 1 : (T > 89 ? 89 : T) ))
echo "target: ${T}%"
cd deploy && RETENTION_TARGET_PERCENT=$T RETENTION_MIN_AGE_HOURS=1 docker compose up -d elasticsearch-retention && cd ..
docker compose -f deploy/docker-compose.yml logs --since 2m elasticsearch-retention and the _cat/indices command from 7a.
Expected: the worker logs starting — … target=<T>% min_age=1h …, then either WARNING: disk …% > target …% and no Logster index is older than the 1h floor — nothing deleted, or retention: deleted … lines naming only indices whose newest data is older than 1 h. The write indices (logster-*-000001, or whichever _cat/aliases/logster-*-write names) are never deleted.
7c. Non-destructive — when the user says no. The target is PCT+1, clamped to 1–89, so usage stays below it and nothing is eligible for deletion. At 89 % or more no valid target lies above usage (min(PCT+1, 89) ≤ usage), so this variant would delete too: skip it, and record why.
PCT=$(curl -s -u elastic:logster-dev-es-password 'localhost:9200/_cluster/stats?filter_path=nodes.fs' | python3 -c 'import json,sys; f=json.load(sys.stdin)["nodes"]["fs"]; print(int((f["total_in_bytes"]-f["available_in_bytes"])/f["total_in_bytes"]*100))')
if [ "$PCT" -ge 89 ]; then echo "disk ${PCT}% >= 89%: no non-destructive target exists, skipping"; else
T=$(( PCT + 1 )); T=$(( T < 1 ? 1 : (T > 89 ? 89 : T) ))
echo "target: ${T}%"
cd deploy && RETENTION_TARGET_PERCENT=$T RETENTION_MIN_AGE_HOURS=1 docker compose up -d elasticsearch-retention && cd ..
fi
starting — … target=<T>% min_age=1h …, no retention: deleted line, and _cat/indices lists the same indices as in 7a.
7d. Either way, restore this box's target (the one chosen in Step 3b):
cd deploy && RETENTION_TARGET_PERCENT=$(cat "${TMPDIR:-/tmp}/logster-t19-target") docker compose up -d elasticsearch-retention && cd ..
starting — … target=<that value>% min_age=24h ….
- [ ] Step 8: Record the results
Write into the PR description: - the baseline and after counts (samples A and B from Step 5); - the hydration before/after numbers; - the best_compression ratio; - screenshots or notes for Step 6, and the Playwright evidence (screenshots, console messages, network status codes); - the Step 7 log lines, and which variant (7b or 7c) the user chose; - the retention target this box was left on.
Task 20: Live verification: OpenShift on CRC (air-gapped posture)
Files: none (runbook). This needs the CRC environment (memory: e2e OpenShift access how-to; tests/e2e-openshift/*). Confirm with the user that CRC is up and who runs it before starting.
Air-gapped posture here means every image comes from the CRC internal registry via global.imageRegistry, and no pod pulls from the internet. That is exactly how the customer runs the chart.
- [ ] Step 1: Build and load this branch's images into the CRC internal registry
Build the images locally (cd deploy && docker compose --profile services build && cd ..). Tag them logster-<svc>:es-ilm-rc and load them into the CRC registry project, the same way tests/e2e-openshift/load-images-crc.sh does for a release zip. The images changed by this work are logster-elasticsearch-init, logster-inference, logster-api and logster-dashboard; load the full set anyway so the tag is consistent.
Package this branch's chart, locally from the repo root:
Expected:Successfully packaged chart and saved it to: /tmp/tmp.…/logster-1.2.1-es-ilm-rc.tgz. --version sets the version in the packaged Chart.yaml (what helm history shows in its CHART column) as well as the file name; the repo's Chart.yaml stays at version: 1.2.1. Without it the file would be logster-1.2.1.tgz and the release history would show logster-1.2.1 for both revisions, indistinguishable from the v1.2.1 release chart. Copy the printed file to the CRC host the same way the images got there. The Helm commands below run on the CRC host with CHART set to that copy's path, for example CHART=~/logster-1.2.1-es-ilm-rc.tgz.
-
[ ] Step 2: Upgrade test from the current release
-
Install the current release (v1.2.1) with
tests/e2e-openshift/install-app-crc.shand let ingest run. Use the RHEL-8 sensor, or the stress producer against the Kafka route. - Record
logster-events/_countandlogster-inferences/_count. Then take the hydration "before" sample (used by the Hydration guard below) from the v1.2.1 inference pod, now, before anything is upgraded. On the CRC host, port-forward to theinferenceService'smetricsport (8002 in the chart and in v1.2.1); no pod or image is pulled:Record theoc port-forward -n "$NS" svc/inference 18002:8002 >/dev/null 2>&1 & PF=$! timeout 30 sh -c 'until curl -sf -o /dev/null localhost:18002/metrics; do sleep 1; done' curl -s localhost:18002/metrics | grep -E '^logster_phantom_hydration_(requests_total|latency_seconds_(bucket|count|sum))' kill $PFresult="error"share oflogster_phantom_hydration_requests_total(summed overplatform) and thele="0.5"bucket's share of_count. These are totals since the v1.2.1 pod started. If nologster_phantom_hydration_lines print, there was no hydration traffic: record that, and the reference numbers in the Hydration guard become the baseline. - Upgrade to this chart exactly as the customer doc says (Task 18), once with Helm 3.10.0 (the oldest Helm that can render the chart;
~/helm3/v3.10.0/helmon the CRC host) and, on a second fresh v1.2.1 baseline, with Helm 4.2.4 (current release, server-side apply):(H=~/helm3/v3.10.0/helm $H get values logster -n "$NS" -o yaml > logster-values.yaml $H upgrade logster "$CHART" -n "$NS" -f logster-values.yaml \ --set images.tag=es-ilm-rc --timeout 20m --wait rm logster-values.yaml--set images.tagis only needed because the test images carry a non-release tag; a packaged release chart already defaults to its own tag.)
Expected:
- helm upgrade --wait completes, with no deadlock.
- oc get job elasticsearch-init is Complete.
- The Vector pod shows wait-es-layout Completed.
- oc logs deploy/elasticsearch-retention shows starting — … and no WARNING.
- tests/e2e-openshift/verify-ingest.sh passes with the pattern defaults.
- The event count after ingest ≥ the recorded count plus new events.
- The legacy counts may be above the item-2 record: under the default RollingUpdate the old Vector pod keeps writing to the legacy indices until the new one is ready (spec §8.2). Once oc rollout status deploy/vector -n "$NS" reports the new pod, the legacy counts stay flat. Sample twice, ≥ 60 s apart, with ingest running. Sample 1, right after the rollout status returns: date +%s > /tmp/logster-t20-s1; oc exec -n "$NS" deploy/elasticsearch-retention -- python3 /templates/eshttp.py GET '/_cat/indices/logster-*?v&h=index,docs.count&s=index'. Then run tests/e2e-openshift/verify-ingest.sh (the item above), which waits on ingest itself. Then take sample 2 with the same oc exec, once at least 60 s have passed since sample 1. If verify-ingest finished sooner, wait with a Monitor until-loop instead of a foreground sleep: until [ "$(date +%s)" -ge $(( $(cat /tmp/logster-t20-s1) + 60 )) ]; do sleep 5; done. logster-events and logster-inferences are identical in both samples, while logster-events-000001 is higher in the second.
- Aliases and legacy adoption are as in Task 19 Step 4.
Also verify the customer's B7 pattern on the upgraded release: helm upgrade logster "$CHART" -n "$NS" --reuse-values --set splunkReceiver.service.type=LoadBalancer. Expected: it succeeds, RETENTION_MIN_AGE_HOURS renders "24", and images stay on es-ilm-rc.
Then record what a plain --reuse-values upgrade from v1.2.1 does (on a throwaway baseline): expected per Helm's documented behaviour, it keeps v1.2.1's chart defaults including images.tag, which is why the customer doc uses helm get values … -f instead. Put the observed result in the PR.
- [ ] Hydration guard (graph-building input must not change — spec review F5)
The "before" sample was taken in Step 2 item 2. After at least 30 minutes of ingest on the upgraded release, scrape the new inference pod the same way (its counters started from zero at the upgrade, so they cover only the new layout) and compare:
oc port-forward -n "$NS" svc/inference 18002:8002 >/dev/null 2>&1 & PF=$!
timeout 30 sh -c 'until curl -sf -o /dev/null localhost:18002/metrics; do sleep 1; done'
curl -s localhost:18002/metrics | grep -E '^logster_phantom_hydration_(requests_total|latency_seconds_(bucket|count|sum))'
kill $PF
result="error" share of logster_phantom_hydration_requests_total does not rise (errors include the 0.5 s timeouts), and the le="0.5" latency bucket still holds ≥ 99.9 % of _count. Reference (100k events, chart ES sizing, 4 concurrent callers): 1 index p99 40 ms; 36 daily indices p99 160 ms, worst 174 ms, 0/400 over 0.5 s. If errors rise, stop and report — do not change phantom_hydration.timeout_seconds without the user's decision (it changes graph-building behaviour).
- [ ] Browser verification (Playwright), dashboard served from OpenShift
Run this on the upgraded release, before the rollback. Load the Playwright MCP tools first with ToolSearch select:mcp__playwright__browser_navigate,mcp__playwright__browser_snapshot,mcp__playwright__browser_click,mcp__playwright__browser_type,mcp__playwright__browser_console_messages,mcp__playwright__browser_network_requests,mcp__playwright__browser_take_screenshot.
- Route check (router + TLS), then a plain-HTTP tunnel for the browser. The CRC router serves a self-signed certificate, which the Playwright browser rejects, so the browser goes through a tunnel to the dashboard Service and the Route is checked separately with curl.
- Route, on the CRC host:
DHOST=$(oc get route dashboard -n "$NS" -o jsonpath='{.spec.host}'); curl -sk -o /dev/null -w '%{http_code}\n' "https://$DHOST/". Expected:200(dashboard-logster.apps-crc.testingfor the defaultNS=logster;-kbecause of the self-signed router certificate). - Port-forward, on the CRC host, in the background.
dashboardis the chart's Service (templates/app/services.yaml: port5000, namedhttp): - SSH tunnel, from this dev box, as a background process (Bash tool
run_in_background):ssh -N -L 15000:localhost:15000 chad@logster-cluster - Check, on this dev box:
curl -s -o /dev/null -w '%{http_code}\n' http://localhost:15000/. Expected:200. mcp__playwright__browser_navigatetohttp://localhost:15000/. If the login page appears (AUTH_MODE=local), ask the user for the dashboard credentials; never reset or change the password.- Storage.
mcp__playwright__browser_click"System Health" in the top bar, thenmcp__playwright__browser_snapshot. The Storage section lists exactly one row per family (logster-events,logster-inferences,logster-graphs, andlogster-alertsif that index exists), with no…-000001rows. Each row's docs count is the legacy index plus its rollover indices summed: compare withoc exec -n "$NS" deploy/elasticsearch-retention -- python3 /templates/eshttp.py GET '/_cat/indices/logster-events*?h=index,docs.count'(the two reads differ only by what was ingested in between). The "Days until early deletion" tile is shown, and "Days to emergency sweep" is gone.mcp__playwright__browser_take_screenshotwith filenamees-ilm-t20-storage.png. - Inference detail. On "Dashboard", in the "Inference Log" section, open one inference from before the upgrade (v1.2.1 ingest) and one from after. Each opens the investigation modal with its events listed. Screenshot
es-ilm-t20-inference.png. - Graph view (the chart's
dashboard.enableGraphViewdefaults to"true"; skip this item only if the release overrides it to false). Open the graph from the inference's row or its investigation modal: the graph renders with nodes. Screenshotes-ilm-t20-graph.png. - Recent attacks. The sidebar "Alert Feed" loads, and
mcp__playwright__browser_network_requestsshowsGET /api/attacks/recent→ 200 (an empty feed is fine if the range has no attacks). Screenshotes-ilm-t20-alerts.png. - Search. Click "Search" in the top bar, then
mcp__playwright__browser_typethe endpoint id of a host that sent events after the upgrade (the RHEL-8 sensor's hostname, ordesktop-stress-0000if you used the stress producer) and submit. Results render, andGET /api/search…→ 200. Screenshotes-ilm-t20-search.png. - Console.
mcp__playwright__browser_console_messages: zeroerrorentries across the whole session. - Stop both background processes. On this dev box:
pkill -f 'ssh -N -L 15000:localhost:15000 chad@logster-cluster'(or stop the background shell). On the CRC host:kill "$(cat /tmp/logster-pf-dashboard.pid)" && rm -f /tmp/logster-pf-dashboard.pid. Expected:curl -s -o /dev/null -w '%{http_code}\n' http://localhost:15000/on the dev box now prints000.
Evidence to keep for Step 7 and the task report: the Route curl status, the five screenshots (the tool prints where it saved each file), the browser_console_messages output, and the browser_network_requests lines for /api/storage, /api/inference/…, /api/graph/…, /api/attacks/recent and /api/search with their status codes.
- [ ] Step 3: Rollback test
logster-events: its count grows.
- The dashboard loads.
Then roll forward again with the Step 2 upgrade command. Expected: Vector switches back to the aliases; the legacy origination_date moves to the newest document and the legacy index carries index.lifecycle.indexing_complete: "true"; and within a few minutes oc exec -n "$NS" deploy/elasticsearch-retention -- python3 /templates/eshttp.py GET '/logster-*/_ilm/explain?only_errors=true' returns {"indices":{}} — no legacy-named index stuck in the family policy's hot/rollover step. (If the legacy index had aged out before the rollback, v1.2.1's writers re-created it under the new template; adoption releases it, as integration test_14 proves.)
- [ ] Step 4: Fresh install with
--wait
tests/e2e-openshift/teardown-crc.sh, then install this chart fresh with helm install logster "$CHART" … --wait --timeout 20m (the Step 1 package; flags as in install-app-crc.sh, plus --set images.tag=es-ilm-rc).
Expected:
- The install completes.
- logster-*-000001 write indices exist, and there are no legacy indices.
- verify-ingest.sh passes.
- [ ] Step 5: Air-gap check
oc get pods -n "$NS" -o jsonpath='{range .items[*]}{range .spec.initContainers[*]}{.image}{"\n"}{end}{range .spec.containers[*]}{.image}{"\n"}{end}{end}' | sort -u
oc get events -n "$NS" | grep -i -E 'pull|ErrImage' | tail
ErrImagePull events.
- [ ] Step 6: Verify the customer-facing commands
Run each command in the "Checking status" and "Write-alias squatter" sections of data-retention.md (Task 18) on CRC. For the squatter, create one on purpose first: scale Vector to 0, delete the logster-graphs-write alias with POST _aliases remove, create a concrete index logster-graphs-write, then follow the fix steps. Fix the doc wherever a command needs to change, and keep every command single-line so it pastes cleanly.
- [ ] Step 7: Record the results in the PR description: helm output, verify-ingest output, counts (including the two post-upgrade legacy samples), the hydration before/after numbers, the Playwright evidence, rollback observations, and the image list.
Task 21: Final review and hand-off
- [ ] Step 1: Full test matrix
Run each command on its own. Two suites carry recorded pre-existing failures, so never chain with &&.
venv/bin/python -m unittest discover -s deploy/elasticsearch -p 'test_*.py' 2>&1 | tail -3
venv/bin/python -m pytest services/api/tests services/inference/tests/test_graph_sink.py -q 2>&1 | tail -2
venv/bin/python -m pytest services/inference/tests -q 2>&1 | grep -E '^FAILED|passed|failed'
(cd services/dashboard/server && npm test 2>&1 | grep -E '^ℹ (tests|pass|fail)')
(cd services/dashboard/client && npm test 2>&1 | grep -E '^✖|^ℹ (tests|pass|fail)')
(cd services/dashboard/client && npm run build 2>&1 | tail -3)
helm lint deploy/openshift/logster --set elasticsearch.password=ci --set license.existingSecret=ci-license
OK. API tests and test_graph_sink.py: all pass. Dashboard server: ℹ fail 0. Build: ✓ built in …. helm lint: 1 chart(s) linted, 0 chart(s) failed.
- Full inference suite: 8 failed, 373 passed, and the FAILED lines are exactly the 8 recorded baseline failures: test_script_graph_integration.py::test_v1_inference_persists_the_graph_passed_to_classifier, test_script_pinning.py::test_all_script_nodes_survive_budget_pruning, and six in test_service_license.py (test_process_event_consumes_license_and_buffers, test_network_event_buffers_without_calling_check, test_only_process_and_file_consume_license, test_over_cap_counted_event_still_buffered, test_expired_license_terminates_on_counted_event, test_expired_license_does_not_terminate_on_uncounted_event). If one of them has been fixed elsewhere, the counts shift by that much; any FAILED line not in this list is a regression to fix before hand-off.
- Client: ℹ tests 12, ℹ pass 11, ℹ fail 1, and the only ✖ test is eBPF process event uses eBPF label (recorded baseline, also failing on dev).
- Report the two known-failure sets to the user in the hand-off as pre-existing, not "all green".
- [ ] Step 2: Whole-branch review
Use superpowers:requesting-code-review for a fresh reviewer on the most capable model, pointed at the spec, this plan and git diff dev...HEAD. Fix confirmed findings and rerun Step 1.
- [ ] Step 3: Hand off
Report to the user:
- what changed, per task;
- test and live-verification results;
- the best_compression ratio;
- any contingency that fired in Task 9 or 10;
- that the CLAUDE.md update (Task 18 Step 4) is local-only: CLAUDE.md is gitignored (.gitignore:249), so it is in no commit or PR.
Ask whether to commit (as Abdullah Chaudhry, no Co-Authored-By) and open a PR. The PR description should include the release-notes items from spec §14: the new layout, removed knobs, the one-time legacy granularity, rollback behavior and reduced disk protection while rolled back, "at least N, at most N+1 days", ingestion pausing for up to Kafka's retention in a disk emergency, and upgrading with helm get values … > file + helm upgrade -f file (works on Helm 3.10+).