Data Retention Safety — 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: Make data retention foolproof — the system should never crash from out-of-space, regardless of PVC size.

Architecture: Three layers of protection: (1) Kafka ships safe retention defaults so the broker never fills its disk, (2) Elasticsearch re-enables disk watermarks as a circuit breaker and adds an emergency retention sweep that accelerates cleanup under disk pressure, (3) a dashboard warning banner alerts users before storage becomes critical.

Tech Stack: Helm templates (YAML), Python stdlib (retention.py, init.py — no pip), React + Lucide icons (dashboard client), Express (dashboard server)

Spec: docs/superpowers/specs/2026-09-20-data-retention-safety-design.md

Global Constraints

• Python scripts (init.py, retention.py) use stdlib only — no pip, no requests, no third-party packages. They run on the logster-elasticsearch-init image (Alpine + python3).
• Helm values that are large integers MUST be quoted strings (Helm renders large YAML ints in scientific notation, which Strimzi silently rejects).
• The retention block in values.yaml is at top level (.Values.retention.*), not nested under elasticsearch.
• RETENTION_RUN_ONCE env var must continue to work in retention.py (test harness uses it).
• Dashboard server follows the existing pattern: env-var config in config.js, route in routes/, client fetches via fetchRaw().
• No commits unless explicitly asked.

---

Task 1: Kafka Safe Retention Defaults

Files: - Modify: deploy/openshift/logster/values.yaml:68-77 (kafka.config block) - Create: deploy/openshift/logster/templates/NOTES.txt - Modify: deploy/docker-compose.yml:14-30 (kafka environment block)

Interfaces: - Consumes: nothing - Produces: Kafka retention config that later tasks can assume is active

• [ ] Step 1: Add retention settings to values.yaml kafka.config

In deploy/openshift/logster/values.yaml, add three entries to the kafka.config block (after line 77, before topics:). The chart has 51 total partitions (5 topics × 6 + 1 × 12 + 1 × 6 + 1 × 3). Budget: 70% of 10Gi = 7Gi. Per-partition: 7Gi / 51 ≈ 140 MiB.

  config:
    message.max.bytes: "10485760"
    replica.fetch.max.bytes: "10485760"
    auto.create.topics.enable: "true"
    num.partitions: "6"
    offsets.topic.replication.factor: "1"
    transaction.state.log.replication.factor: "1"
    transaction.state.log.min.isr: "1"
    default.replication.factor: "1"
    min.insync.replicas: "1"
    # Retention — sized for the default 10Gi PVC. The chart defines 51
    # partitions total; 51 × 140 MiB ≈ 7 Gi (70% of 10Gi), leaving 30%
    # for in-flight segments + KRaft metadata. Adjust proportionally if
    # you change storageSize or partition counts.
    log.retention.hours: "72"
    log.retention.bytes: "146800640"
    log.segment.bytes: "52428800"

• [ ] Step 2: Add retention env vars to docker-compose.yml kafka service

In deploy/docker-compose.yml, add three env vars to the kafka: service environment: block, after KAFKA_REPLICA_FETCH_MAX_BYTES (line 29):

            KAFKA_LOG_RETENTION_HOURS: 72
            KAFKA_LOG_RETENTION_BYTES: 146800640
            KAFKA_LOG_SEGMENT_BYTES: 52428800

• [ ] Step 3: Create NOTES.txt with storage warnings

Create deploy/openshift/logster/templates/NOTES.txt:

Logster has been installed in namespace {{ .Release.Namespace }}.

== Storage ==

Kafka PVC: {{ .Values.kafka.storageSize }}
  Retention is sized for 10Gi (51 partitions × 140 MiB ≈ 70%).
  If you changed kafka.storageSize, recalculate log.retention.bytes:
    (storageSize × 0.7) / total_partitions

{{- if (index .Values.kafka.config "log.retention.bytes") }}
  log.retention.bytes = {{ index .Values.kafka.config "log.retention.bytes" }}
  log.retention.hours = {{ index .Values.kafka.config "log.retention.hours" }}
  log.segment.bytes   = {{ index .Values.kafka.config "log.segment.bytes" }}
{{- end }}

  IMPORTANT: When using helm upgrade --set for byte values, use --set-string:
    --set-string 'kafka.config.log\.retention\.bytes=146800640'
  Helm converts large integers to scientific notation, which Strimzi
  silently rejects (the value is ignored with no error).

