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

catalog-backend: separated out state writing + give state a ttl

Browse Source 
Co-authored-by: Fredrik Adelöw <freben@gmail.com>
Co-authored-by: blam <ben@blam.sh>
Co-authored-by: Johan Haals <johan.haals@gmail.com>
Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>
This commit is contained in:
Patrik Oldsberg
2021-09-22 14:58:55 +02:00
committed by Johan Haals
parent b562c68091
commit 69acb4e7ba
6 changed files with 192 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts
+87 -2
View File
@@ -28,6 +28,7 @@ describe('DefaultCatalogProcessingEngine', () => {
transaction: jest.fn(),
getProcessableEntities: jest.fn(),
updateProcessedEntity: jest.fn(),
updateEntityCache: jest.fn(),
} as unknown as jest.Mocked<DefaultProcessingDatabase>;
const orchestrator: jest.Mocked<CatalogProcessingOrchestrator> = {
process: jest.fn(),
@@ -84,7 +85,7 @@ describe('DefaultCatalogProcessingEngine', () => {
metadata: { name: 'test' },
},
resultHash: '',
state: [],
state: [] as any,
nextUpdateAt: DateTime.now(),
lastDiscoveryAt: DateTime.now(),
},
@@ -221,16 +222,100 @@ describe('DefaultCatalogProcessingEngine', () => {
expect(hash.digest).toBeCalledTimes(1);
expect(db.updateProcessedEntity).toBeCalledTimes(1);
});
expect(db.updateEntityCache).not.toHaveBeenCalled();
db.getProcessableEntities
.mockReset()
.mockResolvedValueOnce({ items: [refreshState] })
.mockResolvedValueOnce({
items: [{ ...refreshState, state: { something: 'different' } }],
})
.mockResolvedValue({ items: [] });
await waitForExpect(() => {
expect(orchestrator.process).toBeCalledTimes(2);
expect(hash.digest).toBeCalledTimes(2);
expect(db.updateProcessedEntity).toBeCalledTimes(1);
expect(db.updateEntityCache).toBeCalledTimes(1);
});
expect(db.updateEntityCache).toHaveBeenCalledWith(expect.anything(), {
id: '',
state: { ttl: 5 },
});
await engine.stop();
});
it('should decrease the state ttl if there are errors', async () => {
const entity = {
apiVersion: '1',
kind: 'Location',
metadata: { name: 'test' },
};
const refreshState = {
id: '',
entityRef: '',
unprocessedEntity: entity,
resultHash: 'the matching hash',
state: { some: 'value', ttl: 1 },
nextUpdateAt: DateTime.now(),
lastDiscoveryAt: DateTime.now(),
};
hash.digest.mockReturnValue('the matching hash');
orchestrator.process.mockResolvedValue({
ok: false,
errors: [],
});
const engine = new DefaultCatalogProcessingEngine(
getVoidLogger(),
[],
db,
orchestrator,
stitcher,
() => hash,
);
db.transaction.mockImplementation(cb => cb((() => {}) as any));
await engine.start();
db.getProcessableEntities
.mockResolvedValueOnce({
items: [refreshState],
})
.mockResolvedValue({ items: [] });
await waitForExpect(() => {
expect(db.updateEntityCache).toBeCalledTimes(1);
});
expect(db.updateEntityCache).toHaveBeenCalledWith(expect.anything(), {
id: '',
state: { some: 'value', ttl: 0 },
});
// Second run, the TTL should now reach 0 and the cache should be cleared
db.getProcessableEntities
.mockResolvedValueOnce({
items: [
{
...refreshState,
state: db.updateEntityCache.mock.calls[0][1].state,
},
],
})
.mockResolvedValue({ items: [] });
db.updateEntityCache.mockReset();
await waitForExpect(() => {
expect(db.updateEntityCache).toBeCalledTimes(1);
});
expect(db.updateEntityCache).toHaveBeenCalledWith(expect.anything(), {
id: '',
state: {},
});
await engine.stop();
plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts
+26 -3
View File
@@ -38,6 +38,8 @@ import {
EntityProviderMutation,
} from './types';
const CACHE_TTL = 5;
class Connection implements EntityProviderConnection {
readonly validateEntityEnvelope = entityEnvelopeSchemaValidator();
@@ -151,6 +153,29 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
track.markProcessorsCompleted(result);
if (result.ok) {
if (stableStringify(state) !== stableStringify(result.state)) {
await this.processingDatabase.transaction(async tx => {
await this.processingDatabase.updateEntityCache(tx, {
id,
state: {
ttl: CACHE_TTL,
...result.state,
},
});
});
}
} else {
const maybeTtl = state?.ttl;
const ttl = Number.isInteger(maybeTtl) ? (maybeTtl as number) : 0;
await this.processingDatabase.transaction(async tx => {
await this.processingDatabase.updateEntityCache(tx, {
id,
state: ttl > 0 ? { ...state, ttl: ttl - 1 } : {},
});
});
}
for (const error of result.errors) {
// TODO(freben): Try to extract the location out of the unprocessed
// entity and add as meta to the log lines
@@ -167,8 +192,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
hashBuilder = hashBuilder
.update(stableStringify({ ...result.completedEntity }))
.update(stableStringify([...result.deferredEntities]))
.update(stableStringify([...result.relations]))
.update(stableStringify(result.state));
.update(stableStringify([...result.relations]));
}
const resultHash = hashBuilder.digest('hex');
@@ -208,7 +232,6 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
id,
processedEntity: result.completedEntity,
resultHash,
state: result.state,
errors: errorsString,
relations: result.relations,
deferredEntities: result.deferredEntities,
plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts
+48 -8
View File
@@ -212,7 +212,6 @@ describe('Default Processing Database', () => {
id,
processedEntity,
resultHash: '',
state: {},
relations: [],
deferredEntities: [],
}),
@@ -231,7 +230,6 @@ describe('Default Processing Database', () => {
id,
processedEntity,
resultHash: '',
state: {},
relations: [],
deferredEntities: [],
locationKey: 'key',
@@ -284,14 +282,11 @@ describe('Default Processing Database', () => {
last_discovery_at: '2021-04-01 13:37:00',
});
const state = { hello: { t: 'something' } };
await db.transaction(tx =>
db.updateProcessedEntity(tx, {
id,
processedEntity,
resultHash: '',
state,
relations: [],
deferredEntities: [],
locationKey: 'key',
@@ -306,7 +301,6 @@ describe('Default Processing Database', () => {
expect(entities[0].processed_entity).toEqual(
JSON.stringify(processedEntity),
);
expect(entities[0].cache).toEqual(JSON.stringify(state));
expect(entities[0].errors).toEqual("['something broke']");
expect(entities[0].location_key).toEqual('key');
},
@@ -348,7 +342,6 @@ describe('Default Processing Database', () => {
id,
processedEntity,
resultHash: '',
state: {},
relations: relations,
deferredEntities: [],
}),
@@ -400,7 +393,6 @@ describe('Default Processing Database', () => {
id,
processedEntity,
resultHash: '',
state: {},
relations: [],
deferredEntities,
}),
@@ -418,6 +410,54 @@ describe('Default Processing Database', () => {
);
});
describe('updateEntityCache', () => {
it.each(databases.eachSupportedId())(
'updates the entityCache, %p',
async databaseId => {
const { knex, db } = await createDatabase(databaseId);
const id = '123';
await insertRefreshStateRow(knex, {
entity_id: id,
entity_ref: 'location:default/fakelocation',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: '2021-04-01 13:37:00',
last_discovery_at: '2021-04-01 13:37:00',
});
const state = { hello: { t: 'something' } };
await db.transaction(tx =>
db.updateEntityCache(tx, {
id,
state,
}),
);
const entities = await knex<DbRefreshStateRow>(
'refresh_state',
).select();
expect(entities.length).toBe(1);
expect(entities[0].cache).toEqual(JSON.stringify(state));
await db.transaction(tx =>
db.updateEntityCache(tx, {
id,
state: undefined,
}),
);
const entities2 = await knex<DbRefreshStateRow>(
'refresh_state',
).select();
expect(entities2.length).toBe(1);
expect(entities2[0].cache).toEqual('{}');
},
60_000,
);
});
describe('replaceUnprocessedEntities', () => {
const createLocations = async (db: Knex, entityRefs: string[]) => {
for (const ref of entityRefs) {
plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts
+13 -2
View File
@@ -40,6 +40,7 @@ import {
UpdateProcessedEntityOptions,
ListAncestorsOptions,
ListAncestorsResult,
UpdateEntityCacheOptions,
} from './types';
// The number of items that are sent per batch to the database layer, when
@@ -69,7 +70,6 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
id,
processedEntity,
resultHash,
state,
errors,
relations,
deferredEntities,
@@ -79,7 +79,6 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
.update({
processed_entity: JSON.stringify(processedEntity),
result_hash: resultHash,
cache: JSON.stringify(state),
errors,
location_key: locationKey,
})
@@ -140,6 +139,18 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
.where('entity_id', id);
}
async updateEntityCache(
txOpaque: Transaction,
options: UpdateEntityCacheOptions,
): Promise<void> {
const tx = txOpaque as Knex.Transaction;
const { id, state } = options;
await tx<DbRefreshStateRow>('refresh_state')
.update({ cache: JSON.stringify(state ?? {}) })
.where('entity_id', id);
}
private deduplicateRelations(rows: DbRelationsRow[]): DbRelationsRow[] {
return lodash.uniqBy(
rows,
plugins/catalog-backend/src/next/database/types.ts
+15 -3
View File
@@ -15,7 +15,7 @@
*/
import { Entity, EntityRelationSpec } from '@backstage/catalog-model';
import { JsonValue } from '@backstage/config';
import { JsonObject } from '@backstage/config';
import { DateTime } from 'luxon';
import { Transaction } from '../../database/types';
import { DeferredEntity } from '../processing/types';
@@ -36,13 +36,17 @@ export type UpdateProcessedEntityOptions = {
id: string;
processedEntity: Entity;
resultHash: string;
state?: JsonValue;
errors?: string;
relations: EntityRelationSpec[];
deferredEntities: DeferredEntity[];
locationKey?: string;
};
export type UpdateEntityCacheOptions = {
id: string;
state?: JsonObject;
};
export type UpdateProcessedEntityErrorsOptions = {
id: string;
errors?: string;
@@ -57,7 +61,7 @@ export type RefreshStateItem = {
resultHash: string;
nextUpdateAt: DateTime;
lastDiscoveryAt: DateTime; // remove?
state?: JsonValue;
state?: JsonObject;
errors?: string;
locationKey?: string;
};
@@ -118,6 +122,14 @@ export interface ProcessingDatabase {
options: UpdateProcessedEntityOptions,
): Promise<void>;
/**
* Updates the cache associated with an entity.
*/
updateEntityCache(
txOpaque: Transaction,
options: UpdateEntityCacheOptions,
): Promise<void>;
/**
* Updates only the errors of a processed entity
*/
plugins/catalog-backend/src/next/processing/types.ts
+3 -3
View File
@@ -15,17 +15,17 @@
*/
import { Entity, EntityRelationSpec } from '@backstage/catalog-model';
import { JsonValue } from '@backstage/config';
import { JsonObject } from '@backstage/config';
export type EntityProcessingRequest = {
entity: Entity;
state?: JsonValue; // Versions for multiple deployments etc
state?: JsonObject; // Versions for multiple deployments etc
};
export type EntityProcessingResult =
| {
ok: true;
state: JsonValue;
state: JsonObject;
completedEntity: Entity;
deferredEntities: DeferredEntity[];
relations: EntityRelationSpec[];