A free endpoint fails in predictable ways. Malformed input. Concurrent bursts. Slow responses. Empty payloads. Rate limits. Each failure has a signature. Each signature is discoverable in one afternoon.
This tutorial builds a chaos probe. It injects five failures against MonkeyCode's free model endpoints, hosted on its free server option. You learn how your client behaves before your users do. Disclosure: This article was prepared as part of MonkeyCode's product outreach.
Why break it on purpose
A 200 status says "I finished". It does not say "I finished correctly". A free endpoint can return 200 with an empty payload. Your client accepts it. Your user sees a blank screen.
Chaos probes find these gaps on purpose. They turn surprise into a checklist. You run the probe once. You fix what breaks. You run it again after every dependency change.
The five failures
| # | Failure | What it simulates |
|---|---|---|
| 1 | Oversized context | A prompt beyond the window |
| 2 | Concurrent burst | Twenty users at once |
| 3 | Short timeout | A slow generation |
| 4 | Invalid model name | A config typo |
| 5 | Quota hammer | A loop that burns the daily limit |
Each failure maps to a real incident. Each one is reproducible in under ten minutes.
Step 1: Build the harness
The harness is one Python file. It stores every probe result in SQLite. It prints a row after each scenario.
import concurrent.futures
import os
import sqlite3
import time
from openai import OpenAI
DB_PATH = "probe.db"
def get_client():
return OpenAI(
base_url=os.environ["MODEL_BASE_URL"],
api_key=os.environ["MODEL_API_KEY"],
)
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.executescript("""
CREATE TABLE IF NOT EXISTS probe_runs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
scenario TEXT NOT NULL,
status INTEGER NOT NULL,
latency_ms INTEGER NOT NULL,
payload_ok INTEGER NOT NULL,
error_type TEXT,
created_at TEXT NOT NULL
);
""")
conn.commit()
return conn
def record(conn, scenario, status, latency_ms, payload_ok, error_type=None):
conn.execute(
"INSERT INTO probe_runs (scenario, status, latency_ms, payload_ok, error_type)"
" VALUES (?, ?, ?, ?, ?)",
(scenario, status, latency_ms, payload_ok, error_type),
)
conn.commit()
def one_call(client, model, messages, timeout=30):
start = time.perf_counter()
status = 0
payload_ok = 0
error_type = None
try:
resp = client.chat.completions.create(
model=model, messages=messages, timeout=timeout
)
status = 200
content = resp.choices[0].message.content or ""
payload_ok = 1 if content.strip() else 0
except Exception as exc:
error_type = type(exc).__name__
status = getattr(exc, "status_code", 0)
latency_ms = int((time.perf_counter() - start) * 1000)
return status, latency_ms, payload_ok, error_type
Set the endpoint values first:
export MODEL_BASE_URL="https://your-endpoint.example"
export MODEL_API_KEY="your-key"
Verify the harness before any scenario:
python3 -c "
import probe
conn = probe.init_db()
print(conn.execute('SELECT COUNT(*) FROM probe_runs').fetchone()[0])
"
Expect 0. The database exists. The table is empty.
Step 2: Inject input failures
Two failures come from bad input. An oversized prompt. A wrong model name. Both are config errors waiting to happen.
def scenario_oversized(conn, client, model):
messages = [{"role": "user", "content": "word " * 50000}]
status, lat, ok, err = one_call(client, model, messages)
record(conn, "oversized", status, lat, ok, err)
print(f"oversized status={status} latency={lat}ms payload_ok={ok} error={err}")
def scenario_bad_model(conn, client):
messages = [{"role": "user", "content": "Hi."}]
status, lat, ok, err = one_call(client, "model-does-not-exist", messages)
record(conn, "bad_model", status, lat, ok, err)
print(f"bad_model status={status} latency={lat}ms payload_ok={ok} error={err}")
Run both:
python3 -c "
import probe
conn = probe.init_db()
client = probe.get_client()
probe.scenario_oversized(conn, client, 'your-model')
probe.scenario_bad_model(conn, client)
"
Verify two rows exist:
sqlite3 probe.db "SELECT scenario, status, payload_ok FROM probe_runs;"
An oversized prompt should return a 4xx. A bad model should return a 4xx too. If either returns 200, the endpoint is hiding errors. That is a finding.
Step 3: Inject load failures
Two failures come from load. A concurrent burst. A sequential hammer. Both reveal rate limits.
def scenario_burst(conn, client, model):
messages = [{"role": "user", "content": "Say 'ok'."}]
def fire(_):
return one_call(client, model, messages, timeout=60)
with concurrent.futures.ThreadPoolExecutor(max_workers=20) as ex:
results = list(ex.map(fire, range(20)))
for status, lat, ok, err in results:
record(conn, "burst", status, lat, ok, err)
ok_count = sum(1 for r in results if r[0] == 200)
latencies = sorted(r[1] for r in results)
print(f"burst 20 calls, {ok_count} ok, p50={latencies[10]}ms")
def scenario_hammer(conn, client, model, limit=30):
messages = [{"role": "user", "content": "Reply with one word."}]
for i in range(limit):
status, lat, ok, err = one_call(client, model, messages)
record(conn, "hammer", status, lat, ok, err)
if status == 429:
print(f"hammer first 429 at attempt {i + 1}")
return
print(f"hammer no 429 after {limit} attempts")
Run both:
python3 -c "
import probe
conn = probe.init_db()
client = probe.get_client()
probe.scenario_burst(conn, client, 'your-model')
probe.scenario_hammer(conn, client, 'your-model')
"
Verify the output. The burst tells you the concurrency ceiling. The hammer tells you the quota shape. A 429 is a fact, not a suggestion.
Step 4: Inject the timeout failure
The slow generation is the hardest failure to catch. The endpoint is healthy. The response just takes too long.
def scenario_short_timeout(conn, client, model):
messages = [{"role": "user", "content": "Write a very long story about a whale."}]
status, lat, ok, err = one_call(client, model, messages, timeout=2)
record(conn, "short_timeout", status, lat, ok, err)
print(f"short_timeout status={status} latency={lat}ms payload_ok={ok} error={err}")
Run it:
python3 -c "
import probe
conn = probe.init_db()
client = probe.get_client()
probe.scenario_short_timeout(conn, client, 'your-model')
"
Expect a timeout exception. The status should be 0. The error type should be Timeout or APITimeoutError. If the call completes in two seconds, your prompt is too easy. Use a longer generation request.
Step 5: Read the verdict
One query turns five scenarios into a report.
SELECT scenario,
COUNT(*) AS runs,
SUM(CASE WHEN status = 200 THEN 1 ELSE 0 END) AS ok,
SUM(CASE WHEN payload_ok = 0 THEN 1 ELSE 0 END) AS empty_payloads,
CAST(AVG(latency_ms) AS INT) AS avg_latency_ms
FROM probe_runs
GROUP BY scenario;
Run it:
sqlite3 probe.db "SELECT scenario, COUNT(*) AS runs, SUM(CASE WHEN status = 200 THEN 1 ELSE 0 END) AS ok, SUM(CASE WHEN payload_ok = 0 THEN 1 ELSE 0 END) AS empty_payloads, CAST(AVG(latency_ms) AS INT) AS avg_latency_ms FROM probe_runs GROUP BY scenario;"
The table tells you which failures are real for your setup.
Step 6: Apply the fixes
Each finding maps to one fix.
| Finding | Fix |
|---|---|
| 200 with empty payload | Validate content before use |
| 429 on burst | Add a client-side semaphore |
| Timeout on long output | Stream the response |
| 4xx on bad model | Pin model names in a config file |
| 429 after N calls | Track daily usage |
Apply the fix. Rerun the scenario. The probe doubles as a regression test.
Who should skip this probe
This probe is a diagnosis, not a cure. It burns quota on purpose. Run it when you can afford the calls.
Skip it if you have a paid SLA. Use a real chaos tool like Chaos Mesh instead. Skip it if you cannot tolerate any dropped request. Fix the root cause first.
Start with the five failures anyway. A free endpoint is a black box until you break it. Break it on purpose. Record what happens. Fix the gaps. Then break it again.
If you build a probe like this, share your failure table. The sixth failure is usually the one you did not inject.
Top comments (0)