refactor(telemetry): sample OSS hot-path events at 10% to reduce PostHog volume (#4771)

This commit is contained in:
Gabriel Stein
2026-04-11 16:25:38 +05:30
committed by GitHub
parent 789cc9d607
commit 9d82e2329d
6 changed files with 501 additions and 4 deletions
+27
View File
@@ -14,6 +14,25 @@ try {
const POSTHOG_API_KEY = "phc_hgJkUVJFYtmaJqrvf6CYN67TIQ8yhXAkWzUn9AMU4yX";
const POSTHOG_HOST = "https://us.i.posthog.com/i/v0/e/";
// Default sampling rate for hot-path OSS events. Lifecycle events always fire at 100%.
// Override via MEM0_TELEMETRY_SAMPLE_RATE env var. Mirrors mem0/memory/telemetry.py.
const DEFAULT_SAMPLE_RATE = 0.1;
const MEM0_TELEMETRY_SAMPLE_RATE: number = ((): number => {
try {
const raw = process?.env?.MEM0_TELEMETRY_SAMPLE_RATE;
if (raw !== undefined) {
const parsed = Number(raw);
if (Number.isFinite(parsed) && parsed >= 0 && parsed <= 1) {
return parsed;
}
}
} catch {}
return DEFAULT_SAMPLE_RATE;
})();
// Events that bypass sampling. Keep in sync with _captureEvent call sites in memory/index.ts.
const LIFECYCLE_EVENTS: ReadonlySet<string> = new Set(["init", "reset"]);
class UnifiedTelemetry implements TelemetryClient {
private apiKey: string;
private host: string;
@@ -78,6 +97,12 @@ async function captureClientEvent(
return;
}
// >= so that rate=0 drops everything and rate=1 keeps everything (Math.random() ∈ [0, 1)).
const isLifecycle = LIFECYCLE_EVENTS.has(eventName);
if (!isLifecycle && Math.random() >= MEM0_TELEMETRY_SAMPLE_RATE) {
return;
}
const eventData: TelemetryEventData = {
function: `${instance.constructor.name}`,
method: eventName,
@@ -86,6 +111,8 @@ async function captureClientEvent(
client_version: version,
client_source: "nodejs",
...additionalData,
// sample_rate set AFTER the spread so callers can never override it
sample_rate: isLifecycle ? 1.0 : MEM0_TELEMETRY_SAMPLE_RATE,
};
await telemetry.captureEvent(
@@ -23,6 +23,8 @@ export interface TelemetryEventData {
timestamp?: string;
client_source: "browser" | "nodejs";
client_version: string;
/** Set by the sampling layer so PostHog dashboards can extrapolate via 1/sample_rate. */
sample_rate?: number;
[key: string]: any;
}
@@ -0,0 +1,211 @@
/// <reference types="jest" />
/**
* Telemetry sampling — unit tests.
*
* Sampling lives inside captureClientEvent in src/utils/telemetry.ts. It uses
* Math.random() compared against MEM0_TELEMETRY_SAMPLE_RATE (default 0.1) to
* drop hot-path events; lifecycle events ('init', 'reset') always fire.
*
* These tests target captureClientEvent directly. Existing OSS tests that mock
* captureClientEvent (vector-stores-compat, dimension-autodetect, config-manager)
* are unaffected because the function signature and contract are unchanged.
*/
import type { TelemetryInstance } from "../src/utils/telemetry.types";
// Helper to make a fake TelemetryInstance.
function makeInstance(
overrides: Partial<TelemetryInstance> = {},
): TelemetryInstance {
return {
telemetryId: "test-id",
constructor: { name: "Memory" },
host: "https://test.example.com",
...overrides,
};
}
describe("telemetry sampling", () => {
let originalFetch: typeof global.fetch;
let fetchMock: jest.Mock;
let randomSpy: jest.SpyInstance;
beforeEach(() => {
originalFetch = global.fetch;
fetchMock = jest.fn().mockResolvedValue({
ok: true,
text: jest.fn().mockResolvedValue(""),
});
global.fetch = fetchMock as any;
randomSpy = jest.spyOn(Math, "random");
});
afterEach(() => {
global.fetch = originalFetch;
randomSpy.mockRestore();
jest.resetModules();
});
describe("lifecycle events always fire", () => {
it("init event fires even at the highest random value", async () => {
randomSpy.mockReturnValue(0.999);
const { captureClientEvent } = await import("../src/utils/telemetry");
await captureClientEvent("init", makeInstance());
expect(fetchMock).toHaveBeenCalledTimes(1);
});
it("reset event fires even at the highest random value", async () => {
randomSpy.mockReturnValue(0.999);
const { captureClientEvent } = await import("../src/utils/telemetry");
await captureClientEvent("reset", makeInstance());
expect(fetchMock).toHaveBeenCalledTimes(1);
});
it("init event payload has sample_rate: 1.0", async () => {
randomSpy.mockReturnValue(0.999);
const { captureClientEvent } = await import("../src/utils/telemetry");
await captureClientEvent("init", makeInstance());
const body = JSON.parse(fetchMock.mock.calls[0][1].body);
expect(body.properties.sample_rate).toBe(1.0);
});
});
describe("hot-path events are sampled", () => {
it("add event is dropped when random > sample_rate", async () => {
randomSpy.mockReturnValue(0.99); // > 0.1 default rate
const { captureClientEvent } = await import("../src/utils/telemetry");
await captureClientEvent("add", makeInstance());
expect(fetchMock).not.toHaveBeenCalled();
});
it("search event is dropped when random > sample_rate", async () => {
randomSpy.mockReturnValue(0.99);
const { captureClientEvent } = await import("../src/utils/telemetry");
await captureClientEvent("search", makeInstance());
expect(fetchMock).not.toHaveBeenCalled();
});
it("add event passes when random < sample_rate", async () => {
randomSpy.mockReturnValue(0.05); // < 0.1 default rate
const { captureClientEvent } = await import("../src/utils/telemetry");
await captureClientEvent("add", makeInstance());
expect(fetchMock).toHaveBeenCalledTimes(1);
});
it("hot-path event payload has sample_rate: 0.1", async () => {
randomSpy.mockReturnValue(0.05);
const { captureClientEvent } = await import("../src/utils/telemetry");
await captureClientEvent("add", makeInstance());
const body = JSON.parse(fetchMock.mock.calls[0][1].body);
expect(body.properties.sample_rate).toBe(0.1);
});
it("sample_rate cannot be overridden by additionalData", async () => {
randomSpy.mockReturnValue(0.05);
const { captureClientEvent } = await import("../src/utils/telemetry");
await captureClientEvent("add", makeInstance(), { sample_rate: 0.99 });
const body = JSON.parse(fetchMock.mock.calls[0][1].body);
expect(body.properties.sample_rate).toBe(0.1);
});
});
describe("env var override", () => {
afterEach(() => {
delete process.env.MEM0_TELEMETRY_SAMPLE_RATE;
});
it("rate 1.0 sends every event including hot-path at high random", async () => {
process.env.MEM0_TELEMETRY_SAMPLE_RATE = "1.0";
jest.resetModules();
randomSpy.mockReturnValue(0.999);
const { captureClientEvent } = await import("../src/utils/telemetry");
await captureClientEvent("add", makeInstance());
// 0.999 > 1.0 is false, so the gate never trips
expect(fetchMock).toHaveBeenCalledTimes(1);
});
it("rate 0.0 drops every hot-path event including at random 0", async () => {
// The gate is `random >= rate`, so rate=0 drops at random=0 (0 >= 0 is true).
process.env.MEM0_TELEMETRY_SAMPLE_RATE = "0.0";
jest.resetModules();
randomSpy.mockReturnValue(0.0);
const { captureClientEvent } = await import("../src/utils/telemetry");
await captureClientEvent("add", makeInstance());
expect(fetchMock).not.toHaveBeenCalled();
});
it("rate 0.0 drops at any random value", async () => {
process.env.MEM0_TELEMETRY_SAMPLE_RATE = "0.0";
jest.resetModules();
randomSpy.mockReturnValue(0.5);
const { captureClientEvent } = await import("../src/utils/telemetry");
await captureClientEvent("add", makeInstance());
expect(fetchMock).not.toHaveBeenCalled();
});
it("rate 0.0 still passes lifecycle events", async () => {
process.env.MEM0_TELEMETRY_SAMPLE_RATE = "0.0";
jest.resetModules();
randomSpy.mockReturnValue(0.999);
const { captureClientEvent } = await import("../src/utils/telemetry");
await captureClientEvent("init", makeInstance());
expect(fetchMock).toHaveBeenCalledTimes(1);
});
it("rate 0.5 passes events at random below 0.5", async () => {
process.env.MEM0_TELEMETRY_SAMPLE_RATE = "0.5";
jest.resetModules();
randomSpy.mockReturnValue(0.3);
const { captureClientEvent } = await import("../src/utils/telemetry");
await captureClientEvent("add", makeInstance());
expect(fetchMock).toHaveBeenCalledTimes(1);
});
it("invalid env var falls back to default rate", async () => {
process.env.MEM0_TELEMETRY_SAMPLE_RATE = "not a number";
jest.resetModules();
randomSpy.mockReturnValue(0.05);
const { captureClientEvent } = await import("../src/utils/telemetry");
await captureClientEvent("add", makeInstance());
expect(fetchMock).toHaveBeenCalledTimes(1);
});
it("out-of-range env var (negative) falls back to default rate", async () => {
process.env.MEM0_TELEMETRY_SAMPLE_RATE = "-0.5";
jest.resetModules();
randomSpy.mockReturnValue(0.05);
const { captureClientEvent } = await import("../src/utils/telemetry");
await captureClientEvent("add", makeInstance());
expect(fetchMock).toHaveBeenCalledTimes(1);
});
it("out-of-range env var (> 1) falls back to default rate", async () => {
process.env.MEM0_TELEMETRY_SAMPLE_RATE = "5";
jest.resetModules();
randomSpy.mockReturnValue(0.05);
const { captureClientEvent } = await import("../src/utils/telemetry");
await captureClientEvent("add", makeInstance());
expect(fetchMock).toHaveBeenCalledTimes(1);
});
});
describe("contract preservation", () => {
it("lifecycle event includes additionalData", async () => {
const { captureClientEvent } = await import("../src/utils/telemetry");
await captureClientEvent("init", makeInstance(), {
api_version: "v1.1",
client_type: "Memory",
});
const body = JSON.parse(fetchMock.mock.calls[0][1].body);
expect(body.properties.api_version).toBe("v1.1");
expect(body.properties.client_type).toBe("Memory");
expect(body.properties.sample_rate).toBe(1.0);
});
it("captureClientEvent without telemetryId is a no-op", async () => {
const { captureClientEvent } = await import("../src/utils/telemetry");
await captureClientEvent("add", makeInstance({ telemetryId: "" }));
expect(fetchMock).not.toHaveBeenCalled();
});
});
});
+54 -3
View File
@@ -2,6 +2,7 @@ import atexit
import logging
import os
import platform
import random
import sys
import threading
@@ -22,16 +23,65 @@ if not isinstance(MEM0_TELEMETRY, bool):
logging.getLogger("posthog").setLevel(logging.CRITICAL + 1)
logging.getLogger("urllib3").setLevel(logging.CRITICAL + 1)
_logger = logging.getLogger(__name__)
# Default sampling rate for hot-path OSS events. Lifecycle events always fire at 100%.
# Override via MEM0_TELEMETRY_SAMPLE_RATE env var.
_DEFAULT_SAMPLE_RATE = 0.1
def _parse_sample_rate(raw):
"""Parse MEM0_TELEMETRY_SAMPLE_RATE env var. Never raises."""
try:
value = float(raw)
except (TypeError, ValueError):
_logger.debug("MEM0_TELEMETRY_SAMPLE_RATE %r is not a number, defaulting to %s", raw, _DEFAULT_SAMPLE_RATE)
return _DEFAULT_SAMPLE_RATE
if not 0.0 <= value <= 1.0:
_logger.debug("MEM0_TELEMETRY_SAMPLE_RATE %s out of [0.0, 1.0], defaulting to %s", value, _DEFAULT_SAMPLE_RATE)
return _DEFAULT_SAMPLE_RATE
return value
MEM0_TELEMETRY_SAMPLE_RATE = _parse_sample_rate(os.environ.get("MEM0_TELEMETRY_SAMPLE_RATE", str(_DEFAULT_SAMPLE_RATE)))
# Events that bypass sampling and always fire. Keep this set in sync with the
# event names passed to capture_event() in mem0/memory/main.py.
_LIFECYCLE_EVENTS = frozenset({"mem0.init", "mem0.reset", "mem0._create_procedural_memory"})
def _sampling_before_send(msg):
"""PostHog before_send hook: drop sampled hot-path events, annotate survivors with sample_rate."""
if not isinstance(msg, dict):
return None
event_name = msg.get("event", "")
is_lifecycle = event_name in _LIFECYCLE_EVENTS
# >= so that rate=0 drops everything and rate=1 keeps everything (random ∈ [0, 1)).
if not is_lifecycle and random.random() >= MEM0_TELEMETRY_SAMPLE_RATE:
return None
# Annotate so PostHog dashboards can extrapolate true counts via 1/sample_rate.
properties = msg.setdefault("properties", {})
properties["sample_rate"] = 1.0 if is_lifecycle else MEM0_TELEMETRY_SAMPLE_RATE
return msg
class AnonymousTelemetry:
def __init__(self, vector_store=None):
def __init__(self, vector_store=None, before_send=None):
if not MEM0_TELEMETRY:
self.posthog = None
self.user_id = None
return
self.posthog = Posthog(project_api_key=PROJECT_API_KEY, host=HOST)
try:
self.posthog = Posthog(project_api_key=PROJECT_API_KEY, host=HOST, before_send=before_send)
except TypeError:
# posthog <4.5.0 does not accept before_send; fall back without sampling.
_logger.debug("posthog.Posthog does not accept before_send; upgrade to >=4.5.0 for sampling")
self.posthog = Posthog(project_api_key=PROJECT_API_KEY, host=HOST)
self.user_id = get_or_create_user_id(vector_store)
def capture_event(self, event_name, properties=None, user_email=None):
@@ -87,7 +137,7 @@ def _get_oss_telemetry():
# Double-checked locking
if _oss_telemetry_instance is not None:
return _oss_telemetry_instance
_oss_telemetry_instance = AnonymousTelemetry()
_oss_telemetry_instance = AnonymousTelemetry(before_send=_sampling_before_send)
atexit.register(_shutdown_oss_telemetry)
return _oss_telemetry_instance
@@ -102,6 +152,7 @@ def _shutdown_oss_telemetry():
# Module-level client telemetry singleton (used by capture_client_event).
# No before_send — hosted MemoryClient traffic must never be sampled.
client_telemetry = AnonymousTelemetry()
atexit.register(client_telemetry.close)
+1 -1
View File
@@ -17,7 +17,7 @@ dependencies = [
"qdrant-client>=1.9.1",
"pydantic>=2.7.3",
"openai>=1.90.0",
"posthog>=3.5.0",
"posthog>=4.5.0",
"pytz>=2024.1",
"sqlalchemy>=2.0.31",
"protobuf>=5.29.6,<7.0.0",
+206
View File
@@ -0,0 +1,206 @@
"""Tests for OSS telemetry sampling.
Tests target _sampling_before_send and its wiring directly. Existing tests in
test_telemetry.py cover opt-out, singleton, and lifecycle behavior and are
unaffected by sampling.
"""
from unittest.mock import patch
import pytest
import mem0.memory.telemetry as telemetry_module
class TestParseSampleRate:
"""Verify the env var parsing helper. Must never raise."""
@pytest.mark.parametrize(
"raw,expected",
[
("0.0", 0.0),
("0.1", 0.1),
("0.5", 0.5),
("1.0", 1.0),
("0.05", 0.05),
],
)
def test_valid_values(self, raw, expected):
assert telemetry_module._parse_sample_rate(raw) == expected
@pytest.mark.parametrize("raw", ["abc", "", "5.0", "-1", "-0.5", "1.5", "2", "nan"])
def test_invalid_values_fall_back_to_default(self, raw):
assert telemetry_module._parse_sample_rate(raw) == telemetry_module._DEFAULT_SAMPLE_RATE
def test_none_falls_back(self):
assert telemetry_module._parse_sample_rate(None) == telemetry_module._DEFAULT_SAMPLE_RATE
class TestSamplingBeforeSend:
"""Verify _sampling_before_send: the PostHog hook that drops/keeps events."""
def _msg(self, event_name, properties=None):
return {"event": event_name, "properties": properties or {}}
def test_lifecycle_event_passes_through_at_high_random(self):
"""Lifecycle events always fire even when random is near 1.0."""
with patch("mem0.memory.telemetry.random.random", return_value=0.999):
for event in ("mem0.init", "mem0.reset", "mem0._create_procedural_memory"):
result = telemetry_module._sampling_before_send(self._msg(event))
assert result is not None
assert result["properties"]["sample_rate"] == 1.0
def test_hot_path_event_dropped_when_random_above_rate(self):
"""At default rate 0.1, random=0.99 should drop the event."""
with patch.object(telemetry_module, "MEM0_TELEMETRY_SAMPLE_RATE", 0.1):
with patch("mem0.memory.telemetry.random.random", return_value=0.99):
assert telemetry_module._sampling_before_send(self._msg("mem0.add")) is None
def test_hot_path_event_passes_when_random_below_rate(self):
"""At default rate 0.1, random=0.05 should pass the event through."""
with patch.object(telemetry_module, "MEM0_TELEMETRY_SAMPLE_RATE", 0.1):
with patch("mem0.memory.telemetry.random.random", return_value=0.05):
result = telemetry_module._sampling_before_send(self._msg("mem0.add"))
assert result is not None
assert result["properties"]["sample_rate"] == 0.1
def test_hot_path_annotates_sample_rate(self):
with patch.object(telemetry_module, "MEM0_TELEMETRY_SAMPLE_RATE", 0.1):
with patch("mem0.memory.telemetry.random.random", return_value=0.0):
result = telemetry_module._sampling_before_send(self._msg("mem0.search"))
assert result["properties"]["sample_rate"] == 0.1
def test_lifecycle_annotates_sample_rate_one(self):
result = telemetry_module._sampling_before_send(self._msg("mem0.init"))
assert result["properties"]["sample_rate"] == 1.0
def test_rate_zero_drops_all_hot_path(self):
with patch.object(telemetry_module, "MEM0_TELEMETRY_SAMPLE_RATE", 0.0):
# Gate is `random >= rate`, so 0.0 >= 0.0 → drop. random ∈ [0, 1) means
# rate=0 drops every event.
with patch("mem0.memory.telemetry.random.random", return_value=0.0):
assert telemetry_module._sampling_before_send(self._msg("mem0.add")) is None
def test_rate_zero_still_passes_lifecycle(self):
with patch.object(telemetry_module, "MEM0_TELEMETRY_SAMPLE_RATE", 0.0):
with patch("mem0.memory.telemetry.random.random", return_value=0.999):
assert telemetry_module._sampling_before_send(self._msg("mem0.init")) is not None
def test_rate_one_passes_all_events(self):
with patch.object(telemetry_module, "MEM0_TELEMETRY_SAMPLE_RATE", 1.0):
with patch("mem0.memory.telemetry.random.random", return_value=0.999):
# 0.999 > 1.0 is False, so the gate never trips
assert telemetry_module._sampling_before_send(self._msg("mem0.add")) is not None
def test_does_not_override_caller_supplied_sample_rate_for_hot_path(self):
"""If a caller pre-populates sample_rate, our value still wins (we're authoritative)."""
with patch.object(telemetry_module, "MEM0_TELEMETRY_SAMPLE_RATE", 0.1):
with patch("mem0.memory.telemetry.random.random", return_value=0.0):
msg = self._msg("mem0.add", {"sample_rate": 0.99, "other": "x"})
result = telemetry_module._sampling_before_send(msg)
assert result["properties"]["sample_rate"] == 0.1
assert result["properties"]["other"] == "x"
def test_handles_missing_properties_dict(self):
"""If a msg arrives without a properties key, we create it before annotating."""
msg = {"event": "mem0.init"} # no properties
result = telemetry_module._sampling_before_send(msg)
assert result is not None
assert result["properties"]["sample_rate"] == 1.0
def test_handles_missing_event_field(self):
"""Defensive: a msg with no event field is treated as a hot-path event (gets sampled)."""
with patch.object(telemetry_module, "MEM0_TELEMETRY_SAMPLE_RATE", 0.1):
with patch("mem0.memory.telemetry.random.random", return_value=0.99):
# event is missing -> treated as hot-path -> dropped at high random
assert telemetry_module._sampling_before_send({"properties": {}}) is None
def test_handles_non_dict_input(self):
"""Defensive: if PostHog ever passes us a non-dict, we drop the event."""
assert telemetry_module._sampling_before_send("not a dict") is None
assert telemetry_module._sampling_before_send(None) is None
assert telemetry_module._sampling_before_send(42) is None
assert telemetry_module._sampling_before_send(["list"]) is None
class TestBeforeSendWiring:
"""Verify the hook is wired into the OSS singleton but NOT client_telemetry."""
def setup_method(self):
telemetry_module._oss_telemetry_instance = None
telemetry_module._oss_telemetry_shutting_down = False
def teardown_method(self):
telemetry_module._oss_telemetry_instance = None
telemetry_module._oss_telemetry_shutting_down = False
def test_oss_singleton_constructed_with_before_send_hook(self):
"""_get_oss_telemetry() should pass _sampling_before_send to AnonymousTelemetry."""
with patch.object(telemetry_module, "MEM0_TELEMETRY", True):
with patch("mem0.memory.telemetry.Posthog") as mock_posthog_cls:
with patch("mem0.memory.telemetry.get_or_create_user_id", return_value="u"):
with patch("atexit.register"):
telemetry_module._get_oss_telemetry()
# Verify Posthog() was constructed with before_send=_sampling_before_send
_, kwargs = mock_posthog_cls.call_args
assert kwargs.get("before_send") is telemetry_module._sampling_before_send
def test_client_telemetry_constructed_without_before_send(self):
"""The module-level client_telemetry must NOT receive before_send.
client_telemetry is created at import time, so this test verifies the
existing instance has no sampling hook attached to its underlying Posthog.
"""
# client_telemetry is created at module import time. We verify the
# construction call site by re-constructing AnonymousTelemetry() the
# same way and checking the kwargs.
with patch.object(telemetry_module, "MEM0_TELEMETRY", True):
with patch("mem0.memory.telemetry.Posthog") as mock_posthog_cls:
with patch("mem0.memory.telemetry.get_or_create_user_id", return_value="u"):
telemetry_module.AnonymousTelemetry() # mirrors client_telemetry construction
_, kwargs = mock_posthog_cls.call_args
# before_send is None (default), not _sampling_before_send
assert kwargs.get("before_send") is None
def test_anonymous_telemetry_falls_back_when_posthog_rejects_before_send(self):
"""If posthog (older version) rejects before_send, construction still succeeds."""
with patch.object(telemetry_module, "MEM0_TELEMETRY", True):
with patch("mem0.memory.telemetry.Posthog") as mock_posthog_cls:
with patch("mem0.memory.telemetry.get_or_create_user_id", return_value="u"):
# First call (with before_send) raises TypeError; second call succeeds.
mock_posthog_cls.side_effect = [TypeError("unexpected kwarg before_send"), object()]
at = telemetry_module.AnonymousTelemetry(before_send=telemetry_module._sampling_before_send)
# Constructor was called twice: once with before_send, once without
assert mock_posthog_cls.call_count == 2
assert at.posthog is not None
class TestSamplingDoesNotBreakExistingBehavior:
"""Regression checks: sampling must be invisible above the SDK boundary."""
def setup_method(self):
telemetry_module._oss_telemetry_instance = None
telemetry_module._oss_telemetry_shutting_down = False
def teardown_method(self):
telemetry_module._oss_telemetry_instance = None
telemetry_module._oss_telemetry_shutting_down = False
def test_capture_event_signature_unchanged(self):
"""The public capture_event() function still accepts the same args."""
from inspect import signature
params = list(signature(telemetry_module.capture_event).parameters)
assert params == ["event_name", "memory_instance", "additional_data"]
def test_capture_client_event_signature_unchanged(self):
from inspect import signature
params = list(signature(telemetry_module.capture_client_event).parameters)
assert params == ["event_name", "instance", "additional_data"]
def test_default_sample_rate_is_ten_percent(self):
"""Module loads with the documented default."""
# We can't assert MEM0_TELEMETRY_SAMPLE_RATE directly because the user
# may have set the env var, so assert the default constant instead.
assert telemetry_module._DEFAULT_SAMPLE_RATE == 0.1