From 5693bf894179e9ac57697e38318c9aa52f520779 Mon Sep 17 00:00:00 2001 From: Oleksandr Didukh Date: Fri, 24 Jul 2026 14:57:35 +0200 Subject: [PATCH 1/2] feat(openframe): stamp team_id on CDC-captured tables for shared-DB Debezium MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Under Fleet shared-DB multi-tenancy one MySQL serves every tenant, so a CDC record must carry its own tenant discriminator: the Debezium connector's SMTs are stateless and cannot join host_id -> hosts.team_id, and activity_past rows have no host reference at all. This stamps a team_id column on the four CDC-captured tables so the shared-plane stream consumer can resolve the tenant per event. - Migration 20260722000001_AddTeamIdToCdcTables (openframe goose pipeline): nullable team_id on activity_past, activity_host_past, query_results, policy_membership. No index, no FK (a team deletion must not rewrite/cascade over historical CDC rows). Idempotent via columnExists guards. - Write-path stamping, all wrapped in OPENFRAME(mysql-multitenancy) markers: * query_results + sync policy_membership stamp from the request team pin (fleet.OpenframeTeamID(ctx) — the agent plane is always pinned in shared mode); * async policy_membership + activity_host_past stamp from the row's own host via (SELECT team_id FROM hosts WHERE id = ?) — those run from unpinned cron contexts; * activity_past stamps from the request pin — the only tenant signal for host-less (user/team-level) activities. Backward compatible: unpinned writes (flag off, or background crons writing host-less activities) keep the statements byte-identical and leave team_id NULL. No SELECT * struct-scan readers exist on these tables, so the added column cannot break reads. Verified with MySQL-backed tests (migration idempotency + stamping across pinned/unpinned/async paths) and flag-off regressions. Co-Authored-By: Claude Fable 5 --- openframe/docs/fork-file-manifest.md | 2 + openframe/docs/mysql-multitenancy-feature.md | 25 +++++++ .../activity/internal/mysql/new_activity.go | 32 ++++++++- .../internal/mysql/new_activity_test.go | 67 ++++++++++++++++++ .../20260722000001_AddTeamIdToCdcTables.go | 47 +++++++++++++ .../mysql/migrations_openframe_test.go | 69 +++++++++++++++++++ server/datastore/mysql/policies.go | 56 +++++++++++++-- server/datastore/mysql/query_results.go | 21 +++++- 8 files changed, 310 insertions(+), 9 deletions(-) create mode 100644 server/datastore/mysql/migrations/openframe/20260722000001_AddTeamIdToCdcTables.go diff --git a/openframe/docs/fork-file-manifest.md b/openframe/docs/fork-file-manifest.md index 4680c0c503d..bacac844c16 100644 --- a/openframe/docs/fork-file-manifest.md +++ b/openframe/docs/fork-file-manifest.md @@ -212,6 +212,7 @@ openframe/scripts/test_host_assignments.sh openframe/scripts/verify.sh server/datastore/mysql/migrations/openframe/20260301000001_AddPolicyHostsJoinTable.go server/datastore/mysql/migrations/openframe/20260301000002_AddQueryHostsJoinTable.go +server/datastore/mysql/migrations/openframe/20260722000001_AddTeamIdToCdcTables.go server/datastore/mysql/migrations/openframe/migration.go server/datastore/mysql/migrations_openframe_test.go server/datastore/redis/keyprefix.go @@ -247,6 +248,7 @@ go.mod go.sum orbit/cmd/orbit/orbit.go orbit/pkg/osquery/osquery.go +server/activity/internal/mysql/new_activity.go server/archtest/README.md server/archtest/test_files/dependency/dependency.go server/config/config.go diff --git a/openframe/docs/mysql-multitenancy-feature.md b/openframe/docs/mysql-multitenancy-feature.md index 5976e4ec568..3bf99f0b0c5 100644 --- a/openframe/docs/mysql-multitenancy-feature.md +++ b/openframe/docs/mysql-multitenancy-feature.md @@ -48,6 +48,7 @@ Applied by `fleet prepare db`. Idempotent (`columnExists`/`indexExists` guards). | `20260629000001_AddTeamsOpenframeTenantUUID` | `teams.openframe_tenant_uuid CHAR(36)` + unique key | the UUID→team bridge | | `20260626000001_ScopeHostIdentityUniqueToTeam` | `hosts` virtual col `openframe_team_key = IFNULL(team_id,0)` + `UNIQUE(osquery_host_id, openframe_team_key)`, drop global `UNIQUE(osquery_host_id)` | host identity unique **per team** — the same device can exist in two tenants | | `20260620000001_ScopeLabelUniqueNameToTeam` | same generated-column pattern for `labels.name` | label names unique per team; built-ins stay global | +| `20260722000001_AddTeamIdToCdcTables` | nullable `team_id` on `activity_past`, `activity_host_past`, `query_results`, `policy_membership` (no index, no FK) | the Debezium CDC tables must be self-describing on a shared DB — see "CDC team stamping" below | The `IFNULL(team_id,0)` sentinel collapses all NULL-team rows onto key `0` (team ids start at 1), so **flag-off / pre-backfill the uniqueness is bit-for-bit the old global uniqueness**, and the @@ -102,6 +103,30 @@ clusters' migration Jobs pointed at one DB, the first wins and migrates; the res the schema already applied and no-op. Session-scoped (auto-released if the Job dies). Flag-off runs are untouched. +## CDC team stamping (Debezium on the shared DB) + +The OpenFrame Debezium pipeline captures `activity_past`, `activity_host_past`, `query_results` +and `policy_membership`. On a shared DB one connector serves every tenant, and its SMTs are +stateless — a CDC record must therefore carry its own tenant discriminator. The fork stamps the +`team_id` column (added by `20260722000001`) at write time: + +| Table | Stamp source | Where | +|---|---|---| +| `query_results` | request team pin (`fleet.OpenframeTeamID(ctx)` — the agent plane is always pinned in shared mode) | `OverwriteQueryResultRows` | +| `policy_membership` (sync) | request team pin | `RecordPolicyQueryExecutions` | +| `policy_membership` (async collector) | the row's own host, `(SELECT team_id FROM hosts WHERE id = ?)` — the cron context is never pinned | `AsyncBatchInsertPolicyMembership` | +| `activity_past` | request team pin — the only tenant signal for host-less (user/team-level) activities | `server/activity/internal/mysql/new_activity.go` | +| `activity_host_past` | the row's own host via subselect (host activities can come from unpinned crons; the host is authoritative anyway) | same | + +Unpinned writes (flag off, or background crons writing host-less activities) keep the original +statement **byte-identical** and leave `team_id` NULL; the downstream consumer drops NULL-team +events (fail closed). The `teams` table (with the `openframe_tenant_uuid` bridge) is also added +to the connector's capture list platform-side, so consumers can resolve `team_id` → tenant UUID +without touching Fleet's API. Platform side of this pipeline: shared connector registration in +`openframe-saas-shared`, and shared-plane consumption in that repo's `openframe-saas-stream` +(the MeshCentral pattern — per-event tenant resolution, gated by +`openframe.fleet.multi-tenancy.enabled`). + ## Helm / config wiring (`charts/fleet/`) `values.yaml` adds `fleet.openframe.multiTenancy` (`enabled: false` default; `tenantUuid` / diff --git a/server/activity/internal/mysql/new_activity.go b/server/activity/internal/mysql/new_activity.go index 133ea85eaa0..3b1a25d2296 100644 --- a/server/activity/internal/mysql/new_activity.go +++ b/server/activity/internal/mysql/new_activity.go @@ -9,6 +9,9 @@ import ( "github.com/fleetdm/fleet/v4/server/activity/api" "github.com/fleetdm/fleet/v4/server/activity/internal/types" "github.com/fleetdm/fleet/v4/server/contexts/ctxerr" + // >>> OPENFRAME(mysql-multitenancy): team pin helpers for CDC team_id stamping + "github.com/fleetdm/fleet/v4/server/fleet" + // <<< OPENFRAME(mysql-multitenancy) platform_mysql "github.com/fleetdm/fleet/v4/server/platform/mysql" "github.com/jmoiron/sqlx" ) @@ -72,6 +75,18 @@ func (ds *Datastore) NewActivity( cols = append(cols, "user_email") } + // >>> OPENFRAME(mysql-multitenancy): stamp the tenant team onto CDC-captured rows so the + // shared-DB Debezium pipeline can resolve the tenant per event — openframe/docs/mysql-multitenancy-feature.md. + // activity_past has no host reference, so the request's team pin is the only tenant signal + // (user/team-level activities). Unpinned contexts (flag off, or background crons) keep the + // original statement byte-identical and leave team_id NULL. + teamID, teamPinned := fleet.OpenframeTeamID(ctx) + if teamPinned { + cols = append(cols, "team_id") + args = append(args, teamID) + } + // <<< OPENFRAME(mysql-multitenancy) + return platform_mysql.WithRetryTxx(ctx, ds.primary, func(tx sqlx.ExtContext) error { const insertActStmt = `INSERT INTO activity_past (%s) VALUES (%s)` sqlStmt := fmt.Sprintf(insertActStmt, strings.Join(cols, ","), strings.Repeat("?,", len(cols)-1)+"?") @@ -84,13 +99,28 @@ func (ds *Datastore) NewActivity( // This supposes a reasonable amount of hosts per activity, to revisit if we // get in the 10K+. if ah, ok := activity.(types.ActivityHosts); ok { - const insertActHostStmt = `INSERT INTO activity_host_past (host_id, activity_id) VALUES ` + // >>> OPENFRAME(mysql-multitenancy): stamp each row with its host's team via a scalar + // subselect (host activities can be written from unpinned cron contexts, and the host is + // the authoritative tenant signal anyway) — engaged when the multitenancy flag is on, + // or when the caller context is team-pinned. Flag off keeps the statement byte-identical. + stampTeam := teamPinned || fleet.IsOpenframeMultitenancy() + insertActHostStmt := `INSERT INTO activity_host_past (host_id, activity_id) VALUES ` + if stampTeam { + insertActHostStmt = `INSERT INTO activity_host_past (host_id, activity_id, team_id) VALUES ` + } + // <<< OPENFRAME(mysql-multitenancy) var sb strings.Builder if hostIDs := ah.HostIDs(); len(hostIDs) > 0 { sb.WriteString(insertActHostStmt) actID, _ := res.LastInsertId() for _, hid := range hostIDs { + // >>> OPENFRAME(mysql-multitenancy) + if stampTeam { + sb.WriteString(fmt.Sprintf("(%d, %d, (SELECT team_id FROM hosts WHERE id = %d)),", hid, actID, hid)) + continue + } + // <<< OPENFRAME(mysql-multitenancy) sb.WriteString(fmt.Sprintf("(%d, %d),", hid, actID)) } diff --git a/server/activity/internal/mysql/new_activity_test.go b/server/activity/internal/mysql/new_activity_test.go index db7bae6646b..d9d1ebda72a 100644 --- a/server/activity/internal/mysql/new_activity_test.go +++ b/server/activity/internal/mysql/new_activity_test.go @@ -2,6 +2,7 @@ package mysql import ( "context" + "database/sql" "encoding/json" "testing" "time" @@ -9,6 +10,10 @@ import ( "github.com/fleetdm/fleet/v4/server/activity/api" "github.com/fleetdm/fleet/v4/server/activity/internal/testutils" "github.com/fleetdm/fleet/v4/server/activity/internal/types" + // >>> OPENFRAME(mysql-multitenancy) + "github.com/fleetdm/fleet/v4/server/datastore/mysql/migrations/openframe" + "github.com/fleetdm/fleet/v4/server/fleet" + // <<< OPENFRAME(mysql-multitenancy) "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -29,6 +34,9 @@ func TestNewActivity(t *testing.T) { {"HostAssociation", testNewActivityHostAssociation}, {"HostOnly", testNewActivityHostOnly}, {"DeletedUser", testNewActivityDeletedUser}, + // >>> OPENFRAME(mysql-multitenancy) + {"OpenframeTeamStamping", testNewActivityOpenframeTeamStamping}, + // <<< OPENFRAME(mysql-multitenancy) } for _, c := range cases { t.Run(c.name, func(t *testing.T) { @@ -244,3 +252,62 @@ func mustJSON(t *testing.T, v any) []byte { require.NoError(t, err) return b } + +// >>> OPENFRAME(mysql-multitenancy): CDC team_id stamping — openframe/docs/mysql-multitenancy-feature.md +// testNewActivityOpenframeTeamStamping verifies that with the multitenancy flag on: +// - a team-pinned context stamps activity_past.team_id (the only tenant signal for host-less +// activities); +// - activity_host_past rows are stamped with their host's own team via the scalar subselect +// (host activities may be written from unpinned cron contexts). +// +// The test schema comes from schema.sql, which does not include the OpenFrame migration +// pipeline, so the CDC migration is applied directly first. +func testNewActivityOpenframeTeamStamping(t *testing.T, env *testEnv) { + // No t.Setenv (the harness runs parallel): the host-activity stamping gate is + // (multitenancy env || ctx pin), and the pinned context below engages it. + ctx := webhookCtx(t) + + // Apply the openframe CDC migration (idempotent) — adds team_id to the captured tables. + tx, err := env.DB.Begin() + require.NoError(t, err) + require.NoError(t, openframe.Up_20260722000001(tx)) + require.NoError(t, tx.Commit()) + + // A real team row (hosts.team_id has an FK to teams). + res, err := env.DB.ExecContext(t.Context(), `INSERT INTO teams (name) VALUES ('cdc-team')`) + require.NoError(t, err) + teamID64, err := res.LastInsertId() + require.NoError(t, err) + hostTeamID := uint(teamID64) //nolint:gosec // test value from LastInsertId + hostID := env.InsertHost(t, "cdc.local", &hostTeamID) + + userID := env.InsertUser(t, "cdcuser", "cdc@example.com") + user := &api.User{ID: userID, Name: "cdcuser", Email: "cdc@example.com"} + + // Pin the request to a different team than the host's to tell the two stamps apart. + const pinnedTeam = uint(77) + pinnedCtx := fleet.NewOpenframeTeamContext(ctx, pinnedTeam) + + activity := hostActivity{ + dummyActivity: dummyActivity{name: "ran_script", details: map[string]any{"host_id": float64(hostID)}}, + hostIDs: []uint{hostID}, + } + require.NoError(t, env.ds.NewActivity(pinnedCtx, user, activity, mustJSON(t, activity.details), time.Now())) + + // activity_past carries the request pin. + var actTeam sql.NullInt64 + var actID uint + require.NoError(t, env.DB.QueryRowContext(t.Context(), + `SELECT id, team_id FROM activity_past WHERE activity_type = 'ran_script'`).Scan(&actID, &actTeam)) + require.True(t, actTeam.Valid) + require.EqualValues(t, pinnedTeam, actTeam.Int64) + + // activity_host_past carries the HOST's team (subselect), not the pin. + var hostActTeam sql.NullInt64 + require.NoError(t, env.DB.QueryRowContext(t.Context(), + `SELECT team_id FROM activity_host_past WHERE activity_id = ? AND host_id = ?`, actID, hostID).Scan(&hostActTeam)) + require.True(t, hostActTeam.Valid) + require.EqualValues(t, hostTeamID, hostActTeam.Int64) +} + +// <<< OPENFRAME(mysql-multitenancy) diff --git a/server/datastore/mysql/migrations/openframe/20260722000001_AddTeamIdToCdcTables.go b/server/datastore/mysql/migrations/openframe/20260722000001_AddTeamIdToCdcTables.go new file mode 100644 index 00000000000..908a1db01fc --- /dev/null +++ b/server/datastore/mysql/migrations/openframe/20260722000001_AddTeamIdToCdcTables.go @@ -0,0 +1,47 @@ +package openframe + +import ( + "database/sql" + "fmt" +) + +func init() { + MigrationClient.AddMigration(Up_20260722000001, Down_20260722000001) +} + +// Up_20260722000001 adds a nullable `team_id` column to the four tables captured by the +// OpenFrame Debezium CDC pipeline: activity_past, activity_host_past, query_results and +// policy_membership. Under shared-database multitenancy one MySQL serves every tenant, so a +// CDC record must carry its own tenant discriminator — the connector's SMTs are stateless and +// cannot join host_id → hosts.team_id, and activity_past rows have no host reference at all. +// The column is stamped at write time from the request's team pin (fleet.OpenframeTeamID) or, +// for writes without a pinned context (async policy membership, host activities), from the +// host's own team via a scalar subselect. +// +// Rows written with the multitenancy flag off (or by unpinned background jobs) keep team_id +// NULL — byte-identical to pre-migration behavior. No index: nothing queries these tables by +// team; the column exists solely so the change-stream consumer can resolve the tenant. +// +// Intentionally NO foreign key to teams(id): a team deletion must not rewrite or cascade over +// millions of historical CDC rows. +// +// Idempotent. +func Up_20260722000001(tx *sql.Tx) error { + for _, table := range []string{"activity_past", "activity_host_past", "query_results", "policy_membership"} { + hasCol, err := columnExists(tx, table, "team_id") + if err != nil { + return fmt.Errorf("checking %s.team_id column: %w", table, err) + } + if hasCol { + continue + } + if _, err := tx.Exec(fmt.Sprintf("ALTER TABLE %s ADD COLUMN team_id INT UNSIGNED NULL", table)); err != nil { + return fmt.Errorf("adding team_id column to %s: %w", table, err) + } + } + return nil +} + +func Down_20260722000001(tx *sql.Tx) error { + return nil +} diff --git a/server/datastore/mysql/migrations_openframe_test.go b/server/datastore/mysql/migrations_openframe_test.go index 09dc263c85f..43d9545591f 100644 --- a/server/datastore/mysql/migrations_openframe_test.go +++ b/server/datastore/mysql/migrations_openframe_test.go @@ -12,10 +12,12 @@ package mysql import ( "context" + "database/sql" "testing" "time" "github.com/fleetdm/fleet/v4/server/fleet" + "github.com/fleetdm/fleet/v4/server/test" "github.com/stretchr/testify/require" ) @@ -260,3 +262,70 @@ func TestOpenframeMigrationLock(t *testing.T) { require.NoError(t, err) release2() } + +// TestMigrateOpenframeCdcTeamIdStamping verifies 20260722000001 (team_id on the Debezium +// CDC-captured tables) plus the write-path stamping: +// - unpinned context (flag off / fork-main behavior) leaves team_id NULL; +// - a team-pinned context stamps query_results rows; +// - the async policy-membership collector (no pinned context) stamps each row from its +// host's own team via the flag-gated scalar subselect. +func TestMigrateOpenframeCdcTeamIdStamping(t *testing.T) { + ds := CreateMySQLDS(t) + ctx := context.Background() + + require.NoError(t, ds.MigrateOpenframe(ctx)) + for _, table := range []string{"activity_past", "activity_host_past", "query_results", "policy_membership"} { + var n int + require.NoError(t, ds.writer(ctx).GetContext(ctx, &n, + "SELECT COUNT(*) FROM information_schema.COLUMNS WHERE table_schema = DATABASE() AND table_name = ? AND column_name = 'team_id'", + table)) + require.Equal(t, 1, n, "expected %s.team_id after MigrateOpenframe", table) + } + + user := test.NewUser(t, ds, "CDC User", "cdc@example.com", true) + query := test.NewQuery(t, ds, nil, "CDC Query", "SELECT 1", user.ID, true) + host := test.NewHost(t, ds, "cdc-host", "192.168.1.10", "cdc-key", "cdc-uuid", time.Now()) + + rows := []*fleet.ScheduledQueryResultRow{ + {QueryID: query.ID, HostID: host.ID, LastFetched: time.Now().UTC().Truncate(time.Second)}, + } + + // Unpinned: team_id stays NULL — byte-identical to fork-main. + _, err := ds.OverwriteQueryResultRows(ctx, rows, fleet.DefaultMaxQueryReportRows) + require.NoError(t, err) + var teamIDs []sql.NullInt64 + require.NoError(t, ds.writer(ctx).SelectContext(ctx, &teamIDs, + "SELECT team_id FROM query_results WHERE host_id = ?", host.ID)) + require.Len(t, teamIDs, 1) + require.False(t, teamIDs[0].Valid) + + // Pinned: stamped with the pin (overwrite deletes + reinserts the host's rows). + pinned := fleet.NewOpenframeTeamContext(ctx, 42) + _, err = ds.OverwriteQueryResultRows(pinned, rows, fleet.DefaultMaxQueryReportRows) + require.NoError(t, err) + require.NoError(t, ds.writer(ctx).SelectContext(ctx, &teamIDs, + "SELECT team_id FROM query_results WHERE host_id = ?", host.ID)) + require.Len(t, teamIDs, 1) + require.True(t, teamIDs[0].Valid) + require.EqualValues(t, 42, teamIDs[0].Int64) + + // Async policy membership: stamping comes from the host row via the flag-gated subselect. + // The gate is (multitenancy env || ctx pin); the pinned context triggers it here without + // t.Setenv (forbidden — CreateMySQLDS forces t.Parallel), and the SQL ignores the pin value + // and reads hosts.team_id, exactly as the unpinned production cron path does. + team, err := ds.NewTeam(ctx, &fleet.Team{Name: "cdc-team"}) + require.NoError(t, err) + require.NoError(t, ds.AddHostsToTeam(ctx, fleet.NewAddHostsToTeamParams(&team.ID, []uint{host.ID}))) + pol, err := ds.NewGlobalPolicy(ctx, &user.ID, fleet.PolicyPayload{Name: "cdc-policy", Query: "SELECT 1"}) + require.NoError(t, err) + + passes := true + require.NoError(t, ds.AsyncBatchInsertPolicyMembership(pinned, []fleet.PolicyMembershipResult{ + {PolicyID: pol.ID, HostID: host.ID, Passes: &passes}, + })) + var memberTeam sql.NullInt64 + require.NoError(t, ds.writer(ctx).GetContext(ctx, &memberTeam, + "SELECT team_id FROM policy_membership WHERE policy_id = ? AND host_id = ?", pol.ID, host.ID)) + require.True(t, memberTeam.Valid) + require.EqualValues(t, team.ID, memberTeam.Int64) +} diff --git a/server/datastore/mysql/policies.go b/server/datastore/mysql/policies.go index 0afdbb52fa7..31993134749 100644 --- a/server/datastore/mysql/policies.go +++ b/server/datastore/mysql/policies.go @@ -907,6 +907,21 @@ func (ds *Datastore) RecordPolicyQueryExecutions(ctx context.Context, host *flee return err } + // >>> OPENFRAME(mysql-multitenancy): stamp the tenant team onto CDC-captured rows so the + // shared-DB Debezium pipeline can resolve the tenant per event — openframe/docs/mysql-multitenancy-feature.md. + // This path runs with an authenticated-host context, which is team-pinned whenever the + // multitenancy flag is on; unpinned (flag off) keeps the original statement byte-identical. + membershipCols := "updated_at, policy_id, host_id, passes" + membershipOnDup := "updated_at=VALUES(updated_at), passes=VALUES(passes)" + membershipRowShape := "(?,?,?,?)" + teamID, teamPinned := fleet.OpenframeTeamID(ctx) + if teamPinned { + membershipCols += ", team_id" + membershipOnDup += ", team_id=VALUES(team_id)" + membershipRowShape = "(?,?,?,?,?)" + } + // <<< OPENFRAME(mysql-multitenancy) + vals := []interface{}{} bindvars := []string{} if len(needsWrite) > 0 { @@ -919,8 +934,13 @@ func (ds *Datastore) RecordPolicyQueryExecutions(ctx context.Context, host *flee for _, policyID := range orderedIDs { matches := results[policyID] - bindvars = append(bindvars, "(?,?,?,?)") + bindvars = append(bindvars, membershipRowShape) vals = append(vals, updated, policyID, host.ID, matches) + // >>> OPENFRAME(mysql-multitenancy) + if teamPinned { + vals = append(vals, teamID) + } + // <<< OPENFRAME(mysql-multitenancy) } } @@ -940,9 +960,13 @@ func (ds *Datastore) RecordPolicyQueryExecutions(ctx context.Context, host *flee query := fmt.Sprintf( // INSERT IGNORE skips rows whose policy_id no longer exists (policy deleted // after query was distributed but before results arrived). - `INSERT IGNORE INTO policy_membership (updated_at, policy_id, host_id, passes) - VALUES %s ON DUPLICATE KEY UPDATE updated_at=VALUES(updated_at), passes=VALUES(passes)`, + // OPENFRAME(mysql-multitenancy): column list / dup-update carry team_id when the + // request is team-pinned (see membershipCols above). + `INSERT IGNORE INTO policy_membership (%s) + VALUES %s ON DUPLICATE KEY UPDATE %s`, + membershipCols, strings.Join(bindvars, ","), + membershipOnDup, ) if _, err := tx.ExecContext(ctx, query, vals...); err != nil { return ctxerr.Wrapf(ctx, err, "insert policy_membership (%v)", vals) @@ -2242,17 +2266,39 @@ func (ds *Datastore) AsyncBatchInsertPolicyMembership(ctx context.Context, batch // INSERT IGNORE, to avoid failing if policy / host does not exist (as this // runs asynchronously, they could get deleted in between the data being // received and being upserted). + // >>> OPENFRAME(mysql-multitenancy): stamp the tenant team onto CDC-captured rows so the + // shared-DB Debezium pipeline can resolve the tenant per event — openframe/docs/mysql-multitenancy-feature.md. + // This async collector runs from a cron with no team-pinned context, so the team comes from + // the row's own host via a scalar subselect — engaged when the multitenancy flag is on (or + // the caller context is explicitly team-pinned, which only tests do on this path). Flag off + // keeps the original statement byte-identical. + _, ctxPinned := fleet.OpenframeTeamID(ctx) + multitenancy := ctxPinned || fleet.IsOpenframeMultitenancy() sql := `INSERT IGNORE INTO policy_membership (policy_id, host_id, passes) VALUES ` - sql += strings.Repeat(`(?, ?, ?),`, len(batch)) + rowShape := `(?, ?, ?),` + if multitenancy { + sql = `INSERT IGNORE INTO policy_membership (policy_id, host_id, passes, team_id) VALUES ` + rowShape = `(?, ?, ?, (SELECT team_id FROM hosts WHERE id = ?)),` + } + sql += strings.Repeat(rowShape, len(batch)) sql = strings.TrimSuffix(sql, ",") sql += ` ON DUPLICATE KEY UPDATE updated_at = VALUES(updated_at), passes = VALUES(passes)` + if multitenancy { + sql += `, team_id = VALUES(team_id)` + } - vals := make([]interface{}, 0, len(batch)*3) + vals := make([]interface{}, 0, len(batch)*4) + // <<< OPENFRAME(mysql-multitenancy) hostIDs := make([]uint, 0, len(batch)) // Group incoming results per host for flip detection. incomingByHost := make(map[uint]map[uint]*bool, len(batch)) for _, tup := range batch { vals = append(vals, tup.PolicyID, tup.HostID, tup.Passes) + // >>> OPENFRAME(mysql-multitenancy): extra bind for the team subselect + if multitenancy { + vals = append(vals, tup.HostID) + } + // <<< OPENFRAME(mysql-multitenancy) hostIDs = append(hostIDs, tup.HostID) m, ok := incomingByHost[tup.HostID] if !ok { diff --git a/server/datastore/mysql/query_results.go b/server/datastore/mysql/query_results.go index 48a5e5f9574..3d3fd23847d 100644 --- a/server/datastore/mysql/query_results.go +++ b/server/datastore/mysql/query_results.go @@ -47,17 +47,32 @@ func (ds *Datastore) OverwriteQueryResultRows(ctx context.Context, rows []*fleet } // Insert the new rows + // >>> OPENFRAME(mysql-multitenancy): stamp the tenant team onto CDC-captured rows so the + // shared-DB Debezium pipeline can resolve the tenant per event — openframe/docs/mysql-multitenancy-feature.md. + // This path runs with an authenticated-host context, which is team-pinned whenever the + // multitenancy flag is on; unpinned (flag off) keeps the original statement byte-identical. + insertCols := "query_id, host_id, last_fetched, data" + rowShape := "(?, ?, ?, ?)" + teamID, teamPinned := fleet.OpenframeTeamID(ctx) + if teamPinned { + insertCols += ", team_id" + rowShape = "(?, ?, ?, ?, ?)" + } valueStrings := make([]string, 0, len(rows)) - valueArgs := make([]interface{}, 0, len(rows)*4) + valueArgs := make([]interface{}, 0, len(rows)*5) for _, row := range rows { - valueStrings = append(valueStrings, "(?, ?, ?, ?)") + valueStrings = append(valueStrings, rowShape) valueArgs = append(valueArgs, queryID, hostID, row.LastFetched, row.Data) + if teamPinned { + valueArgs = append(valueArgs, teamID) + } } //nolint:gosec // SQL query is constructed using constant strings insertStmt := ` - INSERT IGNORE INTO query_results (query_id, host_id, last_fetched, data) VALUES + INSERT IGNORE INTO query_results (` + insertCols + `) VALUES ` + strings.Join(valueStrings, ",") + // <<< OPENFRAME(mysql-multitenancy) result, err = tx.ExecContext(ctx, insertStmt, valueArgs...) if err != nil { From 8a0960e8cbd2c645eab8830a137a8d7d8bde88c1 Mon Sep 17 00:00:00 2001 From: Oleksandr Didukh Date: Fri, 24 Jul 2026 17:58:43 +0200 Subject: [PATCH 2/2] fix(openframe): address CodeRabbit review on CDC team_id stamping - policies.go: wrap the team_id-carrying policy_membership INSERT in the >>> / <<< OPENFRAME(mysql-multitenancy) sentinel markers so the grep-based fork-audit convention finds it on upstream merges (was a bare comment). - new_activity_test.go: scope the activity_past lookup to the test's own user_id + ORDER BY id DESC LIMIT 1 so it stays deterministic regardless of subtest isolation. Co-Authored-By: Claude Fable 5 --- server/activity/internal/mysql/new_activity_test.go | 3 ++- server/datastore/mysql/policies.go | 6 ++++-- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/server/activity/internal/mysql/new_activity_test.go b/server/activity/internal/mysql/new_activity_test.go index d9d1ebda72a..0504bb821bd 100644 --- a/server/activity/internal/mysql/new_activity_test.go +++ b/server/activity/internal/mysql/new_activity_test.go @@ -298,7 +298,8 @@ func testNewActivityOpenframeTeamStamping(t *testing.T, env *testEnv) { var actTeam sql.NullInt64 var actID uint require.NoError(t, env.DB.QueryRowContext(t.Context(), - `SELECT id, team_id FROM activity_past WHERE activity_type = 'ran_script'`).Scan(&actID, &actTeam)) + `SELECT id, team_id FROM activity_past WHERE activity_type = 'ran_script' AND user_id = ? ORDER BY id DESC LIMIT 1`, + userID).Scan(&actID, &actTeam)) require.True(t, actTeam.Valid) require.EqualValues(t, pinnedTeam, actTeam.Int64) diff --git a/server/datastore/mysql/policies.go b/server/datastore/mysql/policies.go index 31993134749..5f1156933d2 100644 --- a/server/datastore/mysql/policies.go +++ b/server/datastore/mysql/policies.go @@ -957,17 +957,19 @@ func (ds *Datastore) RecordPolicyQueryExecutions(ctx context.Context, host *flee err = ds.withTx(ctx, func(tx sqlx.ExtContext) error { if len(vals) > 0 { + // >>> OPENFRAME(mysql-multitenancy): the column list and ON DUPLICATE KEY UPDATE + // clause carry team_id when the request is team-pinned (see membershipCols above) — + // openframe/docs/mysql-multitenancy-feature.md. Upstream hardcodes the columns. query := fmt.Sprintf( // INSERT IGNORE skips rows whose policy_id no longer exists (policy deleted // after query was distributed but before results arrived). - // OPENFRAME(mysql-multitenancy): column list / dup-update carry team_id when the - // request is team-pinned (see membershipCols above). `INSERT IGNORE INTO policy_membership (%s) VALUES %s ON DUPLICATE KEY UPDATE %s`, membershipCols, strings.Join(bindvars, ","), membershipOnDup, ) + // <<< OPENFRAME(mysql-multitenancy) if _, err := tx.ExecContext(ctx, query, vals...); err != nil { return ctxerr.Wrapf(ctx, err, "insert policy_membership (%v)", vals) }