chore: release v2.60.2

push

.env.prod.example

@@ -134,6 +134,10 @@ LANGFUSE_CSP_ENFORCE_HTTPS="true"
 # Used to determine the Sentry sample rate
 # LANGFUSE_TRACING_SAMPLE_RATE=
 
+# NewRelic
+# NEW_RELIC_API_KEY= 
+# OTLP_ENDPOINT=
+
 # Cloudflare Turnstile
 # NEXT_PUBLIC_TURNSTILE_SITE_KEY=
 # TURNSTILE_SECRET_KEY=
@@ -153,4 +157,5 @@ LANGFUSE_CSP_ENFORCE_HTTPS="true"
 # Admin API
 # ADMIN_API_KEY=
 
+
 ### END Langfuse Cloud Config

.gitignore

@@ -1,4 +1,5 @@
 # See https://help.github.com/articles/ignoring-files/ for more about ignoring files.
+**/newrelic_agent.log
 
 # dependencies
 /node_modules

package.json

@@ -1,6 +1,6 @@
 {
   "name": "langfuse",
-  "version": "2.60.0",
+  "version": "2.60.2",
   "author": "engineering@langfuse.com",
   "license": "MIT",
   "private": true,
@@ -79,5 +79,8 @@
         "pr": ":rocket: _This pull request is included in v${version}. See [${releaseName}](${releaseUrl}) for release notes._"
       }
     }
+  },
+  "dependencies": {
+    "newrelic": "^11.22.0"
   }
 }

packages/shared/prisma/generated/types.ts

@@ -286,7 +286,7 @@ export type ObservationView = {
     trace_id: string | null;
     project_id: string;
     type: ObservationType;
-    start_time: Generated<Timestamp>;
+    start_time: Timestamp;
     end_time: Timestamp | null;
     name: string | null;
     metadata: unknown | null;
@@ -294,7 +294,8 @@ export type ObservationView = {
     level: Generated<ObservationLevel>;
     status_message: string | null;
     version: string | null;
-    created_at: Generated<Timestamp>;
+    created_at: Timestamp;
+    updated_at: Timestamp;
     model: string | null;
     modelParameters: unknown | null;
     input: unknown | null;
@@ -438,6 +439,8 @@ export type TraceView = {
     input: unknown | null;
     output: unknown | null;
     session_id: string | null;
+    created_at: Timestamp;
+    updated_at: Timestamp;
     duration: number | null;
 };
 export type User = {

packages/shared/prisma/migrations/20240705152639_traces_view_add_created_at_updated_at/migration.sql

@@ -0,0 +1,20 @@
+-- Drop and create to be able to change columns, otherwise new t.* cols cannot be added
+
+DROP VIEW IF EXISTS traces_view;
+CREATE VIEW traces_view AS
+WITH observations_metrics AS (
+    SELECT
+        trace_id,
+        project_id,
+        EXTRACT(EPOCH FROM COALESCE(MAX(o.end_time), MAX(o.start_time))) - EXTRACT(EPOCH FROM MIN(o.start_time))::double precision AS duration
+    FROM
+        observations o
+    GROUP BY
+        project_id, trace_id
+)
+SELECT
+    t.*,
+    o.duration
+FROM
+    traces t
+    LEFT JOIN observations_metrics o ON t.id = o.trace_id and t.project_id = o.project_id

packages/shared/prisma/migrations/20240705154048_observation_view_add_created_at_updated_at/migration.sql

@@ -0,0 +1,89 @@
+DROP VIEW IF EXISTS "observations_view"; -- Drop view as column was added in 20240704103900_observations_view_read_from_calculated and update view must have same columns
+CREATE VIEW "observations_view" AS -- Specify the columns that should be returned in the view, as calculated columns are added but exist in the observations table already
+SELECT
+    o.id,
+    o.name,
+    o.start_time,
+    o.end_time,
+    o.parent_observation_id,
+    o.type,
+    o.trace_id,
+    o.metadata,
+    o.model,
+    o."modelParameters",
+    o.input,
+    o.output,
+    o.level,
+    o.status_message,
+    o.completion_start_time,
+    o.completion_tokens,
+    o.prompt_tokens,
+    o.total_tokens,
+    o.version,
+    o.project_id,
+    o.created_at,
+    o.updated_at,
+    o.unit,
+    o.prompt_id,
+    o.input_cost,
+    o.output_cost,
+    o.total_cost,
+    o.internal_model,
+    m.id AS "model_id",
+    m.start_date AS "model_start_date",
+    m.input_price,
+    m.output_price,
+    m.total_price,
+    m.tokenizer_config AS "tokenizer_config",
+    CASE 
+        WHEN o.calculated_input_cost IS NULL AND o.input_cost IS NULL AND o.output_cost IS NULL AND o.total_cost IS NULL THEN
+            o.prompt_tokens::decimal * m.input_price
+        ELSE
+            COALESCE(o.calculated_input_cost, o.input_cost)
+    END AS "calculated_input_cost",
+    CASE 
+        WHEN o.calculated_output_cost IS NULL AND o.input_cost IS NULL AND o.output_cost IS NULL AND o.total_cost IS NULL THEN
+            o.completion_tokens::decimal * m.output_price
+        ELSE
+            COALESCE(o.calculated_output_cost, o.output_cost)
+    END AS "calculated_output_cost",
+    CASE 
+        WHEN o.calculated_total_cost IS NULL AND o.input_cost IS NULL AND o.output_cost IS NULL AND o.total_cost IS NULL THEN
+            CASE 
+                WHEN m.total_price IS NOT NULL AND o.total_tokens IS NOT NULL THEN
+                    m.total_price * o.total_tokens
+                ELSE
+                    o.prompt_tokens::decimal * m.input_price + 
+                    o.completion_tokens::decimal * m.output_price
+            END
+        ELSE
+            COALESCE(o.calculated_total_cost, o.total_cost)
+    END AS "calculated_total_cost",
+    CASE WHEN o.end_time IS NULL THEN NULL ELSE (EXTRACT(EPOCH FROM o."end_time") - EXTRACT(EPOCH FROM o."start_time"))::double precision END AS "latency",
+    CASE WHEN o.completion_start_time IS NOT NULL AND o.start_time IS NOT NULL THEN EXTRACT(EPOCH FROM (completion_start_time - start_time))::double precision ELSE NULL END as "time_to_first_token"
+    
+FROM
+    observations o
+LEFT JOIN LATERAL (
+    SELECT
+        models.*
+    FROM
+        models
+    WHERE (models.project_id = o.project_id OR models.project_id IS NULL)
+    AND models.model_name = o.internal_model
+    AND (models.start_date < o.start_time OR models.start_date IS NULL)
+    AND o.unit::TEXT = models.unit
+    ORDER BY
+        models.project_id ASC, -- in postgres, NULLs are sorted last when ordering ASC
+        models.start_date DESC NULLS LAST -- now, NULLs are sorted last when ordering DESC as well
+    LIMIT 1
+) m ON TRUE
+
+
+-- requirements:
+-- 1. The view should return all columns from the observations table
+-- 2. The view should match with only one model for each observation if:
+--     a. The model has the same project_id as the observation, otherwise the model without project_id. 
+--     b. The model has the same model_name as the observation
+--     c. The model has a start_date that is less than the observation start_time, otherwise the model without start_date
+--     d. The model has the same unit as the observation

packages/shared/prisma/schema.prisma

@@ -272,6 +272,8 @@ view TraceView {
     input      Json?
     output     Json?
     sessionId  String?  @map("session_id")
+    createdAt  DateTime @map("created_at")
+    updatedAt  DateTime @map("updated_at")
 
     // calculated fields
     duration Float? @map("duration") // can be null if no observations in trace
@@ -352,7 +354,7 @@ view ObservationView {
     traceId             String?          @map("trace_id")
     projectId           String           @map("project_id")
     type                ObservationType
-    startTime           DateTime         @default(now()) @map("start_time")
+    startTime           DateTime         @map("start_time")
     endTime             DateTime?        @map("end_time")
     name                String?
     metadata            Json?
@@ -360,7 +362,8 @@ view ObservationView {
     level               ObservationLevel @default(DEFAULT)
     statusMessage       String?          @map("status_message")
     version             String?
-    createdAt           DateTime         @default(now()) @map("created_at")
+    createdAt           DateTime         @map("created_at")
+    updateAt            DateTime         @map("updated_at")
 
     // GENERATION ONLY
     model               String?

web/Dockerfile

@@ -1,10 +1,10 @@
-FROM --platform=${BUILDPLATFORM:-linux/amd64} node:20-alpine3.20 AS alpine
+FROM --platform=${TARGETPLATFORM:-linux/amd64} node:20-alpine3.20 AS alpine
 
 # It's important to update the index before installing packages to ensure you're getting the latest versions.
 # Check https://github.com/nodejs/docker-node/tree/b4117f9333da4138b03a546ec926ef50a31506c3#nodealpine to understand why libc6-compat might be needed.
 RUN apk update && apk upgrade --no-cache libcrypto3 libssl3 libc6-compat busybox ssl_client
 
-FROM --platform=${BUILDPLATFORM:-linux/amd64} alpine AS base
+FROM --platform=${TARGETPLATFORM:-linux/amd64} alpine AS base
 RUN npm install turbo@^1.13.3 --global
 ENV PNPM_HOME="/pnpm"
 ENV PATH="$PNPM_HOME:$PATH"
@@ -12,7 +12,7 @@ RUN corepack enable
 RUN corepack prepare pnpm@8.15.5 --activate
 
 
-FROM --platform=${BUILDPLATFORM:-linux/amd64} base AS pruner
+FROM --platform=${TARGETPLATFORM:-linux/amd64} base AS pruner
 
 WORKDIR /app
 
@@ -20,7 +20,7 @@ COPY . .
 RUN turbo prune --scope=web --docker
 
 
-FROM --platform=${BUILDPLATFORM:-linux/amd64} base AS builder
+FROM --platform=${TARGETPLATFORM:-linux/amd64} base AS builder
 
 WORKDIR /app
 

web/package.json

@@ -1,6 +1,6 @@
 {
   "name": "web",
-  "version": "2.60.0",
+  "version": "2.60.2",
   "private": true,
   "license": "MIT",
   "engines": {
@@ -36,6 +36,18 @@
     "@marsidev/react-turnstile": "^0.5.4",
     "@mui/x-tree-view": "^7.6.2",
     "@next-auth/prisma-adapter": "^1.0.7",
+    "@opentelemetry/api": "^1.9.0",
+    "@opentelemetry/auto-instrumentations-node": "^0.48.0",
+    "@opentelemetry/exporter-jaeger": "^1.25.1",
+    "@opentelemetry/exporter-trace-otlp-http": "^0.52.1",
+    "@opentelemetry/resource-detector-aws": "^1.5.2",
+    "@opentelemetry/resource-detector-container": "^0.3.11",
+    "@opentelemetry/resources": "^1.25.1",
+    "@opentelemetry/sdk-node": "^0.52.1",
+    "@opentelemetry/sdk-trace-node": "^1.25.1",
+    "@opentelemetry/semantic-conventions": "^1.25.1",
+    "@opentelemetry/winston-transport": "^0.5.0",
+    "@prisma/instrumentation": "^5.16.1",
     "@radix-ui/react-accordion": "^1.1.2",
     "@radix-ui/react-alert-dialog": "^1.0.5",
     "@radix-ui/react-avatar": "^1.0.4",

web/src/components/token-usage-badge.tsx

@@ -53,7 +53,7 @@ export const TokenUsageBadge = (
   if (props.inline)
     return (
       <span>
-        {`${numberFormatter(usage.promptTokens, 0)} → ${numberFormatter(usage.promptTokens, 0)} (∑ ${numberFormatter(usage.totalTokens, 0)})`}
+        {`${numberFormatter(usage.promptTokens, 0)} → ${numberFormatter(usage.completionTokens, 0)} (∑ ${numberFormatter(usage.totalTokens, 0)})`}
       </span>
     );
 

web/src/constants/VERSION.ts

@@ -1 +1 @@
-export const VERSION = "v2.60.0";
+export const VERSION = "v2.60.2";

web/src/new-relic-instrumentation.ts

@@ -0,0 +1,51 @@
+import { NodeSDK } from "@opentelemetry/sdk-node";
+import { getNodeAutoInstrumentations } from "@opentelemetry/auto-instrumentations-node";
+import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
+import { Resource } from "@opentelemetry/resources";
+import { SEMRESATTRS_SERVICE_NAME } from "@opentelemetry/semantic-conventions";
+import { BatchSpanProcessor } from "@opentelemetry/sdk-trace-node";
+import { PrismaInstrumentation } from "@prisma/instrumentation";
+import {
+  awsEksDetector,
+  awsEc2Detector,
+} from "@opentelemetry/resource-detector-aws";
+import {
+  hostDetector,
+  osDetector,
+  processDetector,
+} from "@opentelemetry/resources/build/src/detectors/platform";
+import { envDetector } from "@opentelemetry/resources";
+import { containerDetector } from "@opentelemetry/resource-detector-container";
+
+const sdk = new NodeSDK({
+  resource: new Resource({
+    [SEMRESATTRS_SERVICE_NAME]: "web",
+  }),
+  spanProcessors: [
+    new BatchSpanProcessor(
+      new OTLPTraceExporter({
+        url:
+          process.env.OTLP_ENDPOINT ||
+          "https://otlp.eu01.nr-data.net/v1/traces",
+        headers: {
+          "api-key": process.env.NEW_RELIC_API_KEY,
+        },
+      }),
+    ),
+  ],
+  resourceDetectors: [
+    containerDetector,
+    envDetector,
+    hostDetector,
+    osDetector,
+    processDetector,
+    awsEksDetector,
+    awsEc2Detector,
+  ],
+  instrumentations: [
+    getNodeAutoInstrumentations(),
+    new PrismaInstrumentation(),
+  ],
+});
+
+sdk.start();

web/src/pages/api/public/ingestion.ts

@@ -494,6 +494,7 @@ export const sendToWorkerIfEnvironmentConfigured = async (
               ).toString("base64"),
           },
           body: JSON.stringify(body),
+          signal: AbortSignal.timeout(2 * 1000),
         });
       }
     }

web/src/sentry.server.config.ts

@@ -12,6 +12,7 @@ if (process.env.NEXT_PUBLIC_SENTRY_DSN)
     // of transactions for performance monitoring.
     // We recommend adjusting this value in production
     tracesSampleRate: env.LANGFUSE_TRACING_SAMPLE_RATE,
+
     profilesSampleRate: 0.1,
     integrations: [
       // Add profiling integration to list of integrations

worker/Dockerfile

@@ -1,10 +1,11 @@
-FROM --platform=${BUILDPLATFORM:-linux/amd64} node:20-alpine3.20 AS alpine
+FROM --platform=${TARGETPLATFORM:-linux/amd64} node:20-alpine3.20 AS alpine
 
 # It's important to update the index before installing packages to ensure you're getting the latest versions.
 # Check https://github.com/nodejs/docker-node/tree/b4117f9333da4138b03a546ec926ef50a31506c3#nodealpine to understand why libc6-compat might be needed.
 RUN apk update && apk upgrade --no-cache libcrypto3 libssl3 libc6-compat busybox ssl_client
 
-FROM --platform=${BUILDPLATFORM:-linux/amd64} alpine AS base
+
+FROM --platform=${TARGETPLATFORM:-linux/amd64} alpine AS base
 RUN npm install turbo@^1.13.3 --global
 ENV PNPM_HOME="/pnpm"
 ENV PATH="$PNPM_HOME:$PATH"
@@ -12,7 +13,7 @@ RUN corepack enable
 RUN corepack prepare pnpm@8.15.5 --activate
 
 
-FROM --platform=${BUILDPLATFORM:-linux/amd64} base AS pruner
+FROM --platform=${TARGETPLATFORM:-linux/amd64} base AS pruner
 
 WORKDIR /app
 
@@ -21,7 +22,7 @@ RUN turbo prune --scope=worker --docker
 
 
 
-FROM --platform=${BUILDPLATFORM:-linux/amd64} base AS builder
+FROM --platform=${TARGETPLATFORM:-linux/amd64} base AS builder
 
 WORKDIR /app
 
@@ -57,8 +58,10 @@ RUN addgroup --system --gid 1001 expressjs
 RUN adduser --system --uid 1001 expressjs
 USER expressjs
 COPY --from=builder --chown=expressjs:expressjs /app .
+COPY --chown=expressjs:expressjs ./worker/newrelic.js ./newrelic.js
 
 EXPOSE 3030
 ENV PORT=3030
 
-CMD ["node", "worker/dist/index.js"]
+CMD ["node", "--experimental-loader=newrelic/esm-loader.mjs", "worker/dist/index.js"]
+

worker/newrelic.js

@@ -0,0 +1,26 @@
+"use strict";
+
+/**
+ * New Relic agent configuration.
+ *
+ * See lib/config.defaults.js in the agent distribution for a more complete
+ * description of configuration variables and their potential values.
+ */
+exports.config = {
+  /**
+   * Array of application names.
+   */
+  app_name: ["worker"],
+  /**
+   * Your New Relic license key.
+   */
+  license_key: process.env.NEW_RELIC_API_KEY,
+  logging: {
+    enabled: false,
+  },
+  application_logging: {
+    forwarding: {
+      enabled: false,
+    },
+  },
+};

worker/package.json

@@ -1,6 +1,6 @@
 {
   "name": "worker",
-  "version": "2.60.0",
+  "version": "2.60.2",
   "description": "",
   "license": "MIT",
   "private": true,
@@ -10,7 +10,7 @@
   "scripts": {
     "test": "dotenv -e ../.env -- vitest run --reporter=basic --pool=forks",
     "coverage": "vitest run --coverage",
-    "start": "dotenv -e ../.env -- node dist/index.js",
+    "start": "dotenv -e ../.env -- node --experimental-loader=newrelic/esm-loader.mjs dist/index.js",
     "build": "tsc",
     "dev": "dotenv -e ../.env -- nodemon src/index.ts",
     "lint": "eslint . --ext .js,.jsx,.ts,.tsx",
@@ -35,6 +35,7 @@
     "js-tiktoken": "^1.0.12",
     "kysely": "^0.27.3",
     "lodash": "^4.17.21",
+    "newrelic": "^11.22.0",
     "pg": "^8.11.5",
     "pino": "^9.2.0",
     "pino-http": "^9.0.0",
@@ -50,6 +51,7 @@
     "@types/express": "^4.17.21",
     "@types/express-serve-static-core": "^4.19.3",
     "@types/lodash": "^4.17.5",
+    "@types/newrelic": "^9.14.4",
     "@types/node": "^20.11.19",
     "@types/pg": "^8.11.6",
     "@typescript-eslint/parser": "^7.12.0",

worker/src/constants/VERSION.ts

@@ -1 +1 @@
-export const VERSION = "v2.60.0";
+export const VERSION = "v2.60.2";

worker/src/index.ts

@@ -1,3 +1,4 @@
+import "newrelic";
 import app from "./app";
 import { env } from "./env";
 import logger from "./logger";

worker/src/instrumentation.ts

@@ -1,37 +1,12 @@
-import { nodeProfilingIntegration } from "@sentry/profiling-node";
-import { env } from "./env";
-import * as Sentry from "@sentry/node";
+import newrelic from "newrelic";
 
-Sentry.init({
-  dsn: String(env.SENTRY_DSN),
-  integrations: [
-    Sentry.httpIntegration(),
-    Sentry.expressIntegration(),
-    nodeProfilingIntegration(),
-    Sentry.redisIntegration(),
-    Sentry.prismaIntegration(),
-  ],
-
-  // Add Tracing by setting tracesSampleRate
-  // We recommend adjusting this value in production
-  tracesSampleRate: 0.5,
-
-  // Set sampling rate for profiling
-  // This is relative to tracesSampleRate
-  profilesSampleRate: 0.1,
-});
-
-type CallbackAsyncFn<T> = (span?: Sentry.Span) => Promise<T>;
+type CallbackAsyncFn<T> = () => Promise<T>;
 
 export async function instrumentAsync<T>(
   ctx: { name: string },
   callback: CallbackAsyncFn<T>
 ): Promise<T> {
-  if (env.SENTRY_DSN) {
-    return Sentry.startSpan(ctx, async (span) => {
-      return callback(span);
-    });
-  } else {
+  return newrelic.startSegment(ctx.name, true, async function () {
     return callback();
-  }
+  });
 }

worker/src/queues/batchExportQueue.ts

@@ -24,43 +24,36 @@ export const batchExportJobExecutor = redis
   ? new Worker<TQueueJobTypes[QueueName.BatchExport]>(
       QueueName.BatchExport,
       async (job: Job<TQueueJobTypes[QueueName.BatchExport]>) => {
-        return instrumentAsync(
-          { name: "batchExportJobExecutor" },
-          async (span) => {
-            try {
-              logger.info("Executing Batch Export Job", job.data.payload);
-              await handleBatchExportJob(job.data.payload);
+        return instrumentAsync({ name: "batchExportJobExecutor" }, async () => {
+          try {
+            logger.info("Executing Batch Export Job", job.data.payload);
+            await handleBatchExportJob(job.data.payload);
 
-              logger.info("Finished Batch Export Job", job.data.payload);
+            logger.info("Finished Batch Export Job", job.data.payload);
 
-              return true;
-            } catch (e) {
-              const displayError =
-                e instanceof BaseError
-                  ? e.message
-                  : "An internal error occurred";
+            return true;
+          } catch (e) {
+            const displayError =
+              e instanceof BaseError ? e.message : "An internal error occurred";
 
-              await kyselyPrisma.$kysely
-                .updateTable("batch_exports")
-                .set("status", BatchExportStatus.FAILED)
-                .set("finished_at", new Date())
-                .set("log", displayError)
-                .where("id", "=", job.data.payload.batchExportId)
-                .where("project_id", "=", job.data.payload.projectId)
-                .execute();
+            await kyselyPrisma.$kysely
+              .updateTable("batch_exports")
+              .set("status", BatchExportStatus.FAILED)
+              .set("finished_at", new Date())
+              .set("log", displayError)
+              .where("id", "=", job.data.payload.batchExportId)
+              .where("project_id", "=", job.data.payload.projectId)
+              .execute();
 
-              logger.error(
-                e,
-                `Failed Batch Export job for id ${job.data.payload.batchExportId} ${e}`
-              );
-              Sentry.captureException(e);
+            logger.error(
+              e,
+              `Failed Batch Export job for id ${job.data.payload.batchExportId} ${e}`
+            );
+            Sentry.captureException(e);
 
-              throw e;
-            } finally {
-              span?.end();
-            }
+            throw e;
           }
-        );
+        });
       },
       {
         connection: redis,

worker/src/queues/evalQueue.ts

@@ -26,7 +26,7 @@ export const evalJobCreator = redis
   ? new Worker<TQueueJobTypes[QueueName.TraceUpsert]>(
       QueueName.TraceUpsert,
       async (job: Job<TQueueJobTypes[QueueName.TraceUpsert]>) => {
-        return instrumentAsync({ name: "evalJobCreator" }, async (span) => {
+        return instrumentAsync({ name: "evalJobCreator" }, async () => {
           try {
             await createEvalJobs({ event: job.data.payload });
             return true;
@@ -37,8 +37,6 @@ export const evalJobCreator = redis
             );
             Sentry.captureException(e);
             throw e;
-          } finally {
-            span?.end();
           }
         });
       },
@@ -58,7 +56,7 @@ export const evalJobExecutor = redis
   ? new Worker<TQueueJobTypes[QueueName.EvaluationExecution]>(
       QueueName.EvaluationExecution,
       async (job: Job<TQueueJobTypes[QueueName.EvaluationExecution]>) => {
-        return instrumentAsync({ name: "evalJobExecutor" }, async (span) => {
+        return instrumentAsync({ name: "evalJobExecutor" }, async () => {
           try {
             logger.info("Executing Evaluation Execution Job", job.data);
             await evaluate({ event: job.data.payload });
@@ -92,8 +90,6 @@ export const evalJobExecutor = redis
             }
 
             throw e;
-          } finally {
-            span?.end();
           }
         });
       },