paulononato/backstage
1
0
Fork 0
Code Issues Pull Requests Actions 48 Packages Projects Releases Wiki Activity

WIP refreshing entities

Browse Source 
Signed-off-by: Johan Haals <johan.haals@gmail.com>
This commit is contained in:
Johan Haals
2021-09-10 14:20:15 +02:00
parent 20cce38c84
commit 9c9353e2d3
5 changed files with 181 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts
+20 -16
View File
@@ -73,11 +73,12 @@ class Connection implements EntityProviderConnection {
});
});
} else if (mutation.type === 'refresh') {
await db.transaction(async tx => {
await db.refreshUnprocessedEntities(tx, {
match: mutation.match,
});
});
// await db.transaction(async tx => {
// await db.refreshUnprocessedEntities(tx, {
// match: mutation.match,
// });
// });
console.log('wopoop');
}
}
@@ -245,17 +246,20 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
}
async refresh(options: EntityRefreshOptions) {
await Promise.all(
this.entityProviders.map(async provider => {
try {
await provider.refresh?.(options);
} catch (e) {
throw new Error(
`Provider ${provider.getProviderName()} failed refresh, ${e}`,
);
}
}),
);
await this.processingDatabase.transaction(async tx => {
await this.processingDatabase.refreshUnprocessedEntities(tx, options);
});
// await Promise.all(
// this.entityProviders.map(async provider => {
// try {
// await provider.refresh?.(options);
// } catch (e) {
// throw new Error(
// `Provider ${provider.getProviderName()} failed refresh, ${e}`,
// );
// }
// }),
// );
}
}
plugins/catalog-backend/src/next/DefaultLocationStore.ts
+14 -17
View File
@@ -137,23 +137,20 @@ export class DefaultLocationStore implements LocationStore, EntityProvider {
});
}
async refresh(options: EntityRefreshOptions) {
const match: RefreshStateMatch = {};
// locationKey?: string;
// entityRef?: string;
if (options.entityRef) {
match.entityRef = options.entityRef;
}
if (options.locationRef) {
// TODO
}
await this.connection.applyMutation({
type: 'refresh',
match,
});
async refresh(_options: EntityRefreshOptions) {
// const match: RefreshStateMatch = {};
// // locationKey?: string;
// // entityRef?: string;
// if (options.entityRef) {
// match.entityRef = options.entityRef;
// }
// if (options.locationRef) {
// // TODO
// }
// await this.connection.applyMutation({
// type: 'refresh',
// match,
// });
}
private async locations(dbOrTx: Knex.Transaction | Knex = this.db) {
plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts
+83
View File
@@ -1017,4 +1017,87 @@ describe('Default Processing Database', () => {
60_000,
);
});
describe('refreshEntities', () => {
it.each(databases.eachSupportedId())(
'should refresh the location further up the tree, %p',
async databaseId => {
const { knex, db } = await createDatabase(databaseId);
await knex<DbRefreshStateRow>('refresh_state').insert({
entity_id: '7',
entity_ref: 'location:default/myloc',
unprocessed_entity: JSON.stringify({
kind: 'Location',
apiVersion: '1.0.0',
metadata: {
name: 'xyz',
},
} as Entity),
errors: '[]',
next_update_at: '2031-01-01 23:00:00',
last_discovery_at: '2021-04-01 13:37:00',
});
await knex<DbRefreshStateRow>('refresh_state').insert({
entity_id: '8',
entity_ref: 'component:default/mycomp',
unprocessed_entity: JSON.stringify({
kind: 'Component',
apiVersion: '1.0.0',
metadata: {
name: 'xyz',
},
} as Entity),
errors: '[]',
next_update_at: '2031-01-01 23:00:00',
last_discovery_at: '2021-04-01 13:37:00',
});
await knex<DbRefreshStateRow>('refresh_state').insert({
entity_id: '9',
entity_ref: 'api:default/myapi',
unprocessed_entity: JSON.stringify({
kind: 'Api',
apiVersion: '1.0.0',
metadata: {
name: 'xyz',
},
} as Entity),
errors: '[]',
next_update_at: '2031-01-01 23:00:00',
last_discovery_at: '2021-04-01 13:37:00',
});
await insertRefRow(knex, {
source_entity_ref: 'component:default/mycomp',
target_entity_ref: 'api:default/myapi',
});
await insertRefRow(knex, {
source_entity_ref: 'location:default/myloc',
target_entity_ref: 'component:default/mycomp',
});
await insertRefRow(knex, {
source_key: 'ConfigLocationProvider',
target_entity_ref: 'location:default/myloc',
});
await db.transaction(async tx => {
await db.refreshUnprocessedEntities(tx, {
entityRef: 'api:default/myapi',
});
});
const [result] = await knex<DbRefreshStateRow>('refresh_state')
.where('entity_ref', 'api:default/myapi')
.select();
// TODO: This is going to break after 2031
expect(parseDate(result.next_update_at).year).toEqual(
DateTime.local().year,
);
},
);
});
});
plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts
+58 -2
View File
@@ -14,9 +14,13 @@
* limitations under the License.
*/
import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
import {
Entity,
LOCATION_ANNOTATION,
stringifyEntityRef,
} from '@backstage/catalog-model';
import { JsonObject } from '@backstage/config';
import { ConflictError } from '@backstage/errors';
import { ConflictError, NotFoundError } from '@backstage/errors';
import { Knex } from 'knex';
import lodash from 'lodash';
import { v4 as uuid } from 'uuid';
@@ -351,6 +355,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
unprocessed_entity: serializedEntity,
location_key: locationKey,
last_discovery_at: tx.fn.now(),
next_update_at: tx.fn.now(),
})
.where('entity_ref', entityRef)
.andWhere(inner => {
@@ -509,11 +514,62 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
};
}
// TODO(jhaals): Rename this to refreshEntity?
async refreshUnprocessedEntities(
txOpaque: Transaction,
options: RefreshUnprocessedEntitiesOptions,
): Promise<void> {
const tx = txOpaque as Knex.Transaction;
if ('entityRef' in options) {
const { entityRef } = options;
const [result] = await tx<DbRefreshStateRow>('refresh_state')
.where({ entity_ref: entityRef })
.select();
if (!result) {
throw new NotFoundError(`EntityRef ${entityRef} not found`);
}
const refs = await tx<DbRefreshStateReferencesRow>(
'refresh_state_references',
)
.where({ target_entity_ref: entityRef })
.select();
for (const ref of refs) {
if (ref.source_entity_ref?.startsWith('location:')) {
const updateResult = await tx<DbRefreshStateRow>('refresh_state')
.where({ entity_ref: ref.source_entity_ref })
.update({ next_update_at: tx.fn.now() });
if (updateResult === 0) {
throw new ConflictError(
`Failed to schedule ${ref.source_entity_ref} for refresh`,
);
}
}
}
/*
For a given entityRef:
- Fetch entity
- recursively select from refresh_state_references where target is our entityRef.
Continue until we find a location.
- Process and run addUnprocessedEntities with a flag telling it to bump the timestamp for all deferred entities.
*/
/*
For a given URL
- Fetch entity based on managed by location URL?
- repeat for entityRef
*/
}
if ('locationRef' in options) {
// TODO(jhaals): managed by or origin?
// const entity: Entity = JSON.parse(result.processed_entity);
// // TODO: ONly required for URLz
// const managedBy = entity.metadata?.annotations?.[LOCATION_ANNOTATION]
// console.log(managedBy);
}
}
async transaction<T>(fn: (tx: Transaction) => Promise<T>): Promise<T> {
plugins/catalog-backend/src/next/database/types.ts
+6 -3
View File
@@ -85,9 +85,12 @@ export type RefreshStateMatch = {
parentOfEntityRef: string;
};
export type RefreshUnprocessedEntitiesOptions = {
match: RefreshStateMatch;
};
export type RefreshUnprocessedEntitiesOptions =
| {
// match: RefreshStateMatch;
entityRef: string;
}
| { locationRef: string };
export interface ProcessingDatabase {
transaction<T>(fn: (tx: Transaction) => Promise<T>): Promise<T>;