feat: Support Otel logging

genkit-tools/cli/src/utils/manager-utils.ts

@@ -15,6 +15,7 @@
  */
 
 import {
+  LocalFileLogStore,
   LocalFileTraceStore,
   startTelemetryServer,
 } from '@genkit-ai/telemetry-server';
@@ -48,6 +49,10 @@ export async function resolveTelemetryServer(options: {
         storeRoot: options.projectRoot,
         indexRoot: options.projectRoot,
       }),
+      logStore: new LocalFileLogStore({
+        storeRoot: options.projectRoot,
+        indexRoot: options.projectRoot,
+      }),
       corsOrigin: options.corsOrigin,
     });
   }

genkit-tools/common/src/types/index.ts

@@ -23,6 +23,7 @@ export * from './document';
 export * from './env';
 export * from './eval';
 export * from './evaluator';
+export * from './log';
 export * from './model';
 export * from './prompt';
 export * from './reflection';

genkit-tools/common/src/types/log.ts

@@ -0,0 +1,66 @@
+/**
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { extendZodWithOpenApi } from '@asteasolutions/zod-to-openapi';
+import * as z from 'zod';
+
+import { InstrumentationLibrarySchema } from './trace';
+
+extendZodWithOpenApi(z);
+
+/**
+ * Zod schema for log record data.
+ */
+export const LogRecordSchema = z
+  .object({
+    logId: z.string(), // Server-generated if omitted
+    traceId: z.string().optional(),
+    spanId: z.string().optional(),
+    timestamp: z
+      .number()
+      .describe('log recorded time in milliseconds since the epoch'),
+    severityNumber: z.number().optional(),
+    severityText: z.string().optional(),
+    body: z.unknown().optional(), // Represents any value: string, number, boolean, array, or object
+    attributes: z.record(z.string(), z.unknown()).optional(),
+    instrumentationLibrary: InstrumentationLibrarySchema.optional(),
+  })
+  .openapi('LogRecordData');
+
+export type LogRecordData = z.infer<typeof LogRecordSchema>;
+
+/**
+ * Log query parameters.
+ */
+export const LogQuerySchema = z.object({
+  limit: z.coerce.number().optional(),
+  continuationToken: z.string().optional(),
+  traceId: z.string().optional(),
+  spanId: z.string().optional(),
+});
+
+export type LogQuery = z.infer<typeof LogQuerySchema>;
+
+export interface LogQueryResponse {
+  logs: LogRecordData[];
+  continuationToken?: string;
+}
+
+export interface LogStore {
+  init(): Promise<void>;
+  save(logs: LogRecordData[]): Promise<void>;
+  list(query?: LogQuery): Promise<LogQueryResponse>;
+}

genkit-tools/telemetry-server/src/file-log-store.ts

