root/sp
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat(insights): multi-project Sentry archive

Browse Source 
The sentry-archive job was hardcoded to a single SENTRY_PROJECT slug.
Now that sase has a second Sentry project (sase-web, browser SDK), the
worker has to pull both — otherwise frontend events live only in
Sentry's UI and never reach the panel's sentry_events table for the
behavioral-insight pipeline to pick up.

Changes:

- SENTRY_PROJECT becomes a CSV (e.g. "python,sase-web") parsed once at
  module load into PROJECTS[]. Backward-compatible: a single value
  keeps working exactly as before.
- Issues are fetched org-wide in a single loop (unchanged endpoint)
  but each issue's Sentry project slug is now persisted via the new
  `sentrySourceProject` column on sentry_issues.
- Events are project-scoped on Sentry's side, so the job iterates
  projects and calls listEventsPage(slug, cursor) for each. Per-project
  pagination + dedup is preserved. A project-level failure is recorded
  in `error` but doesn't abort the other projects.
- New result shape: `perProject: { [slug]: { fetched, inserted, duplicate } }`
  Pipeline log gains a `[python=N/M sase-web=N/M]` breakdown so it's
  obvious at a glance which project produced what.
- Schema: nullable `sentrySourceProject` on both SentryIssue and
  SentryEvent (+ composite index per projectKey for queries that want
  to filter "panel project = sase AND sentry project = sase-web").
  Migration is a pure nullable add — `prisma db push` on container
  start is safe. Existing rows stay null until next sync touches them.

After merge, set `SENTRY_PROJECT=python,sase-web` on the panel-worker
Coolify app and redeploy.
This commit is contained in:
Semih
2026-06-01 18:37:10 +03:00
parent 928728dc66
commit 8a81cb77bf
4 changed files with 147 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

122
apps/worker/src/jobs/sentry-archive.ts
View File

