diff --git a/.env.example b/.env.example index 23beccf116..584fa781e9 100644 --- a/.env.example +++ b/.env.example @@ -123,7 +123,7 @@ NODE_MAX_OLD_SPACE_SIZE=6144 # LANGFUSE_BASE_URL= # Optional Langfuse fanout for tenant-scoped Langfuse projects. -# The collector is opt-in: add docker-compose.langfuse-fanout.yml, +# The fanout gateway is opt-in: add docker-compose.langfuse-fanout.yml, # deploy-compose.langfuse-fanout.yml, or enable helm langfuseFanout. # Tenant public/secret keys are read from LibreChat tenant app configuration. # Tenant Langfuse base URLs must be set in tenant app configuration and match @@ -132,22 +132,44 @@ NODE_MAX_OLD_SPACE_SIZE=6144 # See otel/langfuse-fanout/README.md. # LANGFUSE_FANOUT_ENABLED=false # LANGFUSE_FANOUT_COLLECTOR_URL=http://langfuse-fanout-collector:4318 -# Emergency switch: unset/false defaults enabled; set true to keep central collector export but skip tenant trace/score export. +# Emergency switch: unset/false defaults enabled; set true to keep central fanout export but skip tenant trace/score export. # LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED=false # Langfuse Cloud base URL options: https://cloud.langfuse.com (EU), # https://us.cloud.langfuse.com (US), https://jp.cloud.langfuse.com (JP). -# Collector-only central trace export URL. LibreChat feedback scores use +# Gateway-only central trace/media export URL. LibreChat feedback scores use # LANGFUSE_BASE_URL, so set both URLs to the same non-EU region when applicable. # LANGFUSE_FANOUT_CENTRAL_BASE_URL=https://cloud.langfuse.com -# Collector-only Basic auth header for central trace export. LibreChat feedback +# Gateway-only Basic auth header for central trace/media export. LibreChat feedback # scores use LANGFUSE_PUBLIC_KEY/LANGFUSE_SECRET_KEY instead. # LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=Basic -# Compose's included collector config supports the three listed destination keys. -# Add custom keys only when the collector routing table/exporters are updated too. +# Compose's included gateway config supports the three listed destination keys. +# Add custom keys only when the gateway is started with matching destination URLs. # LANGFUSE_FANOUT_TENANT_DESTINATIONS=eu=https://cloud.langfuse.com,us=https://us.cloud.langfuse.com,jp=https://jp.cloud.langfuse.com +# Compose's collector config routes only these destination keys. The gateway +# fails startup when LANGFUSE_FANOUT_TENANT_DESTINATIONS contains another key. +# LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS=eu,us,jp +# Gateway base URL used to build one-time media upload URLs. Compose sets this +# to its private service URL; Helm derives an internal service URL unless set. +# LANGFUSE_FANOUT_PUBLIC_URL=http://langfuse-fanout-collector:4318 +# Internal gateway-to-collector trace endpoint. Compose sets this automatically. +# LANGFUSE_FANOUT_TRACE_COLLECTOR_URL=http://langfuse-fanout-otel:4319 +# Redis-backed one-time upload plans let multiple gateway pods handle Langfuse +# media create/upload requests. Compose sets this to its private Redis service. +# LANGFUSE_FANOUT_REDIS_URI=redis://langfuse-fanout-redis:6379 +# LANGFUSE_FANOUT_REDIS_USERNAME= +# LANGFUSE_FANOUT_REDIS_PASSWORD= +# LANGFUSE_FANOUT_REDIS_KEY_PREFIX=langfuse-fanout +# Internal collector receiver bind address. Helm uses 127.0.0.1 because the +# collector is a sidecar; Compose uses 0.0.0.0 on the private fanout network. +# LANGFUSE_FANOUT_OTEL_RECEIVER_ENDPOINT=0.0.0.0:4319 +# Static Compose collector destination URLs. Helm derives these from values. # LANGFUSE_FANOUT_TENANT_EU_BASE_URL=https://cloud.langfuse.com # LANGFUSE_FANOUT_TENANT_US_BASE_URL=https://us.cloud.langfuse.com # LANGFUSE_FANOUT_TENANT_JP_BASE_URL=https://jp.cloud.langfuse.com +# LANGFUSE_FANOUT_UPSTREAM_TIMEOUT=30s +# Optional bearer token for scraping the fanout gateway /metrics endpoint. +# If unset, /metrics returns 401. The gateway also accepts METRICS_SECRET when present. +# LANGFUSE_FANOUT_METRICS_SECRET= # LANGFUSE_FANOUT_MEMORY_LIMIT_MIB=256 # LANGFUSE_FANOUT_MEMORY_SPIKE_LIMIT_MIB=64 # LANGFUSE_FANOUT_BATCH_TIMEOUT=1s diff --git a/.github/workflows/eslint-ci.yml b/.github/workflows/eslint-ci.yml index a988196b5b..3ab8528b04 100644 --- a/.github/workflows/eslint-ci.yml +++ b/.github/workflows/eslint-ci.yml @@ -11,10 +11,6 @@ on: - 'api/**' - 'client/**' - 'packages/**' - - 'helm/librechat/templates/langfuse-fanout-*.yaml' - - 'helm/librechat/tests/langfuse_fanout_*.sh' - - 'helm/librechat/values.yaml' - - 'otel/langfuse-fanout/**' - '.github/workflows/eslint-ci.yml' jobs: @@ -130,21 +126,3 @@ jobs: echo "::error::Or rely on the lint-staged pre-commit hook (do not bypass with --no-verify)." exit 1 fi - - - name: Check Langfuse fanout collector config drift - run: | - BASE_SHA=$(jq --raw-output .pull_request.base.sha "$GITHUB_EVENT_PATH") - mapfile -d '' -t CHANGED_FILES < <( - git diff -z --name-only --diff-filter=ACMRTUXB "$BASE_SHA" HEAD | - grep -zE '^(otel/langfuse-fanout/|helm/librechat/(templates/langfuse-fanout-|tests/langfuse_fanout_|values\.yaml$))' || true - ) - - if [[ ${#CHANGED_FILES[@]} -eq 0 ]]; then - echo "No Langfuse fanout collector files changed. Skipping drift check." - exit 0 - fi - - echo "Files triggering drift check:" - printf '%s\n' "${CHANGED_FILES[@]}" - - helm/librechat/tests/langfuse_fanout_config_drift_test.sh diff --git a/deploy-compose.langfuse-fanout.yml b/deploy-compose.langfuse-fanout.yml index 54a30e33df..15f57c1d88 100644 --- a/deploy-compose.langfuse-fanout.yml +++ b/deploy-compose.langfuse-fanout.yml @@ -12,6 +12,42 @@ services: - langfuse-fanout langfuse-fanout-collector: + build: + context: . + dockerfile: otel/langfuse-fanout/Dockerfile + image: librechat-langfuse-fanout:local + restart: always + depends_on: + - langfuse-fanout-otel + - langfuse-fanout-redis + env_file: + - .env + environment: + - LANGFUSE_FANOUT_CENTRAL_BASE_URL=${LANGFUSE_FANOUT_CENTRAL_BASE_URL:-https://cloud.langfuse.com} + - LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=${LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER:?Set LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER to central Langfuse Basic auth header} + - LANGFUSE_FANOUT_TENANT_DESTINATIONS=${LANGFUSE_FANOUT_TENANT_DESTINATIONS:-eu=https://cloud.langfuse.com,us=https://us.cloud.langfuse.com,jp=https://jp.cloud.langfuse.com} + - LANGFUSE_FANOUT_UPSTREAM_TIMEOUT=${LANGFUSE_FANOUT_UPSTREAM_TIMEOUT:-30s} + - LANGFUSE_FANOUT_PUBLIC_URL=${LANGFUSE_FANOUT_PUBLIC_URL:-http://langfuse-fanout-collector:4318} + - LANGFUSE_FANOUT_TRACE_COLLECTOR_URL=http://langfuse-fanout-otel:4319 + - LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS=eu,us,jp + - LANGFUSE_FANOUT_REDIS_URI=${LANGFUSE_FANOUT_REDIS_URI:-redis://langfuse-fanout-redis:6379} + - LANGFUSE_FANOUT_REDIS_USERNAME=${LANGFUSE_FANOUT_REDIS_USERNAME:-} + - LANGFUSE_FANOUT_REDIS_PASSWORD=${LANGFUSE_FANOUT_REDIS_PASSWORD:-} + - LANGFUSE_FANOUT_REDIS_KEY_PREFIX=${LANGFUSE_FANOUT_REDIS_KEY_PREFIX:-langfuse-fanout} + expose: + - '4318' + networks: + - langfuse-fanout + + langfuse-fanout-redis: + image: redis:7.4-alpine + restart: always + expose: + - '6379' + networks: + - langfuse-fanout + + langfuse-fanout-otel: image: otel/opentelemetry-collector-contrib:0.143.0 restart: always command: ['--config=/etc/otelcol/otelcol.yaml'] @@ -20,6 +56,7 @@ services: environment: - LANGFUSE_FANOUT_CENTRAL_BASE_URL=${LANGFUSE_FANOUT_CENTRAL_BASE_URL:-https://cloud.langfuse.com} - LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=${LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER:?Set LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER to central Langfuse Basic auth header} + - LANGFUSE_FANOUT_OTEL_RECEIVER_ENDPOINT=0.0.0.0:4319 - LANGFUSE_FANOUT_TENANT_EU_BASE_URL=${LANGFUSE_FANOUT_TENANT_EU_BASE_URL:-https://cloud.langfuse.com} - LANGFUSE_FANOUT_TENANT_US_BASE_URL=${LANGFUSE_FANOUT_TENANT_US_BASE_URL:-https://us.cloud.langfuse.com} - LANGFUSE_FANOUT_TENANT_JP_BASE_URL=${LANGFUSE_FANOUT_TENANT_JP_BASE_URL:-https://jp.cloud.langfuse.com} @@ -28,10 +65,10 @@ services: - LANGFUSE_FANOUT_BATCH_TIMEOUT=${LANGFUSE_FANOUT_BATCH_TIMEOUT:-1s} - LANGFUSE_FANOUT_BATCH_SEND_SIZE=${LANGFUSE_FANOUT_BATCH_SEND_SIZE:-128} - LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT=${LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT:-1000} - expose: - - '4318' volumes: - ./otel/langfuse-fanout/otelcol.yaml:/etc/otelcol/otelcol.yaml:ro + expose: + - '4319' networks: - langfuse-fanout diff --git a/docker-compose.langfuse-fanout.yml b/docker-compose.langfuse-fanout.yml index 54a30e33df..15f57c1d88 100644 --- a/docker-compose.langfuse-fanout.yml +++ b/docker-compose.langfuse-fanout.yml @@ -12,6 +12,42 @@ services: - langfuse-fanout langfuse-fanout-collector: + build: + context: . + dockerfile: otel/langfuse-fanout/Dockerfile + image: librechat-langfuse-fanout:local + restart: always + depends_on: + - langfuse-fanout-otel + - langfuse-fanout-redis + env_file: + - .env + environment: + - LANGFUSE_FANOUT_CENTRAL_BASE_URL=${LANGFUSE_FANOUT_CENTRAL_BASE_URL:-https://cloud.langfuse.com} + - LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=${LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER:?Set LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER to central Langfuse Basic auth header} + - LANGFUSE_FANOUT_TENANT_DESTINATIONS=${LANGFUSE_FANOUT_TENANT_DESTINATIONS:-eu=https://cloud.langfuse.com,us=https://us.cloud.langfuse.com,jp=https://jp.cloud.langfuse.com} + - LANGFUSE_FANOUT_UPSTREAM_TIMEOUT=${LANGFUSE_FANOUT_UPSTREAM_TIMEOUT:-30s} + - LANGFUSE_FANOUT_PUBLIC_URL=${LANGFUSE_FANOUT_PUBLIC_URL:-http://langfuse-fanout-collector:4318} + - LANGFUSE_FANOUT_TRACE_COLLECTOR_URL=http://langfuse-fanout-otel:4319 + - LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS=eu,us,jp + - LANGFUSE_FANOUT_REDIS_URI=${LANGFUSE_FANOUT_REDIS_URI:-redis://langfuse-fanout-redis:6379} + - LANGFUSE_FANOUT_REDIS_USERNAME=${LANGFUSE_FANOUT_REDIS_USERNAME:-} + - LANGFUSE_FANOUT_REDIS_PASSWORD=${LANGFUSE_FANOUT_REDIS_PASSWORD:-} + - LANGFUSE_FANOUT_REDIS_KEY_PREFIX=${LANGFUSE_FANOUT_REDIS_KEY_PREFIX:-langfuse-fanout} + expose: + - '4318' + networks: + - langfuse-fanout + + langfuse-fanout-redis: + image: redis:7.4-alpine + restart: always + expose: + - '6379' + networks: + - langfuse-fanout + + langfuse-fanout-otel: image: otel/opentelemetry-collector-contrib:0.143.0 restart: always command: ['--config=/etc/otelcol/otelcol.yaml'] @@ -20,6 +56,7 @@ services: environment: - LANGFUSE_FANOUT_CENTRAL_BASE_URL=${LANGFUSE_FANOUT_CENTRAL_BASE_URL:-https://cloud.langfuse.com} - LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=${LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER:?Set LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER to central Langfuse Basic auth header} + - LANGFUSE_FANOUT_OTEL_RECEIVER_ENDPOINT=0.0.0.0:4319 - LANGFUSE_FANOUT_TENANT_EU_BASE_URL=${LANGFUSE_FANOUT_TENANT_EU_BASE_URL:-https://cloud.langfuse.com} - LANGFUSE_FANOUT_TENANT_US_BASE_URL=${LANGFUSE_FANOUT_TENANT_US_BASE_URL:-https://us.cloud.langfuse.com} - LANGFUSE_FANOUT_TENANT_JP_BASE_URL=${LANGFUSE_FANOUT_TENANT_JP_BASE_URL:-https://jp.cloud.langfuse.com} @@ -28,10 +65,10 @@ services: - LANGFUSE_FANOUT_BATCH_TIMEOUT=${LANGFUSE_FANOUT_BATCH_TIMEOUT:-1s} - LANGFUSE_FANOUT_BATCH_SEND_SIZE=${LANGFUSE_FANOUT_BATCH_SEND_SIZE:-128} - LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT=${LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT:-1000} - expose: - - '4318' volumes: - ./otel/langfuse-fanout/otelcol.yaml:/etc/otelcol/otelcol.yaml:ro + expose: + - '4319' networks: - langfuse-fanout diff --git a/helm/librechat/readme.md b/helm/librechat/readme.md index 19699cb1f6..e30addf8c5 100755 --- a/helm/librechat/readme.md +++ b/helm/librechat/readme.md @@ -56,24 +56,42 @@ https:///api/admin/oauth/openid/callback ## Langfuse Fanout -The chart can optionally deploy an OpenTelemetry Collector that forwards -tenant-scoped Langfuse traces to both a central Langfuse project and the tenant -Langfuse project. It is disabled by default. +The chart can optionally deploy a Langfuse fanout gateway with an internal +OpenTelemetry Collector sidecar. The gateway handles Langfuse media fanout and +proxies traces to the collector; the collector forwards tenant-scoped Langfuse +traces to both a central Langfuse project and the tenant Langfuse project. It is +disabled by default. When enabled, the chart also sets `LANGFUSE_FANOUT_ENABLED` and `LANGFUSE_FANOUT_COLLECTOR_URL` for the LibreChat app unless those values are already provided in `librechat.configEnv`. Set `librechat.configEnv.LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED=true` to keep -central trace export flowing through the collector while disabling tenant trace +central trace export flowing through the fanout gateway while disabling tenant trace and score export. When omitted, false, or blank, tenant export remains available if tenant keys and a known destination are configured. Langfuse tenant base URLs are selected from the startup-configured destination -map rendered into LibreChat and the collector. Tenant API keys can still be added +map rendered into LibreChat and the fanout gateway. Tenant API keys can still be added through tenant app configuration at runtime without restarting either component. -Collector batching can be tuned with `langfuseFanout.batchTimeout` and -`langfuseFanout.batchSendSize`; defaults are `1s` and `128`. +The internal collector provides trace memory limiting, batching, tenant routing, +and removal of LibreChat-only routing attributes before export. + +The fanout gateway stores one-time media upload plans in Redis so media create +and byte-upload requests can land on different gateway replicas. Set +`langfuseFanout.redis.uri` for an external Redis service, or enable the bundled +Redis chart with `redis.enabled=true` and let the chart derive the internal URI. +Scale the gateway manually with `langfuseFanout.replicaCount`; the chart does +not create a fanout HPA. +The internal collector receiver is bound to `127.0.0.1:4319` by default because +only the gateway sidecar should send traces to it. + +The gateway exposes Prometheus metrics at `/metrics`. Configure +`langfuseFanout.metrics.secret.name` and `.key` to pass a bearer token secret to +the gateway; if omitted, `/metrics` returns 401. Use +`langfuseFanout.service.annotations` for scrape annotations when your cluster +uses annotation-based discovery. The gateway container also has configurable +`/healthz` liveness and readiness probes under `langfuseFanout`. See [`otel/langfuse-fanout/README.md`](../../otel/langfuse-fanout/README.md) for the central Langfuse secret and values example. diff --git a/helm/librechat/templates/_helpers.tpl b/helm/librechat/templates/_helpers.tpl index 9cf16894f1..140b7470e4 100755 --- a/helm/librechat/templates/_helpers.tpl +++ b/helm/librechat/templates/_helpers.tpl @@ -86,6 +86,38 @@ already use the same lowercase key shape for collector routing to match. {{- end -}} {{- end }} +{{/* +Render the fanout destination list consumed by LibreChat and the fanout gateway. +*/}} +{{- define "librechat.langfuseFanout.tenantDestinationsEnv" -}} +{{- $tenantDestinations := list -}} +{{- range $name, $destination := .Values.langfuseFanout.tenant.destinations -}} +{{- include "librechat.langfuseFanout.validateDestinationKey" $name -}} +{{- $tenantDestinations = append $tenantDestinations (printf "%s=%s" $name $destination.baseUrl) -}} +{{- end -}} +{{- join "," $tenantDestinations -}} +{{- end }} + +{{/* +Render the fanout destination key list consumed by the gateway as a startup +guard against media destinations the collector cannot route traces to. +*/}} +{{- define "librechat.langfuseFanout.tenantDestinationKeysEnv" -}} +{{- $tenantDestinationKeys := list -}} +{{- range $name, $_destination := .Values.langfuseFanout.tenant.destinations -}} +{{- include "librechat.langfuseFanout.validateDestinationKey" $name -}} +{{- $tenantDestinationKeys = append $tenantDestinationKeys $name -}} +{{- end -}} +{{- join "," $tenantDestinationKeys -}} +{{- end }} + +{{/* +Bundled Redis URI used when the Redis subchart is enabled. +*/}} +{{- define "librechat.bundledRedisURI" -}} +{{- printf "redis://%s-master.%s.svc.cluster.local:6379" (include "common.names.fullname" .Subcharts.redis) (.Release.Namespace | lower) -}} +{{- end }} + {{/* RAG Selector labels */}} diff --git a/helm/librechat/templates/configmap-env.yaml b/helm/librechat/templates/configmap-env.yaml index 12742d69ea..2865b498f3 100755 --- a/helm/librechat/templates/configmap-env.yaml +++ b/helm/librechat/templates/configmap-env.yaml @@ -19,7 +19,7 @@ data: USE_REDIS: "true" {{- end }} {{- if and (not (dig "configEnv" "REDIS_URI" "" .Values.librechat)) .Values.redis.enabled }} - REDIS_URI: redis://{{ include "common.names.fullname" .Subcharts.redis }}-master.{{ .Release.Namespace | lower }}.svc.cluster.local:6379 + REDIS_URI: {{ include "librechat.bundledRedisURI" . }} {{- end }} {{- if and $adminPanelUrl (not $configAdminPanelUrl) }} ADMIN_PANEL_URL: {{ $adminPanelUrl | quote }} diff --git a/helm/librechat/templates/langfuse-fanout-configmap.yaml b/helm/librechat/templates/langfuse-fanout-configmap.yaml index bba55a42ea..6d061877ad 100644 --- a/helm/librechat/templates/langfuse-fanout-configmap.yaml +++ b/helm/librechat/templates/langfuse-fanout-configmap.yaml @@ -18,7 +18,7 @@ data: otlp: protocols: http: - endpoint: 0.0.0.0:4318 + endpoint: ${env:LANGFUSE_FANOUT_OTEL_RECEIVER_ENDPOINT} include_metadata: true traces_url_path: /api/public/otel/v1/traces diff --git a/helm/librechat/templates/langfuse-fanout-deployment.yaml b/helm/librechat/templates/langfuse-fanout-deployment.yaml index 84c360f4bb..13564130e2 100644 --- a/helm/librechat/templates/langfuse-fanout-deployment.yaml +++ b/helm/librechat/templates/langfuse-fanout-deployment.yaml @@ -1,4 +1,15 @@ {{- if .Values.langfuseFanout.enabled }} +{{- $redisURI := .Values.langfuseFanout.redis.uri }} +{{- if and (not $redisURI) .Values.redis.enabled }} +{{- $redisURI = include "librechat.bundledRedisURI" . }} +{{- end }} +{{- if not $redisURI }} +{{- fail "langfuseFanout.redis.uri is required when langfuseFanout.enabled=true unless redis.enabled=true" }} +{{- end }} +{{- $publicURL := .Values.langfuseFanout.publicUrl }} +{{- if not $publicURL }} +{{- $publicURL = printf "http://%s.%s.svc.cluster.local:%v" (include "librechat.langfuseFanout.fullname" .) (.Release.Namespace | lower) .Values.langfuseFanout.service.port }} +{{- end }} apiVersion: apps/v1 kind: Deployment metadata: @@ -13,7 +24,7 @@ spec: template: metadata: annotations: - checksum/langfuse-fanout-config: {{ include (print $.Template.BasePath "/langfuse-fanout-configmap.yaml") . | sha256sum }} + checksum/langfuse-fanout-config: {{ toYaml .Values.langfuseFanout | sha256sum }} {{- with .Values.langfuseFanout.podAnnotations }} {{- toYaml . | nindent 8 }} {{- end }} @@ -24,10 +35,9 @@ spec: {{- end }} spec: containers: - - name: otelcol + - name: langfuse-fanout image: "{{ .Values.langfuseFanout.image.repository }}:{{ .Values.langfuseFanout.image.tag }}" imagePullPolicy: {{ .Values.langfuseFanout.image.pullPolicy }} - args: ["--config=/etc/otelcol/otelcol.yaml"] ports: - name: otlp-http containerPort: 4318 @@ -40,6 +50,66 @@ spec: secretKeyRef: name: {{ required "langfuseFanout.central.authHeaderSecret.name is required when langfuseFanout.enabled=true" .Values.langfuseFanout.central.authHeaderSecret.name | quote }} key: {{ .Values.langfuseFanout.central.authHeaderSecret.key | quote }} + {{- /* Validate destination keys before rendering the env map. */}} + {{- range $name, $_ := .Values.langfuseFanout.tenant.destinations }} + {{- include "librechat.langfuseFanout.validateDestinationKey" $name }} + {{- end }} + - name: LANGFUSE_FANOUT_TENANT_DESTINATIONS + value: {{ include "librechat.langfuseFanout.tenantDestinationsEnv" . | quote }} + - name: LANGFUSE_FANOUT_UPSTREAM_TIMEOUT + value: {{ .Values.langfuseFanout.upstreamTimeout | quote }} + - name: LANGFUSE_FANOUT_TRACE_COLLECTOR_URL + value: {{ .Values.langfuseFanout.traceCollectorUrl | quote }} + - name: LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS + value: {{ include "librechat.langfuseFanout.tenantDestinationKeysEnv" . | quote }} + - name: LANGFUSE_FANOUT_REDIS_URI + value: {{ $redisURI | quote }} + {{- with .Values.langfuseFanout.redis.username }} + - name: LANGFUSE_FANOUT_REDIS_USERNAME + value: {{ . | quote }} + {{- end }} + {{- with .Values.langfuseFanout.redis.passwordSecret.name }} + - name: LANGFUSE_FANOUT_REDIS_PASSWORD + valueFrom: + secretKeyRef: + name: {{ . | quote }} + key: {{ $.Values.langfuseFanout.redis.passwordSecret.key | quote }} + {{- end }} + - name: LANGFUSE_FANOUT_REDIS_KEY_PREFIX + value: {{ .Values.langfuseFanout.redis.keyPrefix | quote }} + - name: LANGFUSE_FANOUT_PUBLIC_URL + value: {{ $publicURL | quote }} + {{- with .Values.langfuseFanout.metrics.secret.name }} + - name: LANGFUSE_FANOUT_METRICS_SECRET + valueFrom: + secretKeyRef: + name: {{ . | quote }} + key: {{ $.Values.langfuseFanout.metrics.secret.key | quote }} + {{- end }} + livenessProbe: + {{- toYaml .Values.langfuseFanout.livenessProbe | nindent 12 }} + readinessProbe: + {{- toYaml .Values.langfuseFanout.readinessProbe | nindent 12 }} + resources: + {{- toYaml .Values.langfuseFanout.resources | nindent 12 }} + - name: otelcol + image: "{{ .Values.langfuseFanout.otelCollector.image.repository }}:{{ .Values.langfuseFanout.otelCollector.image.tag }}" + imagePullPolicy: {{ .Values.langfuseFanout.otelCollector.image.pullPolicy }} + args: ["--config=/etc/otelcol/otelcol.yaml"] + ports: + - name: otlp-internal + containerPort: 4319 + protocol: TCP + env: + - name: LANGFUSE_FANOUT_CENTRAL_BASE_URL + value: {{ .Values.langfuseFanout.central.baseUrl | quote }} + - name: LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER + valueFrom: + secretKeyRef: + name: {{ required "langfuseFanout.central.authHeaderSecret.name is required when langfuseFanout.enabled=true" .Values.langfuseFanout.central.authHeaderSecret.name | quote }} + key: {{ .Values.langfuseFanout.central.authHeaderSecret.key | quote }} + - name: LANGFUSE_FANOUT_OTEL_RECEIVER_ENDPOINT + value: {{ .Values.langfuseFanout.otelCollector.receiverEndpoint | quote }} {{- range $name, $destination := .Values.langfuseFanout.tenant.destinations }} {{- include "librechat.langfuseFanout.validateDestinationKey" $name }} - name: LANGFUSE_FANOUT_TENANT_{{ $name | upper | replace "-" "_" }}_BASE_URL @@ -61,7 +131,7 @@ spec: subPath: otelcol.yaml readOnly: true resources: - {{- toYaml .Values.langfuseFanout.resources | nindent 12 }} + {{- toYaml .Values.langfuseFanout.otelCollector.resources | nindent 12 }} volumes: - name: config configMap: diff --git a/helm/librechat/templates/langfuse-fanout-service.yaml b/helm/librechat/templates/langfuse-fanout-service.yaml index 10028c6309..ea625f653b 100644 --- a/helm/librechat/templates/langfuse-fanout-service.yaml +++ b/helm/librechat/templates/langfuse-fanout-service.yaml @@ -5,6 +5,12 @@ metadata: name: {{ include "librechat.langfuseFanout.fullname" . }} labels: {{- include "librechat.langfuseFanout.labels" . | nindent 4 }} + {{- with .Values.langfuseFanout.service.annotations }} + annotations: + {{- range $key, $value := . }} + {{ $key }}: {{ $value | quote }} + {{- end }} + {{- end }} spec: type: {{ .Values.langfuseFanout.service.type }} ports: diff --git a/helm/librechat/tests/langfuse_fanout_config_drift_test.sh b/helm/librechat/tests/langfuse_fanout_config_drift_test.sh deleted file mode 100755 index c3afc9ae1c..0000000000 --- a/helm/librechat/tests/langfuse_fanout_config_drift_test.sh +++ /dev/null @@ -1,212 +0,0 @@ -#!/usr/bin/env bash -# Regression test for Langfuse fanout collector config drift. -# -# Compose mounts otel/langfuse-fanout/otelcol.yaml directly. Helm renders the -# same collector topology into a ConfigMap from templates/langfuse-fanout-configmap.yaml. -# This test renders Helm with the default tenant destinations and compares the -# semantic collector topology, not raw YAML formatting. - -set -euo pipefail - -SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" -CHART_DIR="$(cd "${SCRIPT_DIR}/.." && pwd)" -REPO_ROOT="$(cd "${CHART_DIR}/../.." && pwd)" -STATIC_CONFIG="${REPO_ROOT}/otel/langfuse-fanout/otelcol.yaml" -RENDER_CHART_DIR="$(mktemp -d -t librechat-fanout-chart.XXXXXX)" -RENDERED_FILE="$(mktemp -t librechat-fanout-config-render.XXXXXX)" -STATIC_SUMMARY="$(mktemp -t librechat-fanout-static-summary.XXXXXX)" -HELM_SUMMARY="$(mktemp -t librechat-fanout-helm-summary.XXXXXX)" -trap 'rm -rf "${RENDER_CHART_DIR}"; rm -f "${RENDERED_FILE}" "${STATIC_SUMMARY}" "${HELM_SUMMARY}"' EXIT - -if ! command -v helm >/dev/null 2>&1; then - echo "FAIL: helm not on PATH" >&2 - exit 1 -fi - -if ! command -v node >/dev/null 2>&1; then - echo "FAIL: node not on PATH" >&2 - exit 1 -fi - -mkdir -p "${RENDER_CHART_DIR}/templates" -awk '/^dependencies:/{ exit } { print }' "${CHART_DIR}/Chart.yaml" > "${RENDER_CHART_DIR}/Chart.yaml" -cp "${CHART_DIR}/values.yaml" "${RENDER_CHART_DIR}/values.yaml" -cp "${CHART_DIR}/templates/_helpers.tpl" "${RENDER_CHART_DIR}/templates/_helpers.tpl" -cp "${CHART_DIR}/templates/langfuse-fanout-configmap.yaml" \ - "${RENDER_CHART_DIR}/templates/langfuse-fanout-configmap.yaml" - -helm template librechat "${RENDER_CHART_DIR}" \ - --set langfuseFanout.enabled=true \ - --set langfuseFanout.central.authHeaderSecret.name=langfuse-central \ - --show-only templates/langfuse-fanout-configmap.yaml \ - > "${RENDERED_FILE}" - -NODE_PATH="${REPO_ROOT}/node_modules${NODE_PATH:+:${NODE_PATH}}" \ -STATIC_CONFIG="${STATIC_CONFIG}" \ -RENDERED_FILE="${RENDERED_FILE}" \ -STATIC_SUMMARY="${STATIC_SUMMARY}" \ -HELM_SUMMARY="${HELM_SUMMARY}" \ -node <<'NODE' -const fs = require('fs'); -const yaml = require('js-yaml'); - -const DEFAULT_DESTINATIONS = ['eu', 'jp', 'us']; - -function fail(message) { - console.error(`FAIL: ${message}`); - process.exit(1); -} - -function requireObject(value, path) { - if (value == null || typeof value !== 'object' || Array.isArray(value)) { - fail(`${path} must be an object`); - } - return value; -} - -function sorted(value) { - return [...(value ?? [])].sort(); -} - -function extractHelmCollectorConfig(renderedFile) { - const docs = yaml.loadAll(fs.readFileSync(renderedFile, 'utf8')).filter(Boolean); - const configMap = docs.find( - (doc) => - doc.kind === 'ConfigMap' && - doc.metadata?.name === 'librechat-librechat-langfuse-fanout-config', - ); - if (!configMap) { - fail('missing rendered Langfuse fanout ConfigMap'); - } - const body = configMap.data?.['otelcol.yaml']; - if (typeof body !== 'string' || body.trim() === '') { - fail('rendered ConfigMap is missing data.otelcol.yaml'); - } - return yaml.load(body); -} - -function extractRouteDestinations(config) { - const routes = config.connectors?.['routing/langfuse_tenant_destination']?.table; - if (!Array.isArray(routes)) { - fail('routing/langfuse_tenant_destination.table must be an array'); - } - - return Object.fromEntries( - routes.map((route) => { - const condition = route.condition; - const match = /attributes\["librechat\.langfuse\.destination"\]\s*==\s*"([^"]+)"/.exec( - condition, - ); - if (!match) { - fail(`unsupported routing condition: ${condition}`); - } - return [ - match[1], - { - context: route.context, - pipelines: sorted(route.pipelines), - }, - ]; - }).sort(([left], [right]) => left.localeCompare(right)), - ); -} - -function summarize(config) { - requireObject(config, 'collector config'); - const processors = requireObject(config.processors, 'processors'); - const exporters = requireObject(config.exporters, 'exporters'); - const pipelines = requireObject(config.service?.pipelines, 'service.pipelines'); - - const destinations = Object.keys(extractRouteDestinations(config)).sort(); - const tenantExporters = Object.fromEntries( - destinations.map((destination) => { - const exporter = exporters[`otlphttp/tenant_${destination}`]; - return [ - destination, - { - endpoint: exporter?.endpoint, - authenticator: exporter?.auth?.authenticator, - headers: exporter?.headers ?? {}, - }, - ]; - }), - ); - const tenantPipelines = Object.fromEntries( - destinations.map((destination) => { - const pipeline = pipelines[`traces/tenant_${destination}`]; - return [ - destination, - { - receivers: sorted(pipeline?.receivers), - processors: sorted(pipeline?.processors), - exporters: sorted(pipeline?.exporters), - }, - ]; - }), - ); - const tenantBatches = Object.fromEntries( - destinations.map((destination) => { - const batch = processors[`batch/by_auth_${destination}`]; - return [ - destination, - { - timeout: batch?.timeout, - sendBatchSize: batch?.send_batch_size, - metadataKeys: sorted(batch?.metadata_keys), - metadataCardinalityLimit: batch?.metadata_cardinality_limit, - }, - ]; - }), - ); - - return { - destinations, - extensions: config.service?.extensions ?? [], - headersSetter: config.extensions?.['headers_setter/tenant_passthrough'] ?? {}, - receiver: config.receivers?.otlp?.protocols?.http ?? {}, - routes: extractRouteDestinations(config), - tenantExportFilter: processors['filter/tenant_export']?.traces?.span ?? [], - dropRoutingAttributes: processors['attributes/drop_librechat_routing']?.actions ?? [], - centralExporter: exporters['otlphttp/central'] ?? {}, - centralBatch: { - timeout: processors['batch/central']?.timeout, - sendBatchSize: processors['batch/central']?.send_batch_size, - }, - centralPipeline: pipelines['traces/central'] ?? {}, - tenantRouterPipeline: pipelines['traces/tenant'] ?? {}, - tenantExporters, - tenantPipelines, - tenantBatches, - }; -} - -const staticConfig = yaml.load(fs.readFileSync(process.env.STATIC_CONFIG, 'utf8')); -const helmConfig = extractHelmCollectorConfig(process.env.RENDERED_FILE); -const staticSummary = summarize(staticConfig); -const helmSummary = summarize(helmConfig); - -if (JSON.stringify(staticSummary.destinations) !== JSON.stringify(DEFAULT_DESTINATIONS)) { - fail( - `static collector destinations changed from default set: ${JSON.stringify( - staticSummary.destinations, - )}`, - ); -} -if (JSON.stringify(helmSummary.destinations) !== JSON.stringify(DEFAULT_DESTINATIONS)) { - fail( - `Helm default collector destinations changed from default set: ${JSON.stringify( - helmSummary.destinations, - )}`, - ); -} - -fs.writeFileSync(process.env.STATIC_SUMMARY, `${JSON.stringify(staticSummary, null, 2)}\n`); -fs.writeFileSync(process.env.HELM_SUMMARY, `${JSON.stringify(helmSummary, null, 2)}\n`); -NODE - -if ! diff -u "${STATIC_SUMMARY}" "${HELM_SUMMARY}"; then - echo "FAIL: static Compose collector config and Helm-rendered collector config drifted" >&2 - exit 1 -fi - -echo "PASS: Langfuse fanout Compose and Helm collector configs match for default destinations" diff --git a/helm/librechat/tests/langfuse_fanout_selector_test.sh b/helm/librechat/tests/langfuse_fanout_selector_test.sh index fc3987373f..abae0ff1f9 100755 --- a/helm/librechat/tests/langfuse_fanout_selector_test.sh +++ b/helm/librechat/tests/langfuse_fanout_selector_test.sh @@ -28,23 +28,22 @@ cp "${CHART_DIR}/templates/langfuse-fanout-service.yaml" \ "${RENDER_CHART_DIR}/templates/langfuse-fanout-service.yaml" cp "${CHART_DIR}/templates/langfuse-fanout-deployment.yaml" \ "${RENDER_CHART_DIR}/templates/langfuse-fanout-deployment.yaml" -cp "${CHART_DIR}/templates/langfuse-fanout-configmap.yaml" \ - "${RENDER_CHART_DIR}/templates/langfuse-fanout-configmap.yaml" helm template librechat "${RENDER_CHART_DIR}" \ --set langfuseFanout.enabled=true \ --set langfuseFanout.central.authHeaderSecret.name=langfuse-central \ + --set langfuseFanout.redis.uri=redis://langfuse-fanout-redis:6379 \ --show-only templates/service.yaml \ --show-only templates/langfuse-fanout-service.yaml \ --show-only templates/langfuse-fanout-deployment.yaml \ - --show-only templates/langfuse-fanout-configmap.yaml \ > "${RENDERED_FILE}" if helm template librechat "${RENDER_CHART_DIR}" \ --set langfuseFanout.enabled=true \ --set langfuseFanout.central.authHeaderSecret.name=langfuse-central \ + --set langfuseFanout.redis.uri=redis://langfuse-fanout-redis:6379 \ --set langfuseFanout.tenant.destinations.EU.baseUrl=https://cloud.langfuse.com \ - --show-only templates/langfuse-fanout-configmap.yaml \ + --show-only templates/langfuse-fanout-deployment.yaml \ > /dev/null 2> "${INVALID_RENDER_ERROR}"; then echo "FAIL: Helm accepted invalid uppercase Langfuse fanout destination key" >&2 exit 1 @@ -87,17 +86,25 @@ function isSubset(subset, labels) { return Object.entries(subset ?? {}).every(([key, value]) => labels?.[key] === value); } +function envValue(env, name) { + return (env ?? []).find((entry) => entry.name === name)?.value; +} + const mainService = find('Service', 'librechat-librechat'); const fanoutService = find('Service', 'librechat-librechat-langfuse-fanout'); const fanoutDeployment = find('Deployment', 'librechat-librechat-langfuse-fanout'); -const fanoutConfigMap = find('ConfigMap', 'librechat-librechat-langfuse-fanout-config'); +const fanoutContainer = fanoutDeployment.spec?.template?.spec?.containers?.find( + (container) => container.name === 'langfuse-fanout', +); +if (!fanoutContainer) { + fail('missing langfuse-fanout container'); +} const mainSelector = mainService.spec?.selector ?? {}; const fanoutSelector = fanoutService.spec?.selector ?? {}; const fanoutMatchLabels = fanoutDeployment.spec?.selector?.matchLabels ?? {}; const fanoutPodLabels = fanoutDeployment.spec?.template?.metadata?.labels ?? {}; const fanoutMetadataLabels = fanoutDeployment.metadata?.labels ?? {}; -const fanoutConfigLabels = fanoutConfigMap.metadata?.labels ?? {}; if (isSubset(mainSelector, fanoutPodLabels)) { fail('main Service selector matches fanout pod labels'); @@ -114,8 +121,20 @@ if (mainSelector['app.kubernetes.io/name'] === fanoutSelector['app.kubernetes.io if (fanoutMetadataLabels['app.kubernetes.io/name'] !== fanoutSelector['app.kubernetes.io/name']) { fail('fanout Deployment metadata labels do not use fanout app name'); } -if (fanoutConfigLabels['app.kubernetes.io/name'] !== fanoutSelector['app.kubernetes.io/name']) { - fail('fanout ConfigMap labels do not use fanout app name'); +if (envValue(fanoutContainer.env, 'LANGFUSE_FANOUT_REDIS_URI') !== 'redis://langfuse-fanout-redis:6379') { + fail('fanout Deployment did not render configured Redis URI'); +} +if ( + envValue(fanoutContainer.env, 'LANGFUSE_FANOUT_PUBLIC_URL') !== + 'http://librechat-librechat-langfuse-fanout.default.svc.cluster.local:4318' +) { + fail('fanout Deployment did not render derived public URL'); +} +if (fanoutContainer.livenessProbe?.httpGet?.path !== '/healthz') { + fail('fanout Deployment missing /healthz liveness probe'); +} +if (fanoutContainer.readinessProbe?.httpGet?.path !== '/healthz') { + fail('fanout Deployment missing /healthz readiness probe'); } console.log('PASS: Langfuse fanout selectors are isolated from the main LibreChat Service'); diff --git a/helm/librechat/values.yaml b/helm/librechat/values.yaml index 88e7150a61..bcefc74342 100755 --- a/helm/librechat/values.yaml +++ b/helm/librechat/values.yaml @@ -288,20 +288,44 @@ langfuseFanout: enabled: false replicaCount: 1 image: - repository: otel/opentelemetry-collector-contrib - tag: "0.143.0" + repository: librechat-langfuse-fanout + tag: "latest" pullPolicy: IfNotPresent + otelCollector: + receiverEndpoint: 127.0.0.1:4319 + image: + repository: otel/opentelemetry-collector-contrib + tag: "0.143.0" + pullPolicy: IfNotPresent + resources: {} service: type: ClusterIP port: 4318 + annotations: {} central: baseUrl: https://cloud.langfuse.com authHeaderSecret: name: "" key: LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER + metrics: + secret: + # Optional bearer token secret for scraping the gateway's /metrics endpoint. + # When omitted, /metrics returns 401. + name: "" + key: METRICS_SECRET + redis: + # Redis stores short-lived Langfuse media upload plans so multiple gateway + # replicas can handle create/upload requests. If empty and redis.enabled is + # true, the chart derives the bundled Redis service URI. + uri: "" + username: "" + passwordSecret: + name: "" + key: REDIS_PASSWORD + keyPrefix: langfuse-fanout tenant: # Destination map keys are emitted by LibreChat as trace attributes and - # matched by the collector. Use lowercase keys matching ^[a-z][a-z0-9_-]*$. + # matched by the gateway. Use lowercase keys matching ^[a-z][a-z0-9_-]*$. destinations: eu: baseUrl: https://cloud.langfuse.com @@ -309,11 +333,24 @@ langfuseFanout: baseUrl: https://us.cloud.langfuse.com jp: baseUrl: https://jp.cloud.langfuse.com + upstreamTimeout: 30s + # Optional override for one-time media upload URLs returned by the gateway. + # When empty, the chart derives the internal fanout Service URL. + publicUrl: "" + traceCollectorUrl: http://127.0.0.1:4319 memoryLimitMiB: 256 memorySpikeLimitMiB: 64 batchTimeout: 1s batchSendSize: 128 metadataCardinalityLimit: 1000 + livenessProbe: + httpGet: + path: /healthz + port: 4318 + readinessProbe: + httpGet: + path: /healthz + port: 4318 resources: {} podAnnotations: {} podLabels: {} diff --git a/otel/langfuse-fanout/Dockerfile b/otel/langfuse-fanout/Dockerfile new file mode 100644 index 0000000000..22be19c5c7 --- /dev/null +++ b/otel/langfuse-fanout/Dockerfile @@ -0,0 +1,13 @@ +FROM golang:1.23-alpine AS builder +WORKDIR /src +COPY otel/langfuse-fanout/go.mod otel/langfuse-fanout/go.sum* ./ +RUN go mod download +COPY otel/langfuse-fanout/ ./ +RUN CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" -o /out/langfuse-fanout ./cmd/langfuse-fanout + +FROM alpine:3.22 +RUN addgroup -S app && adduser -S app -G app +COPY --from=builder /out/langfuse-fanout /usr/local/bin/langfuse-fanout +USER app +EXPOSE 4318 +ENTRYPOINT ["/usr/local/bin/langfuse-fanout"] diff --git a/otel/langfuse-fanout/README.md b/otel/langfuse-fanout/README.md index 74c4c97971..f78fa16bab 100644 --- a/otel/langfuse-fanout/README.md +++ b/otel/langfuse-fanout/README.md @@ -1,59 +1,106 @@ -# Langfuse Fanout Collector +# Langfuse Fanout Gateway LibreChat can send tenant-scoped agent traces to a tenant Langfuse project and -also copy those traces to a central Langfuse project. This is optional and is -disabled unless you explicitly deploy the fanout collector. +also copy those traces to a central Langfuse project. When trace payloads +contain Langfuse media references, the gateway also copies the media upload to +central and tenant Langfuse storage. This is optional and is disabled unless you +explicitly deploy the fanout gateway. + +The deployment is a hybrid: + +- the Go gateway is the only endpoint LibreChat talks to; +- trace requests are proxied to an internal OpenTelemetry collector; +- the collector owns trace memory limiting, batching, routing, and export; +- the Go gateway owns Langfuse media create/upload/patch fanout. ## How It Works - Agent traces use Langfuse OTLP ingestion. -- LibreChat sends tenant traces to the local collector when +- LibreChat sends tenant traces to the local fanout gateway when `LANGFUSE_FANOUT_ENABLED=true` and `LANGFUSE_FANOUT_COLLECTOR_URL` points at - the collector. + the fanout gateway. +- The gateway forwards trace requests to the internal OpenTelemetry collector + at `LANGFUSE_FANOUT_TRACE_COLLECTOR_URL`. - The collector exports every trace to the central Langfuse project using `LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER`. This prebuilt header is collector-only; the LibreChat app derives central score auth from `LANGFUSE_PUBLIC_KEY` and `LANGFUSE_SECRET_KEY`. -- The collector also exports the same trace to the tenant Langfuse project by - routing a LibreChat-stamped destination key to one of the configured tenant - Langfuse base URLs, then forwarding the tenant `Authorization` header that - LibreChat attaches to the OTLP request. -- Tenant export is conditional. LibreChat marks traces as tenant-exportable only - when tenant keys are configured and `LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED` - is not true; unmarked traces are still exported to central but are dropped by - the tenant pipeline. -- LibreChat's routing attributes are consumed by the collector and deleted - before central or tenant export, so they are not forwarded to Langfuse. +- The collector also exports tenant-enabled traces to the tenant Langfuse + project by routing on `librechat.langfuse.destination`, then forwarding the + tenant `Authorization` header that LibreChat attaches to the OTLP request. +- For tenant-exportable runs, LibreChat uses a destination-scoped gateway URL + like `http://langfuse-fanout-collector:4318/tenant/us`. Langfuse media upload + requests do not carry span attributes, so this path gives the gateway the + destination needed to copy media into the tenant's Langfuse region. For + traces on this path, the gateway restores the internal tenant routing + attributes before handing the request to the collector. +- Before export, the collector deletes the internal `librechat.langfuse.*` + routing attributes from central and tenant traces. +- Langfuse media upload is fanned out by calling `POST /api/public/media` on + central and tenant Langfuse, returning a one-time gateway upload URL, then + uploading the received bytes to each upstream presigned upload URL. The SDK's + `PATCH /api/public/media/{mediaId}` status call is also fanned out. +- Tenant export is conditional. LibreChat uses a destination-scoped gateway URL + only when tenant keys are configured, the tenant base URL matches a configured + startup destination, and `LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED` is not true. + Other traces are still exported to central through the gateway without tenant + auth. - User feedback scores use Langfuse's direct REST API from the LibreChat API process. Central scores use LibreChat's normal central Langfuse env config; tenant scores use tenant app configuration when tenant fanout is enabled. Tenant Langfuse keys are expected to come from LibreChat app configuration, for example from an admin panel or another configuration data source. They are not -defined in this collector config. +defined in this gateway config. ## Limitations - Langfuse base URLs are startup configuration. `LANGFUSE_FANOUT_CENTRAL_BASE_URL` - and the tenant destination map must be known when LibreChat and the collector - start. Tenant app configuration may choose any configured tenant destination. + and `LANGFUSE_FANOUT_TENANT_DESTINATIONS` must be known when LibreChat and the + gateway start. Tenant app configuration may choose any configured tenant + destination. - Tenant Langfuse API keys can be added, changed, or disabled in tenant app - configuration at runtime without restarting LibreChat or the collector. + configuration at runtime without restarting LibreChat or the gateway. - Tenant app configuration must set a Langfuse base URL matching one of the startup destinations before tenant trace/score export is enabled; keys alone are treated as central-only. - `LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED=true` can be set on LibreChat as an emergency switch to stop tenant trace and score export while keeping central - collector export active. When omitted, false, or blank, tenant export remains + gateway export active. When omitted, false, or blank, tenant export remains available if tenant keys and a known destination are configured. - This supports Langfuse Cloud and self-hosted Langfuse as long as each allowed - tenant base URL is configured at LibreChat/collector startup. Runtime tenant + tenant base URL is configured at LibreChat/gateway startup. Runtime tenant config selects from those known destinations; it does not inject arbitrary - export URLs into the collector. -- The provided Compose collector config is a three-region Langfuse Cloud preset - (`eu`, `us`, `jp`). For self-hosted or additional destination keys, - use Helm values or update/generate the collector routing table and exporters - in lockstep with `LANGFUSE_FANOUT_TENANT_DESTINATIONS`. + export URLs into the gateway. +- The provided Compose gateway config is a three-region Langfuse Cloud preset + (`eu`, `us`, `jp`). Compose's static collector config routes only those keys; + the gateway fails startup when `LANGFUSE_FANOUT_TENANT_DESTINATIONS` contains + a key outside `LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS`. For self-hosted or + additional destination keys, update the collector config too or use Helm. +- Helm binds the internal collector receiver to `127.0.0.1:4319` because the + collector is a sidecar. Compose binds it to `0.0.0.0:4319` on the private + `langfuse-fanout` network. Do not publish the internal collector receiver + outside the fanout deployment; tenant routing validation happens in the + gateway before traces reach the collector. +- The gateway stores short-lived one-time media upload plans in Redis. This lets + media create and byte-upload requests land on different gateway replicas. + Compose includes a private Redis container; Helm can derive the URI from the + bundled Redis chart or use an explicit `langfuseFanout.redis.uri`. +- The gateway requires an explicit public/internal base URL for one-time upload + URLs. Compose sets `LANGFUSE_FANOUT_PUBLIC_URL` to its private gateway + service URL. Helm derives the fanout Service DNS name unless `publicUrl` is + set. +- Media fanout is not transactional across central and tenant projects. If one + destination accepts `POST /api/public/media` and another fails, LibreChat sees + a gateway error and will not upload bytes, but the successful destination may + retain a short-lived, unused media record. +- Trace batching is handled by the collector. By default it flushes after 128 + items or 1 second, and tenant batches are separated by the request + `Authorization` metadata. +- The gateway exposes Prometheus metrics at `/metrics` using the same bearer + token shape as LibreChat. Set `LANGFUSE_FANOUT_METRICS_SECRET`, or provide + `METRICS_SECRET` in the gateway environment. When neither is set, `/metrics` + returns 401. ## Docker Compose @@ -64,18 +111,29 @@ Set the central Langfuse destination in `.env`: # region as LANGFUSE_FANOUT_CENTRAL_BASE_URL when applicable. LANGFUSE_BASE_URL=https://cloud.langfuse.com -# Used by the collector for central trace export. +# Used by the gateway for central trace and media export. LANGFUSE_FANOUT_CENTRAL_BASE_URL=https://cloud.langfuse.com LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=Basic -# Compose's included collector config supports these three destination keys. -# Do not add a custom key unless the collector routing table also has a matching pipeline/exporter. +# Compose's included gateway config supports these three destination keys. LANGFUSE_FANOUT_TENANT_DESTINATIONS=eu=https://cloud.langfuse.com,us=https://us.cloud.langfuse.com,jp=https://jp.cloud.langfuse.com +LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS=eu,us,jp LANGFUSE_FANOUT_TENANT_EU_BASE_URL=https://cloud.langfuse.com LANGFUSE_FANOUT_TENANT_US_BASE_URL=https://us.cloud.langfuse.com LANGFUSE_FANOUT_TENANT_JP_BASE_URL=https://jp.cloud.langfuse.com LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED=false +LANGFUSE_FANOUT_UPSTREAM_TIMEOUT=30s +LANGFUSE_FANOUT_PUBLIC_URL=http://langfuse-fanout-collector:4318 +LANGFUSE_FANOUT_REDIS_URI=redis://langfuse-fanout-redis:6379 +LANGFUSE_FANOUT_REDIS_USERNAME= +LANGFUSE_FANOUT_REDIS_PASSWORD= +LANGFUSE_FANOUT_REDIS_KEY_PREFIX=langfuse-fanout +LANGFUSE_FANOUT_OTEL_RECEIVER_ENDPOINT=0.0.0.0:4319 +LANGFUSE_FANOUT_METRICS_SECRET= +LANGFUSE_FANOUT_MEMORY_LIMIT_MIB=256 +LANGFUSE_FANOUT_MEMORY_SPIKE_LIMIT_MIB=64 LANGFUSE_FANOUT_BATCH_TIMEOUT=1s LANGFUSE_FANOUT_BATCH_SEND_SIZE=128 +LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT=1000 ``` Langfuse Cloud base URL options: @@ -98,8 +156,9 @@ For the deployed compose stack: docker compose -f deploy-compose.yml -f deploy-compose.langfuse-fanout.yml up -d ``` -The override sets `LANGFUSE_FANOUT_ENABLED=true` and points LibreChat at -`http://langfuse-fanout-collector:4318`. +The override builds the fanout gateway image, sets `LANGFUSE_FANOUT_ENABLED=true`, and points LibreChat at +`http://langfuse-fanout-collector:4318`. It also starts an internal +`langfuse-fanout-otel` service on the private fanout network for trace export. ## Helm @@ -110,9 +169,13 @@ kubectl create secret generic langfuse-central \ --from-literal=LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER='Basic ' ``` -Enable the collector in values: +Enable the gateway in values. Use either the bundled Redis chart as shown here +or set `langfuseFanout.redis.uri` to an external Redis service. ```yaml +redis: + enabled: true + langfuseFanout: enabled: true central: @@ -120,6 +183,10 @@ langfuseFanout: authHeaderSecret: name: langfuse-central key: LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER + metrics: + secret: + name: librechat-metrics + key: METRICS_SECRET tenant: destinations: eu: @@ -128,30 +195,91 @@ langfuseFanout: baseUrl: https://us.cloud.langfuse.com jp: baseUrl: https://jp.cloud.langfuse.com + upstreamTimeout: 30s + publicUrl: "" + otelCollector: + receiverEndpoint: 127.0.0.1:4319 + redis: + uri: "" + username: "" + passwordSecret: + name: "" + key: REDIS_PASSWORD + keyPrefix: langfuse-fanout + memoryLimitMiB: 256 + memorySpikeLimitMiB: 64 batchTimeout: 1s batchSendSize: 128 + metadataCardinalityLimit: 1000 ``` -The chart renders the collector Deployment, Service, and collector ConfigMap, -and injects `LANGFUSE_FANOUT_ENABLED` plus `LANGFUSE_FANOUT_COLLECTOR_URL` into -the LibreChat app ConfigMap when they are not already supplied in -`librechat.configEnv`. +The chart renders one fanout Deployment with two containers: the gateway on +`4318` and an internal OpenTelemetry collector on `4319`. The Service exposes +only the gateway. The chart also injects `LANGFUSE_FANOUT_ENABLED` plus +`LANGFUSE_FANOUT_COLLECTOR_URL` into the LibreChat app ConfigMap when they are +not already supplied in `librechat.configEnv`. + +Set `langfuseFanout.redis.uri` when using an external Redis service. If Redis +requires auth, set `langfuseFanout.redis.username` and point +`langfuseFanout.redis.passwordSecret.name`/`.key` at an existing Kubernetes +Secret. When using the bundled Redis chart with auth enabled, create a password +Secret for the gateway or provide an explicit authenticated URI. +Prefer `passwordSecret` over embedding credentials in `redis.uri`, because the +URI is rendered directly into the Deployment environment. +Scale the gateway manually with `langfuseFanout.replicaCount`; the chart does +not create a fanout HPA. The gateway container has configurable `/healthz` +liveness and readiness probes under `langfuseFanout`. + +Useful gateway metrics include: + +- `langfuse_fanout_http_requests_total` +- `langfuse_fanout_upstream_requests_total` +- `langfuse_fanout_trace_exports_total` +- `langfuse_fanout_media_upload_plans_created_total` +- `langfuse_fanout_media_upload_plans_completed_total` +- `langfuse_fanout_media_upload_plan_misses_total` +- `langfuse_fanout_media_upload_plan_store_errors_total` +- `langfuse_fanout_media_upload_bytes` +- `langfuse_fanout_media_divergence_total` + +`langfuse_fanout_media_divergence_total{kind="media_id"}` is the correctness +signal for trace/media token fanout. `kind="upload_url_presence"` records that +some destinations returned an upload URL while others treated the media as +already uploaded. ## Notes -- The collector only handles traces. Feedback scores go directly to Langfuse's - REST API from the LibreChat API process. +- The gateway handles Langfuse media uploads and proxies traces to the internal + collector. Feedback scores go directly to Langfuse's REST API from the + LibreChat API process. - `LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER` must be a full Basic auth header and is - consumed by the collector only. The app does not use it for scores. -- `LANGFUSE_FANOUT_CENTRAL_BASE_URL` is also consumed by the collector only. + consumed by the fanout deployment only. The app does not use it for scores. +- `LANGFUSE_FANOUT_CENTRAL_BASE_URL` is also consumed by the fanout deployment only. For non-EU central feedback scores, set LibreChat's normal `LANGFUSE_BASE_URL` to the same central Langfuse region. - Tenant destinations default to the three configured Langfuse Cloud regions. Add or override `langfuseFanout.tenant.destinations` in Helm for self-hosted or - custom destinations. Compose's included collector config is static; if you add - custom destination keys there, update or regenerate `otelcol.yaml` as well. -- `LANGFUSE_FANOUT_BATCH_TIMEOUT` and `LANGFUSE_FANOUT_BATCH_SEND_SIZE` tune - the OTel batch processors. The defaults (`1s`, `128`) favor low latency and - modest memory use; high-volume deployments may increase them for throughput. -- `LANGFUSE_FANOUT_COLLECTOR_URL` is the local collector URL used by LibreChat, - not a Langfuse Cloud base URL. + custom destinations. +- `LANGFUSE_FANOUT_UPSTREAM_TIMEOUT` tunes the timeout for gateway calls to + Langfuse APIs and presigned media upload URLs. +- `LANGFUSE_FANOUT_PUBLIC_URL` pins the base URL returned for the SDK's + one-time media upload. The gateway fails startup when it is unset or invalid; + this avoids trusting request `Host` or `X-Forwarded-Host` headers. +- `LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS` is a startup guard that must contain + every key in `LANGFUSE_FANOUT_TENANT_DESTINATIONS`; this prevents media + fanout from accepting a destination the collector cannot route traces to. +- `LANGFUSE_FANOUT_REDIS_URI`, optional `LANGFUSE_FANOUT_REDIS_USERNAME`, + optional `LANGFUSE_FANOUT_REDIS_PASSWORD`, and + `LANGFUSE_FANOUT_REDIS_KEY_PREFIX` configure the shared one-time media upload + plan store. The gateway fails startup without a Redis URI. +- `LANGFUSE_FANOUT_OTEL_RECEIVER_ENDPOINT` controls the internal collector + receiver bind address. +- `LANGFUSE_FANOUT_METRICS_SECRET` protects the gateway `/metrics` endpoint. + If unset, the gateway falls back to `METRICS_SECRET` when present. +- `LANGFUSE_FANOUT_MEMORY_LIMIT_MIB`, + `LANGFUSE_FANOUT_MEMORY_SPIKE_LIMIT_MIB`, `LANGFUSE_FANOUT_BATCH_TIMEOUT`, + `LANGFUSE_FANOUT_BATCH_SEND_SIZE`, and + `LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT` tune the internal collector. +- `LANGFUSE_FANOUT_COLLECTOR_URL` is the local gateway URL used by LibreChat. + The env name is kept for compatibility with the original collector shape; it + is not a Langfuse Cloud base URL. diff --git a/otel/langfuse-fanout/cmd/langfuse-fanout/main.go b/otel/langfuse-fanout/cmd/langfuse-fanout/main.go new file mode 100644 index 0000000000..fb38bf605d --- /dev/null +++ b/otel/langfuse-fanout/cmd/langfuse-fanout/main.go @@ -0,0 +1,1177 @@ +package main + +import ( + "bytes" + "compress/gzip" + "context" + "crypto/rand" + "crypto/subtle" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "log" + "net/http" + "net/url" + "os" + "os/signal" + "strconv" + "strings" + "sync" + "syscall" + "time" + + "github.com/prometheus/client_golang/prometheus/promhttp" + "github.com/redis/go-redis/v9" + tracepb "go.opentelemetry.io/proto/otlp/collector/trace/v1" + commonv1 "go.opentelemetry.io/proto/otlp/common/v1" + tracev1 "go.opentelemetry.io/proto/otlp/trace/v1" + "google.golang.org/protobuf/proto" +) + +const ( + defaultListenAddr = ":4318" + defaultTraceCollector = "http://127.0.0.1:4319" + centralName = "central" + tenantPrefix = "/tenant/" + mediaUploadProxyPath = "/__langfuse-fanout/media-upload/" + otelTracePath = "/api/public/otel/v1/traces" + mediaPath = "/api/public/media" + metricsPath = "/metrics" + tenantExportAttribute = "librechat.langfuse.tenant_export.enabled" + tenantDestAttribute = "librechat.langfuse.destination" +) + +type config struct { + listenAddr string + traceCollectorURL string + publicURL string + metricsSecret string + traceDestinationKeys map[string]bool + central destination + tenants map[string]string + redis redisConfig + uploadStore uploadPlanStore + client *http.Client +} + +type redisConfig struct { + uri string + username string + password string + keyPrefix string +} + +type destination struct { + name string + baseURL string + authorization string +} + +type route struct { + destination string + path string +} + +type uploadDestination struct { + Name string `json:"name"` + UploadURL string `json:"uploadUrl"` +} + +type uploadPlan struct { + ExpiresAt time.Time `json:"expiresAt"` + Destinations []uploadDestination `json:"destinations"` + ContentLength int64 `json:"contentLength"` +} + +type uploadPlanStore interface { + Put(ctx context.Context, uploadID string, plan uploadPlan) error + Take(ctx context.Context, uploadID string) (uploadPlan, bool, error) + Ping(ctx context.Context) error + Close() error +} + +type gateway struct { + cfg config + metrics *gatewayMetrics + metricsHTTP http.Handler +} + +type mediaUploadResponse struct { + UploadURL *string `json:"uploadUrl"` + MediaID string `json:"mediaId"` +} + +func main() { + cfg, err := loadConfig() + if err != nil { + log.Fatalf("failed to load config: %v", err) + } + uploadStore, err := newRedisUploadPlanStore(cfg.redis) + if err != nil { + log.Fatalf("failed to initialize Redis upload plan store: %v", err) + } + defer uploadStore.Close() + cfg.uploadStore = uploadStore + + gw := newGateway(cfg) + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + mux := http.NewServeMux() + mux.HandleFunc("/", gw.handle) + + server := &http.Server{ + Addr: cfg.listenAddr, + Handler: mux, + ReadHeaderTimeout: 10 * time.Second, + } + log.Printf("langfuse fanout gateway listening on %s", cfg.listenAddr) + errCh := make(chan error, 1) + go func() { + errCh <- server.ListenAndServe() + }() + + select { + case <-ctx.Done(): + shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if err := server.Shutdown(shutdownCtx); err != nil { + log.Printf("server shutdown failed: %v", err) + } + case err := <-errCh: + if err != nil && !errors.Is(err, http.ErrServerClosed) { + log.Fatal(err) + } + } +} + +func newGateway(cfg config) *gateway { + if cfg.uploadStore == nil { + panic("langfuse fanout gateway requires an upload plan store") + } + metrics := newGatewayMetrics() + return &gateway{ + cfg: cfg, + metrics: metrics, + metricsHTTP: promhttp.HandlerFor(metrics.registry, promhttp.HandlerOpts{}), + } +} + +func loadConfig() (config, error) { + centralBaseURL := normalizeBaseURL(os.Getenv("LANGFUSE_FANOUT_CENTRAL_BASE_URL")) + centralAuth := strings.TrimSpace(os.Getenv("LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER")) + if centralBaseURL == "" { + return config{}, errors.New("LANGFUSE_FANOUT_CENTRAL_BASE_URL is required") + } + if centralAuth == "" { + return config{}, errors.New("LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER is required") + } + + tenants := map[string]string{} + for _, item := range strings.Split(os.Getenv("LANGFUSE_FANOUT_TENANT_DESTINATIONS"), ",") { + item = strings.TrimSpace(item) + if item == "" { + continue + } + key, value, ok := strings.Cut(item, "=") + if !ok { + continue + } + key = normalizeDestinationKey(key) + if key == "" { + continue + } + if baseURL := normalizeBaseURL(value); baseURL != "" { + tenants[key] = baseURL + } + } + traceDestinationKeys := parseDestinationKeys(os.Getenv("LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS")) + if len(tenants) > 0 && len(traceDestinationKeys) == 0 { + return config{}, errors.New("LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS is required when LANGFUSE_FANOUT_TENANT_DESTINATIONS is set") + } + for key := range tenants { + if len(traceDestinationKeys) > 0 && !traceDestinationKeys[key] { + return config{}, fmt.Errorf("tenant destination %q is not present in LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS", key) + } + } + rawPublicURL := strings.TrimSpace(os.Getenv("LANGFUSE_FANOUT_PUBLIC_URL")) + publicURL := normalizeBaseURL(rawPublicURL) + if publicURL == "" { + return config{}, errors.New("LANGFUSE_FANOUT_PUBLIC_URL must be an absolute HTTP(S) URL") + } + redisURI := strings.TrimSpace(os.Getenv("LANGFUSE_FANOUT_REDIS_URI")) + if redisURI == "" { + return config{}, errors.New("LANGFUSE_FANOUT_REDIS_URI is required for media upload plan storage") + } + + return config{ + listenAddr: envOrDefault("LANGFUSE_FANOUT_LISTEN_ADDR", defaultListenAddr), + traceCollectorURL: normalizeCollectorURL(envOrDefault("LANGFUSE_FANOUT_TRACE_COLLECTOR_URL", defaultTraceCollector)), + publicURL: publicURL, + metricsSecret: firstNonEmptyEnv("LANGFUSE_FANOUT_METRICS_SECRET", "METRICS_SECRET"), + traceDestinationKeys: traceDestinationKeys, + central: destination{ + name: centralName, + baseURL: centralBaseURL, + authorization: centralAuth, + }, + tenants: tenants, + redis: redisConfig{ + uri: redisURI, + username: strings.TrimSpace(os.Getenv("LANGFUSE_FANOUT_REDIS_USERNAME")), + password: strings.TrimSpace(os.Getenv("LANGFUSE_FANOUT_REDIS_PASSWORD")), + keyPrefix: envOrDefault("LANGFUSE_FANOUT_REDIS_KEY_PREFIX", "langfuse-fanout"), + }, + client: &http.Client{ + Timeout: parseDurationEnv("LANGFUSE_FANOUT_UPSTREAM_TIMEOUT", 30*time.Second), + }, + }, nil +} + +func (g *gateway) handle(w http.ResponseWriter, r *http.Request) { + startedAt := time.Now() + recorder := &statusRecorder{ResponseWriter: w, status: http.StatusOK} + w = recorder + defer func() { + if r.URL.Path != metricsPath && g.metrics != nil { + g.metrics.recordHTTP(r.Method, normalizeMetricPath(r.URL.Path), recorder.status, time.Since(startedAt)) + } + }() + + route := parseRoute(r.URL.Path) + switch { + case route.path == otelTracePath && r.Method == http.MethodPost: + g.handleTraces(w, r, route) + case route.path == mediaPath && r.Method == http.MethodPost: + g.handleMediaCreate(w, r, route) + case strings.HasPrefix(route.path, mediaPath+"/") && r.Method == http.MethodGet: + g.handleMediaGet(w, r, route) + case strings.HasPrefix(route.path, mediaPath+"/") && r.Method == http.MethodPatch: + g.handleMediaPatch(w, r, route) + case strings.HasPrefix(r.URL.Path, mediaUploadProxyPath) && r.Method == http.MethodPut: + g.handleMediaUpload(w, r) + case r.URL.Path == "/healthz": + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok")) + case r.URL.Path == metricsPath && r.Method == http.MethodGet: + g.handleMetrics(w, r) + default: + http.Error(w, "langfuse fanout gateway only supports OTLP traces and media upload APIs", http.StatusNotImplemented) + } +} + +func (g *gateway) handleTraces(w http.ResponseWriter, r *http.Request, route route) { + body, err := readMaybeGzip(r) + if err != nil { + http.Error(w, "failed to read request body", http.StatusBadRequest) + return + } + + contentType := r.Header.Get("Content-Type") + if route.destination != "" && + g.cfg.tenants[route.destination] != "" && + strings.TrimSpace(r.Header.Get("Authorization")) != "" { + body, err = addTenantRouteAttributes(body, contentType, route.destination) + if err != nil { + http.Error(w, "failed to add OTLP tenant routing attributes", http.StatusBadRequest) + return + } + } + + contentEncoding := "" + if strings.EqualFold(r.Header.Get("Content-Encoding"), "gzip") { + contentEncoding = "gzip" + body, err = gzipBytes(body) + if err != nil { + http.Error(w, "failed to encode request body", http.StatusInternalServerError) + return + } + } + + resp, err := g.forwardTraceToCollector(r.Context(), r.Header, body, contentType, contentEncoding) + if err != nil { + g.recordTraceExport(route, "error") + http.Error(w, fmt.Sprintf("trace collector export failed: %v", err), http.StatusBadGateway) + return + } + defer resp.Body.Close() + + g.recordTraceExport(route, "success") + copyResponseHeaders(w.Header(), resp.Header) + w.WriteHeader(resp.StatusCode) + _, _ = io.Copy(w, resp.Body) +} + +func (g *gateway) handleMediaCreate(w http.ResponseWriter, r *http.Request, route route) { + body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 2<<20)) + if err != nil { + http.Error(w, "failed to read media create request", http.StatusBadRequest) + return + } + if err := g.cfg.uploadStore.Ping(r.Context()); err != nil { + g.recordUploadPlanStoreError("ping") + http.Error(w, fmt.Sprintf("media upload plan store unavailable: %v", err), http.StatusBadGateway) + return + } + + destinations := g.mediaDestinations(route, r.Header.Get("Authorization")) + if len(destinations) == 0 { + http.Error(w, "no media destinations configured", http.StatusBadGateway) + return + } + + type mediaCreateResult struct { + destination destination + response mediaUploadResponse + err error + } + responses := make([]mediaCreateResult, len(destinations)) + var wg sync.WaitGroup + for index, dest := range destinations { + index, dest := index, dest + wg.Add(1) + go func() { + defer wg.Done() + response, err := g.postMediaCreate(r.Context(), dest, body, r.Header.Get("Content-Type")) + responses[index] = mediaCreateResult{destination: dest, response: response, err: err} + }() + } + wg.Wait() + for _, result := range responses { + if result.err != nil { + http.Error(w, fmt.Sprintf("%s media create failed: %v", result.destination.name, result.err), http.StatusBadGateway) + return + } + } + + mediaID := responses[0].response.MediaID + if mediaID == "" { + http.Error(w, "upstream media create returned empty mediaId", http.StatusBadGateway) + return + } + // Langfuse derives mediaId from the content hash today, so all fanout + // destinations should converge on the same id for the same POST body. + for _, response := range responses[1:] { + if response.response.MediaID != mediaID { + log.Printf( + "upstream media IDs differ: %s=%s %s=%s", + responses[0].destination.name, + mediaID, + response.destination.name, + response.response.MediaID, + ) + g.recordMediaDivergence("media_id", response.destination.name) + http.Error(w, "upstream media IDs differ across destinations", http.StatusBadGateway) + return + } + } + + uploadPlan := uploadPlan{ + ExpiresAt: time.Now().Add(time.Hour), + Destinations: []uploadDestination{}, + } + var requestBody struct { + ContentLength int64 `json:"contentLength"` + } + _ = json.Unmarshal(body, &requestBody) + uploadPlan.ContentLength = requestBody.ContentLength + + for _, response := range responses { + if response.response.UploadURL == nil || *response.response.UploadURL == "" { + if len(uploadPlan.Destinations) > 0 { + g.recordMediaDivergence("upload_url_presence", response.destination.name) + } + continue + } + uploadPlan.Destinations = append(uploadPlan.Destinations, uploadDestination{ + Name: response.destination.name, + UploadURL: *response.response.UploadURL, + }) + } + + result := mediaUploadResponse{MediaID: mediaID} + if len(uploadPlan.Destinations) > 0 { + uploadID, err := randomID() + if err != nil { + http.Error(w, "failed to create media upload id", http.StatusInternalServerError) + return + } + if err := g.storeUpload(r.Context(), uploadID, uploadPlan); err != nil { + g.recordUploadPlanStoreError("put") + http.Error(w, fmt.Sprintf("failed to store media upload plan: %v", err), http.StatusBadGateway) + return + } + g.recordUploadPlanCreated(uploadPlan) + uploadURL := g.absoluteURL(mediaUploadProxyPath + uploadID) + result.UploadURL = &uploadURL + } + + writeJSON(w, http.StatusCreated, result) +} + +func (g *gateway) handleMediaPatch(w http.ResponseWriter, r *http.Request, route route) { + body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 1<<20)) + if err != nil { + http.Error(w, "failed to read media patch request", http.StatusBadRequest) + return + } + + destinations := g.mediaDestinations(route, r.Header.Get("Authorization")) + if len(destinations) == 0 { + http.Error(w, "no media destinations configured", http.StatusBadGateway) + return + } + + type patchResult struct { + destination string + err error + } + results := make([]patchResult, len(destinations)) + var wg sync.WaitGroup + for index, dest := range destinations { + index, dest := index, dest + wg.Add(1) + go func() { + defer wg.Done() + results[index] = patchResult{ + destination: dest.name, + err: g.patchMedia(r.Context(), dest, route.path, body, r.Header.Get("Content-Type")), + } + }() + } + wg.Wait() + for _, result := range results { + if result.err != nil { + http.Error(w, fmt.Sprintf("%s media patch failed: %v", result.destination, result.err), http.StatusBadGateway) + return + } + } + w.WriteHeader(http.StatusNoContent) +} + +func (g *gateway) handleMediaGet(w http.ResponseWriter, r *http.Request, route route) { + destinations := g.mediaDestinations(route, r.Header.Get("Authorization")) + if len(destinations) == 0 { + http.Error(w, "no media destinations configured", http.StatusBadGateway) + return + } + target := destinations[0] + if route.destination != "" { + target = destinations[len(destinations)-1] + } + resp := g.getMedia(r.Context(), target, route.path, r.URL.RawQuery) + if resp == nil { + http.Error(w, "media get failed", http.StatusBadGateway) + return + } + defer resp.Body.Close() + copyResponseHeaders(w.Header(), resp.Header) + w.WriteHeader(resp.StatusCode) + _, _ = io.Copy(w, resp.Body) +} + +func (g *gateway) getMedia(ctx context.Context, target destination, path string, rawQuery string) *http.Response { + upstreamURL := target.baseURL + path + if rawQuery != "" { + upstreamURL += "?" + rawQuery + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, upstreamURL, nil) + if err != nil { + return nil + } + req.Header.Set("Authorization", target.authorization) + resp, err := g.doUpstream(req, "media_get", target.name) + if err != nil { + return nil + } + return resp +} + +func (g *gateway) handleMetrics(w http.ResponseWriter, r *http.Request) { + if g.cfg.metricsSecret == "" { + w.WriteHeader(http.StatusUnauthorized) + return + } + const prefix = "Bearer " + auth := r.Header.Get("Authorization") + if len(auth) < len(prefix) || !strings.EqualFold(auth[:len(prefix)], prefix) { + w.WriteHeader(http.StatusUnauthorized) + return + } + token := strings.TrimSpace(auth[len(prefix):]) + if subtle.ConstantTimeCompare([]byte(token), []byte(g.cfg.metricsSecret)) != 1 { + w.WriteHeader(http.StatusUnauthorized) + return + } + g.metricsHTTP.ServeHTTP(w, r) +} + +func (g *gateway) handleMediaUpload(w http.ResponseWriter, r *http.Request) { + uploadID := strings.TrimPrefix(r.URL.Path, mediaUploadProxyPath) + + plan, ok, err := g.takeUpload(r.Context(), uploadID) + if err != nil { + g.recordUploadPlanStoreError("take") + http.Error(w, "failed to load media upload plan", http.StatusBadGateway) + return + } + if !ok { + g.recordUploadPlanMiss() + http.Error(w, "unknown or expired upload", http.StatusNotFound) + return + } + + body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxUploadBytes(plan.ContentLength))) + if err != nil { + if err := g.restoreUpload(r.Context(), uploadID, plan); err != nil { + g.recordUploadPlanStoreError("restore") + } + http.Error(w, "failed to read upload body", http.StatusBadRequest) + return + } + + type uploadResult struct { + destination string + status int + err error + } + results := make([]uploadResult, len(plan.Destinations)) + var wg sync.WaitGroup + for index, dest := range plan.Destinations { + index, dest := index, dest + wg.Add(1) + go func() { + defer wg.Done() + code, err := g.putMedia(r.Context(), dest, body, r.Header) + results[index] = uploadResult{destination: dest.Name, status: code, err: err} + }() + } + wg.Wait() + status := http.StatusOK + for _, result := range results { + if result.err != nil { + if err := g.restoreUpload(r.Context(), uploadID, plan); err != nil { + g.recordUploadPlanStoreError("restore") + } + http.Error(w, fmt.Sprintf("%s upload failed: %v", result.destination, result.err), http.StatusBadGateway) + return + } + if result.status > status { + status = result.status + } + } + g.recordUploadPlanCompleted(plan) + w.WriteHeader(status) +} + +func (g *gateway) mediaDestinations(route route, tenantAuth string) []destination { + destinations := []destination{g.cfg.central} + if route.destination == "" { + return destinations + } + baseURL := g.cfg.tenants[route.destination] + if baseURL == "" || strings.TrimSpace(tenantAuth) == "" { + return destinations + } + return append(destinations, destination{ + name: "tenant_" + route.destination, + baseURL: baseURL, + authorization: strings.TrimSpace(tenantAuth), + }) +} + +func (g *gateway) forwardTraceToCollector(ctx context.Context, headers http.Header, body []byte, contentType string, contentEncoding string) (*http.Response, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, g.cfg.traceCollectorURL+otelTracePath, bytes.NewReader(body)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", contentTypeOrDefault(contentType, "application/x-protobuf")) + if value := strings.TrimSpace(headers.Get("Authorization")); value != "" { + req.Header.Set("Authorization", value) + } + if contentEncoding != "" { + req.Header.Set("Content-Encoding", contentEncoding) + } + resp, err := g.doUpstream(req, "trace_collector", "collector") + if err != nil { + return nil, err + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + defer resp.Body.Close() + text, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return nil, fmt.Errorf("status %d: %s", resp.StatusCode, strings.TrimSpace(string(text))) + } + return resp, nil +} + +func (g *gateway) postMediaCreate(ctx context.Context, dest destination, body []byte, contentType string) (mediaUploadResponse, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, dest.baseURL+mediaPath, bytes.NewReader(body)) + if err != nil { + return mediaUploadResponse{}, err + } + req.Header.Set("Authorization", dest.authorization) + req.Header.Set("Content-Type", contentTypeOrDefault(contentType, "application/json")) + resp, err := g.doUpstream(req, "media_create", dest.name) + if err != nil { + return mediaUploadResponse{}, err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + text, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return mediaUploadResponse{}, fmt.Errorf("status %d: %s", resp.StatusCode, strings.TrimSpace(string(text))) + } + var result mediaUploadResponse + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return mediaUploadResponse{}, err + } + return result, nil +} + +func (g *gateway) patchMedia(ctx context.Context, dest destination, path string, body []byte, contentType string) error { + req, err := http.NewRequestWithContext(ctx, http.MethodPatch, dest.baseURL+path, bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Authorization", dest.authorization) + req.Header.Set("Content-Type", contentTypeOrDefault(contentType, "application/json")) + return g.doExpect2xx("media_patch", dest.name, req) +} + +func (g *gateway) putMedia(ctx context.Context, dest uploadDestination, body []byte, originalHeaders http.Header) (int, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodPut, dest.UploadURL, bytes.NewReader(body)) + if err != nil { + return 0, err + } + if value := originalHeaders.Get("Content-Type"); value != "" { + if !allowedUploadContentType(value) { + return 0, fmt.Errorf("unsupported upload content type %q", value) + } + req.Header.Set("Content-Type", value) + } + if value := originalHeaders.Get("Content-Encoding"); value != "" { + req.Header.Set("Content-Encoding", value) + } + if isAzureUploadURL(dest.UploadURL) { + if value := originalHeaders.Get("x-ms-blob-type"); value != "" { + req.Header.Set("x-ms-blob-type", value) + } + } else if !isGCSUploadURL(dest.UploadURL) { + if value := originalHeaders.Get("x-amz-checksum-sha256"); value != "" { + req.Header.Set("x-amz-checksum-sha256", value) + } + } + resp, err := g.doUpstream(req, "media_upload", dest.Name) + if err != nil { + return 0, err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + text, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return resp.StatusCode, fmt.Errorf("status %d: %s", resp.StatusCode, strings.TrimSpace(string(text))) + } + return resp.StatusCode, nil +} + +func (g *gateway) doExpect2xx(operation string, destination string, req *http.Request) error { + resp, err := g.doUpstream(req, operation, destination) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + text, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return fmt.Errorf("status %d: %s", resp.StatusCode, strings.TrimSpace(string(text))) + } + return nil +} + +func (g *gateway) doUpstream(req *http.Request, operation string, destination string) (*http.Response, error) { + startedAt := time.Now() + resp, err := g.cfg.client.Do(req) + if err != nil { + if g.metrics != nil { + g.metrics.recordUpstream(operation, destination, "error", time.Since(startedAt)) + } + return nil, err + } + if g.metrics != nil { + g.metrics.recordUpstream(operation, destination, statusClass(resp.StatusCode), time.Since(startedAt)) + } + return resp, nil +} + +func (g *gateway) recordTraceExport(route route, result string) { + if g.metrics == nil { + return + } + destination := centralName + if route.destination != "" { + destination = "tenant_" + route.destination + } + g.metrics.recordTraceExport(destination, result) +} + +func (g *gateway) recordMediaDivergence(kind string, destination string) { + if g.metrics != nil { + g.metrics.recordMediaDivergence(kind, destination) + } +} + +func (g *gateway) recordUploadPlanCreated(plan uploadPlan) { + if g.metrics != nil { + g.metrics.recordUploadPlanCreated(plan.ContentLength) + } +} + +func (g *gateway) recordUploadPlanCompleted(plan uploadPlan) { + if len(plan.Destinations) == 0 { + return + } + if g.metrics != nil { + g.metrics.recordUploadPlanCompleted() + } +} + +func (g *gateway) recordUploadPlanMiss() { + if g.metrics != nil { + g.metrics.recordUploadPlanMiss() + } +} + +func (g *gateway) recordUploadPlanStoreError(operation string) { + if g.metrics != nil { + g.metrics.recordUploadPlanStoreError(operation) + } +} + +func addTenantRouteAttributes(body []byte, contentType string, destination string) ([]byte, error) { + if isJSONContentType(contentType) { + return addJSONTenantRouteAttributes(body, destination) + } + return addProtobufTenantRouteAttributes(body, destination) +} + +func addProtobufTenantRouteAttributes(body []byte, destination string) ([]byte, error) { + var request tracepb.ExportTraceServiceRequest + if err := proto.Unmarshal(body, &request); err != nil { + return nil, err + } + + for _, resourceSpan := range request.ResourceSpans { + for _, scopeSpan := range resourceSpan.ScopeSpans { + for _, span := range scopeSpan.Spans { + upsertSpanStringAttribute(span, tenantDestAttribute, destination) + upsertSpanStringAttribute(span, tenantExportAttribute, "true") + } + } + } + + return proto.Marshal(&request) +} + +func upsertSpanStringAttribute(span *tracev1.Span, key string, value string) { + for _, attribute := range span.Attributes { + if attribute.Key == key { + attribute.Value = stringAnyValue(value) + return + } + } + span.Attributes = append(span.Attributes, &commonv1.KeyValue{ + Key: key, + Value: stringAnyValue(value), + }) +} + +func stringAnyValue(value string) *commonv1.AnyValue { + return &commonv1.AnyValue{ + Value: &commonv1.AnyValue_StringValue{StringValue: value}, + } +} + +func addJSONTenantRouteAttributes(body []byte, destination string) ([]byte, error) { + var request map[string]any + if err := json.Unmarshal(body, &request); err != nil { + return nil, err + } + resourceSpans, _ := request["resourceSpans"].([]any) + for _, resourceSpan := range resourceSpans { + resourceSpanMap, _ := resourceSpan.(map[string]any) + scopeSpans, _ := resourceSpanMap["scopeSpans"].([]any) + for _, scopeSpan := range scopeSpans { + scopeSpanMap, _ := scopeSpan.(map[string]any) + spans, _ := scopeSpanMap["spans"].([]any) + for _, span := range spans { + spanMap, _ := span.(map[string]any) + upsertJSONSpanStringAttribute(spanMap, tenantDestAttribute, destination) + upsertJSONSpanStringAttribute(spanMap, tenantExportAttribute, "true") + } + } + } + return json.Marshal(request) +} + +func upsertJSONSpanStringAttribute(span map[string]any, key string, value string) { + attrs, _ := span["attributes"].([]any) + for _, attr := range attrs { + attrMap, _ := attr.(map[string]any) + if attrMap["key"] == key { + attrMap["value"] = map[string]any{"stringValue": value} + return + } + } + span["attributes"] = append(attrs, map[string]any{ + "key": key, + "value": map[string]any{"stringValue": value}, + }) +} + +func stringValue(value *commonv1.AnyValue) string { + if value == nil { + return "" + } + if stringValue := value.GetStringValue(); stringValue != "" { + return stringValue + } + if value.GetBoolValue() { + return "true" + } + return "" +} + +func (g *gateway) storeUpload(ctx context.Context, uploadID string, plan uploadPlan) error { + return g.cfg.uploadStore.Put(ctx, uploadID, plan) +} + +func (g *gateway) takeUpload(ctx context.Context, uploadID string) (uploadPlan, bool, error) { + if !validUploadID(uploadID) { + return uploadPlan{}, false, nil + } + plan, ok, err := g.cfg.uploadStore.Take(ctx, uploadID) + if err != nil || !ok { + return uploadPlan{}, ok, err + } + return plan, true, nil +} + +func (g *gateway) restoreUpload(ctx context.Context, uploadID string, plan uploadPlan) error { + if time.Now().After(plan.ExpiresAt) { + return nil + } + return g.cfg.uploadStore.Put(ctx, uploadID, plan) +} + +type redisUploadPlanStore struct { + client *redis.Client + prefix string +} + +func newRedisUploadPlanStore(cfg redisConfig) (*redisUploadPlanStore, error) { + options, err := redis.ParseURL(cfg.uri) + if err != nil { + return nil, fmt.Errorf("parse LANGFUSE_FANOUT_REDIS_URI: %w", err) + } + if cfg.username != "" { + options.Username = cfg.username + } + if cfg.password != "" { + options.Password = cfg.password + } + client := redis.NewClient(options) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := client.Ping(ctx).Err(); err != nil { + _ = client.Close() + return nil, fmt.Errorf("ping Redis: %w", err) + } + return &redisUploadPlanStore{ + client: client, + prefix: normalizeRedisKeyPrefix(cfg.keyPrefix), + }, nil +} + +func (s *redisUploadPlanStore) Put(ctx context.Context, uploadID string, plan uploadPlan) error { + if !validUploadID(uploadID) { + return errors.New("invalid upload id") + } + ttl := time.Until(plan.ExpiresAt) + if ttl <= 0 { + return errors.New("upload plan already expired") + } + body, err := json.Marshal(plan) + if err != nil { + return err + } + return s.client.Set(ctx, s.key(uploadID), body, ttl).Err() +} + +func (s *redisUploadPlanStore) Take(ctx context.Context, uploadID string) (uploadPlan, bool, error) { + if !validUploadID(uploadID) { + return uploadPlan{}, false, nil + } + value, err := redisTakeScript.Run(ctx, s.client, []string{s.key(uploadID)}).Text() + if errors.Is(err, redis.Nil) { + return uploadPlan{}, false, nil + } + if err != nil { + return uploadPlan{}, false, err + } + var plan uploadPlan + if err := json.Unmarshal([]byte(value), &plan); err != nil { + return uploadPlan{}, false, err + } + return plan, true, nil +} + +func (s *redisUploadPlanStore) Ping(ctx context.Context) error { + return s.client.Ping(ctx).Err() +} + +func (s *redisUploadPlanStore) Close() error { + return s.client.Close() +} + +func (s *redisUploadPlanStore) key(uploadID string) string { + return s.prefix + ":media-upload:" + uploadID +} + +var redisTakeScript = redis.NewScript(` +local value = redis.call("GET", KEYS[1]) +if value then + redis.call("DEL", KEYS[1]) +end +return value +`) + +func normalizeRedisKeyPrefix(value string) string { + value = strings.TrimSpace(value) + if value == "" { + return "langfuse-fanout" + } + return strings.TrimRight(value, ":") +} + +func validUploadID(value string) bool { + if value == "" || len(value) > 128 { + return false + } + for _, r := range value { + switch { + case r >= 'a' && r <= 'f': + case r >= '0' && r <= '9': + default: + return false + } + } + return true +} + +func parseRoute(path string) route { + if !strings.HasPrefix(path, tenantPrefix) { + return route{path: path} + } + rest := strings.TrimPrefix(path, tenantPrefix) + destination, suffix, ok := strings.Cut(rest, "/") + if !ok { + return route{path: path} + } + normalizedDestination := normalizeDestinationKey(destination) + if normalizedDestination == "" { + return route{path: path} + } + return route{destination: normalizedDestination, path: "/" + suffix} +} + +func readMaybeGzip(r *http.Request) ([]byte, error) { + if !strings.EqualFold(r.Header.Get("Content-Encoding"), "gzip") { + return io.ReadAll(io.LimitReader(r.Body, 20<<20)) + } + reader, err := gzip.NewReader(r.Body) + if err != nil { + return nil, err + } + defer reader.Close() + return io.ReadAll(io.LimitReader(reader, 20<<20)) +} + +func gzipBytes(body []byte) ([]byte, error) { + var buffer bytes.Buffer + writer := gzip.NewWriter(&buffer) + if _, err := writer.Write(body); err != nil { + return nil, err + } + if err := writer.Close(); err != nil { + return nil, err + } + return buffer.Bytes(), nil +} + +func (g *gateway) absoluteURL(path string) string { + return g.cfg.publicURL + path +} + +func writeJSON(w http.ResponseWriter, status int, value any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(value) +} + +func copyResponseHeaders(target http.Header, source http.Header) { + for key, values := range source { + if strings.EqualFold(key, "Content-Length") { + continue + } + for _, value := range values { + target.Add(key, value) + } + } +} + +func normalizeBaseURL(value string) string { + value = strings.TrimSpace(value) + if value == "" { + return "" + } + parsed, err := url.Parse(value) + if err != nil || parsed.Host == "" { + return "" + } + if parsed.Scheme != "http" && parsed.Scheme != "https" { + return "" + } + parsed.Path = strings.TrimRight(parsed.Path, "/") + parsed.RawQuery = "" + parsed.Fragment = "" + return strings.TrimRight(parsed.String(), "/") +} + +func normalizeCollectorURL(value string) string { + value = strings.TrimSpace(value) + if value == "" { + return defaultTraceCollector + } + return strings.TrimRight(value, "/") +} + +func normalizeDestinationKey(value string) string { + value = strings.TrimSpace(strings.ToLower(value)) + if value == "" { + return "" + } + var builder strings.Builder + for _, r := range value { + switch { + case r >= 'a' && r <= 'z': + builder.WriteRune(r) + case r >= '0' && r <= '9': + builder.WriteRune(r) + case r == '_' || r == '-': + builder.WriteRune(r) + default: + builder.WriteRune('_') + } + } + normalized := builder.String() + if normalized == "" || normalized[0] < 'a' || normalized[0] > 'z' { + return "" + } + return normalized +} + +func parseDestinationKeys(value string) map[string]bool { + result := map[string]bool{} + for _, item := range strings.Split(value, ",") { + key := normalizeDestinationKey(item) + if key != "" { + result[key] = true + } + } + return result +} + +func contentTypeOrDefault(value string, fallback string) string { + if strings.TrimSpace(value) == "" { + return fallback + } + return value +} + +func isJSONContentType(value string) bool { + mediaType := strings.ToLower(strings.TrimSpace(strings.Split(value, ";")[0])) + return mediaType == "application/json" || strings.HasSuffix(mediaType, "+json") +} + +func isGCSUploadURL(value string) bool { + parsed, err := url.Parse(value) + if err != nil { + return false + } + host := parsed.Hostname() + return host == "storage.googleapis.com" || strings.HasSuffix(host, ".storage.googleapis.com") +} + +func isAzureUploadURL(value string) bool { + parsed, err := url.Parse(value) + if err != nil { + return false + } + host := parsed.Hostname() + return strings.Contains(host, ".blob.core.") || strings.Contains(host, ".blob.storage.") +} + +func allowedUploadContentType(value string) bool { + mediaType := strings.ToLower(strings.TrimSpace(strings.Split(value, ";")[0])) + return mediaType == "application/octet-stream" || + mediaType == "application/pdf" || + strings.HasPrefix(mediaType, "image/") || + strings.HasPrefix(mediaType, "audio/") || + strings.HasPrefix(mediaType, "video/") +} + +func envOrDefault(key string, fallback string) string { + if value := strings.TrimSpace(os.Getenv(key)); value != "" { + return value + } + return fallback +} + +func firstNonEmptyEnv(keys ...string) string { + for _, key := range keys { + if value := strings.TrimSpace(os.Getenv(key)); value != "" { + return value + } + } + return "" +} + +func parseDurationEnv(key string, fallback time.Duration) time.Duration { + value := strings.TrimSpace(os.Getenv(key)) + if value == "" { + return fallback + } + duration, err := time.ParseDuration(value) + if err == nil { + return duration + } + seconds, err := strconv.Atoi(value) + if err != nil { + return fallback + } + return time.Duration(seconds) * time.Second +} + +func randomID() (string, error) { + var bytes [16]byte + if _, err := rand.Read(bytes[:]); err != nil { + return "", err + } + return hex.EncodeToString(bytes[:]), nil +} + +func maxUploadBytes(contentLength int64) int64 { + if contentLength <= 0 { + return 256 << 20 + } + return contentLength + (1 << 20) +} diff --git a/otel/langfuse-fanout/cmd/langfuse-fanout/main_test.go b/otel/langfuse-fanout/cmd/langfuse-fanout/main_test.go new file mode 100644 index 0000000000..6ad7ac4917 --- /dev/null +++ b/otel/langfuse-fanout/cmd/langfuse-fanout/main_test.go @@ -0,0 +1,888 @@ +package main + +import ( + "bytes" + "compress/gzip" + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "sync" + "testing" + "time" + + tracepb "go.opentelemetry.io/proto/otlp/collector/trace/v1" + commonv1 "go.opentelemetry.io/proto/otlp/common/v1" + resourcev1 "go.opentelemetry.io/proto/otlp/resource/v1" + tracev1 "go.opentelemetry.io/proto/otlp/trace/v1" + "google.golang.org/protobuf/proto" +) + +func TestLoadConfigRejectsDestinationsMissingTraceRoutes(t *testing.T) { + t.Setenv("LANGFUSE_FANOUT_CENTRAL_BASE_URL", "https://cloud.langfuse.com") + t.Setenv("LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER", "Basic central") + t.Setenv("LANGFUSE_FANOUT_PUBLIC_URL", "http://fanout.local:4318") + t.Setenv("LANGFUSE_FANOUT_TENANT_DESTINATIONS", "eu=https://cloud.langfuse.com,ca=https://example.com") + t.Setenv("LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS", "eu,us,jp") + + _, err := loadConfig() + if err == nil || !strings.Contains(err.Error(), `tenant destination "ca"`) { + t.Fatalf("expected missing trace route error, got %v", err) + } +} + +func TestLoadConfigRejectsTenantDestinationsWithoutTraceKeys(t *testing.T) { + t.Setenv("LANGFUSE_FANOUT_CENTRAL_BASE_URL", "https://cloud.langfuse.com") + t.Setenv("LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER", "Basic central") + t.Setenv("LANGFUSE_FANOUT_PUBLIC_URL", "http://fanout.local:4318") + t.Setenv("LANGFUSE_FANOUT_TENANT_DESTINATIONS", "eu=https://cloud.langfuse.com") + t.Setenv("LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS", "") + + _, err := loadConfig() + if err == nil || !strings.Contains(err.Error(), "LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS is required") { + t.Fatalf("expected missing trace keys error, got %v", err) + } +} + +func TestLoadConfigRejectsInvalidPublicURL(t *testing.T) { + t.Setenv("LANGFUSE_FANOUT_CENTRAL_BASE_URL", "https://cloud.langfuse.com") + t.Setenv("LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER", "Basic central") + t.Setenv("LANGFUSE_FANOUT_PUBLIC_URL", "ftp://example.com") + + _, err := loadConfig() + if err == nil || !strings.Contains(err.Error(), "LANGFUSE_FANOUT_PUBLIC_URL") { + t.Fatalf("expected invalid public URL error, got %v", err) + } +} + +func TestLoadConfigRequiresRedisURI(t *testing.T) { + t.Setenv("LANGFUSE_FANOUT_CENTRAL_BASE_URL", "https://cloud.langfuse.com") + t.Setenv("LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER", "Basic central") + t.Setenv("LANGFUSE_FANOUT_PUBLIC_URL", "http://fanout.local:4318") + + _, err := loadConfig() + if err == nil || !strings.Contains(err.Error(), "LANGFUSE_FANOUT_REDIS_URI") { + t.Fatalf("expected missing Redis URI error, got %v", err) + } +} + +func TestNormalizeBaseURLAllowsOnlyHTTPAndHTTPS(t *testing.T) { + if got := normalizeBaseURL("http://localhost:3000/path/"); got != "http://localhost:3000/path" { + t.Fatalf("http URL normalized to %q", got) + } + if got := normalizeBaseURL("https://cloud.langfuse.com/"); got != "https://cloud.langfuse.com" { + t.Fatalf("https URL normalized to %q", got) + } + if got := normalizeBaseURL("file:///tmp/langfuse"); got != "" { + t.Fatalf("file URL should be rejected, got %q", got) + } +} + +func TestTraceProxyForwardsExistingRoutingAttributesToCollector(t *testing.T) { + t.Parallel() + + var collectorTrace []byte + collector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != otelTracePath { + t.Fatalf("unexpected collector path %s", r.URL.Path) + } + if got := r.Header.Get("Authorization"); got != "Basic tenant" { + t.Fatalf("collector auth = %q", got) + } + collectorTrace, _ = io.ReadAll(r.Body) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte("{}")) + })) + defer collector.Close() + + gw := newTestGatewayWithCollector(collector.URL) + body := buildTraceRequest(t, map[string]string{ + tenantExportAttribute: "true", + tenantDestAttribute: "eu", + "kept": "value", + }) + + req := httptest.NewRequest(http.MethodPost, otelTracePath, bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/x-protobuf") + req.Header.Set("Authorization", "Basic tenant") + resp := httptest.NewRecorder() + + gw.handle(resp, req) + if resp.Code != http.StatusOK { + t.Fatalf("status = %d, body = %s", resp.Code, resp.Body.String()) + } + if resp.Body.String() != "{}" { + t.Fatalf("expected collector response body, got %s", resp.Body.String()) + } + if len(collectorTrace) == 0 { + t.Fatal("expected collector export") + } + + attrs := parseTraceAttributes(t, collectorTrace) + if attrs[tenantExportAttribute] != "true" || attrs[tenantDestAttribute] != "eu" { + t.Fatalf("collector trace missing routing attrs: %#v", attrs) + } + if attrs["kept"] != "value" { + t.Fatalf("collector trace lost kept attr: %#v", attrs) + } +} + +func TestGzipTraceProxyAddsRoutingAttributesFromPath(t *testing.T) { + t.Parallel() + + var collectorTrace []byte + var collectorEncoding string + collector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + collectorEncoding = r.Header.Get("Content-Encoding") + reader, err := gzip.NewReader(r.Body) + if err != nil { + t.Fatalf("collector gzip reader: %v", err) + } + defer reader.Close() + collectorTrace, _ = io.ReadAll(reader) + _, _ = w.Write([]byte(`{"partialSuccess":{}}`)) + })) + defer collector.Close() + + var zipped bytes.Buffer + zipper := gzip.NewWriter(&zipped) + if _, err := zipper.Write(buildTraceRequest(t, map[string]string{"kept": "value"})); err != nil { + t.Fatal(err) + } + if err := zipper.Close(); err != nil { + t.Fatal(err) + } + + gw := newTestGatewayWithCollector(collector.URL) + req := httptest.NewRequest(http.MethodPost, tenantPrefix+"eu"+otelTracePath, bytes.NewReader(zipped.Bytes())) + req.Header.Set("Content-Type", "application/x-protobuf") + req.Header.Set("Content-Encoding", "gzip") + req.Header.Set("Authorization", "Basic tenant") + resp := httptest.NewRecorder() + + gw.handle(resp, req) + if resp.Code != http.StatusOK { + t.Fatalf("status = %d, body = %s", resp.Code, resp.Body.String()) + } + if collectorEncoding != "gzip" { + t.Fatalf("collector encoding = %q", collectorEncoding) + } + if resp.Body.String() != `{"partialSuccess":{}}` { + t.Fatalf("expected collector response body, got %s", resp.Body.String()) + } + attrs := parseTraceAttributes(t, collectorTrace) + if attrs[tenantExportAttribute] != "true" || attrs[tenantDestAttribute] != "eu" || attrs["kept"] != "value" { + t.Fatalf("collector trace attrs = %#v", attrs) + } +} + +func TestJSONTraceProxyAddsRoutingAttributesFromPath(t *testing.T) { + t.Parallel() + + var collectorTrace []byte + var collectorContentType string + collector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != otelTracePath { + t.Fatalf("unexpected collector path %s", r.URL.Path) + } + if got := r.Header.Get("Authorization"); got != "Basic tenant" { + t.Fatalf("collector auth = %q", got) + } + collectorContentType = r.Header.Get("Content-Type") + collectorTrace, _ = io.ReadAll(r.Body) + _, _ = w.Write([]byte("{}")) + })) + defer collector.Close() + + gw := newTestGatewayWithCollector(collector.URL) + body := buildJSONTraceRequest(t, map[string]any{ + "kept": "value", + }) + + req := httptest.NewRequest(http.MethodPost, tenantPrefix+"eu"+otelTracePath, bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Basic tenant") + resp := httptest.NewRecorder() + + gw.handle(resp, req) + if resp.Code != http.StatusOK { + t.Fatalf("status = %d, body = %s", resp.Code, resp.Body.String()) + } + if len(collectorTrace) == 0 { + t.Fatal("expected collector export") + } + if collectorContentType != "application/json" { + t.Fatalf("content type = %q", collectorContentType) + } + attrs := parseJSONTraceAttributes(t, collectorTrace) + if attrs[tenantExportAttribute] != "true" || attrs[tenantDestAttribute] != "eu" { + t.Fatalf("collector trace missing routing attrs: %#v", attrs) + } + if attrs["kept"] != "value" { + t.Fatalf("collector trace lost kept attr: %#v", attrs) + } +} + +func TestMediaUploadFansOutToCentralAndTenant(t *testing.T) { + t.Parallel() + + var mu sync.Mutex + uploads := map[string]string{} + upstream := func(name string) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodPost && r.URL.Path == mediaPath: + uploadURL := "http://" + r.Host + "/upload/" + name + writeJSON(w, http.StatusCreated, mediaUploadResponse{ + MediaID: "same-media-id", + UploadURL: &uploadURL, + }) + case r.Method == http.MethodPut && r.URL.Path == "/upload/"+name: + body, _ := io.ReadAll(r.Body) + mu.Lock() + uploads[name] = string(body) + mu.Unlock() + w.WriteHeader(http.StatusOK) + case r.Method == http.MethodPatch && r.URL.Path == mediaPath+"/same-media-id": + w.WriteHeader(http.StatusNoContent) + default: + http.NotFound(w, r) + } + })) + } + central := upstream("central") + defer central.Close() + tenant := upstream("tenant") + defer tenant.Close() + + store := newFakeUploadPlanStore() + createGateway := newTestGatewayWithStore(central.URL, map[string]string{"eu": tenant.URL}, store) + uploadGateway := newTestGatewayWithStore(central.URL, map[string]string{"eu": tenant.URL}, store) + createBody := `{"traceId":"trace","contentType":"image/png","contentLength":5,"sha256Hash":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","field":"input"}` + req := httptest.NewRequest(http.MethodPost, tenantPrefix+"eu"+mediaPath, strings.NewReader(createBody)) + req.Header.Set("Authorization", "Basic tenant") + resp := httptest.NewRecorder() + + createGateway.handle(resp, req) + if resp.Code != http.StatusCreated { + t.Fatalf("create status = %d, body = %s", resp.Code, resp.Body.String()) + } + var create mediaUploadResponse + if err := json.NewDecoder(resp.Body).Decode(&create); err != nil { + t.Fatal(err) + } + if create.MediaID != "same-media-id" || create.UploadURL == nil || !strings.Contains(*create.UploadURL, mediaUploadProxyPath) { + t.Fatalf("unexpected create response: %#v", create) + } + uploadID := strings.TrimPrefix(newUploadURLPath(t, *create.UploadURL), mediaUploadProxyPath) + store.mu.Lock() + storedPlan := store.plans[uploadID] + store.mu.Unlock() + storedPlanJSON, err := json.Marshal(storedPlan) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(storedPlanJSON), "Basic ") { + t.Fatalf("stored upload plan leaked authorization: %s", storedPlanJSON) + } + + uploadReq := httptest.NewRequest(http.MethodPut, *create.UploadURL, strings.NewReader("hello")) + uploadReq.Header.Set("Content-Type", "image/png") + uploadResp := httptest.NewRecorder() + uploadGateway.handle(uploadResp, uploadReq) + if uploadResp.Code != http.StatusOK { + t.Fatalf("upload status = %d, body = %s", uploadResp.Code, uploadResp.Body.String()) + } + + patchReq := httptest.NewRequest(http.MethodPatch, tenantPrefix+"eu"+mediaPath+"/same-media-id", strings.NewReader(`{"uploadHttpStatus":200}`)) + patchReq.Header.Set("Authorization", "Basic tenant") + patchResp := httptest.NewRecorder() + uploadGateway.handle(patchResp, patchReq) + if patchResp.Code != http.StatusNoContent { + t.Fatalf("patch status = %d, body = %s", patchResp.Code, patchResp.Body.String()) + } + + mu.Lock() + defer mu.Unlock() + if uploads["central"] != "hello" || uploads["tenant"] != "hello" { + t.Fatalf("uploads = %#v", uploads) + } +} + +func TestMediaUploadRejectsInvalidIDBeforeReadingBody(t *testing.T) { + t.Parallel() + + gw := newTestGateway("http://central.invalid", nil) + reader := &failingReader{} + req := httptest.NewRequest(http.MethodPut, mediaUploadProxyPath+"not-valid", reader) + resp := httptest.NewRecorder() + + gw.handle(resp, req) + if resp.Code != http.StatusNotFound { + t.Fatalf("status = %d, body = %s", resp.Code, resp.Body.String()) + } + if reader.read { + t.Fatal("invalid upload id should not read request body") + } +} + +func TestMediaUploadIsOneTime(t *testing.T) { + t.Parallel() + + var uploads int + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPut || r.URL.Path != "/upload" { + http.NotFound(w, r) + return + } + uploads++ + w.WriteHeader(http.StatusOK) + })) + defer upstream.Close() + + store := newFakeUploadPlanStore() + uploadID := "abcdef1234" + store.Put(context.Background(), uploadID, uploadPlan{ + ExpiresAt: time.Now().Add(time.Hour), + ContentLength: 5, + Destinations: []uploadDestination{{ + Name: "central", + UploadURL: upstream.URL + "/upload", + }}, + }) + gw := newTestGatewayWithStore(upstream.URL, nil, store) + + for index, expectedStatus := range []int{http.StatusOK, http.StatusNotFound} { + req := httptest.NewRequest(http.MethodPut, mediaUploadProxyPath+uploadID, strings.NewReader("hello")) + req.Header.Set("Content-Type", "image/png") + resp := httptest.NewRecorder() + gw.handle(resp, req) + if resp.Code != expectedStatus { + t.Fatalf("attempt %d status = %d, body = %s", index+1, resp.Code, resp.Body.String()) + } + } + if uploads != 1 { + t.Fatalf("uploads = %d", uploads) + } +} + +func TestMediaUploadOversizeRestoresPlanForRetry(t *testing.T) { + t.Parallel() + + var uploads int + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPut || r.URL.Path != "/upload" { + http.NotFound(w, r) + return + } + uploads++ + w.WriteHeader(http.StatusOK) + })) + defer upstream.Close() + + store := newFakeUploadPlanStore() + uploadID := "abcdef1234" + store.Put(context.Background(), uploadID, uploadPlan{ + ExpiresAt: time.Now().Add(time.Hour), + ContentLength: 1, + Destinations: []uploadDestination{{ + Name: "central", + UploadURL: upstream.URL + "/upload", + }}, + }) + gw := newTestGatewayWithStore(upstream.URL, nil, store) + + oversizeReq := httptest.NewRequest( + http.MethodPut, + mediaUploadProxyPath+uploadID, + strings.NewReader(strings.Repeat("x", int(maxUploadBytes(1))+1)), + ) + oversizeReq.Header.Set("Content-Type", "image/png") + oversizeResp := httptest.NewRecorder() + gw.handle(oversizeResp, oversizeReq) + if oversizeResp.Code != http.StatusBadRequest { + t.Fatalf("oversize status = %d, body = %s", oversizeResp.Code, oversizeResp.Body.String()) + } + + retryReq := httptest.NewRequest(http.MethodPut, mediaUploadProxyPath+uploadID, strings.NewReader("ok")) + retryReq.Header.Set("Content-Type", "image/png") + retryResp := httptest.NewRecorder() + gw.handle(retryResp, retryReq) + if retryResp.Code != http.StatusOK { + t.Fatalf("retry status = %d, body = %s", retryResp.Code, retryResp.Body.String()) + } + if uploads != 1 { + t.Fatalf("uploads = %d", uploads) + } +} + +func TestMediaUploadUnsupportedContentTypeRestoresPlanForRetry(t *testing.T) { + t.Parallel() + + var uploads int + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPut || r.URL.Path != "/upload" { + http.NotFound(w, r) + return + } + uploads++ + w.WriteHeader(http.StatusOK) + })) + defer upstream.Close() + + store := newFakeUploadPlanStore() + uploadID := "abcdef1234" + store.Put(context.Background(), uploadID, uploadPlan{ + ExpiresAt: time.Now().Add(time.Hour), + ContentLength: 5, + Destinations: []uploadDestination{{ + Name: "central", + UploadURL: upstream.URL + "/upload", + }}, + }) + gw := newTestGatewayWithStore(upstream.URL, nil, store) + + badReq := httptest.NewRequest(http.MethodPut, mediaUploadProxyPath+uploadID, strings.NewReader("hello")) + badReq.Header.Set("Content-Type", "text/html") + badResp := httptest.NewRecorder() + gw.handle(badResp, badReq) + if badResp.Code != http.StatusBadGateway { + t.Fatalf("bad content-type status = %d, body = %s", badResp.Code, badResp.Body.String()) + } + + retryReq := httptest.NewRequest(http.MethodPut, mediaUploadProxyPath+uploadID, strings.NewReader("hello")) + retryReq.Header.Set("Content-Type", "image/png") + retryResp := httptest.NewRecorder() + gw.handle(retryResp, retryReq) + if retryResp.Code != http.StatusOK { + t.Fatalf("retry status = %d, body = %s", retryResp.Code, retryResp.Body.String()) + } + if uploads != 1 { + t.Fatalf("uploads = %d", uploads) + } +} + +func TestMediaCreateUsesConfiguredPublicUploadURL(t *testing.T) { + t.Parallel() + + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != mediaPath { + http.NotFound(w, r) + return + } + uploadURL := "http://" + r.Host + "/upload" + writeJSON(w, http.StatusCreated, mediaUploadResponse{ + MediaID: "same-media-id", + UploadURL: &uploadURL, + }) + })) + defer upstream.Close() + + gw := newTestGateway(upstream.URL, nil) + gw.cfg.publicURL = "https://fanout.example.com/base" + req := httptest.NewRequest(http.MethodPost, mediaPath, strings.NewReader(`{"contentLength":5}`)) + req.Host = "attacker.example.com" + req.Header.Set("X-Forwarded-Host", "attacker.example.com") + resp := httptest.NewRecorder() + + gw.handle(resp, req) + if resp.Code != http.StatusCreated { + t.Fatalf("status = %d, body = %s", resp.Code, resp.Body.String()) + } + var create mediaUploadResponse + if err := json.NewDecoder(resp.Body).Decode(&create); err != nil { + t.Fatal(err) + } + if create.UploadURL == nil || !strings.HasPrefix(*create.UploadURL, "https://fanout.example.com/base/") { + t.Fatalf("unexpected upload URL: %#v", create.UploadURL) + } + if strings.Contains(*create.UploadURL, "attacker.example.com") { + t.Fatalf("upload URL trusted request host: %s", *create.UploadURL) + } +} + +func TestMediaGetUsesTenantDestinationForTenantRoute(t *testing.T) { + t.Parallel() + + var centralGets int + central := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + centralGets++ + writeJSON(w, http.StatusOK, map[string]string{"url": "central"}) + })) + defer central.Close() + var tenantGets int + tenant := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + tenantGets++ + if r.URL.Path != mediaPath+"/media-id" { + t.Fatalf("unexpected tenant path %s", r.URL.Path) + } + if got := r.Header.Get("Authorization"); got != "Basic tenant" { + t.Fatalf("tenant auth = %q", got) + } + writeJSON(w, http.StatusOK, map[string]string{"url": "tenant"}) + })) + defer tenant.Close() + + gw := newTestGateway(central.URL, map[string]string{"eu": tenant.URL}) + req := httptest.NewRequest(http.MethodGet, tenantPrefix+"eu"+mediaPath+"/media-id", nil) + req.Header.Set("Authorization", "Basic tenant") + resp := httptest.NewRecorder() + + gw.handle(resp, req) + if resp.Code != http.StatusOK { + t.Fatalf("status = %d, body = %s", resp.Code, resp.Body.String()) + } + if centralGets != 0 || tenantGets != 1 { + t.Fatalf("centralGets=%d tenantGets=%d", centralGets, tenantGets) + } + if !strings.Contains(resp.Body.String(), "tenant") { + t.Fatalf("unexpected body: %s", resp.Body.String()) + } +} + +func TestMediaGetUsesCentralDestinationForCentralRoute(t *testing.T) { + t.Parallel() + + var centralGets int + central := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + centralGets++ + if got := r.Header.Get("Authorization"); got != "Basic central" { + t.Fatalf("central auth = %q", got) + } + writeJSON(w, http.StatusOK, map[string]string{"url": "central"}) + })) + defer central.Close() + var tenantGets int + tenant := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + tenantGets++ + w.WriteHeader(http.StatusOK) + })) + defer tenant.Close() + + gw := newTestGateway(central.URL, map[string]string{"eu": tenant.URL}) + req := httptest.NewRequest(http.MethodGet, mediaPath+"/media-id", nil) + resp := httptest.NewRecorder() + + gw.handle(resp, req) + if resp.Code != http.StatusOK { + t.Fatalf("status = %d, body = %s", resp.Code, resp.Body.String()) + } + if centralGets != 1 || tenantGets != 0 { + t.Fatalf("centralGets=%d tenantGets=%d", centralGets, tenantGets) + } + if !strings.Contains(resp.Body.String(), "central") { + t.Fatalf("unexpected body: %s", resp.Body.String()) + } +} + +func TestMediaCreateRejectsDifferentMediaIDs(t *testing.T) { + t.Parallel() + + upstream := func(mediaID string) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != mediaPath { + http.NotFound(w, r) + return + } + writeJSON(w, http.StatusCreated, mediaUploadResponse{MediaID: mediaID}) + })) + } + central := upstream("central-id") + defer central.Close() + tenant := upstream("tenant-id") + defer tenant.Close() + + gw := newTestGateway(central.URL, map[string]string{"eu": tenant.URL}) + req := httptest.NewRequest(http.MethodPost, tenantPrefix+"eu"+mediaPath, strings.NewReader(`{"contentLength":0}`)) + req.Header.Set("Authorization", "Basic tenant") + resp := httptest.NewRecorder() + + gw.handle(resp, req) + if resp.Code != http.StatusBadGateway { + t.Fatalf("status = %d, body = %s", resp.Code, resp.Body.String()) + } + if !strings.Contains(resp.Body.String(), "media IDs differ") { + t.Fatalf("unexpected body: %s", resp.Body.String()) + } + + metrics := scrapeMetrics(t, gw) + if !strings.Contains(metrics, `langfuse_fanout_media_divergence_total{destination="tenant_eu",kind="media_id"} 1`) { + t.Fatalf("missing media_id divergence metric:\n%s", metrics) + } +} + +func TestMetricsEndpointRequiresBearerToken(t *testing.T) { + t.Parallel() + + gw := newTestGateway("http://central.invalid", nil) + + unauthorized := httptest.NewRecorder() + gw.handle(unauthorized, httptest.NewRequest(http.MethodGet, metricsPath, nil)) + if unauthorized.Code != http.StatusUnauthorized { + t.Fatalf("unauthorized status = %d", unauthorized.Code) + } + + wrong := httptest.NewRecorder() + wrongReq := httptest.NewRequest(http.MethodGet, metricsPath, nil) + wrongReq.Header.Set("Authorization", "Bearer wrong") + gw.handle(wrong, wrongReq) + if wrong.Code != http.StatusUnauthorized { + t.Fatalf("wrong token status = %d", wrong.Code) + } + + authorized := httptest.NewRecorder() + authorizedReq := httptest.NewRequest(http.MethodGet, metricsPath, nil) + authorizedReq.Header.Set("Authorization", "Bearer test-secret") + gw.handle(authorized, authorizedReq) + if authorized.Code != http.StatusOK { + t.Fatalf("authorized status = %d, body = %s", authorized.Code, authorized.Body.String()) + } + if !strings.Contains(authorized.Body.String(), "go_goroutines") { + t.Fatalf("missing gateway metrics:\n%s", authorized.Body.String()) + } +} + +func TestTraceProxyRecordsPrometheusMetrics(t *testing.T) { + t.Parallel() + + collector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte("{}")) + })) + defer collector.Close() + + gw := newTestGatewayWithCollector(collector.URL) + body := buildTraceRequest(t, nil) + req := httptest.NewRequest(http.MethodPost, tenantPrefix+"eu"+otelTracePath, bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/x-protobuf") + req.Header.Set("Authorization", "Basic tenant") + resp := httptest.NewRecorder() + + gw.handle(resp, req) + if resp.Code != http.StatusOK { + t.Fatalf("status = %d, body = %s", resp.Code, resp.Body.String()) + } + + metrics := scrapeMetrics(t, gw) + if !strings.Contains(metrics, `langfuse_fanout_trace_exports_total{destination="tenant_eu",result="success"} 1`) { + t.Fatalf("missing trace export metric:\n%s", metrics) + } + if !strings.Contains(metrics, `langfuse_fanout_upstream_requests_total{destination="collector",operation="trace_collector",status_class="2xx"} 1`) { + t.Fatalf("missing upstream collector metric:\n%s", metrics) + } +} + +func newTestGateway(centralURL string, tenants map[string]string) *gateway { + return newTestGatewayWithStore(centralURL, tenants, newFakeUploadPlanStore()) +} + +func newTestGatewayWithStore(centralURL string, tenants map[string]string, store uploadPlanStore) *gateway { + return newGateway(config{ + traceCollectorURL: "http://collector.invalid", + publicURL: "http://fanout.local:4318", + metricsSecret: "test-secret", + central: destination{ + name: centralName, + baseURL: centralURL, + authorization: "Basic central", + }, + tenants: tenants, + uploadStore: store, + client: &http.Client{Timeout: 5 * time.Second}, + }) +} + +func newTestGatewayWithCollector(collectorURL string) *gateway { + gateway := newTestGateway("http://central.invalid", map[string]string{"eu": "http://tenant.invalid"}) + gateway.cfg.traceCollectorURL = collectorURL + return gateway +} + +func buildTraceRequest(t *testing.T, attrs map[string]string) []byte { + t.Helper() + spanAttrs := make([]*commonv1.KeyValue, 0, len(attrs)) + for key, value := range attrs { + spanAttrs = append(spanAttrs, &commonv1.KeyValue{ + Key: key, + Value: &commonv1.AnyValue{ + Value: &commonv1.AnyValue_StringValue{StringValue: value}, + }, + }) + } + request := &tracepb.ExportTraceServiceRequest{ + ResourceSpans: []*tracev1.ResourceSpans{{ + Resource: &resourcev1.Resource{}, + ScopeSpans: []*tracev1.ScopeSpans{{ + Spans: []*tracev1.Span{{ + TraceId: []byte("1234567890123456"), + SpanId: []byte("12345678"), + Name: "test-span", + Attributes: spanAttrs, + }}, + }}, + }}, + } + body, err := proto.Marshal(request) + if err != nil { + t.Fatal(err) + } + return body +} + +func buildJSONTraceRequest(t *testing.T, attrs map[string]any) []byte { + t.Helper() + spanAttrs := make([]map[string]any, 0, len(attrs)) + for key, value := range attrs { + anyValue := map[string]any{} + switch typed := value.(type) { + case bool: + anyValue["boolValue"] = typed + default: + anyValue["stringValue"] = typed + } + spanAttrs = append(spanAttrs, map[string]any{ + "key": key, + "value": anyValue, + }) + } + request := map[string]any{ + "resourceSpans": []any{ + map[string]any{ + "resource": map[string]any{}, + "scopeSpans": []any{ + map[string]any{ + "spans": []any{ + map[string]any{ + "traceId": "31323334353637383930313233343536", + "spanId": "3132333435363738", + "name": "test-span", + "attributes": spanAttrs, + }, + }, + }, + }, + }, + }, + } + body, err := json.Marshal(request) + if err != nil { + t.Fatal(err) + } + return body +} + +func parseTraceAttributes(t *testing.T, body []byte) map[string]string { + t.Helper() + var request tracepb.ExportTraceServiceRequest + if err := proto.Unmarshal(body, &request); err != nil { + t.Fatal(err) + } + result := map[string]string{} + for _, resourceSpan := range request.ResourceSpans { + for _, scopeSpan := range resourceSpan.ScopeSpans { + for _, span := range scopeSpan.Spans { + for _, attr := range span.Attributes { + result[attr.Key] = stringValue(attr.Value) + } + } + } + } + return result +} + +func parseJSONTraceAttributes(t *testing.T, body []byte) map[string]string { + t.Helper() + var request map[string]any + if err := json.Unmarshal(body, &request); err != nil { + t.Fatal(err) + } + result := map[string]string{} + resourceSpans, _ := request["resourceSpans"].([]any) + for _, resourceSpan := range resourceSpans { + resourceSpanMap, _ := resourceSpan.(map[string]any) + scopeSpans, _ := resourceSpanMap["scopeSpans"].([]any) + for _, scopeSpan := range scopeSpans { + scopeSpanMap, _ := scopeSpan.(map[string]any) + spans, _ := scopeSpanMap["spans"].([]any) + for _, span := range spans { + spanMap, _ := span.(map[string]any) + attrs, _ := spanMap["attributes"].([]any) + for _, attr := range attrs { + attrMap, _ := attr.(map[string]any) + valueMap, _ := attrMap["value"].(map[string]any) + if key, ok := attrMap["key"].(string); ok { + if value, ok := valueMap["stringValue"].(string); ok { + result[key] = value + } + } + } + } + } + } + return result +} + +func scrapeMetrics(t *testing.T, gw *gateway) string { + t.Helper() + req := httptest.NewRequest(http.MethodGet, metricsPath, nil) + req.Header.Set("Authorization", "Bearer test-secret") + resp := httptest.NewRecorder() + gw.handle(resp, req) + if resp.Code != http.StatusOK { + t.Fatalf("metrics status = %d, body = %s", resp.Code, resp.Body.String()) + } + return resp.Body.String() +} + +func newUploadURLPath(t *testing.T, value string) string { + t.Helper() + parsed, err := url.Parse(value) + if err != nil { + t.Fatal(err) + } + return parsed.Path +} + +type fakeUploadPlanStore struct { + mu sync.Mutex + plans map[string]uploadPlan +} + +func newFakeUploadPlanStore() *fakeUploadPlanStore { + return &fakeUploadPlanStore{plans: map[string]uploadPlan{}} +} + +func (s *fakeUploadPlanStore) Put(_ context.Context, uploadID string, plan uploadPlan) error { + s.mu.Lock() + defer s.mu.Unlock() + s.plans[uploadID] = plan + return nil +} + +func (s *fakeUploadPlanStore) Take(_ context.Context, uploadID string) (uploadPlan, bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + plan, ok := s.plans[uploadID] + delete(s.plans, uploadID) + return plan, ok, nil +} + +func (s *fakeUploadPlanStore) Ping(_ context.Context) error { + return nil +} + +func (s *fakeUploadPlanStore) Close() error { + return nil +} + +type failingReader struct { + read bool +} + +func (r *failingReader) Read(_ []byte) (int, error) { + r.read = true + return 0, errors.New("read should not be called") +} diff --git a/otel/langfuse-fanout/cmd/langfuse-fanout/metrics.go b/otel/langfuse-fanout/cmd/langfuse-fanout/metrics.go new file mode 100644 index 0000000000..22259732e8 --- /dev/null +++ b/otel/langfuse-fanout/cmd/langfuse-fanout/metrics.go @@ -0,0 +1,242 @@ +package main + +import ( + "fmt" + "net/http" + "strings" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/collectors" +) + +type gatewayMetrics struct { + registry *prometheus.Registry + httpRequests *prometheus.CounterVec + httpDuration *prometheus.HistogramVec + upstreamRequests *prometheus.CounterVec + upstreamDuration *prometheus.HistogramVec + traceExports *prometheus.CounterVec + mediaDivergence *prometheus.CounterVec + uploadPlansCreated prometheus.Counter + uploadPlansCompleted prometheus.Counter + uploadPlanMisses prometheus.Counter + uploadPlanStoreErrors *prometheus.CounterVec + uploadBytes prometheus.Histogram +} + +func newGatewayMetrics() *gatewayMetrics { + registry := prometheus.NewRegistry() + registry.MustRegister( + collectors.NewGoCollector(), + collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}), + ) + + metrics := &gatewayMetrics{ + registry: registry, + httpRequests: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "langfuse_fanout_http_requests_total", + Help: "Total HTTP requests handled by the Langfuse fanout gateway.", + }, []string{"method", "path", "status"}), + httpDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "langfuse_fanout_http_request_duration_seconds", + Help: "HTTP request duration for the Langfuse fanout gateway.", + Buckets: []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30}, + }, []string{"method", "path", "status"}), + upstreamRequests: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "langfuse_fanout_upstream_requests_total", + Help: "Total upstream requests made by the Langfuse fanout gateway.", + }, []string{"operation", "destination", "status_class"}), + upstreamDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "langfuse_fanout_upstream_request_duration_seconds", + Help: "Upstream request duration for Langfuse and collector calls.", + Buckets: []float64{0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30}, + }, []string{"operation", "destination", "status_class"}), + traceExports: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "langfuse_fanout_trace_exports_total", + Help: "Total trace export attempts through the Langfuse fanout gateway.", + }, []string{"destination", "result"}), + mediaDivergence: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "langfuse_fanout_media_divergence_total", + Help: "Media fanout upstream response divergence by kind. Values are counts only; no media IDs or URLs are exposed.", + }, []string{"kind", "destination"}), + uploadPlansCreated: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "langfuse_fanout_media_upload_plans_created_total", + Help: "Total media upload fanout plans stored for SDK byte upload.", + }), + uploadPlansCompleted: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "langfuse_fanout_media_upload_plans_completed_total", + Help: "Total media upload fanout plans consumed by SDK byte upload.", + }), + uploadPlanMisses: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "langfuse_fanout_media_upload_plan_misses_total", + Help: "Total media upload attempts that did not find a stored fanout plan.", + }), + uploadPlanStoreErrors: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "langfuse_fanout_media_upload_plan_store_errors_total", + Help: "Total Redis upload plan store errors by operation.", + }, []string{"operation"}), + uploadBytes: prometheus.NewHistogram(prometheus.HistogramOpts{ + Name: "langfuse_fanout_media_upload_bytes", + Help: "Configured media upload content length for fanout upload plans.", + Buckets: []float64{1_000, 10_000, 100_000, 1_000_000, 5_000_000, 10_000_000, 25_000_000, 50_000_000, 100_000_000, 250_000_000}, + }), + } + + registry.MustRegister( + metrics.httpRequests, + metrics.httpDuration, + metrics.upstreamRequests, + metrics.upstreamDuration, + metrics.traceExports, + metrics.mediaDivergence, + metrics.uploadPlansCreated, + metrics.uploadPlansCompleted, + metrics.uploadPlanMisses, + metrics.uploadPlanStoreErrors, + metrics.uploadBytes, + ) + return metrics +} + +func (m *gatewayMetrics) recordHTTP(method string, path string, status int, duration time.Duration) { + if m == nil { + return + } + labels := prometheus.Labels{ + "method": method, + "path": path, + "status": fmt.Sprintf("%d", status), + } + m.httpRequests.With(labels).Inc() + m.httpDuration.With(labels).Observe(duration.Seconds()) +} + +func (m *gatewayMetrics) recordUpstream(operation string, destination string, statusClass string, duration time.Duration) { + if m == nil { + return + } + labels := prometheus.Labels{ + "operation": operation, + "destination": normalizeMetricLabel(destination), + "status_class": statusClass, + } + m.upstreamRequests.With(labels).Inc() + m.upstreamDuration.With(labels).Observe(duration.Seconds()) +} + +func (m *gatewayMetrics) recordTraceExport(destination string, result string) { + if m == nil { + return + } + m.traceExports.WithLabelValues(normalizeMetricLabel(destination), result).Inc() +} + +func (m *gatewayMetrics) recordMediaDivergence(kind string, destination string) { + if m == nil { + return + } + m.mediaDivergence.WithLabelValues(normalizeMetricLabel(kind), normalizeMetricLabel(destination)).Inc() +} + +func (m *gatewayMetrics) recordUploadPlanCreated(contentLength int64) { + if m == nil { + return + } + m.uploadPlansCreated.Inc() + if contentLength > 0 { + m.uploadBytes.Observe(float64(contentLength)) + } +} + +func (m *gatewayMetrics) recordUploadPlanCompleted() { + if m == nil { + return + } + m.uploadPlansCompleted.Inc() +} + +func (m *gatewayMetrics) recordUploadPlanMiss() { + if m == nil { + return + } + m.uploadPlanMisses.Inc() +} + +func (m *gatewayMetrics) recordUploadPlanStoreError(operation string) { + if m == nil { + return + } + m.uploadPlanStoreErrors.WithLabelValues(normalizeMetricLabel(operation)).Inc() +} + +type statusRecorder struct { + http.ResponseWriter + status int +} + +func (w *statusRecorder) WriteHeader(status int) { + w.status = status + w.ResponseWriter.WriteHeader(status) +} + +func normalizeMetricPath(path string) string { + route := parseRoute(path) + switch { + case route.path == otelTracePath: + if route.destination == "" { + return otelTracePath + } + return tenantPrefix + "#destination" + otelTracePath + case route.path == mediaPath: + if route.destination == "" { + return mediaPath + } + return tenantPrefix + "#destination" + mediaPath + case strings.HasPrefix(route.path, mediaPath+"/"): + if route.destination == "" { + return mediaPath + "/#mediaId" + } + return tenantPrefix + "#destination" + mediaPath + "/#mediaId" + case strings.HasPrefix(path, mediaUploadProxyPath): + return mediaUploadProxyPath + "#uploadId" + case path == "/healthz": + return "/healthz" + default: + return "/#path" + } +} + +func normalizeMetricLabel(value string) string { + value = strings.TrimSpace(strings.ToLower(value)) + if value == "" { + return "unknown" + } + var builder strings.Builder + for _, r := range value { + switch { + case r >= 'a' && r <= 'z': + builder.WriteRune(r) + case r >= '0' && r <= '9': + builder.WriteRune(r) + case r == '_' || r == '-' || r == ':': + builder.WriteRune(r) + default: + builder.WriteRune('_') + } + if builder.Len() >= 80 { + break + } + } + if builder.Len() == 0 { + return "unknown" + } + return builder.String() +} + +func statusClass(status int) string { + if status <= 0 { + return "error" + } + return fmt.Sprintf("%dxx", status/100) +} diff --git a/otel/langfuse-fanout/go.mod b/otel/langfuse-fanout/go.mod new file mode 100644 index 0000000000..eed620d0e8 --- /dev/null +++ b/otel/langfuse-fanout/go.mod @@ -0,0 +1,29 @@ +module github.com/danny-avila/librechat/otel/langfuse-fanout + +go 1.23.0 + +require ( + github.com/prometheus/client_golang v1.23.2 + github.com/redis/go-redis/v9 v9.17.2 + go.opentelemetry.io/proto/otlp v1.9.0 + google.golang.org/protobuf v1.36.10 +) + +require ( + github.com/beorn7/perks v1.0.1 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect + github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 // indirect + github.com/kr/text v0.2.0 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/prometheus/client_model v0.6.2 // indirect + github.com/prometheus/common v0.66.1 // indirect + github.com/prometheus/procfs v0.16.1 // indirect + go.yaml.in/yaml/v2 v2.4.2 // indirect + golang.org/x/net v0.43.0 // indirect + golang.org/x/sys v0.35.0 // indirect + golang.org/x/text v0.28.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20250825161204-c5933d9347a5 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20250825161204-c5933d9347a5 // indirect + google.golang.org/grpc v1.75.1 // indirect +) diff --git a/otel/langfuse-fanout/go.sum b/otel/langfuse-fanout/go.sum new file mode 100644 index 0000000000..a662c24fa9 --- /dev/null +++ b/otel/langfuse-fanout/go.sum @@ -0,0 +1,90 @@ +github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= +github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 h1:8Tjv8EJ+pM1xP8mK6egEbD1OgnVTyacbefKhmbLhIhU= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2/go.mod h1:pkJQ2tZHJ0aFOVEEot6oZmaVEZcRme73eIFmhiVuRWs= +github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= +github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= +github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o= +github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg= +github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= +github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= +github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9ZoGs= +github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA= +github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg= +github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is= +github.com/redis/go-redis/v9 v9.17.2 h1:P2EGsA4qVIM3Pp+aPocCJ7DguDHhqrXNhVcEp4ViluI= +github.com/redis/go-redis/v9 v9.17.2/go.mod h1:u410H11HMLoB+TP67dz8rL9s6QW2j76l0//kSOd3370= +github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ= +github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA= +go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A= +go.opentelemetry.io/otel v1.37.0 h1:9zhNfelUvx0KBfu/gb+ZgeAfAgtWrfHJZcAqFC228wQ= +go.opentelemetry.io/otel v1.37.0/go.mod h1:ehE/umFRLnuLa/vSccNq9oS1ErUlkkK71gMcN34UG8I= +go.opentelemetry.io/otel/metric v1.37.0 h1:mvwbQS5m0tbmqML4NqK+e3aDiO02vsf/WgbsdpcPoZE= +go.opentelemetry.io/otel/metric v1.37.0/go.mod h1:04wGrZurHYKOc+RKeye86GwKiTb9FKm1WHtO+4EVr2E= +go.opentelemetry.io/otel/sdk v1.37.0 h1:ItB0QUqnjesGRvNcmAcU0LyvkVyGJ2xftD29bWdDvKI= +go.opentelemetry.io/otel/sdk v1.37.0/go.mod h1:VredYzxUvuo2q3WRcDnKDjbdvmO0sCzOvVAiY+yUkAg= +go.opentelemetry.io/otel/sdk/metric v1.37.0 h1:90lI228XrB9jCMuSdA0673aubgRobVZFhbjxHHspCPc= +go.opentelemetry.io/otel/sdk/metric v1.37.0/go.mod h1:cNen4ZWfiD37l5NhS+Keb5RXVWZWpRE+9WyVCpbo5ps= +go.opentelemetry.io/otel/trace v1.37.0 h1:HLdcFNbRQBE2imdSEgm/kwqmQj1Or1l/7bW6mxVK7z4= +go.opentelemetry.io/otel/trace v1.37.0/go.mod h1:TlgrlQ+PtQO5XFerSPUYG0JSgGyryXewPGyayAWSBS0= +go.opentelemetry.io/proto/otlp v1.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjceRb/A= +go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI= +go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU= +golang.org/x/net v0.43.0 h1:lat02VYK2j4aLzMzecihNvTlJNQUq316m2Mr9rnM6YE= +golang.org/x/net v0.43.0/go.mod h1:vhO1fvI4dGsIjh73sWfUVjj3N7CA9WkKJNQm2svM6Jg= +golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI= +golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +golang.org/x/text v0.28.0 h1:rhazDwis8INMIwQ4tpjLDzUhx6RlXqZNPEM0huQojng= +golang.org/x/text v0.28.0/go.mod h1:U8nCwOR8jO/marOQ0QbDiOngZVEBB7MAiitBuMjXiNU= +gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk= +gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E= +google.golang.org/genproto/googleapis/api v0.0.0-20250825161204-c5933d9347a5 h1:BIRfGDEjiHRrk0QKZe3Xv2ieMhtgRGeLcZQ0mIVn4EY= +google.golang.org/genproto/googleapis/api v0.0.0-20250825161204-c5933d9347a5/go.mod h1:j3QtIyytwqGr1JUDtYXwtMXWPKsEa5LtzIFN1Wn5WvE= +google.golang.org/genproto/googleapis/rpc v0.0.0-20250825161204-c5933d9347a5 h1:eaY8u2EuxbRv7c3NiGK0/NedzVsCcV6hDuU5qPX5EGE= +google.golang.org/genproto/googleapis/rpc v0.0.0-20250825161204-c5933d9347a5/go.mod h1:M4/wBTSeyLxupu3W3tJtOgB14jILAS/XWPSSa3TAlJc= +google.golang.org/grpc v1.75.1 h1:/ODCNEuf9VghjgO3rqLcfg8fiOP0nSluljWFlDxELLI= +google.golang.org/grpc v1.75.1/go.mod h1:JtPAzKiq4v1xcAB2hydNlWI2RnF85XXcV0mhKXr2ecQ= +google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE= +google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/otel/langfuse-fanout/otelcol.yaml b/otel/langfuse-fanout/otelcol.yaml index 406f833806..21b1567d28 100644 --- a/otel/langfuse-fanout/otelcol.yaml +++ b/otel/langfuse-fanout/otelcol.yaml @@ -9,7 +9,7 @@ receivers: otlp: protocols: http: - endpoint: 0.0.0.0:4318 + endpoint: ${env:LANGFUSE_FANOUT_OTEL_RECEIVER_ENDPOINT} include_metadata: true traces_url_path: /api/public/otel/v1/traces @@ -20,12 +20,12 @@ connectors: - context: span condition: attributes["librechat.langfuse.destination"] == "eu" pipelines: [traces/tenant_eu] - - context: span - condition: attributes["librechat.langfuse.destination"] == "us" - pipelines: [traces/tenant_us] - context: span condition: attributes["librechat.langfuse.destination"] == "jp" pipelines: [traces/tenant_jp] + - context: span + condition: attributes["librechat.langfuse.destination"] == "us" + pipelines: [traces/tenant_us] processors: memory_limiter: @@ -51,16 +51,17 @@ processors: send_batch_size: ${env:LANGFUSE_FANOUT_BATCH_SEND_SIZE} metadata_keys: [authorization] metadata_cardinality_limit: ${env:LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT} - batch/by_auth_us: - timeout: ${env:LANGFUSE_FANOUT_BATCH_TIMEOUT} - send_batch_size: ${env:LANGFUSE_FANOUT_BATCH_SEND_SIZE} - metadata_keys: [authorization] - metadata_cardinality_limit: ${env:LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT} batch/by_auth_jp: timeout: ${env:LANGFUSE_FANOUT_BATCH_TIMEOUT} send_batch_size: ${env:LANGFUSE_FANOUT_BATCH_SEND_SIZE} metadata_keys: [authorization] metadata_cardinality_limit: ${env:LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT} + batch/by_auth_us: + timeout: ${env:LANGFUSE_FANOUT_BATCH_TIMEOUT} + send_batch_size: ${env:LANGFUSE_FANOUT_BATCH_SEND_SIZE} + metadata_keys: [authorization] + metadata_cardinality_limit: ${env:LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT} + exporters: otlphttp/central: # Langfuse Cloud base URL options: https://cloud.langfuse.com (EU), @@ -75,18 +76,19 @@ exporters: authenticator: headers_setter/tenant_passthrough headers: x-langfuse-ingestion-version: '4' - otlphttp/tenant_us: - endpoint: '${env:LANGFUSE_FANOUT_TENANT_US_BASE_URL}/api/public/otel' - auth: - authenticator: headers_setter/tenant_passthrough - headers: - x-langfuse-ingestion-version: '4' otlphttp/tenant_jp: endpoint: '${env:LANGFUSE_FANOUT_TENANT_JP_BASE_URL}/api/public/otel' auth: authenticator: headers_setter/tenant_passthrough headers: x-langfuse-ingestion-version: '4' + otlphttp/tenant_us: + endpoint: '${env:LANGFUSE_FANOUT_TENANT_US_BASE_URL}/api/public/otel' + auth: + authenticator: headers_setter/tenant_passthrough + headers: + x-langfuse-ingestion-version: '4' + service: extensions: [headers_setter/tenant_passthrough] pipelines: @@ -102,11 +104,11 @@ service: receivers: [routing/langfuse_tenant_destination] processors: [attributes/drop_librechat_routing, batch/by_auth_eu] exporters: [otlphttp/tenant_eu] - traces/tenant_us: - receivers: [routing/langfuse_tenant_destination] - processors: [attributes/drop_librechat_routing, batch/by_auth_us] - exporters: [otlphttp/tenant_us] traces/tenant_jp: receivers: [routing/langfuse_tenant_destination] processors: [attributes/drop_librechat_routing, batch/by_auth_jp] exporters: [otlphttp/tenant_jp] + traces/tenant_us: + receivers: [routing/langfuse_tenant_destination] + processors: [attributes/drop_librechat_routing, batch/by_auth_us] + exporters: [otlphttp/tenant_us]