Customer.io Load & Scale
Overview
Load testing and scaling strategies for high-volume Customer.io integrations: k6 load test scripts, scaling architecture selection based on volume tier, Kubernetes HPA autoscaling, message queue buffering, and rate-limit-aware batch processing.
Scaling Architecture by Volume
| Daily Events | Architecture | Key Components |
|---|
| < 100K | Direct API | Singleton client, retry, connection pooling |
| 100K - 1M | Batched API | Event queue, batch processor, rate limiter |
| 1M - 10M | Queue-backed | Redis/Kafka queue, worker pool, backpressure |
| > 10M | Distributed | Multiple workspaces, sharded queues, regional routing |
Customer.io rate limit is ~100 req/sec per workspace. Plan your architecture around this.
Instructions
Step 1: k6 Load Test Script
import http from "k6/http";
import { check, sleep } from "k6";
import { Counter, Trend } from "k6/metrics";
const SITE_ID = __ENV.CUSTOMERIO_SITE_ID;
const API_KEY = __ENV.CUSTOMERIO_TRACK_API_KEY;
const BASE_URL = "https://track.customer.io/api/v1";
const AUTH = `${SITE_ID}:${API_KEY}`;
const identifyLatency = new Trend("cio_identify_latency");
const trackLatency = new Trend("cio_track_latency");
const errors = new Counter("cio_errors");
export const options = {
scenarios: {
identify_load: {
executor: "ramping-arrival-rate",
startRate: 10,
timeUnit: "1s",
preAllocatedVUs: 20,
maxVUs: 50,
stages: [
{ duration: "30s", target: 50 },
{ duration: "60s", target: 80 },
{ duration: "30s", target: 10 },
],
},
},
thresholds: {
cio_identify_latency: ["p(95)<500", "p(99)<2000"],
cio_track_latency: ["p(95)<500", "p(99)<2000"],
cio_errors: ["count<50"],
},
};
export default function () {
const userId = `k6-load-${__VU}-${__ITER}`;
const headers = {
"Content-Type": "application/json",
Authorization: `Basic ${encoding.b64encode(AUTH)}`,
};
const identifyRes = http.put(
`${BASE_URL}/customers/${userId}`,
JSON.stringify({
email: `${userId}@loadtest.example.com`,
_load_test: true,
created_at: Math.floor(Date.now() / 1000),
}),
{ headers }
);
identifyLatency.add(identifyRes.timings.duration);
check(identifyRes, { "identify 200": (r) => r.status === 200 }) || errors.add(1);
const trackRes = http.post(
`${BASE_URL}/customers/${userId}/events`,
JSON.stringify({
name: "load_test_event",
data: { iteration: __ITER, vu: __VU },
}),
{ headers }
);
trackLatency.add(trackRes.timings.duration);
check(trackRes, { "track 200": (r) => r.status === 200 }) || errors.add(1);
sleep(0.1);
}
export function teardown() {
console.log("Load test complete. Clean up k6-load-* users in CIO dashboard.");
}
Run:
k6 run --env CUSTOMERIO_SITE_ID="$CUSTOMERIO_SITE_ID" \
--env CUSTOMERIO_TRACK_API_KEY="$CUSTOMERIO_TRACK_API_KEY" \
load-tests/customerio.js
Step 2: Queue-Based Architecture
import { Queue, Worker, QueueEvents } from "bullmq";
import { TrackClient, RegionUS } from "customerio-node";
import Bottleneck from "bottleneck";
const REDIS_URL = process.env.REDIS_URL ?? "redis://localhost:6379";
const limiter = new Bottleneck({
maxConcurrent: 15,
reservoir: 80,
reservoirRefreshAmount: 80,
reservoirRefreshInterval: 1000,
});
const eventQueue = new Queue("cio:events", {
connection: { url: REDIS_URL },
defaultJobOptions: {
attempts: 5,
backoff: { type: "exponential", delay: 2000 },
removeOnComplete: { count: 10000 },
: { : },
},
});
(): <> {
eventQueue.(, { userId, data, : .() });
}
(): {
cio = (
process..!,
process..!,
{ : }
);
worker = (
,
(job) => {
limiter.( () => {
(job. === ) {
cio.(job.., job..);
} {
cio.(job.., job..);
}
});
},
{
: { : },
concurrency,
}
);
worker.(, {
.();
});
events = (, {
: { : },
});
( () => {
counts = eventQueue.();
.(
+
);
}, );
}
Step 3: Kubernetes HPA Autoscaling
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: cio-worker-hpa
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: cio-event-worker
minReplicas: 2
maxReplicas: 20
metrics:
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
- type: Pods
pods:
metric:
name: cio_queue_depth
target:
type: AverageValue
averageValue: "500"
behavior:
scaleUp:
stabilizationWindowSeconds: 60
policies:
- type: Pods
value: 4
periodSeconds:
Step 4: Batch Sender for Bulk Operations
import { TrackClient, RegionUS } from "customerio-node";
import Bottleneck from "bottleneck";
export async function batchSend(
operations: Array<{
type: "identify" | "track";
userId: string;
data: Record<string, any>;
}>,
ratePerSec = 80
): Promise<{ succeeded: number; failed: number }> {
const cio = new TrackClient(
process.env.CUSTOMERIO_SITE_ID!,
process.env.CUSTOMERIO_TRACK_API_KEY!,
{ region: RegionUS }
);
const limiter = new Bottleneck({
maxConcurrent: 15,
reservoir: ratePerSec,
reservoirRefreshAmount: ratePerSec,
reservoirRefreshInterval: 1000,
});
let succeeded = 0;
let failed = 0;
const promises = operations.(
limiter.( () => {
{
(op. === ) {
cio.(op., op.);
} {
cio.(op., op.);
}
succeeded++;
} {
failed++;
}
((succeeded + failed) % === ) {
.();
}
})
);
.(promises);
{ succeeded, failed };
}
Install: npm install bottleneck bullmq
Load Test Checklist
Error Handling
| Issue | Solution |
|---|
| 429 during load test | Reduce rate, check limiter config |
| Queue backlog growing | Scale workers, increase concurrency |
| Memory pressure | Limit batch and queue sizes, enable GC |
| k6 VU exhaustion | Increase preAllocatedVUs and maxVUs |
Resources
Next Steps
After load testing, proceed to customerio-known-pitfalls for anti-patterns to avoid.
