From ec3f7c058c984e29ff7b7d8f512b73ca39ea09db Mon Sep 17 00:00:00 2001 From: Mateo Hernandez Date: Tue, 3 Nov 2026 15:22:28 -0300 Subject: [PATCH 1/4] feat(databricks): add unity catalog table and volume sync CXH-2723. Tables and volumes hang from the schema, each with its own privilege set and a read-only owner. The tree is unchanged from d10a7e9e, the state reviewed and approved on #73. Co-Authored-By: Claude Opus 5 --- README.md | 30 ++- baton_capabilities.json | 155 ++++++++++++ docs/connector.mdx | 16 +- docs/docs-info.md | 9 +- pkg/connector/catalog.go | 1 - pkg/connector/connector.go | 4 +- pkg/connector/connector_test.go | 5 +- pkg/connector/metastore.go | 1 - pkg/connector/raw_id_test.go | 13 + pkg/connector/resource_types.go | 16 ++ pkg/connector/schema.go | 53 +++- pkg/connector/schema_child.go | 357 ++++++++++++++++++++++++++ pkg/connector/securable_helpers.go | 94 +++++-- pkg/connector/table.go | 358 +++++++++++++++++++++++++++ pkg/connector/volume.go | 208 ++++++++++++++++ pkg/databricks/models.go | 22 ++ pkg/databricks/unity_catalog.go | 102 ++++++++ pkg/databricks/unity_catalog_test.go | 40 +++ pkg/databricks/vars.go | 14 ++ 19 files changed, 1449 insertions(+), 49 deletions(-) create mode 100644 pkg/connector/schema_child.go create mode 100644 pkg/connector/table.go create mode 100644 pkg/connector/volume.go diff --git a/README.md b/README.md index 85bae015..1e799d3d 100644 --- a/README.md +++ b/README.md @@ -83,6 +83,9 @@ baton resources - Roles - Metastores - Catalogs (under their metastore) +- Schemas (under their catalog) +- Tables (under their schema) +- Volumes (under their schema) The connector fetches all resources from the account and every workspace the service principal can access. To limit the scope, pass a comma-separated list of @@ -95,9 +98,9 @@ other metastores and their grants from ConductorOne. Catalogs must also be visible through a listed workspace. A list that names no workspace on the account is rejected during credential validation, before any sync runs. -Metastores and catalogs are opt-in: each resource type stays off until enabled -for the connector in ConductorOne. Catalog sync uses the existing OAuth -credentials and configuration fields. +Metastores, catalogs, schemas, tables, and volumes are Unity Catalog securables +and are opt-in: each type stays off until it is enabled for the connector in +ConductorOne. They use the existing OAuth credentials and configuration fields. ## Authentication @@ -223,6 +226,8 @@ account └── metastore └── catalog └── schema + ├── table + └── volume ``` A metastore exposes its privileges as entitlements (`CREATE_CATALOG`, @@ -264,11 +269,11 @@ type) are excluded, regardless of ownership. ## Unity Catalog provisioning -A metastore or catalog privilege can be granted to and revoked from a user, -group or service principal. The `owner` entitlement is read-only: granting or -revoking it fails with an `InvalidArgument` error. +A privilege on a metastore, catalog, schema, table, or volume can be granted to +and revoked from a user, group or service principal. The `owner` entitlement is +read-only: granting or revoking it fails with an `InvalidArgument` error. -If a catalog becomes unreachable, provisioning must distinguish deletion from +If a securable becomes unreachable, provisioning must distinguish deletion from a workspace binding or visibility change. Requests can fail with `FailedPrecondition` when deletion cannot be proved. @@ -276,9 +281,14 @@ a workspace binding or visibility change. Requests can fail with The service principal must own the metastore, or be assigned **Admin** directly on at least one running workspace attached to it. Admin inherited only through a -group is not enough. Databricks answers an under-permissioned read with HTTP 200 -and only the rows the caller may see, so the connector reads a metastore only -through a workspace that passes that check. When no attached workspace does, the +group is not enough. The connector reads a metastore only through a workspace +that passes that check. Listing every schema takes ownership of the catalog, or +`MANAGE`, `USE_SCHEMA`, or `ALL_PRIVILEGES` on that catalog. Listing every table +or volume needs more: ownership of the metastore, the catalog, or the schema, +direct Admin when the workspace admins group owns the catalog, or +`ALL_PRIVILEGES` on the catalog. `MANAGE` and `USE_SCHEMA` are not enough for +tables or volumes. Databricks answers an under-permissioned read with HTTP 200 +and only the rows the caller may see. The connector refuses that read, so the sync fails and ConductorOne keeps the grants it already has. For a catalog's full grants list, the service principal must own the metastore diff --git a/baton_capabilities.json b/baton_capabilities.json index 9633f1b3..407402fc 100644 --- a/baton_capabilities.json +++ b/baton_capabilities.json @@ -377,6 +377,82 @@ ] } }, + { + "resourceType": { + "id": "table", + "displayName": "Table", + "traits": [ + "TRAIT_APP" + ], + "annotations": [ + { + "@type": "type.googleapis.com/c1.connector.v2.OptInRequired" + }, + { + "@type": "type.googleapis.com/c1.connector.v2.CapabilityPermissions", + "permissions": [ + { + "permission": "unity-catalog" + }, + { + "permission": "scim" + }, + { + "permission": "access-management" + }, + { + "permission": "provisioning" + }, + { + "permission": "Account admin" + }, + { + "permission": "Workspace admin" + }, + { + "permission": "USE_CATALOG" + }, + { + "permission": "USE_SCHEMA" + } + ] + } + ] + }, + "capabilities": [ + "CAPABILITY_SYNC", + "CAPABILITY_PROVISION" + ], + "permissions": { + "permissions": [ + { + "permission": "unity-catalog" + }, + { + "permission": "scim" + }, + { + "permission": "access-management" + }, + { + "permission": "provisioning" + }, + { + "permission": "Account admin" + }, + { + "permission": "Workspace admin" + }, + { + "permission": "USE_CATALOG" + }, + { + "permission": "USE_SCHEMA" + } + ] + }, + "optInRequired": true + }, { "resourceType": { "id": "user", @@ -418,6 +494,85 @@ ] } }, + { + "resourceType": { + "id": "volume", + "displayName": "Volume", + "traits": [ + "TRAIT_APP" + ], + "annotations": [ + { + "@type": "type.googleapis.com/c1.connector.v2.OptInRequired" + }, + { + "@type": "type.googleapis.com/c1.connector.v2.CapabilityPermissions", + "permissions": [ + { + "permission": "unity-catalog" + }, + { + "permission": "scim" + }, + { + "permission": "access-management" + }, + { + "permission": "provisioning" + }, + { + "permission": "Account admin" + }, + { + "permission": "Workspace admin" + }, + { + "permission": "USE_CATALOG" + }, + { + "permission": "USE_SCHEMA" + } + ] + }, + { + "@type": "type.googleapis.com/c1.connector.v2.SkipEntitlements" + } + ] + }, + "capabilities": [ + "CAPABILITY_SYNC", + "CAPABILITY_PROVISION" + ], + "permissions": { + "permissions": [ + { + "permission": "unity-catalog" + }, + { + "permission": "scim" + }, + { + "permission": "access-management" + }, + { + "permission": "provisioning" + }, + { + "permission": "Account admin" + }, + { + "permission": "Workspace admin" + }, + { + "permission": "USE_CATALOG" + }, + { + "permission": "USE_SCHEMA" + } + ] + }, + "optInRequired": true + }, { "resourceType": { "id": "workspace", diff --git a/docs/connector.mdx b/docs/connector.mdx index cfb7b05f..bd815ebd 100644 --- a/docs/connector.mdx +++ b/docs/connector.mdx @@ -17,7 +17,9 @@ sidebarTitle: "Databricks" | Roles | | | | Schemas | | | | Service principals | | | +| Tables | | | | Users | | | +| Volumes | | | | Workspaces | | | The Databricks connector supports [automatic account provisioning and deprovisioning](/product/admin/account-provisioning). @@ -30,25 +32,25 @@ Revoking the workspace **member** entitlement removes the principal's entire wor [This connector syncs non-human identities](/product/admin/nhi) and displays them on the **Identities overview** dashboard. -## Metastore, catalog and schema access in C1 +## Metastore, catalog, schema, table and volume access in C1 The connector syncs Databricks [Unity Catalog](https://docs.databricks.com/aws/en/data-governance/unity-catalog/manage-privileges/privileges) -metastores, the catalogs in them, and the schemas in those catalogs as C1 resources. Each privilege Databricks +metastores, the catalogs in them, the schemas in those catalogs, and the tables and volumes in those schemas as C1 resources. Each privilege Databricks records on one of them (`CREATE_CATALOG` on a metastore, `USE_CATALOG` or -`SELECT` on a catalog, `USE_SCHEMA` on a schema, and so on) becomes an entitlement in C1, so you can +`SELECT` on a catalog, `USE_SCHEMA` on a schema, `SELECT` on a table, `READ_VOLUME` on a volume, and so on) becomes an entitlement in C1, so you can review who holds it, request it, and revoke it for users, groups and service principals. A grant held by a group is expandable so C1 can resolve the group's members as holders. Ownership is shown as a read-only `owner` -entitlement — it cannot be granted or revoked through C1. +entitlement — it cannot be granted or revoked through C1. Views, materialized views, metric views and streaming tables are tables; `table_type` picks which privileges can be requested. `ALL_PRIVILEGES` on a view, and `MODIFY` on a streaming table, are shown when they already exist and cannot be requested. -The Metastores, Catalogs and Schemas resource types are turned off by default. Turn on the ones you want from the connector's configuration page in C1. Catalogs appear under their metastore and schemas under their catalog, and all of them use the same OAuth credentials; no additional configuration fields are needed. +The Metastores, Catalogs, Schemas, Tables and Volumes resource types are turned off by default. Turn on the ones you want from the connector's configuration page in C1. Each appears under its parent securable, and all of them use the same OAuth credentials; no additional configuration fields are needed. **Notes on Unity Catalog access in C1:** -- Ownership is shown but cannot be provisioned. An owner that is not a Databricks account identity, such as a system account or a workspace admins group, is not shown at all. +- Ownership is shown but cannot be provisioned. An owner that is not an account identity, such as `System user`, is not shown. A workspace admins group is shown when it resolves as an account group. - **MANAGE** and **ALL_PRIVILEGES** are not offered on a metastore, because Databricks rejects both at that level. - A catalog is synced when any running workspace in scope attached to its metastore lists it, and the same catalog name is kept once per metastore. A catalog that only excluded workspaces can reach is not synced. - The built-in `system` catalog and catalogs Databricks identifies as `SYSTEM_CATALOG` are excluded regardless of ownership. This includes `samples` when Databricks reports that type. @@ -214,6 +216,8 @@ Workspace **Admin** does not list every schema, except when the catalog is owned A catalog whose schemas cannot all be listed fails the sync. `system` and `samples` are not a special case of that listing: the catalog sync leaves them out, when their privilege list cannot be proven complete, before their schemas are read. +Listing every table or volume needs more than **USE SCHEMA** or **MANAGE**. The service principal must own the metastore, the catalog, or the schema, be a direct Admin when the workspace admins group owns the catalog, or hold **ALL PRIVILEGES** on the catalog. Otherwise the sync fails. + **Done.** Here's the set of credentials you'll need when setting up the connector: - Account ID diff --git a/docs/docs-info.md b/docs/docs-info.md index 31b6f469..4738d241 100644 --- a/docs/docs-info.md +++ b/docs/docs-info.md @@ -6,7 +6,7 @@ While developing the connector, please fill out this form. This information is n > The connector syncs the Databricks account, workspaces, users, groups, service principals, and roles. > - > It also syncs Unity Catalog metastores and their catalogs. Both resource types are opt-in — each stays off until enabled per resource type in ConductorOne. + > It also syncs Unity Catalog metastores, catalogs, schemas, tables, and volumes. Those resource types are opt-in — each stays off until enabled per resource type in ConductorOne. 2. Can the connector provision any resources? If so, which ones? @@ -14,7 +14,7 @@ While developing the connector, please fill out this form. This information is n > > - **User accounts**: create and delete account users. > - **Entitlements**: grant and revoke role and membership assignments on accounts, workspaces, groups, service principals, and roles. - > - **Unity Catalog privileges**: grant and revoke a metastore or catalog privilege for a user, group or service principal. The `owner` entitlement is read-only and cannot be granted or revoked. + > - **Unity Catalog privileges**: grant and revoke a privilege on a metastore, catalog, schema, table, or volume for a user, group or service principal. The `owner` entitlement is read-only and cannot be granted or revoked. ## Connector credentials @@ -57,7 +57,8 @@ While developing the connector, please fill out this form. This information is n - **Direct grants only.** Databricks inherits privileges downward. The connector reports the assignments stored on the object. The inheritance endpoint does not return principal IDs. - **Full grants visibility is required.** Metastore ownership or direct workspace Admin permits metastore reads. Catalog reads require one of the catalog permissions listed above; workspace Admin applies only when that workspace's admins group owns the catalog. Admin inherited through a group does not count. The sync fails instead of reporting a partial list, so ConductorOne keeps the grants it already has. -- **Ownership is read-only.** It is shown and cannot be granted or revoked. An owner that is not an account identity (`System user`, `_workspace_admins_…`) is not shown. +- **A partial list fails the sync.** A privilege read on a catalog, schema, table, or volume is complete when the service principal owns the metastore, owns the object or a parent, or holds `MANAGE`. `ALL_PRIVILEGES` does not include `MANAGE`. A schema listing is also complete with `USE_SCHEMA` or `ALL_PRIVILEGES`, or when the catalog is owned by the workspace admins group and the service principal is a direct Admin. A table or volume listing is not: it needs ownership of the metastore, catalog, or schema, that same Admin standing, or `ALL_PRIVILEGES` on the catalog. Without that standing the sync fails naming the object, and ConductorOne keeps what it already has. `system` and `samples` (any `SYSTEM_CATALOG`) are left out of the catalog sync when their privilege list cannot be proven complete. While the service principal owns the metastore, both are synced, including their schemas and tables. +- **Ownership is read-only.** It is shown and cannot be granted or revoked. An owner that does not resolve to an account identity, such as `System user`, is not shown. A workspace admins group is shown when it resolves as an account group. - **No `MANAGE` or `ALL_PRIVILEGES` on a metastore.** Databricks rejects both, so they are not offered. - **Read through one running workspace.** Privileges are not served on the account plane. The connector reads each metastore through the first running workspace in scope where it can see every privilege. With no running workspace in scope, the metastore is not synced and ConductorOne removes it. A workspace that is still provisioning fails the sync until it is `RUNNING`. An unreadable assignment list fails the sync and ConductorOne keeps the metastore. - **`workspaces` scopes metastores.** An allowlist that names no workspace on the account fails at credential validation, before any sync. A name that is also in `databricks-exclude-workspaces` is reported separately from a name the account does not have. @@ -65,4 +66,4 @@ While developing the connector, please fill out this form. This information is n - **Catalog discovery visits every attached workspace in scope.** The connector asks each running workspace attached to the metastore, including for `ISOLATED` catalogs bound to only some workspaces. Each catalog name is kept once per metastore. A catalog bound only to workspaces the sync excludes is not listed, and ConductorOne removes it with its grants. - **System catalogs are excluded regardless of ownership.** The built-in `system` catalog and any catalog with `CatalogType == "SYSTEM_CATALOG"` are excluded, including `samples` when it has that type. - **Seeing a catalog needs more than `MANAGE`.** The list holds the catalogs the service principal owns or holds `USE_CATALOG` on, or every catalog when it is the metastore admin. `MANAGE` alone reads a catalog's privileges but does not make it appear. -- **Not synced yet:** tables, volumes, functions, registered models, external locations, storage credentials, connections, shares, recipients, and tags. +- **Not synced yet:** functions, registered models, external locations, storage credentials, connections, shares, recipients, and tags. diff --git a/pkg/connector/catalog.go b/pkg/connector/catalog.go index 36eba554..895a0ca6 100644 --- a/pkg/connector/catalog.go +++ b/pkg/connector/catalog.go @@ -122,7 +122,6 @@ var errCatalogGone = errors.New("catalog no longer exists") type catalogBuilder struct { securableOps - // workspaces is the configured allowlist, empty when none was given. workspaces map[string]struct{} } diff --git a/pkg/connector/connector.go b/pkg/connector/connector.go index f06c4efc..45d0ca84 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -45,6 +45,8 @@ func (d *Databricks) ResourceSyncers(_ context.Context) []connectorbuilder.Resou newMetastoreBuilder(d.client, d.clientID, workspaces, principals), newCatalogBuilder(d.client, d.clientID, workspaces, principals), newSchemaBuilder(d.client, d.clientID, workspaces, principals), + newTableBuilder(d.client, d.clientID, workspaces, principals), + newVolumeBuilder(d.client, d.clientID, workspaces, principals), } } @@ -76,7 +78,7 @@ func (d *Databricks) Metadata(_ context.Context) (*v2.ConnectorMetadata, error) return &v2.ConnectorMetadata{ DisplayName: "Databricks", Description: "Connector syncing Databricks workspaces, users, groups, service principals and roles to Baton, " + - "along with Unity Catalog data assets — metastores, catalogs and schemas — and the privileges granted on them", + "along with Unity Catalog data assets — metastores, catalogs, schemas, tables and volumes — and the privileges granted on them", AccountCreationSchema: &v2.ConnectorAccountCreationSchema{ FieldMap: map[string]*v2.ConnectorAccountCreationSchema_Field{ "email": { diff --git a/pkg/connector/connector_test.go b/pkg/connector/connector_test.go index b2d96dcb..e5e41cbc 100644 --- a/pkg/connector/connector_test.go +++ b/pkg/connector/connector_test.go @@ -87,7 +87,10 @@ func TestResourceSyncersAlwaysRegisterTheUnityCatalogBuilders(t *testing.T) { for _, syncer := range d.ResourceSyncers(ctx) { registered = append(registered, syncer.ResourceType(ctx).GetId()) } - for _, want := range []string{metastoreResourceType.Id, catalogResourceType.Id, schemaResourceType.Id} { + for _, want := range []string{ + metastoreResourceType.Id, catalogResourceType.Id, schemaResourceType.Id, + tableResourceType.Id, volumeResourceType.Id, + } { if !slices.Contains(registered, want) { t.Errorf("registered syncers = %v, want them to include %s", registered, want) } diff --git a/pkg/connector/metastore.go b/pkg/connector/metastore.go index 8e3a2d28..5efb8970 100644 --- a/pkg/connector/metastore.go +++ b/pkg/connector/metastore.go @@ -26,7 +26,6 @@ var _ connectorbuilder.StaticEntitlementSyncerV2 = (*metastoreBuilder)(nil) type metastoreBuilder struct { securableOps - // workspaces is the configured allowlist, empty when none was given. workspaces map[string]struct{} } diff --git a/pkg/connector/raw_id_test.go b/pkg/connector/raw_id_test.go index d0552e9d..c28a39db 100644 --- a/pkg/connector/raw_id_test.go +++ b/pkg/connector/raw_id_test.go @@ -84,6 +84,19 @@ func TestResourcesHaveRawId(t *testing.T) { &v2.ResourceId{ResourceType: catalogResourceType.Id, Resource: "ms1::cat"}, schemaRoute{})), "ms1::cat.sch", }, + { + // A table and a volume extend the schema's id with their own name. + "table", + build(tableResource(&databricks.Table{Name: "orders"}, "ms1", "cat", "sch", + &v2.ResourceId{ResourceType: schemaResourceType.Id, Resource: "ms1::cat.sch"}, schemaChildRoute{})), + "ms1::cat.sch.orders", + }, + { + "volume", + build(volumeResource(&databricks.Volume{Name: "files"}, "ms1", "cat", "sch", + &v2.ResourceId{ResourceType: schemaResourceType.Id, Resource: "ms1::cat.sch"}, schemaChildRoute{})), + "ms1::cat.sch.files", + }, } for _, tc := range cases { diff --git a/pkg/connector/resource_types.go b/pkg/connector/resource_types.go index 5f192937..b4174a93 100644 --- a/pkg/connector/resource_types.go +++ b/pkg/connector/resource_types.go @@ -101,4 +101,20 @@ var ( Traits: []v2.ResourceType_Trait{v2.ResourceType_TRAIT_APP}, Annotations: annotationsForStaticSecurableResourceType(securablePermissions(permissionUseCatalog, permissionUseSchema)...), } + + tableResourceType = &v2.ResourceType{ + Id: "table", + DisplayName: "Table", + Annotations: annotationsForSecurableResourceType(securablePermissions(permissionUseCatalog, permissionUseSchema)...), + } + + volumeResourceType = &v2.ResourceType{ + Id: "volume", + DisplayName: "Volume", + Annotations: annotations.New( + &v2.OptInRequired{}, + &v2.SkipEntitlements{}, + capabilityPermissions(securablePermissions(permissionUseCatalog, permissionUseSchema)...), + ), + } ) diff --git a/pkg/connector/schema.go b/pkg/connector/schema.go index 76103b87..cbed4df6 100644 --- a/pkg/connector/schema.go +++ b/pkg/connector/schema.go @@ -29,6 +29,10 @@ var _ connectorbuilder.StaticEntitlementSyncerV2 = (*schemaBuilder)(nil) const ( securableTypeSchema = "SCHEMA" + // errorCodeSchemaGone is what Databricks names in error_code when the schema + // a request addresses does not exist. + errorCodeSchemaGone = "SCHEMA_DOES_NOT_EXIST" + // informationSchema is the read-only schema Databricks puts in every catalog. // Its privileges are not grantable, so it is left out of the sync. informationSchema = "information_schema" @@ -57,10 +61,6 @@ const ( // schemaListRemedy is what a refusal to sync a catalog whose schema listing // would be partial tells the operator. schemaListRemedy = "the connector's service principal must own the metastore or the catalog, or hold MANAGE, USE_SCHEMA or ALL_PRIVILEGES on the catalog" - - // errorCodeSchemaGone is what Databricks names in error_code when the schema - // a request addresses does not exist. - errorCodeSchemaGone = "SCHEMA_DOES_NOT_EXIST" ) // schemaPrivileges is the catalog set minus the three that stop at the catalog. @@ -90,7 +90,6 @@ var schemaLevel = securableLevel{ type schemaBuilder struct { securableOps - // workspaces is the configured allowlist, empty when none was given. workspaces map[string]struct{} } @@ -169,7 +168,11 @@ func schemaResource( nil, rs.WithResourceProfile(profile), rs.WithParentResourceID(parent), - rs.WithAnnotation(&v2.RawId{Id: resourceID}), + rs.WithAnnotation( + &v2.ChildResourceType{ResourceTypeId: tableResourceType.Id}, + &v2.ChildResourceType{ResourceTypeId: volumeResourceType.Id}, + &v2.RawId{Id: resourceID}, + ), ) } @@ -361,8 +364,44 @@ func (b *schemaBuilder) pushRoute( catalogName, schemaListRemedy) } +// schemaThrough finds one schema in the listing of workspace, nil when that +// workspace does not list it. Its caller has already proven the listing holds +// every schema in the catalog, so an absence here is the schema being gone. +func schemaThrough( + ctx context.Context, + client *databricks.Client, + workspace, catalogName, schemaName string, + annos *annotations.Annotations, +) (*databricks.Schema, error) { + pageToken := "" + for { + schemas, next, rateLimit, err := client.ListSchemas(ctx, workspace, catalogName, pageToken, ResourcesPageSize) + noteRateLimit(annos, rateLimit) + if err != nil { + return nil, schemaListingError(catalogName, workspace, err) + } + + for _, schema := range schemas { + if schema.Name == schemaName { + return &schema, nil + } + } + + if next == "" { + return nil, nil + } + if next == pageToken { + return nil, status.Errorf(codes.Internal, + "databricks-connector: the schemas of catalog %s repeat page token %q through workspace %s", + catalogName, next, workspace) + } + pageToken = next + } +} + // schemaListingError returns NotFound only when Databricks names the missing -// catalog; any other refusal leaves the schema listing unknown. +// catalog. Any other refusal leaves the listing unknown rather than empty: a +// short list would read as deletions in C1. func schemaListingError(catalogName, workspace string, err error) error { if databricks.IsSecurableGone(err, errorCodeCatalogGone) { return uhttp.WrapErrors(codes.NotFound, diff --git a/pkg/connector/schema_child.go b/pkg/connector/schema_child.go new file mode 100644 index 00000000..8ecfecfa --- /dev/null +++ b/pkg/connector/schema_child.go @@ -0,0 +1,357 @@ +package connector + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "strconv" + "strings" + + "github.com/conductorone/baton-databricks/pkg/databricks" + v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" + "github.com/conductorone/baton-sdk/pkg/annotations" + "github.com/conductorone/baton-sdk/pkg/pagination" + rs "github.com/conductorone/baton-sdk/pkg/types/resource" + "github.com/conductorone/baton-sdk/pkg/uhttp" + "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap/ctxzap" + "go.uber.org/zap" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// profileKeySchemaOwner carries the owner of the parent schema. Owning the +// schema is one of the things that makes a table's or a volume's privilege read +// complete, and Grants is handed only the table, so List writes the owner down +// instead of listing the schemas again on every call. +const profileKeySchemaOwner = "schema_owner" + +// schemaChildRoute is the workspace a schema's tables or volumes are listed +// through, carried in the page token so later pages neither re-resolve the route +// nor prove again that the parent listing is complete. The owners ride along so +// each resource's profile can carry them. See profileKeySchemaOwner. +type schemaChildRoute struct { + Workspace string `json:"w"` + WorkspaceID string `json:"i"` + WorkspaceName string `json:"n"` + MetastoreOwner string `json:"o"` + CatalogOwner string `json:"c"` + SchemaOwner string `json:"s"` +} + +// listSchemaChildren pages one level that hangs from a schema — a table or a +// volume. pushSchemaChildRoute re-runs the schema syncer's completeness proof +// on the first page of every schema, once for tables and again for volumes. +// The privilege reads in that proof bypass the HTTP cache. The owners the +// proof learns are written onto each resource, so Grants does not repeat it. +// +// fetch reads one page through the routed workspace and turns it into resources. +func listSchemaChildren( + ctx context.Context, + client *databricks.Client, + clientID string, + workspaces map[string]struct{}, + resourceType *v2.ResourceType, + noun string, + parentResourceID *v2.ResourceId, + attr rs.SyncOpAttrs, + fetch func(ctx context.Context, route schemaChildRoute, metastoreID, catalogName, schemaName, pageToken string) ( + []*v2.Resource, string, *v2.RateLimitDescription, error), +) ([]*v2.Resource, *rs.SyncOpResults, error) { + if parentResourceID.GetResourceType() != schemaResourceType.Id { + return nil, nil, nil + } + + metastoreID, catalogName, schemaName, err := parseSchemaID(parentResourceID.GetResource()) + if err != nil { + return nil, nil, err + } + // The schema syncer leaves Databricks' own read-only schema out, so nothing + // under it is ever synced either. + if schemaName == informationSchema { + return nil, nil, nil + } + + annos := annotations.Annotations{} + bag := &pagination.Bag{} + if err := bag.Unmarshal(attr.PageToken.Token); err != nil { + return nil, nil, status.Errorf(codes.InvalidArgument, "databricks-connector: invalid %s page token: %v", noun, err) + } + + if bag.Current() == nil { + if err := pushSchemaChildRoute(ctx, client, clientID, workspaces, resourceType, noun, + metastoreID, catalogName, schemaName, bag, &annos); err != nil { + return nil, &rs.SyncOpResults{Annotations: annos}, err + } + if bag.Current() == nil { + ctxzap.Extract(ctx).Debug("databricks-connector: no workspace in sync scope lists the schema; listing none of its children", + zap.String("resource_type", resourceType.Id), + zap.String("schema", parentResourceID.GetResource())) + + return nil, &rs.SyncOpResults{Annotations: annos}, nil + } + } + + var route schemaChildRoute + if err := json.Unmarshal([]byte(bag.ResourceID()), &route); err != nil { + return nil, nil, status.Errorf(codes.InvalidArgument, "databricks-connector: invalid %s page token: %v", noun, err) + } + + rv, next, rateLimit, err := fetch(ctx, route, metastoreID, catalogName, schemaName, bag.PageToken()) + noteRateLimit(&annos, rateLimit) + if err != nil { + return nil, &rs.SyncOpResults{Annotations: annos}, + schemaChildListingError(noun, catalogName+schemaNameSeparator+schemaName, route.Workspace, err) + } + + nextToken, err := bag.NextToken(next) + if err != nil { + return nil, &rs.SyncOpResults{Annotations: annos}, err + } + + return rv, &rs.SyncOpResults{Annotations: annos, NextPageToken: nextToken}, nil +} + +// childListRemedy is what a refusal to sync a schema whose table or volume +// listing is partial tells the operator to fix. USE_SCHEMA and MANAGE list +// schemas; they do not list every table or volume. +const childListRemedy = "the connector's service principal must own the metastore, the catalog or the schema, " + + "or hold ALL_PRIVILEGES on the catalog" + +// pushSchemaChildRoute records a workspace that can list every table or volume +// of the schema. It walks the running in-scope workspaces until one lists the +// catalog and the child listing is complete. An error from a workspace stops +// the walk. An empty result is an error: it would make ConductorOne delete +// the tables and volumes. +func pushSchemaChildRoute( + ctx context.Context, + client *databricks.Client, + clientID string, + workspaces map[string]struct{}, + resourceType *v2.ResourceType, + noun string, + metastoreID, catalogName, schemaName string, + bag *pagination.Bag, + annos *annotations.Annotations, +) error { + scope, err := scopedWorkspaces(ctx, client, workspaces, annos) + if err != nil { + return err + } + + usable, inScope, err := scope.attached(ctx, client, metastoreID, annos) + if err != nil { + return assignmentListError(metastoreID, err) + } + if len(usable) == 0 { + if inScope { + return status.Errorf(codes.FailedPrecondition, + "databricks-connector: metastore %s has no running workspace in sync scope", metastoreID) + } + + return nil + } + + owner, err := metastoreOwner(ctx, client, metastoreID, annos) + if err != nil { + return err + } + + fullName := catalogName + schemaNameSeparator + schemaName + var sawCatalog bool + for i := range usable { + workspace := usable[i] + catalog, err := catalogThrough(ctx, client, workspace.DeploymentName, metastoreID, catalogName, annos) + if err != nil { + return err + } + if catalog == nil { + continue + } + sawCatalog = true + + route := schemaChildRoute{ + Workspace: workspace.DeploymentName, + WorkspaceID: strconv.Itoa(workspace.ID), + WorkspaceName: workspace.Name, + MetastoreOwner: owner, + CatalogOwner: catalog.Owner, + } + trust := securableTrust{ + ownPrincipal: clientID, + metastoreOwner: owner, + parentOwner: catalog.Owner, + fullName: catalogName, + workspace: route.Workspace, + workspaceID: route.WorkspaceID, + } + schemasComplete, err := schemaListingIsComplete(ctx, client, trust, annos) + if err != nil { + return err + } + + schema, err := schemaThrough(ctx, client, route.Workspace, catalogName, schemaName, annos) + if err != nil { + return err + } + if schemasComplete && schema == nil { + ctxzap.Extract(ctx).Debug("databricks-connector: the schema is no longer listed; listing none of its children", + zap.String("resource_type", resourceType.Id), + zap.String("schema", fullName), + zap.String("workspace", route.Workspace)) + + return nil + } + // Owning the schema is enough to list its children. Anything else a + // partial list hides cannot be told from a schema that is gone. + if schema == nil || (!schemasComplete && !ownsPrincipal(clientID, schema.Owner)) { + continue + } + + trust.schemaOwner = schema.Owner + childrenComplete, err := childListingIsComplete(ctx, client, trust, annos) + if err != nil { + return err + } + if !childrenComplete { + continue + } + route.SchemaOwner = schema.Owner + + encoded, err := json.Marshal(route) + if err != nil { + return err + } + bag.Push(pagination.PageState{ResourceTypeID: resourceType.Id, ResourceID: string(encoded)}) + + return nil + } + if !sawCatalog { + return status.Errorf(codes.FailedPrecondition, + "databricks-connector: no workspace in sync scope lists catalog %s of metastore %s", catalogName, metastoreID) + } + + return status.Errorf(codes.PermissionDenied, + "databricks-connector: cannot list every %s in schema %s; %s", noun, fullName, childListRemedy) +} + +// childListingIsComplete is true when ListTables or ListVolumes returns every +// child of the schema. USE_SCHEMA and MANAGE make the schema list complete. +// They do not make the table or volume list complete. +func childListingIsComplete( + ctx context.Context, + client *databricks.Client, + trust securableTrust, + annos *annotations.Annotations, +) (bool, error) { + if ownsPrincipal(trust.ownPrincipal, trust.metastoreOwner) || + ownsPrincipal(trust.ownPrincipal, trust.parentOwner) || + ownsPrincipal(trust.ownPrincipal, trust.schemaOwner) { + return true, nil + } + + if ownerWorkspaceID, ok := catalogAdminWorkspaceID(trust.parentOwner); ok { + admin, err := isWorkspaceAdmin(ctx, client, trust.ownPrincipal, ownerWorkspaceID, annos) + if err != nil || admin { + return admin, err + } + } + + complete, err := holdsPrivilegeOn(ctx, client, trust, databricks.SecurableCatalog, trust.fullName, + []string{privilegeAllPrivileges}, []string{errorCodeCatalogGone}, annos) + if errors.Is(err, errSecurableGone) { + return false, errCatalogGone + } + + return complete, err +} + +// schemaChildListingError reads a workspace that refuses the listing as unknown, +// not as listing nothing: a short list would read as deletions in C1, and the SDK +// reads a NotFound as a securable with no children. +func schemaChildListingError(noun, schemaFullName, workspace string, err error) error { + switch apiStatus(err) { + case http.StatusUnauthorized, http.StatusForbidden, http.StatusNotFound: + return uhttp.WrapErrors(codes.FailedPrecondition, + fmt.Sprintf("databricks-connector: the %ss of schema %s cannot be listed through workspace %s, "+ + "so which of them are in scope is unknown; %s", noun, schemaFullName, workspace, childListRemedy), + err) + } + + return fmt.Errorf("databricks-connector: failed to list the %ss of schema %s through workspace %s: %w", + noun, schemaFullName, workspace, err) +} + +// schemaChildResourceID keys a table or a volume by its metastore and its dotted +// full name, which extends the schema's "{metastore_id}::{catalog}.{schema}". +func schemaChildResourceID(metastoreID, catalogName, schemaName, name string) string { + return metastoreID + securableKeySeparator + + strings.Join([]string{catalogName, schemaName, name}, schemaNameSeparator) +} + +// parseSchemaChildID reads a table or volume resource id, +// "{metastore_id}::{catalog}.{schema}.{name}". +func parseSchemaChildID(resourceID, noun string) (string, string, string, string, error) { + invalid := status.Errorf(codes.InvalidArgument, "databricks-connector: invalid %s resource id %q", noun, resourceID) + + metastoreID, fullName, ok := strings.Cut(resourceID, securableKeySeparator) + if !ok { + return "", "", "", "", invalid + } + + parts := strings.Split(fullName, schemaNameSeparator) + if len(parts) != 3 || metastoreID == "" || parts[0] == "" || parts[1] == "" || parts[2] == "" { + return "", "", "", "", invalid + } + + return metastoreID, parts[0], parts[1], parts[2], nil +} + +// schemaChildProfile is what every table and volume carries, before the level +// adds its own name and type. +func schemaChildProfile(metastoreID, catalogName, schemaName, fullName, owner, securableType string, route schemaChildRoute) map[string]any { + return map[string]any{ + profileKeyMetastoreID: metastoreID, + profileKeyMetastoreOwner: route.MetastoreOwner, + profileKeyFullName: fullName, + profileKeyCatalogName: catalogName, + profileKeyCatalogOwner: route.CatalogOwner, + profileKeySchemaName: schemaName, + profileKeySchemaOwner: route.SchemaOwner, + profileKeyOwner: owner, + profileKeySecurableType: securableType, + profileKeyWorkspace: route.Workspace, + profileKeyWorkspaceID: route.WorkspaceID, + profileKeyWorkspaceName: route.WorkspaceName, + } +} + +// schemaChildGrantsAreTrustworthy is true for an owner of the table or volume, its schema, its catalog, or the metastore, or for a holder of MANAGE on any of those. +func schemaChildGrantsAreTrustworthy( + ctx context.Context, + client *databricks.Client, + trust securableTrust, + securableType string, + annos *annotations.Annotations, +) (bool, error) { + // An owner holds every privilege implicitly, which is not reported as a grant. + if ownsPrincipal(trust.ownPrincipal, trust.metastoreOwner) || + ownsPrincipal(trust.ownPrincipal, trust.parentOwner) || + ownsPrincipal(trust.ownPrincipal, trust.schemaOwner) || + ownsPrincipal(trust.ownPrincipal, trust.owner) { + return true, nil + } + + // The workspace admins group acts as the owner for a direct Admin of its workspace. + if ownerWorkspaceID, ok := catalogAdminWorkspaceID(trust.parentOwner); ok { + admin, err := isWorkspaceAdmin(ctx, client, trust.ownPrincipal, ownerWorkspaceID, annos) + if err != nil || admin { + return admin, err + } + } + + // MANAGE on the securable, its schema or its catalog makes the privilege + // list complete, and one effective read reports all three. + return holdsEffectivePrivilegeOn(ctx, client, trust, securableType, trust.fullName, managePrivilege, annos) +} diff --git a/pkg/connector/securable_helpers.go b/pkg/connector/securable_helpers.go index f9969364..3b5ac930 100644 --- a/pkg/connector/securable_helpers.go +++ b/pkg/connector/securable_helpers.go @@ -39,6 +39,15 @@ type securableOps struct { level securableLevel } +// withLevel returns the ops addressed at another level. A table's privilege set +// follows its table_type, so that level is decided per resource rather than once +// on the builder. +func (o securableOps) withLevel(level securableLevel) securableOps { + o.level = level + + return o +} + type securableLevel struct { // securableType is the endpoint path segment, lowercase. It is not the // uppercase securable_type a listed securable carries. @@ -65,6 +74,16 @@ type securableLevel struct { // of absence: an unbound or unprivileged workspace answers 404 too. goneErrorCodes []string + // ambiguousGoneCodes are codes a missing securable shares with a workspace + // that cannot serve it, so on their own they prove nothing. A level that + // names any must also set confirmGone. + ambiguousGoneCodes []string + + // confirmGone reads the securable itself to settle an ambiguous code. A + // level whose codes are all unambiguous leaves it nil. + confirmGone func(ctx context.Context, client *databricks.Client, fullName, workspace string, + annos *annotations.Annotations) (bool, error) + // trusts reports whether the connector reads every privilege assignment on // the securable. What proves that differs per level, so the rule travels // with the level rather than being decided by the shared path. @@ -83,10 +102,15 @@ type securableTrust struct { // metastoreOwner is empty on a metastore, whose own owner is in owner. metastoreOwner string - // parentOwner is the owner of the securable one level up, which only a - // schema has: the levels above it hang from the metastore directly. + // parentOwner is the owner of the catalog the securable belongs to, which a + // schema, a table and a volume have: the levels above a catalog hang from + // the metastore directly. parentOwner string + // schemaOwner is the owner of the schema the securable belongs to, which + // only a table and a volume have. + schemaOwner string + fullName string workspace string workspaceID string @@ -108,21 +132,23 @@ func ensureGrantsAreTrustworthy( owner, _ := rs.GetProfileStringValue(profile, profileKeyOwner) metastoreOwner, _ := rs.GetProfileStringValue(profile, profileKeyMetastoreOwner) workspaceID, _ := rs.GetProfileStringValue(profile, profileKeyWorkspaceID) - // Only a schema hangs from a securable other than the metastore, so at the - // other levels this key is absent and the owner above is the metastore's. + // A metastore and a catalog hang from the metastore directly, so these keys + // are absent there and the only owner above is the metastore's. parentOwner, _ := rs.GetProfileStringValue(profile, profileKeyCatalogOwner) + schemaOwner, _ := rs.GetProfileStringValue(profile, profileKeySchemaOwner) trustworthy, err := level.trusts(ctx, client, securableTrust{ ownPrincipal: clientID, owner: owner, metastoreOwner: metastoreOwner, parentOwner: parentOwner, + schemaOwner: schemaOwner, fullName: fullName, workspace: workspace, workspaceID: workspaceID, }, annos) if err != nil { - return trustReadError(level, fullName, workspace, err) + return trustReadError(ctx, client, level, fullName, workspace, err, annos) } if !trustworthy { return status.Errorf(codes.PermissionDenied, @@ -136,11 +162,20 @@ func ensureGrantsAreTrustworthy( // trustReadError classifies the 404 a trust read left for the level. A proven // absence reports errSecurableGone; an unproven one is refused, because read as // an empty result ConductorOne deletes everything the securable held. -func trustReadError(level securableLevel, fullName, workspace string, err error) error { +func trustReadError( + ctx context.Context, + client *databricks.Client, + level securableLevel, + fullName, workspace string, + err error, + annos *annotations.Annotations, +) error { if !errors.Is(err, errTrustRead404) { return err } - if databricks.IsSecurableGone(err, level.goneErrorCodes...) { + // A volume names the generic code, which a workspace that cannot serve it + // returns too, so absence is confirmed by reading the securable itself. + if securableIsGone(ctx, client, level, fullName, workspace, err, annos) { return errSecurableGone } @@ -195,10 +230,7 @@ func (o securableOps) syncGrants( assignments, nextPageToken, rateLimit, err := client.ListPermissions(ctx, workspace, level.securableType, fullName, "", pageToken) noteRateLimit(&annos, rateLimit) if err != nil { - // A securable deleted between List and this read is not a failed sync. - // Only a level that names its gone codes can reach this: with none, - // IsSecurableGone is false and every 404 still refuses the read. - if databricks.IsSecurableGone(err, level.goneErrorCodes...) { + if securableIsGone(ctx, client, level, fullName, workspace, err, &annos) { return nil, &rs.SyncOpResults{Annotations: annos}, nil } return nil, &rs.SyncOpResults{Annotations: annos}, refuseUnreadable(fmt.Sprintf( @@ -227,9 +259,6 @@ func (o securableOps) syncGrants( options := groupSecurableGrantOptions(principalId) for _, privilege := range assignment.Privileges { if !slices.Contains(level.privileges, privilege) { - // A privilege Databricks added after this connector shipped is not - // an entitlement, so it cannot be emitted. Say so rather than - // dropping it without a trace. l.Debug("databricks-connector: the securable reports a privilege this connector does not model", zap.String("securable", resourceID), zap.String("privilege", privilege)) @@ -446,10 +475,7 @@ func (o securableOps) revoke( }) noteRateLimit(&annos, rateLimit) if err != nil { - // Databricks names the missing securable in error_code. That is the only - // answer that proves it is gone, so there is nothing left to revoke: a - // bare 404 also comes back from a workspace the securable is not bound to. - if databricks.IsSecurableGone(err, level.goneErrorCodes...) { + if securableIsGone(ctx, client, level, change.fullName, change.workspace, err, &annos) { annos.Update(&v2.GrantAlreadyRevoked{}) return annos, nil @@ -490,3 +516,35 @@ func (o securableOps) revoke( func isSecurableGoneSentinel(err error) bool { return errors.Is(err, errCatalogGone) || errors.Is(err, errMetastoreGone) } + +// securableIsGone reports whether err proves the securable no longer exists. An +// unambiguous code settles it alone. A code a missing securable shares with a +// workspace that cannot serve it is settled by reading the securable, and a read +// that does not answer leaves the absence unproven rather than assuming it. +func securableIsGone( + ctx context.Context, + client *databricks.Client, + level securableLevel, + fullName, workspace string, + err error, + annos *annotations.Annotations, +) bool { + if databricks.IsSecurableGone(err, level.goneErrorCodes...) { + return true + } + if level.confirmGone == nil || !databricks.IsSecurableGone(err, level.ambiguousGoneCodes...) { + return false + } + + gone, confirmErr := level.confirmGone(ctx, client, fullName, workspace, annos) + if confirmErr != nil { + ctxzap.Extract(ctx).Debug("databricks-connector: could not confirm whether the securable is gone", + zap.String("securable", fullName), + zap.String("workspace", workspace), + zap.Error(confirmErr)) + + return false + } + + return gone +} diff --git a/pkg/connector/table.go b/pkg/connector/table.go new file mode 100644 index 00000000..23e0f65a --- /dev/null +++ b/pkg/connector/table.go @@ -0,0 +1,358 @@ +package connector + +import ( + "context" + "fmt" + "slices" + "strings" + + "github.com/conductorone/baton-databricks/pkg/databricks" + v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" + "github.com/conductorone/baton-sdk/pkg/annotations" + "github.com/conductorone/baton-sdk/pkg/connectorbuilder" + ent "github.com/conductorone/baton-sdk/pkg/types/entitlement" + rs "github.com/conductorone/baton-sdk/pkg/types/resource" + "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap/ctxzap" + "go.uber.org/zap" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +var _ connectorbuilder.ResourceSyncerV2 = (*tableBuilder)(nil) + +const ( + securableTypeTable = "TABLE" + + // errorCodeTableGone is what Databricks names in error_code when the table a + // request addresses does not exist. + errorCodeTableGone = "TABLE_DOES_NOT_EXIST" + + // The canonical UPPER_SNAKE form the API answers with. + privilegeAllPrivileges = "ALL_PRIVILEGES" + privilegeRefresh = "REFRESH" + + // A view and a materialized view are both securable_type "table", so + // table_type is the only field that says which privileges apply. + tableTypeManaged = "MANAGED" + tableTypeExternal = "EXTERNAL" + tableTypeManagedShallowClone = "MANAGED_SHALLOW_CLONE" + tableTypeExternalShallowClone = "EXTERNAL_SHALLOW_CLONE" + tableTypeForeign = "FOREIGN" + tableTypeView = "VIEW" + tableTypeMaterializedView = "MATERIALIZED_VIEW" + tableTypeStreamingTable = "STREAMING_TABLE" + tableTypeMetricView = "METRIC_VIEW" + + profileKeyTableName = "table_name" + profileKeyTableType = "table_type" + + // tableWriteRemedy is what a 403 on the PATCH tells the operator. + tableWriteRemedy = "the connector must own the metastore, the parent catalog, the parent schema or the table, " + + "or hold MANAGE on the table, the parent schema or the parent catalog" + + // tableTrustRemedy is what a refusal to report a partial privilege list + // tells the operator. + tableTrustRemedy = "the connector's service principal must own the metastore, the parent catalog, the parent schema or the table, " + + "hold MANAGE on the table, the parent schema or the parent catalog, or be assigned Admin directly on the workspace whose " + + "admins group owns the parent catalog" +) + +// tablePrivileges is the set a real table takes. BROWSE, and everything that +// creates a securable, stop higher up. +// https://docs.databricks.com/aws/en/data-governance/unity-catalog/manage-privileges/privileges +var tablePrivileges = []string{ + privilegeAllPrivileges, + "APPLY_TAG", + "DELETE", + "INSERT", + privilegeManage, + "MODIFY", + "READ_METADATA", + "SELECT", + "UPDATE", +} + +// viewPrivileges matches the privileges reference, ALL_PRIVILEGES included. The +// REST grants API still answers an added ALL_PRIVILEGES on a view with a 400, so +// Grant rejects that write while Grants still reports one SQL already granted. +var viewPrivileges = []string{ + privilegeAllPrivileges, + "APPLY_TAG", + privilegeManage, + "READ_METADATA", + "SELECT", +} + +// A materialized view accepts ALL_PRIVILEGES, and a permissions read answers +// with only it, not the SELECT and REFRESH it implies. +var materializedViewPrivileges = append(slices.Clone(viewPrivileges), privilegeRefresh) + +// A streaming table takes SELECT and REFRESH. MODIFY is reported when a +// principal already holds it and is not offered: the streaming-tables reference +// lists SELECT and REFRESH only, and a grant of MODIFY has not been seen to succeed. +var streamingTablePrivileges = []string{ + privilegeAllPrivileges, + "APPLY_TAG", + privilegeManage, + "MODIFY", + "READ_METADATA", + privilegeRefresh, + "SELECT", +} + +// A foreign table is read-only: its rows live in the federated source, so the +// DML trio and MODIFY cannot be granted on one. +// https://docs.databricks.com/aws/en/data-governance/unity-catalog/access-control/privileges-reference +var foreignTablePrivileges = slices.DeleteFunc(slices.Clone(tablePrivileges), func(privilege string) bool { + return slices.Contains([]string{"DELETE", "INSERT", "MODIFY", "UPDATE"}, privilege) +}) + +// anyTablePrivileges is for a table whose variant is not knowable. REFRESH is in +// it so a held one is not dropped, and Grant still refuses to ask for it: only a +// materialized view or a streaming table accepts that write. +var anyTablePrivileges = append(slices.Clone(tablePrivileges), privilegeRefresh) + +type tableBuilder struct { + securableOps + workspaces map[string]struct{} +} + +func newTableBuilder(client *databricks.Client, clientID string, workspaces map[string]struct{}, sync syncPrincipals) *tableBuilder { + return &tableBuilder{ + securableOps: securableOps{client: client, clientID: clientID, sync: sync, level: tableLevel(privilegesForTableType(""))}, + workspaces: workspaces, + } +} + +func (b *tableBuilder) ResourceType(_ context.Context) *v2.ResourceType { + return tableResourceType +} + +// tablePermissionsName: the permissions endpoint addresses a table by its dotted +// full name. The metastore is carried by the host. +func tablePermissionsName(resourceID string) (string, error) { + _, catalogName, schemaName, name, err := parseSchemaChildID(resourceID, "table") + if err != nil { + return "", err + } + + return strings.Join([]string{catalogName, schemaName, name}, schemaNameSeparator), nil +} + +func tableGrantsAreTrustworthy( + ctx context.Context, + client *databricks.Client, + trust securableTrust, + annos *annotations.Annotations, +) (bool, error) { + return schemaChildGrantsAreTrustworthy(ctx, client, trust, databricks.SecurableTable, annos) +} + +// tableLevel is built per table rather than declared once: table_type decides +// which privileges this one reports and accepts. See privilegesForTableType. +func tableLevel(privileges []string) securableLevel { + return securableLevel{ + securableType: databricks.SecurableTable, + noun: "table", + privileges: privileges, + writeRemedy: tableWriteRemedy, + trustRemedy: tableTrustRemedy, + permissionsName: tablePermissionsName, + trusts: tableGrantsAreTrustworthy, + // Dropping a parent drops the table with it, so any of the three codes + // means there is no table left to revoke on. + goneErrorCodes: []string{errorCodeTableGone, errorCodeSchemaGone, errorCodeCatalogGone}, + } +} + +func tableResource( + table *databricks.Table, + metastoreID, catalogName, schemaName string, + parent *v2.ResourceId, + route schemaChildRoute, +) (*v2.Resource, error) { + fullName := strings.Join([]string{catalogName, schemaName, table.Name}, schemaNameSeparator) + profile := schemaChildProfile(metastoreID, catalogName, schemaName, fullName, table.Owner, securableTypeTable, route) + profile[profileKeyTableName] = table.Name + profile[profileKeyTableType] = table.TableType + + resourceID := schemaChildResourceID(metastoreID, catalogName, schemaName, table.Name) + + // The display name is the full name rather than the leaf: main.sales.orders + // and main.finance.orders would otherwise both read "orders" in C1. + return rs.NewResource( + fullName, + tableResourceType, + resourceID, + rs.WithResourceProfile(profile), + rs.WithParentResourceID(parent), + rs.WithAnnotation(&v2.RawId{Id: resourceID}), + ) +} + +// List pages the tables, views and materialized views of one schema. +func (b *tableBuilder) List(ctx context.Context, parentResourceID *v2.ResourceId, attr rs.SyncOpAttrs) ([]*v2.Resource, *rs.SyncOpResults, error) { + return listSchemaChildren(ctx, b.client, b.clientID, b.workspaces, tableResourceType, "table", parentResourceID, attr, + func(ctx context.Context, route schemaChildRoute, metastoreID, catalogName, schemaName, pageToken string) ( + []*v2.Resource, string, *v2.RateLimitDescription, error) { + tables, next, rateLimit, err := b.client.ListTables(ctx, route.Workspace, catalogName, schemaName, pageToken, ResourcesPageSize) + if err != nil { + return nil, "", rateLimit, err + } + + var rv []*v2.Resource + for _, table := range tables { + if table.Name == "" { + continue + } + + built, err := tableResource(&table, metastoreID, catalogName, schemaName, parentResourceID, route) + if err != nil { + return nil, "", rateLimit, fmt.Errorf( + "databricks-connector: failed to build the resource for table %s.%s.%s: %w", catalogName, schemaName, table.Name, err) + } + rv = append(rv, built) + } + + return rv, next, rateLimit, nil + }) +} + +// isViewLike reports a view or a metric view. Both reject the table DML set, and +// the grants API rejects adding ALL_PRIVILEGES on either. +func isViewLike(tableType string) bool { + switch strings.ToUpper(strings.TrimSpace(tableType)) { + case tableTypeView, tableTypeMetricView: + return true + default: + return false + } +} + +// refreshIsRequestable reports the table types whose grants API accepts REFRESH. +// The privileges reference lists it on a materialized view, and a streaming table +// takes it too. On every other type, including one this connector does not name, +// a held REFRESH is still reported and the grant is not offered. +func refreshIsRequestable(tableType string) bool { + switch strings.ToUpper(strings.TrimSpace(tableType)) { + case tableTypeMaterializedView, tableTypeStreamingTable: + return true + default: + return false + } +} + +// privilegesForTableType returns the set for a known table_type. MANAGED, +// EXTERNAL and both shallow clones stay on the table set. A foreign table is +// read-only. Anything else, including a type Databricks adds later, gets the +// union, so a privilege already held is reported rather than dropped. +func privilegesForTableType(tableType string) []string { + if isViewLike(tableType) { + return viewPrivileges + } + + switch strings.ToUpper(strings.TrimSpace(tableType)) { + case tableTypeMaterializedView: + return materializedViewPrivileges + case tableTypeStreamingTable: + return streamingTablePrivileges + case tableTypeManaged, tableTypeExternal, tableTypeManagedShallowClone, tableTypeExternalShallowClone: + return tablePrivileges + case tableTypeForeign: + return foreignTablePrivileges + default: + return anyTablePrivileges + } +} + +// privilegesForTable is the set this table reports. A missing table_type falls +// back to the union rather than to one variant. +func privilegesForTable(ctx context.Context, resource *v2.Resource) []string { + variant, _ := rs.GetProfileStringValue(rs.GetProfile(resource), profileKeyTableType) + if variant == "" { + ctxzap.Extract(ctx).Debug("databricks-connector: table has no table_type; reporting every variant's privileges", + zap.String("table", resource.GetId().GetResource())) + + return anyTablePrivileges + } + + return privilegesForTableType(variant) +} + +// Entitlements are per table, not static: the privilege set follows table_type, +// and a privilege the grants API refuses must not be offered as requestable. +// ConductorOne would otherwise advertise requests that always fail. +func (b *tableBuilder) Entitlements(ctx context.Context, resource *v2.Resource, _ rs.SyncOpAttrs) ([]*v2.Entitlement, *rs.SyncOpResults, error) { + variant, _ := rs.GetProfileStringValue(rs.GetProfile(resource), profileKeyTableType) + grantable := ent.WithGrantableTo(userResourceType, groupResourceType, servicePrincipalResourceType) + + privileges := privilegesForTable(ctx, resource) + rv := make([]*v2.Entitlement, 0, len(privileges)+1) + for _, privilege := range privileges { + options := []ent.EntitlementOption{ + ent.WithDisplayName(fmt.Sprintf("%s %s", resource.GetDisplayName(), privilege)), + ent.WithDescription(fmt.Sprintf("%s privilege on table %s in Databricks", privilege, resource.GetDisplayName())), + } + // Databricks rejects adding these three, so they are reported where a + // principal already holds them and are never requestable. + reportOnly := (privilege == privilegeAllPrivileges && isViewLike(variant)) || + (privilege == privilegeRefresh && !refreshIsRequestable(variant)) || + (privilege == "MODIFY" && strings.EqualFold(strings.TrimSpace(variant), tableTypeStreamingTable)) + if !reportOnly { + options = append([]ent.EntitlementOption{grantable}, options...) + } + rv = append(rv, ent.NewPermissionEntitlement(resource, privilege, options...)) + } + + rv = append(rv, ent.NewOwnershipEntitlement(resource, ownerEntitlement, + ent.WithDisplayName(fmt.Sprintf("%s owner", resource.GetDisplayName())), + ent.WithDescription(fmt.Sprintf("Owns table %s in Databricks", resource.GetDisplayName())), + )) + + return rv, nil, nil +} + +func (b *tableBuilder) Grants(ctx context.Context, resource *v2.Resource, attr rs.SyncOpAttrs) ([]*v2.Grant, *rs.SyncOpResults, error) { + return b.withLevel(tableLevel(privilegesForTable(ctx, resource))).syncGrants(ctx, resource, attr) +} + +func (b *tableBuilder) Grant(ctx context.Context, principal *v2.Resource, entitlement *v2.Entitlement) (annotations.Annotations, error) { + resource := entitlement.GetResource() + if err := refuseUngrantableTablePrivilege(resource, entitlement); err != nil { + return nil, err + } + + return b.withLevel(tableLevel(privilegesForTable(ctx, resource))).grant(ctx, principal.GetId(), entitlement) +} + +// refuseUngrantableTablePrivilege rejects the two privileges a table reports but +// the grants API refuses to add, before any PATCH goes out. Revoke is not held to +// this: a privilege SQL granted still has to be removable. +func refuseUngrantableTablePrivilege(resource *v2.Resource, entitlement *v2.Entitlement) error { + privilege, err := parseEntitlementPrivilege(entitlement.GetId()) + if err != nil { + return err + } + + variant, _ := rs.GetProfileStringValue(rs.GetProfile(resource), profileKeyTableType) + normalized := strings.ToUpper(strings.TrimSpace(variant)) + switch { + case privilege == privilegeAllPrivileges && isViewLike(variant): + return status.Errorf(codes.InvalidArgument, + "databricks-connector: the grants API rejects ALL_PRIVILEGES on a %s; grant SELECT or MANAGE instead", normalized) + case privilege == privilegeRefresh && !refreshIsRequestable(variant): + return status.Errorf(codes.InvalidArgument, + "databricks-connector: REFRESH can be granted on a materialized view or a streaming table, not on table type %q", normalized) + case privilege == "MODIFY" && normalized == tableTypeStreamingTable: + return status.Errorf(codes.InvalidArgument, + "databricks-connector: MODIFY on a streaming table is reported when already held; the grants API reference does not list it as grantable") + } + + return nil +} + +func (b *tableBuilder) Revoke(ctx context.Context, revoked *v2.Grant) (annotations.Annotations, error) { + resource := revoked.GetEntitlement().GetResource() + + return b.withLevel(tableLevel(privilegesForTable(ctx, resource))).revoke(ctx, revoked.GetPrincipal().GetId(), revoked.GetEntitlement()) +} diff --git a/pkg/connector/volume.go b/pkg/connector/volume.go new file mode 100644 index 00000000..c037218f --- /dev/null +++ b/pkg/connector/volume.go @@ -0,0 +1,208 @@ +package connector + +import ( + "context" + "fmt" + "strings" + + "github.com/conductorone/baton-databricks/pkg/databricks" + v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" + "github.com/conductorone/baton-sdk/pkg/annotations" + "github.com/conductorone/baton-sdk/pkg/connectorbuilder" + ent "github.com/conductorone/baton-sdk/pkg/types/entitlement" + rs "github.com/conductorone/baton-sdk/pkg/types/resource" +) + +var _ connectorbuilder.StaticEntitlementSyncerV2 = (*volumeBuilder)(nil) + +const ( + securableTypeVolume = "VOLUME" + + // errorCodeGenericGone is Databricks' generic 404 code, which a missing + // volume and a workspace that cannot serve one both answer with. + errorCodeGenericGone = "RESOURCE_DOES_NOT_EXIST" + + profileKeyVolumeName = "volume_name" + profileKeyVolumeType = "volume_type" + + // volumeWriteRemedy is what a 403 on the PATCH tells the operator. + volumeWriteRemedy = "the connector must own the metastore, the parent catalog, the parent schema or the volume, " + + "or hold MANAGE on the volume, the parent schema or the parent catalog" + + // volumeTrustRemedy is what a refusal to report a partial privilege list + // tells the operator. + volumeTrustRemedy = "the connector's service principal must own the metastore, the parent catalog, the parent schema or the volume, " + + "hold MANAGE on the volume, the parent schema or the parent catalog, or be assigned Admin directly on the workspace whose " + + "admins group owns the parent catalog" +) + +// volumePrivileges is the set a volume takes: its files are read and written +// rather than queried, so none of the table DML applies. +// https://docs.databricks.com/aws/en/data-governance/unity-catalog/manage-privileges/privileges +var volumePrivileges = []string{ + privilegeAllPrivileges, + "APPLY_TAG", + privilegeManage, + "READ_METADATA", + "READ_VOLUME", + "WRITE_VOLUME", +} + +var volumeLevel = securableLevel{ + securableType: databricks.SecurableVolume, + noun: "volume", + privileges: volumePrivileges, + writeRemedy: volumeWriteRemedy, + trustRemedy: volumeTrustRemedy, + permissionsName: volumePermissionsName, + trusts: volumeGrantsAreTrustworthy, + // Dropping a parent drops the volume with it, and both parent codes name + // what is missing. The generic code does not, so it is confirmed first. + goneErrorCodes: []string{errorCodeSchemaGone, errorCodeCatalogGone}, + ambiguousGoneCodes: []string{errorCodeGenericGone}, + confirmGone: volumeIsGone, +} + +// volumeIsGone reads the volume itself after the permissions endpoint answered +// the generic 404. The volume endpoint answers 200 while the volume exists, so +// anything but a gone code there means the permissions read failed for another +// reason and the absence stays unproven. +func volumeIsGone( + ctx context.Context, + client *databricks.Client, + fullName, workspace string, + annos *annotations.Annotations, +) (bool, error) { + volume, rateLimit, err := client.GetVolume(ctx, workspace, fullName) + noteRateLimit(annos, rateLimit) + if err == nil { + return volume == nil, nil + } + if databricks.IsSecurableGone(err, errorCodeGenericGone, errorCodeSchemaGone, errorCodeCatalogGone) { + return true, nil + } + + return false, err +} + +type volumeBuilder struct { + securableOps + workspaces map[string]struct{} +} + +func newVolumeBuilder(client *databricks.Client, clientID string, workspaces map[string]struct{}, sync syncPrincipals) *volumeBuilder { + return &volumeBuilder{ + securableOps: securableOps{client: client, clientID: clientID, sync: sync, level: volumeLevel}, + workspaces: workspaces, + } +} + +func (b *volumeBuilder) ResourceType(_ context.Context) *v2.ResourceType { + return volumeResourceType +} + +// volumePermissionsName: the permissions endpoint addresses a volume by its +// dotted full name. The metastore is carried by the host. +func volumePermissionsName(resourceID string) (string, error) { + _, catalogName, schemaName, name, err := parseSchemaChildID(resourceID, "volume") + if err != nil { + return "", err + } + + return strings.Join([]string{catalogName, schemaName, name}, schemaNameSeparator), nil +} + +func volumeGrantsAreTrustworthy( + ctx context.Context, + client *databricks.Client, + trust securableTrust, + annos *annotations.Annotations, +) (bool, error) { + return schemaChildGrantsAreTrustworthy(ctx, client, trust, databricks.SecurableVolume, annos) +} + +func volumeResource( + volume *databricks.Volume, + metastoreID, catalogName, schemaName string, + parent *v2.ResourceId, + route schemaChildRoute, +) (*v2.Resource, error) { + fullName := strings.Join([]string{catalogName, schemaName, volume.Name}, schemaNameSeparator) + profile := schemaChildProfile(metastoreID, catalogName, schemaName, fullName, volume.Owner, securableTypeVolume, route) + profile[profileKeyVolumeName] = volume.Name + profile[profileKeyVolumeType] = volume.VolumeType + + resourceID := schemaChildResourceID(metastoreID, catalogName, schemaName, volume.Name) + + // The display name is the full name rather than the leaf: main.sales.files + // and main.finance.files would otherwise both read "files" in C1. + return rs.NewResource( + fullName, + volumeResourceType, + resourceID, + rs.WithResourceProfile(profile), + rs.WithParentResourceID(parent), + rs.WithAnnotation(&v2.RawId{Id: resourceID}), + ) +} + +// List pages the volumes of one schema. +func (b *volumeBuilder) List(ctx context.Context, parentResourceID *v2.ResourceId, attr rs.SyncOpAttrs) ([]*v2.Resource, *rs.SyncOpResults, error) { + return listSchemaChildren(ctx, b.client, b.clientID, b.workspaces, volumeResourceType, "volume", parentResourceID, attr, + func(ctx context.Context, route schemaChildRoute, metastoreID, catalogName, schemaName, pageToken string) ( + []*v2.Resource, string, *v2.RateLimitDescription, error) { + volumes, next, rateLimit, err := b.client.ListVolumes(ctx, route.Workspace, catalogName, schemaName, pageToken, ResourcesPageSize) + if err != nil { + return nil, "", rateLimit, err + } + + var rv []*v2.Resource + for _, volume := range volumes { + if volume.Name == "" { + continue + } + + built, err := volumeResource(&volume, metastoreID, catalogName, schemaName, parentResourceID, route) + if err != nil { + return nil, "", rateLimit, fmt.Errorf( + "databricks-connector: failed to build the resource for volume %s.%s.%s: %w", catalogName, schemaName, volume.Name, err) + } + rv = append(rv, built) + } + + return rv, next, rateLimit, nil + }) +} + +func (b *volumeBuilder) StaticEntitlements(_ context.Context, _ rs.SyncOpAttrs) ([]*v2.Entitlement, *rs.SyncOpResults, error) { + grantable := ent.WithGrantableTo(userResourceType, groupResourceType, servicePrincipalResourceType) + rv := make([]*v2.Entitlement, 0, len(volumePrivileges)+1) + for _, privilege := range volumePrivileges { + rv = append(rv, ent.NewPermissionEntitlement(nil, privilege, + grantable, + ent.WithDescription(fmt.Sprintf("%s privilege on a volume in Databricks", privilege)), + )) + } + + rv = append(rv, ent.NewOwnershipEntitlement(nil, ownerEntitlement, + ent.WithDescription("Owns a volume in Databricks"), + )) + + return rv, nil, nil +} + +func (b *volumeBuilder) Entitlements(context.Context, *v2.Resource, rs.SyncOpAttrs) ([]*v2.Entitlement, *rs.SyncOpResults, error) { + return nil, nil, nil +} + +func (b *volumeBuilder) Grants(ctx context.Context, resource *v2.Resource, attr rs.SyncOpAttrs) ([]*v2.Grant, *rs.SyncOpResults, error) { + return b.syncGrants(ctx, resource, attr) +} + +func (b *volumeBuilder) Grant(ctx context.Context, principal *v2.Resource, entitlement *v2.Entitlement) (annotations.Annotations, error) { + return b.grant(ctx, principal.GetId(), entitlement) +} + +func (b *volumeBuilder) Revoke(ctx context.Context, revoked *v2.Grant) (annotations.Annotations, error) { + return b.revoke(ctx, revoked.GetPrincipal().GetId(), revoked.GetEntitlement()) +} diff --git a/pkg/databricks/models.go b/pkg/databricks/models.go index e20a9c07..e7ab5004 100644 --- a/pkg/databricks/models.go +++ b/pkg/databricks/models.go @@ -190,6 +190,18 @@ type Schema struct { Owner string `json:"owner"` } +type Table struct { + Name string `json:"name"` + Owner string `json:"owner"` + TableType string `json:"table_type"` +} + +type Volume struct { + Name string `json:"name"` + Owner string `json:"owner"` + VolumeType string `json:"volume_type"` +} + type PrivilegeAssignment struct { Principal string `json:"principal"` PrincipalID json.Number `json:"principal_id"` @@ -236,6 +248,16 @@ type schemasResponse struct { NextPageToken string `json:"next_page_token"` } +type tablesResponse struct { + Tables []Table `json:"tables"` + NextPageToken string `json:"next_page_token"` +} + +type volumesResponse struct { + Volumes []Volume `json:"volumes"` + NextPageToken string `json:"next_page_token"` +} + type permissionsResponse struct { PrivilegeAssignments []PrivilegeAssignment `json:"privilege_assignments"` NextPageToken string `json:"next_page_token"` diff --git a/pkg/databricks/unity_catalog.go b/pkg/databricks/unity_catalog.go index 339a8737..5340c63c 100644 --- a/pkg/databricks/unity_catalog.go +++ b/pkg/databricks/unity_catalog.go @@ -26,6 +26,12 @@ const ( // https://docs.databricks.com/api/workspace/schemas/list unityCatalogSchemasEndpoint = "/api/2.1/unity-catalog/schemas" + // https://docs.databricks.com/api/workspace/tables/list + unityCatalogTablesEndpoint = "/api/2.1/unity-catalog/tables" + + // https://docs.databricks.com/api/workspace/volumes/list + unityCatalogVolumesEndpoint = "/api/2.1/unity-catalog/volumes" + // GET https://docs.databricks.com/api/workspace/grants/get // PATCH https://docs.databricks.com/api/workspace/grants/update unityCatalogPermissionsEndpoint = "/api/2.1/unity-catalog/permissions" @@ -41,10 +47,15 @@ const ( SecurableMetastore = "metastore" SecurableCatalog = "catalog" SecurableSchema = "schema" + SecurableTable = "table" + SecurableVolume = "volume" // The permissions endpoint rejects any max_results from 1 to 149 with a 400. // Zero selects the paginated response that replaces the unpaginated form. permissionsMaxResults uint = 0 + + // The tables endpoint rejects a max_results above this ceiling. + unityCatalogTablesMaxResults uint = 50 ) // ListMetastores (GET /api/2.0/accounts/{account_id}/metastores). Not paginated. @@ -124,6 +135,32 @@ func (c *Client) ListCatalogs( return res.Catalogs, res.NextPageToken, ratelimitData, nil } +// GetVolume (GET /api/2.1/unity-catalog/volumes/{full_name}) reads one volume +// through workspaceId, which is a deployment name. The permissions endpoint +// answers a missing volume with the generic RESOURCE_DOES_NOT_EXIST, which an +// unreachable workspace also answers, so this read is what separates the two: +// measured on a live tenant it returns 200 for a volume that exists and 404 for +// one that does not, and it names SCHEMA_DOES_NOT_EXIST when the parent is what +// is missing. +func (c *Client) GetVolume( + ctx context.Context, + workspaceId, fullName string, +) ( + *Volume, + *v2.RateLimitDescription, + error, +) { + var res Volume + + u := c.workspaceUrl(workspaceId).JoinPath(unityCatalogVolumesEndpoint, fullName) + ratelimitData, err := c.Get(ctx, u, &res) + if err != nil { + return nil, ratelimitData, nameWorkspace403Remedy(workspaceId, fmt.Errorf("failed to read volume %s: %w", fullName, err)) + } + + return &res, ratelimitData, nil +} + // GetCatalog (GET /api/2.1/unity-catalog/catalogs/{name}) reads one catalog // through workspaceId, which is a deployment name. It answers what draining // ListCatalogs cannot: a 404 naming CATALOG_DOES_NOT_EXIST is the catalog not @@ -192,6 +229,71 @@ func (c *Client) ListSchemas( return res.Schemas, res.NextPageToken, ratelimitData, nil } +// ListTables (GET /api/2.1/unity-catalog/tables?catalog_name=&schema_name=) pages +// the tables, views and materialized views in a schema. schema_name is mandatory, +// and a max_results above the endpoint's ceiling of 50 is clamped rather than sent. +func (c *Client) ListTables( + ctx context.Context, + workspaceId, catalogName, schemaName, pageToken string, + maxResults uint, +) ( + []Table, + string, + *v2.RateLimitDescription, + error, +) { + if maxResults > unityCatalogTablesMaxResults { + maxResults = unityCatalogTablesMaxResults + } + + var res tablesResponse + + u := c.workspaceUrl(workspaceId).JoinPath(unityCatalogTablesEndpoint) + ratelimitData, err := c.Get(ctx, u, &res, &unityCatalogVars{ + catalogName: catalogName, + schemaName: schemaName, + maxResults: &maxResults, + pageToken: pageToken, + // Without this the listing carries every column and property of every + // table, which is the bulk of the response and none of it is read here. + omitTablePayload: true, + }) + if err != nil { + return nil, "", ratelimitData, nameWorkspace403Remedy(workspaceId, + fmt.Errorf("failed to list tables in schema %s.%s: %w", catalogName, schemaName, err)) + } + + return res.Tables, res.NextPageToken, ratelimitData, nil +} + +// ListVolumes (GET /api/2.1/unity-catalog/volumes?catalog_name=&schema_name=) pages a schema's volumes. +func (c *Client) ListVolumes( + ctx context.Context, + workspaceId, catalogName, schemaName, pageToken string, + maxResults uint, +) ( + []Volume, + string, + *v2.RateLimitDescription, + error, +) { + var res volumesResponse + + u := c.workspaceUrl(workspaceId).JoinPath(unityCatalogVolumesEndpoint) + ratelimitData, err := c.Get(ctx, u, &res, &unityCatalogVars{ + catalogName: catalogName, + schemaName: schemaName, + maxResults: &maxResults, + pageToken: pageToken, + }) + if err != nil { + return nil, "", ratelimitData, nameWorkspace403Remedy(workspaceId, + fmt.Errorf("failed to list volumes in schema %s.%s: %w", catalogName, schemaName, err)) + } + + return res.Volumes, res.NextPageToken, ratelimitData, nil +} + // ListPermissions (GET /api/2.1/unity-catalog/permissions/{securable_type}/{full_name}). // The principal query filters effectively, not literally: an inherited assignment // comes back naming the group it is inherited through, so a caller must read every diff --git a/pkg/databricks/unity_catalog_test.go b/pkg/databricks/unity_catalog_test.go index 856771a3..cad0882e 100644 --- a/pkg/databricks/unity_catalog_test.go +++ b/pkg/databricks/unity_catalog_test.go @@ -6,6 +6,9 @@ import ( "io" "net/http" "net/http/httptest" + "net/url" + "reflect" + "strconv" "testing" v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" @@ -95,6 +98,43 @@ func TestUpdatePermissionsDrainsFromPatchToken(t *testing.T) { } } +// omit_username is deliberately absent: it also drops owner, and the drop is +// silent — the response still parses and every owner simply becomes empty. The +// exact-set assertion is what makes adding it fail here. +func TestListTablesSendsOnlyTheOmitColumnsAndPropertiesPayloadVars(t *testing.T) { + t.Setenv("BATON_HTTP_CACHE_TTL", "0") + + var query url.Values + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + query = r.URL.Query() + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(tablesResponse{Tables: []Table{{Name: "t"}}}) + })) + t.Cleanup(srv.Close) + + client := srv.Client() + client.Transport = rewriteHost(srv.Listener.Addr().String(), client.Transport) + c, err := NewClient(context.Background(), client, "example.cloud.databricks.com", "accounts.cloud.databricks.com", "acc-1", srv.URL, &NoAuth{}, nil) + if err != nil { + t.Fatalf("NewClient: %v", err) + } + + if _, _, _, err := c.ListTables(context.Background(), "ws", "main", "default", "", unityCatalogTablesMaxResults); err != nil { + t.Fatalf("ListTables: %v", err) + } + + want := url.Values{ + "catalog_name": []string{"main"}, + "schema_name": []string{"default"}, + "max_results": []string{strconv.FormatUint(uint64(unityCatalogTablesMaxResults), 10)}, + "omit_columns": []string{"true"}, + "omit_properties": []string{"true"}, + } + if !reflect.DeepEqual(query, want) { + t.Fatalf("query = %v, want exactly %v", query, want) + } +} + func TestListCatalogsCanIncludeUnboundCatalogs(t *testing.T) { t.Setenv("BATON_HTTP_CACHE_TTL", "0") var includeUnbound string diff --git a/pkg/databricks/vars.go b/pkg/databricks/vars.go index 6fac93ad..04634dcd 100644 --- a/pkg/databricks/vars.go +++ b/pkg/databricks/vars.go @@ -174,10 +174,15 @@ func NewNameVars(name string, etag string) *NameVars { // maxResults is a pointer: its presence is the signal. type unityCatalogVars struct { catalogName string + schemaName string principal string maxResults *uint pageToken string includeUnbound bool + + // omit_username is deliberately not sent alongside these: it also removes + // owner, which every securable's profile and its owner grant depend on. + omitTablePayload bool } func (v *unityCatalogVars) Apply(params *url.Values) { @@ -185,6 +190,10 @@ func (v *unityCatalogVars) Apply(params *url.Values) { params.Add("catalog_name", v.catalogName) } + if v.schemaName != "" { + params.Add("schema_name", v.schemaName) + } + if v.principal != "" { params.Add("principal", v.principal) } @@ -199,4 +208,9 @@ func (v *unityCatalogVars) Apply(params *url.Values) { if v.includeUnbound { params.Set("include_unbound", "true") } + + if v.omitTablePayload { + params.Add("omit_columns", "true") + params.Add("omit_properties", "true") + } } From 605ac6ab6ae9e0605102731b756b0477650716d3 Mon Sep 17 00:00:00 2001 From: Mateo Hernandez Date: Wed, 4 Nov 2026 16:15:02 -0300 Subject: [PATCH 2/4] fix(databricks): read one schema instead of every schema of the catalog Listing a schema's tables or volumes resolved its route on the first page, once per child type, and schemaThrough found the schema by paging the whole catalog's schema listing. That is one full listing walk per schema per child type: a catalog with 1,000 schemas spends about 21,000 listing requests, and 43,000 requests in all, before the first table is listed. The uhttp GET cache hid it while a catalog's listing fit in the 5 MB it holds by default, which a catalog of 30 KB schema rows does not. schemaThrough now reads the one schema it needs. The same catalog costs 2,000 schema reads instead, and the cost is linear in the schema count rather than quadratic. Error classification is unchanged. A 404 naming SCHEMA_DOES_NOT_EXIST is read exactly like a schema absent from the listing was: proof of deletion only for a caller that sees the whole catalog, since Databricks answers the same way for a schema the caller merely cannot see. Every other refusal, including a 404 carrying no code, still leaves the schema unknown rather than absent -- read as an empty result the SDK would skip the children and ConductorOne would delete every table and volume under the schema. Co-Authored-By: Claude Opus 5 --- pkg/connector/schema.go | 49 ++++--- pkg/connector/schema_child.go | 7 +- pkg/connector/schema_child_test.go | 192 +++++++++++++++++++++++++++ pkg/connector/schema_test.go | 16 +++ pkg/databricks/models.go | 3 + pkg/databricks/unity_catalog.go | 42 ++++++ pkg/databricks/unity_catalog_test.go | 57 ++++++++ 7 files changed, 339 insertions(+), 27 deletions(-) create mode 100644 pkg/connector/schema_child_test.go diff --git a/pkg/connector/schema.go b/pkg/connector/schema.go index cbed4df6..b056560f 100644 --- a/pkg/connector/schema.go +++ b/pkg/connector/schema.go @@ -364,39 +364,36 @@ func (b *schemaBuilder) pushRoute( catalogName, schemaListRemedy) } -// schemaThrough finds one schema in the listing of workspace, nil when that -// workspace does not list it. Its caller has already proven the listing holds -// every schema in the catalog, so an absence here is the schema being gone. +// errSchemaGone marks a schema Databricks named as missing. +var errSchemaGone = errors.New("schema no longer exists") + +// schemaThrough reads one schema through workspace. It reports errSchemaGone +// when Databricks names the schema missing, which proves nothing on its own: a +// caller without catalog-wide visibility is answered the same way for a schema +// it merely cannot see, so the caller decides what the absence means. Every +// other refusal, a 404 carrying no code included, leaves the schema unknown +// rather than absent: read as an empty result the SDK would skip the children +// and ConductorOne would delete every table and volume under it. func schemaThrough( ctx context.Context, client *databricks.Client, workspace, catalogName, schemaName string, annos *annotations.Annotations, ) (*databricks.Schema, error) { - pageToken := "" - for { - schemas, next, rateLimit, err := client.ListSchemas(ctx, workspace, catalogName, pageToken, ResourcesPageSize) - noteRateLimit(annos, rateLimit) - if err != nil { - return nil, schemaListingError(catalogName, workspace, err) - } - - for _, schema := range schemas { - if schema.Name == schemaName { - return &schema, nil - } - } - - if next == "" { - return nil, nil - } - if next == pageToken { - return nil, status.Errorf(codes.Internal, - "databricks-connector: the schemas of catalog %s repeat page token %q through workspace %s", - catalogName, next, workspace) - } - pageToken = next + schema, rateLimit, err := client.GetSchema(ctx, workspace, catalogName+schemaNameSeparator+schemaName) + noteRateLimit(annos, rateLimit) + switch { + case err == nil: + case databricks.IsSecurableGone(err, errorCodeSchemaGone): + return nil, errSchemaGone + case status.Code(err) == codes.InvalidArgument: + // The answer does not describe the schema asked for. + return nil, err + default: + return nil, schemaListingError(catalogName, workspace, err) } + + return schema, nil } // schemaListingError returns NotFound only when Databricks names the missing diff --git a/pkg/connector/schema_child.go b/pkg/connector/schema_child.go index 8ecfecfa..cc386149 100644 --- a/pkg/connector/schema_child.go +++ b/pkg/connector/schema_child.go @@ -192,11 +192,16 @@ func pushSchemaChildRoute( } schema, err := schemaThrough(ctx, client, route.Workspace, catalogName, schemaName, annos) + // A schema named as missing is read like one absent from the listing: + // proof of deletion only for a caller that sees the whole catalog. + if errors.Is(err, errSchemaGone) { + schema, err = nil, nil + } if err != nil { return err } if schemasComplete && schema == nil { - ctxzap.Extract(ctx).Debug("databricks-connector: the schema is no longer listed; listing none of its children", + ctxzap.Extract(ctx).Debug("databricks-connector: the schema is no longer there; listing none of its children", zap.String("resource_type", resourceType.Id), zap.String("schema", fullName), zap.String("workspace", route.Workspace)) diff --git a/pkg/connector/schema_child_test.go b/pkg/connector/schema_child_test.go new file mode 100644 index 00000000..613a8d48 --- /dev/null +++ b/pkg/connector/schema_child_test.go @@ -0,0 +1,192 @@ +package connector + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "slices" + "testing" + + v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" + "github.com/conductorone/baton-sdk/pkg/pagination" + rs "github.com/conductorone/baton-sdk/pkg/types/resource" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// serveSchema answers GET /schemas/{full_name}. +func (f *fakeSchemaAccount) serveSchema(w http.ResponseWriter, r *http.Request) { + workspace := r.Header.Get("X-Test-Workspace") + f.schemaReads = append(f.schemaReads, workspace) + if code, ok := f.schemaGetStatus[workspace]; ok { + body := `{"message":"refused"}` + if f.schemaGetErrorCode != "" { + body = fmt.Sprintf(`{"error_code":%q,"message":"refused"}`, f.schemaGetErrorCode) + } + http.Error(w, body, code) + + return + } + + owner := "someone" + if len(f.schemas) > 0 { + owner, _ = f.schemas[0]["owner"].(string) + } + _ = json.NewEncoder(w).Encode(map[string]any{"name": "sales", "full_name": "main.sales", "owner": owner}) +} + +func (f *fakeSchemaAccount) serveSchemaChildren(w http.ResponseWriter, r *http.Request, key string, rows []map[string]any) { + f.childReads = append(f.childReads, key) + if got := r.URL.Query().Get("schema_name"); got == "" { + http.Error(w, `{"message":"schema_name is required"}`, http.StatusBadRequest) + + return + } + _ = json.NewEncoder(w).Encode(map[string]any{key: rows}) +} + +var testSchemaResourceID = &v2.ResourceId{ResourceType: schemaResourceType.Id, Resource: "ms-1::main.sales"} + +func newTestChildBuildersAs(t *testing.T, f *fakeSchemaAccount, clientID string) (*tableBuilder, *volumeBuilder) { + t.Helper() + t.Setenv("BATON_HTTP_CACHE_TTL", "0") + if f.metastoreOwner == "" { + f.metastoreOwner = testOwnPrincipal + } + if f.assignments == nil { + f.assignments = map[string][]map[string]any{} + } + client, _ := newWorkspaceRoutedClient(t, f.handler(t)) + + return newTableBuilder(client, clientID, nil, syncPrincipals{groups: true}), + newVolumeBuilder(client, clientID, nil, syncPrincipals{groups: true}) +} + +func listAllChildren(t *testing.T, + list func(context.Context, *v2.ResourceId, rs.SyncOpAttrs) ([]*v2.Resource, *rs.SyncOpResults, error), +) ([]*v2.Resource, error) { + t.Helper() + + var all []*v2.Resource + token := "" + for range 1000 { + resources, results, err := list(context.Background(), testSchemaResourceID, rs.SyncOpAttrs{PageToken: pagination.Token{Token: token}}) + if err != nil { + return all, err + } + all = append(all, resources...) + if results == nil || results.NextPageToken == "" { + return all, nil + } + token = results.NextPageToken + } + t.Fatal("List never returned an empty page token") + + return nil, nil +} + +// A schema's children are listed after reading that one schema, not after +// walking the catalog's whole schema listing: the walk runs once per schema and +// again per child type, which is quadratic in the catalog's schema count. +func TestSchemaChildListReadsOneSchemaInsteadOfDrainingTheCatalog(t *testing.T) { + // sales sits last, so walking the listing pays for every page of the catalog. + manySchemas := make([]map[string]any, 0, 300) + for i := range 299 { + manySchemas = append(manySchemas, map[string]any{"name": fmt.Sprintf("s%03d", i), "owner": "eng"}) + } + manySchemas = append(manySchemas, map[string]any{"name": "sales", "owner": "eng"}) + + f := &fakeSchemaAccount{ + metastoreOwner: "eng", + catalogOwner: "eng", + schemas: manySchemas, + tables: []map[string]any{{"name": "orders", "owner": "eng", "table_type": "MANAGED"}}, + volumes: []map[string]any{{"name": "files", "owner": "eng", "volume_type": "MANAGED"}}, + } + tables, volumes := newTestChildBuildersAs(t, f, "eng") + + listedTables, err := listAllChildren(t, tables.List) + if err != nil { + t.Fatalf("table List: %v", err) + } + listedVolumes, err := listAllChildren(t, volumes.List) + if err != nil { + t.Fatalf("volume List: %v", err) + } + + if len(listedTables) != 1 || len(listedVolumes) != 1 { + t.Fatalf("listed %d table(s) and %d volume(s), want 1 of each", len(listedTables), len(listedVolumes)) + } + if f.schemaPages != 0 { + t.Errorf("the schemas listing was paged %d time(s); listing a schema's children must not walk the catalog's schemas", f.schemaPages) + } + if want := []string{"wsa", "wsa"}; !slices.Equal(f.schemaReads, want) { + t.Errorf("single-schema reads = %v, want %v: one per child type", f.schemaReads, want) + } +} + +// A schema named as missing is proof of deletion only for a caller that sees the +// whole catalog. Reading the other case as "no children" would make ConductorOne +// delete every table and volume under it. +func TestSchemaChildListTellsADeletedSchemaFromAnInvisibleOne(t *testing.T) { + tests := []struct { + name string + clientID string + catalogOwner string + errorCode string + wantErr codes.Code + }{ + { + name: "the connector owns the catalog, so the schema is gone", + clientID: "eng", catalogOwner: "eng", errorCode: errorCodeSchemaGone, + }, + { + name: "the connector cannot see the whole catalog, so the sync fails", + clientID: "app-9", catalogOwner: "someone", errorCode: errorCodeSchemaGone, + wantErr: codes.PermissionDenied, + }, + { + // A 404 Databricks does not put a code on says nothing about the + // schema. Read as an empty result it would delete every table and + // volume under it, so the sync fails instead. + name: "a 404 naming nothing is not proof the schema is gone", + clientID: "eng", catalogOwner: "eng", + wantErr: codes.FailedPrecondition, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + f := &fakeSchemaAccount{ + metastoreOwner: "someone", + catalogOwner: test.catalogOwner, + schemas: []map[string]any{{"name": "sales", "owner": "someone"}}, + schemaGetStatus: map[string]int{"wsa": http.StatusNotFound}, + schemaGetErrorCode: test.errorCode, + } + tables, _ := newTestChildBuildersAs(t, f, test.clientID) + + listed, err := listAllChildren(t, tables.List) + if test.wantErr != codes.OK { + if status.Code(err) != test.wantErr { + t.Fatalf("List error = %v, want %s", err, test.wantErr) + } + + return + } + if err != nil { + t.Fatalf("List: %v", err) + } + if len(listed) != 0 { + t.Errorf("listed %d table(s), want none under a schema that is gone", len(listed)) + } + // The empty result must come from the schema read proving the + // absence, not from an earlier silent exit. + if len(f.schemaReads) == 0 || len(f.childReads) != 0 { + t.Errorf("schema reads = %v, child listings = %v: want the empty result to come from the schema read", + f.schemaReads, f.childReads) + } + }) + } +} diff --git a/pkg/connector/schema_test.go b/pkg/connector/schema_test.go index 84bc5ca5..e7c91c0b 100644 --- a/pkg/connector/schema_test.go +++ b/pkg/connector/schema_test.go @@ -33,6 +33,14 @@ type fakeSchemaAccount struct { schemas []map[string]any // schemaStatus fails the schemas listing with that status. schemaStatus int + // schemaGetStatus fails the single-schema read on that deployment name, with + // schemaGetErrorCode as the vendor error_code. SCHEMA_DOES_NOT_EXIST is what + // both a deleted schema and one the caller cannot see answer. + schemaGetStatus map[string]int + schemaGetErrorCode string + // tables and volumes are what a schema's children listings return. + tables []map[string]any + volumes []map[string]any // assignments maps a securable's full name to the privilege assignments on it. assignments map[string][]map[string]any // ownPrivileges answers a read filtered by principal, keyed by @@ -58,6 +66,8 @@ type fakeSchemaAccount struct { routed int schemaPages int + schemaReads []string + childReads []string catalogPages int patches []map[string]any permHosts []string @@ -114,6 +124,12 @@ func (f *fakeSchemaAccount) handler(t *testing.T) http.HandlerFunc { f.serveCatalog(w, r) case strings.HasSuffix(path, "/unity-catalog/schemas"): f.serveSchemas(w, r) + case strings.Contains(path, "/unity-catalog/schemas/"): + f.serveSchema(w, r) + case strings.HasSuffix(path, "/unity-catalog/tables"): + f.serveSchemaChildren(w, r, "tables", f.tables) + case strings.HasSuffix(path, "/unity-catalog/volumes"): + f.serveSchemaChildren(w, r, "volumes", f.volumes) case strings.Contains(path, "/unity-catalog/effective-permissions/"): _, securable, _ := strings.Cut(path, "/unity-catalog/effective-permissions/") f.serveEffectivePermissions(w, r, securable) diff --git a/pkg/databricks/models.go b/pkg/databricks/models.go index e7ab5004..8c801289 100644 --- a/pkg/databricks/models.go +++ b/pkg/databricks/models.go @@ -188,6 +188,9 @@ type Catalog struct { type Schema struct { Name string `json:"name"` Owner string `json:"owner"` + + // FullName is "{catalog}.{schema}", which a single schema read is checked against. + FullName string `json:"full_name"` } type Table struct { diff --git a/pkg/databricks/unity_catalog.go b/pkg/databricks/unity_catalog.go index 5340c63c..bde9d0c1 100644 --- a/pkg/databricks/unity_catalog.go +++ b/pkg/databricks/unity_catalog.go @@ -200,6 +200,48 @@ func (c *Client) GetCatalog( return &res, ratelimitData, nil } +// GetSchema (GET /api/2.1/unity-catalog/schemas/{full_name}) reads one schema +// through workspaceId, which is a deployment name. full_name is "{catalog}.{schema}". +// It separates the two answers draining ListSchemas runs together: a 404 naming +// SCHEMA_DOES_NOT_EXIST is Databricks naming the schema missing, which a caller +// that merely cannot see it is answered with too, and any other refusal is this +// workspace not serving it. Which of the two a named 404 is belongs to the +// caller. The name is checked against the assembled URL for the same reason +// permissionsUrl checks it. +func (c *Client) GetSchema( + ctx context.Context, + workspaceId, fullName string, +) ( + *Schema, + *v2.RateLimitDescription, + error, +) { + var res Schema + + if fullName == "" { + return nil, nil, invalidArgument("schema full name is required") + } + + u := c.workspaceUrl(workspaceId).JoinPath(unityCatalogSchemasEndpoint, fullName) + if !addressesSecurable(u, "/"+fullName) { + return nil, nil, invalidArgument("cannot address schema %q: the name does not survive URL assembly", fullName) + } + + ratelimitData, err := c.Get(ctx, u, &res) + if err != nil { + return nil, ratelimitData, nameWorkspace403Remedy(workspaceId, fmt.Errorf("failed to read schema %s: %w", fullName, err)) + } + + // An answer about another schema, or one that names no schema at all, does not + // describe the one asked for, and its owner is what decides whether the + // children listing is complete. + if res.FullName != fullName { + return nil, ratelimitData, invalidArgument("asked Databricks for schema %q and it answered with %q", fullName, res.FullName) + } + + return &res, ratelimitData, nil +} + // ListSchemas (GET /api/2.1/unity-catalog/schemas?catalog_name=) reads one page of // the schemas of catalogName, through workspaceId, which is a deployment name. // Unlike the catalogs listing, this one pages, so the cursor is passed through. diff --git a/pkg/databricks/unity_catalog_test.go b/pkg/databricks/unity_catalog_test.go index cad0882e..a0783b18 100644 --- a/pkg/databricks/unity_catalog_test.go +++ b/pkg/databricks/unity_catalog_test.go @@ -220,6 +220,63 @@ func TestGetCatalogRefusesAMismatchedAnswer(t *testing.T) { } } +// A schema read addresses one securable by its dotted full name. The same URL +// check GetCatalog makes applies: a name holding "?", "#" or "%" would address +// a different schema, and its owner is what decides whether the connector can +// list every table and volume under it. +func TestGetSchemaRefusesANameThatChangesTheURLOrAMismatchedAnswer(t *testing.T) { + t.Setenv("BATON_HTTP_CACHE_TTL", "0") + + var requested []string + var answer any = Schema{Name: "other", FullName: "main.other", Owner: "eng"} + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requested = append(requested, r.URL.Path) + _ = json.NewEncoder(w).Encode(answer) + })) + t.Cleanup(srv.Close) + + client := srv.Client() + client.Transport = rewriteHost(srv.Listener.Addr().String(), client.Transport) + c, err := NewClient(context.Background(), client, "example.cloud.databricks.com", "accounts.cloud.databricks.com", "acc-1", srv.URL, &NoAuth{}, nil) + if err != nil { + t.Fatalf("NewClient: %v", err) + } + + for _, fullName := range []string{"main.sales?x", "main.sales#x", "main.sales%x", ""} { + got, _, err := c.GetSchema(context.Background(), "ws", fullName) + if err == nil { + t.Fatalf("GetSchema(%q) returned schema %+v, want an error", fullName, got) + } + if status.Code(err) != codes.InvalidArgument { + t.Fatalf("GetSchema(%q) returned %v, want InvalidArgument", fullName, status.Code(err)) + } + } + if len(requested) != 0 { + t.Fatalf("GetSchema sent %v, want no request at all", requested) + } + + // An answer about another schema does not describe the one asked for. Neither + // does one that names no schema at all: its owner would still be read as this + // schema's, and that owner decides whether the children listing is complete. + for _, body := range []any{ + Schema{Name: "other", FullName: "main.other", Owner: "eng"}, + map[string]any{"name": "sales", "owner": "eng"}, + } { + answer = body + requested = nil + got, _, err := c.GetSchema(context.Background(), "ws", "main.sales") + if err == nil { + t.Fatalf("GetSchema returned schema %+v for answer %+v, want an error", got, body) + } + if status.Code(err) != codes.InvalidArgument { + t.Fatalf("GetSchema returned %v for answer %+v, want InvalidArgument", status.Code(err), body) + } + if want := []string{"/api/2.1/unity-catalog/schemas/main.sales"}; !reflect.DeepEqual(requested, want) { + t.Fatalf("GetSchema requested %v, want %v", requested, want) + } + } +} + type hostRewriter struct { host http.RoundTripper addr string From d5596f4e1b42e57b1f7848718328feb8b44b7be9 Mon Sep 17 00:00:00 2001 From: Mateo Hernandez Date: Wed, 4 Nov 2026 16:28:27 -0300 Subject: [PATCH 3/4] fix(databricks): walk on when a workspace does not serve the catalog Listing a schema's tables or volumes returned the catalog read's error for every refusal, where the schema syncer walks on to the next workspace. A catalog bound to one workspace of several is answered by the others with a 404, and which of the two answers Databricks gives decides which way this breaks. A 404 carrying no error code reached the SDK as codes.NotFound, which it reads as a warning: the listing finished green with no resources and ConductorOne deleted every table and volume under the schema. A 404 naming CATALOG_DOES_NOT_EXIST raised errCatalogGone, which failed the whole sync. Both now continue the walk, matching the schema syncer. A catalog every workspace names as gone, with none refusing the read, empties the listing: the catalog syncer drops it in the same sync. A refusal still fails the sync and keeps what ConductorOne already has. Co-Authored-By: Claude Opus 5 --- pkg/connector/schema_child.go | 33 +++++++++++++-- pkg/connector/schema_child_test.go | 66 ++++++++++++++++++++++++++++++ 2 files changed, 96 insertions(+), 3 deletions(-) diff --git a/pkg/connector/schema_child.go b/pkg/connector/schema_child.go index cc386149..f968ccfc 100644 --- a/pkg/connector/schema_child.go +++ b/pkg/connector/schema_child.go @@ -159,11 +159,28 @@ func pushSchemaChildRoute( } fullName := catalogName + schemaNameSeparator + schemaName - var sawCatalog bool + var sawCatalog, sawGone bool + var lastDeny error for i := range usable { workspace := usable[i] catalog, err := catalogThrough(ctx, client, workspace.DeploymentName, metastoreID, catalogName, annos) if err != nil { + // A workspace that does not serve the catalog says nothing about the + // schema below it, so the walk goes on. Returning the error instead + // reaches the SDK as NotFound for a 404 carrying no code, which it + // reads as a warning: the children would be emitted as none and + // ConductorOne would delete them. + if errors.Is(err, errCatalogGone) { + sawGone = true + + continue + } + if deniesTrust(err) { + lastDeny = err + + continue + } + return err } if catalog == nil { @@ -233,8 +250,18 @@ func pushSchemaChildRoute( return nil } if !sawCatalog { - return status.Errorf(codes.FailedPrecondition, - "databricks-connector: no workspace in sync scope lists catalog %s of metastore %s", catalogName, metastoreID) + // Every workspace named the catalog as gone and none refused the read. + // The catalog syncer drops it in the same sync, so this empties what + // ConductorOne is already removing. + if sawGone && lastDeny == nil { + return nil + } + msg := fmt.Sprintf("databricks-connector: no workspace in sync scope lists catalog %s of metastore %s", catalogName, metastoreID) + if lastDeny != nil { + msg = fmt.Sprintf("%s: %s", msg, lastDeny.Error()) + } + + return status.Errorf(codes.FailedPrecondition, "%s", msg) } return status.Errorf(codes.PermissionDenied, diff --git a/pkg/connector/schema_child_test.go b/pkg/connector/schema_child_test.go index 613a8d48..bfcf34a5 100644 --- a/pkg/connector/schema_child_test.go +++ b/pkg/connector/schema_child_test.go @@ -190,3 +190,69 @@ func TestSchemaChildListTellsADeletedSchemaFromAnInvisibleOne(t *testing.T) { }) } } + +// A workspace that does not serve the catalog answers the catalog read with a +// 404 carrying no error code, the same answer a routing or permission miss +// gives. The schema syncer walks on to the next workspace; listing a schema's +// children must do the same. Returning that error instead reaches the SDK as +// codes.NotFound, which it reads as a warning: the listing finishes green with +// no resources and ConductorOne deletes every table and volume under the schema. +func TestSchemaChildListTriesTheNextWorkspaceWhenOneDoesNotServeTheCatalog(t *testing.T) { + for _, errorCode := range []string{"", errorCodeCatalogGone} { + name := "bare 404" + if errorCode != "" { + name = errorCode + } + t.Run(name, func(t *testing.T) { + f := &fakeSchemaAccount{ + second: true, + metastoreOwner: "eng", + catalogOwner: "eng", + catalogStatus: map[string]int{"wsa": http.StatusNotFound}, + catalogErrorCode: errorCode, + schemas: []map[string]any{{"name": "sales", "owner": "eng"}}, + tables: []map[string]any{{"name": "orders", "owner": "eng", "table_type": "MANAGED"}}, + } + tables, _ := newTestChildBuildersAs(t, f, "eng") + + listed, err := listAllChildren(t, tables.List) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(listed) != 1 { + t.Fatalf("listed %d table(s), want 1 through the second workspace", len(listed)) + } + if workspace, _ := rs.GetProfileStringValue(rs.GetProfile(listed[0]), profileKeyWorkspace); workspace != "wsb" { + t.Errorf("profile[%s] = %q, want wsb", profileKeyWorkspace, workspace) + } + }) + } +} + +// Every workspace naming the catalog as gone, with none refusing the read, is +// the catalog being dropped. The catalog syncer removes it in the same sync, so +// the children listing empties what ConductorOne is already removing instead of +// failing the sync. +func TestSchemaChildListIsEmptyWhenEveryWorkspaceSaysTheCatalogIsGone(t *testing.T) { + f := &fakeSchemaAccount{ + second: true, + metastoreOwner: "eng", + catalogOwner: "eng", + catalogStatus: map[string]int{"wsa": http.StatusNotFound, "wsb": http.StatusNotFound}, + catalogErrorCode: errorCodeCatalogGone, + schemas: []map[string]any{{"name": "sales", "owner": "eng"}}, + tables: []map[string]any{{"name": "orders", "owner": "eng", "table_type": "MANAGED"}}, + } + tables, _ := newTestChildBuildersAs(t, f, "eng") + + listed, err := listAllChildren(t, tables.List) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(listed) != 0 { + t.Errorf("listed %d table(s), want none under a catalog that is gone", len(listed)) + } + if want := []string{"wsa", "wsb"}; !slices.Equal(f.catalogReads, want) { + t.Errorf("catalog reads = %v, want %v: both workspaces are asked before the listing empties", f.catalogReads, want) + } +} From 34cecee6c258d5678fb4fd23a4fad14f1cf54d08 Mon Sep 17 00:00:00 2001 From: Mateo Hernandez Date: Thu, 5 Nov 2026 11:59:03 -0300 Subject: [PATCH 4/4] fix(databricks): keep a parent-catalog 404 from emptying a schema's children schemaThrough classified every refusal of the single-schema read with schemaListingError, which maps CATALOG_DOES_NOT_EXIST to codes.NotFound. The SDK reads NotFound from a List as a warning, not a failure: isWarning (sync/syncer.go) returns true for it and syncOneAction (sync/parallel_syncer.go) logs "skipping sync action" and carries on. The listing then completes having emitted nothing and ConductorOne deletes every table and volume under that schema. A parent-catalog 404 is a real answer there: measured on the tenant, GET /schemas/{full} returns CATALOG_DOES_NOT_EXIST once the catalog is dropped. So the read now refuses as unreadable, which reaches C1 as FailedPrecondition and keeps what it already has. Only a code naming the schema itself still reports errSchemaGone, and the caller decides what that absence means. The message drops the privilege remedy: it is the answer to a partial listing, not to an arbitrary refusal, and on a 403 it contradicted the remedy the client layer already appends. schemaReadError is inlined into the branch that uses it, since it classified nothing. Side effect worth naming: a 403 now reaches C1 as PermissionDenied and a 401 as Unauthenticated, where both were FailedPrecondition. Neither is a warning, so neither changes what C1 keeps. Co-Authored-By: Claude Opus 5 --- pkg/connector/schema.go | 16 +++++++------- pkg/connector/schema_child.go | 7 +++--- pkg/connector/schema_child_test.go | 35 ++++++++++++++++++++++++++++++ 3 files changed, 46 insertions(+), 12 deletions(-) diff --git a/pkg/connector/schema.go b/pkg/connector/schema.go index b056560f..21a23c6b 100644 --- a/pkg/connector/schema.go +++ b/pkg/connector/schema.go @@ -368,19 +368,17 @@ func (b *schemaBuilder) pushRoute( var errSchemaGone = errors.New("schema no longer exists") // schemaThrough reads one schema through workspace. It reports errSchemaGone -// when Databricks names the schema missing, which proves nothing on its own: a -// caller without catalog-wide visibility is answered the same way for a schema -// it merely cannot see, so the caller decides what the absence means. Every -// other refusal, a 404 carrying no code included, leaves the schema unknown -// rather than absent: read as an empty result the SDK would skip the children -// and ConductorOne would delete every table and volume under it. +// when Databricks names the schema itself missing, which proves nothing on its +// own, so the caller decides what that absence means. Every other refusal leaves +// the schema unknown: read as empty, C1 would delete its tables and volumes. func schemaThrough( ctx context.Context, client *databricks.Client, workspace, catalogName, schemaName string, annos *annotations.Annotations, ) (*databricks.Schema, error) { - schema, rateLimit, err := client.GetSchema(ctx, workspace, catalogName+schemaNameSeparator+schemaName) + fullName := catalogName + schemaNameSeparator + schemaName + schema, rateLimit, err := client.GetSchema(ctx, workspace, fullName) noteRateLimit(annos, rateLimit) switch { case err == nil: @@ -390,7 +388,9 @@ func schemaThrough( // The answer does not describe the schema asked for. return nil, err default: - return nil, schemaListingError(catalogName, workspace, err) + return nil, refuseUnreadable(fmt.Sprintf( + "databricks-connector: schema %s cannot be read through workspace %s, "+ + "so its tables and volumes are unknown", fullName, workspace), err) } return schema, nil diff --git a/pkg/connector/schema_child.go b/pkg/connector/schema_child.go index f968ccfc..04e6df15 100644 --- a/pkg/connector/schema_child.go +++ b/pkg/connector/schema_child.go @@ -120,10 +120,9 @@ const childListRemedy = "the connector's service principal must own the metastor "or hold ALL_PRIVILEGES on the catalog" // pushSchemaChildRoute records a workspace that can list every table or volume -// of the schema. It walks the running in-scope workspaces until one lists the -// catalog and the child listing is complete. An error from a workspace stops -// the walk. An empty result is an error: it would make ConductorOne delete -// the tables and volumes. +// of the schema, walking the running in-scope workspaces until one serves the +// catalog and the child listing is complete. A workspace that does not serve it +// is skipped; an empty result is an error unless every workspace named it gone. func pushSchemaChildRoute( ctx context.Context, client *databricks.Client, diff --git a/pkg/connector/schema_child_test.go b/pkg/connector/schema_child_test.go index bfcf34a5..86196403 100644 --- a/pkg/connector/schema_child_test.go +++ b/pkg/connector/schema_child_test.go @@ -6,6 +6,7 @@ import ( "fmt" "net/http" "slices" + "strings" "testing" v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" @@ -154,6 +155,14 @@ func TestSchemaChildListTellsADeletedSchemaFromAnInvisibleOne(t *testing.T) { clientID: "eng", catalogOwner: "eng", wantErr: codes.FailedPrecondition, }, + { + // CATALOG_DOES_NOT_EXIST names the parent, not this schema, and + // one workspace is not every workspace, so it does not prove the + // subtree is gone. + name: "a 404 naming the parent catalog is not proof either", + clientID: "eng", catalogOwner: "eng", errorCode: errorCodeCatalogGone, + wantErr: codes.FailedPrecondition, + }, } for _, test := range tests { @@ -256,3 +265,29 @@ func TestSchemaChildListIsEmptyWhenEveryWorkspaceSaysTheCatalogIsGone(t *testing t.Errorf("catalog reads = %v, want %v: both workspaces are asked before the listing empties", f.catalogReads, want) } } + +// Every workspace refusing the catalog read is not the catalog being gone. The +// listing must fail and keep what ConductorOne already has, and the refusal has +// to reach the operator rather than be swallowed into a generic message. +func TestSchemaChildListFailsWhenEveryWorkspaceRefusesTheCatalog(t *testing.T) { + f := &fakeSchemaAccount{ + second: true, + metastoreOwner: "eng", + catalogOwner: "eng", + catalogStatus: map[string]int{"wsa": http.StatusForbidden, "wsb": http.StatusForbidden}, + schemas: []map[string]any{{"name": "sales", "owner": "eng"}}, + tables: []map[string]any{{"name": "orders", "owner": "eng", "table_type": "MANAGED"}}, + } + tables, _ := newTestChildBuildersAs(t, f, "eng") + + listed, err := listAllChildren(t, tables.List) + if status.Code(err) != codes.FailedPrecondition { + t.Fatalf("List error = %v, want FailedPrecondition", err) + } + if len(listed) != 0 { + t.Errorf("listed %d table(s), want none", len(listed)) + } + if !strings.Contains(err.Error(), "403") { + t.Errorf("List error = %v, want the refusal named in the message", err) + } +}