Elasticsearch PVC: {{ .Values.elasticsearch.storageSize }}
  Retention: events={{ .Values.retention.eventsDays }}d, inferences={{ .Values.retention.inferencesDays }}d, graphs={{ .Values.retention.graphsDays }}d
  Emergency sweep triggers at {{ .Values.retention.emergencyThresholdPercent | default 80 }}% disk.

Dashboard: https://{{ (index .Values "dashboard" "host" | default "<route-host>") }}
  Default login: admin / admin (password change required on first login).

• [ ] Step 4: Verify Helm template renders correctly

cd deploy/openshift/logster
helm template test . 2>&1 | grep -A3 'log.retention'
helm template test . 2>&1 | grep -A5 'NOTES'

Expected: the three retention settings appear in the Kafka CR config section as quoted strings, and the NOTES output shows the storage section.

---

Task 2: Elasticsearch Disk Watermarks

Files: - Modify: deploy/elasticsearch/init.py:105-114 (watermark settings)

Interfaces: - Consumes: nothing - Produces: ES watermarks active at startup, which Task 3's emergency sweep and flood-stage recovery depend on

• [ ] Step 1: Replace disabled watermarks with explicit thresholds in init.py

In deploy/elasticsearch/init.py, replace lines 105–114 (the watermark-disable block) with:

    # Disk watermarks protect against disk-full corruption. On a single-node
    # cluster the flood_stage is the primary protection: it makes indices
    # read-only at 95% disk, preventing writes that would corrupt data.
    # The low/high watermarks are harmless on single-node (they affect shard
    # allocation/relocation, which requires other nodes). The retention
    # worker's emergency sweep targets 80%/90%, giving it a window to free
    # space before the flood stage fires.
    print("[init] configuring disk-watermark thresholds", flush=True)
    must("PUT", "/_cluster/settings",
         json.dumps({"persistent": {
             "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%",
         }}),
         what="disk-watermark settings")

• [ ] Step 2: Verify init.py runs cleanly against a local ES

cd deploy && docker compose up -d elasticsearch
docker compose run --rm elasticsearch-init
docker compose exec elasticsearch python3 -c "
import urllib.request, json
r = urllib.request.urlopen('http://localhost:9200/_cluster/settings')
s = json.loads(r.read())
print(json.dumps(s['persistent'], indent=2))
"

Expected output includes "cluster.routing.allocation.disk.threshold_enabled": "true" and the three watermark percentages.

---

Task 3: Elasticsearch Emergency Retention Sweep

Files: - Modify: deploy/elasticsearch/retention.py (add disk-usage polling, emergency sweep, flood-stage recovery) - Modify: deploy/openshift/logster/values.yaml:110-118 (add threshold config to retention block) - Modify: deploy/openshift/logster/templates/elasticsearch/retention-deployment.yaml:27-38 (pass new env vars) - Modify: deploy/docker-compose.yml:137-143 (add new env vars to elasticsearch-retention service) - Create: deploy/elasticsearch/test_retention.py (tests for the new logic)

Interfaces: - Consumes: ES watermarks from Task 2 (flood stage at 95% that this task auto-recovers from) - Produces: self-healing retention that the dashboard banner (Task 4) can reference in its warning messages

• [ ] Step 1: Add disk-usage query function to retention.py

Add after the es_reachable() function (line 116):

def get_disk_usage():
    """Return (used_bytes, total_bytes, percent) or (0, 0, 0) on failure."""
    status, body = request("GET", "/_cluster/stats", timeout=10)
    if status is None or status >= 400:
        log(f"disk usage check failed: status={status}")
        return 0, 0, 0
    try:
        data = json.loads(body)
        fs = data.get("nodes", {}).get("fs", {})
        total = fs.get("total_in_bytes", 0)
        available = fs.get("available_in_bytes", 0)
        used = total - available
        pct = (used / total * 100) if total > 0 else 0
        return used, total, pct
    except (json.JSONDecodeError, KeyError, ZeroDivisionError):
        log(f"disk usage parse failed: {body[:200]}")
        return 0, 0, 0

• [ ] Step 2: Add flood-stage recovery function

Add after get_disk_usage():

