diff --git a/.changeset/observability-unification.md b/.changeset/observability-unification.md new file mode 100644 index 00000000..51270e27 --- /dev/null +++ b/.changeset/observability-unification.md @@ -0,0 +1,7 @@ +--- +'@inkcre/core': patch +'@inkcre/client-web': patch +'@inkcre/client-webext': patch +--- + +统一观测记录的采集与查询，默认使用内置信息库，选定 Loki 时通过具备能力的 Peer 读写；Web 提供全局筛选、关联时间线与明确的后端/历史状态，Job 生命周期记录不再依赖额外 OTEL 出口开关。 diff --git a/apps/client-web/src/components/common/AppSidePanel/AppSidePanel.vue b/apps/client-web/src/components/common/AppSidePanel/AppSidePanel.vue index 4ba66e20..f7ca2966 100644 --- a/apps/client-web/src/components/common/AppSidePanel/AppSidePanel.vue +++ b/apps/client-web/src/components/common/AppSidePanel/AppSidePanel.vue @@ -65,6 +65,13 @@ const NAV_ITEMS: NavItem[] = [ matches: ['/peers'], requiresReady: true, }, + { + name: 'observability', + labelKey: 'sidePanel.observability', + to: '/observability', + matches: ['/observability'], + requiresReady: true, + }, { name: 'settings', labelKey: 'sidePanel.settings', diff --git a/apps/client-web/src/components/obsrv/LogEntry/LogEntry.md b/apps/client-web/src/components/obsrv/LogEntry/LogEntry.md deleted file mode 100644 index 52e1be9b..00000000 --- a/apps/client-web/src/components/obsrv/LogEntry/LogEntry.md +++ /dev/null @@ -1,5 +0,0 @@ -# LogEntry - -显示一条日志的时间、严重级别和正文。传入 `log: Log` 或 `logId: number`，二选一；传 ID 时通过现有查询接口读取日志。 - -摘要行是原生按钮，点击、Enter 或 Space 均可展开／收起详情，`aria-expanded` 同步反映状态。详情展示 Trace ID、Span ID、严重级别数值和非空 attributes。时间使用等宽元数据角色，日志正文使用 body-sm；长正文可以换行，结构化属性保留等宽代码格式并在局部滚动。 diff --git a/apps/client-web/src/components/obsrv/LogEntry/LogEntry.scss b/apps/client-web/src/components/obsrv/LogEntry/LogEntry.scss deleted file mode 100644 index bbd40eb8..00000000 --- a/apps/client-web/src/components/obsrv/LogEntry/LogEntry.scss +++ /dev/null @@ -1,99 +0,0 @@ -.log-entry { - display: flex; - flex-direction: column; - gap: sys-var(space, xs); - - &__main { - width: 100%; - border: 0; - background: transparent; - text-align: left; - - &:focus-visible { - outline: 2px solid sys-var(color, border, strong); - outline-offset: 2px; - } - display: flex; - flex-wrap: wrap; - gap: sys-var(space, sm); - align-items: center; - cursor: pointer; - padding: sys-var(space, xs); - border-radius: sys-var(radius, none); - transition: background-color 0.2s; - - &:hover { - background-color: sys-var(color, surface, subtle-hover); - } - } - - &__chevron { - @include apply-icon(sm); - color: sys-var(color, text, subtle); - transition: transform 0.2s; - flex-shrink: 0; - - &--expanded { - transform: rotate(90deg); - } - } - - .log-time { - @include apply-font(label-md, $mono: true); - color: sys-var(color, text, subtle); - min-width: 80px; - } - - .log-severity { - @include apply-font(label-md); - color: sys-var(color, text, base); - min-width: 60px; - } - - .log-body { - @include apply-font(body-sm); - color: sys-var(color, text, base); - flex: 1 1 12rem; - min-width: 0; - overflow-wrap: anywhere; - } - - &__details { - display: flex; - flex-direction: column; - gap: sys-var(space, xs); - padding: sys-var(space, sm) sys-var(space, md); - background-color: sys-var(color, surface, subtle); - border-radius: sys-var(radius, none); - } - - &__detail-row { - display: flex; - flex-wrap: wrap; - gap: sys-var(space, sm); - justify-content: space-between; - width: 100%; - } - - &__detail-label { - @include apply-font(label-md); - color: sys-var(color, text, subtle); - min-width: 120px; - flex-shrink: 0; - } - - &__detail-value { - @include apply-font(body-sm, $mono: true); - color: sys-var(color, text, base); - flex: 1 1 12rem; - min-width: 0; - overflow-wrap: anywhere; - - &--code { - padding: sys-var(space, xs); - border-radius: sys-var(radius, none); - margin: 0; - white-space: pre-wrap; - } - } -} diff --git a/apps/client-web/src/components/obsrv/LogEntry/LogEntry.ts b/apps/client-web/src/components/obsrv/LogEntry/LogEntry.ts deleted file mode 100644 index 23970486..00000000 --- a/apps/client-web/src/components/obsrv/LogEntry/LogEntry.ts +++ /dev/null @@ -1,9 +0,0 @@ -import type { Log } from '@inkcre/core' - -// --- Types --- - -// --- Props --- -export type LogEntryProps = { log: Log; logId?: never } | { log?: never; logId: number } - -// --- Emits --- -export const logEntryEmits = {} as const diff --git a/apps/client-web/src/components/obsrv/LogEntry/LogEntry.vue b/apps/client-web/src/components/obsrv/LogEntry/LogEntry.vue deleted file mode 100644 index f534c218..00000000 --- a/apps/client-web/src/components/obsrv/LogEntry/LogEntry.vue +++ /dev/null @@ -1,66 +0,0 @@ - - - - - diff --git a/package.json b/package.json index 6197e3b1..efd405d4 100644 --- a/package.json +++ b/package.json @@ -40,7 +40,7 @@ "build": "pnpm check:workspace && pnpm -r run build && pnpm check:package", "check": "pnpm format:check && pnpm lint && pnpm type-check && pnpm build && pnpm check:sdk", "ci": "pnpm check && pnpm test:e2e", - "check:sdk": "pnpm --filter @inkcre/core build && node --test scripts/tests/core-independent.test.mjs" + "check:sdk": "pnpm --filter @inkcre/core build && node --test scripts/tests/core-independent.test.mjs scripts/tests/observability.test.mjs" }, "devDependencies": { "@changesets/cli": "3.0.0", diff --git a/packages/core/README.md b/packages/core/README.md index 789da5ba..77c9fb01 100644 --- a/packages/core/README.md +++ b/packages/core/README.md @@ -23,3 +23,13 @@ if (error) throw error 这是按实际用途切出的消费边界，不是新的平台适配包。当前独立入口不承诺覆盖根入口所有高层领域 facade 或 Host UI；需要既有 Vue 生命周期的功能继续从兼容入口消费。Core SDK 版本仍由包自身发布，数据库协议与 MF Host ABI 仍按各自原合同判断。 `node --test scripts/tests/core-independent.test.mjs` 从本仓根检查真实 npm tarball：仓库外导入、无 UI 声明编译、并发目标隔离、序列化和二进制请求。测试使用受控传输，不连接真实数据库；构建与本地检查不能替代跨语言端到端验收。 + +## 统一观测 + +独立入口导出 `ObservabilityClient`，构造参数为显式 `DatabaseConnection`；构造不读取网络或启动后台任务。`query({since, until, limit, ...filters})` 返回有界窗口与 `truncated`，过滤先于截断；截断时缩窄窗口，不把返回页摘要解释为全库统计。`status()` 说明选中后端的配置与查询能力，不能将 `configured` 当作已验证网络就绪；未知的丢弃数是 `null`。 + +缺少 `inkcre.observability.backend` selector 时使用内置 PostgreSQL；selector 为 Loki 时通过已公告 `core.observability.v1` 的 Peer 读写。首次使用或显式调用 `prepare()` 后，实例固定所选后端；修改 selector 后须创建新实例。外部凭据仅属于执行 Peer，不进入浏览器、公共配置或调用参数。外部故障和未知配置明确失败，不悄悄改查 PostgreSQL。切换后只查询当前后端，历史保留在原处，不自动合并或迁移。 + +`emit(record)` 等待后端接收结果并报告失败；外部 Peer 接受进入队列不等于已经持久化。应用兼容入口的 operation/Job 记录队列有界，失败不改变业务结果，使用捕获的连接和 Peer 身份；默认记录不依赖 OTEL 开关。OTEL 继续作为额外出口。severity 沿用 OTEL 1–24：DEBUG 5、INFO 9、WARN 13、ERROR 17、FATAL 21。 + +旧 `Log` 是 deprecated 的 PostgreSQL 物理行/数字游标兼容 API，可能用于读取旧 PG 历史，不能代表当前选定后端。所有新的普通消费和第一方观测界面使用 `ObservabilityClient`。本次提供事件、日志和操作摘要，不声称具备完整分布式 trace 存储或指标后端。 diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index a9ea4e3d..bcfefaaf 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -140,6 +140,9 @@ export { export { Cron, CronForm } from './cron' // Observability +export * from './obsrv/contracts' +export { ObservabilityClient } from './obsrv/client' +export { createObservabilityClient } from './obsrv/app' export { Log, type LogRef, diff --git a/packages/core/src/obsrv/app.ts b/packages/core/src/obsrv/app.ts new file mode 100644 index 00000000..123d1322 --- /dev/null +++ b/packages/core/src/obsrv/app.ts @@ -0,0 +1,12 @@ +import { configStore } from '../config' +import { signDatabaseToken } from '../auth/sign' +import { ObservabilityClient } from './client' + +/** The compatibility entry snapshots App selection; the SDK constructor stays explicit. */ +export function createObservabilityClient(): ObservabilityClient { + const { INKCRE_PGREST_URL, INKCRE_JWT_SECRET } = configStore.metaConfig + return new ObservabilityClient({ + baseUrl: INKCRE_PGREST_URL, + tokenProvider: () => signDatabaseToken(INKCRE_JWT_SECRET), + }) +} diff --git a/packages/core/src/obsrv/client.ts b/packages/core/src/obsrv/client.ts new file mode 100644 index 00000000..9aad8cbd --- /dev/null +++ b/packages/core/src/obsrv/client.ts @@ -0,0 +1,266 @@ +import { z } from 'zod' +import { DatabaseClient, type DatabaseConnection } from '../base/client' +import { PeerCapabilitySnapshotSchema, PEER_HTTP_PROTOCOL } from '../peer/contracts' +import { + OBSERVABILITY_CAPABILITY, + ObservationPageSchema, + ObservationQuerySchema, + ObservationRecordSchema, + ObservabilityStatusSchema, + ObservabilityUnavailableError, + type ObservationQuery, + type ObservationRecord, + type ObservationPage, + type ObservabilityStatus, +} from './contracts' + +const BackendSchema = z.object({ backend: z.enum(['postgresql', 'loki']) }) +const Scalar = z.union([z.string().max(1024), z.number().finite(), z.boolean(), z.null()]) +const InputSchema = ObservationRecordSchema.extend({ + event_name: z.string().max(128).nullable(), + severity_text: z.string().max(32), + body: z.string().max(8192), + trace_id: z.string().max(128).nullable(), + span_id: z.string().max(128).nullable(), + attributes: z + .record(z.string().max(128), z.union([Scalar, z.array(Scalar).max(32)])) + .refine((value) => Object.keys(value).length <= 64), +}) +const EndpointSchema = z.object({ + method: z.literal('POST'), + url: z.url().refine((value) => { + const url = new URL(value) + return ['http:', 'https:'].includes(url.protocol) && !url.username && !url.password && !url.hash + }), +}) +const HISTORY = 'Only the selected backend is queried. Data from previous backends remains there.' + +/** One explicit deployment snapshot; queries never consult a selected App connection. */ +export class ObservabilityClient { + private readonly connection: DatabaseConnection + private readonly logs: DatabaseClient<'logs'> + private readonly configs: DatabaseClient<'configs'> + private readonly peers: DatabaseClient<'peers'> + private selectedBackend?: Promise<'postgresql' | 'loki'> + + constructor(connection: DatabaseConnection) { + this.connection = { ...connection } + this.logs = new DatabaseClient('logs', undefined, this.connection) + this.configs = new DatabaseClient('configs', undefined, this.connection) + this.peers = new DatabaseClient('peers', undefined, this.connection) + } + + private backend(): Promise<'postgresql' | 'loki'> { + this.selectedBackend ??= this.readBackend().catch((error: unknown) => { + this.selectedBackend = undefined + throw error + }) + return this.selectedBackend + } + + /** Bind this instance once; changing the deployment selector requires a new instance. */ + async prepare(): Promise { + await this.backend() + } + + private async readBackend(): Promise<'postgresql' | 'loki'> { + const result = await this.configs + .from() + .select('schema,value') + .eq('key', 'inkcre.observability.backend') + .abortSignal(AbortSignal.timeout(10_000)) + .maybeSingle() + if (result.error) + throw new ObservabilityUnavailableError('Observability configuration could not be read.') + if (!result.data) return 'postgresql' + if (result.data.schema !== 'inkcre.observability.backend.v1') + throw new ObservabilityUnavailableError('Unsupported observability configuration version.') + return BackendSchema.parse(result.data.value).backend + } + + private async externalProvider(action: 'status' | 'query' | 'ingest' = 'status'): Promise<{ + endpoint: z.infer + status: ObservabilityStatus + }> { + const result = await this.peers + .from() + .select('id,capabilities') + .gt('lease_expires_at', 'now') + .order('id') + .abortSignal(AbortSignal.timeout(10_000)) + if (result.error) + throw new ObservabilityUnavailableError('Observability Peer discovery failed.') + const deadline = Date.now() + 10_000 + let diagnostic: + | { endpoint: z.infer; status: ObservabilityStatus } + | undefined + for (const peer of result.data ?? []) { + if (Date.now() >= deadline) break + const capabilities = PeerCapabilitySnapshotSchema.safeParse(peer.capabilities) + if (!capabilities.success) continue + const capability = capabilities.data.find((item) => item.id === OBSERVABILITY_CAPABILITY) + if (!capability || capability.inbound.protocol !== PEER_HTTP_PROTOCOL) continue + const parsed = EndpointSchema.safeParse(capability.inbound.parameters) + if (!parsed.success) continue + const candidate = await this.dispatch( + parsed.data, + { action: 'status' }, + Math.min(3000, deadline - Date.now()) + ).catch(() => null) + const status = ObservabilityStatusSchema.safeParse(candidate) + if (!status.success || status.data.backend !== 'loki') continue + diagnostic ??= { endpoint: parsed.data, status: status.data } + const ready = + status.data.configured && + (action === 'ingest' ? status.data.recording : status.data.queryable) + if (ready) return { endpoint: parsed.data, status: status.data } + } + if (action === 'status' && diagnostic) return diagnostic + throw new ObservabilityUnavailableError( + 'No live Peer provides the selected observability backend.' + ) + } + + private async dispatch( + endpoint: z.infer, + body: unknown, + timeoutMs = 30_000 + ): Promise { + // After provider selection, dispatch once. Never redirect a failed write to another backend. + const response = await (this.connection.fetch ?? globalThis.fetch)(endpoint.url, { + method: endpoint.method, + headers: { + Authorization: `Bearer ${await this.connection.tokenProvider()}`, + 'Content-Type': 'application/json', + }, + body: JSON.stringify(body), + redirect: 'error', + signal: AbortSignal.timeout(Math.max(1, timeoutMs)), + }) + if (!response.ok) + throw new ObservabilityUnavailableError( + `Observability backend request failed (${response.status}).` + ) + return response.json() + } + + async status(): Promise { + const backend = await this.backend() + if (backend === 'loki') { + return (await this.externalProvider()).status + } + return { + backend, + configured: true, + recording: true, + queryable: true, + scope: 'selected_backend', + history: HISTORY, + dropped: null, + error: null, + } + } + + async query(input: ObservationQuery): Promise { + const query = ObservationQuerySchema.parse(input) + const backend = await this.backend() + if (backend === 'loki') { + const { endpoint } = await this.externalProvider('query') + const page = ObservationPageSchema.parse( + await this.dispatch(endpoint, { action: 'query', expected_backend: backend, query }) + ) + if (page.backend !== backend) + throw new ObservabilityUnavailableError('Observability backend changed during this query.') + return page + } + let request = this.logs + .from() + .select() + .gte('timestamp', query.since) + .lt('timestamp', query.until) + .order('timestamp', { ascending: false }) + .order('id', { ascending: false }) + .limit(query.limit + 1) + if (query.kind === 'log') + request = request.or( + 'attributes->>"inkcre.record.kind".eq.log,attributes->>"inkcre.record.kind".is.null' + ) + else if (query.kind) + request = request.contains('attributes', { 'inkcre.record.kind': query.kind }) + if (query.severity_min !== undefined) + request = request.gte('severity_number', query.severity_min) + if (query.job_id !== undefined) + request = request.or( + `trace_id.eq.job.${query.job_id},attributes->>"inkcre.job.id".eq.${query.job_id},attributes->>job_id.eq.${query.job_id}` + ) + if (query.trace_id) request = request.eq('trace_id', query.trace_id) + if (query.peer_id) request = request.contains('attributes', { 'inkcre.peer.id': query.peer_id }) + if (query.operation) + request = request.contains('attributes', { 'inkcre.operation': query.operation }) + if (query.text) + request = request.filter('body', 'imatch', query.text.replace(/[.*+?^${}()|[\]\\]/g, '\\$&')) + const result = await request.abortSignal(AbortSignal.timeout(10_000)) + if (result.error) + throw new ObservabilityUnavailableError( + 'Observations could not be read from the selected backend.' + ) + const rows = result.data ?? [] + const records = rows.slice(0, query.limit).map((row) => { + const attributes = z.record(z.string(), z.unknown()).parse(row.attributes ?? {}) + return ObservationRecordSchema.parse({ + ...row, + id: attributes['inkcre.record.id'] ?? `pg.${row.id}`, + kind: attributes['inkcre.record.kind'] ?? 'log', + event_name: attributes['inkcre.event.name'] ?? null, + attributes, + }) + }) + return { + records, + truncated: rows.length > query.limit, + since: query.since, + until: query.until, + backend, + } + } + + async emit(record: ObservationRecord): Promise { + const input = InputSchema.parse(record) + const backend = await this.backend() + if (backend === 'loki') { + const { endpoint } = await this.externalProvider('ingest') + const result = z.object({ accepted: z.number().int(), dropped: z.number().int() }).parse( + await this.dispatch(endpoint, { + action: 'ingest', + expected_backend: backend, + records: [input], + }) + ) + if (result.accepted !== 1 || result.dropped !== 0) + throw new ObservabilityUnavailableError( + 'The selected backend did not accept this observation.' + ) + return + } + const result = await this.logs + .insert({ + timestamp: input.timestamp, + severity_number: input.severity_number, + severity_text: input.severity_text, + body: input.body, + trace_id: input.trace_id, + span_id: input.span_id, + attributes: { + ...input.attributes, + 'inkcre.record.id': input.id, + 'inkcre.record.kind': input.kind, + 'inkcre.event.name': input.event_name, + }, + }) + .abortSignal(AbortSignal.timeout(10_000)) + if (result.error) + throw new ObservabilityUnavailableError( + 'Observation could not be stored in the selected backend.' + ) + } +} diff --git a/packages/core/src/obsrv/contracts.ts b/packages/core/src/obsrv/contracts.ts new file mode 100644 index 00000000..7eeaee41 --- /dev/null +++ b/packages/core/src/obsrv/contracts.ts @@ -0,0 +1,61 @@ +import { z } from 'zod' + +export const OBSERVABILITY_CAPABILITY = 'core.observability.v1' +export const ObservationRecordSchema = z.object({ + id: z.string().min(1).max(128), + timestamp: z.iso.datetime({ offset: true }), + kind: z.enum(['log', 'event', 'span']), + event_name: z.string().nullable(), + severity_number: z.number().int().min(1).max(24), + severity_text: z.string(), + body: z.string(), + trace_id: z.string().nullable(), + span_id: z.string().nullable(), + attributes: z.record(z.string(), z.unknown()), +}) +export type ObservationRecord = z.infer + +export const ObservationQuerySchema = z + .object({ + since: z.iso.datetime({ offset: true }), + until: z.iso.datetime({ offset: true }), + limit: z.number().int().min(1).max(500).default(100), + kind: z.enum(['log', 'event', 'span']).optional(), + severity_min: z.number().int().min(1).max(24).optional(), + job_id: z.number().int().positive().optional(), + trace_id: z.string().min(1).max(128).optional(), + peer_id: z.uuid().optional(), + operation: z.string().min(1).max(128).optional(), + text: z.string().max(256).optional(), + }) + .refine((value) => Date.parse(value.since) < Date.parse(value.until), { + message: 'Observation time range must end after it starts.', + }) + .refine((value) => Date.parse(value.until) - Date.parse(value.since) <= 31 * 86400 * 1000, { + message: 'Observation time range must not exceed 31 days.', + }) +export type ObservationQuery = z.input +export const ObservationPageSchema = z.object({ + records: z.array(ObservationRecordSchema), + truncated: z.boolean(), + since: z.iso.datetime({ offset: true }), + until: z.iso.datetime({ offset: true }), + backend: z.enum(['postgresql', 'loki']), +}) +export type ObservationPage = z.infer + +export const ObservabilityStatusSchema = z.object({ + backend: z.enum(['postgresql', 'loki']), + configured: z.boolean(), + recording: z.boolean(), + queryable: z.boolean(), + scope: z.literal('selected_backend'), + history: z.string(), + dropped: z.number().int().nonnegative().nullable(), + error: z.string().nullable(), +}) +export type ObservabilityStatus = z.infer + +export class ObservabilityUnavailableError extends Error { + override name = 'ObservabilityUnavailableError' +} diff --git a/packages/core/src/obsrv/index.ts b/packages/core/src/obsrv/index.ts index 810e9b67..a40100f0 100644 --- a/packages/core/src/obsrv/index.ts +++ b/packages/core/src/obsrv/index.ts @@ -1,2 +1,5 @@ export * from './log' export * from './telemetry' +export * from './contracts' +export * from './client' +export * from './app' diff --git a/packages/core/src/obsrv/log.ts b/packages/core/src/obsrv/log.ts index 681e1a32..a186841a 100644 --- a/packages/core/src/obsrv/log.ts +++ b/packages/core/src/obsrv/log.ts @@ -6,8 +6,9 @@ export type LogRef = number export const LogRefZ = z.number() /** - * Log model for observability logs. - * Used for tracking events, errors, and structured logging. + * @deprecated Physical PostgreSQL row compatibility API, including historical rows. + * Use ObservabilityClient for the selected backend. Numeric IDs/cursors cannot + * represent external observations and this legacy API intentionally does not. */ export class Log extends Z.class({ id: LogRefZ, diff --git a/packages/core/src/obsrv/recording.ts b/packages/core/src/obsrv/recording.ts new file mode 100644 index 00000000..965f1973 --- /dev/null +++ b/packages/core/src/obsrv/recording.ts @@ -0,0 +1,70 @@ +import { configStore } from '../config' +import { createObservabilityClient } from './app' +import type { ObservabilityClient } from './client' +import type { ObservationRecord } from './contracts' + +export interface ObservationRecorder { + client: ObservabilityClient + peerId: string +} + +const pending = new Set>() +let warned = false +let current: { url: string; secret: string; recorder: ObservationRecorder } | undefined + +export function resetObservationRecorder(): void { + current = undefined +} + +/** Capture App ownership before await; callers must retain this recorder for their operation. */ +export function captureObservationRecorder(): ObservationRecorder | undefined { + const meta = configStore.metaConfig + if (!meta.INKCRE_PGREST_URL) return undefined + if ( + current?.url === meta.INKCRE_PGREST_URL && + current.secret === meta.INKCRE_JWT_SECRET && + current.recorder.peerId === meta.INKCRE_PEER_ID + ) + return current.recorder + if (pending.size >= 128) return undefined + const recorder = { client: createObservabilityClient(), peerId: meta.INKCRE_PEER_ID } + current = { url: meta.INKCRE_PGREST_URL, secret: meta.INKCRE_JWT_SECRET, recorder } + const preparation = recorder.client + .prepare() + .catch(() => { + if (!warned) console.warn('[Observability] Backend selection is unavailable.') + warned = true + }) + .finally(() => pending.delete(preparation)) + pending.add(preparation) + return recorder +} + +/** Diagnostic I/O is bounded and must not turn a completed business command into a failure. */ +export function recordObservation( + recorder: ObservationRecorder | undefined, + record: ObservationRecord +): void { + if (!recorder) return + if (pending.size >= 128) { + if (!warned) console.warn('[Observability] Recording queue is full; observations were dropped.') + warned = true + return + } + const completion = recorder.client + .emit({ + ...record, + attributes: { ...record.attributes, 'inkcre.peer.id': recorder.peerId }, + }) + .catch(() => { + if (!warned) + console.warn('[Observability] Recording failed; business work remains available.') + warned = true + }) + .finally(() => pending.delete(completion)) + pending.add(completion) +} + +export async function flushObservations(): Promise { + await Promise.allSettled(pending) +} diff --git a/packages/core/src/obsrv/telemetry.ts b/packages/core/src/obsrv/telemetry.ts index 9a0a61c1..f8411c3b 100644 --- a/packages/core/src/obsrv/telemetry.ts +++ b/packages/core/src/obsrv/telemetry.ts @@ -17,6 +17,13 @@ import type { MetaConfig } from '../config/schema' import type { LoggerProvider } from '@opentelemetry/sdk-logs' import type { MeterProvider } from '@opentelemetry/sdk-metrics' import type { WebTracerProvider } from '@opentelemetry/sdk-trace-web' +import { + captureObservationRecorder, + recordObservation, + flushObservations, + resetObservationRecorder, + type ObservationRecorder, +} from './recording' // Shared config is public to admitted browsers; never accept credential-bearing URLs. const PublicUrl = z.url().refine((value) => { @@ -57,6 +64,7 @@ let logger: ReturnType | undefined let duration: ReturnType['createHistogram']> | undefined let droppedCarrier: ReturnType['createCounter']> | undefined const operationLoggerKey = createContextKey('inkcre.operation.logger') +const observationRecorderKey = createContextKey('inkcre.operation.recorder') let initialization = Promise.resolve() /** Serialized connection cutover; disabled peers do not even read shared telemetry configuration. */ @@ -227,10 +235,13 @@ async function bounded(operation: Promise): Promise { } } export async function flushTelemetry(): Promise { + await bounded(flushObservations()) if (providers.length) await bounded(Promise.allSettled(providers.map((provider) => provider.forceFlush()))) } export async function shutdownTelemetry(): Promise { + await bounded(flushObservations()) + resetObservationRecorder() const closing = providers providers = [] tracer = undefined @@ -252,20 +263,46 @@ export async function observeOperation( parent: Context = context.active() ): Promise { const result: OperationResult = { outcome: 'success' } - if (!tracer) return operation(parent, undefined, result) - const span = tracer.startSpan(name, options, parent) + const recorder = + (parent.getValue(observationRecorderKey) as ObservationRecorder | undefined) ?? + captureObservationRecorder() + const span = tracer?.startSpan(name, options, parent) const operationDuration = duration const operationLogger = logger - const active = trace.setSpan(parent, span).setValue(operationLoggerKey, operationLogger) + const active = (span ? trace.setSpan(parent, span) : parent) + .setValue(operationLoggerKey, operationLogger) + .setValue(observationRecorderKey, recorder) const started = performance.now() try { return await operation(active, span, result) } catch (error) { result.outcome = 'error' - span.setStatus({ code: SpanStatusCode.ERROR }) + span?.setStatus({ code: SpanStatusCode.ERROR }) throw error } finally { - span.end() + span?.end() + const durationMs = performance.now() - started + const traceContext = trace.getSpanContext(active) + const failed = result.outcome === 'error' || result.outcome === 'timeout' + recordObservation(recorder, { + id: crypto.randomUUID(), + timestamp: new Date().toISOString(), + kind: 'span', + event_name: 'inkcre.operation.completed', + severity_number: failed ? 17 : 9, + severity_text: failed ? 'ERROR' : 'INFO', + body: 'inkcre.operation.completed', + trace_id: traceContext?.traceId ?? null, + span_id: traceContext?.spanId ?? null, + attributes: { + 'inkcre.operation': name, + duration_ms: durationMs, + outcome: result.outcome, + ...(typeof options.attributes?.['inkcre.job.id'] === 'number' + ? { 'inkcre.job.id': options.attributes['inkcre.job.id'] } + : {}), + }, + }) const attributes = { 'inkcre.operation': name, 'inkcre.outcome': result.outcome } operationDuration?.record((performance.now() - started) / 1000, { operation: name, @@ -319,6 +356,19 @@ export function emitJobEvent( parent: Context, status?: string ): void { + const traceContext = trace.getSpanContext(parent) + recordObservation(parent.getValue(observationRecorderKey) as ObservationRecorder | undefined, { + id: crypto.randomUUID(), + timestamp: new Date().toISOString(), + kind: 'event', + event_name: name, + severity_number: status === 'failed' || status === 'timed_out' ? 17 : 9, + severity_text: status === 'failed' || status === 'timed_out' ? 'ERROR' : 'INFO', + body: name, + trace_id: traceContext?.traceId ?? `job.${jobId}`, + span_id: traceContext?.spanId ?? null, + attributes: { 'inkcre.job.id': jobId, ...(status ? { 'inkcre.job.status': status } : {}) }, + }) const operationLogger = parent.getValue(operationLoggerKey) as typeof logger operationLogger?.emit({ context: parent, diff --git a/packages/core/src/sdk.ts b/packages/core/src/sdk.ts index bc2c529d..54f8222d 100644 --- a/packages/core/src/sdk.ts +++ b/packages/core/src/sdk.ts @@ -10,3 +10,5 @@ export * from './database' export * from './peer/contracts' export * from './info-base/resolvers/contracts' export { PostgreSQLBlobStorage, type PostgreSQLBlobPointer } from './info-base/storages/blob' +export * from './obsrv/contracts' +export { ObservabilityClient } from './obsrv/client' diff --git a/scripts/tests/observability.test.mjs b/scripts/tests/observability.test.mjs new file mode 100644 index 00000000..6548afc8 --- /dev/null +++ b/scripts/tests/observability.test.mjs @@ -0,0 +1,362 @@ +import assert from 'node:assert/strict' +import { test } from 'node:test' +import { + ObservabilityClient, + ObservabilityUnavailableError, +} from '../../packages/core/dist/sdk/sdk.js' + +const range = { since: '2026-11-04T00:00:00Z', until: '2026-11-04T23:59:59Z' } +const record = { + id: 'observation-1', + timestamp: '2026-11-04T10:00:00Z', + kind: 'event', + event_name: 'job.closed', + severity_number: 9, + severity_text: 'INFO', + body: 'job.closed', + trace_id: 'job.8', + span_id: null, + attributes: { 'inkcre.job.id': 8 }, +} +const selector = { schema: 'inkcre.observability.backend.v1', value: { backend: 'loki' } } +const peer = { + id: '00000000-0000-4000-8000-000000000001', + capabilities: [ + { + id: 'core.observability.v1', + inbound: { + protocol: 'core.peer.protocol.http.v1', + parameters: { method: 'POST', url: 'https://peer.example/observability' }, + }, + }, + ], +} + +void test('default database query is bounded, preserves legacy records and captures connection', async () => { + const calls = [] + const connection = { + baseUrl: 'https://a.example', + tokenProvider: async () => 'token-a', + fetch: async (input, init) => { + const url = new URL(input) + calls.push({ url, headers: new Headers(init.headers) }) + if (url.pathname === '/configs') return Response.json(null) + return Response.json([ + { + ...record, + id: 7, + kind: undefined, + event_name: undefined, + body: 'x'.repeat(8193), + attributes: { job_id: '8', legacy: { nested: ['metadata'] } }, + }, + { ...record, id: 8 }, + ]) + }, + } + const client = new ObservabilityClient(connection) + connection.baseUrl = 'https://b.example' + connection.tokenProvider = async () => 'token-b' + const result = await client.query({ + ...range, + limit: 1, + job_id: 8, + kind: 'log', + text: 'literal.*(x)', + severity_min: 9, + }) + assert.equal(result.backend, 'postgresql') + assert.equal(result.records[0].id, 'pg.7') + assert.equal(result.records[0].kind, 'log') + assert.equal(result.records[0].body.length, 8193) + assert.deepEqual(result.records[0].attributes.legacy, { nested: ['metadata'] }) + assert.equal(result.truncated, true) + const logCall = calls[1] + assert.equal(logCall.url.origin, 'https://a.example') + assert.equal(logCall.headers.get('Authorization'), 'Bearer token-a') + assert.equal(logCall.url.searchParams.get('limit'), '2') + assert.equal(logCall.url.searchParams.get('body'), 'imatch.literal\\.\\*\\(x\\)') + assert.equal(logCall.url.searchParams.get('severity_number'), 'gte.9') + assert.deepEqual(logCall.url.searchParams.getAll('timestamp'), [ + `gte.${range.since}`, + `lt.${range.until}`, + ]) + assert.ok( + logCall.url.searchParams + .getAll('or') + .some((value) => value.includes('attributes->>"inkcre.record.kind"')) + ) + assert.ok( + logCall.url.searchParams + .getAll('or') + .includes('(trace_id.eq.job.8,attributes->>"inkcre.job.id".eq.8,attributes->>job_id.eq.8)') + ) +}) + +void test('selected external backend delegates query and writes using Peer authority, never database logs', async () => { + const calls = [] + const client = new ObservabilityClient({ + baseUrl: 'https://db.example', + tokenProvider: async () => 'peer-token', + fetch: async (input, init) => { + const url = new URL(input) + calls.push(url) + if (url.pathname === '/configs') return Response.json(selector) + if (url.pathname === '/peers') return Response.json([peer]) + assert.equal(url.href, 'https://peer.example/observability') + assert.equal(new Headers(init.headers).get('Authorization'), 'Bearer peer-token') + assert.equal(init.redirect, 'error') + const body = JSON.parse(init.body) + if (body.action === 'ingest') { + assert.deepEqual(body.records, [record]) + return Response.json({ accepted: 1, dropped: 0 }) + } + if (body.action === 'status') + return Response.json({ + backend: 'loki', + configured: true, + recording: true, + queryable: true, + scope: 'selected_backend', + history: 'Previous backend data remains there.', + dropped: null, + error: null, + }) + return Response.json({ ...range, records: [record], truncated: false, backend: 'loki' }) + }, + }) + assert.deepEqual((await client.query(range)).records, [record]) + assert.equal((await client.status()).dropped, null) + await client.emit(record) + assert.ok(calls.every((url) => url.pathname !== '/logs')) +}) + +void test('unknown selection, missing capability, external failure and backend disagreement never fall back', async () => { + for (const failure of ['selection', 'capability', 'request', 'mismatch']) { + let queriedDatabase = false + const client = new ObservabilityClient({ + baseUrl: 'https://db.example', + tokenProvider: async () => 'token', + fetch: async (input) => { + const url = new URL(input) + if (url.pathname === '/configs') + return Response.json( + failure === 'selection' ? { schema: 'future.v2', value: {} } : selector + ) + if (url.pathname === '/peers') return Response.json(failure === 'capability' ? [] : [peer]) + if (url.pathname === '/logs') { + queriedDatabase = true + return Response.json([]) + } + return failure === 'request' + ? new Response('unavailable', { status: 503 }) + : Response.json({ ...range, records: [], truncated: false, backend: 'postgresql' }) + }, + }) + await assert.rejects(client.query(range), ObservabilityUnavailableError) + assert.equal(queriedDatabase, false) + } +}) + +void test('database ingestion preserves portable identity, kind and event metadata', async () => { + let stored + const client = new ObservabilityClient({ + baseUrl: 'https://db.example', + tokenProvider: async () => 'token', + fetch: async (input, init) => { + if (new URL(input).pathname === '/configs') return Response.json(null) + stored = JSON.parse(init.body) + return new Response(null, { status: 201 }) + }, + }) + await client.emit(record) + assert.equal(stored.attributes['inkcre.record.id'], record.id) + assert.equal(stored.attributes['inkcre.record.kind'], 'event') + assert.equal(stored.attributes['inkcre.event.name'], 'job.closed') + assert.equal(stored.id, undefined) + await assert.rejects(client.emit({ ...record, body: 'x'.repeat(8193) })) + for (const [key, length] of [ + ['event_name', 128], + ['severity_text', 32], + ['trace_id', 128], + ['span_id', 128], + ]) + await assert.rejects(client.emit({ ...record, [key]: 'x'.repeat(length + 1) })) + await assert.rejects(client.emit({ ...record, attributes: { nested: { value: 1 } } })) + await assert.rejects(client.emit({ ...record, attributes: { large: 'x'.repeat(1025) } })) + await assert.rejects(client.emit({ ...record, attributes: { large: Array(33).fill(1) } })) + assert.equal(stored.body, record.body) + await assert.rejects(client.query({ ...range, until: range.since })) + await assert.rejects(client.query({ ...range, limit: 501 })) +}) + +void test('an initialized instance keeps its selected backend until a new instance is created', async () => { + let selected = null + let reads = 0 + const connection = { + baseUrl: 'https://db.example', + tokenProvider: async () => 'token', + fetch: async (input) => { + const url = new URL(input) + if (url.pathname === '/configs') { + reads++ + return Response.json(selected) + } + if (url.pathname === '/logs') return Response.json([]) + if (url.pathname === '/peers') return Response.json([]) + throw new Error('Unexpected call') + }, + } + const old = new ObservabilityClient(connection) + await old.prepare() + selected = selector + assert.equal((await old.query(range)).backend, 'postgresql') + assert.equal(reads, 1) + await assert.rejects( + new ObservabilityClient(connection).query(range), + ObservabilityUnavailableError + ) + await assert.rejects(old.query({ since: '2026-01-01T00:00:00Z', until: '2026-11-01T00:00:00Z' })) +}) + +void test('a not-ready Peer is skipped before dispatch and writes carry atomic backend expectation', async () => { + const stale = { + ...peer, + id: '00000000-0000-4000-8000-000000000000', + capabilities: [ + { + ...peer.capabilities[0], + inbound: { + ...peer.capabilities[0].inbound, + parameters: { method: 'POST', url: 'https://stale.example/observability' }, + }, + }, + ], + } + let stored = false + const client = new ObservabilityClient({ + baseUrl: 'https://db.example', + tokenProvider: async () => 'token', + fetch: async (input, init) => { + const url = new URL(input) + if (url.pathname === '/configs') return Response.json(selector) + if (url.pathname === '/peers') return Response.json([stale, peer]) + const body = JSON.parse(init.body) + if (body.action === 'status') + return Response.json({ + backend: 'loki', + configured: url.hostname !== 'stale.example', + recording: url.hostname !== 'stale.example', + queryable: url.hostname !== 'stale.example', + scope: 'selected_backend', + history: '', + dropped: null, + error: null, + }) + assert.equal(url.hostname, 'peer.example') + assert.equal(body.expected_backend, 'loki') + stored = true + return Response.json({ accepted: 1, dropped: 0 }) + }, + }) + await client.emit(record) + assert.equal(stored, true) +}) + +void test('Job lifecycle and terminal severities are recorded without optional OTEL or raw content', async () => { + const { JobManager, configStore, flushTelemetry } = + await import('../../packages/core/dist/index.js') + const originalFetch = globalThis.fetch + const priorMeta = { ...configStore.metaConfig } + const dbJob = { + id: 8, + type: 'test.observation', + parameters: { private_input: 'must-not-be-recorded' }, + state: {}, + abort_requested: false, + timeout_seconds: 30, + status: 'pending', + created_at: range.since, + started_at: null, + closed_at: null, + } + const records = [] + let configReads = 0 + let outcome = 'finished' + globalThis.fetch = async (input, init) => { + const url = new URL(input) + if (url.pathname === '/configs') { + configReads++ + return Response.json(null) + } + if (url.pathname === '/logs') { + records.push(JSON.parse(init.body)) + return new Response(null, { status: 201 }) + } + if (url.pathname === '/jobs') { + if (init.method === 'PATCH') Object.assign(dbJob, JSON.parse(init.body)) + const singleton = new Headers(init.headers).get('Accept')?.includes('object+json') + return Response.json(singleton ? dbJob : [dbJob]) + } + throw new Error(`Unexpected mocked request ${url.pathname}`) + } + Object.assign(configStore.metaConfig, { + INKCRE_PGREST_URL: 'https://test.example', + INKCRE_JWT_SECRET: 'test-only-observation-secret', + telemetry_enabled: false, + }) + await new Promise((resolve) => queueMicrotask(resolve)) + JobManager.registerHandler('test.observation', { + parameters: { parse: (value) => value }, + canHandle: () => true, + handle: async (job, _parameters, signal) => { + job.state = { count: 11 } + if (outcome === 'timed_out') + await new Promise((resolve) => signal.addEventListener('abort', resolve, { once: true })) + if (outcome === 'aborted') { + dbJob.abort_requested = true + await new Promise((resolve) => setImmediate(resolve)) + await JobManager.checkAbortRequests() + } + }, + }) + try { + assert.equal(await JobManager.run(8), true) + await flushTelemetry() + assert.equal(dbJob.status, 'finished') + const events = records.filter((item) => + item.attributes?.['inkcre.event.name']?.startsWith('job.') + ) + assert.deepEqual( + events.map((item) => item.attributes['inkcre.event.name']).sort((a, b) => a.localeCompare(b)), + ['job.closed', 'job.started'] + ) + assert.ok( + events.every((item) => item.severity_number === 9 && item.attributes['inkcre.job.id'] === 8) + ) + assert.equal(JSON.stringify(records).includes('must-not-be-recorded'), false) + for (const [terminalStatus, severity, expectedOutcome] of [ + ['timed_out', 17, 'timeout'], + ['aborted', 9, 'cancelled'], + ]) { + outcome = terminalStatus + records.length = 0 + Object.assign(dbJob, { status: 'pending', abort_requested: false, timeout_seconds: 1 }) + assert.equal(await JobManager.run(8), true) + await flushTelemetry() + assert.equal(dbJob.status, terminalStatus) + const closed = records.find((item) => item.attributes['inkcre.event.name'] === 'job.closed') + const execution = records.find( + (item) => item.attributes['inkcre.operation'] === 'job.execute' + ) + assert.equal(closed.severity_number, severity) + assert.equal(execution.severity_number, severity) + assert.equal(execution.attributes.outcome, expectedOutcome) + } + assert.equal(configReads, 1, 'all operations share one captured backend selection') + } finally { + JobManager.unregisterHandler('test.observation') + Object.assign(configStore.metaConfig, priorMeta) + globalThis.fetch = originalFetch + } +}) diff --git a/tests/e2e/peer-database.spec.ts b/tests/e2e/peer-database.spec.ts index 1b1584ff..593d5a69 100644 --- a/tests/e2e/peer-database.spec.ts +++ b/tests/e2e/peer-database.spec.ts @@ -1903,14 +1903,17 @@ test('Job preserves readable errors and logs and distinguishes failed and missin } else await route.continue() }) await page.getByRole('button', { name: 'Retry', exact: true }).click() - const logs = page.locator('.logs-viewer') + const logs = page.getByRole('region', { name: 'Observability', exact: true }) await expect(logs.getByText('Existing log survives failure')).toBeVisible() await expect(logs.getByText('Updating automatically')).toBeVisible() await expect(logs.getByRole('status')).toHaveCount(0) await expect.poll(() => logReads).toBe(2) await expect(logs.getByRole('status', { name: 'Loading logs…' })).toBeVisible() releaseLogs() - await expect(logs.getByText('Unable to load logs.')).toBeVisible() + await expect(logs.getByText(/^Observation query failed\./)).toBeVisible() + await expect( + logs.getByText('Previously returned records remain below and have not been refreshed.') + ).toBeVisible() await expect(logs.getByText('Existing log survives failure')).toBeVisible() await expect(logs.getByRole('status')).toHaveCount(0) rejectLogs = false