Clean up webhook events

by running a periodic garbage collector. This is a pattern we use for other resources.

fixes #12430

Author: AlexTugarev

components/gitpod-db/src/typeorm/webhook-event-db-impl.ts

@@ -51,4 +51,20 @@ export class WebhookEventDBImpl implements WebhookEventDB {
         query.limit(limit);
         return query.getMany();
     }
+
+    public async deleteOldEvents(ageInDays: number, limit?: number): Promise<void> {
+        const repo = await this.getRepo();
+        const d = new Date();
+        d.setDate(d.getDate() - ageInDays);
+        const expirationDate = d.toISOString();
+        const query = repo
+            .createQueryBuilder("event")
+            .update()
+            .set({ deleted: true })
+            .where("event.creationTime < :expirationDate", { expirationDate });
+        if (typeof limit === "number") {
+            query.limit(limit);
+        }
+        await query.execute();
+    }
 }

components/gitpod-db/src/webhook-event-db.ts

@@ -11,4 +11,5 @@ export interface WebhookEventDB {
     createEvent(parts: Omit<WebhookEvent, "id" | "creationTime">): Promise<WebhookEvent>;
     updateEvent(id: string, update: Partial<WebhookEvent>): Promise<void>;
     findByCloneUrl(cloneUrl: string, limit?: number): Promise<WebhookEvent[]>;
+    deleteOldEvents(ageInDays: number, limit?: number): Promise<void>;
 }

components/server/src/container-module.ts

@@ -115,6 +115,7 @@ import {
     getExperimentsClientForBackend,
 } from "@gitpod/gitpod-protocol/lib/experiments/configcat-server";
 import { VerificationService } from "./auth/verification-service";
+import { WebhookEventGarbageCollector } from "./projects/webhook-event-garbage-collector";
 
 export const productionContainerModule = new ContainerModule((bind, unbind, isBound, rebind) => {
     bind(Config).toConstantValue(ConfigFile.fromFile());
@@ -287,4 +288,6 @@ export const productionContainerModule = new ContainerModule((bind, unbind, isBo
         .inSingletonScope();
 
     bind(VerificationService).toSelf().inSingletonScope();
+
+    bind(WebhookEventGarbageCollector).toSelf().inSingletonScope();
 });

components/server/src/projects/webhook-event-garbage-collector.ts

@@ -0,0 +1,52 @@
+/**
+ * Copyright (c) 2022 Gitpod GmbH. All rights reserved.
+ * Licensed under the GNU Affero General Public License (AGPL).
+ * See License-AGPL.txt in the project root for license information.
+ */
+
+import { injectable, inject } from "inversify";
+import * as opentracing from "opentracing";
+import { DBWithTracing, TracedWorkspaceDB, WorkspaceDB, WebhookEventDB } from "@gitpod/gitpod-db/lib";
+import { Disposable } from "@gitpod/gitpod-protocol";
+import { ConsensusLeaderQorum } from "../consensus/consensus-leader-quorum";
+import { TraceContext } from "@gitpod/gitpod-protocol/lib/util/tracing";
+import { log } from "@gitpod/gitpod-protocol/lib/util/logging";
+import { repeat } from "@gitpod/gitpod-protocol/lib/util/repeat";
+
+@injectable()
+export class WebhookEventGarbageCollector {
+    static readonly GC_CYCLE_INTERVAL_SECONDS = 4 * 60; // every 6 minutes
+
+    @inject(WebhookEventDB) protected readonly db: WebhookEventDB;
+
+    @inject(ConsensusLeaderQorum) protected readonly leaderQuorum: ConsensusLeaderQorum;
+    @inject(TracedWorkspaceDB) protected readonly workspaceDB: DBWithTracing<WorkspaceDB>;
+
+    public async start(intervalSeconds?: number): Promise<Disposable> {
+        const intervalSecs = intervalSeconds || WebhookEventGarbageCollector.GC_CYCLE_INTERVAL_SECONDS;
+        return repeat(async () => {
+            try {
+                if (await this.leaderQuorum.areWeLeader()) {
+                    await this.collectObsoleteWebhookEvents();
+                }
+            } catch (err) {
+                log.error("webhook event garbage collector", err);
+            }
+        }, intervalSecs * 1000);
+    }
+
+    protected async collectObsoleteWebhookEvents() {
+        const span = opentracing.globalTracer().startSpan("collectObsoleteWebhookEvents");
+        log.debug("webhook-event-gc: start collecting...");
+        try {
+            await this.db.deleteOldEvents(10 /* days */, 600 /* limit per run */);
+            log.debug("webhook-event-gc: done collecting.");
+        } catch (err) {
+            TraceContext.setError({ span }, err);
+            log.error("webhook-event-gc: error collecting webhook events: ", err);
+            throw err;
+        } finally {
+            span.finish();
+        }
+    }
+}

components/server/src/server.ts

@@ -48,6 +48,7 @@ import { DebugApp } from "@gitpod/gitpod-protocol/lib/util/debug-app";
 import { LocalMessageBroker } from "./messaging/local-message-broker";
 import { WsConnectionHandler } from "./express/ws-connection-handler";
 import { InstallationAdminController } from "./installation-admin/installation-admin-controller";
+import { WebhookEventGarbageCollector } from "./projects/webhook-event-garbage-collector";
 
 @injectable()
 export class Server<C extends GitpodClient, S extends GitpodServer> {
@@ -75,6 +76,7 @@ export class Server<C extends GitpodClient, S extends GitpodServer> {
     @inject(OneTimeSecretServer) protected readonly oneTimeSecretServer: OneTimeSecretServer;
 
     @inject(PeriodicDbDeleter) protected readonly periodicDbDeleter: PeriodicDbDeleter;
+    @inject(WebhookEventGarbageCollector) protected readonly webhookEventGarbageCollector: WebhookEventGarbageCollector;
 
     @inject(BearerAuth) protected readonly bearerAuth: BearerAuth;
 
@@ -269,6 +271,11 @@ export class Server<C extends GitpodClient, S extends GitpodServer> {
         // Start DB updater
         this.startDbDeleter().catch((err) => log.error("starting DB deleter", err));
 
+        // Start WebhookEvent GC
+        this.webhookEventGarbageCollector
+            .start()
+            .catch((err) => log.error("webhook-event-gc: error during startup", err));
+
         this.app = app;
         log.info("server initialized.");
     }