Merge pull request #6827 from backstage/freben/refresh_hash

Avoid duplicate work by comparing previous processing rounds with the next

.changeset/cuddly-flowers-film.md

@@ -0,0 +1,5 @@
+---
+'@backstage/plugin-catalog-backend': patch
+---
+
+Avoid duplicate work by comparing previous processing rounds with the next

plugins/catalog-backend/knexfile.js

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2021 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// This file makes it possible to run "yarn knex migrate:make some_file_name"
+// to assist in making new migrations
+module.exports = {
+  client: 'sqlite3',
+  connection: ':memory:',
+  useNullAsDefault: true,
+  migrations: {
+    directory: './migrations',
+  },
+};

plugins/catalog-backend/migrations/20210813143113_add_refresh_state_hash.js

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2021 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// @ts-check
+
+/**
+ * @param {import('knex').Knex} knex
+ */
+exports.up = async function up(knex) {
+  await knex.schema.alterTable('refresh_state', table => {
+    table
+      .text('result_hash')
+      .nullable()
+      .comment(
+        'A hash of the processed contents, used to avoid duplicate work',
+      );
+  });
+};
+
+/**
+ * @param {import('knex').Knex} knex
+ */
+exports.down = async function down(knex) {
+  await knex.schema.alterTable('refresh_state', table => {
+    table.dropColumn('result_hash');
+  });
+};

plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts

@@ -15,6 +15,7 @@
  */
 
 import { getVoidLogger } from '@backstage/backend-common';
+import { Hash } from 'crypto';
 import waitForExpect from 'wait-for-expect';
 import { DefaultProcessingDatabase } from './database/DefaultProcessingDatabase';
 import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine';
