Decouple notification cadence and harden env parsing

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:

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__":

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,
     )

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,15 +294,15 @@ 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
             from .notifiers import send_notification_if_configured
 
             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

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

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)

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