Skip to content

Commit ecd909e

Browse files
PavelKopeckyPavelKopecky
authored
feat!: move probe logs to timescale db (#896)
1 parent 5eeb61e commit ecd909e

22 files changed

Lines changed: 1515 additions & 245 deletions

‎config/default.cjs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -113,6 +113,13 @@ module.exports = {
113113
probeLimit: {
114114
asnCityPerUser: 2,
115115
},
116+
probeLogScopes: {
117+
activeWindow: 864_000, // 10 days in seconds
118+
readCacheTtl: 3_600_000, // 1 hour in ms
119+
minReporters: 10,
120+
fleetShare: 0.5,
121+
maxScopesPerReporter: 64,
122+
},
116123
adminData: {
117124
syncInterval: 60000,
118125
},

‎config/development.cjs‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,9 @@ module.exports = {
3434
adminData: {
3535
syncInterval: 5000,
3636
},
37+
probeLogScopes: {
38+
minReporters: 2,
39+
},
3740
reconnectProbesDelay: 0,
3841
sigtermDelay: 0,
3942
};

‎migrations/time-series-1/20260815120000_create-probe-logs.js‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
export const up = async (db) => {
2+
await db.schema.createTable('probe_log_counter', (table) => {
3+
table.uuid('probeUuid').primary();
4+
table.bigInteger('lastAllocatedId').notNullable();
5+
});
6+
7+
await db.schema.createTable('probe_log', (table) => {
8+
table.uuid('probeUuid').notNullable();
9+
table.bigInteger('probeLogId').notNullable();
10+
table.timestamp('timestamp', { useTz: true });
11+
table.timestamp('receivedAt', { useTz: true }).notNullable();
12+
table.string('level', 8);
13+
table.string('scope', 64);
14+
table.string('message', 8192).notNullable();
15+
});
16+
17+
await db.raw(`SELECT create_hypertable('probe_log', by_range('receivedAt', INTERVAL '1 day'))`);
18+
await db.raw(`SELECT add_retention_policy('probe_log', INTERVAL '30 days')`);
19+
await db.raw(`CREATE INDEX probe_log_probe_cursor_idx ON probe_log ("probeUuid", "probeLogId" DESC)`);
20+
await db.raw(`CREATE INDEX probe_log_probe_scope_cursor_idx ON probe_log ("probeUuid", "scope", "probeLogId" DESC)`);
21+
};
22+
23+
export const down = () => {};

‎src/lib/http/server.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@ import { registerAlternativeIpRoute } from '../../alternative-ip/route/alternati
3838
import { registerLimitsRoute } from '../../limits/route/get-limits.js';
3939
import { blacklist } from './middleware/blacklist.js';
4040
import { registerGetProbeLogsRoute } from '../../probe/route/get-probe-logs.js';
41+
import { registerGetProbeLogScopesRoute } from '../../probe/route/get-probe-log-scopes.js';
4142
import { registerRestartProbeRoute } from '../../probe/route/restart-probe.js';
4243
import { captureMiddlewareChainSpan, captureMiddlewareSpan } from '../metrics.js';
4344
import type { IoContext } from '../server.js';
@@ -84,6 +85,8 @@ export const getHttpServer = (ioContext: IoContext) => {
8485
registerGetProbesRoute(apiRouter, ioContext);
8586
// GET /probes/:id/logs
8687
registerGetProbeLogsRoute(apiRouter, ioContext);
88+
// GET /probes/log-scopes
89+
registerGetProbeLogScopesRoute(apiRouter);
8790
// POST /probes/:id/restart
8891
registerRestartProbeRoute(apiRouter, ioContext);
8992
// POST /send-code

‎src/lib/redis/scripts.ts‎

Lines changed: 121 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -226,5 +226,125 @@ const claimTimedOutMeasurements = defineScript({
226226
},
227227
});
228228

229-
export const scripts = { recordProgress, recordProgressAppend, recordResult, markFinishedByTimeout, claimTimedOutMeasurements };
229+
const registerProbeLogScopes = defineScript({
230+
SCRIPT: `
231+
local reporterKey = KEYS[1]
232+
local knownScopesKey = KEYS[2]
233+
local activeWindow = tonumber(ARGV[1])
234+
local maxScopes = tonumber(ARGV[2])
235+
local minReporters = tonumber(ARGV[3])
236+
local reporterIdentity = ARGV[4]
237+
local now = tonumber(redis.call('TIME')[1])
238+
local cutoff = now - activeWindow
239+
240+
redis.call('ZREMRANGEBYSCORE', reporterKey, '-inf', cutoff)
241+
242+
local scores = redis.call('ZMSCORE', reporterKey, unpack(ARGV, 5))
243+
local newScopes = 0
244+
245+
for index = 1, #scores do
246+
if not scores[index] then
247+
newScopes = newScopes + 1
248+
end
249+
end
250+
251+
if redis.call('ZCARD', reporterKey) + newScopes > maxScopes then
252+
return 0
253+
end
254+
255+
local zaddArguments = {}
256+
257+
for index = 5, #ARGV do
258+
table.insert(zaddArguments, now)
259+
table.insert(zaddArguments, ARGV[index])
260+
end
261+
262+
redis.call('ZADD', reporterKey, 'GT', unpack(zaddArguments))
263+
redis.call('EXPIRE', reporterKey, activeWindow)
264+
265+
local knownScopes = redis.call('SMISMEMBER', knownScopesKey, unpack(ARGV, 5))
266+
local newKnownScopes = {}
267+
268+
for index = 3, #KEYS do
269+
local scopeKey = KEYS[index]
270+
local scopeIndex = index - 2
271+
272+
redis.call('ZADD', scopeKey, 'GT', now, reporterIdentity)
273+
redis.call('ZREMRANGEBYSCORE', scopeKey, '-inf', cutoff)
274+
redis.call('EXPIRE', scopeKey, activeWindow)
275+
276+
if knownScopes[scopeIndex] == 0 and redis.call('ZCARD', scopeKey) >= minReporters then
277+
table.insert(newKnownScopes, ARGV[index + 2])
278+
end
279+
end
280+
281+
if #newKnownScopes > 0 then
282+
redis.call('SADD', knownScopesKey, unpack(newKnownScopes))
283+
end
284+
285+
return 1
286+
`,
287+
parseCommand (
288+
parser: CommandParser,
289+
reporterKey: string,
290+
knownScopesKey: string,
291+
scopeKeys: string[],
292+
reporterIdentity: string,
293+
activeWindow: number,
294+
maxScopes: number,
295+
minReporters: number,
296+
scopes: string[],
297+
) {
298+
parser.push((scopeKeys.length + 2).toString());
299+
parser.pushKey(reporterKey);
300+
parser.pushKey(knownScopesKey);
301+
scopeKeys.forEach(scopeKey => parser.pushKey(scopeKey));
302+
303+
parser.push(
304+
activeWindow.toString(),
305+
maxScopes.toString(),
306+
minReporters.toString(),
307+
reporterIdentity,
308+
...scopes,
309+
);
310+
},
311+
transformReply (reply: number) {
312+
return reply === 1;
313+
},
314+
});
315+
316+
const countProbeLogScopeReporters = defineScript({
317+
NUMBER_OF_KEYS: 2,
318+
SCRIPT: `
319+
local now = tonumber(redis.call('TIME')[1])
320+
local cutoff = now - tonumber(ARGV[2])
321+
322+
redis.call('ZREMRANGEBYSCORE', KEYS[2], '-inf', cutoff)
323+
local count = redis.call('ZCARD', KEYS[2])
324+
325+
if count < tonumber(ARGV[3]) then
326+
redis.call('SREM', KEYS[1], ARGV[1])
327+
end
328+
329+
return count
330+
`,
331+
parseCommand (parser: CommandParser, knownScopesKey: string, scopeKey: string, scope: string, activeWindow: number, minReporters: number) {
332+
parser.pushKey(knownScopesKey);
333+
parser.pushKey(scopeKey);
334+
parser.push(scope, activeWindow.toString(), minReporters.toString());
335+
},
336+
transformReply (reply: number) {
337+
return reply;
338+
},
339+
});
340+
341+
export const scripts = {
342+
recordProgress,
343+
recordProgressAppend,
344+
recordResult,
345+
markFinishedByTimeout,
346+
claimTimedOutMeasurements,
347+
registerProbeLogScopes,
348+
countProbeLogScopeReporters,
349+
};
230350
export type RedisScripts = typeof scripts;

‎src/lib/ws/gateway.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ import { handleIsIPv4SupportedUpdate, handleIsIPv6SupportedUpdate } from '../../
1414
import { handleNewLogs } from '../../probe/handler/logs.js';
1515
import { handleAltIps } from '../../probe/handler/alt-ips.js';
1616
import { handleAdoptionServerStart } from '../../probe/handler/local-adoption-server.js';
17+
import { handleLogScopes } from '../../probe/handler/log-scopes.js';
1718
import { handleSettingsUpdate } from '../../probe/handler/settings.js';
1819
import type { IoContext } from '../server.js';
1920
import { metricsAgent } from '../metrics.js';
@@ -45,6 +46,7 @@ export const initGateway = (ioContext: IoContext) => {
4546
// Handlers
4647
subscribeWithHandler(socket, 'probe:status:update', handleStatusUpdate(probe));
4748
subscribeWithAckHandler(socket, 'probe:logs', handleNewLogs(probe));
49+
subscribeWithHandler(socket, 'probe:log-scopes', handleLogScopes(probe));
4850
subscribeWithAckHandler(socket, 'probe:alt-ips', handleAltIps(probe, altIpsClient));
4951
subscribeWithHandler(socket, 'probe:isIPv6Supported:update', handleIsIPv6SupportedUpdate(probe));
5052
subscribeWithHandler(socket, 'probe:isIPv4Supported:update', handleIsIPv4SupportedUpdate(probe));

‎src/probe/handler/log-scopes.ts‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
import { logScopesSchema } from '../schema/probe-response-schema.js';
2+
import { getProbeLogScopesStorage } from '../log-scopes-storage.js';
3+
import { ProbeValidatorError } from '../../lib/probe-validator.js';
4+
import type { SocketProbe } from '../types.js';
5+
6+
const probeLogScopesStorage = getProbeLogScopesStorage();
7+
8+
export const handleLogScopes = (probe: SocketProbe) => async (scopes: string[]): Promise<void> => {
9+
const validation = logScopesSchema.validate(scopes);
10+
11+
if (validation.error) {
12+
throw validation.error;
13+
}
14+
15+
if (!await probeLogScopesStorage.writeScopes(probe.ipAddress, validation.value)) {
16+
throw new ProbeValidatorError('Probe log scope limit exceeded.');
17+
}
18+
};

‎src/probe/log-scopes-storage.ts‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
1+
import config from 'config';
2+
import { getPersistentRedisClient, type RedisClient } from '../lib/redis/persistent-client.js';
3+
import { getIpKey } from '../lib/ws/helper/probe-ip-limit.js';
4+
5+
export const KNOWN_SCOPES_KEY = 'gp:log-scopes';
6+
export const SCOPE_KEY_PREFIX = 'gp:log-scope:';
7+
export const REPORTER_SCOPES_KEY_PREFIX = 'gp:log-scopes:reporter:';
8+
export const scopeActiveWindow = config.get<number>('probeLogScopes.activeWindow');
9+
export const scopeReadCacheTtl = config.get<number>('probeLogScopes.readCacheTtl');
10+
export const minScopeReporters = config.get<number>('probeLogScopes.minReporters');
11+
export const scopeFleetShare = config.get<number>('probeLogScopes.fleetShare');
12+
export const maxScopesPerReporter = config.get<number>('probeLogScopes.maxScopesPerReporter');
13+
14+
export class ProbeLogScopesStorage {
15+
private cachedScopes: string[] | undefined;
16+
private cacheExpiresAt = 0;
17+
18+
constructor (private readonly redis: RedisClient) {}
19+
20+
async writeScopes (ipAddress: string, scopes: string[]): Promise<boolean> {
21+
const uniqueScopes = [ ...new Set(scopes) ];
22+
23+
if (uniqueScopes.length === 0) {
24+
return true;
25+
}
26+
27+
const reporterIdentity = getIpKey(ipAddress);
28+
const scopeKeys = uniqueScopes.map(scope => `${SCOPE_KEY_PREFIX}${scope}`);
29+
30+
// Keep the reporter cap, forward indexes, and known-scope publication atomic across concurrent reports and reads.
31+
return this.redis.registerProbeLogScopes(
32+
`${REPORTER_SCOPES_KEY_PREFIX}${reporterIdentity}`,
33+
KNOWN_SCOPES_KEY,
34+
scopeKeys,
35+
reporterIdentity,
36+
scopeActiveWindow,
37+
maxScopesPerReporter,
38+
minScopeReporters,
39+
uniqueScopes,
40+
);
41+
}
42+
43+
async readScopes (): Promise<string[]> {
44+
if (this.cachedScopes !== undefined && Date.now() < this.cacheExpiresAt) {
45+
return this.cachedScopes;
46+
}
47+
48+
const knownScopes = await this.redis.sMembers(KNOWN_SCOPES_KEY);
49+
50+
const counts = await Promise.all(knownScopes.map(scope => this.redis.countProbeLogScopeReporters(
51+
KNOWN_SCOPES_KEY,
52+
`${SCOPE_KEY_PREFIX}${scope}`,
53+
scope,
54+
scopeActiveWindow,
55+
minScopeReporters,
56+
)));
57+
58+
const totalReporters = counts.reduce((max, count) => Math.max(max, count), 0);
59+
const threshold = Math.max(totalReporters * scopeFleetShare, minScopeReporters);
60+
61+
this.cachedScopes = knownScopes
62+
.filter((_, index) => counts[index]! >= threshold)
63+
.sort();
64+
65+
this.cacheExpiresAt = Date.now() + scopeReadCacheTtl;
66+
67+
return this.cachedScopes;
68+
}
69+
}
70+
71+
let probeLogScopesStorage: ProbeLogScopesStorage;
72+
73+
export const getProbeLogScopesStorage = () => {
74+
if (!probeLogScopesStorage) {
75+
probeLogScopesStorage = new ProbeLogScopesStorage(getPersistentRedisClient());
76+
}
77+
78+
return probeLogScopesStorage;
79+
};

0 commit comments

Comments
 (0)