Alerting

from pathlib import Path
import sys

Path("ops.py").write_text('''import json
import os
import subprocess
import sys
import time
import urllib.error
import urllib.request
from contextlib import contextmanager

KEY = "sk-gateway-1234"          # the master key written in every config.yaml


@contextmanager
def running(port, env=None):
    """Start the gateway on `port` from config.yaml, wait until it answers, stop it afterwards."""
    litellm = os.path.join(os.path.dirname(sys.executable), "litellm")
    cmd = [litellm, "--config", "config.yaml", "--port", str(port)]
    proc = subprocess.Popen(cmd, env=env, stdout=open("gateway.log", "w"), stderr=subprocess.STDOUT)
    try:
        for _ in range(90):
            if proc.poll() is not None:
                raise RuntimeError("the gateway stopped; read gateway.log")
            try:
                urllib.request.urlopen(f"http://127.0.0.1:{port}/health/liveliness")
                break
            except OSError:
                time.sleep(1)
        yield f"http://127.0.0.1:{port}"
    finally:
        proc.terminate()
        proc.wait(10)


def call(url, path, body=None, key=KEY, headers=None):
    """Send one request. Returns (status, text, headers) and does not raise on 4xx or 5xx."""
    head = dict(headers or {})
    if key:
        head["Authorization"] = f"Bearer {key}"
    if body is not None:
        head["Content-Type"] = "application/json"
    data = json.dumps(body).encode() if body is not None else None
    try:
        reply = urllib.request.urlopen(urllib.request.Request(url + path, data=data, headers=head), timeout=60)
    except urllib.error.HTTPError as error:
        reply = error
    return reply.getcode(), reply.read().decode(), {k.lower(): v for k, v in reply.headers.items()}


def chat(url, model, text="Reply with one word: ok", key=KEY, headers=None, **extra):
    """One chat request. Extra keyword arguments are added to the JSON body."""
    body = {"model": model, "messages": [{"role": "user", "content": text}], "max_tokens": 60, **extra}
    return call(url, "/chat/completions", body, key, headers)
''')
sys.path.insert(0, ".")        # lets the programs below run: from ops import running, call, chat
print("wrote ops.py")

wrote ops.py

import json
import os
import re
import threading
import time
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from ops import running, call, chat

received = []

class Hook(BaseHTTPRequestHandler):                      # stands in for Slack
    def do_POST(self):
        size = int(self.headers["Content-Length"])
        received.append(json.loads(self.rfile.read(size))["text"])
        self.send_response(200)
        self.end_headers()
    def log_message(self, *args):                        # keep the output clean
        pass

server = HTTPServer(("127.0.0.1", 4729), Hook)
threading.Thread(target=server.serve_forever, daemon=True).start()

TEMPLATE = """model_list:
  - model_name: fast
    litellm_params:
      model: groq/openai/gpt-oss-20b
      api_key: os.environ/GROQ_API_KEY
  - model_name: broken
    litellm_params:
      model: groq/openai/gpt-oss-120b
      api_key: gsk_not_a_real_key
general_settings:
  master_key: sk-gateway-1234
  alerting: ["slack"]
  alerting_threshold: 0.0001
{extra}"""
RUNS = [
    ("every alert type", ""),
    ("llm_exceptions only, prompt hidden",
     "  alert_types: [\"llm_exceptions\"]\nlitellm_settings:\n  redact_messages_in_exceptions: true\n"),
]
env = {**os.environ, "SLACK_WEBHOOK_URL": "http://127.0.0.1:4729/hook"}

def kinds():
    return sorted(re.search(r"Alert type: `(\w+)`", m).group(1) for m in received)

try:
    for label, extra in RUNS:
        received.clear()
        Path("config.yaml").write_text(TEMPLATE.format(extra=extra))
        with running(4721, env=env) as url:
            print(label)
            print("  test alert:", call(url, "/health/services?service=slack")[0])
            print("  requests:", chat(url, "broken")[0], chat(url, "fast")[0])
            for _ in range(40):                          # alerts leave the gateway in batches
                if "llm_exceptions" in kinds():
                    break
                time.sleep(1)
            time.sleep(12)                               # room for the other alerts to arrive
        print("  alerts:", kinds())
        failure = next(m for m in received if "llm_exceptions" in m)
        print("  failure alert shows the prompt:", "Messages:" in failure)
finally:
    server.shutdown()
print()
print("\n".join(l for l in failure.splitlines() if not l.startswith("Timestamp")))

every alert type
  test alert: 200
  requests: 401 200
  alerts: ['budget_alerts', 'llm_exceptions', 'llm_requests_hanging', 'llm_requests_hanging', 'llm_too_slow']
  failure alert shows the prompt: True
llm_exceptions only, prompt hidden
  test alert: 200
  requests: 401 200
  alerts: ['llm_exceptions']
  failure alert shows the prompt: False

Alert type: `llm_exceptions`
Level: `High`

Message: LLM API call failed: `litellm.BadRequestError: GroqException - {"error":{"message":"Invalid API Key","type":"invalid_request_error","code":"invalid_api_key"}}


LiteLLM: model group 'broken' failed with the error above. No fallback was attempted.
Model: groq/openai/gpt-oss-120b
API Base: `https://api.groq.com/openai/v1`
model_group: `broken`

deployment: `groq/openai/gpt-oss-120b`
`