@@ -0,0 +1,263 @@
+/**
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import {
+  LogRecordData,
+  type LogQuery,
+  type LogQueryResponse,
+  type LogStore,
+} from '@genkit-ai/tools-common';
+import { logger } from '@genkit-ai/tools-common/utils';
+import { Mutex } from 'async-mutex';
+import fs from 'fs';
+import lockfile from 'lockfile';
+import path from 'path';
+
+function lockFile(file: string) {
+  return `${file}.lock`;
+}
+
+export class LocalFileLogStore implements LogStore {
+  private readonly storeRoot: string;
+  private readonly indexRoot: string;
+  private readonly index: LogIndex;
+  private mutex = new Mutex();
+
+  constructor(options: { storeRoot: string; indexRoot: string }) {
+    this.storeRoot = path.resolve(options.storeRoot, '.genkit/logs');
+    fs.mkdirSync(this.storeRoot, { recursive: true });
+    this.indexRoot = path.resolve(options.indexRoot, '.genkit/logs_idx');
+    fs.mkdirSync(this.indexRoot, { recursive: true });
+    logger.debug(
+      `[Telemetry Server] initialized local file log store at root: ${this.storeRoot}`
+    );
+    this.index = new LogIndex(this.indexRoot);
+  }
+
+  async init(): Promise<void> {
+    // Indexes and store are append-only.
+  }
+
+  private getCurrentLogFile(): string {
+    const now = new Date();
+    // format YYYY-MM-DD-HH
+    const pad = (n: number) => n.toString().padStart(2, '0');
+    const dateStr = `${now.getFullYear()}-${pad(now.getMonth() + 1)}-${pad(now.getDate())}-${pad(now.getHours())}`;
+    return path.resolve(this.storeRoot, `logs-${dateStr}.jsonl`);
+  }
+
+  async save(logs: LogRecordData[]): Promise<void> {
+    if (logs.length === 0) return;
+
+    await this.mutex.runExclusive(() => {
+      const logFile = this.getCurrentLogFile();
+      const relativeFileName = path.basename(logFile);
+
+      let currentOffset = 0;
+      if (fs.existsSync(logFile)) {
+        currentOffset = fs.statSync(logFile).size;
+      }
+
+      const indexEntries: LogIndexEntry[] = [];
+      let offsetTracker = currentOffset;
+
+      const linesToAppend = logs.map((log) => {
+        // Ensure log id exists if not provided
+        if (!log.logId && log.traceId && log.spanId) {
+          log.logId = `${log.traceId}-${log.spanId}-${log.timestamp}-${Math.random().toString(36).substring(7)}`;
+        } else if (!log.logId) {
+          log.logId = `${log.timestamp}-${Math.random().toString(36).substring(7)}`;
+        }
+        const line = JSON.stringify(log) + '\n';
+        const length = Buffer.byteLength(line, 'utf8');
+
+        indexEntries.push({
+          traceId: log.traceId,
+          spanId: log.spanId,
+          timestamp: log.timestamp,
+          severityText: log.severityText,
+          severityNumber: log.severityNumber,
+          file: relativeFileName,
+          offset: offsetTracker,
+          length: length,
+        });
+
+        offsetTracker += length;
+        return line;
+      });
+
+      fs.appendFileSync(logFile, linesToAppend.join(''));
+      this.index.add(indexEntries);
+    });
+  }
+
+  async list(query?: LogQuery): Promise<LogQueryResponse> {
+    const startFromIndex = query?.continuationToken
+      ? Number.parseInt(query.continuationToken)
+      : 0;
+    const limit = query?.limit ?? 100;
+
+    const searchResult = this.index.search({
+      limit,
+      startFromIndex,
+      traceId: query?.traceId,
+      spanId: query?.spanId,
+    });
+
+    const logs: LogRecordData[] = [];
+
+    for (const entry of searchResult.entries) {
+      const logFile = path.resolve(this.storeRoot, entry.file);
+      if (fs.existsSync(logFile)) {
+        const buffer = Buffer.alloc(entry.length);
+        const fd = fs.openSync(logFile, 'r');
+        fs.readSync(fd, buffer, 0, entry.length, entry.offset);
+        fs.closeSync(fd);
+        try {
+          logs.push(
+            JSON.parse(buffer.toString('utf8').trim()) as LogRecordData
+          );
+        } catch (e) {
+          logger.error(`Error parsing log at ${entry.file}:${entry.offset}`);
+        }
+      }
+    }
+
+    return {
+      logs,
+      continuationToken: searchResult.pageLastIndex
+        ? `${searchResult.pageLastIndex}`
+        : undefined,
+    };
+  }
+}
+
+export interface LogIndexEntry {
+  traceId?: string;
+  spanId?: string;
+  timestamp: number;
+  severityText?: string;
+  severityNumber?: number;
+  file: string;
+  offset: number;
+  length: number;
+}
+
+export interface LogIndexSearchResult {
+  pageLastIndex?: number;
+  entries: LogIndexEntry[];
+}
+
+export class LogIndex {
+  private currentIndexFile: string;
+
+  constructor(private indexRoot: string) {
+    this.currentIndexFile = path.resolve(
+      this.indexRoot,
+      this.newIndexFileName()
+    );
+    fs.mkdirSync(this.indexRoot, { recursive: true });
+  }
+
+  private newIndexFileName() {
+    return `idx_${(Date.now() + '').padStart(17, '0')}.jsonl`;
+  }
+
+  listIndexFiles() {
+    return fs.readdirSync(this.indexRoot).filter((f) => f.startsWith('idx_'));
+  }
+
+  add(entries: LogIndexEntry[]) {
+    if (entries.length === 0) return;
+    try {
+      lockfile.lockSync(lockFile(this.currentIndexFile));
+      const lines = entries.map((e) => JSON.stringify(e) + '\n').join('');
+      fs.appendFileSync(this.currentIndexFile, lines);
+    } catch (err) {
+      logger.error(
+        `Failed to lock log index file ${this.currentIndexFile}: ${err}`
+      );
+    } finally {
+      if (fs.existsSync(lockFile(this.currentIndexFile))) {
+        lockfile.unlockSync(lockFile(this.currentIndexFile));
+      }
+    }
+  }
+
+  search(query: {
+    limit: number;
+    startFromIndex: number;
+    traceId?: string;
+    spanId?: string;
+  }): LogIndexSearchResult {
+    const indexFiles = this.listIndexFiles().sort().reverse();
+
+    let skipped = 0;
+    const entries: LogIndexEntry[] = [];
+    let hasMore = false;
+
+    for (const idxFile of indexFiles) {
+      if (hasMore) break;
+
+      const idxTxt = fs.readFileSync(
+        path.resolve(this.indexRoot, idxFile),
+        'utf8'
+      );
+      const fileData = idxTxt
+        .split('\n')
+        .filter((l) => l.trim().length > 0)
+        .map((l) => {
+          try {
+            return JSON.parse(l) as LogIndexEntry;
+          } catch {
+            return undefined;
+          }
+        })
+        .filter((d): d is LogIndexEntry => {
+          if (!d) return false;
+          if (query.traceId && d.traceId !== query.traceId) return false;
+          if (query.spanId && d.spanId !== query.spanId) return false;
+          return true;
+        });
+
+      fileData.sort((a, b) => b.timestamp - a.timestamp); // Newest first
+
+      for (const entry of fileData) {
+        if (skipped < query.startFromIndex) {
+          skipped++;
+          continue;
+        }
+
+        if (entries.length < query.limit) {
+          entries.push(entry);
+        } else {
+          hasMore = true;
+          break;
+        }
+      }
+    }
+
+    const result: LogIndexSearchResult = {
+      entries,
+    };
+
+    if (hasMore) {
+      result.pageLastIndex = query.startFromIndex + query.limit;
+    }
+
+    return result;
+  }
+}