Load‑Testing the OpenClaw Rating API WebSocket

a. k6 Script Example

k6 is a modern load‑testing tool that supports WebSocket out of the box. Below is a minimal script that opens 5 000 secure connections, authenticates, and subscribes to the /rating/stream channel.

import ws from 'k6/ws';
import { check, sleep } from 'k6';
import { Counter } from 'k6/metrics';

export const options = {
  stages: [
    { duration: '2m', target: 5000 }, // ramp‑up to 5k connections
    { duration: '5m', target: 5000 }, // hold steady
    { duration: '2m', target: 0 },    // ramp‑down
  ],
  thresholds: {
    'ws_connect_duration': ['p(95)<200'], // 95% of handshakes < 200 ms
    'ws_message_latency': ['p(99)<100'], // 99% of messages < 100 ms
  },
};

const errors = new Counter('ws_errors');

export default function () {
  const url = 'wss://api.openclaw.io/rating';
  const params = { headers: { Authorization: `Bearer ${__ENV.API_TOKEN}` } };

  const response = ws.connect(url, params, function (socket) {
    socket.on('open', function () {
      console.log('WebSocket opened');
      // Subscribe to rating updates
      socket.send(JSON.stringify({ action: 'subscribe', channel: 'rating_updates' }));
    });

    socket.on('message', function (msg) {
      const start = Date.now();
      const data = JSON.parse(msg);
      // Simple latency check
      const latency = Date.now() - start;
      check(latency, { 'latency  l  r && r.status === 101 });
}

This script uses environment variables for the API token, making it safe for CI pipelines. Adjust stages to match your expected peak concurrency.

b. wrk2 Command Line

For raw throughput testing, wrk2 can generate a constant request rate. While it does not natively support WebSocket, you can combine it with websocket‑client scripts. Below is a one‑liner that spawns 2 000 connections and sends a ping every second.

wrk -t12 -c2000 -d5m \
  --latency \
  --script=websocket.lua \
  https://api.openclaw.io/rating

The accompanying websocket.lua (Lua script for wrk2) looks like this:

init = function(args)
  ws = require("ws")
  token = os.getenv("API_TOKEN")
end

request = function()
  local conn = ws.connect("wss://api.openclaw.io/rating", {
    headers = { ["Authorization"] = "Bearer " .. token }
  })
  conn:send('{"action":"subscribe","channel":"rating_updates"}')
  return conn
end

response = function(status, headers, body)
  -- wrk2 only cares about HTTP status; WebSocket frames are handled in the Lua script
  return true
end

Even though wrk2 is primarily HTTP‑focused, this hybrid approach gives you a high‑resolution view of raw network throughput and TLS handshake latency.

c. Custom Python asyncio Script

When you need full control over message payloads, timing, and error handling, a Python asyncio script is the most flexible option. The example below uses websockets and ssl to open 10 000 concurrent connections.

import asyncio
import ssl
import json
import time
import os
import websockets
from collections import Counter

API_URL = "wss://api.openclaw.io/rating"
TOKEN = os.getenv("API_TOKEN")
CONCURRENCY = 10000
DURATION = 300  # seconds

latencies = []
errors = Counter()

ssl_context = ssl.create_default_context()
ssl_context.check_hostname = True
ssl_context.verify_mode = ssl.CERT_REQUIRED

async def worker(id):
    try:
        async with websockets.connect(
            API_URL,
            ssl=ssl_context,
            extra_headers={"Authorization": f"Bearer {TOKEN}"}
        ) as ws:
            # Subscribe
            await ws.send(json.dumps({"action": "subscribe", "channel": "rating_updates"}))
            start = time.time()
            while time.time() - start < DURATION:
                msg = await ws.recv()
                recv_time = time.time()
                # Simple latency measurement (assuming server echoes a timestamp)
                payload = json.loads(msg)
                if "timestamp" in payload:
                    latency = (recv_time - payload["timestamp"]) * 1000  # ms
                    latencies.append(latency)
    except Exception as e:
        errors["exceptions"] += 1
        print(f"Worker {id} error:", e)

async def main():
    tasks = [asyncio.create_task(worker(i)) for i in range(CONCURRENCY)]
    await asyncio.gather(*tasks, return_exceptions=True)

if __name__ == "__main__":
    asyncio.run(main())
    if latencies:
        print(f"Avg latency: {sum(latencies)/len(latencies):.2f} ms")
        print(f"P95 latency: {sorted(latencies)[int(0.95*len(latencies))]:.2f} ms")
    print("Errors:", dict(errors))

This script records per‑message latency, aggregates statistics, and prints a concise summary at the end of the run. Adjust CONCURRENCY and DURATION to match your target load profile.

1️⃣ Handshake Latency

• If the 95th percentile exceeds 200 ms, investigate certificate chain length or CPU‑bound TLS offloading.
• High variance often points to network jitter; consider colocating your load‑generator with the API endpoint.

2️⃣ Message Latency

• Target sub‑100 ms for AI‑agent use‑cases. Anything above 150 ms can degrade model inference timing.
• Plot latency over time; a gradual increase may indicate memory leaks or GC pauses in the server.
• Check for “out‑of‑order” frames – they often reveal back‑pressure in the message broker.

3️⃣ Error Counters

• 1006 (Abnormal Closure) – usually a server‑side timeout; raise the keep‑alive interval.
• 1011 (Internal Error) – indicates unhandled exceptions; review server logs for stack traces.
• Connection‑refused spikes often mean the OS ran out of file descriptors; increase ulimit -n.