From 1bdcd446db558b6197a12b8c3e55a0acc37e2e51 Mon Sep 17 00:00:00 2001 From: HsiuChuanHsu Date: Sun, 16 Aug 2026 15:17:40 +0800 Subject: [PATCH 1/3] fix: compress offloaded node status to reduce database write load. Fixes #13290 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: 刘达 Signed-off-by: HsiuChuanHsu --- docs/database-migrations.md | 12 ++ persist/sqldb/migrate.go | 6 + persist/sqldb/offload_node_status_repo.go | 27 ++- .../offload_node_status_repo_mysql_test.go | 170 ++++++++++++++++++ 4 files changed, 211 insertions(+), 4 deletions(-) create mode 100644 persist/sqldb/offload_node_status_repo_mysql_test.go diff --git a/docs/database-migrations.md b/docs/database-migrations.md index e1e5f42cbbc0..05a294f744e0 100644 --- a/docs/database-migrations.md +++ b/docs/database-migrations.md @@ -255,6 +255,12 @@ drop index argo_archived_workflows_i1 on argo_archived_workflows; -- Step 68 create index argo_archived_workflows_i1 on argo_archived_workflows (clustername, instanceid, namespace, startedat DESC); +-- Step 69 +alter table argo_workflows add column compressednodes longtext; + +-- Step 70 +update argo_workflows set compressednodes = '' where compressednodes is null; + ``` ### PostgreSQL @@ -493,6 +499,12 @@ drop index argo_archived_workflows_i1; -- Step 68 create index argo_archived_workflows_i1 on argo_archived_workflows (clustername, instanceid, namespace, startedat DESC); +-- Step 69 +alter table argo_workflows add column compressednodes text; + +-- Step 70 +update argo_workflows set compressednodes = '' where compressednodes is null; + ``` ## Sync Database diff --git a/persist/sqldb/migrate.go b/persist/sqldb/migrate.go index 72ce7e43f388..ac17f8f41d59 100644 --- a/persist/sqldb/migrate.go +++ b/persist/sqldb/migrate.go @@ -231,6 +231,12 @@ func MigrateChanges(clusterName, tableName string, dbType sqldb.DBType) []sqldb. sqldb.Postgres: sqldb.AnsiSQLChange(`drop index argo_archived_workflows_i1`), }), sqldb.AnsiSQLChange(`create index argo_archived_workflows_i1 on argo_archived_workflows (clustername, instanceid, namespace, startedat DESC)`), + // store compressed node status to cut the volume written on offload. + sqldb.ByType(dbType, sqldb.TypedChanges{ + sqldb.MySQL: sqldb.AnsiSQLChange(`alter table ` + tableName + ` add column compressednodes longtext`), + sqldb.Postgres: sqldb.AnsiSQLChange(`alter table ` + tableName + ` add column compressednodes text`), + }), + sqldb.AnsiSQLChange(`update ` + tableName + ` set compressednodes = '' where compressednodes is null`), } } diff --git a/persist/sqldb/offload_node_status_repo.go b/persist/sqldb/offload_node_status_repo.go index 6a0218bc4213..7756f2daca71 100644 --- a/persist/sqldb/offload_node_status_repo.go +++ b/persist/sqldb/offload_node_status_repo.go @@ -12,6 +12,7 @@ import ( wfv1 "github.com/argoproj/argo-workflows/v4/pkg/apis/workflow/v1alpha1" "github.com/argoproj/argo-workflows/v4/util/env" + "github.com/argoproj/argo-workflows/v4/util/file" "github.com/argoproj/argo-workflows/v4/util/logging" "github.com/argoproj/argo-workflows/v4/util/sqldb" ) @@ -45,6 +46,8 @@ type nodesRecord struct { UUIDVersion Namespace string `db:"namespace"` Nodes string `db:"nodes"` + // Base64-encoded compressed node status; empty on legacy rows written before compression. + CompressedNodes string `db:"compressednodes"` } type nodeOffloadRepo struct { @@ -84,7 +87,9 @@ func (wdc *nodeOffloadRepo) Save(ctx context.Context, uid, namespace string, nod Version: version, }, Namespace: namespace, - Nodes: marshalled, + // nodes is json not null; payload actually lives in CompressedNodes. + Nodes: "null", + CompressedNodes: file.CompressEncodeString(ctx, marshalled), } logCtx := wdc.log.WithFields(logging.Fields{"uid": uid, "version": version}) @@ -135,8 +140,15 @@ func (wdc *nodeOffloadRepo) Get(ctx context.Context, uid, version string) (wfv1. if err != nil { return err } + nodesJSON := r.Nodes + if r.CompressedNodes != "" { + nodesJSON, err = file.DecodeDecompressString(ctx, r.CompressedNodes) + if err != nil { + return err + } + } n := &wfv1.Nodes{} - err = json.Unmarshal([]byte(r.Nodes), n) + err = json.Unmarshal([]byte(nodesJSON), n) if err != nil { return err } @@ -155,7 +167,7 @@ func (wdc *nodeOffloadRepo) List(ctx context.Context, namespace string) (map[UUI err := wdc.sessionProxy.With(ctx, func(s db.Session) error { var records []nodesRecord err := s.SQL(). - Select("uid", "version", "nodes"). + Select("uid", "version", "nodes", "compressednodes"). From(wdc.tableName). Where(db.Cond{"clustername": wdc.clusterName}). And(namespaceEqual(namespace)). @@ -166,8 +178,15 @@ func (wdc *nodeOffloadRepo) List(ctx context.Context, namespace string) (map[UUI res = make(map[UUIDVersion]wfv1.Nodes) for _, r := range records { + nodesJSON := r.Nodes + if r.CompressedNodes != "" { + nodesJSON, err = file.DecodeDecompressString(ctx, r.CompressedNodes) + if err != nil { + return err + } + } nodes := &wfv1.Nodes{} - err = json.Unmarshal([]byte(r.Nodes), nodes) + err = json.Unmarshal([]byte(nodesJSON), nodes) if err != nil { return err } diff --git a/persist/sqldb/offload_node_status_repo_mysql_test.go b/persist/sqldb/offload_node_status_repo_mysql_test.go new file mode 100644 index 000000000000..d0b1d532ec62 --- /dev/null +++ b/persist/sqldb/offload_node_status_repo_mysql_test.go @@ -0,0 +1,170 @@ +//go:build !windows + +package sqldb + +// Integration tests for #13290: offload now stores node status COMPRESSED in the +// compressednodes column, cutting the volume written on every update of a large +// workflow. The MySQL 8.4 container pins max_allowed_packet to 16MB, which also makes +// the size reduction observable: a Save of ~13MB of raw nodes only fits once compressed. + +import ( + "context" + "encoding/json" + "fmt" + "strconv" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + testcontainers "github.com/testcontainers/testcontainers-go" + testmysql "github.com/testcontainers/testcontainers-go/modules/mysql" + "github.com/testcontainers/testcontainers-go/wait" + "github.com/upper/db/v4" + + "github.com/argoproj/argo-workflows/v4/config" + wfv1 "github.com/argoproj/argo-workflows/v4/pkg/apis/workflow/v1alpha1" + "github.com/argoproj/argo-workflows/v4/util/logging" + usqldb "github.com/argoproj/argo-workflows/v4/util/sqldb" +) + +// setupOffloadRepo starts MySQL 8.4 with max_allowed_packet pinned to 16MB, migrates the +// argo_workflows offload table, and returns the offload repo plus the session proxy. +func setupOffloadRepo(ctx context.Context, t *testing.T) (OffloadNodeStatusRepo, *usqldb.SessionProxy) { + t.Helper() + + c, err := testmysql.Run(ctx, + "mysql:8.4", + testmysql.WithDatabase("argo"), + testmysql.WithUsername("argo"), + testmysql.WithPassword("argo"), + // Pin the ceiling to 16MB so ~13MB raw nodes would fail pre-fix but pass compressed. + testcontainers.WithCmdArgs("--max-allowed-packet=16777216"), + testcontainers.WithWaitStrategy( + wait.ForAll( + wait.ForLog("port: 3306 MySQL Community Server").WithStartupTimeout(120*time.Second), + wait.ForListeningPort("3306/tcp"), + )), + ) + require.NoError(t, err) + t.Cleanup(func() { + if termErr := testcontainers.TerminateContainer(c); termErr != nil { + t.Logf("failed to terminate container: %s", termErr) + } + }) + + host, err := c.Host(ctx) + require.NoError(t, err) + p, err := c.MappedPort(ctx, "3306/tcp") + require.NoError(t, err) + port, err := strconv.Atoi(p.Port()) + require.NoError(t, err) + + proxy, err := usqldb.NewSessionProxy(ctx, usqldb.SessionProxyConfig{ + DBConfig: config.DBConfig{ + MySQL: &config.MySQLConfig{ + DatabaseConfig: config.DatabaseConfig{Database: "argo", Host: host, Port: port}, + }, + }, + Username: "argo", + Password: "argo", + }) + require.NoError(t, err) + t.Cleanup(func() { proxy.Close() }) + + require.NoError(t, Migrate(ctx, proxy.Session(), "test", "argo_workflows", proxy.DBType())) + + repo, err := NewOffloadNodeStatusRepo(ctx, logging.RequireLoggerFromContext(ctx), proxy, "test", "argo_workflows") + require.NoError(t, err) + return repo, proxy +} + +// makeNodes builds a wfv1.Nodes whose marshalled JSON is >= target bytes. +func makeNodes(t *testing.T, target int) wfv1.Nodes { + t.Helper() + nodes := wfv1.Nodes{} + chunk := strings.Repeat("x", 64*1024) // 64KB per node + i := 0 + for { + id := fmt.Sprintf("node-%06d", i) + nodes[id] = wfv1.NodeStatus{ID: id, Name: id, Message: chunk} + i++ + if i%16 == 0 { + b, err := json.Marshal(nodes) + require.NoError(t, err) + if len(b) >= target { + return nodes + } + } + } +} + +const mb = 1 << 20 + +// TestOffloadCompression_RoundTrip verifies a ~13MB node status (which pre-fix exceeded the +// 16MB packet ceiling once expanded) now saves compressed, round-trips via Get, and is stored +// with the raw nodes column holding only the "null" placeholder. +func TestOffloadCompression_RoundTrip(t *testing.T) { + ctx := logging.TestContext(t.Context()) + repo, proxy := setupOffloadRepo(ctx, t) + + nodes := makeNodes(t, 13*mb) + uid := "uid-roundtrip" + + version, err := repo.Save(ctx, uid, "default", nodes) + require.NoError(t, err, "compressed Save of ~13MB nodes should succeed under 16MB max_allowed_packet") + + got, err := repo.Get(ctx, uid, version) + require.NoError(t, err) + assert.Equal(t, nodes, got, "Get must return the original nodes") + + // Storage format: compressed payload present, raw nodes column is the placeholder. + r := fetchRow(ctx, t, proxy, uid, version) + assert.NotEmpty(t, r.CompressedNodes, "compressednodes should hold the compressed payload") + assert.Equal(t, "null", r.Nodes, "nodes column should be the json null placeholder") + assert.Less(t, len(r.CompressedNodes), 13*mb, "stored compressed payload should be far smaller than raw") +} + +// TestOffloadCompression_BackwardCompat verifies that a legacy row (raw JSON in nodes, +// empty compressednodes) still reads correctly via both Get and List after the change. +func TestOffloadCompression_BackwardCompat(t *testing.T) { + ctx := logging.TestContext(t.Context()) + repo, proxy := setupOffloadRepo(ctx, t) + + legacyNodes := wfv1.Nodes{"n1": wfv1.NodeStatus{ID: "n1", Name: "n1", Phase: wfv1.NodeSucceeded}} + raw, err := json.Marshal(legacyNodes) + require.NoError(t, err) + + uid, version := "uid-legacy", "fnv:legacy" + err = proxy.With(ctx, func(s db.Session) error { + _, insErr := s.Collection("argo_workflows").Insert(&nodesRecord{ + ClusterName: "test", + UUIDVersion: UUIDVersion{UID: uid, Version: version}, + Namespace: "default", + Nodes: string(raw), + CompressedNodes: "", // legacy: no compression + }) + return insErr + }) + require.NoError(t, err) + + got, err := repo.Get(ctx, uid, version) + require.NoError(t, err) + assert.Equal(t, legacyNodes, got, "Get must read legacy uncompressed rows") + + list, err := repo.List(ctx, "default") + require.NoError(t, err) + assert.Equal(t, legacyNodes, list[UUIDVersion{UID: uid, Version: version}], "List must read legacy uncompressed rows") +} + +func fetchRow(ctx context.Context, t *testing.T, proxy *usqldb.SessionProxy, uid, version string) nodesRecord { + t.Helper() + var r nodesRecord + err := proxy.With(ctx, func(s db.Session) error { + return s.SQL().SelectFrom("argo_workflows"). + Where(db.Cond{"uid": uid}).And(db.Cond{"version": version}).One(&r) + }) + require.NoError(t, err) + return r +} From 8fa15d7d9c5f951b2cbb8b47ab3ac4c04f968a71 Mon Sep 17 00:00:00 2001 From: HsiuChuanHsu Date: Mon, 17 Aug 2026 09:13:41 +0800 Subject: [PATCH 2/3] docs: document offload compression and its rolling-upgrade constraint Signed-off-by: HsiuChuanHsu Co-Authored-By: Claude Opus 5 --- docs/offloading-large-workflows.md | 11 ++++++++++- docs/upgrading.md | 14 ++++++++++++++ 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/docs/offloading-large-workflows.md b/docs/offloading-large-workflows.md index 41273eb7640a..885cece9c031 100644 --- a/docs/offloading-large-workflows.md +++ b/docs/offloading-large-workflows.md @@ -2,7 +2,7 @@ > v2.4 and after -Argo stores workflows as Kubernetes resources (i.e. within EtcD). This creates a limit to their size as resources must be under 1MB. Each resource includes the status of each node, which is stored in the `/status/nodes` field for the resource. This can be over 1MB. If this happens, we try and compress the node status and store it in `/status/compressedNodes`. If the status is still too large, we then try and store it in an SQL database. +Argo stores workflows as Kubernetes resources (i.e. within EtcD). This creates a limit to their size as resources must be under 1MB. Each resource includes the status of each node, which is stored in the `/status/nodes` field for the resource. This can be over 1MB. If this happens, we try and compress the node status and store it in `/status/compressedNodes`. If the status is still too large, we then try and store it in an SQL database. The offloaded node status is itself stored compressed, which reduces the volume written to the database on every update of a large workflow. To enable this feature, configure a Postgres, MySQL, or MariaDB database under `persistence` in [your configuration](workflow-controller-configmap.yaml) and set `nodeStatusOffLoad: true`. @@ -32,6 +32,15 @@ Decompression speed is roughly equal for all three algorithms (about 9 milliseco These numbers are from synthetic data; real workflows may compress differently. +### Offloaded Node Status + +> v4.2 and after + +Node status offloaded to the database is stored compressed in the `compressednodes` column of the `argo_workflows` table, using the algorithm selected above. Rows written before this version keep their uncompressed JSON in the `nodes` column and are read back unchanged, so no data migration is needed. + +!!! Warning + A row written by a controller with this support stores the placeholder `null` in the `nodes` column. A controller without it reads that row as an empty node status. Upgrade every controller sharing a database together, and do not roll back past this version while offloaded workflows are still live. See [Upgrading](upgrading.md#offloaded-node-status-is-stored-compressed). + ## FAQ ### Why aren't my workflows appearing in the database? diff --git a/docs/upgrading.md b/docs/upgrading.md index 1104e3db4440..703b1d94f09b 100644 --- a/docs/upgrading.md +++ b/docs/upgrading.md @@ -5,6 +5,20 @@ For the upgrading guide to a specific version of workflows change the documentat Breaking changes typically (sometimes we don't realise they are breaking) have "!" in the commit message, as per the [conventional commits](https://www.conventionalcommits.org/en/v1.0.0/#summary). +## Upgrading to v4.2 + +### Offloaded node status is stored compressed + +When node status offloading is enabled, the controller now compresses the node status before writing it to the database, storing it in a new `compressednodes` column on the `argo_workflows` table ([#13290](https://github.com/argoproj/argo-workflows/issues/13290)). +This reduces the volume written on every update of a large workflow, which is the dominant cost at scale. +The migration adds the column; no existing data is rewritten. + +Rows written before the upgrade keep their uncompressed JSON in the `nodes` column and are read back unchanged, so a mixed table is fine and there is nothing to migrate. + +A row written *after* the upgrade stores the placeholder `null` in `nodes`, with the real payload in `compressednodes`. +A controller that predates this change reads such a row as an empty node status. +Upgrade every controller sharing a database at the same time, and do not roll back past this version while offloaded workflows are still live. + ## Upgrading to v4.1 ### Controller caches no longer store `managedFields` From 352cfdf15b8169d03a912d215d79b3eab0d073bf Mon Sep 17 00:00:00 2001 From: HsiuChuanHsu Date: Wed, 19 Aug 2026 22:23:13 +0800 Subject: [PATCH 3/3] fix(sqldb): read offloaded rows whose compressednodes is null The backfill in the offload migration runs once. During a rolling upgrade an older replica can insert a row after it has run, leaving compressednodes NULL rather than the empty string, and scanning NULL into a string fails, so Get and List could not read that row back. Make the column sql.NullString and treat NULL the same as empty, which is the legacy shape whose payload lives in nodes. Cover both shapes on MySQL and Postgres: the drivers scan NULL through different code, so one engine is not evidence for the other. Co-Authored-By: Claude Opus 5 Signed-off-by: HsiuChuanHsu --- persist/sqldb/offload_node_status_repo.go | 35 ++-- .../offload_node_status_repo_mysql_test.go | 81 ++++++--- .../offload_node_status_repo_postgres_test.go | 172 ++++++++++++++++++ 3 files changed, 248 insertions(+), 40 deletions(-) create mode 100644 persist/sqldb/offload_node_status_repo_postgres_test.go diff --git a/persist/sqldb/offload_node_status_repo.go b/persist/sqldb/offload_node_status_repo.go index 7756f2daca71..3ffce7dd1805 100644 --- a/persist/sqldb/offload_node_status_repo.go +++ b/persist/sqldb/offload_node_status_repo.go @@ -2,6 +2,7 @@ package sqldb import ( "context" + "database/sql" "encoding/json" "fmt" "hash/fnv" @@ -46,8 +47,18 @@ type nodesRecord struct { UUIDVersion Namespace string `db:"namespace"` Nodes string `db:"nodes"` - // Base64-encoded compressed node status; empty on legacy rows written before compression. - CompressedNodes string `db:"compressednodes"` + // Base64-encoded compressed node status. Nullable: an old replica can insert after + // the one-shot backfill (migrate.go:236-239), landing NULL rather than ''. + CompressedNodes sql.NullString `db:"compressednodes"` +} + +// nodesJSON returns the node status as JSON. A NULL or empty compressednodes means a +// legacy row written before compression, whose payload is in nodes. +func (r nodesRecord) nodesJSON(ctx context.Context) (string, error) { + if r.CompressedNodes.String == "" { + return r.Nodes, nil + } + return file.DecodeDecompressString(ctx, r.CompressedNodes.String) } type nodeOffloadRepo struct { @@ -89,7 +100,7 @@ func (wdc *nodeOffloadRepo) Save(ctx context.Context, uid, namespace string, nod Namespace: namespace, // nodes is json not null; payload actually lives in CompressedNodes. Nodes: "null", - CompressedNodes: file.CompressEncodeString(ctx, marshalled), + CompressedNodes: sql.NullString{String: file.CompressEncodeString(ctx, marshalled), Valid: true}, } logCtx := wdc.log.WithFields(logging.Fields{"uid": uid, "version": version}) @@ -140,12 +151,9 @@ func (wdc *nodeOffloadRepo) Get(ctx context.Context, uid, version string) (wfv1. if err != nil { return err } - nodesJSON := r.Nodes - if r.CompressedNodes != "" { - nodesJSON, err = file.DecodeDecompressString(ctx, r.CompressedNodes) - if err != nil { - return err - } + nodesJSON, err := r.nodesJSON(ctx) + if err != nil { + return err } n := &wfv1.Nodes{} err = json.Unmarshal([]byte(nodesJSON), n) @@ -178,12 +186,9 @@ func (wdc *nodeOffloadRepo) List(ctx context.Context, namespace string) (map[UUI res = make(map[UUIDVersion]wfv1.Nodes) for _, r := range records { - nodesJSON := r.Nodes - if r.CompressedNodes != "" { - nodesJSON, err = file.DecodeDecompressString(ctx, r.CompressedNodes) - if err != nil { - return err - } + nodesJSON, err := r.nodesJSON(ctx) + if err != nil { + return err } nodes := &wfv1.Nodes{} err = json.Unmarshal([]byte(nodesJSON), nodes) diff --git a/persist/sqldb/offload_node_status_repo_mysql_test.go b/persist/sqldb/offload_node_status_repo_mysql_test.go index d0b1d532ec62..cf1662b78cfc 100644 --- a/persist/sqldb/offload_node_status_repo_mysql_test.go +++ b/persist/sqldb/offload_node_status_repo_mysql_test.go @@ -9,6 +9,7 @@ package sqldb import ( "context" + "database/sql" "encoding/json" "fmt" "strconv" @@ -31,7 +32,7 @@ import ( // setupOffloadRepo starts MySQL 8.4 with max_allowed_packet pinned to 16MB, migrates the // argo_workflows offload table, and returns the offload repo plus the session proxy. -func setupOffloadRepo(ctx context.Context, t *testing.T) (OffloadNodeStatusRepo, *usqldb.SessionProxy) { +func setupOffloadRepo(ctx context.Context, t testing.TB) (OffloadNodeStatusRepo, *usqldb.SessionProxy) { t.Helper() c, err := testmysql.Run(ctx, @@ -121,14 +122,19 @@ func TestOffloadCompression_RoundTrip(t *testing.T) { // Storage format: compressed payload present, raw nodes column is the placeholder. r := fetchRow(ctx, t, proxy, uid, version) - assert.NotEmpty(t, r.CompressedNodes, "compressednodes should hold the compressed payload") + assert.NotEmpty(t, r.CompressedNodes.String, "compressednodes should hold the compressed payload") assert.Equal(t, "null", r.Nodes, "nodes column should be the json null placeholder") - assert.Less(t, len(r.CompressedNodes), 13*mb, "stored compressed payload should be far smaller than raw") + assert.Less(t, len(r.CompressedNodes.String), 13*mb, "stored compressed payload should be far smaller than raw") } -// TestOffloadCompression_BackwardCompat verifies that a legacy row (raw JSON in nodes, -// empty compressednodes) still reads correctly via both Get and List after the change. -func TestOffloadCompression_BackwardCompat(t *testing.T) { +// TestOffloadCompression_LegacyRows covers both shapes a pre-compression row can have, +// so an upgrade needs no data migration: the empty string the migration backfills, and a +// genuine SQL NULL an old replica writes after that one-shot backfill has already run +// (see the CompressedNodes field comment). +// +// Worth having on both engines rather than trusting one: the column is longtext here and +// text on Postgres, and the two drivers scan NULL through different code. +func TestOffloadCompression_LegacyRows(t *testing.T) { ctx := logging.TestContext(t.Context()) repo, proxy := setupOffloadRepo(ctx, t) @@ -136,26 +142,51 @@ func TestOffloadCompression_BackwardCompat(t *testing.T) { raw, err := json.Marshal(legacyNodes) require.NoError(t, err) - uid, version := "uid-legacy", "fnv:legacy" - err = proxy.With(ctx, func(s db.Session) error { - _, insErr := s.Collection("argo_workflows").Insert(&nodesRecord{ - ClusterName: "test", - UUIDVersion: UUIDVersion{UID: uid, Version: version}, - Namespace: "default", - Nodes: string(raw), - CompressedNodes: "", // legacy: no compression + const version = "fnv:legacy" + for _, tc := range []struct { + name, uid string + wantValid bool + insert func(db.Session, string) error + }{ + { + name: "backfilled empty string", uid: "uid-legacy", wantValid: true, + insert: func(s db.Session, uid string) error { + _, insErr := s.Collection("argo_workflows").Insert(&nodesRecord{ + ClusterName: "test", + UUIDVersion: UUIDVersion{UID: uid, Version: version}, + Namespace: "default", + Nodes: string(raw), + CompressedNodes: sql.NullString{String: "", Valid: true}, + }) + return insErr + }, + }, + { + // Must be raw SQL: Insert(&nodesRecord{...}) always writes a non-NULL empty + // string, so it cannot reproduce the shape an old replica leaves behind. + name: "genuine SQL NULL", uid: "uid-null", wantValid: false, + insert: func(s db.Session, uid string) error { + _, execErr := s.SQL().Exec( + "insert into argo_workflows (clustername, uid, version, namespace, nodes) values (?, ?, ?, ?, ?)", + "test", uid, version, "default", string(raw)) + return execErr + }, + }, + } { + t.Run(tc.name, func(t *testing.T) { + require.NoError(t, proxy.With(ctx, func(s db.Session) error { return tc.insert(s, tc.uid) })) + require.Equal(t, tc.wantValid, fetchRow(ctx, t, proxy, tc.uid, version).CompressedNodes.Valid, + "row must have the column shape this case claims to test") + + got, err := repo.Get(ctx, tc.uid, version) + require.NoError(t, err, "Get must read legacy rows") + assert.Equal(t, legacyNodes, got) + + list, err := repo.List(ctx, "default") + require.NoError(t, err, "List must read legacy rows") + assert.Equal(t, legacyNodes, list[UUIDVersion{UID: tc.uid, Version: version}]) }) - return insErr - }) - require.NoError(t, err) - - got, err := repo.Get(ctx, uid, version) - require.NoError(t, err) - assert.Equal(t, legacyNodes, got, "Get must read legacy uncompressed rows") - - list, err := repo.List(ctx, "default") - require.NoError(t, err) - assert.Equal(t, legacyNodes, list[UUIDVersion{UID: uid, Version: version}], "List must read legacy uncompressed rows") + } } func fetchRow(ctx context.Context, t *testing.T, proxy *usqldb.SessionProxy, uid, version string) nodesRecord { diff --git a/persist/sqldb/offload_node_status_repo_postgres_test.go b/persist/sqldb/offload_node_status_repo_postgres_test.go new file mode 100644 index 000000000000..e9b148b49dc6 --- /dev/null +++ b/persist/sqldb/offload_node_status_repo_postgres_test.go @@ -0,0 +1,172 @@ +//go:build !windows + +package sqldb + +// Postgres counterpart to offload_node_status_repo_mysql_test.go. +// +// The compression change touches two things that behave differently per database: +// Save writes the Go string "null" into the nodes column, which is `json not null` +// on both engines, and the migration adds compressednodes as `text` on Postgres +// (`longtext` on MySQL). Only the MySQL side was covered, so these tests prove the +// Postgres path. + +import ( + "context" + "database/sql" + "encoding/json" + "strconv" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + testcontainers "github.com/testcontainers/testcontainers-go" + testpostgres "github.com/testcontainers/testcontainers-go/modules/postgres" + "github.com/testcontainers/testcontainers-go/wait" + "github.com/upper/db/v4" + + "github.com/argoproj/argo-workflows/v4/config" + wfv1 "github.com/argoproj/argo-workflows/v4/pkg/apis/workflow/v1alpha1" + "github.com/argoproj/argo-workflows/v4/util/logging" + usqldb "github.com/argoproj/argo-workflows/v4/util/sqldb" +) + +// setupOffloadRepoPostgres starts Postgres, migrates the argo_workflows offload table, +// and returns the offload repo plus the session proxy. +func setupOffloadRepoPostgres(ctx context.Context, t testing.TB) (OffloadNodeStatusRepo, *usqldb.SessionProxy) { + t.Helper() + + c, err := testpostgres.Run(ctx, + "postgres:17.4-alpine", + testpostgres.WithDatabase("argo"), + testpostgres.WithUsername("argo"), + testpostgres.WithPassword("argo"), + testcontainers.WithWaitStrategy( + wait.ForLog("database system is ready to accept connections"). + WithOccurrence(2). + WithStartupTimeout(120*time.Second)), + ) + require.NoError(t, err) + t.Cleanup(func() { + if termErr := testcontainers.TerminateContainer(c); termErr != nil { + t.Logf("failed to terminate container: %s", termErr) + } + }) + + host, err := c.Host(ctx) + require.NoError(t, err) + p, err := c.MappedPort(ctx, "5432/tcp") + require.NoError(t, err) + port, err := strconv.Atoi(p.Port()) + require.NoError(t, err) + + proxy, err := usqldb.NewSessionProxy(ctx, usqldb.SessionProxyConfig{ + DBConfig: config.DBConfig{ + // SSL is left unset, which yields sslmode=disable. + PostgreSQL: &config.PostgreSQLConfig{ + DatabaseConfig: config.DatabaseConfig{Database: "argo", Host: host, Port: port}, + }, + }, + Username: "argo", + Password: "argo", + }) + require.NoError(t, err) + t.Cleanup(func() { proxy.Close() }) + + require.NoError(t, Migrate(ctx, proxy.Session(), "test", "argo_workflows", proxy.DBType())) + + repo, err := NewOffloadNodeStatusRepo(ctx, logging.RequireLoggerFromContext(ctx), proxy, "test", "argo_workflows") + require.NoError(t, err) + return repo, proxy +} + +// TestOffloadCompressionPostgres_RoundTrip verifies that the migration applies on Postgres +// and that a compressed Save round-trips through Get. +// +// 1MiB is deliberate: the MySQL test uses ~13MiB to clear the 16MB max_allowed_packet +// ceiling, but Postgres has no equivalent limit. What is unproven here is the storage +// format and the migration, and 1MiB exercises both without the runtime cost. +func TestOffloadCompressionPostgres_RoundTrip(t *testing.T) { + ctx := logging.TestContext(t.Context()) + repo, proxy := setupOffloadRepoPostgres(ctx, t) + + nodes := makeNodes(t, 1*mb) + uid := "uid-pg-roundtrip" + + version, err := repo.Save(ctx, uid, "default", nodes) + require.NoError(t, err, `Save must accept the "null" placeholder in the json not null nodes column`) + + got, err := repo.Get(ctx, uid, version) + require.NoError(t, err) + assert.Equal(t, nodes, got, "Get must return the original nodes") + + // List has its own decompression branch and its own explicit column list, so Get + // passing does not imply List passes. Reuses the container rather than starting one. + list, err := repo.List(ctx, "default") + require.NoError(t, err) + assert.Equal(t, nodes, list[UUIDVersion{UID: uid, Version: version}], "List must decompress compressed rows") + + r := fetchRow(ctx, t, proxy, uid, version) + assert.NotEmpty(t, r.CompressedNodes.String, "compressednodes should hold the compressed payload") + assert.Equal(t, "null", r.Nodes, "nodes column should be the json null placeholder") + assert.Less(t, len(r.CompressedNodes.String), 1*mb, "stored compressed payload should be smaller than raw") +} + +// TestOffloadCompressionPostgres_LegacyRows covers both shapes a pre-compression row +// can have, so an upgrade needs no data migration: the empty string the migration +// backfills, and a genuine SQL NULL an old replica writes after that one-shot backfill +// has already run (see the CompressedNodes field comment). +func TestOffloadCompressionPostgres_LegacyRows(t *testing.T) { + ctx := logging.TestContext(t.Context()) + repo, proxy := setupOffloadRepoPostgres(ctx, t) + + legacyNodes := wfv1.Nodes{"n1": wfv1.NodeStatus{ID: "n1", Name: "n1", Phase: wfv1.NodeSucceeded}} + raw, err := json.Marshal(legacyNodes) + require.NoError(t, err) + + const version = "fnv:legacy" + for _, tc := range []struct { + name, uid string + wantValid bool + insert func(db.Session, string) error + }{ + { + name: "backfilled empty string", uid: "uid-pg-legacy", wantValid: true, + insert: func(s db.Session, uid string) error { + _, insErr := s.Collection("argo_workflows").Insert(&nodesRecord{ + ClusterName: "test", + UUIDVersion: UUIDVersion{UID: uid, Version: version}, + Namespace: "default", + Nodes: string(raw), + CompressedNodes: sql.NullString{String: "", Valid: true}, + }) + return insErr + }, + }, + { + // Must be raw SQL: Insert(&nodesRecord{...}) always writes a non-NULL empty + // string, so it cannot reproduce the shape an old replica leaves behind. + name: "genuine SQL NULL", uid: "uid-pg-null", wantValid: false, + insert: func(s db.Session, uid string) error { + _, execErr := s.SQL().Exec( + "insert into argo_workflows (clustername, uid, version, namespace, nodes) values ($1, $2, $3, $4, $5)", + "test", uid, version, "default", string(raw)) + return execErr + }, + }, + } { + t.Run(tc.name, func(t *testing.T) { + require.NoError(t, proxy.With(ctx, func(s db.Session) error { return tc.insert(s, tc.uid) })) + require.Equal(t, tc.wantValid, fetchRow(ctx, t, proxy, tc.uid, version).CompressedNodes.Valid, + "row must have the column shape this case claims to test") + + got, err := repo.Get(ctx, tc.uid, version) + require.NoError(t, err, "Get must read legacy rows on Postgres") + assert.Equal(t, legacyNodes, got) + + list, err := repo.List(ctx, "default") + require.NoError(t, err, "List must read legacy rows on Postgres") + assert.Equal(t, legacyNodes, list[UUIDVersion{UID: tc.uid, Version: version}]) + }) + } +}