github-featured / comet-ml--opik

发布

0 发布 8,340 Git 标签

RSS 订阅源
  • 2.0.33-5325 68d8dc2107
    比较

    [NA] [BE] fix(python-backend): one metric exporter per pod (move rq_worker metrics to parent) (#6683)

    frostbyte_neo 发布于 2026-05-13 11:28:28 +00:00

    • fix(python-backend): emit rq_worker metrics from parent only — one exporter per pod

    The pod runs gunicorn (--workers 1 --threads N) with N RQ worker threads
    inside that single process. RQ's default Worker forks a child per job, and
    the inherited OTel SDK in the child has its own PeriodicExportingMetricReader.
    Because the Python SDK's default Resource carries no per-process identifier
    (no service.instance.id, no process.pid), every forked child emits
    system_cpu_time_seconds_total / system_memory_* / process_runtime_*
    with the same label set as the parent and its siblings. The OTel Collector
    forwards each cycle faithfully, Prometheus rejects the remote-write batch
    with HTTP 400 duplicate sample for timestamp ... overrides not allowed,
    and ~1000+ metric points get dropped per export cycle on a busy pod
    (observed in self-hosted-eks against dev-monitoring-opentelemetry-collector,
    where one pod with ~17 concurrent optimizer jobs was emitting system_cpu_time
    at 18×/min while sibling pods on the same image hit the expected 1×/min).

    Fix: make the pod have a single metric exporter chain — the parent's.

    1. main_work_horse (runs post-fork in the child) calls
      metrics.get_meter_provider().shutdown() on the inherited MeterProvider.
      Linux fork is COW, so the shutdown affects only the child's local copy
      of the provider state — the parent's reader/exporter is untouched.
      TracerProvider is intentionally not shut down: span IDs are unique per
      span, traces don't have the duplicate-sample issue, and we want spans
      from auto-instrumentation inside a job (Flask, Requests, ...) to keep
      flowing.

    2. execute_job (runs in the parent, wraps super().execute_job())
      records the per-job counters and histograms — rq_worker.jobs.processed,
      rq_worker.jobs.succeeded, rq_worker.jobs.failed,
      rq_worker.job.processing_time, rq_worker.job.queue_wait_time,
      rq_worker.job.total_time, rq_worker.jobs.concurrent. After
      super().execute_job() returns, job.refresh() pulls the final
      started_at/ended_at/is_failed/exc_info from Redis (the child
      wrote them before exiting) so the parent has the same data the child
      used to. Per-job metrics keep flowing; just from a single exporter.

    3. perform_job (child) loses its metric-emission body and now only
      logs the lifecycle. The metric work is already done in the parent.

    Side benefits:

    • No collector-side change required. The current
      resource/strip-instance-id processor in comet-monitoring/values-dev.yaml
      stays as-is — there is nothing to keep distinct now that only the parent
      emits.
    • One MeterProvider per pod is more memory- and CPU-efficient than the
      "one per forked child" alternative we considered.

    Trade-offs:

    • error_type label on rq_worker.jobs.failed is now best-effort: parsed
      from job.exc_info (the child's traceback string) rather than read from
      a live exception object. Falls back to "UnknownError" if the format
      isn't what we expect. Other failure dimensions (counter increments,
      durations) are unaffected.

    Co-Authored-By: Claude Opus 4.7 (1M context) noreply@anthropic.com

    • test(python-backend): unit tests for MetricsWorker emission + child shutdown

    Six tests covering the two halves of the fix:

    execute_job (parent) — emits per-job counters/histograms via the only
    active MeterProvider in the pod:

    • test_success_records_processed_succeeded_and_durations
      Happy path: jobs.processed +1, jobs.succeeded +1, no jobs.failed,
      processing_time ≈ ended_at - started_at, total_time ≈
      ended_at - created_at, queue_wait_time recorded.

    • test_failed_job_records_error_type_parsed_from_exc_info
      job.is_failed = True with a multi-line traceback string —
      error_type label is extracted from the last "ExceptionClass: msg"
      line. jobs.failed +1, jobs.succeeded untouched.

    • test_hard_execute_job_exception_records_failed_with_exception_class
      super().execute_job raises — exception class name lands on the
      error_type label and the exception propagates.

    • test_concurrent_counter_balances_to_zero_after_a_single_job
      UpDownCounter +1/-1 net to zero on a clean job lifecycle.

    main_work_horse (forked child) — shuts down the inherited MeterProvider:

    • test_shutdown_is_called_then_super_main_work_horse_runs
      shutdown() called once, super().main_work_horse called after.

    • test_shutdown_exception_is_swallowed_and_super_still_runs
      If shutdown() raises (e.g., already shutdown), the job still
      runs — we don't fail jobs because the SDK is in an odd state.

    Test design notes:

    • One session-scoped InMemoryMetricReader since OTel Python's
      set_meter_provider is set-once per process. Tests stay isolated
      by emitting unique function attributes and filtering data points
      on retrieval.

    • The child-side tests monkeypatch metrics.get_meter_provider inside
      the metrics_worker module to return a local mock so the session-wide
      provider is left intact for the parent-side tests in the same file.

    • MetricsWorker is constructed via the real Worker.init over
      fakeredis.FakeStrictRedis — no new/init bypass.

    Co-Authored-By: Claude Opus 4.7 (1M context) noreply@anthropic.com

    • review: improvements to single-exporter-per-pod fix

    Addresses feedback on the original commit:

    • Quote BaseException | None in execute_job so the local-variable
      annotation isn't evaluated on Python < 3.10 (matches the static
      method's existing style).

    • Move concurrent_jobs_counter.add(1) inside the outer try/finally
      pair so the matching add(-1) always fires. Previously, if the +1
      itself raised, no -1 would run — theoretical but ugly.

    • Restore the pre-refactor queue_wait_time SLI: record
      job.started_at - job.created_at from _record_job_completion_metrics
      after job.refresh(), instead of now - job.created_at at execute_job
      entry. Existing dashboards/alerts keyed on this histogram see no
      semantic shift now.

    • Harden _error_type_from_job against multi-line exception messages:
      scan from the end and skip indented continuation lines, matching the
      first column-0 Name: line via regex. Dotted-module prefix is still
      stripped to the leaf class name.

    • Type the metric_attributes parameter as Mapping[str, str] instead
      of bare dict for static-analysis clarity.

    • Tests:

      • Drop autouse=True on the session-scoped MeterProvider fixture so
        it doesn't preempt other test files that might want their own
        provider (OTel's set_meter_provider is set-once per process).
      • Add rq_worker.jobs.processed and rq_worker.jobs.concurrent == 0
        assertions to the is_failed=True and hard-exception cases, matching
        the symmetry of the success test.
      • New test for the multi-line-exception parsing — covers a
        requests.exceptions.ConnectionError traceback with a colon-bearing
        multi-line message and asserts the leaf class name is extracted.

    Co-Authored-By: Claude Opus 4.7 (1M context) noreply@anthropic.com

    • review: harden refresh-failure path; assert shutdown was called

    baz-reviewer flagged two issues:

    1. _record_job_completion_metrics swallowed job.refresh() errors but
      then fell through to getattr(job, "is_failed", False). In RQ,
      Job.is_failed triggers another Redis round-trip (get_status()),
      so a refresh failure (e.g., NoSuchJobError, Redis outage) could
      raise from the finally block — silently dropping the terminal
      metric AND surfacing a finished job as a worker error.

      Fix: track refresh success in a local. When refresh fails, do NOT
      consult is_failed (avoid the second Redis call). Instead, emit
      rq_worker.jobs.failed{error_type="RefreshFailed"} so the terminal
      outcome metric isn't silently dropped, and skip the duration
      histograms (their inputs are stale/absent without a successful
      refresh).

    2. test_shutdown_exception_is_swallowed_and_super_still_runs didn't
      assert that MeterProvider.shutdown() was actually invoked. The
      test would still pass if the child skipped shutdown entirely.

      Fix: add local_provider.shutdown.assert_called_once() so the test
      verifies the expected behavior.

    Plus a new test, test_refresh_failure_emits_explicit_unknown_outcome,
    that asserts:

    • processed counter still +1
    • rq_worker.jobs.failed{error_type="RefreshFailed"} +1
    • is_failed is NEVER consulted (would fail the test if it were)
    • no bogus duration histograms recorded with stale/None timestamps
    • concurrent counter balances back to zero

    Co-Authored-By: Claude Opus 4.7 (1M context) noreply@anthropic.com

    Co-authored-by: Claude Opus 4.7 (1M context) noreply@anthropic.com

    下载附件