OSAC-3428: Implement Provider Adapter framework with Kafka consumer lifecycle

[amito]

Description

Adds a new Go framework package at osac-metering/adapters/ that handles the full Kafka consumer lifecycle for provider metering adapters. Concrete adapter binaries (M360, Cost Management, etc.) import this package, implement the ProviderAdapter interface, and let the framework's Runner manage consumption, retries, dedup, and flushing.

Architecture:

Kafka topics
    │
    ▼
Runner (Sarama ConsumerGroup)
    │
    ├── Dedup cache (CloudEvent ID, 10m TTL)
    ├── Out-of-order tracker (resource_id → last transition_time)
    │
    ▼
ProviderAdapter.Submit(event)
    │
    ├── Success → track offset
    ├── RetryableError → exponential backoff (1s–5m, max 10 attempts)
    └── NonRetryableError → log + skip
    │
    ▼
Flush ticker (default 10s)
    │
    ▼
ProviderAdapter.Flush()
    │
    ├── Success → commit Kafka offsets
    └── Failure → do not commit (redelivery on restart)

Key components:

File	Responsibility
adapter.go	ProviderAdapter interface, MeteringEvent, SubmitResult, error types
runner.go	Runner with Kafka consumer lifecycle, flush scheduling, graceful shutdown
retry.go	Exponential backoff (1s–5m, ±25% jitter, max 10 attempts), error classification
dedup.go	In-memory CloudEvent ID dedup cache with TTL eviction
order.go	Out-of-order detection via transition_time per resource_id
metrics.go	8 Prometheus metrics (osac_adapter_*), handler for adapter HTTP servers
kafka.go	Sarama consumer configuration, TLS/SASL/SCRAM helpers
cmd/echo-adapter/	Echo adapter binary for E2E testing (see below)

Echo adapter (adapters/cmd/echo-adapter/):

Replaces the previous metering-service/cmd/test-adapter/ (which used raw Sarama and did not exercise the framework). The echo adapter implements the full ProviderAdapter interface and exercises the complete Runner lifecycle. It exposes an HTTP query API for E2E test assertions:

Endpoint	Description
GET /events	List events with filters (type, resource_id, since, limit)
GET /events/count	Count matching events
GET /events/{id}	Lookup single event by CloudEvent ID
DELETE /events	Clear buffer for test isolation
GET /metrics	Prometheus metrics
GET /healthz	Health check

Configuration: ECHO_BUFFER_SIZE (default 1000), METRICS_ADDR (default :2112), plus standard KAFKA_* env vars.

JIRA Tasks

• OSAC-3428 — Implement Provider Adapter framework with Kafka consumer lifecycle
• OSAC-3763 — Implement Echo Provider Adapter for integration testing
• Parent epic: OSAC-3411 — Provider Adapter
• Parent feature: OSAC-985 — Metering and Usage Tracking

Not Included / Out of Scope

• DLQ handling — Kafka DLQ topic, publisher, error envelope (OSAC-3436, moved to separate epic OSAC-3666)
• Concrete adapter implementations — M360 adapter (OSAC-3413), Cost Management adapter
• E2E / integration tests with real Kafka — OSAC-3440
• Replay tooling and osac_adapter_replay_lag_events metric

How Has This Been Tested?

Unit-testing

68 unit tests using Ginkgo v2 / Gomega, all passing with race detector enabled:

$ cd osac-metering && go test ./adapters/... -race -v
ok  github.com/osac-project/osac-metering/adapters  2.137s

Test coverage by component:

Test file	Coverage
adapter_test.go	Error type wrapping/unwrapping, interface contracts
dedup_test.go	Duplicate suppression, TTL expiry, concurrent safety
order_test.go	Out-of-order detection, independent resources, TTL eviction
retry_test.go	Backoff intervals, max attempts, non-retryable skip, context cancellation
metrics_test.go	Counter increments, histogram observations, HTTP handler serving
runner_test.go	Full lifecycle: consume → dedup → order check → submit with retry → flush → offset commit; graceful shutdown; error handling; rebalance safety (context cancellation does not lose events)
kafka_test.go	TLS toggle (enabled/disabled), consumer defaults, broker address parsing

