diff --git a/.env.example b/.env.example index da9afc2..df71964 100644 --- a/.env.example +++ b/.env.example @@ -12,3 +12,11 @@ RPC_URL_8453=https://example-rpc RPC_URL_42161=https://example-rpc RPC_URL_80094=https://example-rpc RPC_URL_747474=https://example-rpc +# OpenTelemetry error reporting. Signal-specific values take precedence. +# Sentry provides the complete OTLP logs endpoint and auth header under Project Settings > Client Keys (DSN). +OTEL_EXPORTER_OTLP_LOGS_ENDPOINT= +OTEL_EXPORTER_OTLP_LOGS_HEADERS= +# Generic OTLP base URL. The worker appends /v1/logs. +OTEL_EXPORTER_OTLP_ENDPOINT= +OTEL_EXPORTER_OTLP_HEADERS= +OTEL_SERVICE_NAME=yearn-prices diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index ddbd925..c666148 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -38,6 +38,11 @@ jobs: RPC_URL_42161: op://webops-prod/yearn-price/RPC_URL_42161 RPC_URL_80094: op://webops-prod/yearn-price/RPC_URL_80094 RPC_URL_747474: op://webops-prod/yearn-price/RPC_URL_747474 + OTEL_EXPORTER_OTLP_ENDPOINT: op://webops-prod/yearn-price/OTEL_EXPORTER_OTLP_ENDPOINT + OTEL_EXPORTER_OTLP_HEADERS: op://webops-prod/yearn-price/OTEL_EXPORTER_OTLP_HEADERS + OTEL_EXPORTER_OTLP_LOGS_ENDPOINT: op://webops-prod/yearn-price/OTEL_EXPORTER_OTLP_LOGS_ENDPOINT + OTEL_EXPORTER_OTLP_LOGS_HEADERS: op://webops-prod/yearn-price/OTEL_EXPORTER_OTLP_LOGS_HEADERS + OTEL_SERVICE_NAME: op://webops-prod/yearn-price/OTEL_SERVICE_NAME - name: Upload secrets to Cloudflare run: | jq -n '{ @@ -58,7 +63,12 @@ jobs: RPC_URL_8453: env.RPC_URL_8453, RPC_URL_42161: env.RPC_URL_42161, RPC_URL_80094: env.RPC_URL_80094, - RPC_URL_747474: env.RPC_URL_747474 + RPC_URL_747474: env.RPC_URL_747474, + OTEL_EXPORTER_OTLP_ENDPOINT: env.OTEL_EXPORTER_OTLP_ENDPOINT, + OTEL_EXPORTER_OTLP_HEADERS: env.OTEL_EXPORTER_OTLP_HEADERS, + OTEL_EXPORTER_OTLP_LOGS_ENDPOINT: env.OTEL_EXPORTER_OTLP_LOGS_ENDPOINT, + OTEL_EXPORTER_OTLP_LOGS_HEADERS: env.OTEL_EXPORTER_OTLP_LOGS_HEADERS, + OTEL_SERVICE_NAME: env.OTEL_SERVICE_NAME }' | bunx wrangler secret bulk - name: Run migrations run: bun run migrate diff --git a/src/cache/edge.ts b/src/cache/edge.ts index d57e602..91d9404 100644 --- a/src/cache/edge.ts +++ b/src/cache/edge.ts @@ -1,3 +1,6 @@ +import { captureError } from '../observability' +import type { Env } from '../types' + // Cloudflare edge caching via the Cache API. // // A worker-generated Response with a Cache-Control header only drives the @@ -66,7 +69,12 @@ export async function readEdgeCache(request: Request): Promise { + captureError(ctx, env, error) + }), + ) } diff --git a/src/index.ts b/src/index.ts index 89ef781..cc680f2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -9,6 +9,7 @@ import { withCors, } from './http' import { renderLandingPage } from './lander' +import { captureError } from './observability' import { handleHealth } from './routes/health' import { handleBatchHistorical } from './routes/historical/batch' import { handleHistorical } from './routes/historical/exact' @@ -95,7 +96,7 @@ export default { const response = await routePriceRequest(request, env, pathname) if (request.method === 'GET') { - writeEdgeCache(ctx, request, response) + writeEdgeCache(ctx, env, request, response) } return response } catch (error) { @@ -114,6 +115,7 @@ export default { const headers = error.code === 'NOT_FOUND' && notFoundCacheable ? withCors(notFoundErrorHeaders()) : withCors({ 'cache-control': CACHE_CONTROL_NO_STORE }) + if (error.status >= 500) captureError(ctx, env, error) return jsonError(error, headers) } @@ -125,6 +127,7 @@ export default { error: error instanceof Error ? error.message : String(error), }), ) + captureError(ctx, env, error) return jsonError(new ApiError('INTERNAL_ERROR', 'Unexpected internal error'), withCors({ 'cache-control': CACHE_CONTROL_NO_STORE })) } }, diff --git a/src/observability.ts b/src/observability.ts new file mode 100644 index 0000000..d774701 --- /dev/null +++ b/src/observability.ts @@ -0,0 +1,95 @@ +import type { Env } from './types' + +// Cloudflare Workers can't run the OpenTelemetry Node SDK, so errors are sent as +// OTLP/HTTP JSON log records via fetch. Vendor-neutral: point +// OTEL_EXPORTER_OTLP_ENDPOINT at any OTLP backend (Sentry OTLP, Grafana, a Collector, ...). +const SERVICE_NAME = 'yearn-prices' +const SEVERITY_ERROR = 17 // OTLP severityNumber for ERROR + +type OtlpAttribute = { key: string; value: { stringValue: string } } + +function attr(key: string, value: string): OtlpAttribute { + return { key, value: { stringValue: value } } +} + +function decodeHeaderPart(value: string): string { + try { + return decodeURIComponent(value) + } catch { + return value + } +} + +// OTEL_EXPORTER_OTLP_HEADERS format: "key1=value1,key2=value2". +function parseHeaders(raw?: string): Record { + const headers: Record = { 'content-type': 'application/json' } + if (!raw) return headers + for (const pair of raw.split(',')) { + const idx = pair.indexOf('=') + if (idx > 0) { + headers[decodeHeaderPart(pair.slice(0, idx).trim())] = decodeHeaderPart( + pair.slice(idx + 1).trim(), + ) + } + } + return headers +} + +function buildPayload(serviceName: string, err: Error): unknown { + const attributes = [attr('exception.type', err.name), attr('exception.message', err.message)] + if (err.stack) attributes.push(attr('exception.stacktrace', err.stack)) + + return { + resourceLogs: [ + { + resource: { attributes: [attr('service.name', serviceName)] }, + scopeLogs: [ + { + scope: { name: serviceName }, + logRecords: [ + { + timeUnixNano: String(BigInt(Date.now()) * 1_000_000n), + severityNumber: SEVERITY_ERROR, + severityText: 'ERROR', + body: { stringValue: err.message }, + attributes, + }, + ], + }, + ], + }, + ], + } +} + +export function captureError(ctx: ExecutionContext, env: Env, error: unknown): void { + const logsEndpoint = env.OTEL_EXPORTER_OTLP_LOGS_ENDPOINT + const endpoint = logsEndpoint || env.OTEL_EXPORTER_OTLP_ENDPOINT + if (!endpoint) return + + const err = error instanceof Error ? error : new Error(String(error)) + const normalizedEndpoint = endpoint.replace(/\/$/, '') + const url = logsEndpoint ? normalizedEndpoint : `${normalizedEndpoint}/v1/logs` + const body = JSON.stringify(buildPayload(env.OTEL_SERVICE_NAME || SERVICE_NAME, err)) + + // waitUntil lets the export finish after the response is returned (no added latency). + ctx.waitUntil( + fetch(url, { + method: 'POST', + headers: parseHeaders(env.OTEL_EXPORTER_OTLP_LOGS_HEADERS || env.OTEL_EXPORTER_OTLP_HEADERS), + body, + }).then((response) => { + if (!response.ok) { + console.error(JSON.stringify({ + message: 'otel-export-error', + status: response.status, + })) + } + }).catch((exportError) => { + console.error(JSON.stringify({ + message: 'otel-export-error', + error: exportError instanceof Error ? exportError.message : String(exportError), + })) + }), + ) +} diff --git a/src/types.ts b/src/types.ts index f3c02e4..ec715ad 100644 --- a/src/types.ts +++ b/src/types.ts @@ -12,6 +12,11 @@ export type PriceSource = (typeof SOURCE_PRIORITY)[number] export interface Env { DATABASE_URL: string ENSO_API_KEY?: string + OTEL_EXPORTER_OTLP_ENDPOINT?: string + OTEL_EXPORTER_OTLP_HEADERS?: string + OTEL_EXPORTER_OTLP_LOGS_ENDPOINT?: string + OTEL_EXPORTER_OTLP_LOGS_HEADERS?: string + OTEL_SERVICE_NAME?: string [key: string]: string | undefined } diff --git a/test/edge-cache.test.ts b/test/edge-cache.test.ts index 9bf89b2..be6ba8d 100644 --- a/test/edge-cache.test.ts +++ b/test/edge-cache.test.ts @@ -10,10 +10,13 @@ import { readEdgeCache, writeEdgeCache, } from '../src/cache' +import type { Env } from '../src/types' import { normalizeToEndOfDay } from '../src/utils' const BASE = 'https://svc/api/prices/spot' +const ENV = { DATABASE_URL: 'postgres://x' } satisfies Env + function spotUrl(coins: unknown): string { return `${BASE}?coins=${encodeURIComponent(JSON.stringify(coins))}` } @@ -102,7 +105,7 @@ describe('readEdgeCache / writeEdgeCache', () => { const ctx = { waitUntil: vi.fn() } as unknown as ExecutionContext const written = new Request(spotUrl(['Ethereum:0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2'])) - writeEdgeCache(ctx, written, new Response('{"coins":{}}')) + writeEdgeCache(ctx, ENV, written, new Response('{"coins":{}}')) await (ctx.waitUntil as any).mock.calls[0][0] expect([...store.keys()]).toEqual([canonicalCacheKey(written.url)]) @@ -123,7 +126,7 @@ describe('readEdgeCache / writeEdgeCache', () => { const request = new Request(spotUrl([TOKEN])) const response = new Response('{"coins":{}}') - writeEdgeCache(ctx, request, response) + writeEdgeCache(ctx, ENV, request, response) await (ctx.waitUntil as any).mock.calls[0][0] await expect(response.text()).resolves.toBe('{"coins":{}}') diff --git a/test/observability.test.ts b/test/observability.test.ts new file mode 100644 index 0000000..85cc96b --- /dev/null +++ b/test/observability.test.ts @@ -0,0 +1,136 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { captureError } from '../src/observability' +import type { Env } from '../src/types' + +function context(): ExecutionContext { + return { waitUntil: vi.fn() } as unknown as ExecutionContext +} + +async function pending(ctx: ExecutionContext): Promise { + await (ctx.waitUntil as ReturnType).mock.calls[0][0] +} + +describe('captureError', () => { + afterEach(() => { + vi.unstubAllGlobals() + vi.restoreAllMocks() + }) + + it('uses a signal-specific Sentry endpoint without appending another path', async () => { + const fetchMock = vi.fn().mockResolvedValue(new Response(null, { status: 200 })) + vi.stubGlobal('fetch', fetchMock) + const ctx = context() + const env = { + DATABASE_URL: 'postgres://x', + OTEL_EXPORTER_OTLP_LOGS_ENDPOINT: 'https://o0.ingest.sentry.io/api/0/integration/otlp/v1/logs', + OTEL_EXPORTER_OTLP_LOGS_HEADERS: 'x-sentry-auth=sentry sentry_key=public-key', + } satisfies Env + + captureError(ctx, env, new Error('boom')) + await pending(ctx) + + expect(fetchMock).toHaveBeenCalledWith(env.OTEL_EXPORTER_OTLP_LOGS_ENDPOINT, expect.objectContaining({ + headers: expect.objectContaining({ 'x-sentry-auth': 'sentry sentry_key=public-key' }), + })) + }) + + it('appends the logs path to the generic OTLP base endpoint', async () => { + const fetchMock = vi.fn().mockResolvedValue(new Response(null, { status: 200 })) + vi.stubGlobal('fetch', fetchMock) + const ctx = context() + + captureError(ctx, { + DATABASE_URL: 'postgres://x', + OTEL_EXPORTER_OTLP_ENDPOINT: 'https://collector.example.test/', + }, new Error('boom')) + await pending(ctx) + + expect(fetchMock.mock.calls[0][0]).toBe('https://collector.example.test/v1/logs') + const payload = JSON.parse(fetchMock.mock.calls[0][1].body as string) + expect(payload.resourceLogs[0].scopeLogs[0].logRecords[0].attributes).toEqual( + expect.arrayContaining([ + { key: 'exception.type', value: { stringValue: 'Error' } }, + { key: 'exception.message', value: { stringValue: 'boom' } }, + ]), + ) + }) + + it('reports rejected exports without throwing into the request path', async () => { + vi.stubGlobal('fetch', vi.fn().mockRejectedValue(new Error('socket hang up'))) + const consoleError = vi.spyOn(console, 'error').mockImplementation(() => {}) + const ctx = context() + + captureError(ctx, { + DATABASE_URL: 'postgres://x', + OTEL_EXPORTER_OTLP_ENDPOINT: 'https://collector.example.test', + }, new Error('boom')) + await pending(ctx) + + expect(consoleError).toHaveBeenCalledWith(expect.stringContaining('otel-export-error')) + expect(consoleError).toHaveBeenCalledWith(expect.stringContaining('socket hang up')) + }) + + it('logs non-2xx export responses without throwing into the request path', async () => { + vi.stubGlobal('fetch', vi.fn().mockResolvedValue(new Response(null, { status: 401 }))) + const consoleError = vi.spyOn(console, 'error').mockImplementation(() => {}) + const ctx = context() + + captureError(ctx, { + DATABASE_URL: 'postgres://x', + OTEL_EXPORTER_OTLP_ENDPOINT: 'https://collector.example.test', + }, new Error('boom')) + await pending(ctx) + + expect(consoleError).toHaveBeenCalledWith(expect.stringContaining('otel-export-error')) + expect(consoleError).toHaveBeenCalledWith(expect.stringContaining('401')) + }) + + it('is a no-op when no OTLP endpoint is configured', () => { + const fetchMock = vi.fn() + vi.stubGlobal('fetch', fetchMock) + const ctx = context() + + captureError(ctx, { DATABASE_URL: 'postgres://x' }, new Error('boom')) + + expect(ctx.waitUntil).not.toHaveBeenCalled() + expect(fetchMock).not.toHaveBeenCalled() + }) + + it('percent-decodes OTLP header values and records an exact unix-nano timestamp', async () => { + const fetchMock = vi.fn().mockResolvedValue(new Response(null, { status: 200 })) + vi.stubGlobal('fetch', fetchMock) + vi.spyOn(Date, 'now').mockReturnValue(1_700_000_000_123) + const ctx = context() + + captureError(ctx, { + DATABASE_URL: 'postgres://x', + OTEL_EXPORTER_OTLP_ENDPOINT: 'https://collector.example.test', + OTEL_EXPORTER_OTLP_HEADERS: 'x-custom=foo%2Cbar', + }, new Error('boom')) + await pending(ctx) + + expect(fetchMock.mock.calls[0][1].headers['x-custom']).toBe('foo,bar') + const payload = JSON.parse(fetchMock.mock.calls[0][1].body as string) + expect(payload.resourceLogs[0].scopeLogs[0].logRecords[0].timeUnixNano).toBe( + String(1_700_000_000_123n * 1_000_000n), + ) + }) + + it('does not throw when OTLP headers contain a raw percent', async () => { + const fetchMock = vi.fn().mockResolvedValue(new Response(null, { status: 200 })) + vi.stubGlobal('fetch', fetchMock) + const ctx = context() + + expect(() => { + captureError(ctx, { + DATABASE_URL: 'postgres://x', + OTEL_EXPORTER_OTLP_ENDPOINT: 'https://collector.example.test', + OTEL_EXPORTER_OTLP_HEADERS: 'authorization=Api-Key 50%off', + }, new Error('boom')) + }).not.toThrow() + + await pending(ctx) + + expect(fetchMock.mock.calls[0][1].headers['authorization']).toBe('Api-Key 50%off') + }) +})