debug_env/server/services/traffic.py

"""
TrafficGenerator — sends periodic background requests to the buggy server.

Running as a daemon thread, it hits common endpoints every 2-5 seconds so
that ``app.log`` accumulates realistic-looking access entries that the agent
can query with the ``query_logs`` tool.
"""

import logging
import random
import threading
import urllib.error
import urllib.request

logger = logging.getLogger(__name__)


class TrafficGenerator:
    """Daemon thread that generates synthetic HTTP traffic against a local service."""

    # Endpoints sampled for GET-style traffic
    _GET_ENDPOINTS = ["/health", "/jobs"]

    def __init__(self, port: int) -> None:
        self._port = port
        self._stop_event = threading.Event()
        self._thread: threading.Thread | None = None

    def start(self) -> None:
        """Launch the background traffic daemon thread."""
        self._thread = threading.Thread(target=self._run, daemon=True, name="traffic-gen")
        self._thread.start()
        logger.debug("TrafficGenerator started on port %s", self._port)

    def stop(self) -> None:
        """Signal the daemon thread to exit on its next iteration."""
        self._stop_event.set()

    # ── Internal ───────────────────────────────────────────────────────────────

    def _run(self) -> None:
        while not self._stop_event.is_set():
            endpoint = random.choice(self._GET_ENDPOINTS)
            url = f"http://localhost:{self._port}{endpoint}"
            try:
                with urllib.request.urlopen(url, timeout=2):
                    pass
            except (urllib.error.URLError, OSError):
                pass
            except Exception:
                pass
            # Wait 2–5 s before the next request (or exit early if stopped)
            self._stop_event.wait(random.uniform(2, 5))