mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-08-28 04:37:37 +00:00
🩻 feat(langfuse): add tenant export telemetry (#15247)
This commit is contained in:
Ravi Kumar L
GitHub
parent
f34a49007d
commit
56f0cde9a5
11 changed files with 552 additions and 35 deletions
|
|
@ -260,7 +260,7 @@ Useful gateway metrics include:
|
|||
|
||||
- `langfuse_fanout_http_requests_total`
|
||||
- `langfuse_fanout_upstream_requests_total`
|
||||
- `langfuse_fanout_trace_exports_total`
|
||||
- `langfuse_fanout_trace_exports_total` (`destination`, `result`, and `tenant_id` labels)
|
||||
- `langfuse_fanout_media_upload_plans_created_total`
|
||||
- `langfuse_fanout_media_upload_plans_completed_total`
|
||||
- `langfuse_fanout_media_upload_plan_misses_total`
|
||||
|
|
@ -268,6 +268,20 @@ Useful gateway metrics include:
|
|||
- `langfuse_fanout_media_upload_bytes`
|
||||
- `langfuse_fanout_media_divergence_total`
|
||||
|
||||
LibreChat stamps `librechat.tenant.id`, `librechat.langfuse.export_plan`, and
|
||||
`librechat.langfuse.export_reason` on Langfuse run spans. The gateway reads the
|
||||
tenant ID from each OTLP batch for the trace export counter. Batches without a
|
||||
tenant ID use `<unknown>`; batches containing more than one tenant use `<multiple>`.
|
||||
Angle brackets keep these synthetic values outside LibreChat's accepted tenant-ID grammar.
|
||||
The `tenant_id` label is intentionally high-cardinality and must only receive
|
||||
traffic from trusted LibreChat deployments. Each distinct tenant creates a
|
||||
Prometheus time series for every destination and result combination.
|
||||
|
||||
Successful admin connection updates emit the structured log event
|
||||
`librechat.langfuse.connection.changed`. It includes the tenant, configuration
|
||||
state, destination, verification result, a primary `change`, and all `changes`.
|
||||
It does not include the Langfuse public or secret key.
|
||||
|
||||
`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
|
||||
|
|
|
|||
|
|
@ -42,6 +42,9 @@ const (
|
|||
metricsPath = "/metrics"
|
||||
tenantExportAttribute = "librechat.langfuse.tenant_export.enabled"
|
||||
tenantDestAttribute = "librechat.langfuse.destination"
|
||||
tenantIDAttribute = "librechat.tenant.id"
|
||||
unknownTenantID = "<unknown>"
|
||||
multipleTenantIDs = "<multiple>"
|
||||
)
|
||||
|
||||
type config struct {
|
||||
|
|
@ -269,19 +272,25 @@ func (g *gateway) handle(w http.ResponseWriter, r *http.Request) {
|
|||
func (g *gateway) handleTraces(w http.ResponseWriter, r *http.Request, route route) {
|
||||
body, err := readMaybeGzip(r)
|
||||
if err != nil {
|
||||
g.recordTraceExport(route, "error", unknownTenantID)
|
||||
g.writeGatewayError(w, r, route, "trace_export", http.StatusBadRequest, "failed to read request body", err)
|
||||
return
|
||||
}
|
||||
|
||||
contentType := r.Header.Get("Content-Type")
|
||||
if route.destination != "" &&
|
||||
tenantID := unknownTenantID
|
||||
authorizedTenantRoute := route.destination != "" &&
|
||||
g.cfg.tenants[route.destination] != "" &&
|
||||
strings.TrimSpace(r.Header.Get("Authorization")) != "" {
|
||||
body, err = addTenantRouteAttributes(body, contentType, route.destination)
|
||||
strings.TrimSpace(r.Header.Get("Authorization")) != ""
|
||||
if authorizedTenantRoute {
|
||||
body, tenantID, err = addTenantRouteAttributes(body, contentType, route.destination)
|
||||
if err != nil {
|
||||
g.recordTraceExport(route, "error", tenantID)
|
||||
g.writeGatewayError(w, r, route, "trace_route_attributes", http.StatusBadRequest, "failed to add OTLP tenant routing attributes", err)
|
||||
return
|
||||
}
|
||||
} else if g.metrics != nil {
|
||||
tenantID = extractTraceTenantID(body, contentType)
|
||||
}
|
||||
|
||||
contentEncoding := ""
|
||||
|
|
@ -289,6 +298,7 @@ func (g *gateway) handleTraces(w http.ResponseWriter, r *http.Request, route rou
|
|||
contentEncoding = "gzip"
|
||||
body, err = gzipBytes(body)
|
||||
if err != nil {
|
||||
g.recordTraceExport(route, "error", tenantID)
|
||||
g.writeGatewayError(w, r, route, "trace_gzip", http.StatusInternalServerError, "failed to encode request body", err)
|
||||
return
|
||||
}
|
||||
|
|
@ -296,13 +306,13 @@ func (g *gateway) handleTraces(w http.ResponseWriter, r *http.Request, route rou
|
|||
|
||||
resp, err := g.forwardTraceToCollector(r.Context(), r.Header, body, contentType, contentEncoding)
|
||||
if err != nil {
|
||||
g.recordTraceExport(route, "error")
|
||||
g.recordTraceExport(route, "error", tenantID)
|
||||
g.writeGatewayError(w, r, route, "trace_collector", http.StatusBadGateway, "trace collector export failed", err)
|
||||
return
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
g.recordTraceExport(route, "success")
|
||||
g.recordTraceExport(route, "success", tenantID)
|
||||
copyResponseHeaders(w.Header(), resp.Header)
|
||||
w.WriteHeader(resp.StatusCode)
|
||||
_, _ = io.Copy(w, resp.Body)
|
||||
|
|
@ -734,11 +744,11 @@ func (g *gateway) doUpstreamWithClient(client *http.Client, req *http.Request, o
|
|||
return resp, nil
|
||||
}
|
||||
|
||||
func (g *gateway) recordTraceExport(route route, result string) {
|
||||
func (g *gateway) recordTraceExport(route route, result string, tenantID string) {
|
||||
if g.metrics == nil {
|
||||
return
|
||||
}
|
||||
g.metrics.recordTraceExport(routeDestinationLabel(route), result)
|
||||
g.metrics.recordTraceExport(routeDestinationLabel(route), result, tenantID)
|
||||
}
|
||||
|
||||
func (g *gateway) recordMediaDivergence(kind string, destination string) {
|
||||
|
|
@ -774,18 +784,111 @@ func (g *gateway) recordUploadPlanStoreError(operation string) {
|
|||
}
|
||||
}
|
||||
|
||||
func addTenantRouteAttributes(body []byte, contentType string, destination string) ([]byte, error) {
|
||||
func addTenantRouteAttributes(body []byte, contentType string, destination string) ([]byte, string, error) {
|
||||
if isJSONContentType(contentType) {
|
||||
return addJSONTenantRouteAttributes(body, destination)
|
||||
}
|
||||
return addProtobufTenantRouteAttributes(body, destination)
|
||||
}
|
||||
|
||||
func addProtobufTenantRouteAttributes(body []byte, destination string) ([]byte, error) {
|
||||
func extractTraceTenantID(body []byte, contentType string) string {
|
||||
if isJSONContentType(contentType) {
|
||||
return extractJSONTraceTenantID(body)
|
||||
}
|
||||
return extractProtobufTraceTenantID(body)
|
||||
}
|
||||
|
||||
func resolveTraceTenantID(tenantIDs map[string]struct{}) string {
|
||||
if len(tenantIDs) == 0 {
|
||||
return unknownTenantID
|
||||
}
|
||||
if len(tenantIDs) > 1 {
|
||||
return multipleTenantIDs
|
||||
}
|
||||
for tenantID := range tenantIDs {
|
||||
return tenantID
|
||||
}
|
||||
return unknownTenantID
|
||||
}
|
||||
|
||||
func extractProtobufTraceTenantID(body []byte) string {
|
||||
var request tracepb.ExportTraceServiceRequest
|
||||
if err := proto.Unmarshal(body, &request); err != nil {
|
||||
return nil, err
|
||||
return unknownTenantID
|
||||
}
|
||||
return protobufTraceTenantID(&request)
|
||||
}
|
||||
|
||||
func protobufTraceTenantID(request *tracepb.ExportTraceServiceRequest) string {
|
||||
tenantIDs := make(map[string]struct{})
|
||||
for _, resourceSpan := range request.ResourceSpans {
|
||||
for _, scopeSpan := range resourceSpan.ScopeSpans {
|
||||
for _, span := range scopeSpan.Spans {
|
||||
for _, attribute := range span.Attributes {
|
||||
if attribute.Key != tenantIDAttribute {
|
||||
continue
|
||||
}
|
||||
tenantID := strings.TrimSpace(attribute.Value.GetStringValue())
|
||||
if tenantID != "" {
|
||||
tenantIDs[tenantID] = struct{}{}
|
||||
if len(tenantIDs) > 1 {
|
||||
return multipleTenantIDs
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return resolveTraceTenantID(tenantIDs)
|
||||
}
|
||||
|
||||
func extractJSONTraceTenantID(body []byte) string {
|
||||
var request map[string]any
|
||||
if err := json.Unmarshal(body, &request); err != nil {
|
||||
return unknownTenantID
|
||||
}
|
||||
return jsonTraceTenantID(request)
|
||||
}
|
||||
|
||||
func jsonTraceTenantID(request map[string]any) string {
|
||||
tenantIDs := make(map[string]struct{})
|
||||
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)
|
||||
attributes, _ := spanMap["attributes"].([]any)
|
||||
for _, attribute := range attributes {
|
||||
attributeMap, _ := attribute.(map[string]any)
|
||||
if attributeMap["key"] != tenantIDAttribute {
|
||||
continue
|
||||
}
|
||||
value, _ := attributeMap["value"].(map[string]any)
|
||||
tenantID, _ := value["stringValue"].(string)
|
||||
tenantID = strings.TrimSpace(tenantID)
|
||||
if tenantID != "" {
|
||||
tenantIDs[tenantID] = struct{}{}
|
||||
if len(tenantIDs) > 1 {
|
||||
return multipleTenantIDs
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return resolveTraceTenantID(tenantIDs)
|
||||
}
|
||||
|
||||
func addProtobufTenantRouteAttributes(body []byte, destination string) ([]byte, string, error) {
|
||||
var request tracepb.ExportTraceServiceRequest
|
||||
if err := proto.Unmarshal(body, &request); err != nil {
|
||||
return nil, unknownTenantID, err
|
||||
}
|
||||
tenantID := protobufTraceTenantID(&request)
|
||||
|
||||
for _, resourceSpan := range request.ResourceSpans {
|
||||
for _, scopeSpan := range resourceSpan.ScopeSpans {
|
||||
|
|
@ -796,7 +899,8 @@ func addProtobufTenantRouteAttributes(body []byte, destination string) ([]byte,
|
|||
}
|
||||
}
|
||||
|
||||
return proto.Marshal(&request)
|
||||
updatedBody, err := proto.Marshal(&request)
|
||||
return updatedBody, tenantID, err
|
||||
}
|
||||
|
||||
func upsertSpanStringAttribute(span *tracev1.Span, key string, value string) {
|
||||
|
|
@ -818,11 +922,12 @@ func stringAnyValue(value string) *commonv1.AnyValue {
|
|||
}
|
||||
}
|
||||
|
||||
func addJSONTenantRouteAttributes(body []byte, destination string) ([]byte, error) {
|
||||
func addJSONTenantRouteAttributes(body []byte, destination string) ([]byte, string, error) {
|
||||
var request map[string]any
|
||||
if err := json.Unmarshal(body, &request); err != nil {
|
||||
return nil, err
|
||||
return nil, unknownTenantID, err
|
||||
}
|
||||
tenantID := jsonTraceTenantID(request)
|
||||
resourceSpans, _ := request["resourceSpans"].([]any)
|
||||
for _, resourceSpan := range resourceSpans {
|
||||
resourceSpanMap, _ := resourceSpan.(map[string]any)
|
||||
|
|
@ -837,7 +942,8 @@ func addJSONTenantRouteAttributes(body []byte, destination string) ([]byte, erro
|
|||
}
|
||||
}
|
||||
}
|
||||
return json.Marshal(request)
|
||||
updatedBody, err := json.Marshal(request)
|
||||
return updatedBody, tenantID, err
|
||||
}
|
||||
|
||||
func upsertJSONSpanStringAttribute(span map[string]any, key string, value string) {
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ import (
|
|||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"regexp"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
|
@ -1047,7 +1048,7 @@ func TestTraceProxyRecordsPrometheusMetrics(t *testing.T) {
|
|||
defer collector.Close()
|
||||
|
||||
gw := newTestGatewayWithCollector(collector.URL)
|
||||
body := buildTraceRequest(t, nil)
|
||||
body := buildTraceRequest(t, map[string]string{tenantIDAttribute: "tenant-123"})
|
||||
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")
|
||||
|
|
@ -1059,7 +1060,7 @@ func TestTraceProxyRecordsPrometheusMetrics(t *testing.T) {
|
|||
}
|
||||
|
||||
metrics := scrapeMetrics(t, gw)
|
||||
if !strings.Contains(metrics, `langfuse_fanout_trace_exports_total{destination="tenant_eu",result="success"} 1`) {
|
||||
if !strings.Contains(metrics, `langfuse_fanout_trace_exports_total{destination="tenant_eu",result="success",tenant_id="tenant-123"} 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`) {
|
||||
|
|
@ -1067,6 +1068,71 @@ func TestTraceProxyRecordsPrometheusMetrics(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestExtractTraceTenantID(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
t.Run("protobuf", func(t *testing.T) {
|
||||
body := buildTraceRequest(t, map[string]string{tenantIDAttribute: "tenant-123"})
|
||||
if tenantID := extractTraceTenantID(body, "application/x-protobuf"); tenantID != "tenant-123" {
|
||||
t.Fatalf("tenant ID = %q", tenantID)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("json", func(t *testing.T) {
|
||||
body := []byte(`{"resourceSpans":[{"scopeSpans":[{"spans":[{"attributes":[{"key":"librechat.tenant.id","value":{"stringValue":"tenant-456"}}]}]}]}]}`)
|
||||
if tenantID := extractTraceTenantID(body, "application/json"); tenantID != "tenant-456" {
|
||||
t.Fatalf("tenant ID = %q", tenantID)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("missing", func(t *testing.T) {
|
||||
body := buildTraceRequest(t, nil)
|
||||
if tenantID := extractTraceTenantID(body, "application/x-protobuf"); tenantID != unknownTenantID {
|
||||
t.Fatalf("tenant ID = %q", tenantID)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("multiple", func(t *testing.T) {
|
||||
tenantIDs := map[string]struct{}{"tenant-1": {}, "tenant-2": {}}
|
||||
if tenantID := resolveTraceTenantID(tenantIDs); tenantID != multipleTenantIDs {
|
||||
t.Fatalf("tenant ID = %q", tenantID)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("sentinels are outside tenant ID grammar", func(t *testing.T) {
|
||||
validTenantID := regexp.MustCompile(`^[-a-zA-Z0-9_.]+$`)
|
||||
for _, sentinel := range []string{unknownTenantID, multipleTenantIDs} {
|
||||
if validTenantID.MatchString(sentinel) {
|
||||
t.Fatalf("sentinel %q is a valid tenant ID", sentinel)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestMalformedTenantTraceRecordsErrorMetric(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
gw := newTestGatewayWithCollector("http://collector.invalid")
|
||||
req := httptest.NewRequest(
|
||||
http.MethodPost,
|
||||
tenantPrefix+"eu"+otelTracePath,
|
||||
bytes.NewReader([]byte("not protobuf")),
|
||||
)
|
||||
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.StatusBadRequest {
|
||||
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="error",tenant_id="<unknown>"} 1`) {
|
||||
t.Fatalf("missing invalid trace export metric:\n%s", metrics)
|
||||
}
|
||||
}
|
||||
|
||||
func newTestGateway(centralURL string, tenants map[string]string) *gateway {
|
||||
return newTestGatewayWithStore(centralURL, tenants, newFakeUploadPlanStore())
|
||||
}
|
||||
|
|
|
|||
|
|
@ -55,7 +55,7 @@ func newGatewayMetrics() *gatewayMetrics {
|
|||
traceExports: prometheus.NewCounterVec(prometheus.CounterOpts{
|
||||
Name: "langfuse_fanout_trace_exports_total",
|
||||
Help: "Total trace export attempts through the Langfuse fanout gateway.",
|
||||
}, []string{"destination", "result"}),
|
||||
}, []string{"destination", "result", "tenant_id"}),
|
||||
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.",
|
||||
|
|
@ -125,11 +125,19 @@ func (m *gatewayMetrics) recordUpstream(operation string, destination string, st
|
|||
m.upstreamDuration.With(labels).Observe(duration.Seconds())
|
||||
}
|
||||
|
||||
func (m *gatewayMetrics) recordTraceExport(destination string, result string) {
|
||||
func (m *gatewayMetrics) recordTraceExport(destination string, result string, tenantID string) {
|
||||
if m == nil {
|
||||
return
|
||||
}
|
||||
m.traceExports.WithLabelValues(normalizeMetricLabel(destination), result).Inc()
|
||||
m.traceExports.WithLabelValues(normalizeMetricLabel(destination), result, tenantMetricLabel(tenantID)).Inc()
|
||||
}
|
||||
|
||||
func tenantMetricLabel(tenantID string) string {
|
||||
tenantID = strings.TrimSpace(tenantID)
|
||||
if tenantID == "" {
|
||||
return "unknown"
|
||||
}
|
||||
return tenantID
|
||||
}
|
||||
|
||||
func (m *gatewayMetrics) recordMediaDivergence(kind string, destination string) {
|
||||
|
|
|
|||
|
|
@ -106,6 +106,7 @@ function createHandlers(overrides = {}) {
|
|||
),
|
||||
getMessages: jest.fn().mockResolvedValue([]),
|
||||
invalidateConfigCaches: jest.fn().mockResolvedValue(undefined),
|
||||
recordConnectionUpdate: jest.fn(),
|
||||
...overrides,
|
||||
};
|
||||
const handlers = createAdminLangfuseHandlers(deps);
|
||||
|
|
@ -529,6 +530,18 @@ describe('createAdminLangfuseHandlers', () => {
|
|||
expect(fields['langfuse.projectId']).toBe('project-1');
|
||||
expect(res.body?.secretKey).toBeUndefined();
|
||||
expect(deps.invalidateConfigCaches).toHaveBeenCalledWith('t1');
|
||||
expect(deps.recordConnectionUpdate).toHaveBeenCalledWith({
|
||||
event_name: 'librechat.langfuse.connection.changed',
|
||||
tenant_id: 't1',
|
||||
configured: true,
|
||||
enabled: true,
|
||||
destination: 'eu',
|
||||
change: 'created',
|
||||
changes: ['created'],
|
||||
verification_result: 'success',
|
||||
});
|
||||
expect(JSON.stringify(deps.recordConnectionUpdate.mock.calls)).not.toContain('sk-lf-secret');
|
||||
expect(JSON.stringify(deps.recordConnectionUpdate.mock.calls)).not.toContain('pk-lf-1');
|
||||
});
|
||||
|
||||
it('requires a new secret when connection fields change', async () => {
|
||||
|
|
@ -615,6 +628,7 @@ describe('createAdminLangfuseHandlers', () => {
|
|||
error: 'Langfuse rejected these keys. Check the destination and keys',
|
||||
});
|
||||
expect(deps.patchConfigFields).not.toHaveBeenCalled();
|
||||
expect(deps.recordConnectionUpdate).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('rejects credentials when Langfuse does not return a stable project identity', async () => {
|
||||
|
|
@ -667,6 +681,14 @@ describe('createAdminLangfuseHandlers', () => {
|
|||
expect(deps.patchConfigFields).toHaveBeenCalledTimes(1);
|
||||
expect(deps.patchConfigFields.mock.calls[0][3]['langfuse.enabled']).toBe(true);
|
||||
expect(deps.patchConfigFields.mock.calls[0][3]['langfuse.projectId']).toBe('project-1');
|
||||
expect(deps.recordConnectionUpdate).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
tenant_id: 't1',
|
||||
change: 'enabled',
|
||||
changes: ['enabled'],
|
||||
verification_result: 'skipped',
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it('allows an existing connection to be disabled after its destination is removed', async () => {
|
||||
|
|
|
|||
|
|
@ -28,6 +28,26 @@ const DEFAULT_PRIORITY = 10;
|
|||
const ENCRYPTED_PREFIX = 'v3:';
|
||||
const LANGFUSE_VERIFICATION_TIMEOUT_MS = 10_000;
|
||||
|
||||
type LangfuseConnectionChange =
|
||||
| 'created'
|
||||
| 'credentials_rotated'
|
||||
| 'destination_changed'
|
||||
| 'disabled'
|
||||
| 'enabled'
|
||||
| 'updated';
|
||||
type LangfuseConnectionChanges = [LangfuseConnectionChange, ...LangfuseConnectionChange[]];
|
||||
|
||||
export interface LangfuseConnectionEvent {
|
||||
event_name: 'librechat.langfuse.connection.changed';
|
||||
tenant_id?: string;
|
||||
configured: boolean;
|
||||
enabled: boolean;
|
||||
destination?: string;
|
||||
change: LangfuseConnectionChange;
|
||||
changes: LangfuseConnectionChange[];
|
||||
verification_result: 'skipped' | 'success';
|
||||
}
|
||||
|
||||
export interface AdminLangfuseDeps {
|
||||
findConfigByPrincipal: (
|
||||
principalType: PrincipalType,
|
||||
|
|
@ -51,6 +71,7 @@ export interface AdminLangfuseDeps {
|
|||
) => Promise<IConfig | null>;
|
||||
getMessages: MessageMethods['getMessages'];
|
||||
invalidateConfigCaches?: (tenantId?: string) => Promise<void>;
|
||||
recordConnectionUpdate?: (event: LangfuseConnectionEvent) => void;
|
||||
}
|
||||
|
||||
function getTenantId(req: ServerRequest): string | undefined {
|
||||
|
|
@ -79,6 +100,33 @@ function buildStatus(config: IConfig | null): TLangfuseConnectionStatus {
|
|||
};
|
||||
}
|
||||
|
||||
function getConnectionChanges(
|
||||
stored: TCustomConfig['langfuse'],
|
||||
enabled: boolean,
|
||||
destination: string,
|
||||
publicKey: string,
|
||||
secretKey: string,
|
||||
): LangfuseConnectionChanges {
|
||||
if (!stored?.publicKey || !stored.secretKey) {
|
||||
return ['created'];
|
||||
}
|
||||
const changes: LangfuseConnectionChange[] = [];
|
||||
if (stored.destination !== destination) {
|
||||
changes.push('destination_changed');
|
||||
}
|
||||
if (stored.publicKey !== publicKey || secretKey !== '') {
|
||||
changes.push('credentials_rotated');
|
||||
}
|
||||
if (stored.enabled !== true && enabled) {
|
||||
changes.push('enabled');
|
||||
}
|
||||
if (stored.enabled === true && !enabled) {
|
||||
changes.push('disabled');
|
||||
}
|
||||
const [change, ...additionalChanges] = changes;
|
||||
return change ? [change, ...additionalChanges] : ['updated'];
|
||||
}
|
||||
|
||||
function rejectWhenConnectionUnavailable(res: Response): Response | undefined {
|
||||
if (isLangfuseConnectionAvailable()) {
|
||||
return undefined;
|
||||
|
|
@ -243,6 +291,8 @@ export function createAdminLangfuseHandlers(deps: AdminLangfuseDeps): {
|
|||
toggleConfigActive,
|
||||
getMessages,
|
||||
invalidateConfigCaches,
|
||||
recordConnectionUpdate = (event) =>
|
||||
logger.info({ message: '[adminLangfuse] Connection updated', ...event }),
|
||||
} = deps;
|
||||
|
||||
function findBaseConfig(options?: { includeInactive?: boolean }): Promise<IConfig | null> {
|
||||
|
|
@ -415,11 +465,30 @@ export function createAdminLangfuseHandlers(deps: AdminLangfuseDeps): {
|
|||
updated = await toggleConfigActive(PrincipalType.ROLE, BASE_CONFIG_PRINCIPAL_ID, true);
|
||||
}
|
||||
|
||||
const status = buildStatus(updated ?? existing);
|
||||
const changes = getConnectionChanges(
|
||||
stored,
|
||||
enabled,
|
||||
persistedDestination,
|
||||
publicKey,
|
||||
secretKey,
|
||||
);
|
||||
recordConnectionUpdate({
|
||||
event_name: 'librechat.langfuse.connection.changed',
|
||||
tenant_id: getTenantId(req),
|
||||
configured: status.configured,
|
||||
enabled: status.enabled,
|
||||
destination: status.destination,
|
||||
change: changes[0],
|
||||
changes,
|
||||
verification_result: connectionChanged ? 'success' : 'skipped',
|
||||
});
|
||||
|
||||
invalidateConfigCaches?.(getTenantId(req))?.catch((err) =>
|
||||
logger.error('[adminLangfuse] Cache invalidation failed after update:', err),
|
||||
);
|
||||
|
||||
return res.status(200).json(buildStatus(updated ?? existing));
|
||||
return res.status(200).json(status);
|
||||
} catch (error) {
|
||||
logger.error('[adminLangfuse] updateConnection error:', error);
|
||||
return res.status(500).json({ error: 'Failed to update Langfuse connection' });
|
||||
|
|
|
|||
|
|
@ -2096,10 +2096,19 @@ async function callAndCaptureRunConfig({
|
|||
// ---------------------------------------------------------------------------
|
||||
// Suite: Langfuse run config
|
||||
// ---------------------------------------------------------------------------
|
||||
const exportTelemetry = (plan: string, reason: string, tenantId?: string) => ({
|
||||
...(tenantId ? { 'librechat.tenant.id': tenantId } : {}),
|
||||
'librechat.langfuse.export_plan': plan,
|
||||
'librechat.langfuse.export_reason': reason,
|
||||
});
|
||||
|
||||
describe('Langfuse run config', () => {
|
||||
it('passes deterministic Langfuse trace config without tenant metadata by default', async () => {
|
||||
const callArgs = await callAndCaptureRunConfig();
|
||||
expect(callArgs.langfuse).toEqual({ deterministicTraceId: true });
|
||||
expect(callArgs.langfuse).toEqual({
|
||||
deterministicTraceId: true,
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'fanout_disabled'),
|
||||
});
|
||||
});
|
||||
|
||||
it('adds the explicit request tenant id to Langfuse trace metadata and tags', async () => {
|
||||
|
|
@ -2111,6 +2120,7 @@ describe('Langfuse run config', () => {
|
|||
});
|
||||
expect(callArgs.langfuse).toEqual({
|
||||
deterministicTraceId: true,
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'fanout_disabled', 'tenant-1'),
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
@ -2124,6 +2134,7 @@ describe('Langfuse run config', () => {
|
|||
});
|
||||
expect(callArgs.langfuse).toEqual({
|
||||
deterministicTraceId: true,
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'fanout_disabled', 'tenant-2'),
|
||||
metadata: { 'librechat.tenant.id': 'tenant-2' },
|
||||
tags: ['tenant:tenant-2'],
|
||||
});
|
||||
|
|
@ -2152,6 +2163,7 @@ describe('Langfuse run config', () => {
|
|||
baseUrl: 'http://langfuse-fanout-collector:4318/tenant/eu',
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
librechatTraceAttributes: {
|
||||
...exportTelemetry('tenant_fanout', 'configured', 'tenant-1'),
|
||||
'librechat.langfuse.tenant_export.enabled': 'true',
|
||||
'librechat.langfuse.destination': 'eu',
|
||||
},
|
||||
|
|
@ -2181,6 +2193,7 @@ describe('Langfuse run config', () => {
|
|||
publicKey: 'pk-central',
|
||||
secretKey: 'sk-central',
|
||||
baseUrl: 'https://central.langfuse.example',
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'fanout_disabled', 'tenant-1'),
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
@ -2207,6 +2220,11 @@ describe('Langfuse run config', () => {
|
|||
expect(callArgs.langfuse).toEqual({
|
||||
deterministicTraceId: true,
|
||||
baseUrl: 'http://collector-from-env:4318',
|
||||
librechatTraceAttributes: exportTelemetry(
|
||||
'central_only',
|
||||
'destination_unconfigured',
|
||||
'tenant-1',
|
||||
),
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
@ -2234,6 +2252,7 @@ describe('Langfuse run config', () => {
|
|||
baseUrl: 'http://collector-from-env:4318/tenant/us',
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
librechatTraceAttributes: {
|
||||
...exportTelemetry('tenant_fanout', 'configured', 'tenant-1'),
|
||||
'librechat.langfuse.tenant_export.enabled': 'true',
|
||||
'librechat.langfuse.destination': 'us',
|
||||
},
|
||||
|
|
@ -2284,6 +2303,7 @@ describe('Langfuse run config', () => {
|
|||
secretKey: 'sk-tenant-1',
|
||||
baseUrl: 'http://collector-from-env:4318/tenant/us',
|
||||
librechatTraceAttributes: {
|
||||
...exportTelemetry('tenant_fanout', 'configured', 'tenant-1'),
|
||||
'librechat.langfuse.tenant_export.enabled': 'true',
|
||||
'librechat.langfuse.destination': 'us',
|
||||
},
|
||||
|
|
@ -2317,6 +2337,7 @@ describe('Langfuse run config', () => {
|
|||
publicKey: 'pk-central',
|
||||
secretKey: 'sk-central',
|
||||
baseUrl: 'https://central.langfuse.example',
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'fanout_disabled', 'tenant-1'),
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
@ -2346,10 +2367,10 @@ describe('Langfuse run config', () => {
|
|||
publicKey: 'pk-central',
|
||||
secretKey: 'sk-central',
|
||||
baseUrl: 'https://central.langfuse.example',
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'fanout_disabled', 'tenant-1'),
|
||||
});
|
||||
expect(callArgs.langfuse).not.toMatchObject({
|
||||
baseUrl: 'http://collector-from-env:4318/tenant/eu',
|
||||
librechatTraceAttributes: expect.any(Object),
|
||||
});
|
||||
});
|
||||
|
||||
|
|
@ -2376,6 +2397,11 @@ describe('Langfuse run config', () => {
|
|||
publicKey: 'pk-central',
|
||||
secretKey: 'sk-central',
|
||||
baseUrl: 'https://central.langfuse.example',
|
||||
librechatTraceAttributes: exportTelemetry(
|
||||
'central_only',
|
||||
'collector_unconfigured',
|
||||
'tenant-1',
|
||||
),
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
@ -2404,6 +2430,11 @@ describe('Langfuse run config', () => {
|
|||
expect(callArgs.langfuse).toEqual({
|
||||
deterministicTraceId: true,
|
||||
baseUrl: 'http://collector-from-env:4318',
|
||||
librechatTraceAttributes: exportTelemetry(
|
||||
'central_only',
|
||||
'destination_unconfigured',
|
||||
'tenant-1',
|
||||
),
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
@ -2426,6 +2457,7 @@ describe('Langfuse run config', () => {
|
|||
expect(callArgs.langfuse).toEqual({
|
||||
deterministicTraceId: true,
|
||||
baseUrl: 'http://collector-from-env:4318',
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'missing_credentials', 'tenant-1'),
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
@ -2445,6 +2477,7 @@ describe('Langfuse run config', () => {
|
|||
expect(callArgs.langfuse).toEqual({
|
||||
deterministicTraceId: true,
|
||||
baseUrl: 'http://collector-from-env:4318',
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'tenant_disabled', 'tenant-1'),
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
@ -2472,6 +2505,7 @@ describe('Langfuse run config', () => {
|
|||
expect(callArgs.langfuse).toEqual({
|
||||
deterministicTraceId: true,
|
||||
baseUrl: 'http://collector-from-env:4318',
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'emergency_disabled', 'tenant-1'),
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
@ -2504,6 +2538,7 @@ describe('Langfuse run config', () => {
|
|||
secretKey: 'sk-tenant-1',
|
||||
tags: ['tenant:tenant-1'],
|
||||
librechatTraceAttributes: {
|
||||
...exportTelemetry('tenant_fanout', 'configured', 'tenant-1'),
|
||||
'librechat.langfuse.tenant_export.enabled': 'true',
|
||||
'librechat.langfuse.destination': 'eu',
|
||||
},
|
||||
|
|
@ -2534,6 +2569,7 @@ describe('Langfuse run config', () => {
|
|||
expect(callArgs.langfuse).toEqual({
|
||||
deterministicTraceId: true,
|
||||
baseUrl: 'http://collector-from-env:4318',
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'emergency_disabled', 'tenant-1'),
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
@ -2569,6 +2605,7 @@ describe('Langfuse run config', () => {
|
|||
secretKey: 'sk-tenant-1',
|
||||
tags: ['tenant:tenant-1'],
|
||||
librechatTraceAttributes: {
|
||||
...exportTelemetry('tenant_fanout', 'configured', 'tenant-1'),
|
||||
'librechat.langfuse.tenant_export.enabled': 'true',
|
||||
'librechat.langfuse.destination': 'eu',
|
||||
},
|
||||
|
|
@ -2594,6 +2631,7 @@ describe('Langfuse run config', () => {
|
|||
expect(callArgs.langfuse).toEqual({
|
||||
deterministicTraceId: true,
|
||||
baseUrl: 'http://collector-from-env:4318',
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'tenant_disabled', 'tenant-1'),
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
@ -2617,6 +2655,7 @@ describe('Langfuse run config', () => {
|
|||
expect(callArgs.langfuse).toEqual({
|
||||
deterministicTraceId: true,
|
||||
baseUrl: 'http://collector-from-env:4318',
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'tenant_disabled', 'tenant-1'),
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
|
|||
|
|
@ -4,6 +4,11 @@ process.env.CREDS_KEY =
|
|||
process.env.CREDS_KEY ?? '0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef';
|
||||
|
||||
const CENTRAL_EXPORT_ATTRIBUTE = 'librechat.langfuse.central_export.enabled';
|
||||
const exportTelemetry = (plan: string, reason: string) => ({
|
||||
'librechat.langfuse.export_plan': plan,
|
||||
'librechat.langfuse.export_reason': reason,
|
||||
'librechat.tenant.id': 'tenant-1',
|
||||
});
|
||||
const envKeys = [
|
||||
'LANGFUSE_PUBLIC_KEY',
|
||||
'LANGFUSE_SECRET_KEY',
|
||||
|
|
@ -181,6 +186,7 @@ describe('buildLangfuseConfig', () => {
|
|||
secretKey: 'sk-tenant-1',
|
||||
baseUrl: 'http://langfuse-fanout-collector:4318/tenant/eu',
|
||||
librechatTraceAttributes: {
|
||||
...exportTelemetry('tenant_fanout', 'configured'),
|
||||
'librechat.langfuse.tenant_export.enabled': 'true',
|
||||
'librechat.langfuse.destination': 'eu',
|
||||
},
|
||||
|
|
@ -208,6 +214,7 @@ describe('buildLangfuseConfig', () => {
|
|||
expect(config).toEqual({
|
||||
deterministicTraceId: true,
|
||||
baseUrl: 'http://langfuse-fanout-collector:4318',
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'missing_credentials'),
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
@ -233,6 +240,7 @@ describe('buildLangfuseConfig', () => {
|
|||
expect(config).toEqual({
|
||||
deterministicTraceId: true,
|
||||
baseUrl: 'http://langfuse-fanout-collector:4318',
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'missing_credentials'),
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
@ -249,11 +257,49 @@ describe('buildLangfuseConfig', () => {
|
|||
publicKey: 'pk-central',
|
||||
secretKey: 'sk-central',
|
||||
baseUrl: 'https://central.langfuse.example',
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'fanout_disabled'),
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
});
|
||||
|
||||
it('records when fanout is enabled without a collector URL', async () => {
|
||||
process.env.LANGFUSE_FANOUT_ENABLED = 'true';
|
||||
process.env.LANGFUSE_PUBLIC_KEY = 'pk-central';
|
||||
process.env.LANGFUSE_SECRET_KEY = 'sk-central';
|
||||
const { buildLangfuseConfig } = await import('./config');
|
||||
|
||||
expect(buildLangfuseConfig({ tenantId: 'tenant-1' })).toMatchObject({
|
||||
publicKey: 'pk-central',
|
||||
secretKey: 'sk-central',
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'collector_unconfigured'),
|
||||
});
|
||||
});
|
||||
|
||||
it('records when an enabled tenant connection has an unknown destination', async () => {
|
||||
process.env.LANGFUSE_FANOUT_ENABLED = 'true';
|
||||
process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://collector-from-env:4318';
|
||||
const { encryptV3 } = await import('@librechat/data-schemas');
|
||||
const { buildLangfuseConfig } = await import('./config');
|
||||
|
||||
expect(
|
||||
buildLangfuseConfig({
|
||||
tenantId: 'tenant-1',
|
||||
appConfig: {
|
||||
langfuse: {
|
||||
enabled: true,
|
||||
publicKey: 'pk-tenant-1',
|
||||
secretKey: encryptV3('sk-tenant-1'),
|
||||
destination: 'unknown-destination',
|
||||
},
|
||||
} as unknown as AppConfig,
|
||||
}),
|
||||
).toMatchObject({
|
||||
baseUrl: 'http://collector-from-env:4318',
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'destination_unconfigured'),
|
||||
});
|
||||
});
|
||||
|
||||
it('disables direct central tracing when central export is disabled', async () => {
|
||||
process.env.LANGFUSE_PUBLIC_KEY = 'pk-central';
|
||||
process.env.LANGFUSE_SECRET_KEY = 'sk-central';
|
||||
|
|
@ -271,6 +317,7 @@ describe('buildLangfuseConfig', () => {
|
|||
enabled: false,
|
||||
librechatTraceAttributes: {
|
||||
[CENTRAL_EXPORT_ATTRIBUTE]: 'false',
|
||||
...exportTelemetry('disabled', 'fanout_disabled'),
|
||||
},
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
@ -294,6 +341,7 @@ describe('buildLangfuseConfig', () => {
|
|||
enabled: false,
|
||||
librechatTraceAttributes: {
|
||||
[CENTRAL_EXPORT_ATTRIBUTE]: 'false',
|
||||
...exportTelemetry('disabled', 'tenant_disabled'),
|
||||
},
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
@ -326,6 +374,7 @@ describe('buildLangfuseConfig', () => {
|
|||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
librechatTraceAttributes: {
|
||||
[CENTRAL_EXPORT_ATTRIBUTE]: 'false',
|
||||
...exportTelemetry('tenant_fanout', 'configured'),
|
||||
'librechat.langfuse.tenant_export.enabled': 'true',
|
||||
'librechat.langfuse.destination': 'us',
|
||||
},
|
||||
|
|
@ -359,6 +408,7 @@ describe('buildLangfuseConfig', () => {
|
|||
enabled: false,
|
||||
librechatTraceAttributes: {
|
||||
[CENTRAL_EXPORT_ATTRIBUTE]: 'false',
|
||||
...exportTelemetry('disabled', 'emergency_disabled'),
|
||||
},
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
@ -385,6 +435,7 @@ describe('buildLangfuseConfig', () => {
|
|||
).toEqual({
|
||||
deterministicTraceId: true,
|
||||
baseUrl: 'http://collector-from-env:4318',
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'tenant_disabled'),
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
@ -410,6 +461,7 @@ describe('buildLangfuseConfig', () => {
|
|||
).toEqual({
|
||||
deterministicTraceId: true,
|
||||
baseUrl: 'http://collector-from-env:4318',
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'tenant_disabled'),
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
@ -436,6 +488,7 @@ describe('buildLangfuseConfig', () => {
|
|||
enabled: false,
|
||||
librechatTraceAttributes: {
|
||||
[CENTRAL_EXPORT_ATTRIBUTE]: 'false',
|
||||
...exportTelemetry('disabled', 'tenant_disabled'),
|
||||
},
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
@ -453,6 +506,7 @@ describe('buildLangfuseConfig', () => {
|
|||
deterministicTraceId: true,
|
||||
baseUrl: 'http://collector-from-env:4318',
|
||||
mediaUploadEnabled: false,
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'tenant_disabled'),
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
@ -470,6 +524,7 @@ describe('buildLangfuseConfig', () => {
|
|||
expect(buildLangfuseConfig({ tenantId: 'tenant-1' })).toEqual({
|
||||
deterministicTraceId: true,
|
||||
baseUrl: 'http://collector-from-env:4318',
|
||||
librechatTraceAttributes: exportTelemetry('central_only', 'tenant_disabled'),
|
||||
metadata: { 'librechat.tenant.id': 'tenant-1' },
|
||||
tags: ['tenant:tenant-1'],
|
||||
});
|
||||
|
|
|
|||
|
|
@ -22,10 +22,17 @@ type LangfuseRunConfigWithTraceAttributes = LangfuseRunConfig & {
|
|||
additionalHeaders?: Record<string, string>;
|
||||
};
|
||||
type LangfuseTenantDestination = NonNullable<ReturnType<typeof resolveLangfuseTenantDestination>>;
|
||||
type TenantExportBlockReason =
|
||||
| 'collector_unconfigured'
|
||||
| 'destination_unconfigured'
|
||||
| 'emergency_disabled'
|
||||
| 'fanout_disabled'
|
||||
| 'missing_credentials'
|
||||
| 'tenant_disabled';
|
||||
type LangfuseExportPlan =
|
||||
| { type: 'directCentral' }
|
||||
| { type: 'disabled' }
|
||||
| { type: 'fanoutCollector'; collectorUrl: string }
|
||||
| { type: 'directCentral'; reason: 'collector_unconfigured' | 'fanout_disabled' }
|
||||
| { type: 'disabled'; reason: TenantExportBlockReason }
|
||||
| { type: 'fanoutCollector'; collectorUrl: string; reason: TenantExportBlockReason }
|
||||
| {
|
||||
type: 'tenantFanout';
|
||||
collectorUrl: string;
|
||||
|
|
@ -36,6 +43,9 @@ type LangfuseExportPlan =
|
|||
const TENANT_EXPORT_ATTRIBUTE = 'librechat.langfuse.tenant_export.enabled';
|
||||
const TENANT_DESTINATION_ATTRIBUTE = 'librechat.langfuse.destination';
|
||||
const CENTRAL_EXPORT_ATTRIBUTE = 'librechat.langfuse.central_export.enabled';
|
||||
const EXPORT_PLAN_ATTRIBUTE = 'librechat.langfuse.export_plan';
|
||||
const EXPORT_REASON_ATTRIBUTE = 'librechat.langfuse.export_reason';
|
||||
const TENANT_ID_ATTRIBUTE = 'librechat.tenant.id';
|
||||
const CENTRAL_MEDIA_DISABLED_SEGMENT = 'central-media-disabled';
|
||||
const DEFAULT_BASE_URL = 'https://cloud.langfuse.com';
|
||||
|
||||
|
|
@ -108,29 +118,94 @@ function disableCentralExport(langfuse: LangfuseRunConfigWithTraceAttributes): v
|
|||
};
|
||||
}
|
||||
|
||||
function getTenantExportBlockReason({
|
||||
tenantLangfuseEnabled,
|
||||
hasTenantCredentials,
|
||||
tenantExportEmergencyEnabled,
|
||||
tenantDestination,
|
||||
}: {
|
||||
tenantLangfuseEnabled: boolean;
|
||||
hasTenantCredentials: boolean;
|
||||
tenantExportEmergencyEnabled: boolean;
|
||||
tenantDestination?: LangfuseTenantDestination;
|
||||
}): TenantExportBlockReason {
|
||||
if (!tenantLangfuseEnabled) {
|
||||
return 'tenant_disabled';
|
||||
}
|
||||
if (!hasTenantCredentials) {
|
||||
return 'missing_credentials';
|
||||
}
|
||||
if (!tenantExportEmergencyEnabled) {
|
||||
return 'emergency_disabled';
|
||||
}
|
||||
if (tenantDestination == null) {
|
||||
return 'destination_unconfigured';
|
||||
}
|
||||
return 'missing_credentials';
|
||||
}
|
||||
|
||||
function applyExportPlanTelemetry(
|
||||
langfuse: LangfuseRunConfigWithTraceAttributes,
|
||||
exportPlan: LangfuseExportPlan,
|
||||
tenantId?: string,
|
||||
): void {
|
||||
let exportPlanName = 'central_only';
|
||||
if (exportPlan.type === 'tenantFanout') {
|
||||
exportPlanName = 'tenant_fanout';
|
||||
} else if (exportPlan.type === 'disabled') {
|
||||
exportPlanName = 'disabled';
|
||||
}
|
||||
const exportReason = exportPlan.type === 'tenantFanout' ? 'configured' : exportPlan.reason;
|
||||
|
||||
langfuse.librechatTraceAttributes = {
|
||||
...(langfuse.librechatTraceAttributes ?? {}),
|
||||
...(tenantId ? { [TENANT_ID_ATTRIBUTE]: tenantId } : {}),
|
||||
[EXPORT_PLAN_ATTRIBUTE]: exportPlanName,
|
||||
[EXPORT_REASON_ATTRIBUTE]: exportReason,
|
||||
};
|
||||
}
|
||||
|
||||
function resolveLangfuseExportPlan({
|
||||
centralTraceExportEnabled,
|
||||
fanoutEnabled,
|
||||
fanoutRequested,
|
||||
fanoutCollectorUrl,
|
||||
tenantExportEnabled,
|
||||
tenantLangfuseEnabled,
|
||||
hasTenantCredentials,
|
||||
tenantExportEmergencyEnabled,
|
||||
publicKey,
|
||||
secretKey,
|
||||
tenantDestination,
|
||||
}: {
|
||||
centralTraceExportEnabled: boolean;
|
||||
fanoutEnabled: boolean;
|
||||
fanoutRequested: boolean;
|
||||
fanoutCollectorUrl?: string;
|
||||
tenantExportEnabled: boolean;
|
||||
tenantLangfuseEnabled: boolean;
|
||||
hasTenantCredentials: boolean;
|
||||
tenantExportEmergencyEnabled: boolean;
|
||||
publicKey?: string;
|
||||
secretKey?: string;
|
||||
tenantDestination?: LangfuseTenantDestination;
|
||||
}): LangfuseExportPlan {
|
||||
if (!fanoutEnabled || fanoutCollectorUrl == null) {
|
||||
return centralTraceExportEnabled ? { type: 'directCentral' } : { type: 'disabled' };
|
||||
const reason = fanoutRequested ? 'collector_unconfigured' : 'fanout_disabled';
|
||||
if (centralTraceExportEnabled) {
|
||||
return {
|
||||
type: 'directCentral',
|
||||
reason,
|
||||
};
|
||||
}
|
||||
return { type: 'disabled', reason };
|
||||
}
|
||||
|
||||
const canRouteTenantFanout =
|
||||
tenantExportEnabled && publicKey != null && secretKey != null && tenantDestination != null;
|
||||
tenantLangfuseEnabled &&
|
||||
hasTenantCredentials &&
|
||||
tenantExportEmergencyEnabled &&
|
||||
publicKey != null &&
|
||||
secretKey != null &&
|
||||
tenantDestination != null;
|
||||
|
||||
if (canRouteTenantFanout) {
|
||||
return {
|
||||
|
|
@ -145,10 +220,27 @@ function resolveLangfuseExportPlan({
|
|||
// Direct central export can use the collector normally. Central-suppressed
|
||||
// runs only reach the collector through a concrete tenant fanout route.
|
||||
if (centralTraceExportEnabled) {
|
||||
return { type: 'fanoutCollector', collectorUrl: fanoutCollectorUrl };
|
||||
return {
|
||||
type: 'fanoutCollector',
|
||||
collectorUrl: fanoutCollectorUrl,
|
||||
reason: getTenantExportBlockReason({
|
||||
tenantLangfuseEnabled,
|
||||
hasTenantCredentials,
|
||||
tenantExportEmergencyEnabled,
|
||||
tenantDestination,
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
return { type: 'disabled' };
|
||||
return {
|
||||
type: 'disabled',
|
||||
reason: getTenantExportBlockReason({
|
||||
tenantLangfuseEnabled,
|
||||
hasTenantCredentials,
|
||||
tenantExportEmergencyEnabled,
|
||||
tenantDestination,
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
export function buildLangfuseConfig({
|
||||
|
|
@ -200,6 +292,7 @@ export function buildLangfuseConfig({
|
|||
const tenantCredentials = resolveTenantCredentials(config);
|
||||
const hasTenantCredentials = Boolean(tenantCredentials);
|
||||
const fanoutEnabled = isLangfuseFanoutEnabled();
|
||||
const fanoutRequested = normalizeBoolean(process.env.LANGFUSE_FANOUT_ENABLED) === true;
|
||||
const fanoutCollectorUrl = normalizeString(process.env.LANGFUSE_FANOUT_COLLECTOR_URL);
|
||||
const tenantDestination = resolveLangfuseTenantDestination(config?.destination);
|
||||
const tenantExportEmergencyEnabled = isLangfuseTenantExportEnabled();
|
||||
|
|
@ -222,13 +315,16 @@ export function buildLangfuseConfig({
|
|||
const exportPlan = resolveLangfuseExportPlan({
|
||||
centralTraceExportEnabled,
|
||||
fanoutEnabled,
|
||||
fanoutRequested,
|
||||
fanoutCollectorUrl,
|
||||
tenantExportEnabled:
|
||||
tenantLangfuseEnabled && hasTenantCredentials && tenantExportEmergencyEnabled,
|
||||
tenantLangfuseEnabled,
|
||||
hasTenantCredentials,
|
||||
tenantExportEmergencyEnabled,
|
||||
publicKey: tenantCredentials?.publicKey,
|
||||
secretKey: tenantCredentials?.secretKey,
|
||||
tenantDestination,
|
||||
});
|
||||
applyExportPlanTelemetry(langfuse, exportPlan, normalizedTenantId);
|
||||
|
||||
switch (exportPlan.type) {
|
||||
case 'tenantFanout':
|
||||
|
|
|
|||
|
|
@ -64,6 +64,32 @@ describe('attachRequestContext', () => {
|
|||
});
|
||||
}));
|
||||
|
||||
it('renders tenant Langfuse connection events with identity and outcome fields', () =>
|
||||
tenantStorage.run(context, () => {
|
||||
const result = attachRequestContext({
|
||||
level: 'info',
|
||||
message: '[adminLangfuse] Connection updated',
|
||||
event_name: 'librechat.langfuse.connection.changed',
|
||||
tenant_id: 'tenant-123',
|
||||
configured: true,
|
||||
enabled: true,
|
||||
destination: 'eu',
|
||||
change: 'created',
|
||||
changes: ['created'],
|
||||
verification_result: 'success',
|
||||
});
|
||||
|
||||
const rendered = formatLogContext(result);
|
||||
expect(rendered).toContain('"event_name":"librechat.langfuse.connection.changed"');
|
||||
expect(rendered).toContain('"tenant_id":"tenant-123"');
|
||||
expect(rendered).toContain('"configured":true');
|
||||
expect(rendered).toContain('"enabled":true');
|
||||
expect(rendered).toContain('"destination":"eu"');
|
||||
expect(rendered).toContain('"change":"created"');
|
||||
expect(rendered).toContain('"changes":["created"]');
|
||||
expect(rendered).toContain('"verification_result":"success"');
|
||||
}));
|
||||
|
||||
it('keeps application paths separate from the safe request route', () =>
|
||||
tenantStorage.run(context, () => {
|
||||
const result = attachRequestContext({
|
||||
|
|
|
|||
|
|
@ -26,6 +26,13 @@ const RESERVED_REQUEST_LOG_CONTEXT_KEYS = new Set<string>([
|
|||
|
||||
const STRUCTURED_EVENT_LOG_CONTEXT_KEYS = [
|
||||
'event_name',
|
||||
'tenant_id',
|
||||
'configured',
|
||||
'enabled',
|
||||
'destination',
|
||||
'change',
|
||||
'changes',
|
||||
'verification_result',
|
||||
'auth_strategy',
|
||||
'primary_strategy',
|
||||
'fallback_strategy',
|
||||
|
|
@ -63,6 +70,10 @@ const IDENTITY_FREE_EVENT_NAMES = new Set([
|
|||
'jwt_auth_recovered',
|
||||
'tenant_isolation_error',
|
||||
]);
|
||||
const STRUCTURED_EVENT_NAMES = new Set([
|
||||
...IDENTITY_FREE_EVENT_NAMES,
|
||||
'librechat.langfuse.connection.changed',
|
||||
]);
|
||||
const STRUCTURED_EVENT_LOG_CONTEXT_KEY_SET = new Set<string>(STRUCTURED_EVENT_LOG_CONTEXT_KEYS);
|
||||
const MAX_LOG_CONTEXT_ARRAY_LENGTH = 10;
|
||||
|
||||
|
|
@ -75,6 +86,10 @@ function isIdentityFreeEvent(eventName: unknown): boolean {
|
|||
return typeof eventName === 'string' && IDENTITY_FREE_EVENT_NAMES.has(eventName);
|
||||
}
|
||||
|
||||
function isStructuredEvent(eventName: unknown): boolean {
|
||||
return typeof eventName === 'string' && STRUCTURED_EVENT_NAMES.has(eventName);
|
||||
}
|
||||
|
||||
function getLogTenantId(): string | undefined {
|
||||
const tenantId = getTenantId();
|
||||
return tenantId === SYSTEM_TENANT_ID ? undefined : tenantId;
|
||||
|
|
@ -104,9 +119,10 @@ function normalizeLogContextValue(value: unknown): LogContextValue | undefined {
|
|||
export function formatLogContext(info: LogContextInfo): string {
|
||||
const context: Partial<Record<LogContextKey, LogContextValue>> = {};
|
||||
const omitIdentity = isIdentityFreeEvent(info.event_name);
|
||||
const includeStructuredEvent = isStructuredEvent(info.event_name);
|
||||
|
||||
LOG_CONTEXT_KEYS.forEach((key) => {
|
||||
if (STRUCTURED_EVENT_LOG_CONTEXT_KEY_SET.has(key) && !omitIdentity) {
|
||||
if (STRUCTURED_EVENT_LOG_CONTEXT_KEY_SET.has(key) && !includeStructuredEvent) {
|
||||
return;
|
||||
}
|
||||
if (omitIdentity && (key === 'tenantId' || key === 'userId')) {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue