From 98ca097fab804043a498c72aaa93f1231f192a53 Mon Sep 17 00:00:00 2001 From: Anton Date: Fri, 28 Nov 2025 17:01:10 +0000 Subject: [PATCH] Decouple notification cadence and harden env parsing --- README.md | 1 + logcost/cli.py | 18 ++++++++-- logcost/notifiers.py | 23 ++++++++----- logcost/tracker.py | 40 ++++++++++++++++++---- logcost/utils.py | 36 ++++++++++++++++++++ sidecar.py | 8 +++-- tests/test_notifiers.py | 75 +++++++++++++++++++++++++++++++++++++++++ 7 files changed, 181 insertions(+), 20 deletions(-) create mode 100644 logcost/utils.py create mode 100644 tests/test_notifiers.py diff --git a/README.md b/README.md index 585c297..b23e83a 100644 --- a/README.md +++ b/README.md @@ -194,6 +194,7 @@ import logcost logcost.start_periodic_flush("/var/log/logcost/stats.json") # Stats flushed every 5 minutes (LOGCOST_FLUSH_INTERVAL=300) # Notifications sent every 1 hour (LOGCOST_NOTIFICATION_INTERVAL=3600, configurable) +# Optional: set LOGCOST_NOTIFICATION_TEST_DELAY (seconds) to send a one-time [Test] message after startup ``` Manual notification: diff --git a/logcost/cli.py b/logcost/cli.py index 4542419..18dd941 100644 --- a/logcost/cli.py +++ b/logcost/cli.py @@ -15,7 +15,17 @@ def _load_stats(path: str) -> dict: - return json.loads(Path(path).read_text()) + try: + text = Path(path).read_text() + except FileNotFoundError as exc: + raise RuntimeError(f"Stats file not found: {path}") from exc + except OSError as exc: + raise RuntimeError(f"Unable to read stats file {path}: {exc}") from exc + + try: + return json.loads(text) + except json.JSONDecodeError as exc: + raise RuntimeError(f"Failed to parse stats file {path}: {exc}") from exc def _cmd_analyze(args: argparse.Namespace) -> int: @@ -209,7 +219,11 @@ def build_parser() -> argparse.ArgumentParser: def main(argv: Optional[Sequence[str]] = None) -> int: parser = build_parser() args = parser.parse_args(argv) - return args.func(args) + try: + return args.func(args) + except RuntimeError as exc: + print(f"Error: {exc}", file=sys.stderr) + return 1 if __name__ == "__main__": diff --git a/logcost/notifiers.py b/logcost/notifiers.py index 5cad2fb..e21344f 100644 --- a/logcost/notifiers.py +++ b/logcost/notifiers.py @@ -11,6 +11,7 @@ from urllib import request, error as urllib_error from .analyzer import CostAnalyzer +from .utils import get_env_int def format_bytes(bytes_count: int) -> str: @@ -34,7 +35,8 @@ def _build_slack_message( stats: Dict, provider: str = "gcp", top_n: int = 5, - previous_stats: Optional[Dict] = None + previous_stats: Optional[Dict] = None, + test_notification: bool = False ) -> Dict: """Build Slack message payload from LogCost stats. @@ -56,7 +58,8 @@ def _build_slack_message( total_calls = sum(entry.count for entry in report.entries) # Build header - header = f"*LogCost Report - {provider.upper()}*\n" + header_prefix = "*[Test]* " if test_notification else "" + header = f"{header_prefix}*LogCost Report - {provider.upper()}*\n" header += f"Total: {format_bytes(total_bytes)} ({format_cost(total_cost)})\n" header += f"Log calls: {total_calls:,}\n" @@ -130,8 +133,9 @@ def _build_slack_message( ] }) + text_prefix = "[Test] " if test_notification else "" return { - "text": f"LogCost Report - {format_cost(total_cost)} total cost", + "text": f"{text_prefix}LogCost Report - {format_cost(total_cost)} total cost", "blocks": blocks } @@ -141,7 +145,8 @@ def send_slack_notification( stats: Dict, provider: str = "gcp", top_n: int = 5, - previous_stats: Optional[Dict] = None + previous_stats: Optional[Dict] = None, + test_notification: bool = False ) -> bool: """Send LogCost report to Slack via webhook. @@ -172,7 +177,7 @@ def send_slack_notification( try: # Build message - message = _build_slack_message(stats, provider, top_n, previous_stats) + message = _build_slack_message(stats, provider, top_n, previous_stats, test_notification) # Send POST request req = request.Request( @@ -196,7 +201,8 @@ def send_slack_notification( def send_notification_if_configured( stats: Dict, provider: Optional[str] = None, - previous_stats: Optional[Dict] = None + previous_stats: Optional[Dict] = None, + test_notification: bool = False ) -> bool: """Send notification if webhook URL is configured via environment variable. @@ -223,12 +229,13 @@ def send_notification_if_configured( if provider is None: provider = os.getenv("LOGCOST_PROVIDER", "gcp") - top_n = int(os.getenv("LOGCOST_NOTIFICATION_TOP_N", "5")) + top_n = get_env_int("LOGCOST_NOTIFICATION_TOP_N", 5) return send_slack_notification( webhook_url, stats, provider=provider, top_n=top_n, - previous_stats=previous_stats + previous_stats=previous_stats, + test_notification=test_notification, ) diff --git a/logcost/tracker.py b/logcost/tracker.py index dfc70be..f239211 100644 --- a/logcost/tracker.py +++ b/logcost/tracker.py @@ -19,6 +19,8 @@ from threading import Lock, Thread, Event from typing import Dict, Optional +from .utils import get_env_int, get_env_int_or_none + PRINT_LEVEL = logging.INFO + 5 logging.addLevelName(PRINT_LEVEL, "PRINT") @@ -45,13 +47,19 @@ def __init__(self): self._thread_local = threading.local() # Periodic flush and rotation configuration - self._flush_interval = int(os.getenv("LOGCOST_FLUSH_INTERVAL", "300")) # 5 minutes default - self._max_file_size = int(os.getenv("LOGCOST_MAX_FILE_SIZE", str(10 * 1024 * 1024))) # 10MB default - self._max_backups = int(os.getenv("LOGCOST_MAX_BACKUPS", "5")) + self._flush_interval = get_env_int("LOGCOST_FLUSH_INTERVAL", 300) # 5 minutes default + self._max_file_size = get_env_int("LOGCOST_MAX_FILE_SIZE", 10 * 1024 * 1024) # 10MB default + self._max_backups = get_env_int("LOGCOST_MAX_BACKUPS", 5) self._flush_thread: Optional[Thread] = None self._flush_stop_event = Event() self._auto_flush_enabled = False self._output_path: Optional[str] = None + self._notification_interval = get_env_int("LOGCOST_NOTIFICATION_INTERVAL", 3600) + self._notification_last_sent = 0.0 + test_delay = get_env_int_or_none("LOGCOST_NOTIFICATION_TEST_DELAY") + self._notification_test_delay = test_delay if test_delay is not None else -1 + self._notification_test_sent = False + self._start_time = time.time() def install(self): """Monkey-patch logging.Logger._log to track calls.""" @@ -286,7 +294,7 @@ def add_skip_module(self, module_prefix: str): return self._skip_module_prefixes.add(module_prefix) - def _send_notification_if_configured(self): + def _send_notification_if_configured(self, test_notification: bool = False): """Send notification if webhook is configured via environment variable.""" try: # Import here to avoid circular dependency @@ -294,7 +302,7 @@ def _send_notification_if_configured(self): stats = self.get_stats() if stats: - send_notification_if_configured(stats) + send_notification_if_configured(stats, test_notification=test_notification) except Exception: # Silently fail to avoid breaking the application pass @@ -332,8 +340,26 @@ def _periodic_flush_worker(self): self._rotate_file(self._output_path) self.export(self._output_path) - # Send notification if configured - self._send_notification_if_configured() + # Notification cadence + now = time.time() + send_notification = False + test_notification = False + + if ( + not self._notification_test_sent + and self._notification_test_delay >= 0 + and now - self._start_time >= self._notification_test_delay + ): + send_notification = True + test_notification = True + self._notification_test_sent = True + + if now - self._notification_last_sent >= self._notification_interval: + send_notification = True + + if send_notification: + self._send_notification_if_configured(test_notification=test_notification) + self._notification_last_sent = now except Exception: # Silently fail to avoid breaking the application pass diff --git a/logcost/utils.py b/logcost/utils.py new file mode 100644 index 0000000..6bcf352 --- /dev/null +++ b/logcost/utils.py @@ -0,0 +1,36 @@ +import os +import warnings +from typing import Optional + + +def get_env_int(name: str, default: int) -> int: + """Return an integer environment variable value with fallback.""" + value = os.getenv(name) + if value is None: + return default + + try: + return int(value) + except (TypeError, ValueError): + warnings.warn( + f"Invalid value for {name}: {value!r}. Using default {default}.", + RuntimeWarning, + stacklevel=2, + ) + return default + + +def get_env_int_or_none(name: str) -> Optional[int]: + """Return integer env value or None if unset/invalid.""" + value = os.getenv(name) + if value is None: + return None + try: + return int(value) + except (TypeError, ValueError): + warnings.warn( + f"Invalid value for {name}: {value!r}. Ignoring.", + RuntimeWarning, + stacklevel=2, + ) + return None diff --git a/sidecar.py b/sidecar.py index 492d91f..9d9cf89 100644 --- a/sidecar.py +++ b/sidecar.py @@ -28,6 +28,8 @@ from pathlib import Path from typing import Dict, Optional +from logcost.utils import get_env_int + # Configure logging logging.basicConfig( level=logging.INFO, @@ -41,12 +43,12 @@ class LogCostSidecar: def __init__(self): self.watch_path = Path(os.getenv("LOGCOST_WATCH_PATH", "/var/log/logcost/stats.json")) - self.notification_interval = int(os.getenv("LOGCOST_NOTIFICATION_INTERVAL", "3600")) + self.notification_interval = get_env_int("LOGCOST_NOTIFICATION_INTERVAL", 3600) self.history_dir = Path(os.getenv("LOGCOST_HISTORY_DIR", "/var/log/logcost/history")) - self.history_retention_days = int(os.getenv("LOGCOST_HISTORY_RETENTION", "7")) + self.history_retention_days = get_env_int("LOGCOST_HISTORY_RETENTION", 7) self.slack_webhook = os.getenv("LOGCOST_SLACK_WEBHOOK") self.provider = os.getenv("LOGCOST_PROVIDER", "gcp") - self.top_n = int(os.getenv("LOGCOST_NOTIFICATION_TOP_N", "5")) + self.top_n = get_env_int("LOGCOST_NOTIFICATION_TOP_N", 5) # Create history directory self.history_dir.mkdir(parents=True, exist_ok=True) diff --git a/tests/test_notifiers.py b/tests/test_notifiers.py new file mode 100644 index 0000000..be8bb2f --- /dev/null +++ b/tests/test_notifiers.py @@ -0,0 +1,75 @@ +import json +from types import SimpleNamespace + +from logcost import notifiers + + +def _sample_stats(): + return { + "app.py:10|INFO": { + "file": "app.py", + "line": 10, + "level": "INFO", + "message_template": "Processed %s", + "count": 5, + "bytes": 500, + } + } + + +def test_send_slack_notification_success(monkeypatch): + captured = {} + + class DummyResponse(SimpleNamespace): + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def fake_urlopen(request, timeout=10): + captured["url"] = request.full_url + captured["data"] = json.loads(request.data.decode("utf-8")) + return DummyResponse(status=200) + + monkeypatch.setattr(notifiers.request, "urlopen", fake_urlopen) + + ok = notifiers.send_slack_notification( + "https://hooks.slack.com/services/test", + _sample_stats(), + provider="gcp", + top_n=1, + ) + + assert ok is True + assert captured["url"] == "https://hooks.slack.com/services/test" + assert "blocks" in captured["data"] + + +def test_send_slack_notification_failure(monkeypatch): + def fake_urlopen(*_, **__): + raise notifiers.urllib_error.URLError("boom") + + monkeypatch.setattr(notifiers.request, "urlopen", fake_urlopen) + ok = notifiers.send_slack_notification("https://hooks.slack.com/services/test", _sample_stats()) + assert ok is False + + +def test_send_notification_if_configured(monkeypatch): + monkeypatch.setenv("LOGCOST_SLACK_WEBHOOK", "https://hooks.slack.com/services/test") + monkeypatch.setenv("LOGCOST_PROVIDER", "aws") + monkeypatch.setenv("LOGCOST_NOTIFICATION_TOP_N", "2") + + called = {} + + def fake_send(*args, **kwargs): + called.update(kwargs) + return True + + monkeypatch.setattr(notifiers, "send_slack_notification", fake_send) + + ok = notifiers.send_notification_if_configured(_sample_stats(), test_notification=True) + assert ok is True + assert called["provider"] == "aws" + assert called["top_n"] == 2 + assert called["test_notification"] is True