feat: [TKC-6336] mongo postgres convert - #8171
Open
vsukhin wants to merge 23 commits into
Open
Conversation
vsukhin
changed the title
vsukhin
changed the title
Postgres has been a supported control-plane store for a while, selected by setting API_POSTGRES_DSN instead of API_MONGO_DSN, but there was no way to bring existing data across. Switching the DSN silently started from an empty database: every past Test Workflow execution disappeared from the UI, and execution numbering restarted at 1, so newly scheduled executions collided with the names of old ones. Add a convert binary that copies the two collections that hold data worth moving - testworkflowresults into the seven test_workflow_* tables, and sequences into execution_sequences. It runs as a Kubernetes Job during a cutover, or directly against port-forwarded databases. Nothing else is in scope, because nothing else is in Mongo. Execution logs, outputs and artifacts live in object storage and only their references travel with the execution. Workflow definitions and triggers are custom resources, and cluster configuration is a ConfigMap. The triggers collection holds a leader lease with a one-minute lifetime that the API reacquires on startup, so copying it would move a stale row and nothing else. The write path is CSV plus COPY FROM STDIN, mirroring the control plane's own convert tool, including the per-execution all-or-nothing staging that stops a serialization failure from importing a parent row without its children, and the client-minted signature UUIDs that let a recursive tree satisfy its self-referencing foreign key in file order. Runs are resumable, which the tool this is modelled on is explicitly not. Each batch commits its seven COPY statements and its checkpoint in one transaction, so progress can never be durable without its data or vice versa. A restarted Job resumes from the last committed batch instead of starting over or double-writing - the difference between a tool an operator can retry and one that has to be babysat. Sequences upsert through a temporary table with GREATEST rather than a bare COPY, so a counter can never move backwards. Lowering one would hand a migrated execution's number to a new run, which is the failure the whole sequences phase exists to prevent. workflow_name and status are written directly rather than left to the two denormalization triggers. Those still fire per row during COPY, but their IS DISTINCT FROM guards then match nothing, so each costs an index probe instead of a heap update plus index churn. Documents are decoded raw rather than through the repository, whose read path truncates long config values and blanks sensitive ones - storing that projection would make the loss permanent. Dot-escaped map keys pass through untouched, since both backends store them escaped. Executions are validated before serializing: an empty string escapes to the COPY null sentinel, so an execution missing a name would otherwise abort a whole batch rather than one row. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
vsukhin
force-pushed
the
vsukhin/feature/mongo-postgres-convert
branch
from
August 26, 2026 14:20
2dbd107 to
f85e805
Compare
Collaborator
Author
Contributor
Greptile SummaryThe PR adds a resumable MongoDB-to-PostgreSQL conversion tool, container build, Helm Job, checkpoint schema, verification, and migration tests. The previously reported trailing-failure checkpoint issue is addressed: checkpoint-only final commits advance past invalid documents, and reruns do not process them again. Confidence Score: 5/5The PR appears safe to merge. No blocking failure remains. Important Files Changed
Sequence DiagramsequenceDiagram
participant Job as Convert Job
participant Mongo as MongoDB
participant Converter as Converter
participant Postgres as PostgreSQL
Job->>Converter: Start conversion
Converter->>Postgres: Load checkpoint
Converter->>Mongo: Read documents after last_mongo_id
loop Each batch
Converter->>Converter: Map valid documents and record failures
Converter->>Postgres: Begin transaction
Converter->>Postgres: Copy mapped rows
Converter->>Postgres: Save advanced checkpoint
Converter->>Postgres: Commit transaction
end
Converter->>Postgres: Mark task complete
Converter->>Postgres: Verify migrated counts
Converter-->>Job: Report result
Reviews (4): Last reviewed commit: "fix: values" | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
flush() committed only when the batch held staged rows, but batchEnd also advances for a document --skip-errors declined to migrate. So a run of invalid documents that ended a batch - either falling after the last batch that carried rows, or making up the whole run - moved the position without anything committing it, and the task was marked complete with an older checkpoint. The next run then resumed before those documents, read them again, failed on them again, and reported the same failures. Since a failed record makes the run exit non-zero, a Job with a backoffLimit retried to no purpose and never converged. Gate the flush on whether the position has moved rather than on whether rows were staged. A batch with nothing to copy still commits its checkpoint, so declined documents are recorded once and stepped over from then on, which is what --skip-errors asks for. A commit that carries no rows is no longer counted as a batch, or the statistics would report work that did not happen. The completed flag is still not consulted when resuming, and now says why: the stored position already makes a finished run a no-op, whereas honouring the flag would refuse executions that arrived afterwards - which is exactly the case when an operator converts ahead of the final cutover. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Collaborator
Author
Contributor
There was a problem hiding this comment.
Pull request overview
Adds an OSS-facing MongoDB → PostgreSQL conversion tool and deployment plumbing to migrate historical TestWorkflow executions and execution counters during DB cutover, with resumability via Postgres checkpoints.
Changes:
- Introduces
cmd/convertandpkg/convertmigrators (executions + sequences) with checkpointing and verification. - Adds
convert_checkpointsPostgres migration to persist per-task progress and make runs resumable/idempotent. - Wires build/release + Helm support: Makefile target, Docker build target/image, GitHub workflow tags, and a Helm Job template/values.
Reviewed changes
Copilot reviewed 22 out of 24 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| pkg/database/postgres/sqlc/models.go | Adds a ConvertCheckpoint type in a sqlc-generated models file. |
| pkg/database/postgres/migrations/20260826120000_convert_checkpoints.sql | Creates convert_checkpoints table used by the converter for crash-safe progress tracking. |
| pkg/convert/stats.go | Per-task stats aggregation and printing. |
| pkg/convert/sequences.go | Migrates Mongo sequences counters into execution_sequences with idempotent upsert semantics. |
| pkg/convert/sequences_test.go | Unit tests for sequence name resolution and fold behavior. |
| pkg/convert/executiontime.go | Repairs legacy execution timestamps/durations to avoid epoch/overflow artifacts. |
| pkg/convert/executions.go | Batch migrator for executions with exactly-once checkpoint commit semantics. |
| pkg/convert/executions_row.go | COPY-row serializers and column order definitions for the target tables. |
| pkg/convert/executions_row_test.go | Unit tests for COPY serialization correctness and escaping. |
| pkg/convert/copyutil.go | COPY helpers: escaping, JSON sanitization, timestamp formatting, and COPY streaming. |
| pkg/convert/copyutil_test.go | Unit tests for escaping/sanitization/timestamp formatting. |
| pkg/convert/convert.go | Orchestrates tasks, reset, verify, and run summary reporting. |
| pkg/convert/convert_integration_test.go | Integration tests asserting end-to-end migration correctness and idempotency/resume behavior. |
| pkg/convert/checkpoint.go | Checkpoint load/save/complete/clear helpers against convert_checkpoints. |
| pkg/convert/checkpoint_test.go | Unit test for batch counting semantics. |
| pkg/controlplane/scheduling/sqlc/models.go | Adds a ConvertCheckpoint type in a sqlc-generated models file. |
| Makefile | Adds build-convert and includes it in make build. |
| k8s/helm/testkube/values.yaml | Adds convert: values block for enabling/configuring the conversion Job. |
| k8s/helm/testkube/templates/convert-job.yaml | New Helm Job template to run the convert tool during cutover. |
| k8s/helm/testkube/templates/_helpers.tpl | Adds helper functions for convert Job naming/labels/image/env wiring. |
| docker-bake.hcl | Adds a convert bake target using build/new/convert.Dockerfile. |
| cmd/convert/main.go | CLI entrypoint for the conversion tool (flags/envs, DB connect, migrations, execution). |
| build/new/convert.Dockerfile | Builds and packages the convert binary into an Alpine-based runtime image. |
| .github/workflows/new-build.yaml | Adds build+publish metadata/targets for the convert image in CI. |
Files not reviewed (2)
- pkg/controlplane/scheduling/sqlc/models.go: Generated file
- pkg/database/postgres/sqlc/models.go: Generated file
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 22 out of 24 changed files in this pull request and generated 8 comments.
Files not reviewed (2)
- pkg/controlplane/scheduling/sqlc/models.go: Generated file
- pkg/database/postgres/sqlc/models.go: Generated file
Both migrators deferred stats.finish() and then called it again before printing, so the lifecycle read as though Print needed a settled EndTime to report a duration. It does not: Duration falls back to the time elapsed so far precisely so a caller can defer its single finish call, and the deferred call already covers every return path. Drop the explicit calls and say so on Duration, so the fallback is not mistaken for an oversight and the second call reintroduced. Also collapse the blank line left where ConvertCheckpoint was removed from the two generated model files, which left them failing gofmt. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
ConvertCheckpoint had been removed from both generated model files by hand, but the goose migration still creates convert_checkpoints and sqlc emits a model for every table in the schema, so the removal did not survive generation - the next `make generate` would have put it back, and until then the generated files disagreed with their generator. Regenerate so they match again. The struct is unused, since the convert tool reaches convert_checkpoints through hand-written SQL rather than sqlc, but a generated file is not the place to prune dead code: keeping it out means either dropping the migration and creating the table from the tool, or teaching sqlc to exclude it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Two leaks on the way out, both of which only show up on the path a Job actually takes when it is told to stop. The *sql.DB wrapping the pool for goose was never closed. It borrows connections from the pool and runs an opener goroutine, and pgx documents that closing it leaves the pool alone - which is what makes closing it safe here, since the pool is what the function returns. The repository's own Postgres test helper already closes its wrapper; this now matches. The deferred Mongo disconnect ran on the main context, which by then has been cancelled by the SIGINT/SIGTERM handler set up a few lines above. A cancelled context makes Disconnect give up before closing anything, so the one case the defer exists for - shutting down on a signal - was the case it did not cover. Cleanup gets a fresh context with its own deadline, bounded so a hung close cannot hold the pod open. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Every iteration unmarshalled the document once into a one-field struct to get _id and again into the execution. The first decode still walks the whole document - resolvedWorkflow and every step result included - to reach a field sitting at its head. Look _id up in the raw bson instead. MongoDB writes it first, so the lookup stops almost immediately, and it neither allocates nor leaves bson. Measured on a document carrying two workflow snapshots and 200 step results: 50ns and no allocations, against 1421ns and 12 allocations for the decode it replaces. The lookup also lets a bad _id be reported precisely rather than as a decode failure, and it stays fatal whatever --skip-errors says: the checkpoint stores the position as a hex ObjectID, so an _id it cannot represent leaves no resume point, and continuing would checkpoint the wrong place. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 23 out of 25 changed files in this pull request and generated no new comments.
Files not reviewed (2)
- pkg/controlplane/scheduling/sqlc/models.go: Generated file
- pkg/database/postgres/sqlc/models.go: Generated file
Suppressed comments (4)
Previously missed (2) — in code that hasn't changed since the last review.
pkg/convert/convert.go:238
- Verification uses per-run Failed/Skipped counts to compute the expected Postgres row count. On a resumed run, prior failures/skips are stored only in the checkpoint, so this can produce a false warning (and mark the run failed) even though there is nothing left to migrate.
This issue also appears on line 258 of the same file.
expected := stats.Total - stats.Failed - stats.Skipped
c.log.Infof("Executions: %d in mongo, %d in postgres (%d expected)", stats.Total, pgCount, expected)
if pgCount < expected {
warnings = append(warnings, fmt.Sprintf(
"executions: postgres holds %d rows but %d were expected (%d in mongo, %d failed, %d skipped)",
pkg/convert/stats.go:58
- Stats keeps every individual error string in memory. With --skip-errors on a large/dirty dataset this can grow without bound and potentially OOM the Job even though only a capped subset is ever reported.
This issue also appears on line 113 of the same file.
func (s *Stats) addError(format string, args ...interface{}) {
s.Failed++
s.Errors = append(s.Errors, fmt.Sprintf(format, args...))
}
pkg/convert/convert.go:262
- Same as executions: verification derives the expected count from per-run counters. If the task is re-run after a previous run recorded failures/skips in the checkpoint, verification can warn incorrectly.
expected := stats.Total - stats.Failed - stats.Skipped
c.log.Infof("Sequences: %d in mongo, %d in postgres (%d migratable)", stats.Total, pgCount, expected)
if pgCount < expected {
warnings = append(warnings, fmt.Sprintf(
"sequences: postgres holds %d rows but %d were expected (%d in mongo, %d skipped as non-testworkflow)",
pkg/convert/stats.go:124
- After capping stored error messages, Print should report total error count using the Failed counter (not len(Errors)) and still show how many additional errors were omitted.
if len(s.Errors) == 0 {
return
}
log.Warnf("Encountered %d errors (showing up to %d):", len(s.Errors), maxReportedErrors)
for i, e := range s.Errors {
if i >= maxReportedErrors {
log.Warnf("... and %d more", len(s.Errors)-maxReportedErrors)
break
}
log.Warnf(" %d. %s", i+1, e)
}
Collaborator
Author
Verification compares the whole source collection against the whole target table, but derived the expected row count from the counters of the run that had just finished. Those describe only the documents that run read. A resumed run with nothing left to do reported zero failures, so every document an earlier run had declined looked like a missing row: a warning, a non-zero exit, and a Job that retried a migration which was already complete. Report the totals across all runs alongside the per-run ones and verify against those. The two tasks need different totals, so each says which it means rather than sharing a rule that happens to suit one of them: executions resume from a position and add the checkpoint's totals to their own, while sequences reread the whole collection every time and must not, or the same skipped documents would be counted once per run and drive the expected count negative. Result.Failed still keys off the per-run counters, so a run that had nothing to do and nothing go wrong exits zero even when earlier runs recorded failures. There is nothing left for a retry to achieve. Retained error messages are now capped. A damaged collection run with --skip-errors kept one string per failed document in a process that is already holding a batch. Failed stays exact and Print reports it rather than the number of messages kept, so the summary cannot understate the damage, and the omitted tail is counted against the true total. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 24 out of 26 changed files in this pull request and generated 1 comment.
Files not reviewed (2)
- pkg/controlplane/scheduling/sqlc/models.go: Generated file
- pkg/database/postgres/sqlc/models.go: Generated file
Suppressed comments (1)
Previously missed (1) — in code that hasn't changed since the last review.
pkg/convert/copyutil.go:56
escapeJSONBusesstring(data) == "null", which allocates a new string for every JSONB field serialized. This is on the hot path for execution migration (many JSONB columns per row). Prefer a byte comparison to avoid per-call allocations.
func escapeJSONB(data []byte) string {
if len(data) == 0 || string(data) == "null" {
return copyNull
}
return escapeCopySpecials(string(sanitizeJSONForPG(data)))
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 24 out of 26 changed files in this pull request and generated 1 comment.
Files not reviewed (2)
- pkg/controlplane/scheduling/sqlc/models.go: Generated file
- pkg/database/postgres/sqlc/models.go: Generated file
Suppressed comments (1)
Previously missed (1) — in code that hasn't changed since the last review.
pkg/convert/executions.go:439
- In executions checkpointing, cp.Completed is never cleared during a re-run. If a previous run set completed_at, then a later run that migrates new documents (or is interrupted mid-run) will keep writing completed_at as non-NULL in saveCheckpoint, making the checkpoint look "completed" even while work is in progress or incomplete.
cp.Processed = m.priorProcessed + stats.Processed + counts.executions
cp.Failed = m.priorFailed + stats.Failed
cp.Skipped = m.priorSkipped + stats.Skipped
if err := saveCheckpoint(ctx, tx, cp); err != nil {
return err
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 24 out of 26 changed files in this pull request and generated no new comments.
Files not reviewed (2)
- pkg/controlplane/scheduling/sqlc/models.go: Generated file
- pkg/database/postgres/sqlc/models.go: Generated file
Suppressed comments (1)
Previously missed (1) — in code that hasn't changed since the last review.
pkg/convert/sequences.go:199
- When there are zero migratable rows, write() returns early without persisting the checkpoint. That means a collection containing only non-testworkflow/legacy sequences will be re-scanned and re-warned on every run, and the task will never be marked completed in convert_checkpoints.
if len(rows) == 0 {
m.log.Info("No test workflow sequences to migrate")
return nil
}
Signed-off-by: Vladislav Sukhin <vladislav@kubeshop.io>
Signed-off-by: Vladislav Sukhin <vladislav@kubeshop.io>
The integration fixture built a workflow snapshot with no Spec, and TestWorkflow.ConvertDots guards a nil workflow but then reaches through its Spec unguarded. So the Mongo repository's Insert, which escapes dots before writing, segfaulted while the fixture was still being seeded and took the whole package's tests down with it. Real snapshots always carry a spec, so the fixture was describing data no installation holds. Give it one, and put a dotted label inside it while we are here: the spec travels as a single JSONB column, so that also covers escaping surviving a round trip through it. Two tests now pin this without needing a database. One asserts the fixture survives the escaping and unescaping the repositories apply, which is the call that panicked; the other asserts the escaping round-trips and that the fixture really does contain dotted keys, so it cannot quietly stop testing anything. Waiting for the integration job to catch a fixture that does not match the model was the actual mistake. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Two differences the cross-backend comparison was reading as migration bugs. Timestamps came back in different locations: BSON carries no zone, so the Mongo driver decodes in the local one, while pgx returns TIMESTAMPTZ in UTC. Both name the same instant, but reflect-based equality compares the Location pointer too, so the comparison could only have passed on a machine whose local zone is UTC - and there for the wrong reason. Compare with go-cmp and a comparer that asks whether two times are equal, which also names the differing field instead of printing both executions in full. The fixture also carried an empty resource aggregations report, whose two JSONB columns are both NULL. PostgresRepository.Get rebuilds the report only when one of them holds something, so it came back nil there and intact from Mongo. PostgresRepository.Insert stores an empty report exactly as the migrator does, so this is the repositories disagreeing about an absent child, not the migration losing one. Give the fixture real aggregations. Reports has the same shape of disagreement in the other direction: MongoRepository.Insert defaults a nil Reports to an empty slice, so Mongo serves [] where Postgres serves nil. Nothing trips it today because the fixture populates reports, so one test now pins the rule for every child collection and records why, rather than leaving the next minimal execution added to that comparison to rediscover it. Both fixes are pinned by tests that need no database: one that the comparer ignores location but not instant, one that the fixture populates every child. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
markCheckpointComplete wrote unconditionally, so a dry run left a checkpoint row behind claiming the task had finished. A later real run would then resume from a position nothing had been migrated to, and skip everything up to it. Every other writer on this path already declines a dry run itself; this one being the exception is what let the write through. Give it the flag too, rather than adding a fourth caller-side check to a guard that has now been missed once. Audited the rest while here. The two truncates and the checkpoint clear are reachable only through reset, which returns early on a dry run, and all three saveCheckpoint calls sit inside guarded functions, so the data path is now covered. Schema is the deliberate exception and both the field and the flag now say so: the checkpoint lookup and the verification counts read tables that have to exist first, so a dry run still creates the database and applies migrations. The dry-run test now asserts every table the migration touches is empty, convert_checkpoints included. Checking only the tables a previous leak touched is how the next leak gets missed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
✅ Testkube GitHub IntegrationReview based on commit All tests and quality gates passed.
7 workflows executed
|
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
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.
Pull request description
Checklist (choose whats happened)
Breaking changes
Changes
Fixes