Skip to content

CXP-1296 Refresh every workspace entitlement role on updateUser/updateGroup #92

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
JavierCarnelli-ConductorOne wants to merge 3 commits into
base: staging
Choose a base branch
from
Open

CXP-1296 Refresh every workspace entitlement role on updateUser/updateGroup #92

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b3eb2cf
fix(databricks): refresh every workspace entitlement role on updateUs…
JavierCarnelli-ConductorOne Nov 4, 2026
e494583
improvements
JavierCarnelli-ConductorOne Nov 4, 2026
7db2e2a
fix(databricks): drop duplicate resource changes within an event feed…
JavierCarnelli-ConductorOne Nov 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 31 additions & 4 deletions pkg/connector/audit_event_feed.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import (
"errors"
"fmt"
"net/http"
"slices"
"sort"
"strconv"
"strings"
Expand Down Expand Up @@ -101,15 +102,20 @@ var auditLogActions = map[auditActionKey]auditActionMapping{
grant: &auditGrantMapping{entitlement: groupMemberEntitlement, principalParam: "targetUserId", revoke: true},
},
{auditServiceAccounts, "removeGroup"}: {resourceType: groupResourceType, idParam: "targetGroupId"},
// Workspace entitlements change through two paths that log different rows. The admin UI's
// access toggles log changeDatabricksWorkspaceAcl / changeDatabricksSqlAcl (mapped below to their
// single role). Workspace SCIM writes log only updateUser/updateGroup, with no ACL row and no
// detail about which entitlement changed, so those actions refresh every workspace entitlement role.
// Keep both mappings: each covers a path the other doesn't.
{auditServiceAccounts, "updateGroup"}: {
resourceType: groupResourceType, idParam: "targetGroupId",
roleNames: []string{ClusterCreateRole, InstancePoolCreateRole},
roleNames: entitlements,
},
// "add"/"delete" are the real user-lifecycle events; deleteUser is a parameterless PII purge.
{auditServiceAccounts, "add"}: {resourceType: userResourceType, idParam: "targetUserId"},
{auditServiceAccounts, "updateUser"}: {
resourceType: userResourceType, idParam: "targetUserId",
roleNames: []string{ClusterCreateRole, InstancePoolCreateRole},
roleNames: entitlements,
},
{auditServiceAccounts, "delete"}: {resourceType: userResourceType, idParam: "targetUserId"},
{auditServiceAccounts, "activateUser"}: {resourceType: userResourceType, idParam: "targetUserId"},
Expand Down Expand Up @@ -393,8 +399,11 @@ func (f *auditEventFeed) ListEvents(
}

var events []*v2.Event
var unmappedRows, skippedLookupRows, resourceChangeEvents, grantEvents, revokeEvents int
var unmappedRows, skippedLookupRows, resourceChangeEvents, grantEvents, revokeEvents, duplicateResourceChanges int
principalTypes := principalTypeCache{}
// Keep only the last resource change per resource in a page: each one triggers a full refetch

@mateoHernandez123 mateoHernandez123 Nov 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The four-role refresh on updateUser/updateGroup is right. The per-page collapse added on top of it in 7db2e2a is work the platform already does: its feed consumer dedupes resource-change events for each ListEvents page by (resourceType, resourceID), folds OccurredAt to the most recent, and queues one targeted re-sync per surviving resource. The duplicate 19/1-style events never cost a refetch, so collapsing here only saves bytes on the wire — about 5k tiny events at a full auditLogPageLimit page, well inside the 4 MiB gRPC ceiling.

The shipped audit feeds emit one event per audit row and leave the collapse to the platform:

  • baton-ramp — "emits one ResourceChangeEvent per event", "the platform ... turns ResourceChangeEvents into targeted-sync tasks".
  • baton-teleport — flat append over the audit rows, no page-level map.

The cost of keeping it lands in testdata/expected-events.jsonl, which is now globally coupled: add one seed row naming a resource an earlier row already named and an unrelated line disappears (row 19's alice line goes the moment any later row touches alice). I'd drop lastResourceChange, resourceChangeKey and the slices.DeleteFunc pass and ship the role-mapping change on its own.

// (costly for workspace roles). Grant and revoke events are never collapsed.
lastResourceChange := map[string]int{}
for _, row := range rows {
affected, rowRateLimit, err := mapAuditRowToResource(ctx, f.client, principalTypes, row, f.client.GetAccountId(), workspaceLookup)
if rowRateLimit != nil {
Expand Down Expand Up @@ -450,7 +459,14 @@ func (f *auditEventFeed) ListEvents(
CreateGrantEvent: &v2.CreateGrantEvent{Entitlement: a.grant.entitlement, Principal: a.grant.principal},
}
default:
resourceChangeEvents++
key := resourceChangeKey(a.resourceId, a.parentResourceId)
if prev, seen := lastResourceChange[key]; seen {
events[prev] = nil
duplicateResourceChanges++
} else {
resourceChangeEvents++
}
lastResourceChange[key] = len(events)
event.Event = &v2.Event_ResourceChangeEvent{
ResourceChangeEvent: &v2.ResourceChangeEvent{ResourceId: a.resourceId, ParentResourceId: a.parentResourceId},
}
Expand All @@ -459,6 +475,8 @@ func (f *auditEventFeed) ListEvents(
}
}

events = slices.DeleteFunc(events, func(e *v2.Event) bool { return e == nil })

hasMore := rawRowCount >= auditLogPageLimit
nextCursor, advanced := advanceEventCursor(cursor, rows, lastRawBoundary, hasMore, lagCutoff)
if !advanced {
Expand All @@ -477,6 +495,7 @@ func (f *auditEventFeed) ListEvents(
zap.Int("audit_rows_skipped_lookup_failure", skippedLookupRows),
zap.Int("events_total", len(events)),
zap.Int("events_resource_change", resourceChangeEvents),
zap.Int("events_resource_change_duplicates_dropped", duplicateResourceChanges),
zap.Int("events_create_grant", grantEvents),
zap.Int("events_create_revoke", revokeEvents),
zap.Bool("has_more", hasMore),
Expand All @@ -486,6 +505,14 @@ func (f *auditEventFeed) ListEvents(
return events, &pagination.StreamState{Cursor: encoded, HasMore: hasMore}, annos, nil
}

// resourceChangeKey identifies a resource change by its resource and parent, for dropping
// duplicates within one ListEvents page.
func resourceChangeKey(resourceId, parentResourceId *v2.ResourceId) string {
return fmt.Sprintf("%s\x00%s\x00%s\x00%s",
resourceId.GetResourceType(), resourceId.GetResource(),
parentResourceId.GetResourceType(), parentResourceId.GetResource())
}

func isRetryableLookupError(err error) bool {
switch status.Code(err) {
case codes.Unavailable, codes.ResourceExhausted, codes.DeadlineExceeded, codes.Internal, codes.Unknown, codes.Canceled:
Expand Down
136 changes: 134 additions & 2 deletions pkg/connector/audit_event_feed_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -505,10 +505,52 @@ func TestMapAuditRowToResource(t *testing.T) {
},
want: []wantResource{
{userResourceType.Id, "u-1", accountResourceType.Id, accountId},
{roleResourceType.Id, roleResourceId(WorkspaceAccessRole, workspaceParent), workspaceResourceType.Id, "my-workspace"},
{roleResourceType.Id, roleResourceId(SQLAccessRole, workspaceParent), workspaceResourceType.Id, "my-workspace"},
{roleResourceType.Id, roleResourceId(ClusterCreateRole, workspaceParent), workspaceResourceType.Id, "my-workspace"},
{roleResourceType.Id, roleResourceId(InstancePoolCreateRole, workspaceParent), workspaceResourceType.Id, "my-workspace"},
},
},
{
name: "updateGroup in a workspace refreshes the group and every workspace entitlement role",
row: auditLogRow{
ActionName: "updateGroup",
ServiceName: "accounts",
WorkspaceID: 123,
RequestParams: map[string]string{"targetGroupId": "g-1"},
},
want: []wantResource{
{groupResourceType.Id, groupResourceId(context.Background(), "g-1", accountParent), accountResourceType.Id, accountId},
{roleResourceType.Id, roleResourceId(WorkspaceAccessRole, workspaceParent), workspaceResourceType.Id, "my-workspace"},
{roleResourceType.Id, roleResourceId(SQLAccessRole, workspaceParent), workspaceResourceType.Id, "my-workspace"},
{roleResourceType.Id, roleResourceId(ClusterCreateRole, workspaceParent), workspaceResourceType.Id, "my-workspace"},
{roleResourceType.Id, roleResourceId(InstancePoolCreateRole, workspaceParent), workspaceResourceType.Id, "my-workspace"},
},
},
{
name: "account-level updateUser refreshes only the user",
row: auditLogRow{
ActionName: "updateUser",
ServiceName: "accounts",
WorkspaceID: 0,
RequestParams: map[string]string{"targetUserId": "u-1"},
},
want: []wantResource{
{userResourceType.Id, "u-1", accountResourceType.Id, accountId},
},
},
{
name: "account-level updateGroup refreshes only the group",
row: auditLogRow{
ActionName: "updateGroup",
ServiceName: "accounts",
WorkspaceID: 0,
RequestParams: map[string]string{"targetGroupId": "g-1"},
},
want: []wantResource{
{groupResourceType.Id, groupResourceId(context.Background(), "g-1", accountParent), accountResourceType.Id, accountId},
},
},
{
name: "unknown action is skipped",
row: auditLogRow{ActionName: "someUnityCatalogAction"},
Expand Down Expand Up @@ -1423,8 +1465,10 @@ func TestListEventsCachesPrincipalTypes(t *testing.T) {
if err != nil {
t.Fatalf("ListEvents() error = %v", err)
}
if len(events) != 5 {
t.Fatalf("len(events) = %d, want 5", len(events))
// Three sp-1 grants, plus one g-1 change: both gone-1 rows fall back to the same group
// change, and only the last is kept.
if len(events) != 4 {
t.Fatalf("len(events) = %d, want 4", len(events))
}
for _, id := range []string{"sp-1", "gone-1"} {
for _, kind := range []string{"Users", "Groups", "ServicePrincipals"} {
Expand Down Expand Up @@ -1465,3 +1509,91 @@ func TestResolvePrincipalTypeDoesNotCacheErrors(t *testing.T) {
t.Errorf("lookup made %d calls, want 2 (the error must not be cached)", n)
}
}

// TestListEventsDropsDuplicateResourceChanges verifies that when a page names the same resource
// more than once (the same user, or the same workspace role refreshed by several updateUser rows),
// only the last change is kept, with its own row's ID, and grant events are never collapsed.
func TestListEventsDropsDuplicateResourceChanges(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method == http.MethodGet && strings.HasSuffix(r.URL.Path, "/api/2.0/accounts/acct-1/workspaces") {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode([]map[string]any{{"workspace_id": 1, "workspace_name": "ws1", "deployment_name": "ws1"}})
return
}
if r.Method == http.MethodGet && strings.HasSuffix(r.URL.Path, "/api/2.0/sql/warehouses/wh-1") {
w.Header().Set("Content-Type", "application/json")
fmt.Fprintf(w, `{"id":"wh-1"}`)
return
}
if r.Method == http.MethodGet && strings.HasSuffix(r.URL.Path, "/api/2.0/accounts/acct-1/scim/v2/Users/u-1") {
w.Header().Set("Content-Type", "application/json")
fmt.Fprint(w, `{}`)
return
}
if r.Method != http.MethodPost || !strings.HasSuffix(r.URL.Path, "/api/2.0/sql/statements") {
t.Errorf("unexpected request: %s %s", r.Method, r.URL.Path)
http.NotFound(w, r)
return
}

w.Header().Set("Content-Type", "application/json")
resp := map[string]any{
"statement_id": "stmt-1",
"status": map[string]any{"state": "SUCCEEDED"},
"manifest": map[string]any{
"schema": map[string]any{
"columns": []map[string]any{
{"name": "event_id"}, {"name": "event_time"}, {"name": "workspace_id"},
{"name": "action_name"}, {"name": "service_name"}, {"name": "request_params"},
},
},
},
"result": map[string]any{
"data_array": [][]string{
{"evt-1", "2026-01-01T00:00:00Z", "1", "updateUser", "accounts", `{"targetUserId":"u-1"}`},
{"evt-2", "2026-01-01T00:01:00Z", "1", "updateUser", "accounts", `{"targetUserId":"u-2"}`},
{"evt-3", "2026-01-01T00:02:00Z", "1", "updateUser", "accounts", `{"targetUserId":"u-1"}`},
{"evt-4", "2026-01-01T00:03:00Z", "0", "addPrincipalToGroup", "accounts", `{"targetGroupId":"g-1","targetUserId":"u-1"}`},
{"evt-5", "2026-01-01T00:04:00Z", "0", "addPrincipalToGroup", "accounts", `{"targetGroupId":"g-1","targetUserId":"u-1"}`},
},
},
}
if err := json.NewEncoder(w).Encode(resp); err != nil {
t.Fatalf("failed to encode mock statement response: %v", err)
}
}))
defer server.Close()

target, err := url.Parse(server.URL)
if err != nil {
t.Fatalf("failed to parse test server URL: %v", err)
}

httpClient := &http.Client{Transport: &redirectTransport{target: target}}
client, err := databricks.NewClient(context.Background(), httpClient, "example.cloud.databricks.com", "accounts.cloud.databricks.com", "acct-1", "", &databricks.NoAuth{}, nil)
if err != nil {
t.Fatalf("NewClient() error = %v", err)
}

feed := newAuditEventFeed(client, nil, "wh-1")
events, _, _, err := feed.ListEvents(context.Background(), testEarliestEvent(), &pagination.StreamToken{Cursor: ""})
if err != nil {
t.Fatalf("ListEvents() error = %v", err)
}

// evt-1 is fully superseded: evt-3 names u-1 and the same four roles again. evt-2 keeps only
// u-2, its roles are superseded by evt-3. evt-4 and evt-5: one grant each.
want := []string{"evt-2/0", "evt-3/0", "evt-3/1", "evt-3/2", "evt-3/3", "evt-3/4", "evt-4/0", "evt-5/0"}
got := make([]string, 0, len(events))
for _, e := range events {
got = append(got, e.GetId())
}
if strings.Join(got, ",") != strings.Join(want, ",") {
t.Fatalf("event IDs = %v, want %v", got, want)
}
for _, e := range events[6:] {
if e.GetCreateGrantEvent() == nil {
t.Errorf("event %s = %+v, want a CreateGrantEvent", e.GetId(), e)
}
}
}
11 changes: 8 additions & 3 deletions test-server/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -237,16 +237,21 @@ account, as in full sync.
| 16 | −11h | 0 | workspace / `updatePermissionAssignment` (no `workspace_id`) | principal_id dave | none |
| 17 | −10h | ws-beta | workspace / `deletePermissionAssignment` | workspace_id ws-beta | ws-beta changed |
| 18 | −9h | ws-gamma | accounts / `add` | targetUserId alice | none: eu-west-1 isn't visible from a us-east-1 warehouse |
| 19 | −8h | ws-alpha | accounts / `updateUser` | targetUserId alice | alice changed; `ws-alpha:allow-cluster-create` and `ws-alpha:allow-instance-pool-create` changed |
| 20 | −7h | 0 | accounts / `updateGroup` | targetGroupId analysts | analysts changed |
| 19 | −8h | ws-alpha | accounts / `updateUser` | targetUserId alice | alice changed; `ws-alpha:workspace-access`, `ws-alpha:databricks-sql-access`, `ws-alpha:allow-cluster-create` and `ws-alpha:allow-instance-pool-create` changed |
| 20 | −7h | ws-beta | accounts / `updateGroup` | targetGroupId analysts | analysts changed; `ws-beta:workspace-access`, `ws-beta:databricks-sql-access`, `ws-beta:allow-cluster-create` and `ws-beta:allow-instance-pool-create` changed |
| 21 | −6h | 0 | accounts / `removeGroup` | targetGroupId data-engineers | data-engineers changed |
| 22 | −6h (same time as 21) | 0 | accounts / `deactivateUser` | targetUserId dave | dave changed |
| 23 | −5h30m | ws-alpha | accounts / `changeDatabricksWorkspaceAcl` | targetUserId carol | ws-alpha changed; `ws-alpha:workspace-access` changed |
| 24 | −5h | ws-alpha | clusters / `create` | | none: not requested |
| 25 | −2h | 0 | accounts / `add` | targetUserId bob | none: newer than the 4-hour lag cutoff |
| 26 | −72h | 0 | accounts / `add` | targetUserId carol | none: older than the feed start |

That adds up to 25 events.
The table lists what each row maps to, 31 events in all. Within one poll, `ListEvents` keeps
only the last change for each resource (by resource and parent ID), because a resource change
makes C1 re-read the resource's current state anyway. Grant and revoke events are never
collapsed. Several resources appear more than once (bob in rows 2, 11 and 12, `account_admin` in
10 and 11, the ws-alpha roles in 19 and 23, and others), so the poll emits 21 events, each with
the ID of the last row that named its resource.

## curl examples

Expand Down
2 changes: 1 addition & 1 deletion test-server/seeds.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,7 @@ func seedAudit(start time.Time) []auditRow {
RequestParams: user("scim", userAlice, "alice@example.com")},
{EventID: "e0000001-0000-4000-8000-000000000019", EventTime: at(8, 0), WorkspaceID: wsAlphaID, ServiceName: "accounts", ActionName: "updateUser",
RequestParams: user("scim", userAlice, "alice@example.com")},
{EventID: "e0000001-0000-4000-8000-000000000020", EventTime: at(7, 0), ServiceName: "accounts", ActionName: "updateGroup",
{EventID: "e0000001-0000-4000-8000-000000000020", EventTime: at(7, 0), WorkspaceID: wsBetaID, ServiceName: "accounts", ActionName: "updateGroup",
RequestParams: map[string]string{"endpoint": "scim", "targetGroupId": groupAnalysts, "targetGroupName": "analysts"}},
// Two rows with the same event_time, to exercise the (event_time, event_id) ordering.
{EventID: "e0000001-0000-4000-8000-000000000021", EventTime: tie, ServiceName: "accounts", ActionName: "removeGroup",
Expand Down
Loading
Loading