No real Kafka broker required — tests use mock Sarama consumer group sessions and claims.

E2E testing

The echo adapter binary (adapters/cmd/echo-adapter/) is included in this PR with Helm chart templates and CI image build workflow. E2E integration tests against a development cluster are tracked in OSAC-3440.

Review follow-up changes

Changes made in response to review feedback (masayag, CodeRabbit):

Amended into existing commits:

• Dedup-suppressed messages now track offsets (prevents unnecessary redelivery on restart)
• adapter.Close() is bounded by a 10s deadline in the Runner's shutdown path (addresses unbounded Close concern while preserving io.Closer compatibility)
• Context cancellation during retry backoff no longer tracks the offset — the event is redelivered after rebalance instead of being silently lost
• Echo adapter /healthz now calls adapter.HealthCheck() instead of returning hardcoded 200
• Echo adapter startup log redacts broker addresses (logs count only)
• Echo adapter deployment sets KAFKA_CONSUMER_GROUP to match KafkaUser ACL
• Echo adapter values removed from chart defaults (test-only, provided via CI values)
• Adapters Makefile build-echo-adapter creates bin/ directory before build
• Pre-commit hook for adapters triggers on config file changes (.golangci.yml, Makefile, go.mod, go.sum)

New commits:

• TLS toggle — TLSEnabled field on KafkaConfig; TLS is only configured when explicitly requested. kafka_test.go added with unit tests for TLS paths, consumer defaults, and broker address parsing.
• CI scaffolding — Makefile, .golangci.yml, GitHub Actions unit test job (run-osac-metering-adapters-tests), and pre-commit lint hook (osac-metering-adapters-golangci-lint), matching the sibling metering-service module convention.
• Echo adapter (OSAC-3763) — Replaces metering-service/cmd/test-adapter/ with adapters/cmd/echo-adapter/ that exercises the full Runner framework. Adds HTTP query API (/events, /events/count, /events/{id}, DELETE /events), Containerfile, Helm chart templates, and CI image build workflow (build-metering-echo-adapter-image.yaml).

References

• Design spec: docs/superpowers/specs/2026-08-05-provider-adapter-framework-design.md
• Metering EP: OSAC-985 design
• Implementation plan: docs/superpowers/plans/2026-08-05-provider-adapter-framework.md

Merge criteria

• The commits are squashed in a cohesive manner and have meaningful messages.
• Testing instructions have been added in the PR body (for PRs involving changes that are not immediately obvious).
• The developer has manually tested the changes and verified that the changes work

Summary by CodeRabbit

• New Features

  • Added shared Kafka-based metering event processing with duplicate suppression and out-of-order tracking.
  • Added configurable retries, TLS/SASL authentication, provider health checks, flushing, and offset tracking.
  • Added Prometheus metrics for submissions, retries, duplicates, ordering, lag, and flushing.
  • Added an Echo Adapter with event inspection, filtering, counting, deletion, health, and metrics endpoints.
  • Added optional deployment and OpenShift routing for the Echo Adapter.

• Tests

  • Added comprehensive coverage for processing, retries, metrics, deduplication, ordering, and Kafka configuration.

[openshift-ci-robot]

@amito: This pull request references OSAC-3428 which is a valid jira issue.

Warning: The referenced jira issue has an invalid target version for the target branch this PR targets: expected the task to target the "5.0.0" version, but no target version was set.

Details

In response to this:

Description

Adds a new Go framework package at osac-metering/adapters/ that handles the full Kafka consumer lifecycle for provider metering adapters. Concrete adapter binaries (M360, Cost Management, etc.) import this package, implement the ProviderAdapter interface, and let the framework's Runner manage consumption, retries, dedup, and flushing.

