CXP-897 Incremental sync support - #56
Conversation
Add audit log action mappings for account-admin, workspace-access, and SQL-access role changes, plus a coarser fallback for the cluster-create and instance-pool-create entitlements. mapAuditRowToResource now returns multiple affected resources per audit row so a single action can refresh both a principal and the role(s) it holds. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…-event-feeds # Conflicts: # README.md # pkg/connector/connector.go
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| databricks.StatementParameter{Name: "start_date", Value: cursor.StartAt.Format("2006-01-02"), Type: "DATE"}, | ||
| databricks.StatementParameter{Name: "start_time", Value: cursor.StartAt.Format(time.RFC3339), Type: "TIMESTAMP"}, |
There was a problem hiding this comment.
🟠 Bug: cursor.StartAt is derived from time.Now() and is formatted here in the process's local timezone, but event_date / event_time in system.access.audit are UTC. On a host whose TZ is ahead of UTC (e.g. TZ=Asia/Tokyo), the local calendar date can be one day ahead of the UTC date for the same instant, so event_date >= :start_date prunes the partition containing events that event_time >= :start_time should have matched — those events are silently and permanently skipped. Format both parameters in UTC:
| databricks.StatementParameter{Name: "start_date", Value: cursor.StartAt.Format("2006-01-02"), Type: "DATE"}, | |
| databricks.StatementParameter{Name: "start_time", Value: cursor.StartAt.Format(time.RFC3339), Type: "TIMESTAMP"}, | |
| databricks.StatementParameter{Name: "start_date", Value: cursor.StartAt.UTC().Format("2006-01-02"), Type: "DATE"}, | |
| databricks.StatementParameter{Name: "start_time", Value: cursor.StartAt.UTC().Format(time.RFC3339), Type: "TIMESTAMP"}, |
|
|
||
| target := latest.Add(-auditLogTrailingLag) | ||
| if len(rows) == 0 { | ||
| target = now.Add(-auditLogTrailingLag) | ||
| } | ||
| if target.Before(cursor.StartAt) { | ||
| target = cursor.StartAt | ||
| } | ||
|
|
||
| var idsAtTarget []string | ||
| if target.Equal(latest) { | ||
| idsAtTarget = latestIDs | ||
| } | ||
|
|
||
| return eventPageCursor{StartAt: target, LatestEventSeen: latest, LastEventIDs: idsAtTarget} |
There was a problem hiding this comment.
🟡 Suggestion: once a page drains, StartAt is pulled back to latest - auditLogTrailingLag (4h) but LastEventIDs only remembers rows tied exactly at the new boundary. Every subsequent poll therefore re-reads the whole trailing 4h window and re-emits every event in it as a fresh RESOURCE_CHANGE, triggering the same targeted-sync Get calls over and over (dozens of times per event at a few-minute poll cadence). Consider carrying forward the set of already-emitted event IDs for the whole [StartAt, LatestEventSeen] window rather than just the boundary tie.
Separately, LatestEventSeen is written into the cursor here but never read anywhere — either use it (e.g. for the dedupe window above) or drop the field.
| FROM system.access.audit | ||
| WHERE event_date >= :start_date | ||
| AND event_time >= :start_time | ||
| AND action_name IN (%s) | ||
| ORDER BY event_time ASC | ||
| LIMIT %d | ||
| `, quotedInClause(auditLogActionNames()), auditLogPageLimit) |
There was a problem hiding this comment.
🟡 Suggestion: ORDER BY event_time ASC has no tiebreaker, so ordering among rows sharing an event_time is non-deterministic across the paged calls that hasMore drives. If ≥ auditLogPageLimit (1000) rows ever share one event_time, advanceEventCursor leaves StartAt pinned at that timestamp and the feed stops making progress. Adding , event_id ASC makes the ordering stable and lets you page with a (event_time, event_id) > predicate instead of relying on the ID set.
| func (c *Client) pollStatement(ctx context.Context, workspaceId string, res statementResponse) (statementResponse, error) { | ||
| l := ctxzap.Extract(ctx) | ||
|
|
||
| for res.Status.State == StatementStatePending || res.Status.State == StatementStateRunning { | ||
| select { | ||
| case <-ctx.Done(): | ||
| return res, ctx.Err() | ||
| case <-time.After(statementPollInterval): | ||
| } | ||
|
|
||
| l.Debug("polling databricks sql statement", zap.String("statement_id", res.StatementID), zap.String("state", string(res.Status.State))) | ||
|
|
||
| u := c.workspaceUrl(workspaceId).JoinPath(statementsEndpoint, res.StatementID) |
There was a problem hiding this comment.
🟡 Suggestion: this loop only terminates on a terminal statement state or ctx cancellation — there is no attempt/deadline cap. Validate() now calls ValidateAuditLogAccess through this path, so a warehouse stuck PENDING (cold start, queued, quota) makes connector validation hang for as long as the caller's context allows. Consider a bounded number of polls (or a context.WithTimeout around the statement) and cancelling the statement via DELETE /api/2.0/sql/statements/{id} when giving up so it doesn't keep occupying the warehouse.
| // ever advances forward, and LastEventIDs dedupes rows tied exactly on that boundary. | ||
| type eventPageCursor struct { | ||
| StartAt time.Time `json:"start_at"` | ||
| LatestEventSeen time.Time `json:"latest_event_seen"` |
There was a problem hiding this comment.
🟡 Suggestion: LastEventIDs is unbounded — it grows with the number of rows tied at the boundary event_time. The cursor is round-tripped through ListEventsRequest.cursor, which the SDK proto validates at max_bytes: 4096. Roughly 70+ UUID event IDs in one tie would produce a base64 cursor over that limit and the next ListEvents call would be rejected, wedging the feed. Cap the slice (or switch to a (event_time, event_id) keyset cursor as suggested on the query).
| func (u *userBuilder) Get(ctx context.Context, resourceId *v2.ResourceId, parentResourceId *v2.ResourceId) (*v2.Resource, annotations.Annotations, error) { | ||
| var workspaceId string | ||
| if parentResourceId.GetResourceType() == workspaceResourceType.Id { | ||
| workspaceId = parentResourceId.Resource | ||
| } | ||
|
|
||
| user, rateLimitData, err := u.client.GetUser(ctx, workspaceId, resourceId.Resource) |
There was a problem hiding this comment.
🟡 Suggestion: the SDK's GetResource passes request.GetParentResourceId() straight through, which is nil when C1 has no parent recorded for the resource — and under workspace-token auth userResource deliberately omits WithParentResourceID, so those user resources are stored parentless. parentResourceId is then handed to u.userResource, which dereferences parent.ResourceType at users.go:65 and panics. Same shape in servicePrincipalBuilder.Get → servicePrincipalResource (service-principals.go:32) and roleBuilder.Get → roleResource (roles.go:56). A nil guard in the Get methods (or switching those constructors to parent.GetResourceType()) removes the crash.
| func (w *workspaceBuilder) Get(ctx context.Context, resourceId *v2.ResourceId, parentResourceId *v2.ResourceId) (*v2.Resource, annotations.Annotations, error) { | ||
| workspace, _, err := w.client.GetWorkspace(ctx, resourceId.Resource) | ||
| if err != nil { | ||
| return nil, nil, fmt.Errorf("databricks-connector: failed to get workspace %s: %w", resourceId.Resource, err) | ||
| } |
There was a problem hiding this comment.
🟡 Suggestion: GetWorkspace resolves through ListWorkspaces, which hits the Account API. List above deliberately handles the token-auth case by building minimalWorkspaceResource from the configured deployment names instead. Now that CAPABILITY_TARGETED_SYNC is advertised for workspaces, a targeted sync under workspace-token auth will always fail here. Mirroring List's w.client.IsTokenAuth() branch would keep the two paths consistent.
| } | ||
|
|
||
| nativeId, ok := row.RequestParams[mapping.idParam] | ||
| if !ok || nativeId == "" { | ||
| return nil | ||
| } | ||
|
|
||
| resourceId := &v2.ResourceId{ResourceType: mapping.resourceType.Id, Resource: nativeId} | ||
| if mapping.resourceType == groupResourceType { | ||
| resourceId.Resource = groupResourceId(context.Background(), nativeId, parent) | ||
| } | ||
|
|
||
| affected = append(affected, affectedResource{resourceId: resourceId, parentResourceId: parent}) |
There was a problem hiding this comment.
🟡 Suggestion: when the Account API is available, groups/users/service principals are synced only as children of the account (accountResource), while minimalWorkspaceResource parents them to the workspace only under token auth. Here any audit row carrying a non-zero workspace_id builds the resource under workspaceParent, so in Account-API mode a workspace-scoped SCIM row emits workspace/<deployment>/group/<id>, which is not the ID C1 has synced — the account-parented copy never gets refreshed and the Get may 404. Consider selecting the parent from f.client.IsAccountAPIAvailable() the way groupGrantParent in helpers.go already does.
Nit: context.Background() on line 309 discards the request context; mapAuditRowToResource could take the caller's ctx (or groupResourceId's ctx parameter could be dropped since it's already unused).
Connector PR Review: CXP-897 Incremental sync supportBlocking Issues: 1 | Suggestions: 7 | Threads Resolved: 0 Review SummaryThe new commit drops Security IssuesNone found. The audit-log SQL statement interpolates only compile-time constants ( Correctness Issues
Suggestions
Prompt for AI agents |
| return nil, fmt.Errorf("databricks-connector: sql-warehouse-id is required when incremental sync is enabled") | ||
| } | ||
|
|
||
| auditWorkspaces, _, err := d.client.ListWorkspaces(ctx) |
There was a problem hiding this comment.
[Critical] heads up — this hits the account API (ListWorkspaces) unconditionally, but that's unreachable under workspace-token auth (see the IsTokenAuth() check + d.workspaces fallback right above this in Validate()). Same thing happens in audit_event_feed.go's ListEvents. As-is this fails Validate() for every workspace-token customer who turns on incremental sync. Probably needs the same IsTokenAuth() guard + d.workspaces fallback here.
| cursor = eventPageCursor{StartAt: start} | ||
| } | ||
|
|
||
| workspaces, _, err := f.client.ListWorkspaces(ctx) |
There was a problem hiding this comment.
[Critical] same account-API-unreachable-under-token-auth issue as connector.go's Validate() — no IsTokenAuth() guard, and this struct doesn't even have d.workspaces to fall back to. Also (separate, lower severity): this refetches the whole workspace list on every single poll even though Validate() already fetched it once at startup — might be worth caching instead of hitting the API every cycle.
|
|
||
| // sqlQueryWorkspace deterministically picks the workspace used to run the audit log query | ||
| // and builds the workspace-ID-to-deployment-name lookup used to resolve audit rows. | ||
| func sqlQueryWorkspace(workspaces []databricks.Workspace) (string, map[int64]string) { |
There was a problem hiding this comment.
[High] sqlQueryWorkspace just picks whichever workspace sorts alphabetically first by deployment name and routes the warehouse query through it — but SQL warehouses are workspace-scoped, so if the real warehouse doesn't live in that workspace this just breaks. No config field lets you pin the right one, and --workspaces doesn't help since it's not wired into this path at all. Might need something like a --sql-warehouse-workspace flag.
| return parseAuditLogRows(result) | ||
| } | ||
|
|
||
| func quotedInClause(values []string) string { |
There was a problem hiding this comment.
[Low] this is just strings.Join with extra steps — could be:
quoted := make([]string, len(values))
for i, v := range values {
quoted[i] = "'" + v + "'"
}
return strings.Join(quoted, ", ")| return nil, err | ||
| } | ||
|
|
||
| switch res.Status.State { |
There was a problem hiding this comment.
[Low] nit: this switch only really has two outcomes (success vs error), could just be if res.Status.State != StatementStateSucceeded { ... }. Purely cosmetic.
|
|
||
| // auditLogActions maps audit log action_name values to the resources they affect. | ||
| var auditLogActions = map[string]auditActionMapping{ | ||
| "createGroup": {resourceType: groupResourceType, idParam: "targetGroupId"}, |
There was a problem hiding this comment.
Suggestion: key this mapping by both service_name and action_name, then use the documented IAM event names. The current filter silently excludes common changes: batch membership uses addPrincipalsToGroup/removePrincipalsFromGroup, user creation uses add, group deletion uses removeGroup, and account-admin changes use setAccountAdmin/removeAccountAdmin.
Select service_name in the query and map the documented (service, action) pairs so generic names such as add remain unambiguous. Reference: https://docs.databricks.com/aws/en/admin/account-settings/audit-logs
| WorkspaceTokensField, | ||
| BaseURLField, | ||
| ExcludeWorkspacesField, | ||
| EnableIncrementalSyncField, |
There was a problem hiding this comment.
Suggestion: also add EnableIncrementalSyncField and SQLWarehouseIDField to every auth field group that supports incremental sync. They are in configFields (here) but missing from both group Fields lists below, so the grouped schema does not associate them with a selectable auth mode and SDK validation skips them for that mode.
Keep common feature fields in each applicable auth group. Pattern: https://github.com/ConductorOne/baton-azure-devops/blob/47b239de197e4c4c35da801e08c59ba6009f78e8/pkg/config/config.go#L215-L267
ORDER BY event_time ASC alone gives no deterministic ordering among rows sharing an event_time, so paging via a >= start_time filter plus a remembered ID set can stall forever if a single event_time has >= auditLogPageLimit rows. Order by (event_time, event_id) and page with a composite (event_time, event_id) > predicate instead, so the cursor always advances regardless of how many rows share a timestamp.
pollStatement could block for as long as the caller's context allowed if a warehouse got stuck PENDING/RUNNING (cold start, queued, quota), hanging Validate() indefinitely when incremental sync is enabled. Cap polling at statementPollMaxWait and cancel the statement via DELETE when giving up so it stops occupying the warehouse.
The SDK's GetResource passes request.GetParentResourceId() straight through to Get, which is nil whenever C1 has no parent recorded for the resource (e.g. workspace-token auth, where userResource deliberately omits WithParentResourceID). That nil parent was then dereferenced directly in userResource, servicePrincipalResource, and roleResource, panicking on resync after a RESOURCE_CHANGE event. Use the nil-safe GetResourceType()/GetResource() getters instead, matching the pattern groups.go already used.
Get always called GetWorkspace, which hits the Account API via ListWorkspaces. The Account API is unreachable under workspace-token auth, so any targeted sync of a workspace (advertised via CAPABILITY_TARGETED_SYNC) always failed in that mode, even though List already builds a minimalWorkspaceResource from the configured workspace list to avoid the same call. Get now mirrors that branch.
mapAuditRowToResource picked the user/group/service-principal parent from whether the audit row carried a workspace_id, not from how the resource is actually synced. Users/groups/service principals are only ever synced under the account when the Account API is reachable (accountResource declares them as children only then; groupGrantParent already encodes this rule), so a workspace-scoped row in that mode built an ID that was never synced (e.g. workspace/<deployment>/group/x instead of account/<id>/group/x), and the real resource never got refreshed. Select the parent from IsAccountAPIAvailable() instead, matching groupGrantParent. Also thread the real ctx through instead of context.Background(), now that groupResourceId's ctx parameter is actually used for something worth passing correctly.
Validate() and ListEvents() both called ListWorkspaces unconditionally to resolve the audit-log query workspace, but the Account API is unreachable under workspace-token auth — so incremental sync always failed Validate() and every ListEvents poll for token-auth customers. Add resolveSQLWorkspaces, which builds minimal workspaces from the configured deployment names under token auth (mirroring workspaceBuilder.List's token-auth branch) instead of calling ListWorkspaces, and use it in both places. auditEventFeed now carries the configured workspace list to support this. Workspace-scoped audit rows can't be resolved to a deployment name under token auth this way (no numeric workspace ID is ever learned), so they're skipped by mapAuditRowToResource rather than mis-resolved — a known limitation, not a regression, since incremental sync couldn't run under token auth at all before this.
quotedInClause hand-rolled a strings.Join; use it directly. The two-case switch in ExecuteStatement (success vs everything else) reads clearer as a plain if. Cosmetic only, per review feedback.
sqlQueryWorkspace picked whichever workspace sorted alphabetically first by deployment name to run the system.access.audit query, with no way to route it to the workspace that actually hosts sql-warehouse-id. SQL warehouses only exist in one workspace, so querying through the wrong one 404s — this broke deterministically for any account with more than one workspace, unless the warehouse happened to live in the alphabetically-smallest one. Add --sql-warehouse-workspace to pin the deployment name explicitly, validated against the resolved workspace list in both Validate() and ListEvents() via a shared resolveQueryWorkspace helper. When unset and more than one workspace is available, log a Debug line naming the arbitrarily-picked workspace and pointing at the new flag, so it's diagnosable without escalating to Warn/Error for what is, until set, a config gap rather than a connector fault. Regenerated config_schema.json and pkg/config/conf.gen.go; updated README's incremental sync section and flag list.
| f.sqlWarehouseID, | ||
| statement, | ||
| databricks.StatementParameter{Name: "start_date", Value: cursor.StartAt.UTC().Format("2006-01-02"), Type: "DATE"}, | ||
| databricks.StatementParameter{Name: "start_time", Value: cursor.StartAt.UTC().Format(time.RFC3339), Type: "TIMESTAMP"}, |
There was a problem hiding this comment.
🟠 Bug: time.RFC3339 has no fractional-second component, so start_time is floored to the whole second while cursor.StartAt keeps millisecond precision from parseAuditLogRows. That makes the (event_time, event_id) tiebreaker inert: for a boundary row at 10:00:00.500 the predicate becomes event_time > '10:00:00', which re-matches every already-processed row in that second (the event_time = :start_time branch never fires). Every poll re-emits the tail of the last second, and if ≥auditLogPageLimit rows share one second the full page returns the same rows forever and the feed stalls. Use time.RFC3339Nano (or "2006-01-02 15:04:05.999999") here.
| // mapAuditRowToResource — a known limitation of token auth, not a regression, since | ||
| // incremental sync couldn't run under token auth at all before this. | ||
| func resolveSQLWorkspaces(ctx context.Context, client *databricks.Client, configuredWorkspaces []string) ([]databricks.Workspace, error) { | ||
| if client.IsTokenAuth() { |
There was a problem hiding this comment.
🟡 Suggestion: under token auth every audit row is dropped, not just workspace-scoped ones. The minimal workspaces all have ID == 0, so workspaceLookup can never resolve a real workspace_id (rows with WorkspaceID != 0 return nil at line 291), and account-scoped rows (WorkspaceID == 0) also return nil because accountAPIAvailable is false and workspaceParent is nil (line 308). So enable-incremental-sync passes Validate() and then polls the SQL warehouse forever producing zero events. Consider rejecting incremental sync in Validate() under token auth (or at least logging a Warn and documenting the OAuth requirement in README.md).
| ExcludeWorkspacesField, | ||
| EnableIncrementalSyncField, | ||
| SQLWarehouseIDField, | ||
| SQLWarehouseWorkspaceField, |
There was a problem hiding this comment.
🟡 Suggestion: enable-incremental-sync, sql-warehouse-id, and sql-warehouse-workspace were added to configFields but to neither entry in WithFieldGroups below, unlike every other connector-specific field (workspaces, base-url, databricks-exclude-workspaces, …) which appear in both groups. config_schema.json's fieldGroups confirms the omission. If the UI renders fields per selected auth group, these three won't be settable there — add them to the oauth2 group (and to workspace-token if that mode should support incremental sync).
| } | ||
|
|
||
| var res statementResponse | ||
| if _, err := c.Post(ctx, u, body, &res); err != nil { |
There was a problem hiding this comment.
🟡 Suggestion: the *v2.RateLimitDescription from Post (and from Get in pollStatement/collectStatementResult) is discarded, so ListEvents always returns nil annotations. A 429 from the Statement Execution API won't be surfaced to the SDK's rate-limit handling (mixin C3). Consider returning the rate-limit data alongside *StatementResult and attaching it via annos.WithRateLimiting(...) in ListEvents.
| func decodeEventCursor(s string) eventPageCursor { | ||
| if s == "" { | ||
| return eventPageCursor{} | ||
| } | ||
|
|
||
| raw, err := base64.StdEncoding.DecodeString(s) | ||
| if err != nil { | ||
| return eventPageCursor{} | ||
| } | ||
|
|
||
| var c eventPageCursor | ||
| if err := json.Unmarshal(raw, &c); err != nil { | ||
| return eventPageCursor{} | ||
| } | ||
|
|
||
| return c | ||
| } |
There was a problem hiding this comment.
🟡 Suggestion: a corrupt/undecodable cursor silently resets the watermark to now - auditLogLookback, which permanently skips any event older than one hour that hadn't been emitted yet. The self-heal is reasonable, but it's an invisible data gap — take a ctx and log at Warn (with the decode error) so the skip is observable.
- Use RFC3339Nano for start_time so the (event_time, event_id) tiebreaker keeps sub-second precision, fixing duplicate re-emission at page boundaries. - Reject enable-incremental-sync under workspace token auth at Validate(), since no audit row can ever resolve to a synced resource that way and it would otherwise poll forever for nothing. - Add the three incremental-sync config fields to both field groups so they're selectable in the UI; regenerate config_schema.json. - Propagate rate-limit info from the SQL Statement Execution API through ExecuteStatement into ListEvents' annotations. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Requires another review
decodeEventCursor now logs at Debug when a cursor fails to decode, so the watermark reset is observable instead of invisible. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
| Fields: []field.SchemaField{AccountIdField, WorkspacesField, WorkspaceTokensField, HostnameField, AccountHostnameField, BaseURLField, ExcludeWorkspacesField}, | ||
| Fields: []field.SchemaField{ | ||
| AccountIdField, WorkspacesField, WorkspaceTokensField, HostnameField, AccountHostnameField, BaseURLField, ExcludeWorkspacesField, | ||
| EnableIncrementalSyncField, SQLWarehouseIDField, SQLWarehouseWorkspaceField, |
There was a problem hiding this comment.
🟡 Suggestion: the three incremental-sync fields are now offered under the workspace-token group, but Validate (connector.go:171) hard-fails with "incremental sync is not supported with workspace token auth" whenever IsTokenAuth() is true. That surfaces a config option in the UI that can never succeed for this auth mode. Consider dropping them from this group (and from config_schema.json's token-auth constraint), and noting the OAuth-only requirement in the README's "Incremental sync" section.
There was a problem hiding this comment.
Fixed at f8ebc73
| AccountIdField, WorkspacesField, WorkspaceTokensField, HostnameField, AccountHostnameField, BaseURLField, ExcludeWorkspacesField, | ||
| EnableIncrementalSyncField, SQLWarehouseIDField, SQLWarehouseWorkspaceField, | ||
| }, | ||
| Default: false, |
There was a problem hiding this comment.
🟡 Suggestion: now that Fields: above is a multi-line entry, gofmt collapses the alignment run and would emit Default: false, with a single space (compare the OAuth2 group's Default: true, on line 129). Minor gofmt drift.
| Default: false, | |
| Default: false, |
There was a problem hiding this comment.
Fixed on f8ebc73
Validate() rejects enable-incremental-sync under token auth, so offering those fields in the workspace-token group let the UI present an option that can never succeed. Remove them from that group (they stay in oauth2) and regenerate config_schema.json; note the OAuth-only requirement in the README. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
| return workspaces, nil | ||
| } | ||
|
|
||
| workspaces, _, err := client.ListWorkspaces(ctx) |
There was a problem hiding this comment.
🟠 Bug: resolveSQLWorkspaces ignores the --workspaces allowlist on the OAuth2 path — and since Validate now rejects token auth, that's the only path that runs. client.ListWorkspaces applies --databricks-exclude-workspaces only; the allowlist filter lives in workspaceBuilder.List (matchConfiguredWorkspace, workspaces.go:121). So workspaceLookup covers workspaces that were never synced, and mapAuditRowToResource emits RESOURCE_CHANGE events parented to them, pulling out-of-scope workspaces/roles back in via targeted sync. Filter the ListWorkspaces result through matchConfiguredWorkspace(configuredWorkspaces, ...) when configuredWorkspaces is non-empty.
| return resource, nil, nil | ||
| } | ||
|
|
||
| workspace, _, err := w.client.GetWorkspace(ctx, resourceId.Resource) |
There was a problem hiding this comment.
🟡 Suggestion: the token-auth branch above (line 290) rejects a workspace that isn't in w.workspaces, but this OAuth2 branch doesn't — GetWorkspace only sees the exclude-list filtering done inside ListWorkspaces. A targeted sync for a workspace outside --workspaces will succeed here and re-add a resource the operator scoped out. Apply the same matchConfiguredWorkspace check used in List before returning.
| defer cancel() | ||
|
|
||
| u := c.workspaceUrl(workspaceId).JoinPath(statementsEndpoint, statementId) | ||
| if _, err := c.Delete(ctx, u); err != nil { |
There was a problem hiding this comment.
🟡 Suggestion: the doc comment says this "cancels the statement so it stops occupying the warehouse", but DELETE /api/2.0/sql/statements/{id} is the close operation, which Databricks only honours once the statement has reached a terminal state. A statement still PENDING/RUNNING at statementPollMaxWait therefore keeps running (and keeps billing the warehouse). Use POST /api/2.0/sql/statements/{id}/cancel here instead.
| if cursor.StartAt.IsZero() { | ||
| start := now.Add(-auditLogLookback) | ||
| if earliestEvent != nil && earliestEvent.AsTime().After(start) { | ||
| start = earliestEvent.AsTime() | ||
| } | ||
| cursor = eventPageCursor{StartAt: start} | ||
| } |
There was a problem hiding this comment.
🟡 Suggestion: this takes the later of now-1h and earliestEvent, so when the SDK asks for events since (say) the last full sync 6 hours ago, the window is silently narrowed to 1 hour and those 5 hours of changes are never emitted — they only surface at the next full sync. auditLogLookback reads as a floor for "no watermark yet", not a cap on an explicit request; consider if earliestEvent != nil && earliestEvent.AsTime().Before(start) { start = earliestEvent.AsTime() } (optionally bounded by a documented max lookback).
| if hasMore { | ||
| return eventPageCursor{StartAt: latest, StartAfterEventID: lastEventID, LatestEventSeen: latest} | ||
| } | ||
|
|
||
| target := latest.Add(-auditLogTrailingLag) | ||
| if len(rows) == 0 { | ||
| target = now.Add(-auditLogTrailingLag) | ||
| } | ||
| if target.Before(cursor.StartAt) { | ||
| target = cursor.StartAt | ||
| } |
There was a problem hiding this comment.
🟡 Suggestion: the hasMore branch moves StartAt all the way to the last row's event_time with no trailing lag, and the never-regress floor on line 265-267 then prevents the lag from ever being re-applied. So after a single burst of ≥1000 matching events, StartAt is pinned at "newest row seen" forever and the late-indexed rows that auditLogTrailingLag exists to catch are skipped permanently — the exact failure the constant's comment describes. Consider clamping the full-page advance to min(lastRow.EventTime, now-auditLogTrailingLag), or tracking the lagged floor separately from the paging boundary.
5 participants
No description provided.