Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 36 additions & 6 deletions toolbox/mdcode/src/libts/gcp/dataplex.ts
Original file line number Diff line number Diff line change
Expand Up @@ -165,6 +165,29 @@ export class CatalogClient extends api.ApiClient {
return res;
}

// The type of every aspect on an entry, as project.location.type, each
// once. A FULL read lists every aspect but gives data only to those the
// entry's type requires, so this reads the keys alone, which lookupEntry
// would drop.
async lookupEntryAspectTypes(project: string, location: string,
name: string): Promise<api.ApiResult<string[]>> {
const container = `${catalogContainer(project, location)}:lookupEntry`;
const res = await this._get<Entry>(container, { entry: name, view: 'FULL' });
if (res.status != 200 || !res.result) {
return { status: res.status, message: res.message };
}
// Each type is resolved once: crm caches only the projects it can read,
// so several field-path keys of one unreadable type would each look it up.
const typeRefs = new Set(Object.keys(res.result.aspects ?? {})
.map(key => key.split('@')[0]));
const types = new Set<string>();
for (const typeRef of typeRefs) {
const { aspectType } = await _aspectTypeOfKey(typeRef, this.context);
types.add(_nameToTypeRef(aspectType));
}
return { status: 200, result: [...types] };
}

async modifyEntry(project: string, location: string, entry: Entry,
updateMask?: string[],
aspectKeys?: string[]): Promise<api.ApiResult<Entry>> {
Expand Down Expand Up @@ -445,12 +468,7 @@ export async function _fixEntry(entry: Entry, ctx: context.ApiContext,
const fixedAspects: Record<string, Aspect> = {};
for (const [aspectKey, aspectValue] of Object.entries(entry.aspects)) {
if (!aspectValue || !Object.keys(aspectValue).length) continue;
const atIdx = aspectKey.indexOf('@');
const typeRef = atIdx === -1 ? aspectKey : aspectKey.slice(0, atIdx);
const keyPath = atIdx === -1 ? undefined : aspectKey.slice(atIdx + 1);

const aspectType =
await crm.fixProject(_typeRefToName(typeRef, 'aspect'), ctx);
const { aspectType, keyPath } = await _aspectTypeOfKey(aspectKey, ctx);

const path = aspectValue.path || keyPath;
const baseKey = _nameToTypeRef(aspectType);
Expand All @@ -466,6 +484,18 @@ export async function _fixEntry(entry: Entry, ctx: context.ApiContext,
}
}

// The aspect type an aspect key names, as a resource name with a project ID,
// and the key's `@<path>` part, if any.
async function _aspectTypeOfKey(key: string, ctx: context.ApiContext):
Promise<{ aspectType: string; keyPath?: string }> {
const atIdx = key.indexOf('@');
const typeRef = atIdx === -1 ? key : key.slice(0, atIdx);
return {
aspectType: await crm.fixProject(_typeRefToName(typeRef, 'aspect'), ctx),
keyPath: atIdx === -1 ? undefined : key.slice(atIdx + 1),
};
}

// Constructs canonical names for catalog container resources, identified by project, location and
// optionally, depending on use-case, the entry group.
export function catalogContainer(project: string, location: string, entryGroup: string=''): string {
Expand Down
3 changes: 3 additions & 0 deletions toolbox/mdcode/src/libts/source.ts
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,9 @@ export interface CatalogSource {
// them along with those each entry's type requires. System-generated aspects
// are left out, since push sends every local aspect back.
readonly defaultAspects: readonly string[];
// The aspect types the service writes on this scope's entries beyond those
// their entry types require. Pull leaves them out without a warning.
readonly systemAspects: readonly string[];

entries(ctx: gcp.ApiContext): AsyncGenerator<gcp.Entry, void, unknown>;
localName(entry: gcp.Entry): string;
Expand Down
8 changes: 8 additions & 0 deletions toolbox/mdcode/src/libts/sources/bq-dataset.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,14 @@ export class BigQueryDatasetSource implements CatalogSource {
readonly ingestedEntries = true;
readonly layout = Layouts.STANDARD;
readonly defaultAspects = [md.OVERVIEW_ASPECT];
// BigQuery and Dataplex data scans write these on a table besides the ones
// its type requires.
readonly systemAspects = [
'dataplex-types.global.bigquery-policy',
'dataplex-types.global.data-profile',
'dataplex-types.global.data-quality-scorecard',
'dataplex-types.global.usage',
];

private readonly _datasets: Map<string, bq.Dataset>;

Expand Down
1 change: 1 addition & 0 deletions toolbox/mdcode/src/libts/sources/entrygroup.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ export class EntryGroupSource implements CatalogSource {
readonly ingestedEntries: boolean;
readonly layout = Layouts.STANDARD;
readonly defaultAspects = [md.OVERVIEW_ASPECT];
readonly systemAspects = [];

private readonly _name: string[];
private readonly _entryGroup: dataplex.EntryGroup;
Expand Down
1 change: 1 addition & 0 deletions toolbox/mdcode/src/libts/sources/kb.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ export class KnowledgeBaseSource implements CatalogSource {
readonly ingestedEntries = false;
readonly layout = Layouts.DOCUMENTS;
readonly defaultAspects = [md.OVERVIEW_ASPECT];
readonly systemAspects = [];

private readonly _name: string[];
private readonly _entryGroup: dataplex.EntryGroup;
Expand Down
1 change: 1 addition & 0 deletions toolbox/mdcode/src/libts/sources/semantic-model.ts
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ export class SemanticModelSource implements CatalogSource {
readonly layout = Layouts.SEMANTIC_MODEL;
// A semantic model is pulled by its own path, not by snapshot aspects.
readonly defaultAspects = [];
readonly systemAspects = [];

readonly project: string;
readonly location: string;
Expand Down
27 changes: 25 additions & 2 deletions toolbox/mdcode/src/libts/sync.ts
Original file line number Diff line number Diff line change
Expand Up @@ -54,22 +54,45 @@ export class CatalogSync {
entries.push(entry);
}

const aspectTypes = [...this._snapshot.aspectTypes.keys()];
// Pull fetches the aspect types it knows: those catalog.yaml lists and
// each entry's type requires. It warns about each other aspect type an
// entry carries, apart from the scope's system aspects.
const aspectTypes = this._snapshot.aspectTypes;
const requested = [...aspectTypes.keys()];
const system = new Set(this._snapshot.manifest.source.systemAspects);
const leftOut = new Map<string, number>();
for (const entry of entries) {
// TODO: Handle local modification conflicts.
// TODO: Handle config changes or service deletions that require removing local entries.

const nameParts = entry.name.split('/');
const res = await this._catalog.lookupEntry(nameParts[1], nameParts[3], entry.name,
aspectTypes);
requested);
// The server will respond with 403 permission denied for both resource not exist or
// insufficient permission. We cannot tell if a resource not exist or user does not
// have the access. Thus using 200 for an ensured result.
if (res.status != 200 || !res.result) {
continue;
}

// lookupEntry sets every returned aspect's type.
const stored = new Set(Object.values(res.result.aspects ?? {})
.map(a => gcp._nameToTypeRef(a.aspectType!)));
await this._snapshot._storeEntry(res.result);

const onEntry = await this._catalog.lookupEntryAspectTypes(
nameParts[1], nameParts[3], entry.name);
for (const type of onEntry.result ?? []) {
if (!stored.has(type) && !aspectTypes.has(type) && !system.has(type)) {
leftOut.set(type, (leftOut.get(type) ?? 0) + 1);
}
}
}
for (const [type, count] of leftOut) {
console.warn(
`Warning: pull left out aspect type '${type}' on ${count} ` +
`${count === 1 ? 'entry' : 'entries'}; list it under ` +
`snapshot.aspects in catalog.yaml to pull it.`);
}
return { success: true };
}
Expand Down
52 changes: 52 additions & 0 deletions toolbox/mdcode/tests/libts/gcp/dataplex.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -246,6 +246,58 @@ describe('_fixEntry removed aspects', () => {
});


describe('CatalogClient.lookupEntryAspectTypes', () => {
test('reads the FULL view and names each aspect type once, with a project ID',
async () => {
const client = new CatalogClient(CTX);
const get = spyOn(client as any, '_get').mockImplementation(async () => ({
status: 200,
result: {
name: ENTRY,
aspects: {
// A FULL read gives no data to an aspect the type does not require.
'dataplex-types.global.overview': {},
'655216118709.global.guidelines@Schema.order_id': {},
'dataplex-types.global.guidelines@Schema.customer_id': {},
},
},
}));

const res = await client.lookupEntryAspectTypes('proj', 'us', ENTRY);

expect(get.mock.calls[0].slice(0, 2)).toEqual(
['projects/proj/locations/us:lookupEntry', {entry: ENTRY, view: 'FULL'}]);
expect(res).toEqual({
status: 200,
result: ['dataplex-types.global.overview', 'dataplex-types.global.guidelines'],
});
});

test('looks up an unreadable project once per aspect type', async () => {
const client = new CatalogClient(CTX);
spyOn(client as any, '_get').mockImplementation(async () => ({
status: 200,
result: {
name: ENTRY,
aspects: {
'111.global.guidelines@Schema.order_id': {},
'111.global.guidelines@Schema.customer_id': {},
},
},
}));
// Resource Manager cannot read the project, and crm caches only the
// projects it can read.
const lookup = spyOn(ResourceManagerClient.prototype, 'getProject')
.mockResolvedValue({status: 403} as any);

const res = await client.lookupEntryAspectTypes('proj', 'us', ENTRY);

expect(res.result).toEqual(['111.global.guidelines']);
expect(lookup).toHaveBeenCalledTimes(1);
lookup.mockRestore();
});
});

// Dataplex keeps an aspect that aspectKeys names and the body leaves out
// unless deleteMissingAspects is set, so a caller that means to remove one has
// to send it.
Expand Down
11 changes: 11 additions & 0 deletions toolbox/mdcode/tests/libts/mocks.ts
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,17 @@ export class CatalogClientMock extends gcp.CatalogClient {
return { status: 200, result: { ...entry, aspects: Object.fromEntries(kept) } };
}

async lookupEntryAspectTypes(project: string, location: string, name: string): Promise<gcp.ApiResult<string[]>> {
const entry = this.mockEntries.find(e => e.name == name);
if (!entry) {
return { status: 404, message: 'Not found' };
}
// As Dataplex's FULL view does: it lists every aspect key on the entry,
// with data or not, so every aspect type appears, each once.
const types = Object.keys(entry.aspects ?? {}).map(k => k.split('@')[0]);
return { status: 200, result: [...new Set(types)] };
}

async modifyEntry(project: string, location: string, entry: gcp.Entry, updateMask?: string[], aspectKeys?: string[]): Promise<gcp.ApiResult<gcp.Entry>> {
return this.applyUpdate(entry, updateMask, aspectKeys);
}
Expand Down
20 changes: 19 additions & 1 deletion toolbox/mdcode/tests/libts/scenarios.ts
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,10 @@ function runScenario(scenario: any) {
const snapshot = await kcmac.CatalogSnapshot.fromPath('/', TEST_API_CONTEXT);
const sync = new kcmac.CatalogSync(catalog, snapshot);

// Execute - Snapshot actions
// Execute - Snapshot actions, recording the warnings they print.
const warnings: string[] = [];
const warn = spyOn(console, 'warn').mockImplementation(
(...args: any[]) => { warnings.push(args.join(' ')); });
const actions = scenario.actions ?? [];
for (const actionStep of actions) {
const { action, ...params } = actionStep;
Expand All @@ -119,6 +122,12 @@ function runScenario(scenario: any) {
throw new Error(`Unknown action: ${action}`);
}
}
warn.mockRestore();

// Assert expectations - Warnings
if (scenario.assert?.warnings) {
expect(warnings).toEqual(scenario.assert.warnings);
}

// Assert expectations - Filesystem
if (scenario.assert?.fileSystem) {
Expand Down Expand Up @@ -227,6 +236,15 @@ function main() {
}
);

spyOn(gcp.CatalogClient.prototype, 'lookupEntryAspectTypes').mockImplementation(
async function(project: string, location: string, entry: string) {
if (currentCatalogMock) {
return await currentCatalogMock.lookupEntryAspectTypes(project, location, entry);
}
return { status: 404, message: 'Not found' };
}
);

spyOn(gcp.CatalogClient.prototype, 'modifyEntry').mockImplementation(
async function(project: string, location: string, entry: gcp.Entry, updateMask?: string[], aspectKeys?: string[]) {
if (currentCatalogMock) {
Expand Down
14 changes: 13 additions & 1 deletion toolbox/mdcode/tests/scenarios/pull_bq.yaml
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
name: pull_bq
description: BigQuery dataset
description: BigQuery dataset, whose system-written aspects are left out without a warning
setup:
bigQuery:
datasets:
Expand All @@ -23,6 +23,16 @@ setup:
entryType: projects/dataplex-types/locations/global/entryTypes/bigquery-dataset
- name: projects/p/locations/us/entryGroups/@bigquery/entries/bigquery.googleapis.com/projects/p/datasets/d/tables/t1
entryType: projects/dataplex-types/locations/global/entryTypes/bigquery-table
# Aspects BigQuery writes besides the required ones: left out, with
# no warning.
aspects:
dataplex-types.global.usage:
aspectType: projects/dataplex-types/locations/global/aspectTypes/usage
data:
queryCount: 7
dataplex-types.global.bigquery-policy:
aspectType: projects/dataplex-types/locations/global/aspectTypes/bigquery-policy
data: {}

fileSystem:
catalog.yaml: |
Expand All @@ -39,3 +49,5 @@ assert:
catalog/p.d/t1.yaml:
- contains: "name: p.d/t1"
- contains: "type: dataplex-types.global.bigquery-table"
- notContains: "queryCount"
warnings: []
48 changes: 48 additions & 0 deletions toolbox/mdcode/tests/scenarios/pull_left_out.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,48 @@
name: pull_left_out
description: An entry group pulled with a catalog.yaml that lists no aspects leaves out the aspects it does not list and warns once for each aspect type
setup:
catalog:
entryGroups:
- name: projects/test/locations/us/entryGroups/g1
entryTypes:
- name: projects/test/locations/global/entryTypes/et1
aspectTypes:
- name: projects/test/locations/global/aspectTypes/extra
entries:
- name: projects/test/locations/us/entryGroups/g1/entries/e1
entryType: projects/test/locations/global/entryTypes/et1
aspects:
test.global.extra:
aspectType: projects/test/locations/global/aspectTypes/extra
data:
note: on e1
dataplex-types.global.overview:
aspectType: projects/dataplex-types/locations/global/aspectTypes/overview
data:
content: Text in Knowledge Catalog.
- name: projects/test/locations/us/entryGroups/g1/entries/e2
entryType: projects/test/locations/global/entryTypes/et1
aspects:
test.global.extra:
aspectType: projects/test/locations/global/aspectTypes/extra
data:
note: on e2

fileSystem:
catalog.yaml: |
scope: entryGroup.test.us.g1

actions:
- action: pull

assert:
fileSystem:
catalog/e1.yaml:
- contains: "name: e1"
- notContains: "note: on e1"
- notContains: "Text in Knowledge Catalog."
catalog/e2.yaml:
- notContains: "note: on e2"
warnings:
- "Warning: pull left out aspect type 'test.global.extra' on 2 entries; list it under snapshot.aspects in catalog.yaml to pull it."
- "Warning: pull left out aspect type 'dataplex-types.global.overview' on 1 entry; list it under snapshot.aspects in catalog.yaml to pull it."
Loading