Skip to content

feat(contrib/lovoo/goka): add goka integration with APM and DSM support#5029

Draft
anapsix wants to merge 1 commit into
DataDog:mainfrom
anapsix:feat/lovoo-goka-integration
Draft

feat(contrib/lovoo/goka): add goka integration with APM and DSM support#5029
anapsix wants to merge 1 commit into
DataDog:mainfrom
anapsix:feat/lovoo-goka-integration

Conversation

@anapsix

@anapsix anapsix commented Jul 14, 2026

Copy link
Copy Markdown

What does this PR do?

Adds a new contrib/lovoo/goka integration (module
github.com/DataDog/dd-trace-go/contrib/lovoo/goka/v2, package goka) that instruments the
goka Kafka-streams library with both APM distributed
tracing and Data Streams Monitoring (DSM) in a single integration.

goka builds its consumer group on top of IBM/sarama but owns the
sarama.ConsumerGroupHandler internally (it passes the *Processor itself to
consumerGroup.Consume), so contrib/IBM/sarama cannot attach. This package instruments
through goka's public extension points instead — no changes to goka are required:

  • NewTracer(...) + WrapContext (register via goka.WithContextWrapper): sets the inbound
    DSM direction:in checkpoint and returns a goka.Context whose Emit/Loopback propagate
    the active consume span and a DSM direction:out checkpoint into the outbound headers via
    goka.WithCtxEmitHeaders.
  • WrapCallback: wraps each goka.Input callback so every input message runs inside a
    kafka.consume span, finished when the callback returns and error-tagged on ctx.Fail
    (panics are recovered, tagged, and re-raised so goka still shuts the processor down).
    WithContextWrapper has no completion hook, so the callback is the only place to finish the
    span. This is also where the DSM commit offset is recorded — on successful processing, or, if
    the handler uses ctx.DeferCommit, when that deferred commit succeeds.
  • No kafka.produce spans. goka.Context.Emit is asynchronous and returns no handle, so a
    produce span could never be finished on the emit's real completion nor tagged with an async
    delivery error. Trace continuity is instead preserved by propagating the consume span through
    the emitted message headers, so the downstream consumer continues the trace.
  • EmitHeaders: covers the standalone goka.Emitter path.
  • Options: WithService, WithDataStreams, and WithLoopSuffix (to keep Loopback DSM
    checkpoints aligned when the app changes goka's loop suffix via goka.SetLoopSuffix).
  • A single header carrier satisfies both the tracer and datastreams TextMap interfaces.

Follows the existing Kafka-family contrib conventions (IBM/sarama, twmb/franz-go):
instrumentation.Load with a new PackageLovooGoka, WithService/WithDataStreams options,
the base64 DSM carrier, standard messaging tags, dedicated submodule with
replace => ../../.., and mocktracer + DSM pathway tests. Registers PackageLovooGoka
in instrumentation/packages.go and contribIntegrations in ddtrace/tracer/option.go.

No orchestrion.yml: unlike sarama/franz-go there is no single constructor call to rewrite
(a processor is assembled from a user-defined graph plus options), so the integration is an
explicit one-line opt-in.

Known limitations (inherent to goka's extension points)

  • The kafka.consume span is finished when the callback returns; because emits are
    asynchronous, an emit that fails after the callback (and shuts the partition down) is not
    reflected on the span.
  • DSM commit-offset tracking happens in WrapCallback, so an Input whose callback is not
    wrapped reports no committed offset to DSM.
  • DSM payload size counts the bytes observable at the seam: key + headers on consume, and key +
    value on produce only when the value is a raw []byte/string; codec-encoded values are
    opaque here and reported as 0.
  • goka also invokes the context wrapper for VisitValues callbacks, whose Topic() is the
    visit name; with DSM enabled these produce an inbound checkpoint keyed on that name.

Motivation

There is currently no way to get APM traces or DSM checkpoints from a goka processor without
hand-rolling instrumentation. Related discussion: #5028. Opening as a draft pending
maintainer agreement on the approach (per CONTRIBUTING).

Reviewer's Checklist

  • Changed code has unit tests for its functionality at or near 100% coverage. (93.2% statement coverage; remaining lines are error/ok=false branches. Includes an INTEGRATION-gated end-to-end test.)
  • System-Tests covering this feature have been added and enabled with the va.b.c-dev version tag. (N/A — no agent-facing behavior; happy to add if desired.)
  • There is a benchmark for any new code, or changes to existing code. (None added yet; can add if wanted.)
  • If this interacts with the agent in a new way, a system test has been added. (N/A.)
  • New code is free of linting errors. make lint (golangci-lint, gocritic disabled for the nested module as in scripts/lint.sh) is clean.
  • New code doesn't break existing tests. go test -race ./... passes in the new module; check-supported-config, check-modules (fix-modules → no diff), check-format, copyright, and checklocks all verified locally.
  • Add an appropriate team label so this PR gets put in the right place for the release notes. (Maintainer to assign.)
  • All generated files are up to date. make generate produces no diff (the orchestrion aggregator only imports contribs that ship an orchestrion.yml).
  • Non-trivial go.mod changes, e.g. adding new modules, are reviewed by @DataDog/dd-trace-go-guild. (New submodule + go.work entry — flagging for guild review. make fix-modules produces no diff.)

Add a new contrib package instrumenting github.com/lovoo/goka with both
Datadog APM distributed tracing and Data Streams Monitoring in a single
integration.

goka builds its consumer group on top of IBM/sarama but owns the
sarama.ConsumerGroupHandler internally, so the contrib/IBM/sarama
integration cannot attach. This package instruments through goka's public
extension points instead, requiring no changes to goka:

- NewTracer + WrapContext (goka.WithContextWrapper) sets the inbound DSM
  checkpoint and returns a context whose Emit/Loopback propagate the active
  consume span and a DSM outbound checkpoint into the emitted headers.
- WrapCallback runs each input message inside a kafka.consume span,
  finishing it around the callback (panics recovered and error-tagged) and
  recording the DSM commit offset on success, honoring ctx.DeferCommit.
- No kafka.produce span is created: goka.Context.Emit is asynchronous with
  no completion handle, so trace continuity is preserved by propagating the
  consume span through the outbound headers instead.
- EmitHeaders covers the standalone goka.Emitter path.
- WithService, WithDataStreams, and WithLoopSuffix options; a single header
  carrier satisfies both the tracer and datastreams TextMap interfaces.

Register PackageLovooGoka in instrumentation/packages.go and
contribIntegrations in ddtrace/tracer/option.go. Includes unit tests
(carrier round-trip, DSM no-op when disabled, span parent/child linkage,
error tagging, emit/loopback injection, DeferCommit, context delegation)
and an INTEGRATION-gated end-to-end test.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
@anapsix anapsix force-pushed the feat/lovoo-goka-integration branch from e7c4af0 to c6dbdd9 Compare July 14, 2026 18:34
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant