feat(telemetry): core-node observability foundation — OTLP logs + traces + metrics

packages/cli/src/config.ts

@@ -538,8 +538,65 @@ export interface DkgConfig {
    * on first start and stored in `<DKG_HOME>/auth.token`.
    */
   auth?: { enabled?: boolean; tokens?: string[] };
-  /** Opt-in telemetry streaming to central network dashboard. */
-  telemetry?: { enabled?: boolean };
+  /**
+   * Opt-in telemetry streaming to a central network dashboard.
+   * `enabled` is the master gate: when false, NOTHING is forwarded off the
+   * node (local logging — SQLite + daemon.log — is always on regardless).
+   */
+  telemetry?: {
+    enabled?: boolean;
+    /**
+     * Remote log forwarding (opt-in). Active only when `enabled` is true.
+     */
+    logs?: {
+      /**
+       * Outbound transport for logs. 'none' = local only; 'otlp' = OTLP/HTTP
+       * to an OpenTelemetry collector; 'syslog' = legacy RFC 5424 → Graylog.
+       * Defaults to 'syslog' when unset (preserves prior behaviour).
+       */
+      exporter?: 'none' | 'otlp' | 'syslog';
+      /**
+       * OTLP/HTTP logs endpoint, e.g. http://localhost:4318/v1/logs. Falls
+       * back to the per-network default (TELEMETRY_ENDPOINTS[network].otlpLogs).
+       */
+      endpoint?: string;
+      /** Bearer credential for the operator's collector. Treated as a secret. */
+      token?: string;
+      /** Minimum level forwarded remotely. Local sink keeps everything. Default 'info'. */
+      level?: 'debug' | 'info' | 'warn' | 'error';
+      /** Extra sensitive key names to redact from messages before they leave the node. */
+      redact?: string[];
+      /** Bounded in-memory buffer; drop-oldest on overflow. Default 500. */
+      bufferMaxEntries?: number;
+    };
+    /**
+     * OTel trace export (opt-in, independent of logs). Registers the tracer
+     * ONLY when an endpoint resolves (config or OTEL_EXPORTER_OTLP_* env);
+     * never falls back to a guessed prod URL.
+     */
+    traces?: {
+      enabled?: boolean;
+      /** OTLP traces endpoint, e.g. http://localhost:4318/v1/traces. */
+      endpoint?: string;
+      /** Bearer credential. Treated as a secret. */
+      token?: string;
+      /** Parent-based ratio sampler 0..1. Default 1.0. */
+      sampleRatio?: number;
+    };
+    /**
+     * OTel metric export (opt-in, independent of logs). Registers the meter
+     * ONLY when an endpoint resolves (config or OTEL_EXPORTER_OTLP_* env).
+     */
+    metrics?: {
+      enabled?: boolean;
+      /** OTLP metrics endpoint, e.g. http://localhost:4318/v1/metrics. */
+      endpoint?: string;
+      /** Bearer credential. Treated as a secret. */
+      token?: string;
+      /** PeriodicExportingMetricReader interval. Default 30000ms. */
+      exportIntervalMs?: number;
+    };
+  };
   /** Shared memory (workspace) data TTL in milliseconds. Default: 30 days (2592000000). Set to 0 to disable cleanup. */
   sharedMemoryTtlMs?: number;
   /** @deprecated Legacy alias for sharedMemoryTtlMs */
@@ -736,14 +793,19 @@ export interface DkgConfig {
  * Nodes resolve the correct endpoints from the network they're on.
  * Operators only see a single toggle — no endpoint configuration.
  */
-export const TELEMETRY_ENDPOINTS: Record<string, { syslog: { host: string; port: number }; otlp: string }> = {
+export const TELEMETRY_ENDPOINTS: Record<
+  string,
+  { syslog: { host: string; port: number }; otlp: string; otlpLogs?: string }
+> = {
   testnet: {
     syslog: { host: 'loggly.origin-trail.network', port: 12201 },
     otlp: 'https://telemetry-testnet.origintrail.io/v1/metrics',
+    otlpLogs: 'https://telemetry-testnet.origintrail.io/v1/logs', // OriginTrail-hosted opt-in collector (TBD)
   },
   mainnet: {
-    syslog: { host: 'loggly.origin-trail.network', port: 0 }, // TODO: assign mainnet syslog port
+    syslog: { host: 'loggly.origin-trail.network', port: 0 }, // legacy syslog — OTLP is the mainnet path
     otlp: 'https://telemetry.origintrail.io/v1/metrics',
+    otlpLogs: 'https://telemetry.origintrail.io/v1/logs', // OriginTrail-hosted opt-in collector (TBD)
   },
 };
 

packages/cli/src/daemon/lifecycle.ts

@@ -70,7 +70,7 @@ import {
 } from '@origintrail-official/dkg-chain';
 import { DKGAgent, loadOpWallets, KaNumberAllocator } from '@origintrail-official/dkg-agent';
 import { isExternalBackend } from '@origintrail-official/dkg-storage';
-import { computeNetworkId, createOperationContext, DKGEvent, Logger, PayloadTooLargeError, GET_VIEWS, TrustLevel, validateSubGraphName, validateAssertionName, validateContextGraphId, isSafeIri, assertSafeIri, sparqlIri, contextGraphSharedMemoryUri, contextGraphAssertionUri, contextGraphMetaUri, DEFAULT_PROTOCOL_OUTBOX_BACKOFFS_MS, DEFAULT_PROTOCOL_OUTBOX_MAX_AGE_MS, pickNetworkTunables } from '@origintrail-official/dkg-core';
+import { computeNetworkId, createOperationContext, createLogRedactor, DKGEvent, Logger, PayloadTooLargeError, GET_VIEWS, TrustLevel, validateSubGraphName, validateAssertionName, validateContextGraphId, isSafeIri, assertSafeIri, sparqlIri, contextGraphSharedMemoryUri, contextGraphAssertionUri, contextGraphMetaUri, DEFAULT_PROTOCOL_OUTBOX_BACKOFFS_MS, DEFAULT_PROTOCOL_OUTBOX_MAX_AGE_MS, pickNetworkTunables } from '@origintrail-official/dkg-core';
 import { findReservedSubjectPrefix, isSkolemizedUri } from '@origintrail-official/dkg-publisher';
 import {
   DashboardDB,
@@ -79,6 +79,9 @@ import {
   handleNodeUIRequest,
   ChatMemoryManager,
   LogPushWorker,
+  OtlpLogWorker,
+  initTelemetry,
+  shutdownTelemetry,
   LlmClient,
   SqliteMessageIdempotencyStore,
   SqliteProtocolOutboxStore,
@@ -1966,6 +1969,11 @@ export async function runDaemonInner(
   chatDb = dashDb;
   log("Dashboard DB initialized at " + join(dkgDir(), "node-ui.db"));
 
+  // Redactor for the copy of each record that LEAVES the node. The local
+  // dashboard DB keeps full-fidelity records (it's the operator's own machine);
+  // redaction only protects data crossing the trust boundary to a collector.
+  const redactForRemote = createLogRedactor(config.telemetry?.logs?.redact);
+
   Logger.setSink((entry) => {
     try {
       dashDb.insertLog({
@@ -1979,7 +1987,12 @@ export async function runDaemonInner(
     } catch {
       /* DB write must never break the node */
     }
-    logPusher?.push(entry);
+    // Fan out a single redacted copy to every active remote shipper.
+    if (logPusher || otlpExporter) {
+      const safe = redactForRemote(entry);
+      logPusher?.push(safe);
+      otlpExporter?.push(safe);
+    }
   });
 
   // Extract the plain value from an RDF typed literal like "6"^^<xsd:integer>
@@ -2149,6 +2162,7 @@ export async function runDaemonInner(
     : "mainnet";
   const syslogEndpoint = TELEMETRY_ENDPOINTS[networkKey]?.syslog;
   let logPusher: LogPushWorker | null = null;
+  let otlpExporter: OtlpLogWorker | null = null;
 
   function startLogPusher(): { ok: boolean; error?: string } {
     if (logPusher) return { ok: true };
@@ -2190,8 +2204,115 @@ export async function runDaemonInner(
     log("Telemetry: log streaming disabled");
   }
 
+  function startOtlpExporter(): { ok: boolean; error?: string } {
+    if (otlpExporter) return { ok: true };
+    const endpoint =
+      config.telemetry?.logs?.endpoint || TELEMETRY_ENDPOINTS[networkKey]?.otlpLogs;
+    if (!endpoint) {
+      return {
+        ok: false,
+        error: `OTLP log export is not configured for ${networkKey} (set config.telemetry.logs.endpoint)`,
+      };
+    }
+    const minLevel = config.telemetry?.logs?.level ?? "info";
+    otlpExporter = new OtlpLogWorker({
+      endpoint,
+      token: config.telemetry?.logs?.token,
+      network: networkKey,
+      peerId: agent.peerId,
+      nodeName: config.name,
+      version: nodeVersion,
+      commit: nodeCommit,
+      role: config.nodeRole ?? "edge",
+      minLevel,
+      bufferMaxEntries: config.telemetry?.logs?.bufferMaxEntries,
+      onError: (m) => log(`Telemetry(OTLP): ${m}`),
+    });
+    otlpExporter.start();
+    log(`Telemetry: OTLP log export enabled → ${endpoint} (level ≥ ${minLevel})`);
+    return { ok: true };
+  }
+
+  function stopOtlpExporter(): void {
+    if (!otlpExporter) return;
+    otlpExporter.stop();
+    otlpExporter = null;
+    log("Telemetry: OTLP log export disabled");
+  }
+
+  // Dispatch to the configured log exporter. 'syslog' is the default when
+  // unset (preserves prior behaviour); 'otlp' is the recommended path; 'none'
+  // keeps logs local-only even while telemetry is enabled.
+  function startTelemetry(): { ok: boolean; error?: string } {
+    const mode = config.telemetry?.logs?.exporter ?? "syslog";
+    if (mode === "none") return { ok: true };
+    if (mode === "otlp") return startOtlpExporter();
+    return startLogPusher();
+  }
+
+  function stopTelemetry(): void {
+    stopLogPusher();
+    stopOtlpExporter();
+  }
+
+  // OTel traces + metrics SDK (independent of the log-exporter path above).
+  // Endpoints resolve env-first (standard OTEL_EXPORTER_OTLP_* names) then
+  // config; a signal registers ONLY when its endpoint resolves — never a
+  // guessed prod default. enabled=false ⇒ initTelemetry is a total no-op.
+  if (config.telemetry?.enabled) {
+    const otelBase = process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.replace(/\/$/, "");
+    const tracesEndpoint =
+      process.env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT ||
+      (otelBase ? `${otelBase}/v1/traces` : undefined) ||
+      config.telemetry.traces?.endpoint;
+    const metricsEndpoint =
+      process.env.OTEL_EXPORTER_OTLP_METRICS_ENDPOINT ||
+      (otelBase ? `${otelBase}/v1/metrics` : undefined) ||
+      config.telemetry.metrics?.endpoint;
+    const tracesOn = !!tracesEndpoint && config.telemetry.traces?.enabled !== false;
+    const metricsOn = !!metricsEndpoint && config.telemetry.metrics?.enabled !== false;
+    try {
+      initTelemetry({
+        enabled: true,
+        resource: {
+          serviceName: "dkg-node",
+          serviceVersion: nodeVersion,
+          serviceInstanceId: config.name,
+          network: networkKey,
+          peerId: agent.peerId,
+          nodeName: config.name,
+          nodeRole: config.nodeRole ?? "edge",
+          commit: nodeCommit,
+          chainId: config.chain?.chainId,
+        },
+        traces: tracesOn
+          ? {
+              endpoint: tracesEndpoint,
+              token: config.telemetry.traces?.token,
+              sampleRatio: config.telemetry.traces?.sampleRatio,
+            }
+          : undefined,
+        metrics: metricsOn
+          ? {
+              endpoint: metricsEndpoint,
+              token: config.telemetry.metrics?.token,
+              exportIntervalMs: config.telemetry.metrics?.exportIntervalMs,
+            }
+          : undefined,
+      });
+      if (tracesOn || metricsOn) {
+        log(
+          `Telemetry: OTel SDK registered (traces=${tracesOn ? tracesEndpoint : "off"}, metrics=${metricsOn ? metricsEndpoint : "off"})`,
+        );
+      }
+    } catch (err) {
+      // Telemetry must never block startup.
+      log(`Telemetry: OTel init failed (non-fatal): ${String(err)}`);
+    }
+  }
+
   if (config.telemetry?.enabled) {
-    const r = startLogPusher();
+    const r = startTelemetry();
     if (!r.ok) {
       log(`Telemetry: ${r.error}`);
       config.telemetry.enabled = false;
@@ -2619,10 +2740,10 @@ export async function runDaemonInner(
       enabled: boolean,
     ): Promise<{ ok: boolean; error?: string }> => {
       if (enabled) {
-        const r = startLogPusher();
+        const r = startTelemetry();
         if (!r.ok) return r;
       } else {
-        stopLogPusher();
+        stopTelemetry();
       }
       config.telemetry = { ...config.telemetry, enabled };
       await saveConfig(config);
@@ -3107,6 +3228,9 @@ export async function runDaemonInner(
         clearInterval(pruneTimer);
         rateLimiter.destroy();
         metricsCollector.stop();
+        stopTelemetry();
+        // Flush + shut down the OTel SDK (no-op if never registered).
+        await shutdownTelemetry().catch(() => {});
         natStatusWatcherStop?.();
         resetNatStatus();
         await publisherRuntime

packages/core/package.json

@@ -30,6 +30,7 @@
     "@multiformats/multiaddr": "^13.0.3",
     "@noble/ed25519": "^3.1.0",
     "@noble/hashes": "^2.2.0",
+    "@opentelemetry/api": "^1.9.1",
     "js-yaml": "^4.1.1",
     "libp2p": "^3.3.1",
     "protobufjs": "^8.3.0",

packages/core/src/index.ts

@@ -8,7 +8,12 @@ export * from './publisher-extension.js';
 export * from './imported-artifact-bytes.js';
 export * from './imported-artifact-metadata.js';
 export * from './event-bus.js';
-export { Logger, createOperationContext, type OperationContext, type OperationName, type LogSink } from './logger.js';
+export { Logger, createOperationContext, type OperationContext, type OperationName, type LogSink, type LogRecord } from './logger.js';
+export { createLogRedactor, redactLogEntry, redactMessage, DEFAULT_SENSITIVE_KEYS, REDACTED } from './log-redaction.js';
+export {
+  getTracer, withSpan, linkedSpan, currentTraceIds, activeSpanContext,
+  getMetrics, rebuildMetrics, type WithSpanOpts, type DkgMetrics,
+} from './telemetry-api.js';
 export * from './crypto/index.js';
 export * from './proto/index.js';
 export {