From 69acb4e7baa4d2de6d76e7139723ba6863d37dca Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Wed, 22 Sep 2021 14:58:55 +0200 Subject: [PATCH] catalog-backend: separated out state writing + give state a ttl MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Fredrik Adelöw Co-authored-by: blam Co-authored-by: Johan Haals Signed-off-by: Patrik Oldsberg --- .../DefaultCatalogProcessingEngine.test.ts | 89 ++++++++++++++++++- .../next/DefaultCatalogProcessingEngine.ts | 29 +++++- .../DefaultProcessingDatabase.test.ts | 56 ++++++++++-- .../database/DefaultProcessingDatabase.ts | 15 +++- .../src/next/database/types.ts | 18 +++- .../src/next/processing/types.ts | 6 +- 6 files changed, 192 insertions(+), 21 deletions(-) diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts index c569bf8135..430d7f251f 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts @@ -28,6 +28,7 @@ describe('DefaultCatalogProcessingEngine', () => { transaction: jest.fn(), getProcessableEntities: jest.fn(), updateProcessedEntity: jest.fn(), + updateEntityCache: jest.fn(), } as unknown as jest.Mocked; const orchestrator: jest.Mocked = { 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(); diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index 1955220b2b..346a912009 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -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, diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts index a39d2ba981..c10a37396b 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts @@ -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( + '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( + '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) { diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 263ba1bdfd..33f01b14eb 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -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 { + const tx = txOpaque as Knex.Transaction; + const { id, state } = options; + + await tx('refresh_state') + .update({ cache: JSON.stringify(state ?? {}) }) + .where('entity_id', id); + } + private deduplicateRelations(rows: DbRelationsRow[]): DbRelationsRow[] { return lodash.uniqBy( rows, diff --git a/plugins/catalog-backend/src/next/database/types.ts b/plugins/catalog-backend/src/next/database/types.ts index ab449c8a7a..aabd28afc8 100644 --- a/plugins/catalog-backend/src/next/database/types.ts +++ b/plugins/catalog-backend/src/next/database/types.ts @@ -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; + /** + * Updates the cache associated with an entity. + */ + updateEntityCache( + txOpaque: Transaction, + options: UpdateEntityCacheOptions, + ): Promise; + /** * Updates only the errors of a processed entity */ diff --git a/plugins/catalog-backend/src/next/processing/types.ts b/plugins/catalog-backend/src/next/processing/types.ts index 59a10c084f..8b533714ce 100644 --- a/plugins/catalog-backend/src/next/processing/types.ts +++ b/plugins/catalog-backend/src/next/processing/types.ts @@ -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[];