Architecture:

Kafka topics
   │
   ▼
Runner (Sarama ConsumerGroup)
   │
   ├── Dedup cache (CloudEvent ID, 10m TTL)
   ├── Out-of-order tracker (resource_id → last transition_time)
   │
   ▼
ProviderAdapter.Submit(event)
   │
   ├── Success → track offset
   ├── RetryableError → exponential backoff (1s–5m, max 10 attempts)
   └── NonRetryableError → log + skip
   │
   ▼
Flush ticker (default 10s)
   │
   ▼
ProviderAdapter.Flush()
   │
   ├── Success → commit Kafka offsets
   └── Failure → do not commit (redelivery on restart)

Key components:

File	Responsibility
adapter.go	ProviderAdapter interface, MeteringEvent, SubmitResult, error types
runner.go	Runner with Kafka consumer lifecycle, flush scheduling, graceful shutdown
retry.go	Exponential backoff (1s–5m, ±25% jitter, max 10 attempts), error classification
dedup.go	In-memory CloudEvent ID dedup cache with TTL eviction
order.go	Out-of-order detection via transition_time per resource_id
metrics.go	8 Prometheus metrics (osac_adapter_*), handler for adapter HTTP servers
kafka.go	Sarama consumer configuration, TLS/SASL/SCRAM helpers

JIRA Tasks

• OSAC-3428 — Implement Provider Adapter framework with Kafka consumer lifecycle
• Parent epic: OSAC-3411 — Provider Adapter
• Parent feature: OSAC-985 — Metering and Usage Tracking

Not Included / Out of Scope

• DLQ handling — Kafka DLQ topic, publisher, error envelope (OSAC-3436, moved to separate epic OSAC-3666)
• Concrete adapter implementations — M360 adapter (OSAC-3413), Cost Management adapter
• Helm chart / deployment manifests — separate deployment task
• E2E / integration tests with real Kafka — OSAC-3440
• Replay tooling and osac_adapter_replay_lag_events metric
• Echo adapter binary (adapters/cmd/echo-adapter/) — local smoke-test tool, not included in this PR

How Has This Been Tested?

55 unit tests using Ginkgo v2 / Gomega, all passing with race detector enabled:

$ cd osac-metering && go test ./adapters/... -race -v
ok  github.com/osac-project/osac-metering/adapters  2.137s

Test coverage by component:

Test file	Coverage
adapter_test.go	Error type wrapping/unwrapping, interface contracts
dedup_test.go	Duplicate suppression, TTL expiry, concurrent safety
order_test.go	Out-of-order detection, independent resources, TTL eviction
retry_test.go	Backoff intervals, max attempts, non-retryable skip, context cancellation
metrics_test.go	Counter increments, histogram observations, HTTP handler serving
runner_test.go	Full lifecycle: consume → dedup → order check → submit with retry → flush → offset commit; graceful shutdown; error handling

No real Kafka broker required — tests use mock Sarama consumer group sessions and claims.

References

• Design spec: docs/superpowers/specs/2026-08-05-provider-adapter-framework-design.md
• Metering EP: OSAC-985 design
• Implementation plan: docs/superpowers/plans/2026-08-05-provider-adapter-framework.md

Merge criteria

• The commits are squashed in a cohesive manner and have meaningful messages.
• Testing instructions have been added in the PR body (for PRs involving changes that are not immediately obvious).
• The developer has manually tested the changes and verified that the changes work

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the openshift-eng/jira-lifecycle-plugin repository.

[coderabbitai]

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

Walkthrough

The PR adds the osac-metering/adapters Go module. It defines provider contracts, Kafka security and consumer settings, deduplication, ordering checks, retries, Prometheus metrics, and a Kafka consumer runner. It also adds an echo adapter, Helm integration, and build validation.

Changes

Metering adapter runtime

