Repository navigation
OSAC-3408: State Projection, Heartbeat Generator, and Reconciliation (Phase 2) - #135
Merged
omer-vishlitzky merged 12 commits intoAug 5, 2026
Conversation
|
@omer-vishlitzky: This pull request references OSAC-3408 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 epic to target the "5.0.0" version, but no target version was set. DetailsIn response to this:
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. |
omer-vishlitzky
had a problem deploying
to
e2e-test
August 4, 2026 11:05 — with 
GitHub Actions
Error
GitHub Actions
Error
omer-vishlitzky
had a problem deploying
to
e2e-test
August 4, 2026 11:05 — with
GitHub Actions
Error
GitHub Actions
Error
omer-vishlitzky
had a problem deploying
to
e2e-test
August 4, 2026 11:05 — with
GitHub Actions
Error
GitHub Actions
Error
Contributor
|
Note Reviews pausedIt 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 Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughThe metering service adds PostgreSQL-backed resource projections, state-aware event processing, reconciliation, heartbeat publication, Kafka topic routing, metrics, deployment wiring, and an optional Kafka test adapter. ChangesMetering platform
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related PRs
Suggested labels: Suggested reviewers: Caution Pre-merge checks failedPlease resolve all errors before merging. Addressing warnings is optional.
❌ Failed checks (2 errors, 1 warning)
✅ Passed checks (8 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
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. Comment |
omer-vishlitzky
force-pushed
the
feat/OSAC-3408-state-projection-heartbeat-reconciliation
branch
from
August 4, 2026 11:05
4a5cbb0 to
26a0e66
Compare
omer-vishlitzky
had a problem deploying
to
e2e-test
August 4, 2026 11:06 — with
GitHub Actions
Error
GitHub Actions
Error
omer-vishlitzky
had a problem deploying
to
e2e-test
August 4, 2026 11:06 — with
GitHub Actions
Error
GitHub Actions
Error
omer-vishlitzky
had a problem deploying
to
e2e-test
August 4, 2026 11:06 — with
GitHub Actions
Error
GitHub Actions
Error
omer-vishlitzky
force-pushed
the
feat/OSAC-3408-state-projection-heartbeat-reconciliation
branch
from
August 4, 2026 11:29
26a0e66 to
ecbe1b0
Compare
omer-vishlitzky
had a problem deploying
to
e2e-test
August 4, 2026 11:30 — with
GitHub Actions
Failure
GitHub Actions
Failure
omer-vishlitzky
temporarily deployed
to
e2e-test
August 4, 2026 11:30 — with
GitHub Actions
Inactive
GitHub Actions
Inactive
omer-vishlitzky
temporarily deployed
to
e2e-test
August 4, 2026 11:30 — with
GitHub Actions
Inactive
GitHub Actions
Inactive
omer-vishlitzky
force-pushed
the
feat/OSAC-3408-state-projection-heartbeat-reconciliation
branch
from
August 4, 2026 16:08
ecbe1b0 to
e08550f
Compare
omer-vishlitzky
had a problem deploying
to
e2e-test
August 4, 2026 16:09 — with
GitHub Actions
Error
GitHub Actions
Error
omer-vishlitzky
had a problem deploying
to
e2e-test
August 4, 2026 16:09 — with
GitHub Actions
Error
GitHub Actions
Error
omer-vishlitzky
had a problem deploying
to
e2e-test
August 4, 2026 16:09 — with
GitHub Actions
Error
GitHub Actions
Error
omer-vishlitzky
had a problem deploying
to
e2e-test
August 4, 2026 16:27 — with
GitHub Actions
Failure
GitHub Actions
Failure
omer-vishlitzky
temporarily deployed
to
e2e-test
August 4, 2026 16:27 — with
GitHub Actions
Inactive
GitHub Actions
Inactive
omer-vishlitzky
temporarily deployed
to
e2e-test
August 4, 2026 16:28 — with
GitHub Actions
Inactive
GitHub Actions
Inactive
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 18
🧹 Nitpick comments (14)
osac-metering/metering-service/internal/database/migrations/0_create_metering_resource_state.up.sql (1)
29-33: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueNo current query uses these two indexes.
idx_metering_resource_state_tenantandidx_metering_resource_state_resource_typehave no reader in the projection store.Getuses the primary key,ListBillableuses the partial billable index,ListAllis a full scan, andUpdateLastHeartbeatfilters onresource_id. Both indexes cost write amplification on every upsert and heartbeat batch with no read benefit today.Keep them only if a tenant-scoped or type-scoped query is landing in a near-term phase. Otherwise add them with the migration that introduces the query. The partial
is_billableindex on lines 25-27 is correct and matchesListBillable.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/database/migrations/0_create_metering_resource_state.up.sql` around lines 29 - 33, Remove the unused idx_metering_resource_state_tenant and idx_metering_resource_state_resource_type definitions from this migration; retain the partial is_billable index, and add the removed indexes only alongside a future migration that introduces matching tenant- or resource_type-scoped queries.osac-metering/metering-service/internal/projection/store.go (1)
20-27: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDocument the
(nil, nil)not-found contract onGet.
PostgresStore.Getreturns(nil, nil)when no row matches (postgres.golines 43-45). Nothing on this interface states that. A caller that checks onlyerrand then dereferences the result panics. ThisStoreis consumed by the reconciliation, heartbeat, and watch layers, so the contract needs to be readable at the interface.Add doc comments for the nil result and for
ErrStaleVersion.♻️ Proposed fix
type Store interface { + // Get returns the stored state for resourceID, or (nil, nil) when no row exists. Get(ctx context.Context, resourceID string) (*ResourceState, error) + // Upsert writes state. It returns ErrStaleVersion when the stored + // fulfillment_version is greater than state.FulfillmentVersion. Upsert(ctx context.Context, state ResourceState) error + // Delete removes the row for resourceID. It returns nil when no row exists. Delete(ctx context.Context, resourceID string) error ListBillable(ctx context.Context) ([]ResourceState, error) ListAll(ctx context.Context) ([]ResourceState, error) + // UpdateLastHeartbeat sets last_heartbeat_at to at for every listed resource. + // It ignores resource IDs with no matching row. UpdateLastHeartbeat(ctx context.Context, resourceIDs []string, at time.Time) error }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/projection/store.go` around lines 20 - 27, Document the Store interface contract around Get: state that a missing resource returns (nil, nil), and add documentation for ErrStaleVersion explaining its meaning. Place the comments next to Get and ErrStaleVersion, using the existing symbols without changing behavior.osac-metering/metering-service/internal/database/container.go (2)
266-280: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRetain the last connection error for the timeout message.
Line 275 discards every
pgx.Connecterror. If the password, port, or image is wrong, the caller sees only "timed out waiting for database connection" after 30 seconds and has no cause. Keep the last error and wrap it.♻️ Proposed fix
+ var lastErr error for { select { case <-ctx.Done(): return ctx.Err() case <-deadline: - return fmt.Errorf("timed out waiting for database connection") + return fmt.Errorf("timed out waiting for database connection: %w", lastErr) case <-ticker.C: conn, err := pgx.Connect(ctx, adminURL) if err != nil { + lastErr = err continue }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/database/container.go` around lines 266 - 280, Update the connection retry loop in the database container method to retain the most recent error from pgx.Connect, and when the deadline expires, wrap that error in the timeout message; preserve the existing context-cancellation behavior and successful connection assignment.
141-147: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winHonor
ctxwhen you kill the container.
Stopacceptsctxbut lines 141 and 146 ignore it.postgres_test.gopasses a 30-second context, and that deadline has no effect here. If the container tool hangs,Stopblocks forever and the suite never finishes.Use
exec.CommandContextso the kill respects the caller's deadline.♻️ Proposed fix
- killCmd := exec.Command(c.tool, "kill", c.id) + killCmd := exec.CommandContext(ctx, c.tool, "kill", c.id) if err := killCmd.Run(); err != nil { c.logger.Error(err, "killing database container") }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/database/container.go` around lines 141 - 147, Update the container stop logic in Stop to create killCmd with exec.CommandContext using the caller’s ctx instead of exec.Command, so both killing the container and its existing Run error handling honor the caller’s deadline.osac-metering/metering-service/internal/database/tool.go (1)
94-118: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winExtract the shared migrator setup.
Lines 95-118 duplicate lines 39-63 exactly: parse the URL, set the
pgx5scheme, build theiofsdriver, create the migrator, and defer the two-errorClose. Two copies means a future fix to the close handling lands in only one of them.Extract a helper and call it from both methods.
♻️ Proposed fix
func (t *Tool) newMigrate() (*migrate.Migrate, func(), error) { parsed, err := neturl.Parse(t.url) if err != nil { return nil, nil, fmt.Errorf("parsing database URL: %w", err) } parsed.Scheme = "pgx5" driver, err := iofs.New(migrationsFS, "migrations") if err != nil { return nil, nil, fmt.Errorf("loading migration files: %w", err) } migrations, err := migrate.NewWithSourceInstance("iofs", driver, parsed.String()) if err != nil { return nil, nil, fmt.Errorf("creating migration instance: %w", err) } closeFn := func() { sourceErr, databaseErr := migrations.Close() if sourceErr != nil { t.logger.Error(sourceErr, "closing migration source") } if databaseErr != nil { t.logger.Error(databaseErr, "closing migration database") } } return migrations, closeFn, nil }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/database/tool.go` around lines 94 - 118, Extract the duplicated migrator setup from MigrateToVersion and its counterpart into a shared Tool.newMigrate helper. Move URL parsing, the pgx5 scheme assignment, iofs driver creation, migrator construction, and Close error logging into that helper, returning the migrator, a cleanup function, and any setup error; update both methods to call it and defer the returned cleanup function.osac-metering/metering-service/internal/projection/postgres_test.go (1)
59-65: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winGive
newTestStorea bounded context.Line 60 uses
context.Background()with no deadline.inst.PoolrunsCREATE DATABASE ... TEMPLATE metering_template, and PostgreSQL blocks that statement while any other session is connected to the template database. A hungBeforeEachthen stalls the CI job with no error text.BeforeSuiteandAfterSuiteboth set explicit timeouts, so this helper is the inconsistent one.♻️ Proposed fix
-func newTestStore() (*projection.PostgresStore, *pgxpool.Pool) { - ctx := context.Background() +func newTestStore() (*projection.PostgresStore, *pgxpool.Pool) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() inst := dbContainer.NewInstance() pool, err := inst.Pool(ctx) Expect(err).ToNot(HaveOccurred()) return projection.NewPostgresStore(pool), pool }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/projection/postgres_test.go` around lines 59 - 65, Update newTestStore to use a bounded context with an explicit timeout, matching the timeout pattern already used by BeforeSuite and AfterSuite, and pass it to inst.Pool so database setup cannot hang indefinitely.osac-metering/metering-service/internal/projection/postgres.go (2)
133-160: 🚀 Performance & Scalability | 🔵 Trivial | 🏗️ Heavy liftBoth listings are unbounded, and
ListBillableruns every 60 seconds.Neither query has a
LIMITor a cursor.ListBillablematerializes every billable row, including thebilling_dimensionsJSON for each, and the heartbeat generator calls it on each 60-second tick.ListAllscans the whole table and the reconciler builds a full in-memory map from it every hour. Both grow linearly with the platform's billable resource count.The reconciler already paginates the fulfillment side, so the projection side is the asymmetric half of that comparison.
Two directions, either of which is a follow-up rather than a change for this PR:
- Add a batched or cursor-based variant to the
Storeinterface, keyed onresource_id, and have the heartbeat generator stream batches.- Push the stale-heartbeat filter into SQL. The reconciler currently reads every billable row and compares
LastHeartbeatAtin Go. AWHERE is_billable AND (last_heartbeat_at IS NULL OR last_heartbeat_at < $1)predicate returns only the rows that need work.The second option is the smaller change and removes most of the hourly volume.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/projection/postgres.go` around lines 133 - 160, The ListBillable and ListAll methods in PostgresStore perform unbounded full-table reads; update the projection listing flow to reduce hourly reconciliation volume by adding the stale-heartbeat predicate to ListBillable’s SQL, accepting the cutoff timestamp from the caller and returning only billable rows with a NULL or older last_heartbeat_at. Propagate the required Store interface and heartbeat/reconciler call-site changes while preserving existing resource-state collection behavior.
177-200: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winExtract the shared column list so the positional scan cannot drift.
scanResourceStatescans 12 columns by position. Three SELECT statements repeat that exact column list verbatim: lines 33-36, 135-138, and 150-153. If someone adds a column to one list and not the others, or reorders one, the compiler stays silent and the scan fails or mis-assigns at runtime.Declare the list once and reference it in all three queries.
♻️ Proposed fix
const resourceStateColumns = ` resource_id, resource_type, tenant_id, project_id, current_state, previous_state, is_billable, billable_since, last_heartbeat_at, transition_time, fulfillment_version, billing_dimensions`- row := s.pool.QueryRow(ctx, ` - SELECT resource_id, resource_type, tenant_id, project_id, - current_state, previous_state, is_billable, billable_since, - last_heartbeat_at, transition_time, fulfillment_version, - billing_dimensions - FROM metering_resource_state - WHERE resource_id = $1`, - resourceID) + row := s.pool.QueryRow(ctx, + `SELECT `+resourceStateColumns+` + FROM metering_resource_state + WHERE resource_id = $1`, + resourceID)Apply the same substitution in
ListBillableandListAll. The constant holds only compile-time literal text, so the queries stay parameterized.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/projection/postgres.go` around lines 177 - 200, Declare a shared compile-time resourceStateColumns constant containing the 12 selected columns, then replace the duplicated column lists in ListBillable, ListAll, and the other resource-state SELECT query with that constant. Keep the order aligned with scanResourceState’s row.Scan arguments and preserve existing query parameterization.osac-metering/metering-service/internal/database/database_test.go (1)
128-158: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winNo spec asserts that each
.up.sqlhas a matching.down.sql.The five specs check suffixes, duplicates, naming, the hash, and numbering gaps. None checks reversibility. A migration added with only an up file passes every spec, and the rollback path breaks at the worst possible moment.
knownGapson line 133 is an empty map. Keep it as the documented extension point, and add a pairing spec.💚 Proposed additional spec
It("Has a matching down migration for every up migration", func() { ups, err := filepath.Glob("migrations/*.up.sql") Expect(err).ToNot(HaveOccurred()) Expect(ups).ToNot(BeEmpty()) var missing []string for _, up := range ups { down := strings.TrimSuffix(up, ".up.sql") + ".down.sql" if _, err := os.Stat(down); err != nil { missing = append(missing, filepath.Base(down)) } } Expect(missing).To(BeEmpty(), "missing down migrations: %v", missing) })🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/database/database_test.go` around lines 128 - 158, Add a Ginkgo spec alongside “Has no unexpected gaps in migration numbering” that globs every migrations/*.up.sql, derives the corresponding .down.sql path, and collects missing rollback files before asserting none are missing. Preserve the existing knownGaps extension point and use os.Stat with clear failure output listing missing basenames.osac-metering/metering-service/internal/kafka/publisher_test.go (1)
92-107: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueMove this spec into the
TopicForblock and cover the remaining routes.The spec sits in
Describe("Publish")but callskafka.TopicForonly. It never invokespub.Publish, so its location is misleading. TheTopicForblock at Lines 191-197 is the right home.Two coverage gaps remain:
osac.inference.usage.v1is intopicRoutes(osac-metering/metering-service/internal/kafka/publisher.goLine 55) and has no spec.Publishwith an unrouted event type is untested. That path returns beforejson.Marshaland before any producer expectation is consumed, which is a distinct behavior fromTopicForreturning an error.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/kafka/publisher_test.go` around lines 92 - 107, The lifecycle routing spec should be moved from the Describe("Publish") block into the TopicFor block. Extend TopicFor coverage to include osac.inference.usage.v1, then add a Publish test for an unrouted event type that verifies the early-return behavior without requiring JSON marshaling or producer interaction.osac-metering/metering-service/internal/heartbeat/generator_test.go (2)
44-53: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
updateErris never exercised.
mockStore.updateErris declared and checked, but no spec sets it. TheUpdateLastHeartbeatfailure branch intick(osac-metering/metering-service/internal/heartbeat/generator.goLines 105-107) has no coverage. Add a spec that setsupdateErrand asserts the tick reports the failure, or remove the field.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/heartbeat/generator_test.go` around lines 44 - 53, Add a heartbeat generator spec that configures mockStore.updateErr before invoking tick and asserts the returned or reported result reflects the UpdateLastHeartbeat failure; use the existing mockStore and tick test patterns, or remove updateErr and its failure branch if that behavior is intentionally out of scope.
95-115: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winThe specs depend on wall-clock timing.
Each spec starts
Runwith a 100 ms interval and a 300 ms context timeout, then asserts afterRunreturns. The suite therefore spends about 300 ms per spec and assumes at least one tick lands before cancellation. Under a loaded runner the last tick can race the context deadline, becauseRunselects betweenctx.Done()andticker.C.Cancel from the publisher once the expected work is observed, and assert with
Eventually. The consumer suite already uses that pattern withmockPublisher.cancelFunc(osac-metering/metering-service/internal/watch/consumer_test.go).Also applies to: 150-169
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/heartbeat/generator_test.go` around lines 95 - 115, Update the heartbeat generator specs around gen.Run to avoid fixed wall-clock timeouts: provide the mockPublisher with a cancelFunc, have it cancel the context once the expected heartbeat publications are observed, and assert publication counts and types with Eventually. Apply the same pattern to the additional spec covering the referenced range, preserving each test’s expected work.osac-metering/metering-service/internal/events/mapper.go (1)
25-31: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winTwo billing-dimension accessors on one interface invite divergence.
BillingDimensions() anyandBillingDimensionsMap() map[string]stringdescribe the same source data in two shapes. The event path uses the first, the projection and heartbeat path uses the second. That is howboot_disk_size_gibends up as a number in lifecycle events and a string in heartbeat events (seeosac-metering/metering-service/internal/events/compute_instance.goLines 34-47).Prefer a single typed accessor and derive the map from it, or make both paths serialize identically.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/events/mapper.go` around lines 25 - 31, Unify the billing-dimension representation exposed by the event interface: replace the divergent BillingDimensions and BillingDimensionsMap usage with one typed accessor, then derive the map form for projection and heartbeat consumers. Update compute-instance lifecycle and heartbeat serialization to use the same conversion so fields such as boot_disk_size_gib have identical serialized types.osac-metering/metering-service/internal/events/compute_instance.go (1)
129-142: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueState literals duplicate the trimmed enum names.
resolveUpdatedEventTypecompares against"RUNNING","STOPPED","PAUSED", and"FAILED".CurrentState()produces these strings by trimmingcomputeInstanceStatePrefixfrom the protobuf enum name. A rename in the proto enum silently changes the classification with no compile error.Consider deriving the constants once from the enum values, for example
strings.TrimPrefix(privatev1.ComputeInstanceState_COMPUTE_INSTANCE_STATE_RUNNING.String(), computeInstanceStatePrefix).🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/events/compute_instance.go` around lines 129 - 142, Update resolveUpdatedEventType to compare CurrentState() against constants derived from the corresponding privatev1.ComputeInstanceState enum values by trimming computeInstanceStatePrefix, rather than hard-coded state literals. Apply this for RUNNING, STOPPED, PAUSED, and FAILED while preserving the existing event classification behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@osac-metering/charts/osac-metering/templates/deployment.yaml`:
- Around line 52-59: The default chart must require a database password Secret
because the application always expects DB_PASSWORD_FILE. In
osac-metering/charts/osac-metering/templates/deployment.yaml lines 52-59, use
Helm required for passwordSecret.name; at lines 106-110, always materialize the
mandatory Secret; and at lines 132-133, align the file path with that
unconditional Secret contract. In osac-metering/charts/osac-metering/values.yaml
lines 15-17, provide a valid default Secret name or make an empty value fail
rendering.
- Around line 106-108: Update the secret lookup in the deployment script around
DB_SECRET_KEY to use a key-safe Kubernetes JSONPath expression, preserving
configured keys containing dots such as db.password as literal data keys. Escape
the key or use bracket notation before passing it to oc get secret, while
keeping the existing decoding and password-file output unchanged.
In `@osac-metering/metering-service/cmd/metering-service/main.go`:
- Around line 150-158: Update the interval validation in the configuration
validation flow around heartbeatInterval and reconciliationInterval to parse
each duration, reject values less than or equal to zero, and return the existing
variable-specific validation errors. Add tests covering 0s and -1s for both
HEARTBEAT_INTERVAL and RECONCILIATION_INTERVAL.
- Around line 163-168: Update both dbConfig.poolConfig() and migrationURL() to
build PostgreSQL DSNs with net/url.URL, url.UserPassword, and net.JoinHostPort
instead of interpolating raw values; add sslrootcert via Query().Set so caCert
paths are encoded correctly. Add tests covering passwords containing reserved
URI characters and verify both pool and migration URLs preserve the intended
credentials, endpoint, database, and certificate path.
In `@osac-metering/metering-service/internal/database/container.go`:
- Around line 159-186: Add a mutex field to the Container struct to protect
shared state, then acquire this Container-level lock in NewInstance before
appending to c.instances, and in the init method before incrementing
i.container.count and generating the database name. This ensures concurrent init
calls and NewInstance calls do not race on the shared count or instances slice.
As a secondary refinement, update the database name prefix in the init method
from "metering_template%d" to "metering_test%d" to better reflect that instances
are test clones rather than actual templates.
In `@osac-metering/metering-service/internal/database/database_test.go`:
- Around line 104-125: Update the migration hash calculation in the database
migration integrity spec to read each migration file and hash its filename
together with its contents, preserving deterministic sorted ordering. Replace
the manual Fail in the computedHashText mismatch branch with an Expect assertion
that reports both the stored and computed hashes. Recompute migrations.sha256
using the corresponding content-inclusive scheme.
In `@osac-metering/metering-service/internal/database/tool.go`:
- Around line 38-44: Update Tool.Migrate to honor its ctx parameter by starting
cancellation handling that sends true to the migration container’s GracefulStop
channel when ctx.Done() closes, ensuring migration operations can be cancelled.
Keep the existing migration flow intact and avoid leaking the cancellation
goroutine after migration completes.
In `@osac-metering/metering-service/internal/events/compute_instance.go`:
- Around line 34-47: Unify billing_dimensions serialization through one
canonical typed accessor: in
osac-metering/metering-service/internal/events/mapper.go:25-31, collapse
ResourceMapper’s duplicate accessors and derive any map form from the typed
value; in
osac-metering/metering-service/internal/events/compute_instance.go:34-47, update
computeInstanceMapper.BillingDimensionsMap to use that canonical representation
rather than fmt.Sprintf stringification, with absent dimensions consistently
omitted or represented as null.
In `@osac-metering/metering-service/internal/heartbeat/generator.go`:
- Around line 146-164: Update Generator.updateGauges to reset both
projectionResources and heartbeatLag at the start of every tick before
calculating current values, then retain the existing writes for resource types
in billable. Ensure resource types absent from the current slice expose zero
values rather than stale metrics.
- Around line 113-144: Update Generator.buildHeartbeatEvent to return both the
CloudEvent and the error from ce.SetData instead of discarding it. Propagate
that error to the caller at the heartbeat tick loop, ensuring the resource tick
is skipped and the error is returned when event serialization fails.
- Around line 91-107: When a publish error occurs in the loop over billable
resources, the function returns immediately without updating the heartbeat
timestamps for the IDs that were successfully published. To prevent duplicate
heartbeats on the next tick, capture the publishedIDs accumulated before the
error and call UpdateLastHeartbeat to checkpoint that progress before returning
the publish error from the Publish call on billable[i]. Preserve the current
flow where UpdateLastHeartbeat is called after all publishes succeed.
In `@osac-metering/metering-service/internal/kafka/publisher.go`:
- Around line 46-56: Update NewPublisher and the publisher routing flow to
accept an injected configurable topic route table or topic resolver, using it
when selecting destinations while retaining topicRoutes as the default when none
is provided. Preserve the existing event-type mappings and update construction
sites to pass environment or tenant-specific configuration where available.
In `@osac-metering/metering-service/internal/projection/postgres.go`:
- Around line 82-121: Update Upsert in
osac-metering/metering-service/internal/projection/postgres.go:82-121 to retain
the tx.Exec command tag and return ErrStaleVersion when RowsAffected() is zero,
matching the existing contract at line 78. Update the concurrency spec in
osac-metering/metering-service/internal/projection/postgres_test.go:285-296 to
require any returned error to be ErrStaleVersion, assert the final
FulfillmentVersion equals 3, and add coverage racing two upserts for a
resource_id with no existing row.
In `@osac-metering/metering-service/internal/reconciliation/reconciler.go`:
- Around line 184-194: After a successful publisher.Publish in the
stale-heartbeat branch of the reconciliation loop, call
Store.UpdateLastHeartbeat for the resource using the publication timestamp, and
propagate any update error instead of treating reconciliation as successful. Add
coverage that runs reconciliation twice and verifies the second pass emits no
synthetic heartbeat.
- Around line 125-127: The reconciliation branches around r.store.Upsert must
propagate projection mutation errors instead of logging and continuing to report
success; apply this consistently to the referenced correction paths as well.
Return the error from the reconciliation operation, or otherwise prevent
success/readiness signaling until the durable projection update succeeds, while
preserving deterministic idempotency if publication remains publish-first.
- Around line 131-157: Update the reconciliation logic around the state-drift
branch and FulfillmentVersion handling to reject fulfillment records older than
the projection, preventing rollback; when fs.version is newer, persist it even
if fs.state is unchanged so stale watch events remain rejected. Preserve
existing state-transition and billing updates for valid changes, and add tests
covering both older-fulfillment rejection and newer-version persistence.
In `@osac-metering/metering-service/internal/watch/consumer_test.go`:
- Around line 217-223: The newConsumerWithStore helper is unused and should not
be retained via `_ =`. Remove the helper and discard statement, or replace them
with focused specs that use `watch.NewConsumer` with a seeded store to cover
same-state-and-dimensions skipping, ErrStaleVersion skipping, and the
DELETED-then-Delete path.
In `@osac-metering/metering-service/internal/watch/consumer.go`:
- Around line 147-171: Update the projection construction in the consumer to use
the event-derived transitionTime for both ResourceState.TransitionTime and the
initial BillableSince value. Preserve existing BillableSince when the resource
is already billable, while replacing the ingestion-time now value for newly
billable resources.
---
Nitpick comments:
In `@osac-metering/metering-service/internal/database/container.go`:
- Around line 266-280: Update the connection retry loop in the database
container method to retain the most recent error from pgx.Connect, and when the
deadline expires, wrap that error in the timeout message; preserve the existing
context-cancellation behavior and successful connection assignment.
- Around line 141-147: Update the container stop logic in Stop to create killCmd
with exec.CommandContext using the caller’s ctx instead of exec.Command, so both
killing the container and its existing Run error handling honor the caller’s
deadline.
In `@osac-metering/metering-service/internal/database/database_test.go`:
- Around line 128-158: Add a Ginkgo spec alongside “Has no unexpected gaps in
migration numbering” that globs every migrations/*.up.sql, derives the
corresponding .down.sql path, and collects missing rollback files before
asserting none are missing. Preserve the existing knownGaps extension point and
use os.Stat with clear failure output listing missing basenames.
In
`@osac-metering/metering-service/internal/database/migrations/0_create_metering_resource_state.up.sql`:
- Around line 29-33: Remove the unused idx_metering_resource_state_tenant and
idx_metering_resource_state_resource_type definitions from this migration;
retain the partial is_billable index, and add the removed indexes only alongside
a future migration that introduces matching tenant- or resource_type-scoped
queries.
In `@osac-metering/metering-service/internal/database/tool.go`:
- Around line 94-118: Extract the duplicated migrator setup from
MigrateToVersion and its counterpart into a shared Tool.newMigrate helper. Move
URL parsing, the pgx5 scheme assignment, iofs driver creation, migrator
construction, and Close error logging into that helper, returning the migrator,
a cleanup function, and any setup error; update both methods to call it and
defer the returned cleanup function.
In `@osac-metering/metering-service/internal/events/compute_instance.go`:
- Around line 129-142: Update resolveUpdatedEventType to compare CurrentState()
against constants derived from the corresponding privatev1.ComputeInstanceState
enum values by trimming computeInstanceStatePrefix, rather than hard-coded state
literals. Apply this for RUNNING, STOPPED, PAUSED, and FAILED while preserving
the existing event classification behavior.
In `@osac-metering/metering-service/internal/events/mapper.go`:
- Around line 25-31: Unify the billing-dimension representation exposed by the
event interface: replace the divergent BillingDimensions and
BillingDimensionsMap usage with one typed accessor, then derive the map form for
projection and heartbeat consumers. Update compute-instance lifecycle and
heartbeat serialization to use the same conversion so fields such as
boot_disk_size_gib have identical serialized types.
In `@osac-metering/metering-service/internal/heartbeat/generator_test.go`:
- Around line 44-53: Add a heartbeat generator spec that configures
mockStore.updateErr before invoking tick and asserts the returned or reported
result reflects the UpdateLastHeartbeat failure; use the existing mockStore and
tick test patterns, or remove updateErr and its failure branch if that behavior
is intentionally out of scope.
- Around line 95-115: Update the heartbeat generator specs around gen.Run to
avoid fixed wall-clock timeouts: provide the mockPublisher with a cancelFunc,
have it cancel the context once the expected heartbeat publications are
observed, and assert publication counts and types with Eventually. Apply the
same pattern to the additional spec covering the referenced range, preserving
each test’s expected work.
In `@osac-metering/metering-service/internal/kafka/publisher_test.go`:
- Around line 92-107: The lifecycle routing spec should be moved from the
Describe("Publish") block into the TopicFor block. Extend TopicFor coverage to
include osac.inference.usage.v1, then add a Publish test for an unrouted event
type that verifies the early-return behavior without requiring JSON marshaling
or producer interaction.
In `@osac-metering/metering-service/internal/projection/postgres_test.go`:
- Around line 59-65: Update newTestStore to use a bounded context with an
explicit timeout, matching the timeout pattern already used by BeforeSuite and
AfterSuite, and pass it to inst.Pool so database setup cannot hang indefinitely.
In `@osac-metering/metering-service/internal/projection/postgres.go`:
- Around line 133-160: The ListBillable and ListAll methods in PostgresStore
perform unbounded full-table reads; update the projection listing flow to reduce
hourly reconciliation volume by adding the stale-heartbeat predicate to
ListBillable’s SQL, accepting the cutoff timestamp from the caller and returning
only billable rows with a NULL or older last_heartbeat_at. Propagate the
required Store interface and heartbeat/reconciler call-site changes while
preserving existing resource-state collection behavior.
- Around line 177-200: Declare a shared compile-time resourceStateColumns
constant containing the 12 selected columns, then replace the duplicated column
lists in ListBillable, ListAll, and the other resource-state SELECT query with
that constant. Keep the order aligned with scanResourceState’s row.Scan
arguments and preserve existing query parameterization.
In `@osac-metering/metering-service/internal/projection/store.go`:
- Around line 20-27: Document the Store interface contract around Get: state
that a missing resource returns (nil, nil), and add documentation for
ErrStaleVersion explaining its meaning. Place the comments next to Get and
ErrStaleVersion, using the existing symbols without changing behavior.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository: osac-project/coderabbit/.coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 861993ed-1caf-47cd-b9d1-87fefaf54166
⛔ Files ignored due to path filters (2)
go.work.sumis excluded by!**/*.sumosac-metering/metering-service/go.sumis excluded by!**/*.sum
📒 Files selected for processing (30)
osac-metering/charts/osac-metering/templates/deployment.yamlosac-metering/charts/osac-metering/values.yamlosac-metering/metering-service/cmd/metering-service/main.goosac-metering/metering-service/go.modosac-metering/metering-service/internal/database/container.goosac-metering/metering-service/internal/database/database_suite_test.goosac-metering/metering-service/internal/database/database_test.goosac-metering/metering-service/internal/database/migrations.sha256osac-metering/metering-service/internal/database/migrations/0_create_metering_resource_state.down.sqlosac-metering/metering-service/internal/database/migrations/0_create_metering_resource_state.up.sqlosac-metering/metering-service/internal/database/tool.goosac-metering/metering-service/internal/events/compute_instance.goosac-metering/metering-service/internal/events/mapper.goosac-metering/metering-service/internal/events/mapper_test.goosac-metering/metering-service/internal/heartbeat/generator.goosac-metering/metering-service/internal/heartbeat/generator_test.goosac-metering/metering-service/internal/heartbeat/heartbeat_suite_test.goosac-metering/metering-service/internal/kafka/publisher.goosac-metering/metering-service/internal/kafka/publisher_test.goosac-metering/metering-service/internal/projection/postgres.goosac-metering/metering-service/internal/projection/postgres_test.goosac-metering/metering-service/internal/projection/projection_suite_test.goosac-metering/metering-service/internal/projection/store.goosac-metering/metering-service/internal/projection/types.goosac-metering/metering-service/internal/reconciliation/correction.goosac-metering/metering-service/internal/reconciliation/reconciler.goosac-metering/metering-service/internal/reconciliation/reconciler_test.goosac-metering/metering-service/internal/reconciliation/reconciliation_suite_test.goosac-metering/metering-service/internal/watch/consumer.goosac-metering/metering-service/internal/watch/consumer_test.go
Comment on lines
+52
to
+59
| {{- if .Values.database.passwordSecret.name }} | ||
| - name: DB_SECRET_NAME | ||
| value: {{ .Values.database.passwordSecret.name }} | ||
| - name: DB_SECRET_KEY | ||
| value: {{ .Values.database.passwordSecret.key | default "password" }} | ||
| - name: DB_SECRET_NS | ||
| value: {{ .Release.Namespace }} | ||
| {{- end }} |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Require a database password Secret for the default chart.
passwordSecret.name defaults to empty. The init container then skips the password file, while the application always receives DB_PASSWORD_FILE. Startup fails when poolConfig reads the missing file.
osac-metering/charts/osac-metering/templates/deployment.yaml#L52-L59: use Helmrequiredfor the Secret name instead of optional configuration.osac-metering/charts/osac-metering/templates/deployment.yaml#L106-L110: always materialize the mandatory password Secret.osac-metering/charts/osac-metering/templates/deployment.yaml#L132-L133: align the unconditional file path with the mandatory Secret contract.osac-metering/charts/osac-metering/values.yaml#L15-L17: provide a valid default Secret name or make the blank value fail rendering.
📍 Affects 2 files
osac-metering/charts/osac-metering/templates/deployment.yaml#L52-L59(this comment)osac-metering/charts/osac-metering/templates/deployment.yaml#L106-L110osac-metering/charts/osac-metering/templates/deployment.yaml#L132-L133osac-metering/charts/osac-metering/values.yaml#L15-L17
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@osac-metering/charts/osac-metering/templates/deployment.yaml` around lines 52
- 59, The default chart must require a database password Secret because the
application always expects DB_PASSWORD_FILE. In
osac-metering/charts/osac-metering/templates/deployment.yaml lines 52-59, use
Helm required for passwordSecret.name; at lines 106-110, always materialize the
mandatory Secret; and at lines 132-133, align the file path with that
unconditional Secret contract. In osac-metering/charts/osac-metering/values.yaml
lines 15-17, provide a valid default Secret name or make an empty value fail
rendering.
Comment on lines
+106
to
+108
| if [ -n "${DB_SECRET_NAME:-}" ]; then | ||
| wait_for_secret "${DB_SECRET_NAME}" "${DB_SECRET_NS}" | ||
| oc get secret "${DB_SECRET_NAME}" -n "${DB_SECRET_NS}" -o jsonpath="{.data.${DB_SECRET_KEY}}" | base64 -d > /etc/db-secrets/password |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== locate deployment files =="
fd -a 'deployment\.ya?ml$' . || true
echo "== relevant deployment section =="
if [ -f osac-metering/charts/osac-metering/templates/deployment.yaml ]; then
sed -n '1,180p' osac-metering/charts/osac-metering/templates/deployment.yaml | cat -n
fi
echo "== search references and values =="
rg -n "DB_SECRET_KEY|DB_SECRET_NAME|DB_SECRET_NS|wait_for_secret|osac-metering" osac-metering/charts osac-metering -S || trueRepository: osac-project/osac
Length of output: 2020
🌐 Web query:
Kubernetes kubectl jsonpath dot field separator documentation
💡 Result:
In Kubernetes kubectl, the dot (.) character functions as a child operator when used in JSONPath expressions, indicating nested fields in a JSON object [1]. Because it is a reserved operator, dot notation cannot be used directly to access keys that contain dots (e.g., labels, annotations, or secret data) [2][3]. To handle fields with dot separators or special characters, you should use one of the following approaches: 1. Bracket Notation (Recommended): Use square brackets with the key name enclosed in quotes. This treats the entire string inside the brackets as a literal key [2][3]. Example: kubectl get pod my-pod -o jsonpath='{.metadata.labels["app.kubernetes.io/name"]}' [3] 2. Escaping the Dot: You can escape the dot within a dot-separated path by preceding it with a backslash () [4][5]. Note that when using this method, the backslash itself may need to be escaped depending on your shell environment [6][4]. Example: kubectl get secret mysecret -o jsonpath='{.data..dockerconfigjson}' [5] kubectl get pods -o jsonpath='{.items.metadata.labels.kubernetes.io/hostname}' [1] Summary of syntax options: - Standard fields: Use dot notation (e.g., {.metadata.name}) [2]. - Complex keys (dots/slashes): Use bracket notation (e.g., {.metadata.labels['key.with.dot']}) [2][3]. - Escaping: Use a backslash to treat a dot as a character rather than an operator [1][4]. When working in environments like Windows or when your template contains spaces, ensure you use proper quoting (e.g., double quotes for the template and single quotes for internal literals) to avoid shell parsing errors [6].
Citations:
- 1: https://kubernetes.io/docs/reference/kubectl/jsonpath/
- 2: https://codemia.io/knowledge-hub/path/kubectl_jsonpath_expression_for_named_path
- 3: https://codemia.io/knowledge-hub/path/how_to_access_key_in_a_map_returned_by_kubectl
- 4: JSONpath fails to return keys containing dots in a map kubernetes/kubernetes#23386
- 5: https://www.alexdyas.com/technical/Kubectl-jsonpath-escaping-periods/
- 6: https://github.com/kubernetes/website/blob/master/content/en/docs/reference/kubectl/jsonpath.md
Read Secret keys with a key-safe JSONPath expression.
Kubernetes JSONPath treats . as a child operator, so a configured key like db.password is parsed as nested fields instead of one secret data key. Use bracket notation or escape the key before passing it as the JSONPath path.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@osac-metering/charts/osac-metering/templates/deployment.yaml` around lines
106 - 108, Update the secret lookup in the deployment script around
DB_SECRET_KEY to use a key-safe Kubernetes JSONPath expression, preserving
configured keys containing dots such as db.password as literal data keys. Escape
the key or use bracket notation before passing it to oc get secret, while
keeping the existing decoding and password-file output unchanged.
omer-vishlitzky
force-pushed
the
feat/OSAC-3408-state-projection-heartbeat-reconciliation
branch
from
August 4, 2026 18:49
59bd491 to
40607fd
Compare
omer-vishlitzky
had a problem deploying
to
e2e-test
August 4, 2026 18:49 — with
GitHub Actions
Failure
GitHub Actions
Failure
omer-vishlitzky
had a problem deploying
to
e2e-test
August 4, 2026 18:49 — with
GitHub Actions
Failure
GitHub Actions
Failure
omer-vishlitzky
had a problem deploying
to
e2e-test
August 4, 2026 18:49 — with
GitHub Actions
Failure
GitHub Actions
Failure
Contributor
There was a problem hiding this comment.
Actionable comments posted: 20
♻️ Duplicate comments (1)
osac-metering/metering-service/internal/reconciliation/reconciler.go (1)
125-127: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
RunFullreturns success after a projection mutation fails. All three correction branches publish the correction, log the durable-state failure, and letRunFullreturnnilon line 214. Startup readiness therefore opens while the projection stays stale, and the next pass republishes the same logical correction with a new event ID, which blocks downstream deduplication. Every branch already returns onPublishfailure, so the mutation failure is the only silent path.
osac-metering/metering-service/internal/reconciliation/reconciler.go#L125-L127: return thestore.Upserterror for theMissedCreationbranch instead of logging it.osac-metering/metering-service/internal/reconciliation/reconciler.go#L154-L156: return thestore.Upserterror for theStateDriftbranch instead of logging it.osac-metering/metering-service/internal/reconciliation/reconciler.go#L170-L172: return thestore.Deleteerror for theMissedDeletionbranch instead of logging it.If publish-first ordering must stay, use an outbox or a deterministic idempotency key derived from the resource ID, reason, and version so a retried pass emits the same event ID.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/reconciliation/reconciler.go` around lines 125 - 127, Ensure RunFull propagates projection mutation failures instead of returning success: in osac-metering/metering-service/internal/reconciliation/reconciler.go lines 125-127, return the MissedCreation store.Upsert error; lines 154-156, return the StateDrift store.Upsert error; and lines 170-172, return the MissedDeletion store.Delete error, replacing the current log-only handling. Preserve existing publish-failure behavior; if publish-first ordering remains, use an outbox or deterministic idempotency key based on resource ID, reason, and version.
🧹 Nitpick comments (1)
osac-metering/metering-service/internal/database/container.go (1)
91-99: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueSanitize interpolated identifiers in the DDL statements.
The Go path instruction requires
database/sqlplaceholders and nofmt.Sprintfin queries. PostgreSQL does not accept placeholders for identifiers inCREATE USER,CREATE DATABASE, orALTER DATABASE, so the interpolation cannot be removed here. The interpolated values are thecontainerUserconstant, a locally generated password, and internally generated names, so no external input reaches these statements.Use
pgx.Identifier{...}.Sanitize()for the identifier positions. This keeps the intent explicit and silences the repeated static analysis warnings on lines 91, 98, 113, 138, 185, and 218.The password on line 92 still needs literal quoting.
ALTER ROLE ... PASSWORDalso takes no placeholder, so keep that value generated internally and never derived from input.♻️ Example for the database creation calls
- _, err = c.adminConn.Exec(ctx, - fmt.Sprintf("CREATE DATABASE %s OWNER %s", templateDB, containerUser)) + _, err = c.adminConn.Exec(ctx, + fmt.Sprintf("CREATE DATABASE %s OWNER %s", + pgx.Identifier{templateDB}.Sanitize(), + pgx.Identifier{containerUser}.Sanitize()))🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/database/container.go` around lines 91 - 99, Replace raw identifier interpolation in the DDL statements within the container database setup flow with pgx.Identifier{...}.Sanitize(), covering the CREATE USER, CREATE DATABASE, ALTER DATABASE, and related statements referenced by the warnings. Keep the internally generated password as a literal quoted value where PostgreSQL disallows placeholders, and apply the same sanitization consistently to every interpolated identifier without changing the existing execution flow.Sources: Path instructions, Linters/SAST tools
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@osac-metering/metering-service/cmd/metering-service/main.go`:
- Around line 250-258: Track the heartbeat generator and periodic reconciler
launched near hbGen.Run and reconciler.RunPeriodic with an errgroup or
sync.WaitGroup, wait for both to finish after cancellation, then close the
producer in run. Handle and propagate or log the producer.Close error instead of
discarding it. Apply this in
osac-metering/metering-service/cmd/metering-service/main.go at lines 250-258 and
214-218.
- Around line 145-150: Update readDBURL to trim the file contents and reject an
empty trimmed value by returning an appropriate error before returning the URL;
continue returning the trimmed value for non-empty content.
In `@osac-metering/metering-service/cmd/test-adapter/main.go`:
- Line 28: Replace the count-only maxEvents limit with a total retained-byte
budget below the 128 MiB container limit, accounting for event and slice/runtime
overhead. Update the event ingestion flow to reject or truncate individual
records that exceed the per-record allowance, evict or stop retaining events
when the aggregate budget is reached, and bound the /events response payload
independently. Apply the same byte-budget logic to the related code at the
additional referenced section.
- Around line 87-95: Update consumerGroupHandler.Setup, Cleanup, and
ConsumeClaim in osac-metering/metering-service/cmd/test-adapter/main.go:87-95 to
track the active session and stop consumption when the session context ends. In
osac-metering/metering-service/cmd/test-adapter/main.go:139-149, wrap
non-context group.Consume errors in a capped backoff before retrying, while
preserving immediate context-cancellation termination. In
osac-metering/metering-service/cmd/test-adapter/main.go:151-155, keep /healthz
process-only and add /readyz to report consumer readiness; update readinessProbe
in
osac-metering/charts/osac-metering/templates/test-adapter-deployment.yaml:112-123
to call /readyz.
- Line 134: Update the server lifecycle around srv.ListenAndServe and the signal
handler so shutdown is coordinated and awaited on the main path rather than
performed asynchronously. Preserve signal-triggered shutdown, but capture and
check both ListenAndServe and srv.Shutdown errors, allowing the main flow to
wait for shutdown completion before closing group.
- Around line 156-159: Restrict the /events handler in main.go before reading
query parameters or encoding events by validating the intended test caller and
namespace, returning an appropriate denial response for unauthorized requests;
update the handler’s event-encoding path accordingly. In main.go lines 163-168,
ensure json.NewEncoder(w).Encode(events) is reachable only after this
authorization check. In
osac-metering/charts/osac-metering/templates/test-adapter-service.yaml lines
10-18, configure the Service or related access controls to enforce the same
intended caller/namespace restriction.
- Around line 160-162: Validate the RFC3339Nano parse result in the request
handler’s since query-parameter block before calling store.query. When a
non-empty since value is invalid, return HTTP 400 Bad Request immediately and do
not execute the query; preserve the existing behavior for valid or omitted since
values.
In `@osac-metering/metering-service/Containerfile.test-adapter`:
- Around line 13-17: Add a final-image HEALTHCHECK to the Containerfile after
the test-adapter image setup, targeting the adapter’s /healthz endpoint and
using the container’s available health-check tooling. Ensure the resulting image
declares the check explicitly, rather than relying only on Kubernetes probes.
In `@osac-metering/metering-service/internal/projection/postgres_test.go`:
- Around line 59-63: Update newTestStore to create a bounded context.Context
with a timeout before calling inst.Pool, and ensure the timeout cancellation is
released appropriately. Pass this timed context to pool creation so database
failures cannot stall the tests.
- Around line 147-158: The same-version upsert test incorrectly allows a changed
payload to overwrite the existing projection. Update the `Allows idempotent
upsert with same version` test and the `Upsert` behavior so equal-version writes
are no-ops for identical payloads and changed payloads are rejected or otherwise
preserve the stored state; assert the chosen conflict/unchanged outcome and
ensure `CurrentState` remains `RUNNING`.
- Around line 51-56: Update the AfterSuite cleanup around dbContainer.Stop to
handle its returned error instead of discarding it. Preserve the existing
timeout and cancellation flow, and report or propagate shutdown failures using
the test suite’s established error-handling mechanism.
- Around line 285-295: Update the final FulfillmentVersion assertion in the
concurrent update test to require exact equality with version 3 rather than
accepting any value greater than or equal to 2. Keep the existing success-count
validation and Get call unchanged.
In `@osac-metering/metering-service/internal/reconciliation/reconciler_test.go`:
- Line 94: Update mockStore.UpdateLastHeartbeat to persist the provided
timestamp in LastHeartbeatAt for every resource ID, then extend the
stale-heartbeat test to assert res-stale.LastHeartbeatAt advances after
Reconciler.RunFull. Keep the existing success behavior while ensuring the mock
records each heartbeat update.
- Around line 317-338: Update the no-op fixture in the “emits no corrections
when states match” test to mark the projection as billable and assign a recent
LastHeartbeatAt before running recon.RunFull(ctx). Keep the matching RUNNING
state and empty publication assertion unchanged so the test exercises the
healthy billable-resource path.
In `@osac-metering/metering-service/internal/reconciliation/reconciler.go`:
- Around line 318-333: Update buildSyntheticHeartbeat to return both the
CloudEvent and the error from ce.SetData instead of discarding it. At its call
site, handle the returned error by logging it with the resource ID and
continuing to the next resource, preventing publication and heartbeat updates
when construction fails.
In `@osac-metering/metering-service/internal/watch/consumer.go`:
- Around line 180-196: The deletion branch in the consumer flow must validate
the incoming fulfillment version before removing state. Update the store
operation used by the OBJECT_DELETED handling around c.store.Delete to perform
the version check and deletion atomically, return or recognize
projection.ErrStaleVersion without publishing or treating it as a successful
deletion, and preserve the existing error propagation for other failures.
- Around line 123-147: Update the event-processing flow before buildStateContext
to override isBillable as false when event.GetType() is OBJECT_DELETED,
regardless of mapper.IsBillable(). Add a test covering deletion of a running
resource and assert that the resulting state includes the closing duration.
- Around line 134-140: Update the watch consumer’s projection mutation and event
publication flow, including the unchanged-state shortcut and deletion handling
around publishWithRetry, to persist a pending event in the same transaction as
each projection change. Retry pending outbox records until Kafka acknowledges
them, ensuring replayed updates and deletions are still delivered after
publishWithRetry exhaustion, and add tests covering both failure-and-replay
scenarios.
- Around line 134-140: Update the unchanged-update branch in the event consumer
to compare normalized ProjectID alongside state and billing dimensions before
suppressing the event. For suppressed updates, atomically persist a higher
FulfillmentVersion while retaining the existing projection data, and preserve
stale-version protection when the incoming version is not higher. Add regression
coverage for version 3 following version 1 and for project-only changes.
In `@osac-metering/metering-service/Makefile`:
- Around line 21-22: Declare the build-test-adapter target as phony by adding it
to the Makefile’s .PHONY declaration, preserving the existing target command.
---
Duplicate comments:
In `@osac-metering/metering-service/internal/reconciliation/reconciler.go`:
- Around line 125-127: Ensure RunFull propagates projection mutation failures
instead of returning success: in
osac-metering/metering-service/internal/reconciliation/reconciler.go lines
125-127, return the MissedCreation store.Upsert error; lines 154-156, return the
StateDrift store.Upsert error; and lines 170-172, return the MissedDeletion
store.Delete error, replacing the current log-only handling. Preserve existing
publish-failure behavior; if publish-first ordering remains, use an outbox or
deterministic idempotency key based on resource ID, reason, and version.
---
Nitpick comments:
In `@osac-metering/metering-service/internal/database/container.go`:
- Around line 91-99: Replace raw identifier interpolation in the DDL statements
within the container database setup flow with pgx.Identifier{...}.Sanitize(),
covering the CREATE USER, CREATE DATABASE, ALTER DATABASE, and related
statements referenced by the warnings. Keep the internally generated password as
a literal quoted value where PostgreSQL disallows placeholders, and apply the
same sanitization consistently to every interpolated identifier without changing
the existing execution flow.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository: osac-project/coderabbit/.coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: a76a3d56-c30c-434c-9abd-54df367a88d6
⛔ Files ignored due to path filters (2)
go.work.sumis excluded by!**/*.sumosac-metering/metering-service/go.sumis excluded by!**/*.sum
📒 Files selected for processing (40)
osac-installer/charts/osac/templates/bundled-postgres.yamlosac-installer/charts/osac/values.schema.jsonosac-installer/values/vmaas-ci/values.yamlosac-metering/charts/osac-metering/templates/deployment.yamlosac-metering/charts/osac-metering/templates/test-adapter-deployment.yamlosac-metering/charts/osac-metering/templates/test-adapter-kafka-user.yamlosac-metering/charts/osac-metering/templates/test-adapter-service.yamlosac-metering/charts/osac-metering/values.yamlosac-metering/metering-service/Containerfile.test-adapterosac-metering/metering-service/Makefileosac-metering/metering-service/cmd/metering-service/main.goosac-metering/metering-service/cmd/test-adapter/main.goosac-metering/metering-service/go.modosac-metering/metering-service/internal/database/container.goosac-metering/metering-service/internal/database/database_suite_test.goosac-metering/metering-service/internal/database/database_test.goosac-metering/metering-service/internal/database/migrations.sha256osac-metering/metering-service/internal/database/migrations/0_create_metering_resource_state.down.sqlosac-metering/metering-service/internal/database/migrations/0_create_metering_resource_state.up.sqlosac-metering/metering-service/internal/database/tool.goosac-metering/metering-service/internal/events/compute_instance.goosac-metering/metering-service/internal/events/mapper.goosac-metering/metering-service/internal/events/mapper_test.goosac-metering/metering-service/internal/heartbeat/generator.goosac-metering/metering-service/internal/heartbeat/generator_test.goosac-metering/metering-service/internal/heartbeat/heartbeat_suite_test.goosac-metering/metering-service/internal/kafka/connect.goosac-metering/metering-service/internal/kafka/publisher.goosac-metering/metering-service/internal/kafka/publisher_test.goosac-metering/metering-service/internal/projection/postgres.goosac-metering/metering-service/internal/projection/postgres_test.goosac-metering/metering-service/internal/projection/projection_suite_test.goosac-metering/metering-service/internal/projection/store.goosac-metering/metering-service/internal/projection/types.goosac-metering/metering-service/internal/reconciliation/correction.goosac-metering/metering-service/internal/reconciliation/reconciler.goosac-metering/metering-service/internal/reconciliation/reconciler_test.goosac-metering/metering-service/internal/reconciliation/reconciliation_suite_test.goosac-metering/metering-service/internal/watch/consumer.goosac-metering/metering-service/internal/watch/consumer_test.go
🚧 Files skipped from review as they are similar to previous changes (20)
- osac-metering/metering-service/internal/projection/types.go
- osac-metering/metering-service/internal/reconciliation/reconciliation_suite_test.go
- osac-metering/metering-service/internal/projection/store.go
- osac-metering/metering-service/internal/database/migrations.sha256
- osac-metering/metering-service/internal/heartbeat/heartbeat_suite_test.go
- osac-metering/metering-service/internal/projection/projection_suite_test.go
- osac-metering/metering-service/go.mod
- osac-metering/metering-service/internal/database/migrations/0_create_metering_resource_state.up.sql
- osac-metering/metering-service/internal/database/database_suite_test.go
- osac-metering/metering-service/internal/heartbeat/generator.go
- osac-metering/metering-service/internal/heartbeat/generator_test.go
- osac-metering/metering-service/internal/database/database_test.go
- osac-metering/metering-service/internal/events/mapper.go
- osac-metering/metering-service/internal/kafka/publisher_test.go
- osac-metering/metering-service/internal/events/mapper_test.go
- osac-metering/metering-service/internal/projection/postgres.go
- osac-metering/metering-service/internal/kafka/publisher.go
- osac-metering/metering-service/internal/reconciliation/correction.go
- osac-metering/metering-service/internal/events/compute_instance.go
- osac-metering/charts/osac-metering/templates/deployment.yaml
Comment on lines
+145
to
+150
| func readDBURL(dir string) (string, error) { | ||
| data, err := os.ReadFile(filepath.Join(dir, "url")) | ||
| if err != nil { | ||
| return "", fmt.Errorf("reading database URL from %s/url: %w", dir, err) | ||
| } | ||
| return strings.TrimSpace(string(data)), nil |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
ast-grep outline osac-metering/metering-service/cmd/metering-service/main.go --view expanded
rg -n -C 4 'func readDBURL|pgxpool\.ParseConfig|DB_URL_FILE' \
osac-metering/metering-service/cmd/metering-service/main.go
fd -a '.*test\.go$' osac-metering/metering-service | \
xargs -r rg -n -C 3 'readDBURL|ParseConfig|DB_URL_FILE'Repository: osac-project/osac
Length of output: 2678
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '110,205p' osac-metering/metering-service/cmd/metering-service/main.go
printf '\n--- module pgx version ---\n'
rg -n 'pgx|pgxpool|github.com/jackc/pgx' go.mod go.sum osac-metering/metering-service 2>/dev/null | head -120
printf '\n--- tests containing DB_URL_FILE/readDBURL/ParseConfig ---\n'
python3 - <<'PY'
from pathlib import Path
for p in Path('osac-metering/metering-service').rglob('*.go'):
if p.name.endswith('_test.go') or p.name.endswith('test.go'):
text=p.read_text()
if any(s in text for s in ['readDBURL', 'ParseConfig', 'DB_URL_FILE', 'pgxpool']):
print(f'FILE {p}')
for i,line in enumerate(text.splitlines(),1):
if any(s in line for s in ['readDB URL', 'DB_URL_FILE', 'ParseConfig', 'readDBURL', 'pgxpool']):
lo=max(1,i-4); hi=min(len(text.splitlines()), i+4)
lines=text.splitlines()
for j in range(lo,hi+1):
print(f'{j:4}: {lines[j-1]}')
PYRepository: osac-project/osac
Length of output: 5416
🌐 Web query:
jackc pgx/v5 pgxpool.ParseConfig empty string behavior parse error
💡 Result:
In jackc/pgx/v5, calling pgxpool.ParseConfig("") does not cause a parse error [1][2][3]. Instead, providing an empty string is the standard and intended way to instruct the parser to ignore connection string arguments and rely solely on environment variables (such as PGHOST, PGUSER, PGDATABASE, etc.) and, if necessary, the.pgpass file [4][2][3]. When the connection string is empty, the underlying configuration logic proceeds to load defaults (matching PostgreSQL's libpq behavior) and then merges any values found in the environment [4][2]. This is a common pattern for CI/CD or containerized environments where connection details are already injected via environment variables [3].
Citations:
- 1: https://github.com/jackc/pgx/blob/2c625120/pgconn/config_test.go
- 2: https://git.corp.kornet35.ru/gopkg/pgx/commit/473062b114e54e039d7af4a951e877b430ea0c67
- 3: https://gitea.develooper.com/ntp/common/src/commit/af7683da9a049bd66370bc3dda0c2240940b7864/database/pgdb/pool.go
- 4: https://github.com/jackc/pgx/blob/24fbe353ed5c3f53d379b3ae370229a5638a2868/pgconn/config.go
Reject an empty database URL.
A whitespace-only url value passes strings.TrimSpace and reaches pgxpool.ParseConfig, which accepts empty URLs and falls back to environment/libpq defaults. Treat an empty trimmed value as invalid before pool configuration.
Proposed fix
- return strings.TrimSpace(string(data)), nil
+ url := strings.TrimSpace(string(data))
+ if url == "" {
+ return "", fmt.Errorf("database URL at %s/url is empty", dir)
+ }
+ return url, nil📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| func readDBURL(dir string) (string, error) { | |
| data, err := os.ReadFile(filepath.Join(dir, "url")) | |
| if err != nil { | |
| return "", fmt.Errorf("reading database URL from %s/url: %w", dir, err) | |
| } | |
| return strings.TrimSpace(string(data)), nil | |
| func readDBURL(dir string) (string, error) { | |
| data, err := os.ReadFile(filepath.Join(dir, "url")) | |
| if err != nil { | |
| return "", fmt.Errorf("reading database URL from %s/url: %w", dir, err) | |
| } | |
| url := strings.TrimSpace(string(data)) | |
| if url == "" { | |
| return "", fmt.Errorf("database URL at %s/url is empty", dir) | |
| } | |
| return url, nil |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@osac-metering/metering-service/cmd/metering-service/main.go` around lines 145
- 150, Update readDBURL to trim the file contents and reject an empty trimmed
value by returning an appropriate error before returning the URL; continue
returning the trimmed value for non-empty content.
Comment on lines
+250
to
+258
| hbGen := heartbeat.NewGenerator(store, publisher, logger, heartbeatInterval) | ||
| go func() { | ||
| if err := hbGen.Run(ctx); err != nil { | ||
| logger.Error(err, "heartbeat generator exited with error") | ||
| runCancel() | ||
| } | ||
| }() | ||
|
|
||
| go reconciler.RunPeriodic(ctx, reconInterval) |
Contributor
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Wait for workers before closing shared dependencies.
When consumer.Run returns after context cancellation, run can execute producer.Close() while the heartbeat generator or periodic reconciler still calls the publisher or projection store. This can drop final events or use closed resources. Wait for both workers after cancellation, then close the producer and propagate or log its close error.
osac-metering/metering-service/cmd/metering-service/main.go#L250-L258: track both worker goroutines with anerrgrouporsync.WaitGroup.osac-metering/metering-service/cmd/metering-service/main.go#L214-L218: close the producer only after the worker wait completes, and do not discardproducer.Close()errors.
As per path instructions, “Never ignore error returns.”
📍 Affects 1 file
osac-metering/metering-service/cmd/metering-service/main.go#L250-L258(this comment)osac-metering/metering-service/cmd/metering-service/main.go#L214-L218
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@osac-metering/metering-service/cmd/metering-service/main.go` around lines 250
- 258, Track the heartbeat generator and periodic reconciler launched near
hbGen.Run and reconciler.RunPeriodic with an errgroup or sync.WaitGroup, wait
for both to finish after cancellation, then close the producer in run. Handle
and propagate or log the producer.Close error instead of discarding it. Apply
this in osac-metering/metering-service/cmd/metering-service/main.go at lines
250-258 and 214-218.
Source: Path instructions
| kafkapub "github.com/osac-project/osac-metering/internal/kafka" | ||
| ) | ||
|
|
||
| const maxEvents = 10000 |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Bound the event store by bytes.
maxEvents limits item count, not retained payload size. The Deployment caps this container at 128 MiB. Ten thousand raw events averaging slightly more than 13 KiB exceed that limit before slice, JSON, and runtime overhead. A single oversized Kafka record can also exhaust the process. Track a total byte budget below the container limit, cap oversized records, and bound /events responses.
Also applies to: 42-49
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@osac-metering/metering-service/cmd/test-adapter/main.go` at line 28, Replace
the count-only maxEvents limit with a total retained-byte budget below the 128
MiB container limit, accounting for event and slice/runtime overhead. Update the
event ingestion flow to reject or truncate individual records that exceed the
per-record allowance, evict or stop retaining events when the aggregate budget
is reached, and bound the /events response payload independently. Apply the same
byte-budget logic to the related code at the additional referenced section.
Comment on lines
+87
to
+95
| func (h *consumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil } | ||
| func (h *consumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil } | ||
| func (h *consumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { | ||
| for msg := range claim.Messages() { | ||
| h.store.add(msg.Value) | ||
| session.MarkMessage(msg, "") | ||
| } | ||
| return nil | ||
| } |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== locate files =="
fd -a 'main.go|test-adapter-deployment.yaml' . | sed 's#^\./##'
echo "== inspect main.go outline =="
ast-grep outline osac-metering/metering-service/cmd/test-adapter/main.go --view compact || true
echo "== relevant main.go sections =="
cat -n osac-metering/metering-service/cmd/test-adapter/main.go | sed -n '1,230p'
echo "== relevant deployment section =="
cat -n osac-metering/charts/osac-metering/templates/test-adapter-deployment.yaml | sed -n '90,130p'
echo "== sarama imports in main.go and module versions =="
rg -n 'sarama|sarama\.ConsumerGroup|go-sumtype' -S .Repository: osac-project/osac
Length of output: 14987
Make readiness and retries report actual Kafka consumer state.
Setup never records an active consumer session, ConsumeClaim only stops when the context is canceled, and the retry loop calls group.Consume again immediately after non-context errors. readinessProbe:///healthz currently always returns 200 ok, so the Pod can be marked Ready while producing only HTTP success responses and logging failures.
- Track consumer-session readiness in
consumerGroupHandler. - Add a capped backoff around
group.Consumefor errors other than context cancellation. - Add
/readyzand keep/healthzprocess-only. - Change
readinessProbeto call/readyz.
📍 Affects 2 files
osac-metering/metering-service/cmd/test-adapter/main.go#L87-L95(this comment)osac-metering/metering-service/cmd/test-adapter/main.go#L139-L149osac-metering/metering-service/cmd/test-adapter/main.go#L151-L155osac-metering/charts/osac-metering/templates/test-adapter-deployment.yaml#L112-L123
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@osac-metering/metering-service/cmd/test-adapter/main.go` around lines 87 -
95, Update consumerGroupHandler.Setup, Cleanup, and ConsumeClaim in
osac-metering/metering-service/cmd/test-adapter/main.go:87-95 to track the
active session and stop consumption when the session context ends. In
osac-metering/metering-service/cmd/test-adapter/main.go:139-149, wrap
non-context group.Consume errors in a capped backoff before retrying, while
preserving immediate context-cancellation termination. In
osac-metering/metering-service/cmd/test-adapter/main.go:151-155, keep /healthz
process-only and add /readyz to report consumer readiness; update readinessProbe
in
osac-metering/charts/osac-metering/templates/test-adapter-deployment.yaml:112-123
to call /readyz.
| fmt.Fprintf(os.Stderr, "creating consumer group: %v\n", err) | ||
| os.Exit(1) | ||
| } | ||
| defer func() { _ = group.Close() }() |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate and inspect the target file around main lifecycle and shutdown paths.
fd -a 'main.go$' . | rg 'osac-metering/metering-service/cmd/test-adapter/main.go|main.go$' || true
target='osac-metering/metering-service/cmd/test-adapter/main.go'
if [ -f "$target" ]; then
echo "===== file exists ====="
wc -l "$target"
echo "===== outline ====="
ast-grep outline "$target" || true
echo "===== relevant source 1-240 ====="
sed -n '1,240p' "$target" | nl -ba
fi
echo "===== references to srv.Shutdown/group.Close/server lifecycle ====="
rg -n "srv\.Shutdown|Shutdown\(|group\.Close|ListenAndServe|signal|select|close\(.*\)|err :=" "$target" || trueRepository: osac-project/osac
Length of output: 1728
🏁 Script executed:
#!/bin/bash
set -euo pipefails
# Inspect relevant Go HTTP lifecycle behavior from installed docs/runtime if present.
go version 2>/dev/null || true
if command -v go >/dev/null 2>&1; then
go env GOROOT 2>/dev/null || true
printf '===== net/http Shutdown docs excerpt =====\n'
go doc net/http.Server.Shutdown 2>/dev/null | sed -n '1,80p' || true
fi
# Static verifier: identify whether main returns before waiting for server shutdown
# and whether error returns are discarded in the target function.
python3 - <<'PY'
from pathlib import Path
import re
p = Path('osac-metering/metering-service/cmd/test-adapter/main.go')
text = p.read_text()
lines = text.splitlines()
# Print locations and adjacent context for Shutdown and return.
print("===== file lines with shutdown/return relevant symbols =====")
for i, line in enumerate(lines, 1):
if 'srv.Shutdown' in line or 'group.Close' in line or line.strip().startswith('return ') or 'err :=' in line or '= err' in line:
start=max(1,i-3); end=min(len(lines),i+2)
for j in range(start,end+1):
print(f"{j:5}:{lines[j-1]}")
print("---")
# Behavioral model of deferred close vs return:
# A top-level defer that calls Close during normal exit still returns Close() result as deferred cleanup.
# If Shutdown is called in goroutine and main returns before that goroutine completes,
# then active request drain may be incomplete depending on caller waiting semantics.
print("===== static signal goroutine patterns =====")
for i, line in enumerate(lines, 1):
if 'go func' in line or 'Select' in line or 'sig' in line.lower() or 'Shutdown' in line:
print(f"{i:5}:{lines[i-1]}")
PYRepository: osac-project/osac
Length of output: 205
🏁 Script executed:
#!/bin/bash
set -u
target='osac-metering/metering-service/cmd/test-adapter/main.go'
echo "===== relevant source 1-240 ====="
awk '{printf "%5d:%s\n", NR, $0}' "$target" | sed -n '97,192p'
echo "===== references ====="
grep -nE 'srv\.Shutdown|Shutdown\(|group\.Close|ListenAndServe|signal|select|err :=|return ' "$target" || true
echo "===== package/module Go version hints ====="
for f in go.mod go.sum "$target"; do
if [ -f "$f" ]; then
echo "--- $f"
grep -nE '^(go |require \(|github\.com\/segmentio\/kafka|golang\.org\/x/net/http2)' "$f" | sed -n '1,80p' || true
fi
doneRepository: osac-project/osac
Length of output: 4715
Wait for HTTP shutdown on the main path.
On SIGINT/SIGTERM, the signal goroutine calls srv.Shutdown and ignores the error, but ListenAndServe can return normally afterward. Move shutdown coordination onto the main path, wait for srv.Shutdown to complete, and check the returned errors.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@osac-metering/metering-service/cmd/test-adapter/main.go` at line 134, Update
the server lifecycle around srv.ListenAndServe and the signal handler so
shutdown is coordinated and awaited on the main path rather than performed
asynchronously. Preserve signal-triggered shutdown, but capture and check both
ListenAndServe and srv.Shutdown errors, allowing the main flow to wait for
shutdown completion before closing group.
Source: Path instructions
Comment on lines
+180
to
+196
| if event.GetType() == privatev1.EventType_EVENT_TYPE_OBJECT_DELETED { | ||
| if existing != nil { | ||
| if err := c.store.Delete(ctx, resourceID); err != nil { | ||
| return fmt.Errorf("deleting projection for %s: %w", resourceID, err) | ||
| } | ||
| } | ||
| } else { | ||
| err = c.store.Upsert(ctx, projState) | ||
| if err != nil { | ||
| return received, fmt.Errorf("mapping event %s: %w", resp.GetEvent().GetId(), err) | ||
| if errors.Is(err, projection.ErrStaleVersion) { | ||
| c.logger.Info("stale version, skipping projection update", | ||
| "resource_id", resourceID, "version", version) | ||
| return nil | ||
| } | ||
| return fmt.Errorf("upserting projection for %s: %w", resourceID, err) | ||
| } | ||
| } |
Contributor
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Reject stale deletions before mutation.
If a deletion has a lower fulfillment version than the stored projection, Delete still removes current state and publishes a false deletion event. ErrStaleVersion protects only Upsert.
Make deletion conditional on the incoming fulfillment version in the store. Execute the version check and deletion atomically. Do not publish when the store rejects a stale deletion.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@osac-metering/metering-service/internal/watch/consumer.go` around lines 180 -
196, The deletion branch in the consumer flow must validate the incoming
fulfillment version before removing state. Update the store operation used by
the OBJECT_DELETED handling around c.store.Delete to perform the version check
and deletion atomically, return or recognize projection.ErrStaleVersion without
publishing or treating it as a successful deletion, and preserve the existing
error propagation for other failures.
Comment on lines
+21
to
+22
| build-test-adapter: | ||
| go build -o bin/test-adapter ./cmd/test-adapter |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Declare build-test-adapter as phony.
If a file or directory has this name, make build-test-adapter skips the build. Add the target to .PHONY.
Proposed fix
-.PHONY: build test lint generate image-build clean
+.PHONY: build test lint generate image-build build-test-adapter clean🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@osac-metering/metering-service/Makefile` around lines 21 - 22, Declare the
build-test-adapter target as phony by adding it to the Makefile’s .PHONY
declaration, preserving the existing target command.
Contributor
|
Note GitHub couldn't provide a complete incremental comparison for this pull request, so CodeRabbit is performing a full review instead. This review may take a little longer. |
omer-vishlitzky
had a problem deploying
to
e2e-test
August 5, 2026 15:25 — with
GitHub Actions
Error
GitHub Actions
Error
omer-vishlitzky
had a problem deploying
to
e2e-test
August 5, 2026 15:25 — with
GitHub Actions
Error
GitHub Actions
Error
omer-vishlitzky
force-pushed
the
feat/OSAC-3408-state-projection-heartbeat-reconciliation
branch
from
August 5, 2026 15:28
18d37ec to
ec19437
Compare
omer-vishlitzky
temporarily deployed
to
e2e-test
August 5, 2026 15:30 — with
GitHub Actions
Inactive
GitHub Actions
Inactive
omer-vishlitzky
had a problem deploying
to
e2e-test
August 5, 2026 15:30 — with
GitHub Actions
Error
GitHub Actions
Error
Contributor
Go code review — follow-up on
|
…elm config Migration framework using golang-migrate/v4 with embedded SQL files. Returns error on dirty schema state requiring manual intervention. Schema: metering_resource_state table with resource identity, state, billability tracking, billing dimensions (JSONB), fulfillment_version for stale-update rejection, and partial indexes. Test container via podman/docker with template database for fast per-test isolation. Thread-safe instance creation. Migration collision prevention tests (SHA-256 hash, duplicate/gap detection, naming). Helm: projected volume with database.connection for cert-based PostgreSQL auth (matching fulfillment-service). osac-installer bundled postgres creates metering user, database, and client cert for CI. Assisted-by: Claude Code <noreply@anthropic.com> Signed-off-by: omer-vishlitzky <omer.vishlitzky@gmail.com>
Store interface: Get, Upsert, Delete, ListBillable, ListAll, UpdateLastHeartbeat. PostgreSQL implementation with SELECT FOR UPDATE row-level locking and fulfillment_version stale-update rejection (> not >=, allowing idempotent same-version replays). ResourceState tracks: resource identity, current/previous state, billability, BillableSince, LastHeartbeatAt, transition time, fulfillment version, and billing dimensions (map[string]any). Tests with real PostgreSQL via database.Container: upsert/get round-trip, stale version rejection, same-version idempotent replay, concurrent upsert serialization (version 3 wins over 2), delete, ListBillable filtering, UpdateLastHeartbeat batch update. Assisted-by: Claude Code <noreply@anthropic.com> Signed-off-by: omer-vishlitzky <omer.vishlitzky@gmail.com>
Watch Consumer reads previous state from projection before mapping, upserts after. Skips same-state-same-dimensions updates. Publishes delete events before deleting projection (publish-first for deletes). Fails fast on unknown payload, missing timestamps, missing tenant_id. Handles ErrStaleVersion gracefully (log and continue). State-derived event types: STOPPED/PAUSED to RUNNING emits resumed.v1; first-time RUNNING emits started.v1. Billing interval closes on dimension change while billable (BillableSince resets, duration_seconds computed from old BillableSince). Unified billing_dimensions: all event types use BillingDimensionsMap() (map[string]any). Absent fields omitted, numbers stay numeric. ComputeInstanceBillingDimensions exported for shared use. DimensionsEqual and NilIfEmpty extracted to events package. Reconnect metric: osac_metering_watch_stream_reconnects_total. Tests: fail-fast (unknown payload, missing timestamp, missing tenant), state projection integration (same-state skip, resumed.v1, delete + cleanup, stale version skip, first-seen upsert, dimension-change interval close with BillableSince reset). Assisted-by: Claude Code <noreply@anthropic.com> Signed-off-by: omer-vishlitzky <omer.vishlitzky@gmail.com>
Replace single topic with TopicRouter map: lifecycle, heartbeat, corrections, inference topics. TopicFor returns error for unknown event types. Partition key = osacresourceid extension. Prometheus metrics: osac_metering_events_published_total (topic, event_type, resource_type), osac_metering_kafka_publish_errors_total (topic, error_type), osac_metering_kafka_publish_latency_seconds. Exported ConfigureTLS, ConfigureSASL, SplitAndTrim for reuse by test adapter consumer. Tests: routing for all event types, partition key, CloudEvent JSON serialization, Kafka headers, missing resource ID error, unknown type error, producer config verification. Assisted-by: Claude Code <noreply@anthropic.com> Signed-off-by: omer-vishlitzky <omer.vishlitzky@gmail.com>
Generator queries ListBillable, publishes osac.resource.heartbeat.v1 CloudEvents per resource, updates last_heartbeat_at. Partial checkpoint on Kafka failure — successfully published IDs are checkpointed before returning error. Heartbeat events carry: resource identity, tenant_id, project_id, current_state, duration_seconds (from BillableSince), billing dimensions, and schema_version. CloudEvent extensions include osacresourceid, osacresourcetype, osactenant, osacproject. Prometheus gauges: osac_metering_heartbeat_lag_seconds (max staleness per resource type), osac_metering_state_projection_resources (billable count). Gauges reset each tick to clear stale label sets. Tests: heartbeat publish + last_heartbeat_at update, partial Kafka failure checkpoint, ListBillable error, empty billable set, billing dimensions and osacresourceid in events, context cancellation. Assisted-by: Claude Code <noreply@anthropic.com> Signed-off-by: omer-vishlitzky <omer.vishlitzky@gmail.com>
Reconciler diffs State Projection against fulfillment-service ListComputeInstances (paginated, 500/page). Four correction types: missed_creation, state_drift, billing_dimensions_drift, missed_deletion. Publish-first strategy: corrections published to Kafka before updating projection. ErrStaleVersion handled gracefully (log and continue) for concurrent Watch Consumer updates. Single now timestamp per cycle. Stale heartbeat detection: reloads projection from DB after corrections, emits synthetic heartbeats for billable resources with LastHeartbeatAt older than 2x heartbeat interval. Updates LastHeartbeatAt after publish. Correction events: *string for absent states (null not ""), billing dimensions included, project_id from fulfillment. Shared ComputeInstanceBillingDimensions and ComputeInstanceStatePrefix from events package. Stale threshold derived from heartbeat interval (2x) instead of hardcoded 120s. Metrics: osac_metering_reconciliation_duration_seconds, osac_metering_reconciliation_corrections_total, last_completed_at. Tests: all four correction types, BillableSince on state/dimension drift, stale heartbeat emission, no corrections when match, List API error, publish-first verification, ErrStaleVersion continuation, billing dimensions extraction, periodic cancellation. Assisted-by: Claude Code <noreply@anthropic.com> Signed-off-by: omer-vishlitzky <omer.vishlitzky@gmail.com>
…tion Startup order: DB connect via URL file, migrate, startup reconciliation, set ready, start Watch Consumer + Heartbeat Generator + periodic reconciliation. Readiness probe gates on reconciliationComplete flag plus gRPC connectivity. Workers tracked via sync.WaitGroup — producer and pool close only after all workers finish. Config validation: reject non-positive HEARTBEAT_INTERVAL and RECONCILIATION_INTERVAL. DB_URL_FILE required. KAFKA_TOPIC removed (dead code — routing is in publisher topicRoutes). Assisted-by: Claude Code <noreply@anthropic.com> Signed-off-by: omer-vishlitzky <omer.vishlitzky@gmail.com>
Kafka consumer pod for CI that consumes all metering topics and exposes events via GET /events HTTP API with type/resource_id/since filtering. Solves DNS resolution for cluster-internal Kafka from external test containers. Disabled by default via testAdapter.enabled. Separate KafkaUser with read-only ACLs. RBAC includes test adapter secret in resourceNames when enabled. Containerfile, Makefile target, Helm deployment/service. Assisted-by: Claude Code <noreply@anthropic.com> Signed-off-by: omer-vishlitzky <omer.vishlitzky@gmail.com>
Correctness: - Version guard: skip corrections when projection is ahead of fulfillment snapshot (fs.version < ps.FulfillmentVersion) - Correction events: *string for null semantics, project_id, description, billing_dimensions fields; SetData errors checked - Heartbeat/synthetic heartbeat: SetData errors checked, return (cloudevents.Event, error) - Same-state skip: WARN log when transition times differ (signals replayed upserted-but-unpublished events) - Dirty migration: return error instead of proceeding Readability: - Reconciler: rename RunFull to Reconcile, split into reconcileFulfillmentResources, reconcileMissedDeletions, reconcileStaleHeartbeats - Watch Consumer: extract shouldSkipUpdate, buildProjectionState - Remove Phase 1 nil guard from MapWatchEvent - Restore doc comments on Consumer, Run, Publish, NewProducerConfig - Update heartbeat TODO to describe partial checkpoint behavior Code quality: - Topic constants (TopicLifecycle, TopicHeartbeat, TopicCorrections, TopicInference) and Topics slice in kafka package; main.go and test adapter use shared constants - Intervals parsed once as time.Duration in configFromEnv; parseDurationOrDefault combines parse + positive check - Test adapter uses kafkapub.Topics (no inference in Phase 2) - Container.go: test-only purpose comment - Test adapter: bounded ring buffer comment Test coverage (6 new state transition tests): - FAILED→RUNNING (started.v1), RUNNING→RUNNING (started.v1) - DELETING (updated.v1), UNSPECIFIED (updated.v1) - RUNNING→STOPPED with duration_seconds, RUNNING→FAILED (suspended.v1) Assisted-by: Claude Code <noreply@anthropic.com> Signed-off-by: omer-vishlitzky <omer.vishlitzky@gmail.com>
… comparison
Reconciler:
- Remove BillableSince reset from billing_dimensions_drift branch.
Unlike the Watch Consumer (which processes real-time resize events
and correctly closes/reopens billing intervals), the reconciler
detects drift after the fact. Resetting BillableSince to now would
destroy accrued billable time with no way for adapters to recover
the lost duration.
- Advance fulfillment_version in projection when state and dimensions
match but fulfillment has a newer version. Prevents stale Watch
events from passing the version check later.
DimensionsEqual:
- Normalize numeric types via toFloat64 before comparison. Prevents
false mismatches for large numbers where fmt.Sprintf("%v") produces
scientific notation for float64 (1e+06) vs decimal for int32
(1000000).
Heartbeat:
- Log and continue on buildHeartbeatEvent error instead of aborting
the entire tick. One poisoned resource should not kill heartbeats
for all other billable resources.
Tests:
- Reconciler: BillableSince preserved on dimension drift, version
advanced when state+dims match
- DimensionsEqual: int32 vs float64, large numbers, empty/nil maps
Assisted-by: Claude Code <noreply@anthropic.com>
Signed-off-by: omer-vishlitzky <omer.vishlitzky@gmail.com>
Shared CloudEvent extensions: - Extract SetOSACExtensions helper in events package. Sets osacresourceid, osacresourcetype, osactenant, osacproject on any CloudEvent. Replaces 4 duplicated extension-setting blocks in mapper, heartbeat generator, correction builder, and synthetic heartbeat builder. Composite index: - Add last_heartbeat_at to the billable partial index per design spec: (is_billable, last_heartbeat_at) WHERE is_billable = TRUE. Enables index-only scan for stale heartbeat detection in reconciliation. Test adapter: - Make maxEvents configurable via MAX_EVENTS env var (default 10000) for CI tunability under high-volume test scenarios. Config: - Log warning to stderr when HEARTBEAT_INTERVAL or RECONCILIATION_INTERVAL is set to an invalid or non-positive value and falls back to default. Assisted-by: Claude Code <noreply@anthropic.com> Signed-off-by: omer-vishlitzky <omer.vishlitzky@gmail.com>
Time precision: - Truncate both sides to microsecond in shouldSkipUpdate before comparing transition times. Proto timestamps are nanosecond, PostgreSQL TIMESTAMPTZ is microsecond — without truncation the WARN log fires on every same-state event, not just replays. Shared billability: - Export IsBillableState from events package. Replaces duplicated isBillableState in reconciler. Single source of truth prevents divergence when new billable states are added. Observability: - Debug log when reconciler skips a resource because projection is ahead of fulfillment snapshot (fs.version < ps.FulfillmentVersion). Cleanup: - Remove dead error-handling for heartbeat generator Run (always returns nil; tick errors are logged internally). - ProjectID uses nullIfEmpty in Upsert for NULL vs empty string consistency with PreviousState. - AffectedInterval TODO comment for deferred scope. Assisted-by: Claude Code <noreply@anthropic.com> Signed-off-by: omer-vishlitzky <omer.vishlitzky@gmail.com>
Contributor
|
Note GitHub couldn't provide a complete incremental comparison for this pull request, so CodeRabbit is performing a full review instead. This review may take a little longer. |
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (2)
osac-metering/metering-service/internal/database/container.go (1)
188-193: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
i.namestays set afterCREATE DATABASEfails.
initassignsi.nameat line 180 before it creates the database. If theExecat line 188 fails,i.nameremains non-empty. The nextURLorPoolcall returns early at line 174 and hands back a URL for a database that does not exist. The real error appears once, and every later failure looks like a connection problem.Reset
i.nameon the error path. Capture the name first so the error message keeps it.🐛 Proposed fix
+ name := i.name _, err := i.container.adminConn.Exec(ctx, fmt.Sprintf("CREATE DATABASE %s TEMPLATE metering_template OWNER %s", - i.name, containerUser)) + name, containerUser)) if err != nil { - return fmt.Errorf("creating database %s from template: %w", i.name, err) + i.name = "" + return fmt.Errorf("creating database %s from template: %w", name, err) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/database/container.go` around lines 188 - 193, In the database creation error path within init, capture i.name before attempting Exec, then reset i.name to its uninitialized value before returning the wrapped error. Use the captured name in the error message so subsequent URL or Pool calls retry initialization instead of treating the failed database as existing.osac-metering/metering-service/internal/reconciliation/reconciler.go (1)
377-380: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winA short page ends pagination early and triggers false deletions.
The loop stops as soon as a page returns fewer than
defaultPageSizeitems. Servers are free to cap a page below the requested limit. If that happens,resultomits every later resource.
reconcileMissedDeletionsat lines 255-271 treats each missing resource as deleted. It publishes aMissedDeletioncorrection and callsr.store.Deletefor a resource that is still running. That drops the projection for a live billable resource and stops its billing.Use the response totals to decide termination.
ComputeInstancesListResponseprovidesSizeandTotal. Continue untiloffset + len(items)reachesresp.GetTotal(), and return an error if a nonterminal page makes no progress. Add a test with a short page before the total and assert that no projection is deleted.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/reconciliation/reconciler.go` around lines 377 - 380, Update the pagination loop in reconciler.go around the reconciliation fetch logic to stop using len(items) < defaultPageSize as the termination condition. Use ComputeInstancesListResponse totals instead: keep iterating until offset + len(items) reaches resp.GetTotal(), and return an error if a nonterminal page yields no progress. Adjust the reconcileMissedDeletions path only as needed to preserve live resources, and add a test covering a short page before the total that confirms no projection is deleted.
🧹 Nitpick comments (3)
osac-metering/metering-service/internal/reconciliation/reconciler.go (1)
324-328: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winBound each reconciliation pass with a timeout.
Reconcilereceives the caller's long-lived context. It performs paginated gRPC calls and several database writes. If one call hangs, the pass never returns, the ticker channel is not drained, and reconciliation stalls until process shutdown.Derive a per-pass context with a deadline.
As per path instructions "context.Context for cancellation and timeouts".♻️ Proposed fix
case <-ticker.C: - if err := r.Reconcile(ctx); err != nil { - r.logger.Error(err, "periodic reconciliation failed") - } + passCtx, cancel := context.WithTimeout(ctx, interval) + if err := r.Reconcile(passCtx); err != nil { + r.logger.Error(err, "periodic reconciliation failed") + } + cancel() }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/reconciliation/reconciler.go` around lines 324 - 328, Update the periodic reconciliation branch around r.Reconcile to derive a child context with a per-pass timeout from ctx, invoke Reconcile with that context, and always cancel it after the pass completes. Keep the existing error logging behavior, including timeout errors, unchanged.Source: Path instructions
osac-metering/metering-service/internal/database/container.go (1)
83-121: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winClean up the container when
Startfails partway.
Startlaunchesc.runCmd, then returns onwaitForPort,waitForConnection, migration, or DDL failures without callingStop.--rmwaits for the process to exit, so each failed setup leaves a live PostgreSQL container and temp config file behind. If the suite does not callStopafterdbContainer.Start(ctx)fails, add cleanup for this path.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/database/container.go` around lines 83 - 121, Update the container startup flow in Start to clean up c.runCmd and its temporary resources whenever waitForPort, waitForConnection, user/database creation, migration, or template-database setup fails after the process starts. Ensure cleanup stops the running container before returning the original setup error, while preserving the existing successful startup path and error context.osac-metering/metering-service/internal/events/compute_instance.go (1)
27-31: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse
IsBillableStatefromIsBillable.Lines 27-31 and Lines 56-60 define the same billability rule. The rules can diverge when billable states change. Make
IsBillabledelegate toIsBillableState(m.CurrentState()).Proposed refactor
func (m *computeInstanceMapper) IsBillable() bool { - if s := m.ci.GetStatus(); s != nil { - return s.GetState() == privatev1.ComputeInstanceState_COMPUTE_INSTANCE_STATE_RUNNING - } - return false + return IsBillableState(m.CurrentState()) }Also applies to: 56-60
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@osac-metering/metering-service/internal/events/compute_instance.go` around lines 27 - 31, Update computeInstanceMapper.IsBillable to delegate directly to IsBillableState using m.CurrentState(), removing the duplicated status/state comparison while preserving the existing billability behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@osac-metering/metering-service/internal/events/dimensions.go`:
- Around line 21-23: Update the map comparison loop in the dimensions equality
function to verify that each key from a is present in b before comparing values
with valueEqual. Return false when the key is missing, while preserving the
existing value comparison for present keys.
---
Duplicate comments:
In `@osac-metering/metering-service/internal/database/container.go`:
- Around line 188-193: In the database creation error path within init, capture
i.name before attempting Exec, then reset i.name to its uninitialized value
before returning the wrapped error. Use the captured name in the error message
so subsequent URL or Pool calls retry initialization instead of treating the
failed database as existing.
In `@osac-metering/metering-service/internal/reconciliation/reconciler.go`:
- Around line 377-380: Update the pagination loop in reconciler.go around the
reconciliation fetch logic to stop using len(items) < defaultPageSize as the
termination condition. Use ComputeInstancesListResponse totals instead: keep
iterating until offset + len(items) reaches resp.GetTotal(), and return an error
if a nonterminal page yields no progress. Adjust the reconcileMissedDeletions
path only as needed to preserve live resources, and add a test covering a short
page before the total that confirms no projection is deleted.
---
Nitpick comments:
In `@osac-metering/metering-service/internal/database/container.go`:
- Around line 83-121: Update the container startup flow in Start to clean up
c.runCmd and its temporary resources whenever waitForPort, waitForConnection,
user/database creation, migration, or template-database setup fails after the
process starts. Ensure cleanup stops the running container before returning the
original setup error, while preserving the existing successful startup path and
error context.
In `@osac-metering/metering-service/internal/events/compute_instance.go`:
- Around line 27-31: Update computeInstanceMapper.IsBillable to delegate
directly to IsBillableState using m.CurrentState(), removing the duplicated
status/state comparison while preserving the existing billability behavior.
In `@osac-metering/metering-service/internal/reconciliation/reconciler.go`:
- Around line 324-328: Update the periodic reconciliation branch around
r.Reconcile to derive a child context with a per-pass timeout from ctx, invoke
Reconcile with that context, and always cancel it after the pass completes. Keep
the existing error logging behavior, including timeout errors, unchanged.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository: osac-project/coderabbit/.coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 1888dd31-ae93-4fad-9ac8-6e2c8f4ea3a0
⛔ Files ignored due to path filters (63)
go.work.sumis excluded by!**/*.sumosac-metering/metering-service/go.sumis excluded by!**/*.sumosac-metering/metering-service/internal/api/osac/private/v1/authn_capabilities_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/bare_metal_instance_type_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/baremetal_instance_catalog_item_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/baremetal_instance_catalog_items_service.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/baremetal_instance_catalog_items_service_grpc.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/baremetal_instance_template_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/baremetal_instance_templates_service.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/baremetal_instance_templates_service_grpc.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/baremetal_instance_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/baremetal_instances_service.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/baremetal_instances_service_grpc.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/capabilities_service.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/capabilities_service_grpc.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/cluster_catalog_item_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/cluster_catalog_items_service.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/cluster_catalog_items_service_grpc.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/cluster_template_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/cluster_templates_service.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/cluster_templates_service_grpc.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/cluster_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/cluster_version_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/cluster_versions_service.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/cluster_versions_service_grpc.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/clusters_service.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/clusters_service_grpc.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/compute_instance_catalog_item_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/compute_instance_catalog_items_service.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/compute_instance_catalog_items_service_grpc.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/compute_instance_template_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/compute_instance_templates_service.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/compute_instance_templates_service_grpc.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/compute_instance_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/compute_instances_service.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/compute_instances_service_grpc.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/condition_status_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/event_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/events_service.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/events_service_grpc.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/field_definition_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/host_type_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/host_types_service.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/host_types_service_grpc.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/hub_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/hubs_service.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/hubs_service_grpc.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/instance_type_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/instance_types_service.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/instance_types_service_grpc.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/metadata_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/network_class_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/network_classes_service.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/network_classes_service_grpc.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/security_group_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/security_groups_service.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/security_groups_service_grpc.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/subnet_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/subnets_service.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/subnets_service_grpc.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/virtual_network_type.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/virtual_networks_service.pb.gois excluded by!**/*.pb.goosac-metering/metering-service/internal/api/osac/private/v1/virtual_networks_service_grpc.pb.gois excluded by!**/*.pb.go
📒 Files selected for processing (46)
osac-installer/charts/osac/templates/bundled-postgres.yamlosac-installer/charts/osac/values.schema.jsonosac-installer/values/vmaas-ci/values.yamlosac-metering/charts/osac-metering/templates/deployment.yamlosac-metering/charts/osac-metering/templates/kafka-secrets-rbac.yamlosac-metering/charts/osac-metering/templates/test-adapter-deployment.yamlosac-metering/charts/osac-metering/templates/test-adapter-kafka-user.yamlosac-metering/charts/osac-metering/templates/test-adapter-service.yamlosac-metering/charts/osac-metering/values.yamlosac-metering/metering-service/Containerfile.test-adapterosac-metering/metering-service/Makefileosac-metering/metering-service/buf.gen.yamlosac-metering/metering-service/cmd/metering-service/main.goosac-metering/metering-service/cmd/test-adapter/main.goosac-metering/metering-service/go.modosac-metering/metering-service/internal/database/container.goosac-metering/metering-service/internal/database/database_suite_test.goosac-metering/metering-service/internal/database/database_test.goosac-metering/metering-service/internal/database/migrations.sha256osac-metering/metering-service/internal/database/migrations/0_create_metering_resource_state.down.sqlosac-metering/metering-service/internal/database/migrations/0_create_metering_resource_state.up.sqlosac-metering/metering-service/internal/database/tool.goosac-metering/metering-service/internal/events/compute_instance.goosac-metering/metering-service/internal/events/dimensions.goosac-metering/metering-service/internal/events/extensions.goosac-metering/metering-service/internal/events/mapper.goosac-metering/metering-service/internal/events/mapper_test.goosac-metering/metering-service/internal/heartbeat/generator.goosac-metering/metering-service/internal/heartbeat/generator_test.goosac-metering/metering-service/internal/heartbeat/heartbeat_suite_test.goosac-metering/metering-service/internal/kafka/connect.goosac-metering/metering-service/internal/kafka/kafka_suite_test.goosac-metering/metering-service/internal/kafka/publisher.goosac-metering/metering-service/internal/kafka/publisher_test.goosac-metering/metering-service/internal/kafka/scram.goosac-metering/metering-service/internal/projection/postgres.goosac-metering/metering-service/internal/projection/postgres_test.goosac-metering/metering-service/internal/projection/projection_suite_test.goosac-metering/metering-service/internal/projection/store.goosac-metering/metering-service/internal/projection/types.goosac-metering/metering-service/internal/reconciliation/correction.goosac-metering/metering-service/internal/reconciliation/reconciler.goosac-metering/metering-service/internal/reconciliation/reconciler_test.goosac-metering/metering-service/internal/reconciliation/reconciliation_suite_test.goosac-metering/metering-service/internal/watch/consumer.goosac-metering/metering-service/internal/watch/consumer_test.go
🚧 Files skipped from review as they are similar to previous changes (36)
- osac-metering/metering-service/internal/database/migrations.sha256
- osac-metering/metering-service/internal/projection/types.go
- osac-metering/metering-service/internal/projection/store.go
- osac-metering/charts/osac-metering/templates/kafka-secrets-rbac.yaml
- osac-metering/metering-service/go.mod
- osac-metering/metering-service/internal/heartbeat/heartbeat_suite_test.go
- osac-metering/metering-service/internal/database/migrations/0_create_metering_resource_state.up.sql
- osac-installer/values/vmaas-ci/values.yaml
- osac-metering/metering-service/internal/heartbeat/generator_test.go
- osac-metering/metering-service/internal/database/tool.go
- osac-metering/metering-service/internal/reconciliation/reconciliation_suite_test.go
- osac-metering/metering-service/internal/heartbeat/generator.go
- osac-metering/metering-service/Containerfile.test-adapter
- osac-metering/metering-service/internal/kafka/connect.go
- osac-metering/metering-service/internal/projection/projection_suite_test.go
- osac-metering/metering-service/internal/projection/postgres.go
- osac-metering/metering-service/internal/kafka/publisher.go
- osac-metering/charts/osac-metering/values.yaml
- osac-installer/charts/osac/values.schema.json
- osac-metering/metering-service/Makefile
- osac-metering/charts/osac-metering/templates/deployment.yaml
- osac-metering/charts/osac-metering/templates/test-adapter-kafka-user.yaml
- osac-installer/charts/osac/templates/bundled-postgres.yaml
- osac-metering/metering-service/internal/database/database_suite_test.go
- osac-metering/metering-service/internal/projection/postgres_test.go
- osac-metering/metering-service/internal/reconciliation/correction.go
- osac-metering/charts/osac-metering/templates/test-adapter-deployment.yaml
- osac-metering/charts/osac-metering/templates/test-adapter-service.yaml
- osac-metering/metering-service/cmd/test-adapter/main.go
- osac-metering/metering-service/internal/events/mapper_test.go
- osac-metering/metering-service/internal/watch/consumer_test.go
- osac-metering/metering-service/cmd/metering-service/main.go
- osac-metering/metering-service/internal/watch/consumer.go
- osac-metering/metering-service/internal/database/database_test.go
- osac-metering/metering-service/internal/events/mapper.go
- osac-metering/metering-service/internal/kafka/publisher_test.go
Comment on lines
+21
to
+23
| for k, v := range a { | ||
| if !valueEqual(v, b[k]) { | ||
| return false |
Contributor
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
Check that b contains each key before comparing values.
Line 22 treats a missing key as nil. Two same-size maps with different keys and nil values can compare equal. Reconciliation then skips a required billing-dimension correction.
Proposed fix
for k, v := range a {
- if !valueEqual(v, b[k]) {
+ other, ok := b[k]
+ if !ok || !valueEqual(v, other) {
return false
}
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| for k, v := range a { | |
| if !valueEqual(v, b[k]) { | |
| return false | |
| for k, v := range a { | |
| other, ok := b[k] | |
| if !ok || !valueEqual(v, other) { | |
| return false |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@osac-metering/metering-service/internal/events/dimensions.go` around lines 21
- 23, Update the map comparison loop in the dimensions equality function to
verify that each key from a is present in b before comparing values with
valueEqual. Return false when the key is missing, while preserving the existing
value comparison for present keys.
Contributor
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: masayag, omer-vishlitzky The full list of commands accepted by this bot can be found here. The pull request process is described here DetailsNeeds approval from an approver in each of these files:
Approvers can indicate their approval by writing |
1 of 3 tasks
This was referenced Aug 6, 2026
3 tasks done
6 tasks done
This branch had an error being deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Phase 2 of OSAC metering: adds durable state tracking to the Metering Service.
resumed.v1(STOPPED/PAUSED→RUNNING) andduration_secondsenrichment. Stale-update rejection viafulfillment_version.osac.resource.heartbeat.v1for all billable resources. All-or-nothing per tick. Bounds metering gap to one tick on pod restart.ListComputeInstances. Emits correction events (missed_creation,state_drift,missed_deletion) with publish-first strategy. Stale heartbeat detection emits synthetic catch-up heartbeats for billable resources with no heartbeat in >120s. Readiness probe gates on startup reconciliation.TopicForreturns error (not panic) for unknown types. Prometheus metrics on publish.testAdapter.enabled). Consumes all metering topics, exposes events viaGET /eventsHTTP API. Solves the DNS resolution issue where CI test containers cannot reach cluster-internal Kafka. Reference implementation of the Provider Adapter consumer pattern.Tickets covered
Correctness hardening (post-review)
>not>=), so a Watch event replayed after publish failure can re-process without being silently dropped by the stale version guardBillableSincewhen correcting to non-billable state, preventing stale duration computations in heartbeatsDesign choices
duration_secondscomputed fromtransitionTime - billableSince(event time, not wall clock), preventing inflated durations on processing delay.billingDims()source;BillingDimensions()andBillingDimensionsMap()both derive from it.What Phase 2 does NOT include (deferred)
AffectedIntervalin correction events (adapters not yet consuming it)References
Test plan
cd osac-metering/metering-service && go build ./...passes (both binaries)cd osac-metering/metering-service && ginkgo run -r internal— 97 specs pass (7 suites)helm lint osac-metering/charts/osac-metering/passeshelm lint osac-metering/charts/osac-metering/ --set testAdapter.enabled=truepassesgolangci-lint run ./...— 0 issuesosac.metering.heartbeattopic at ~60s intervalsresumed.v1emitted on STOPPED→RUNNING transitionGET /eventsreturns consumed events🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Monitoring
Bug Fixes