Docs / Connect your data
Your app's code
Count sign-ups, orders or any event in your Node or Python code, and get an alert when it changes.
In short: paste a small counter into your Node or Python code. It sends its counts once an hour, with a zero for every quiet 5 minutes, so if your process stops you get a "stopped" alert.
The detailed steps below work for you or your coding agent. Your agent can do every step that needs no secret; anything secret, such as a webhook URL or a key, you make in the dashboard and paste yourself.
Counts events where they happen (orders placed, e-mails sent, jobs done, logins failed) with a small counter in the process. Once an hour it sends every finished 5-minute bucket, zeros included, as one request: the watch is an explicit source, so if the app stops sending, the no-data alert fires. A failed send keeps its buckets for the next one (up to 6 hours). Set COUNTWATCH_API_KEY (and COUNTWATCH_API_URL=https://staging.countwatch.com) in the app's environment.
Count outcomes, not attempts: count("orders") after the order is saved, not when checkout starts. The calls are those of the client libraries (count(metric, n, dimensions)), so moving to npm install countwatch or pip install countwatch later changes only the import.
Node (18+, no packages)
Save as countwatch-counter.mjs:
// countwatch-counter.mjs — CountWatch counter (Node 18+, no packages). Same calls as the countwatch npm package.
// const cw = countwatch(); cw.count("orders", 1, { shop: "eu" }); … on shutdown: await cw.flush(); cw.stop();
const BUCKET = 300_000;
const MAX_BACK = 6 * 3_600_000 - BUCKET;
export function countwatch({ token = process.env.COUNTWATCH_API_KEY, api = process.env.COUNTWATCH_API_URL ?? "https://api.countwatch.com", everyMs = 3_600_000 } = {}) {
const watches = new Map(); // key -> { metric, dimensions, from, counts: Map<bucket start, n> }
const bucketOf = (t) => Math.floor(t / BUCKET) * BUCKET;
function count(metric, n = 1, dimensions = {}) {
const key = metric + "|" + JSON.stringify(Object.entries(dimensions).sort());
const b = bucketOf(Date.now());
let w = watches.get(key);
if (!w) watches.set(key, (w = { metric, dimensions, from: b, counts: new Map() }));
w.counts.set(b, (w.counts.get(b) ?? 0) + n);
}
let inFlight = null; // one flush at a time: a second call waits for the running one
function flush() {
inFlight ??= send().finally(() => (inFlight = null));
return inFlight;
}
async function send() {
const open = bucketOf(Date.now()); // the bucket still filling is sent next time
const points = [];
for (const w of watches.values()) {
w.from = Math.max(w.from, open - MAX_BACK); // older buckets are no longer accepted: forget them
for (const t of w.counts.keys()) if (t < w.from) w.counts.delete(t);
for (let t = w.from; t < open; t += BUCKET) {
points.push({ metric: w.metric, dimensions: w.dimensions, ts: new Date(t).toISOString(), set: w.counts.get(t) ?? 0 });
}
}
for (let i = 0; i < points.length; i += 10_000) {
const res = await fetch(api + "/v1/ingest", {
method: "POST",
headers: { authorization: "Bearer " + token, "content-type": "application/json" },
body: JSON.stringify({ points: points.slice(i, i + 10_000) }),
signal: AbortSignal.timeout(10_000),
});
if (!res.ok) throw new Error("CountWatch " + res.status + ": " + (await res.text()).slice(0, 200));
const answer = await res.json().catch(() => ({}));
// Refused points (a watch limit, a bad name) are not retried: say so.
if (answer.rejected?.length) console.error("countwatch: points refused:", JSON.stringify(answer.rejected).slice(0, 300));
}
for (const w of watches.values()) {
w.from = Math.max(w.from, open);
for (const t of w.counts.keys()) if (t < open) w.counts.delete(t);
}
}
// The first send waits a random part of the interval, so processes started together (a deploy) send at different times.
const tick = () => flush().catch((e) => console.error("countwatch:", e.message));
let timer = setTimeout(() => {
tick();
timer = setInterval(tick, everyMs);
timer.unref?.();
}, Math.random() * everyMs);
timer.unref?.();
return { count, flush, stop: () => clearTimeout(timer) }; // clears the first timeout or the interval
}
import { countwatch } from "./countwatch-counter.mjs";
const cw = countwatch();
cw.count("orders", 0, { shop: "eu" }); // declare each watch at startup: zeros go out even before the first event
// … where the event happens:
cw.count("orders", 1, { shop: "eu" });
// on shutdown:
process.on("SIGTERM", async () => { await cw.flush().catch(() => {}); cw.stop(); process.exit(0); });
Python (3.8+, standard library)
Save as countwatch_counter.py (not countwatch.py: that name would hide the countwatch package):
# countwatch_counter.py — CountWatch counter (Python 3.8+, standard library only). Same calls as the countwatch
# PyPI package. cw = CountWatch(); cw.count("orders", 1, {"shop": "eu"}); … on shutdown: cw.flush(); cw.stop()
import json, os, random, threading, time, urllib.request
from datetime import datetime, timezone
BUCKET = 300
MAX_BACK = 6 * 3600 - BUCKET
class CountWatch:
def __init__(self, token=None, api=None, every_s=3600):
self.token = token or os.environ["COUNTWATCH_API_KEY"]
self.api = api or os.environ.get("COUNTWATCH_API_URL", "https://api.countwatch.com")
self.every_s = every_s
self.watches = {} # key -> {"metric", "dimensions", "from", "counts": {bucket start: n}}
self.lock = threading.Lock()
self.flushing = threading.Lock() # one flush at a time
self.timer = None
self.stopped = False
# The first send waits a random part of the interval, so processes started together (a deploy) send at different times.
self._schedule(random.uniform(0, every_s))
def count(self, metric, n=1, dimensions=None):
dimensions = dimensions or {}
key = metric + "|" + json.dumps(sorted(dimensions.items()))
b = int(time.time()) // BUCKET * BUCKET
with self.lock:
w = self.watches.setdefault(key, {"metric": metric, "dimensions": dimensions, "from": b, "counts": {}})
w["counts"][b] = w["counts"].get(b, 0) + n
def flush(self):
with self.flushing:
self._send()
def _send(self):
open_b = int(time.time()) // BUCKET * BUCKET # the bucket still filling is sent next time
with self.lock:
points = []
for w in self.watches.values():
w["from"] = max(w["from"], open_b - MAX_BACK) # older buckets are no longer accepted: forget them
w["counts"] = {t: n for t, n in w["counts"].items() if t >= w["from"]}
for t in range(w["from"], open_b, BUCKET):
ts = datetime.fromtimestamp(t, timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
points.append({"metric": w["metric"], "dimensions": w["dimensions"], "ts": ts, "set": w["counts"].get(t, 0)})
for i in range(0, len(points), 10000):
req = urllib.request.Request(
self.api + "/v1/ingest",
data=json.dumps({"points": points[i : i + 10000]}).encode(),
headers={"authorization": "Bearer " + self.token, "content-type": "application/json"},
method="POST",
)
answer = json.loads(urllib.request.urlopen(req, timeout=10).read() or b"{}") # raises on a non-2xx answer: the buckets stay
if answer.get("rejected"): # refused points (a watch limit, a bad name) are not retried: say so
print("countwatch: points refused:", json.dumps(answer["rejected"])[:300])
with self.lock:
for w in self.watches.values():
w["from"] = max(w["from"], open_b)
w["counts"] = {t: n for t, n in w["counts"].items() if t >= open_b}
def _schedule(self, delay=None):
if self.stopped:
return
self.timer = threading.Timer(self.every_s if delay is None else delay, self._tick)
self.timer.daemon = True
self.timer.start()
def _tick(self):
try:
self.flush()
except Exception as e: # noqa: BLE001 — a failed send is retried next time
print("countwatch:", e)
self._schedule()
def stop(self):
self.stopped = True
if self.timer:
self.timer.cancel()
from countwatch_counter import CountWatch
import atexit
cw = CountWatch()
cw.count("orders", 0, {"shop": "eu"}) # declare each watch at startup: zeros go out even before the first event
atexit.register(cw.flush) # send the finished buckets on a normal exit
# … where the event happens:
cw.count("orders", 1, {"shop": "eu"})
Several processes
Each point sets its bucket, so two processes sending the same watch overwrite each other. Give each process its own dimension ({"instance": "web-1"}; mind the plan's watch limit), or count in one place.
History
Count the same events per 5 minutes from your database or logs for the last 89 days, every bucket (an empty one is a 0), with UTC bucket starts, and send them once with POST /v1/import (resolution: "5m", the same metric and dimensions). The API stops an import at a watch's first live bucket, so it never overwrites the counter's sends. Use "1h" or "1d" only when the source keeps nothing finer: coarser counts are spread evenly over their 5-minute buckets.
Check it works
- Data arrives.
check_setup(orcurl -s https://staging.countwatch.com/v1/status -H "Authorization: Bearer $COUNTWATCH_API_KEY"): theorderswatch showslast_request_at, andmissing_buckets_24hstays0(explicit watches). Do whatnext_actionssays. - History is in.
GET /v1/imports/{import_id}showsstatus: doneand the rules chosen fororders, with the alerts they would have raised. (Skipped the import? The watch learns for about 3 weeks; starter rules apply meanwhile.) - An alert reaches you. The person turns on e-mail alerts in the dashboard (https://staging.countwatch.com/dashboard), or you add a webhook (
manage add_destination, thenverify_destination; the endpoint answers the challenge, see https://staging.countwatch.com/agent). Thenmanage {"action": "send_test_alert"}: every active destination receives a test alert (test: true) within a few minutes.
For agents: this page as Markdown.