Layer / File(s)	Summary
Adapter contracts and Kafka configuration osac-metering/adapters/adapter.go, osac-metering/adapters/kafka.go, osac-metering/adapters/go.mod, */Containerfile	Defines provider contracts, event and error types, Kafka TLS/SASL settings, SCRAM authentication, broker parsing, module dependencies, and container build inputs.
Deduplication and ordering state osac-metering/adapters/dedup.go, osac-metering/adapters/order.go, osac-metering/adapters/*_test.go	Adds synchronized CloudEvent deduplication and resource ordering trackers with TTL eviction and concurrency tests.
Submission retry behavior osac-metering/adapters/retry.go, osac-metering/adapters/retry_test.go	Adds jittered exponential backoff, retry limits, context cancellation, error classification, and retry behavior tests.
Prometheus metrics integration osac-metering/adapters/metrics.go, osac-metering/adapters/metrics_test.go	Adds metrics for submissions, retries, deduplication, ordering, lag, and flushing, plus an HTTP metrics handler.
Kafka consumer runner osac-metering/adapters/runner.go, osac-metering/adapters/runner_test.go	Adds consumer lifecycle management, CloudEvent processing, offset tracking and commits, adapter flushing, metrics recording, and end-to-end tests.
Echo adapter and event store osac-metering/adapters/cmd/echo-adapter/*, osac-metering/adapters/Containerfile.echo-adapter, .github/workflows/build-metering-echo-adapter-image.yaml	Adds the echo adapter binary, bounded event store, HTTP endpoints, container image build, and signal-aware shutdown.
Deployment and validation integration osac-metering/charts/osac-metering/*, .github/workflows/*, .pre-commit-config.yaml, osac-metering/adapters/.golangci.yml, osac-metering/adapters/Makefile, osac-metering/metering-service/Makefile	Updates Helm and VMaaS configuration for the echo adapter, adds adapter tests and linting, and removes the old test-adapter build target.

Estimated code review effort: 4 (Complex) | ~60 minutes

Sequence Diagram(s)

sequenceDiagram
  participant Kafka as Kafka consumer group
  participant Runner
  participant EchoAdapter
  participant EventStore
  participant Metrics as Prometheus metrics
  Kafka->>Runner: Deliver CloudEvent
  Runner->>Runner: Check duplicate and ordering state
  Runner->>EchoAdapter: Submit event with retry handling
  EchoAdapter->>EventStore: Store event
  Runner->>Metrics: Record processing outcome
  Runner->>Kafka: Mark processed offset
  Runner->>EchoAdapter: Flush buffered events
  Runner->>Kafka: Commit offsets

Loading

Possibly related PRs

• osac-project/osac#82: Adds Kafka and CloudEvent metering infrastructure extended by this adapters module.
• osac-project/osac#135: Introduces the test adapter that this PR replaces with the shared echo adapter.
• osac-project/osac#171: Adds the test-adapter image workflow that this PR changes to build the echo-adapter image.

Suggested labels: approved, jira/valid-reference

Suggested reviewers: akshaynadkarni, omer-vishlitzky, ajamias

---

Caution

Pre-merge checks failed

Please resolve all errors before merging. Addressing warnings is optional.

• Ignore

❌ Failed checks (1 error, 2 warnings)

Check name	Status	Explanation	Resolution
No-Sensitive-Data-In-Logs	❌ Error	The echo adapter logs raw Kafka broker addresses and topics, and logs every CloudEvent ID; these logs can expose internal hostnames and event/customer identifiers.	Do not log raw broker lists or event identifiers by default. Log redacted broker metadata and aggregate counters, or guard detailed event logging behind a secure, disabled-by-default debug setting.
Docstring Coverage	⚠️ Warning	Docstring coverage is 35.71% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.
Ai-Attribution	⚠️ Warning	The PR description mentions CodeRabbit, but all 13 PR commits contain only Signed-off-by trailers and no Assisted-by or Generated-by attribution.	Add an Assisted-by or Generated-by trailer naming the AI tool to each AI-assisted commit; do not use Co-Authored-By for AI tools.

✅ Passed checks (8 passed)

Check name	Status	Explanation
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.
No-Hardcoded-Secrets	✅ Passed	Scans found no vendor credentials, private keys, embedded-credential URLs, or literal credential assignments; SASL uses env/file inputs and Helm/workflows reference secrets indirectly.
No-Weak-Crypto	✅ Passed	Changed adapters code uses TLS 1.2 and SCRAM-SHA-512 only; searches found no MD5, SHA1, DES, RC4, Blowfish, ECB, custom crypto, or secret/token comparisons.
No-Injection-Vectors	✅ Passed	Changed files contain no SQL concatenation, unsafe deserialization, eval/exec, or os.system; the Kubernetes shell block uses quoted fixed configuration variables, not user input.
Container-Privileges	✅ Passed	Changed manifests set runAsNonRoot, drop all capabilities, and disable privilege escalation; the image uses USER 65532:65532, with no privileged, hostPID, hostNetwork, hostIPC, or SYS_ADMIN setting...
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly identifies the OSAC-3428 provider adapter framework and Kafka consumer lifecycle, which are the main changes in the pull request.

✨ Finishing Touches 💡 1

🛠️ Fix failing CI checks 💡

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[masayag]

Re-review — v3 (current HEAD d554f6d)

All critical and major findings from my previous reviews and from omer-vishlitzky's comments are resolved in this version. The branch was rebased so the referenced commit SHAs in earlier replies no longer exist, but the fixes are confirmed present in the current HEAD.

Resolved

Finding	Fix
ctx-cancellation data-loss regression	runner.go L290-300: else branch skips trackOffset + V(1) log; test added in runner_test.go
Consumer group ACL mismatch	echo-adapter-deployment.yaml now sets KAFKA_CONSUMER_GROUP=osac-metering-echo-adapter
/healthz hardcoded to 200	main.go now calls adapter.HealthCheck(r.Context()), returns 503 on failure
Broker addresses logged in clear text	Now logs broker_count=%d instead of actual addresses
mkdir -p bin missing in Makefile	Fixed
pre-commit hook file pattern too narrow	Now includes .golangci.yml, Makefile, go.{mod,sum}
values.yaml echo adapter defaults	Commented out in base values; environment-specific vmaas-ci/values.yaml provides the real config

Minor suggestions for follow-up (non-blocking)

1. newAdapterMetrics(provider string) — unused parameter (metrics.go L31). The provider arg is never referenced in the function body. Either remove it or use it in metric Help strings.

2. consumerLag gauge registered but never Set() (metrics.go L56-59). It appears as a permanent zero in /metrics output — either wire it from claim.HighWaterMarkOffset() - msg.Offset in ConsumeClaim, or drop it until a follow-up implements it.

3. SubmitResult from Flush() discarded (runner.go L174). The echo-adapter returns {Idempotent: true} but the Runner ignores it. Worth at least a debug-level log for future adapters that return a ProviderEventID.

4. echoAdapter.HealthCheck is a no-op (main.go L80) — always returns nil, so probes will always report healthy even if Kafka is unreachable. Fine for a smoke-test binary, but add a comment so real adapters don't copy the pattern.

5. RetryableError exported but functionally unused (adapter.go L48-52). The framework retries ALL non-NonRetryableError errors, so wrapping in RetryableError is optional. Document this in a doc.go or inline comment to avoid confusion for future adapter contributors.

6. No contributor README yet. A doc.go or README.md describing the Submit/Flush contract, error classification, and how to wire a new adapter binary would lower the bar for M360 and Cost Management implementations.

---

The framework is solid — offset tracking, dedup, retry, graceful shutdown, and the echo-adapter reference are all working correctly. Approving with the above as follow-up suggestions.

[openshift-ci]

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: amito, masayag

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Details

Needs approval from an approver in each of these files:

• OWNERS [amito,masayag]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[omer-vishlitzky]

/lgtm