perf(sqldb): compress offloaded node status. Fixes #13290

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

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?

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`

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`),
 	}
 }
 

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
 			}

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
+}