def clear_flood_stage_blocks():
    """Clear read_only_allow_delete blocks set by ES flood-stage watermark."""
    status, body = request("GET", "/_all/_settings", timeout=10)
    if status is None or status >= 400:
        return
    try:
        indices = json.loads(body)
    except json.JSONDecodeError:
        return
    for idx_name, idx_settings in indices.items():
        blocks = idx_settings.get("settings", {}).get("index", {}).get("blocks", {})
        if blocks.get("read_only_allow_delete") == "true":
            log(f"clearing flood-stage read-only block on {idx_name}")
            s, _ = request(
                "PUT", f"/{idx_name}/_settings",
                json.dumps({"index.blocks.read_only_allow_delete": None}),
            )
            if s and s < 400:
                log(f"  {idx_name}: block cleared")
            else:
                log(f"  {idx_name}: failed to clear block (status={s})")

• [ ] Step 3: Add emergency sweep function

Add after clear_flood_stage_blocks():

def emergency_sweep(events_days, inferences_days, graphs_days, strip_raw_days,
                    target_pct):
    """Progressively halve retention windows until disk drops below target_pct.

    Returns True if disk is now below target, False if it couldn't free enough.
    """
    # Start with halved windows
    e, i, g, s = events_days // 2, inferences_days // 2, graphs_days // 2, strip_raw_days // 2
    # Floor at 1 day so we don't disable rules entirely
    e, i, g, s = max(e, 1), max(i, 1), max(g, 1), max(s, 1)

    for attempt in range(4):  # halve up to 4 times (1/2, 1/4, 1/8, 1/16)
        log(f"emergency sweep attempt {attempt + 1}: "
            f"events={e}d inferences={i}d graphs={g}d strip_raw={s}d")

        strip_raw_event(s)
        delete_older_than("logster-events", e)
        delete_older_than("logster-inferences", i)
        delete_older_than("logster-graphs", g)

        # Force-refresh to get accurate disk stats after deletes
        request("POST", "/_refresh", timeout=30)
        # Force merge to reclaim disk — deleted docs don't free space until
        # segments are merged. max_num_segments=1 is aggressive but this is
        # an emergency.
        request("POST", "/_forcemerge?max_num_segments=1", timeout=OP_TIMEOUT)

        _, _, pct = get_disk_usage()
        log(f"  disk now at {pct:.1f}%")
        if pct < target_pct:
            return True

        # Halve again
        e, i, g, s = max(e // 2, 1), max(i // 2, 1), max(g // 2, 1), max(s // 2, 1)

    return False


def critical_sweep(target_pct):
    """Delete oldest data in 1-day chunks until disk drops below target_pct."""
    indices = ["logster-events", "logster-inferences", "logster-graphs"]

    for chunk_days_ago in range(365, 0, -1):  # start from oldest
        for idx in indices:
            older_than = f"now-{chunk_days_ago}d"
            younger_than = f"now-{chunk_days_ago + 1}d"
            status, body = request(
                "POST",
                f"/{idx}/_delete_by_query"
                "?conflicts=proceed&slices=auto"
                "&ignore_unavailable=true",
                json.dumps({"query": {"range": {"@timestamp": {
                    "gte": younger_than, "lt": older_than,
                }}}}),
                timeout=OP_TIMEOUT,
            )
            if status and status < 400:
                deleted = json.loads(body).get("deleted", 0)
                if deleted > 0:
                    log(f"CRITICAL sweep: {idx} deleted {deleted} docs "
                        f"from day -{chunk_days_ago}")

        request("POST", "/_refresh", timeout=30)
        request("POST", "/_forcemerge?max_num_segments=1", timeout=OP_TIMEOUT)

        _, _, pct = get_disk_usage()
        log(f"  disk now at {pct:.1f}%")
        if pct < target_pct:
            return True

    return False

• [ ] Step 4: Rewrite main() to poll disk and trigger sweeps

Replace the main() function entirely:

def main():
    events_days = _env_days("RETENTION_EVENTS_DAYS", 30)
    inferences_days = _env_days("RETENTION_INFERENCES_DAYS", 90)
    graphs_days = _env_days("RETENTION_GRAPHS_DAYS", 14)
    strip_raw_days = _env_days("RETENTION_STRIP_RAW_EVENT_DAYS", 7)
    interval_hours = _env_days("RETENTION_INTERVAL_HOURS", 24)
    check_interval_min = _env_days("RETENTION_CHECK_INTERVAL_MINUTES", 30)
    emergency_pct = _env_days("RETENTION_EMERGENCY_THRESHOLD_PERCENT", 80)
    critical_pct = _env_days("RETENTION_CRITICAL_THRESHOLD_PERCENT", 90)
    target_pct = _env_days("RETENTION_TARGET_PERCENT", 75)

    if interval_hours <= 0:
        log(f"interval={interval_hours}h is not usable — flooring to 1h")
        interval_hours = 1
    if check_interval_min <= 0:
        check_interval_min = 5

    log(f"starting — events={events_days}d inferences={inferences_days}d "
        f"graphs={graphs_days}d strip_raw_event={strip_raw_days}d "
        f"interval={interval_hours}h check_interval={check_interval_min}m "
        f"emergency={emergency_pct}% critical={critical_pct}% "
        f"target={target_pct}%")

    run_once = (os.environ.get("RETENTION_RUN_ONCE") or "").lower() in (
        "1", "true", "yes")

    check_interval_sec = check_interval_min * 60
    retention_interval_sec = interval_hours * 3600
    last_retention_cycle = 0  # force immediate first cycle

    while True:
        # Wait for ES — survives ES restarts without crash-looping.
        for _ in range(60):
            if es_reachable():
                break
            time.sleep(5)
        else:
            log("ES unreachable for 5 minutes — retrying next cycle")
            if run_once:
                return
            time.sleep(check_interval_sec)
            continue

        now = time.monotonic()

        # --- Scheduled retention cycle (every interval_hours) ---
        if now - last_retention_cycle >= retention_interval_sec:
            strip_raw_event(strip_raw_days)
            delete_older_than("logster-events", events_days)
            delete_older_than("logster-inferences", inferences_days)
            delete_older_than("logster-graphs", graphs_days)
            last_retention_cycle = now
            log(f"retention cycle complete — next in {interval_hours}h")

        # --- Disk usage check (every check_interval_min) ---
        _, _, disk_pct = get_disk_usage()
        if disk_pct > 0:
            log(f"disk usage: {disk_pct:.1f}%")

        if disk_pct >= critical_pct:
            log(f"CRITICAL: disk at {disk_pct:.1f}% (>= {critical_pct}%) — "
                f"aggressive sweep to {target_pct}%")
            if not emergency_sweep(events_days, inferences_days, graphs_days,
                                   strip_raw_days, target_pct):
                log("emergency sweep insufficient — falling back to "
                    "critical sweep (delete oldest data)")
                critical_sweep(target_pct)
            clear_flood_stage_blocks()

        elif disk_pct >= emergency_pct:
            log(f"WARNING: disk at {disk_pct:.1f}% (>= {emergency_pct}%) — "
                f"emergency sweep to {target_pct}%")
            emergency_sweep(events_days, inferences_days, graphs_days,
                            strip_raw_days, target_pct)
            clear_flood_stage_blocks()

        if run_once:
            return
        time.sleep(check_interval_sec)

• [ ] Step 5: Add threshold values to Helm values.yaml

In deploy/openshift/logster/values.yaml, extend the retention: block (after line 115):

retention:
  eventsDays: 30
  inferencesDays: 90
  graphsDays: 14
  stripRawEventDays: 7
  intervalHours: 24
  checkIntervalMinutes: 30
  emergencyThresholdPercent: 80
  criticalThresholdPercent: 90
  targetPercent: 75
  resources:
    requests: {cpu: 10m, memory: 32Mi}
    limits: {cpu: 50m, memory: 64Mi}

• [ ] Step 6: Pass new env vars in retention-deployment.yaml

In deploy/openshift/logster/templates/elasticsearch/retention-deployment.yaml, add after the RETENTION_INTERVAL_HOURS env entry (line 38):

            - name: RETENTION_CHECK_INTERVAL_MINUTES
              value: {{ .Values.retention.checkIntervalMinutes | quote }}
            - name: RETENTION_EMERGENCY_THRESHOLD_PERCENT
              value: {{ .Values.retention.emergencyThresholdPercent | quote }}
            - name: RETENTION_CRITICAL_THRESHOLD_PERCENT
              value: {{ .Values.retention.criticalThresholdPercent | quote }}
            - name: RETENTION_TARGET_PERCENT
              value: {{ .Values.retention.targetPercent | quote }}

• [ ] Step 7: Add new env vars to docker-compose.yml elasticsearch-retention

In deploy/docker-compose.yml, add to the elasticsearch-retention: service environment: block (after line 143):

            RETENTION_CHECK_INTERVAL_MINUTES: ${RETENTION_CHECK_INTERVAL_MINUTES:-30}
            RETENTION_EMERGENCY_THRESHOLD_PERCENT: ${RETENTION_EMERGENCY_THRESHOLD_PERCENT:-80}
            RETENTION_CRITICAL_THRESHOLD_PERCENT: ${RETENTION_CRITICAL_THRESHOLD_PERCENT:-90}
            RETENTION_TARGET_PERCENT: ${RETENTION_TARGET_PERCENT:-75}

• [ ] Step 8: Write tests for the new retention logic

Create deploy/elasticsearch/test_retention.py:

"""Tests for retention.py emergency sweep logic.

Run: python3 -m pytest deploy/elasticsearch/test_retention.py -v
(or: cd deploy/elasticsearch && python3 -m pytest test_retention.py -v)
"""
import json
import types
import unittest
from unittest.mock import patch, MagicMock

# retention.py is not a package — import it as a module from its path.
import importlib.util
import os

_here = os.path.dirname(os.path.abspath(__file__))
_spec = importlib.util.spec_from_file_location("retention",
                                                os.path.join(_here, "retention.py"))
retention = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(retention)


class TestGetDiskUsage(unittest.TestCase):
    def test_parses_cluster_stats(self):
        body = json.dumps({
            "nodes": {"fs": {"total_in_bytes": 1000, "available_in_bytes": 300}}
        })
        with patch.object(retention, "request", return_value=(200, body)):
            used, total, pct = retention.get_disk_usage()
        self.assertEqual(used, 700)
        self.assertEqual(total, 1000)
        self.assertAlmostEqual(pct, 70.0)

    def test_returns_zeros_on_failure(self):
        with patch.object(retention, "request", return_value=(None, "connection refused")):
            used, total, pct = retention.get_disk_usage()
        self.assertEqual((used, total, pct), (0, 0, 0))


class TestClearFloodStageBlocks(unittest.TestCase):
    def test_clears_blocked_indices(self):
        settings = {
            "logster-events": {
                "settings": {"index": {"blocks": {"read_only_allow_delete": "true"}}}
            },
            "logster-inferences": {
                "settings": {"index": {}}
            },
        }
        calls = []
        def mock_request(method, path, data=None, timeout=60):
            if method == "GET":
                return 200, json.dumps(settings)
            calls.append((method, path))
            return 200, "{}"

        with patch.object(retention, "request", side_effect=mock_request):
            retention.clear_flood_stage_blocks()

        self.assertEqual(len(calls), 1)
        self.assertIn("logster-events", calls[0][1])


class TestEmergencySweep(unittest.TestCase):
    def test_stops_when_below_target(self):
        disk_calls = [0]
        def mock_disk():
            disk_calls[0] += 1
            # First check: still high; second: below target
            if disk_calls[0] == 1:
                return 0, 0, 82.0
            return 0, 0, 70.0

        with patch.object(retention, "get_disk_usage", side_effect=mock_disk), \
             patch.object(retention, "strip_raw_event"), \
             patch.object(retention, "delete_older_than"), \
             patch.object(retention, "request", return_value=(200, "{}")):
            result = retention.emergency_sweep(30, 90, 14, 7, 75)

        self.assertTrue(result)
        self.assertEqual(disk_calls[0], 1)  # succeeded on first attempt

    def test_halves_progressively(self):
        """Verify retention windows are halved on each attempt."""
        delete_calls = []
        def mock_delete(idx, days):
            delete_calls.append((idx, days))

        def mock_disk():
            return 0, 0, 85.0  # never drops below target

        with patch.object(retention, "get_disk_usage", side_effect=mock_disk), \
             patch.object(retention, "strip_raw_event"), \
             patch.object(retention, "delete_older_than", side_effect=mock_delete), \
             patch.object(retention, "request", return_value=(200, "{}")):
            result = retention.emergency_sweep(30, 90, 14, 7, 75)

        self.assertFalse(result)
        # First attempt: 15, 45, 7 (halved, floored at 1)
        # Find the events calls
        events_days = [d for idx, d in delete_calls if idx == "logster-events"]
        self.assertEqual(events_days[0], 15)  # 30 // 2
        self.assertEqual(events_days[1], 7)   # 15 // 2


if __name__ == "__main__":
    unittest.main()

• [ ] Step 9: Run the tests

cd deploy/elasticsearch
python3 -m pytest test_retention.py -v

Expected: all tests pass.

• [ ] Step 10: Smoke-test against local ES

cd deploy
docker compose build elasticsearch-init
docker compose up -d elasticsearch
# Wait for healthy
docker compose run --rm -e RETENTION_RUN_ONCE=true \
  -e RETENTION_CHECK_INTERVAL_MINUTES=1 \
  elasticsearch-retention

Expected: logs show disk usage: X.X% and retention cycle complete, no errors.

---

Task 4: Dashboard Storage Warning Banner

Files: - Modify: services/dashboard/server/src/config.js (add threshold env vars) - Modify: services/dashboard/server/src/routes/storage.js (add banner state to response) - Create: services/dashboard/client/src/components/common/StorageBanner.jsx - Create: services/dashboard/client/src/components/common/StorageBanner.css - Modify: services/dashboard/client/src/App.jsx:107-109 (render banner)

Interfaces: - Consumes: /api/storage route (already exists) — ES disk stats from fetchElasticsearch() - Produces: visible warning banner on every dashboard page

• [ ] Step 1: Add storage threshold config to config.js

In services/dashboard/server/src/config.js, add after the ATTACK_PROB_THRESHOLD block (around line 49):

const _rawStorageWarn = parseInt(process.env.STORAGE_WARNING_THRESHOLD, 10);
export const STORAGE_WARNING_THRESHOLD =
  Number.isFinite(_rawStorageWarn) && _rawStorageWarn > 0 && _rawStorageWarn <= 100
    ? _rawStorageWarn
    : 70;

const _rawStorageCrit = parseInt(process.env.STORAGE_CRITICAL_THRESHOLD, 10);
export const STORAGE_CRITICAL_THRESHOLD =
  Number.isFinite(_rawStorageCrit) && _rawStorageCrit > 0 && _rawStorageCrit <= 100
    ? _rawStorageCrit
    : 85;

• [ ] Step 2: Add banner state to the storage route response

In services/dashboard/server/src/routes/storage.js, import the thresholds and compute banner state. Add the import at line 4:

import { ENABLE_SYSTEM_HEALTH, KAFKA_BROKERS, STORAGE_WARNING_THRESHOLD, STORAGE_CRITICAL_THRESHOLD } from '../config.js';

Then in the route handler, before cached = { at: ... } (around line 216), compute the banner:

    // Compute banner state for each storage component
    const esPct = elasticsearch.disk_total_bytes > 0
      ? (elasticsearch.disk_used_bytes / elasticsearch.disk_total_bytes) * 100
      : 0;

    let bannerLevel = 'ok';
    let bannerMessage = '';
    const warnings = [];

    if (esPct >= STORAGE_CRITICAL_THRESHOLD) {
      bannerLevel = 'critical';
      warnings.push(`Elasticsearch disk at ${esPct.toFixed(0)}%`);
    } else if (esPct >= STORAGE_WARNING_THRESHOLD) {
      if (bannerLevel !== 'critical') bannerLevel = 'warning';
      warnings.push(`Elasticsearch disk at ${esPct.toFixed(0)}%`);
    }

    if (bannerLevel === 'critical') {
      bannerMessage = `Storage critical — ${warnings.join(', ')}. Data retention may be shortened automatically to prevent data loss.`;
    } else if (bannerLevel === 'warning') {
      bannerMessage = `Storage warning — ${warnings.join(', ')}. Consider expanding storage or reviewing retention settings.`;
    }

    const result = {
      elasticsearch: { reachable: true, ...elasticsearch },
      kafka,
      banner: { level: bannerLevel, message: bannerMessage },
    };

• [ ] Step 3: Create StorageBanner.css

Create services/dashboard/client/src/components/common/StorageBanner.css:

.storage-banner {
  display: flex;
  align-items: center;
  gap: 10px;
  padding: 10px 20px;
  font-size: 0.875rem;
  line-height: 1.4;
  z-index: 100;
}

.storage-banner--warning {
  background: var(--color-warning-bg, #3d2e00);
  color: var(--color-warning-text, #fbbf24);
  border-bottom: 1px solid var(--color-warning-border, #92700a);
}

.storage-banner--critical {
  background: var(--color-critical-bg, #3b0a0a);
  color: var(--color-critical-text, #fca5a5);
  border-bottom: 1px solid var(--color-critical-border, #991b1b);
}

.storage-banner__icon {
  flex-shrink: 0;
}

.storage-banner__message {
  flex: 1;
}

.storage-banner__dismiss {
  flex-shrink: 0;
  background: none;
  border: none;
  color: inherit;
  cursor: pointer;
  padding: 4px;
  opacity: 0.7;
  border-radius: 4px;
}

.storage-banner__dismiss:hover {
  opacity: 1;
  background: rgba(255, 255, 255, 0.1);
}

• [ ] Step 4: Create StorageBanner.jsx component

Create services/dashboard/client/src/components/common/StorageBanner.jsx:

import { useCallback, useEffect, useRef, useState } from 'react';
import { AlertTriangle, X } from 'lucide-react';
import { fetchRaw } from '../../lib/api';
import './StorageBanner.css';

const POLL_INTERVAL_MS = 60_000;

export default function StorageBanner() {
  const [banner, setBanner] = useState(null);
  const [dismissed, setDismissed] = useState(false);
  const timerRef = useRef(null);

  const poll = useCallback(async () => {
    try {
      const res = await fetchRaw('/api/storage');
      if (!res.ok) return;
      const data = await res.json();
      if (data.banner && data.banner.level !== 'ok') {
        setBanner(data.banner);
        // Reset dismiss state if severity escalated to critical
        if (data.banner.level === 'critical') setDismissed(false);
      } else {
        setBanner(null);
        setDismissed(false);
      }
    } catch {
      // Storage endpoint unavailable — don't show a banner for that
    }
  }, []);

  useEffect(() => {
    poll();
    timerRef.current = setInterval(poll, POLL_INTERVAL_MS);
    return () => clearInterval(timerRef.current);
  }, [poll]);

  if (!banner || dismissed) return null;

  const isCritical = banner.level === 'critical';

  return (
    <div
      className={`storage-banner storage-banner--${banner.level}`}
      role="alert"
    >
      <AlertTriangle size={18} className="storage-banner__icon" />
      <span className="storage-banner__message">{banner.message}</span>
      {!isCritical && (
        <button
          className="storage-banner__dismiss"
          onClick={() => setDismissed(true)}
          aria-label="Dismiss"
        >
          <X size={16} />
        </button>
      )}
    </div>
  );
}

• [ ] Step 5: Wire StorageBanner into App.jsx

In services/dashboard/client/src/App.jsx, add the import near the top with the other component imports:

import StorageBanner from './components/common/StorageBanner';

Then insert <StorageBanner /> between <TopBar> and the layout div (between lines 109 and 111):

      <TopBar page={activePage} onPageChange={changePage} />
      <StorageBanner />
      <div className="layout">

• [ ] Step 6: Test the banner locally

cd deploy && docker compose --profile services up -d dashboard

Open http://localhost:5001, navigate to System Health → verify the storage section still renders. The banner should not appear unless ES disk is above 70%.

To force-test the banner, temporarily lower the threshold:

docker compose --profile services up -d -e STORAGE_WARNING_THRESHOLD=1 dashboard

Reload the dashboard — an amber warning banner should appear at the top of every page. Verify: - It appears on the main dashboard view AND the health view - The dismiss button works (warning level only) - The banner reappears on next poll if still above threshold

---

Verification Checklist

After all four tasks are complete:

• [ ] helm template test deploy/openshift/logster/ renders without errors
• [ ] Kafka CR includes log.retention.hours, log.retention.bytes, log.segment.bytes
• [ ] NOTES.txt displays storage warnings and --set-string reminder
• [ ] deploy/elasticsearch/test_retention.py passes
• [ ] retention.py logs disk usage: X.X% on each check cycle
• [ ] init.py sets watermark thresholds (not threshold_enabled: false)
• [ ] Dashboard banner appears when STORAGE_WARNING_THRESHOLD is set below current usage
• [ ] Banner is dismissible at warning level, pinned at critical level
• [ ] Banner appears on all pages (dashboard, health, endpoints, etc.)