@@ -33,6 +34,10 @@ describe('DefaultCatalogProcessingEngine', () => {
   const stitcher = {
     stitch: jest.fn(),
   } as unknown as jest.Mocked<Stitcher>;
+  const hash = {
+    update: () => hash,
+    digest: jest.fn(),
+  } as unknown as jest.Mocked<Hash>;
 
   beforeEach(() => {
     jest.resetAllMocks();
@@ -57,6 +62,7 @@ describe('DefaultCatalogProcessingEngine', () => {
       db,
       orchestrator,
       stitcher,
+      () => hash,
     );
 
     db.transaction.mockImplementation(cb => cb((() => {}) as any));
@@ -76,6 +82,7 @@ describe('DefaultCatalogProcessingEngine', () => {
               kind: 'Location',
               metadata: { name: 'test' },
             },
+            resultHash: '',
             state: new Map(),
             nextUpdateAt: '',
             lastDiscoveryAt: '',
@@ -117,6 +124,7 @@ describe('DefaultCatalogProcessingEngine', () => {
       db,
       orchestrator,
       stitcher,
+      () => hash,
     );
 
     db.transaction.mockImplementation(cb => cb((() => {}) as any));
@@ -137,6 +145,7 @@ describe('DefaultCatalogProcessingEngine', () => {
               kind: 'Location',
               metadata: { name: 'test' },
             },
+            resultHash: '',
             state: new Map(),
             nextUpdateAt: '',
             lastDiscoveryAt: '',
@@ -158,4 +167,71 @@ describe('DefaultCatalogProcessingEngine', () => {
     });
     await engine.stop();
   });
+
+  it('runs fully when hash mismatches, early-outs when hash matches', async () => {
+    const entity = {
+      apiVersion: '1',
+      kind: 'Location',
+      metadata: { name: 'test' },
+    };
+
+    const refreshState = {
+      id: '',
+      entityRef: '',
+      unprocessedEntity: entity,
+      resultHash: 'the matching hash',
+      state: new Map(),
+      nextUpdateAt: '',
+      lastDiscoveryAt: '',
+    };
+
+    hash.digest.mockReturnValue('the matching hash');
+
+    orchestrator.process.mockResolvedValue({
+      ok: true,
+      completedEntity: entity,
+      relations: [],
+      errors: [],
+      deferredEntities: [],
+      state: new Map(),
+    });
+
+    const engine = new DefaultCatalogProcessingEngine(
+      getVoidLogger(),
+      [],
+      db,
+      orchestrator,
+      stitcher,
+      () => hash,
+    );
+
+    db.transaction.mockImplementation(cb => cb((() => {}) as any));
+
+    db.getProcessableEntities
+      .mockResolvedValueOnce({
+        items: [{ ...refreshState, resultHash: 'NOT RIGHT' }],
+      })
+      .mockResolvedValue({ items: [] });
+
+    await engine.start();
+
+    await waitForExpect(() => {
+      expect(orchestrator.process).toBeCalledTimes(1);
+      expect(hash.digest).toBeCalledTimes(1);
+      expect(db.updateProcessedEntity).toBeCalledTimes(1);
+    });
+
+    db.getProcessableEntities
+      .mockReset()
+      .mockResolvedValueOnce({ items: [refreshState] })
+      .mockResolvedValue({ items: [] });
+
+    await waitForExpect(() => {
+      expect(orchestrator.process).toBeCalledTimes(2);
+      expect(hash.digest).toBeCalledTimes(2);
+      expect(db.updateProcessedEntity).toBeCalledTimes(1);
+    });
+
+    await engine.stop();
+  });
 });

plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts

@@ -20,6 +20,8 @@ import {
   stringifyEntityRef,
 } from '@backstage/catalog-model';
 import { serializeError } from '@backstage/errors';
+import { Hash } from 'crypto';
+import stableStringify from 'fast-json-stable-stringify';
 import { Logger } from 'winston';
 import { ProcessingDatabase, RefreshStateItem } from './database/types';
 import { CatalogProcessingOrchestrator } from './processing/types';
@@ -89,6 +91,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
     private readonly processingDatabase: ProcessingDatabase,
     private readonly orchestrator: CatalogProcessingOrchestrator,
     private readonly stitcher: Stitcher,
+    private readonly createHash: () => Hash,
   ) {}
 
   async start() {
@@ -125,7 +128,14 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
       },
       processTask: async item => {
         try {
-          const { id, state, unprocessedEntity, entityRef, locationKey } = item;
+          const {
+            id,
+            state,
+            unprocessedEntity,
+            entityRef,
+            locationKey,
+            resultHash: previousResultHash,
+          } = item;
           const result = await this.orchestrator.process({
             entity: unprocessedEntity,
             state,
@@ -142,6 +152,23 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
             result.errors.map(e => serializeError(e)),
           );
 
+          let hashBuilder = this.createHash().update(errorsString);
+          if (result.ok) {
+            hashBuilder = hashBuilder
+              .update(stableStringify({ ...result.completedEntity }))
+              .update(stableStringify([...result.deferredEntities]))
+              .update(stableStringify([...result.relations]))
+              .update(stableStringify(Object.fromEntries(result.state)));
+          }
+
+          const resultHash = hashBuilder.digest('hex');
+          if (resultHash === previousResultHash) {
+            // If nothing changed in our produced outputs, we cannot have any
+            // significant effect on our surroundings; therefore, we just abort
+            // without any updates / stitching.
+            return;
+          }
+
           // If the result was marked as not OK, it signals that some part of the
           // processing pipeline threw an exception. This can happen both as part of
           // non-catastrophic things such as due to validation errors, as well as if
@@ -154,6 +181,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
               await this.processingDatabase.updateProcessedEntityErrors(tx, {
                 id,
                 errors: errorsString,
+                resultHash,
               });
             });
             await this.stitcher.stitch(
@@ -167,6 +195,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
             await this.processingDatabase.updateProcessedEntity(tx, {
               id,
               processedEntity: result.completedEntity,
+              resultHash,
               state: result.state,
               errors: errorsString,
               relations: result.relations,

plugins/catalog-backend/src/next/NextCatalogBuilder.ts

@@ -31,6 +31,7 @@ import {
 } from '@backstage/catalog-model';
 import { Config } from '@backstage/config';
 import { ScmIntegrations } from '@backstage/integration';
+import { createHash } from 'crypto';
 import lodash from 'lodash';
 import { Logger } from 'winston';
 import {
@@ -310,6 +311,7 @@ export class NextCatalogBuilder {
       processingDatabase,
       orchestrator,
       stitcher,
+      () => createHash('sha1'),
     );
 
     const locationsCatalog = new DatabaseLocationsCatalog(db);

plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts

@@ -19,8 +19,8 @@ import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils';
 import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
 import { JsonObject } from '@backstage/config';
 import { Knex } from 'knex';
-import { Logger } from 'winston';
 import * as uuid from 'uuid';
+import { Logger } from 'winston';
 import { DatabaseManager } from './DatabaseManager';
 import { DefaultProcessingDatabase } from './DefaultProcessingDatabase';
 import {
@@ -206,6 +206,7 @@ describe('Default Processing Database', () => {
             db.updateProcessedEntity(tx, {
               id,
               processedEntity,
+              resultHash: '',
               state: new Map<string, JsonObject>(),
               relations: [],
               deferredEntities: [],
@@ -224,6 +225,7 @@ describe('Default Processing Database', () => {
         const options = {
           id,
           processedEntity,
+          resultHash: '',
           state: new Map<string, JsonObject>(),
           relations: [],
           deferredEntities: [],
@@ -250,7 +252,11 @@ describe('Default Processing Database', () => {
 
         await db.transaction(tx =>
           expect(
-            db.updateProcessedEntity(tx, { ...options, locationKey: 'fail' }),
+            db.updateProcessedEntity(tx, {
+              ...options,
+              resultHash: '',
+              locationKey: 'fail',
+            }),
           ).rejects.toThrow(
             `Conflicting write of processing result for ${id} with location key 'fail'`,
           ),
@@ -280,6 +286,7 @@ describe('Default Processing Database', () => {
           db.updateProcessedEntity(tx, {
             id,
             processedEntity,
+            resultHash: '',
             state,
             relations: [],
             deferredEntities: [],
@@ -295,7 +302,9 @@ describe('Default Processing Database', () => {
         expect(entities[0].processed_entity).toEqual(
           JSON.stringify(processedEntity),
         );
-        expect(entities[0].cache).toEqual(JSON.stringify(state));
+        expect(entities[0].cache).toEqual(
+          JSON.stringify(Object.fromEntries(state)),
+        );
         expect(entities[0].errors).toEqual("['something broke']");
         expect(entities[0].location_key).toEqual('key');
       },
@@ -336,6 +345,7 @@ describe('Default Processing Database', () => {
           db.updateProcessedEntity(tx, {
             id,
             processedEntity,
+            resultHash: '',
             state: new Map<string, JsonObject>(),
             relations: relations,
             deferredEntities: [],
@@ -387,6 +397,7 @@ describe('Default Processing Database', () => {
           db.updateProcessedEntity(tx, {
             id,
             processedEntity,
+            resultHash: '',
             state: new Map<string, JsonObject>(),
             relations: [],
             deferredEntities,

plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts

@@ -60,6 +60,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
     const {
       id,
       processedEntity,
+      resultHash,
       state,
       errors,
       relations,
@@ -69,7 +70,8 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
     const refreshResult = await tx<DbRefreshStateRow>('refresh_state')
       .update({
         processed_entity: JSON.stringify(processedEntity),
-        cache: JSON.stringify(state),
+        result_hash: resultHash,
+        cache: JSON.stringify(Object.fromEntries(state || [])),
         errors,
         location_key: locationKey,
       })
@@ -122,11 +124,12 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
     options: UpdateProcessedEntityOptions,
   ): Promise<void> {
     const tx = txOpaque as Knex.Transaction;
-    const { id, errors } = options;
+    const { id, errors, resultHash } = options;
 
     await tx<DbRefreshStateRow>('refresh_state')
       .update({
         errors,
+        result_hash: resultHash,
       })
       .where('entity_id', id);
   }
@@ -492,6 +495,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
             processedEntity: i.processed_entity
               ? (JSON.parse(i.processed_entity) as Entity)
               : undefined,
+            resultHash: i.result_hash || '',
             nextUpdateAt: i.next_update_at,
             lastDiscoveryAt: i.last_discovery_at,
             state: i.cache

plugins/catalog-backend/src/next/database/tables.ts

@@ -25,6 +25,7 @@ export type DbRefreshStateRow = {
   entity_ref: string;
   unprocessed_entity: string;
   processed_entity?: string;
+  result_hash?: string;
   cache?: string;
   next_update_at: string;
   last_discovery_at: string; // remove?

plugins/catalog-backend/src/next/database/types.ts

@@ -34,6 +34,7 @@ export type AddUnprocessedEntitiesResult = {};
 export type UpdateProcessedEntityOptions = {
   id: string;
   processedEntity: Entity;
+  resultHash: string;
   state?: Map<string, JsonObject>;
   errors?: string;
   relations: EntityRelationSpec[];
@@ -44,6 +45,7 @@ export type UpdateProcessedEntityOptions = {
 export type UpdateProcessedEntityErrorsOptions = {
   id: string;
   errors?: string;
+  resultHash: string;
 };
 
 export type RefreshStateItem = {
@@ -51,6 +53,7 @@ export type RefreshStateItem = {
   entityRef: string;
   unprocessedEntity: Entity;
   processedEntity?: Entity;
+  resultHash: string;
   nextUpdateAt: string;
   lastDiscoveryAt: string; // remove?
   state: Map<string, JsonObject>;