From c290d0663ac6d605879018fd5d7c014eb7836fda Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carl-Gerhard=20Lindesva=CC=88rd?= Date: Sun, 25 Jan 2026 14:27:29 +0100 Subject: [PATCH 1/8] feat: S3/GCS data export integration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Export a project's analytics events to the user's own S3 or GCS bucket. ClickHouse is the single source of truth for the export — there is no Redis buffer and no per-event ingestion hook, so a stalled/failing bucket can never touch the ingestion path (in either the Kafka or GroupMQ mode). How it works: - A new `inserted_at DateTime64(3)` column on the events table records ingestion time. It DEFAULTs to `created_at` (deterministic for pre-migration parts — a DEFAULT now() would be evaluated lazily at read time and never settle) and is set explicitly at insert time (createEvent + the import production-move), so backdated events (server-side, offline, imports) still get a real insert time. A minmax skip index keeps the windowed scan cheap since inserted_at isn't in the primary key. - The flushExports cron job (every 60s) windows the events table by inserted_at for each (project, integration), batches rows into gzipped JSONL + a manifest, and uploads them. A per-(project, integration) ExportWatermark (Postgres) tracks progress with a composite (inserted_at, id) cursor — the id tie-breaker is required because an import stamps one inserted_at across its whole batch. A safety lag behind now() avoids reading in-flight inserts; the watermark only advances after a successful upload (at-least-once; the manifest is the commit marker). First run starts from now(), so connecting an export doesn't dump all history (historical backfill = reset the watermark). Integrations are organization-scoped, so every active project in the org exports independently and gets its own watermark + object path (layout keyed by project_id). Destinations are a pluggable object-store sink (S3 + GCS adapters) so future targets are small additions. Credentials (S3 secret keys, GCS service-account keys) are encrypted at rest with CREDENTIALS_ENCRYPTION_KEY and decrypted inside the adapters. Export batching and the safety lag are tunable via EXPORT_* env vars. Rebased onto main after the Redpanda/Kafka migration. Claude-Session: https://claude.ai/code/session_01AWucrxmRqWqgWBvr8cC5cj --- .env.example | 15 + apps/api/package.json | 5 + apps/api/tsdown.config.ts | 4 + .../forms/gcs-export-integration.tsx | 185 +++++++ .../forms/s3-export-integration.tsx | 325 +++++++++++ .../components/integrations/integrations.tsx | 24 +- apps/start/src/modals/add-integration.tsx | 16 + apps/worker/package.json | 6 + apps/worker/src/boot-cron.ts | 5 + apps/worker/src/boot-debug.ts | 1 + apps/worker/src/jobs/cron.flush-exports.ts | 290 ++++++++++ apps/worker/src/jobs/cron.ts | 4 + apps/worker/tsdown.config.ts | 4 + packages/common/server/encryption.ts | 123 +++++ packages/common/server/index.ts | 1 + .../18-add-events-inserted-at.ts | 56 ++ packages/db/index.ts | 1 + .../migration.sql | 24 + packages/db/prisma/schema.prisma | 22 + packages/db/src/exports/batch-creator.ts | 219 ++++++++ packages/db/src/exports/export-event.ts | 72 +++ packages/db/src/exports/index.ts | 25 + packages/db/src/exports/manifest.ts | 70 +++ packages/db/src/services/event.service.ts | 9 + packages/db/src/services/import.service.ts | 6 +- packages/integrations/package.json | 7 +- .../src/object-store/gcs-adapter.ts | 165 ++++++ .../integrations/src/object-store/index.ts | 7 + .../src/object-store/s3-adapter.ts | 293 ++++++++++ .../integrations/src/object-store/types.ts | 45 ++ packages/queue/src/queues.ts | 5 + packages/trpc/src/routers/integration.ts | 80 +++ packages/validation/src/index.ts | 69 ++- pnpm-lock.yaml | 512 +++++++++++++++++- 34 files changed, 2682 insertions(+), 13 deletions(-) create mode 100644 apps/start/src/components/integrations/forms/gcs-export-integration.tsx create mode 100644 apps/start/src/components/integrations/forms/s3-export-integration.tsx create mode 100644 apps/worker/src/jobs/cron.flush-exports.ts create mode 100644 packages/common/server/encryption.ts create mode 100644 packages/db/code-migrations/18-add-events-inserted-at.ts create mode 100644 packages/db/prisma/migrations/20260622204749_export_watermarks/migration.sql create mode 100644 packages/db/src/exports/batch-creator.ts create mode 100644 packages/db/src/exports/export-event.ts create mode 100644 packages/db/src/exports/index.ts create mode 100644 packages/db/src/exports/manifest.ts create mode 100644 packages/integrations/src/object-store/gcs-adapter.ts create mode 100644 packages/integrations/src/object-store/index.ts create mode 100644 packages/integrations/src/object-store/s3-adapter.ts create mode 100644 packages/integrations/src/object-store/types.ts diff --git a/.env.example b/.env.example index eea0d9bde..48e75ae5a 100644 --- a/.env.example +++ b/.env.example @@ -8,6 +8,21 @@ CLICKHOUSE_URL="http://localhost:8123/openpanel" # Generate with: openssl rand -hex 32 ENCRYPTION_KEY="" +# Symmetric key used to encrypt object-store export credentials (S3 secret +# access keys, GCS service-account keys) at rest. Only required if you configure +# an S3/GCS data-export integration. Generate with: openssl rand -hex 32 +# CREDENTIALS_ENCRYPTION_KEY="" + +# OBJECT-STORE EXPORT (S3/GCS) tuning — optional, sensible defaults shown. +# The flushExports cron job windows ClickHouse by inserted_at and uploads +# batched files. LAG keeps a safety gap behind now() for in-flight inserts; +# BATCH_SIZE is rows per file; MAX_BATCHES_PER_RUN bounds backlog drain per +# tick; CONCURRENCY is parallel (project,integration) uploads. +# EXPORT_LAG_SECONDS="60" +# EXPORT_BATCH_SIZE="50000" +# EXPORT_MAX_BATCHES_PER_RUN="20" +# EXPORT_CONCURRENCY="4" + # REST BATCH_SIZE="5000" BATCH_INTERVAL="10000" diff --git a/apps/api/package.json b/apps/api/package.json index 3b845cc46..fa42287e6 100644 --- a/apps/api/package.json +++ b/apps/api/package.json @@ -77,5 +77,10 @@ "tsdown": "0.14.2", "typescript": "catalog:", "vitest": "^1.0.0" + }, + "peerDependencies": { + "@aws-sdk/client-s3": "^3.974.0", + "@aws-sdk/client-sts": "^3.974.0", + "@google-cloud/storage": "^7.18.0" } } \ No newline at end of file diff --git a/apps/api/tsdown.config.ts b/apps/api/tsdown.config.ts index c41662a85..fcc2fd9a2 100644 --- a/apps/api/tsdown.config.ts +++ b/apps/api/tsdown.config.ts @@ -10,6 +10,10 @@ const options: Options = { 'pino', 'pino-pretty', '@node-rs/argon2', + // integrations package + '@aws-sdk/client-s3', + '@aws-sdk/client-sts', + '@google-cloud/storage', ], sourcemap: true, platform: 'node', diff --git a/apps/start/src/components/integrations/forms/gcs-export-integration.tsx b/apps/start/src/components/integrations/forms/gcs-export-integration.tsx new file mode 100644 index 000000000..a2bb17cca --- /dev/null +++ b/apps/start/src/components/integrations/forms/gcs-export-integration.tsx @@ -0,0 +1,185 @@ +import { InputWithLabel } from '@/components/forms/input-with-label'; +import { Button } from '@/components/ui/button'; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from '@/components/ui/select'; +import { useAppParams } from '@/hooks/use-app-params'; +import { useTRPC } from '@/integrations/trpc/react'; +import type { RouterOutputs } from '@/trpc/client'; +import { zodResolver } from '@hookform/resolvers/zod'; +import { zCreateGCSExportIntegration } from '@openpanel/validation'; +import { useMutation } from '@tanstack/react-query'; +import { path, mergeDeepRight } from 'ramda'; +import { Controller, useForm } from 'react-hook-form'; +import { toast } from 'sonner'; +import type { z } from 'zod'; + +type IForm = z.infer; + +export function GCSExportIntegrationForm({ + defaultValues, + onSuccess, +}: { + defaultValues?: RouterOutputs['integration']['get']; + onSuccess: () => void; +}) { + const { organizationId } = useAppParams(); + const form = useForm({ + defaultValues: mergeDeepRight( + { + id: defaultValues?.id, + organizationId, + name: '', + config: { + type: 'gcs_export' as const, + bucket: '', + prefix: 'openpanel-exports', + format: 'jsonl_gzip' as const, + serviceAccountKey: '', + }, + }, + defaultValues ?? {}, + ), + resolver: zodResolver(zCreateGCSExportIntegration), + }); + const trpc = useTRPC(); + const mutation = useMutation( + trpc.integration.createOrUpdateExport.mutationOptions({ + onSuccess, + onError(error) { + toast.error(error.message || 'Failed to create integration'); + }, + }), + ); + + const testMutation = useMutation( + trpc.integration.testExportConnection.mutationOptions({ + onSuccess(data) { + if (data.success) { + toast.success('Connection successful! Bucket is accessible.'); + } else { + toast.error(`Connection failed: ${data.error}`); + } + }, + onError(error) { + toast.error(error.message || 'Failed to test connection'); + }, + }), + ); + + const handleSubmit = (values: IForm) => { + mutation.mutate(values); + }; + + const handleError = () => { + toast.error('Please fix validation errors'); + }; + + const handleTest = () => { + const values = form.getValues(); + if (!values.config.bucket || !values.config.serviceAccountKey) { + return toast.error('Bucket and Service Account Key are required'); + } + testMutation.mutate(values); + }; + + return ( +
+ + +
+ + +
+ +
+ + ( + + )} + /> +
+ +
+ +