@@ -1,6 +1,12 @@
import { gzipSync } from "node:zlib";
import { prisma } from "../db";
import { isConfigured, listIssuesPage, listEventsPage, type SentryEventRaw } from "../lib/sentry";
import {
getProjects,
isConfigured,
listEventsPage,
listIssuesPage,
type SentryEventRaw,
} from "../lib/sentry";
import { putBuffer } from "../lib/minio";
const PROJECT_KEY = process.env.INSIGHT_PROJECT_KEY ?? "sase";
@@ -15,6 +21,7 @@ export type SentryArchiveResult = {
eventsInserted: number;
eventsDuplicate: number;
dumped: number;
perProject: Record<string, { fetched: number; inserted: number; duplicate: number }>;
error?: string;
};
@@ -25,17 +32,29 @@ function toDate(v: unknown): Date | null {
}
export async function runArchiveSentry(): Promise<SentryArchiveResult> {
const perProject: SentryArchiveResult["perProject"] = {};
if (!isConfigured()) {
return { issuesSynced: 0, eventsFetched: 0, eventsInserted: 0, eventsDuplicate: 0, dumped: 0, error: "sentry_not_configured" };
return {
issuesSynced: 0,
eventsFetched: 0,
eventsInserted: 0,
eventsDuplicate: 0,
dumped: 0,
perProject,
error: "sentry_not_configured",
};
}
const projects = getProjects();
let issuesSynced = 0;
let eventsFetched = 0;
let eventsInserted = 0;
let eventsDuplicate = 0;
let error: string | undefined;
// ── Issues: upsert latest aggregate state ──
// ── Issues: single org-wide loop covers all projects. ───────────────────
// Each issue carries `project.slug` so rows preserve which Sentry project
// raised them.
try {
let cursor: string | undefined;
for (let page = 0; page < ISSUE_MAX_PAGES; page++) {
@@ -43,8 +62,10 @@ export async function runArchiveSentry(): Promise<SentryArchiveResult> {
if (!data.length) break;
for (const i of data) {
if (!i.id) continue;
const sentrySourceProject = i.project?.slug ?? null;
const row = {
projectKey: PROJECT_KEY,
sentrySourceProject,
shortId: i.shortId ?? null,
title: i.title ?? null,
culprit: i.culprit ?? null,
@@ -73,46 +94,58 @@ export async function runArchiveSentry(): Promise<SentryArchiveResult> {
error = `issues_failed: ${(e as Error).message}`;
}
// ── Events: newest-first, dedup by eventId, stop once caught up ──
try {
let cursor: string | undefined;
for (let page = 0; page < EVENT_MAX_PAGES; page++) {
const { data, nextCursor } = await listEventsPage(cursor);
if (!data.length) break;
// ── Events: per Sentry project, newest-first, stop once caught up. ──────
for (const slug of projects) {
const counters = { fetched: 0, inserted: 0, duplicate: 0 };
perProject[slug] = counters;
try {
let cursor: string | undefined;
for (let page = 0; page < EVENT_MAX_PAGES; page++) {
const { data, nextCursor } = await listEventsPage(slug, cursor);
if (!data.length) break;
const batch = data
.map((e: SentryEventRaw) => {
const eventId = String(e.eventID ?? e.id ?? "");
const ts = toDate(e.dateCreated ?? e.dateReceived);
if (!eventId || !ts) return null;
return {
eventId,
issueId: e.groupID ? String(e.groupID) : null,
projectKey: PROJECT_KEY,
title: e.title ?? null,
message: typeof e.message === "string" ? e.message : null,
level: typeof e.level === "string" ? (e.level as string) : null,
platform: e.platform ?? null,
timestamp: ts,
tags: (e.tags ?? []) as object,
payload: e as object,
};
})
.filter((r): r is NonNullable<typeof r> => r !== null);
const batch = data
.map((e: SentryEventRaw) => {
const eventId = String(e.eventID ?? e.id ?? "");
const ts = toDate(e.dateCreated ?? e.dateReceived);
if (!eventId || !ts) return null;
return {
eventId,
issueId: e.groupID ? String(e.groupID) : null,
projectKey: PROJECT_KEY,
sentrySourceProject: slug,
title: e.title ?? null,
message: typeof e.message === "string" ? e.message : null,
level: typeof e.level === "string" ? (e.level as string) : null,
platform: e.platform ?? null,
timestamp: ts,
tags: (e.tags ?? []) as object,
payload: e as object,
};
})
.filter((r): r is NonNullable<typeof r> => r !== null);
eventsFetched += batch.length;
if (batch.length) {
const res = await prisma.sentryEvent.createMany({ data: batch as never, skipDuplicates: true });
eventsInserted += res.count;
eventsDuplicate += batch.length - res.count;
// A full page with no new rows means we've reached already-archived events.
if (res.count === 0) break;
counters.fetched += batch.length;
eventsFetched += batch.length;
if (batch.length) {
const res = await prisma.sentryEvent.createMany({
data: batch as never,
skipDuplicates: true,
});
counters.inserted += res.count;
counters.duplicate += batch.length - res.count;
eventsInserted += res.count;
eventsDuplicate += batch.length - res.count;
// A full page with no new rows means we've reached already-archived events.
if (res.count === 0) break;
}
if (!nextCursor) break;
cursor = nextCursor;
}
if (!nextCursor) break;
cursor = nextCursor;
} catch (e) {
// Project-level failure doesn't stop the other projects from being archived.
error = error ?? `events_failed[${slug}]: ${(e as Error).message}`;
}
} catch (e) {
error = error ?? `events_failed: ${(e as Error).message}`;
}
// ── Cold dump closed days to MinIO ──
@@ -123,7 +156,15 @@ export async function runArchiveSentry(): Promise<SentryArchiveResult> {
error = error ?? `dump_failed: ${(e as Error).message}`;
}
return { issuesSynced, eventsFetched, eventsInserted, eventsDuplicate, dumped, error };
return {
issuesSynced,
eventsFetched,
eventsInserted,
eventsDuplicate,
dumped,
perProject,
error,
};
}
async function dumpClosedDays(now: Date): Promise<number> {
@@ -150,6 +191,7 @@ async function dumpClosedDays(now: Date): Promise<number> {
JSON.stringify({
eventId: r.eventId,
issueId: r.issueId,
sentrySourceProject: r.sentrySourceProject,
title: r.title,
message: r.message,
level: r.level,

34
apps/worker/src/lib/sentry.ts
View File

@@ -1,13 +1,27 @@
// Sentry API client (read-only) for the archival pipeline. Uses a User Auth
// Token with org:read / project:read / event:read. EU-region orgs use the
// de.sentry.io API base.
//
// Multi-project support: SENTRY_PROJECT is a CSV (e.g. "python,sase-web").
// Issues come from the org-wide endpoint (one call covers all projects), so
// they're fetched in a single loop and the per-issue `project.slug` is
// retained. Events are project-scoped on Sentry's side, so the archive job
// iterates `getProjects()` and calls `listEventsPage(slug, cursor)` per slug.
const TOKEN = process.env.SENTRY_AUTH_TOKEN ?? "";
const ORG = process.env.SENTRY_ORG ?? "";
const PROJECT = process.env.SENTRY_PROJECT ?? "";
const RAW_PROJECTS = process.env.SENTRY_PROJECT ?? "";
const API_BASE = (process.env.SENTRY_API_BASE ?? "https://sentry.io/api/0").replace(/\/$/, "");
const PROJECTS: string[] = RAW_PROJECTS.split(",")
.map((s) => s.trim())
.filter(Boolean);
export function isConfigured(): boolean {
return Boolean(TOKEN && ORG && PROJECT);
return Boolean(TOKEN && ORG && PROJECTS.length > 0);
}
export function getProjects(): string[] {
return [...PROJECTS];
}
const headers = () => ({ Authorization: `Bearer ${TOKEN}`, Accept: "application/json" });
@@ -36,6 +50,12 @@ async function getPage<T>(path: string, params: Record<string, string>): Promise
return { data, nextCursor: parseNextCursor(res.headers.get("link")) };
}
export type SentryProjectRef = {
id?: string;
slug?: string;
name?: string;
};
export type SentryIssueRaw = {
id: string;
shortId?: string;
@@ -50,6 +70,7 @@ export type SentryIssueRaw = {
lastSeen?: string;
permalink?: string;
metadata?: Record<string, unknown>;
project?: SentryProjectRef;
};
export type SentryEventRaw = {
@@ -67,6 +88,9 @@ export type SentryEventRaw = {
};
// One page of org issues (newest activity first). statsPeriod bounds the window.
// Sentry's response includes a `project: { slug, id, name }` per issue so the
// caller can distinguish which Sentry project a row came from without a
// second round-trip.
export function listIssuesPage(cursor?: string, statsPeriod = "90d"): Promise<Page<SentryIssueRaw>> {
return getPage<SentryIssueRaw>(`/organizations/${ORG}/issues/`, {
project: "",
@@ -78,11 +102,11 @@ export function listIssuesPage(cursor?: string, statsPeriod = "90d"): Promise<Pa
}
// One page of project events, full payloads, newest first.
export function listEventsPage(cursor?: string): Promise<Page<SentryEventRaw>> {
return getPage<SentryEventRaw>(`/projects/${ORG}/${PROJECT}/events/`, {
export function listEventsPage(projectSlug: string, cursor?: string): Promise<Page<SentryEventRaw>> {
return getPage<SentryEventRaw>(`/projects/${ORG}/${projectSlug}/events/`, {
full: "true",
...(cursor ? { cursor } : {}),
});
}
export const sentryConfig = { ORG, PROJECT, API_BASE };
export const sentryConfig = { ORG, PROJECTS, API_BASE };

5
apps/worker/src/schedulers/pipeline.ts
View File

@@ -121,8 +121,11 @@ async function runJob(job: Job) {
case "sentry-archive": {
const res = await runArchiveSentry();
if (res.eventsInserted > 0 || res.issuesSynced > 0 || res.dumped > 0 || res.error) {
const perProj = Object.entries(res.perProject)
.map(([slug, c]) => `${slug}=${c.fetched}/${c.inserted}`)
.join(" ");
console.log(
`[pipeline] sentry-archive issues=${res.issuesSynced} events=${res.eventsFetched}/${res.eventsInserted} dup=${res.eventsDuplicate} dumped=${res.dumped}${res.error ? ` error=${res.error}` : ""}`,
`[pipeline] sentry-archive issues=${res.issuesSynced} events=${res.eventsFetched}/${res.eventsInserted} dup=${res.eventsDuplicate} dumped=${res.dumped}${perProj ? ` [${perProj}]` : ""}${res.error ? ` error=${res.error}` : ""}`,
);
}
return res;