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/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` 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..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" @@ -12,6 +13,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 +47,18 @@ type nodesRecord struct { UUIDVersion Namespace string `db:"namespace"` Nodes string `db:"nodes"` + // 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 { @@ -84,7 +98,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: sql.NullString{String: file.CompressEncodeString(ctx, marshalled), Valid: true}, } logCtx := wdc.log.WithFields(logging.Fields{"uid": uid, "version": version}) @@ -135,8 +151,12 @@ func (wdc *nodeOffloadRepo) Get(ctx context.Context, uid, version string) (wfv1. if err != nil { return err } + nodesJSON, err := r.nodesJSON(ctx) + 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 +175,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 +186,12 @@ func (wdc *nodeOffloadRepo) List(ctx context.Context, namespace string) (map[UUI res = make(map[UUIDVersion]wfv1.Nodes) for _, r := range records { + nodesJSON, err := r.nodesJSON(ctx) + 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..cf1662b78cfc --- /dev/null +++ b/persist/sqldb/offload_node_status_repo_mysql_test.go @@ -0,0 +1,201 @@ +//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" + "database/sql" + "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.TB) (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.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), 13*mb, "stored compressed payload should be far smaller than raw") +} + +// 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) + + 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-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}]) + }) + } +} + +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 +} 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}]) + }) + } +}