Merge branch 'main' into fix-Display-all-AI-SDK-LLM-message

Author: sonianuj287

.github/workflows/lib-integration-tests-runner.yml

@@ -15,6 +15,7 @@ on:
           - all
           - openai
           - langchain
+          - langchain_legacy
           - llama_index
           - anthropic
           - aisuite
@@ -70,6 +71,12 @@ jobs:
     uses: ./.github/workflows/lib-langchain-tests.yml
     secrets: inherit
 
+  langchain_legacy_tests:
+    needs: [init_environment]
+    if: contains(fromJSON('["langchain_legacy", "all"]'), needs.init_environment.outputs.LIBS)
+    uses: ./.github/workflows/lib-langchain-legacy-tests.yml
+    secrets: inherit
+  
   llama_index_tests:
     needs: [init_environment]
     if: contains(fromJSON('["llama_index", "all"]'), needs.init_environment.outputs.LIBS)

.github/workflows/lib-langchain-legacy-tests.yml

@@ -0,0 +1,75 @@
+# Workflow to run Langchain tests
+#
+# Please read inputs to provide correct values.
+#
+name: SDK Lib Langchain Tests for langchain < 1.0.0
+run-name: "SDK Lib Langchain Tests ${{ github.ref_name }} by @${{ github.actor }}"
+permissions:
+  contents: read
+env:
+  OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
+  OPENAI_ORG_ID:  ${{ secrets.OPENAI_ORG_ID }}
+  GCP_CREDENTIALS_JSON: ${{ secrets.GCP_CREDENTIALS_JSON }}
+  GOOGLE_CLOUD_LOCATION: us-east1
+  GOOGLE_CLOUD_PROJECT: opik-sdk-tests
+  GOOGLE_API_KEY: ${{ secrets.GOOGLE_API_KEY }}
+  ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
+  AWS_DEFAULT_REGION: us-east-1
+  AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
+  AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
+  OPIK_ENABLE_LITELLM_MODELS_MONITORING: False
+  OPIK_SENTRY_ENABLE: False
+  GROQ_API_KEY: ${{ secrets.GROQ_API_KEY }}
+on:
+  workflow_call:
+
+jobs:
+  tests:
+    name: Langchain Python ${{matrix.python_version}}
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+    defaults:
+      run:
+        working-directory: sdks/python
+
+    strategy:
+      fail-fast: true
+      matrix:
+        python_version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
+
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v4
+
+      - name: Setup Python ${{matrix.python_version}}
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{matrix.python_version}}
+
+      - name: Install opik
+        run: pip install .
+
+      - name: Install test tools
+        run: |
+          cd ./tests
+          pip install --no-cache-dir --disable-pip-version-check -r test_requirements.txt
+
+      - name: Install lib
+        run: |
+          cd ./tests
+          pip install --no-cache-dir --disable-pip-version-check -r library_integration/langchain/legacy_requirements.txt
+
+      - name: change aws role
+        uses: aws-actions/configure-aws-credentials@v4
+        with:
+          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
+          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
+          role-to-assume: ${{ vars.BEDROCK_ROLE }}
+          aws-region: us-east-1
+          role-chaining: true
+          role-skip-session-tagging: true
+
+      - name: Run tests
+        run: |
+          cd ./tests/library_integration/langchain/
+          python -m pytest  -vv .

.github/workflows/lib-langchain-tests.yml

@@ -35,7 +35,7 @@ jobs:
     strategy:
       fail-fast: true
       matrix:
-        python_version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
+        python_version: ["3.10", "3.11", "3.12", "3.13"]
 
     steps:
       - name: Check out code

apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/events/BaseRedisSubscriber.java

@@ -3,6 +3,7 @@
 import com.comet.opik.infrastructure.StreamConfiguration;
 import io.dropwizard.lifecycle.Managed;
 import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.metrics.LongCounter;
 import io.opentelemetry.api.metrics.LongHistogram;
 import io.opentelemetry.api.metrics.Meter;
 import lombok.NonNull;
@@ -51,6 +52,7 @@ public abstract class BaseRedisSubscriber<M> implements Managed {
     protected final LongHistogram messageProcessingTime;
     protected final LongHistogram messageQueueDelay;
     protected final String payloadField;
+    protected final LongCounter backpressureDropCounter;
 
     protected BaseRedisSubscriber(@NonNull StreamConfiguration config, @NonNull RedissonReactiveClient redisson,
             @NonNull String metricsBaseName, @NonNull String payloadField) {
@@ -78,6 +80,11 @@ protected BaseRedisSubscriber(@NonNull StreamConfiguration config, @NonNull Redi
                 .setUnit("ms")
                 .ofLongs()
                 .build();
+
+        this.backpressureDropCounter = meter
+                .counterBuilder("%s_backpressure_drops_total".formatted(metricNamespace))
+                .setDescription("Total number of events dropped due to backpressure")
+                .build();
     }
 
     protected abstract String getMetricNamespace();
@@ -170,7 +177,13 @@ private void enforceConsumerGroup(RStreamReactive<String, M> stream) {
 
     private void setupStreamListener(RStreamReactive<String, M> stream) {
         this.streamSubscription = Flux.interval(config.getPoolingInterval().toJavaDuration())
-                .onBackpressureDrop()
+                .onBackpressureDrop(dropped -> {
+                    log.warn(
+                            "Backpressure drop detected: Unable to keep up with polling intervals. Polling interval tick dropped (sequence number: '{}').",
+                            dropped);
+                    // Record metric for dropped polling intervals
+                    backpressureDropCounter.add(1);
+                })
                 .flatMap(i -> stream.readGroup(config.getConsumerGroupName(), consumerId, redisReadConfig))
                 .onErrorContinue((throwable, object) -> log.error("Error reading from Redis stream", throwable))
                 .flatMapIterable(Map::entrySet)

apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/events/WebhookSubscriber.java

@@ -9,9 +9,11 @@
 import com.comet.opik.api.events.webhooks.WebhookEvent;
 import com.comet.opik.api.resources.v1.events.webhooks.WebhookHttpClient;
 import com.comet.opik.infrastructure.WebhookConfig;
+import com.comet.opik.infrastructure.auth.RequestContext;
 import com.comet.opik.utils.JsonUtils;
 import com.fasterxml.jackson.core.type.TypeReference;
 import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.metrics.LongCounter;
 import jakarta.inject.Inject;
 import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
@@ -24,8 +26,6 @@
 import java.util.List;
 import java.util.Map;
 
-import static com.comet.opik.infrastructure.auth.RequestContext.WORKSPACE_ID;
-
 /**
  * Service responsible for sending webhook events to external HTTP endpoints.
  * This service ONLY sends webhooks and does not handle aggregation or debouncing.
@@ -62,13 +62,21 @@ public class WebhookSubscriber extends BaseRedisSubscriber<WebhookEvent<?>> {
     private static final TypeReference<List<Guardrail>> LIST_GUARDRAIL_TYPE_REFERENCE = new TypeReference<>() {
     };
 
+    private final LongCounter webhookEventProcessedCounter;
+
     @Inject
     public WebhookSubscriber(@NonNull WebhookConfig webhookConfig,
             @NonNull RedissonReactiveClient redisson,
             @NonNull WebhookHttpClient webhookHttpClient) {
         super(webhookConfig, redisson, METRICS_BASE_NAME, WebhookConfig.PAYLOAD_FIELD);
         this.webhookHttpClient = webhookHttpClient;
         this.webhookConfig = webhookConfig;
+
+        // Pre-build metrics during initialization
+        this.webhookEventProcessedCounter = meter
+                .counterBuilder("opik_webhook_events_processed_total")
+                .setDescription("Total number of webhook events processed")
+                .build();
     }
 
     @Override
@@ -81,28 +89,33 @@ protected Mono<Void> processEvent(@NonNull WebhookEvent<?> event) {
         log.debug("Processing webhook event: id='{}', type='{}', url='{}'",
                 event.getId(), event.getEventType(), event.getUrl());
 
-        // Record metrics
-        var attributes = Attributes.builder()
-                .put("event_type", event.getEventType().getValue())
-                .put("workspace_id", event.getWorkspaceId())
-                .build();
-
-        return Mono.defer(() -> validateEvent(event))
-                .then(Mono.defer(() -> webhookHttpClient
-                        .sendWebhook(deserializeEventPayload((WebhookEvent<Map<String, Object>>) event))))
-                .contextWrite(ctx -> ctx.put(WORKSPACE_ID, event.getWorkspaceId()))
+        return validateEvent(event)
+                .then(Mono.fromCallable(() -> Attributes.builder()
+                        .put("event_type", event.getEventType().getValue())
+                        .put("workspace_id", event.getWorkspaceId())
+                        .build())
+                        .flatMap(attributes -> {
+                            @SuppressWarnings("unchecked")
+                            WebhookEvent<Map<String, Object>> webhookEvent = (WebhookEvent<Map<String, Object>>) event;
+
+                            return Mono.fromCallable(() -> deserializeEventPayload(webhookEvent))
+                                    .subscribeOn(Schedulers.boundedElastic())
+                                    .flatMap(webhookHttpClient::sendWebhook)
+                                    .doOnSuccess(unused -> {
+                                        log.info("Successfully sent webhook: id='{}', type='{}', url='{}'",
+                                                event.getId(), event.getEventType(), event.getUrl());
+
+                                        // Record success metrics
+                                        webhookEventProcessedCounter.add(1,
+                                                attributes.toBuilder().put("status", "success").build());
+                                    })
+                                    .onErrorResume(
+                                            throwable -> handlePermanentFailure(event, throwable).then(Mono.empty()));
+                        }))
+                .onErrorResume(
+                        throwable -> handlePermanentFailure(event, throwable).then(Mono.empty()))
+                .contextWrite(ctx -> ctx.put(RequestContext.WORKSPACE_ID, event.getWorkspaceId()))
                 .subscribeOn(Schedulers.boundedElastic())
-                .doOnSuccess(unused -> {
-                    log.info("Successfully sent webhook: id='{}', type='{}', url='{}'",
-                            event.getId(), event.getEventType(), event.getUrl());
-
-                    // Record success metrics
-                    meter.counterBuilder("opik_webhook_events_processed_total")
-                            .setDescription("Total number of webhook events processed")
-                            .build()
-                            .add(1, attributes.toBuilder().put("status", "success").build());
-                })
-                .onErrorResume(throwable -> handlePermanentFailure(event, throwable).then(Mono.empty()))
                 .then();
     }
 
@@ -155,6 +168,7 @@ private Mono<Void> validateEvent(@NonNull WebhookEvent<?> event) {
     public static WebhookEvent<Map<String, Object>> deserializeEventPayload(
             @NonNull WebhookEvent<Map<String, Object>> event) {
         Map<String, Object> payload = event.getPayload();
+        @SuppressWarnings("unchecked")
         List<String> metadatas = (List<String>) payload.getOrDefault("metadata", List.of());
 
         var deserializeMetadata = metadatas.stream()