github-featured / comet-ml--opik

发布

0 发布 8,340 Git 标签

RSS 订阅源
  • 2.0.11-5086 f858952b36
    比较

    [OPIK-5659] [BE] fix: prevent concurrent experiment aggregation and reduce ClickHouse query overload (#6076)

    frostbyte_neo 发布于 2026-04-22 13:34:18 +00:00

    • [OPIK-5659] [BE] fix: prevent concurrent experiment aggregation and reduce ClickHouse query overload

    Root cause: aggregation lock TTL (1 min) was far shorter than the actual processing
    time for large experiments (~87 min at batchSize=1000 for 1M items). When the lock
    expired, multiple nodes reprocessed the same experiment concurrently, each issuing the
    full set of ClickHouse queries per batch — causing 2000+ queries/minute spikes.

    Fixes:

    • Switch from executeWithLockCustomExpire to bestEffortLock with 500ms acquire wait,
      so a locked experiment is immediately skipped rather than queued concurrently
    • Apply .timeout(lockTTL) to the processing Mono so Reactor cancels in-flight R2DBC
      queries when TTL elapses, instead of leaving them running silently after lock expiry
    • Raise aggregationLockTime default from 1 min to 10 min
    • Add retry logic: on timeout, re-publish via debounce up to maxLockExpiryRetries=3;
      reset counter on successful completion
    • All new config fields (lockAcquireWait, maxLockExpiryRetries, retryCounterTtl) are
      externalised in ExperimentDenormalizationConfig — no hardcoded values
    • Resolve getProjectId once per experiment (not per batch) eliminating N-1 redundant
      ClickHouse round trips across the expand() loop
    • Pass traceIds extracted from the already-fetched batch directly to all 5 parallel
      queries, replacing the repeated experiment_items FINAL CTE subquery in each
    • fix(subscriber): reset retry counter only on successful lock+processing

    Move resetRetryCounter into the action Mono so it only runs when this
    node actually acquired the lock and processing completed successfully.
    Previously it ran after bestEffortLock regardless of whether the lock
    was acquired, resetting the shared counter on skip (Mono.empty) paths.

    Also add workspaceId to the doOnError log for consistency with other
    log statements in the same method.

    • test(subscriber): verify retry counter not reset when lock not acquired

    • [OPIK-5659] [BE] refactor: replace FINAL with ORDER BY + LIMIT 1 BY across aggregation queries

    Removes FINAL from experiments/spans/traces/experiment_items/assertion_results
    reads and replaces with explicit ORDER BY (<sort_key>) DESC, last_updated_at DESC

    • LIMIT 1 BY <dedup_key> to avoid the per-query ReplacingSorted merge work.

    Also fixes two bugs:

    • ::trace_ids typo (double colon) → :trace_ids
    • GET_ASSERTIONS_DATA inner subquery was missing the name column and
      LIMIT 1 BY clause, causing "Unknown expression or function identifier
      'name'" at runtime. Aligned with the canonical assertion_results dedup
      pattern used elsewhere in the file.
    • fix(aggregation): keep latest row per id in GET_EXPERIMENT_ITEMS

    ORDER BY ... ASC, last_updated_at ASC before LIMIT 1 BY id kept the
    oldest row per id, which could cause populateExperimentItemAggregates
    to process stale snapshots. Flipped to DESC, last_updated_at DESC to
    match the latest-row dedup pattern used in GET_TRACES_DATA and
    GET_SPANS_DATA.

    • fix(aggregation): keep cursor pagination ordering in GET_EXPERIMENT_ITEMS

    The prior commit flipped ORDER BY to DESC on all columns so LIMIT 1 BY id
    would keep the latest row per id, but that broke cursor pagination: with
    DESC ordering and the filter id > :cursor, each subsequent batch only
    excludes the smallest id of the previous batch and re-returns the rest,
    so the iteration only ever processes the top ~batchSize ids of the
    experiment and never progresses.

    Keep id ASC for forward cursor progression while using last_updated_at
    DESC so LIMIT 1 BY id still selects the latest version per id.

    • perf(aggregation): bulk-insert experiment_item_aggregates via ClickHouse HTTP JSONEachRow

    The prior INSERT used a StringTemplate that rendered one VALUES tuple per
    item with N distinct named parameters (:id0, :id1, ...). R2DBC's
    named-parameter resolution grew super-linearly with batch size, so
    EXPERIMENT_AGGREGATES_BATCH_SIZE=10000 made the Java-side render + bind
    pipeline take ~45s per batch (CH server-side INSERT itself was <200ms).
    At the 60s lock TTL this left ~0 headroom: every attempt processed one
    10k-item batch, then got cancelled. Smaller batch sizes avoided the
    super-linear cost but capped throughput at ~1.5k items/sec.

    Replace the R2DBC path for this INSERT with the ClickHouse v2 HTTP
    client (com.clickhouse:client-v2) POSTing JSONEachRow directly, with
    client-request + server-response compression enabled. The rest of the
    aggregation pipeline keeps using R2DBC.

    At batchSize=10000 the per-batch cost drops from ~45s to ~60-100ms
    (bodyMs ~20-40 + executeMs ~35-60). End-to-end a 1M-item experiment
    now completes aggregation in ~45s as a single attempt, well within
    the 60s lock TTL. ExperimentAggregatesIntegrationTest (126 tests)
    passes.

    Also wires EXPERIMENT_AGGREGATES_BATCH_SIZE through the backend
    docker-compose service, defaulting to 10000.

    • refactor(aggregation): address PR review feedback
    • Log values moved to end of sentence for production greppability
      (ExperimentAggregatesSubscriber, ExperimentAggregatesService).
    • @Builder on BatchResult; call sites use builder pattern.
    • Javadoc on appendJsonRow documenting JSONEachRow contract, JsonUtils
      reuse, shared-StringBuilder rationale, and null-coalescing policy.
    • Javadoc on insertExperimentItems documenting why the ClickHouse v2
      HTTP client + JSONEachRow path is used only for this high-volume
      batch insert (EXPERIMENT_AGGREGATES_BATCH_SIZE can exceed 1K), why
      date_time_input_format is scoped per-request, and the NUM_ROWS_WRITTEN
      return semantics.
    • DatabaseAnalyticsFactory encapsulates v2 Client construction via
      buildClient(), parseQueryParameters() splits queryParameters into
      driver options (R2DBC-specific, not forwarded) and server settings
      (custom_http_params content → Client.Builder.serverSetting).
      DatabaseAnalyticsModule.getDatabaseAnalyticsFactory provider removed.
    • Unit tests (DatabaseAnalyticsFactoryTest) for parseQueryParameters
      and integration tests (DatabaseAnalyticsFactoryIntegrationTest)
      verifying custom_http_params entries land in system.settings via a
      real ClickHouse Testcontainer, that driver options don't leak, and
      that a JSONEachRow round-trip completes.
    • config.yml: new env-var-overridable settings for the retry pipeline
      (lockAcquireWait, maxLockExpiryRetries, retryCounterTtl); updated
      aggregationLockTime default to 10m to match code.
    • refactor(aggregation): add processed count to denormalization job log and guard it

    The "finished processing all experiments" log previously fired at INFO on every
    5s polling tick, even when no experiments were pending. Now the job tracks the
    number of processed experiments and only emits the completion log when at least
    one was actually processed. Value placed at the end of the sentence per the log
    format convention used across the aggregation pipeline.

    • fix(aggregation): correct processedCount label and semantics in logs
    • Per-batch log in ExperimentAggregatesService was labelled batchSize but
      carried result.processedCount(); rename placeholder to processedCount so
      the wording matches the value. Configured batch size is still logged once
      at job start.
    • Denormalization job counter previously incremented in doOnNext before
      flatMap(processExperiment), so the processedExperiments log counted items
      that failed and were skipped by onErrorContinue. Move the increment to
      doOnSuccess on the inner Mono so the count reflects actually processed
      experiments.
    • refactor(aggregation): use count(DISTINCT author) in feedback score aggregation

    Defensive change: count() was correct given the LIMIT 1 BY ..., author
    dedup but silently overcounts if the dedup ever regresses. Switching
    to count(DISTINCT author) keeps the invariant explicit at negligible
    cost (one row per author in the deduped CTE).

    • refactor(aggregation): address PR review nits
    • ExperimentAggregatesSubscriber: extract buildRetryCountKey helper so
      retriggerIfBelowMaxRetries and resetRetryCounter share a single key
      generation site.
    • ExperimentDenormalizationJob: switch processExperiment to return the
      experiment id so the outer doOnNext actually fires per successful
      processing (Mono never triggers doOnNext). Retains the same
      per-experiment log and counter increment semantic.
    • config.yml: lower retryCounterTtl default from 2h to 30m to match
      the worst-case retry cycle (3 attempts at 10m lock TTL) plus a
      modest idle buffer.
    • config-test.yml: add missing lockAcquireWait, maxLockExpiryRetries,
      retryCounterTtl entries to the experimentDenormalization block.
    • fix(aggregation): set retry-counter TTL before publishing

    Previously counter.expire(...) ran only after publisher.publish(...)
    succeeded. If publish failed, the retry counter was incremented
    without a TTL and could linger, causing subsequent lock-expiry
    retries to hit maxLockExpiryRetries prematurely.

    Reordering so expire runs right after the increment guarantees the
    counter always has a TTL and will age out naturally regardless of
    whether the publish succeeds.

    下载附件