[O2B-1564] Ccreate background GAQ summary worker

docker-compose.dev.yml

@@ -10,6 +10,10 @@ services:
       JWT_SECRET: BOOKKEEPING-DEV
       GRPC_INTERNAL_ORIGIN: '[::]:4001'
       GRPC_AUTHENTICATED_ORIGIN: '[::]:4002'
+      GAQ_ENABLE_RECALCULATION: "True"
+      GAQ_RECALCULATION_PERIOD: 10000
+      GAQ_RECALCULATION_MIN_BATCH_SIZE: 5
+      GAQ_RECALCULATION_MAX_BATCH_SIZE: 50
     ports:
       - "4000:4000"
       - "4001:4001"

docker-compose.test.yml

@@ -10,6 +10,10 @@ services:
       JWT_SECRET: BOOKKEEPING-TEST-SUITE
       PAGE_ITEMS_LIMIT: 100
       CCDB_SYNCHRONIZATION_PERIOD: 3153600000000 # 100y in milliseconds, to be sure all the runs are included when testing sync
+      GAQ_ENABLE_RECALCULATION: "True"
+      GAQ_RECALCULATION_PERIOD: 1000
+      GAQ_RECALCULATION_MIN_BATCH_SIZE: 1
+      GAQ_RECALCULATION_MAX_BATCH_SIZE: 1
     restart: "no"
 
   database:

lib/application.js

@@ -15,7 +15,7 @@
 const { webUiServer } = require('./server');
 const { GRPCConfig, ServicesConfig } = require('./config');
 
-const { userCertificate, monalisa: monalisaConfig, ccdb: ccdbConfig, enableHousekeeping } = ServicesConfig;
+const { userCertificate, monalisa: monalisaConfig, ccdb: ccdbConfig, gaq: gaqConfig, enableHousekeeping } = ServicesConfig;
 const { handleLostRunsAndEnvironments } = require('./server/services/housekeeping/handleLostRunsAndEnvironments.js');
 const { isInTestMode } = require('./utilities/env-utils.js');
 const { ScheduledProcessesManager } = require('./server/services/ScheduledProcessesManager.js');
@@ -27,6 +27,7 @@
 const { environmentService } = require('./server/services/environment/EnvironmentService.js');
 const { runService } = require('./server/services/run/RunService.js');
 const { CcdbSynchronizer } = require('./server/externalServicesSynchronization/ccdb/CcdbSynchronizer.js');
+const { gaqWorker } = require('./server/services/gaq/GaqWorker.js');
 const { promises: fs } = require('fs');
 const { MonAlisaClient } = require('./server/externalServicesSynchronization/monalisa/MonAlisaClient.js');
 const https = require('https');
@@ -131,6 +132,16 @@
                     },
                 );
             }
+
+            if (gaqConfig.enableRecalculation) {
+                this.scheduledProcessesManager.schedule(
+                    () => gaqWorker.recalculateGaqSummaries(gaqConfig.minBatchSize, gaqConfig.maxBatchSize),
+                    {
+                        wait: 10 * 1000,
+                        every: gaqConfig.recalculationPeriod,
+                    },
+                );
+            }
         } catch (error) {
             this._logger.errorMessage(`Error while starting: ${error}`);
             return this.stop();
@@ -156,7 +167,7 @@
     }
 
     /**
      * Housekeeping method, it wraps @see handleLostRunsAndEnvironments and logs its results
      *
      * @param {Agent} httpAgent agent to be used by the HTTP client
      * @return {Promise<void>} promise

lib/config/services.js

@@ -27,6 +27,25 @@
     CCDB_RUN_INFO_URL,
 } = process.env ?? {};
 
+/**
+ * Parse a positive integer env var, falling back to the default and warning if the value is invalid
+ *
+ * @param {string|undefined} raw the raw env var value
+ * @param {number} defaultValue value to use when raw is unset or invalid
+ * @param {string} name env var name (for the warning message)
+ * @return {number} the parsed value or the default
+ */
+const parsePositiveInt = (raw, defaultValue, name) => {
+    if (raw === undefined) return defaultValue;
+    const parsed = Number(raw);
+    if (!Number.isInteger(parsed) || parsed <= 0) {
+        // eslint-disable-next-line no-console
+        console.warn(`Invalid ${name}=${JSON.stringify(raw)}; falling back to ${defaultValue}`);
+        return defaultValue;
+    }
+    return parsed;
+};
+
 exports.services = {
     enableHousekeeping: process.env?.ENABLE_HOUSEKEEPING?.toLowerCase() === 'true',
     userCertificate: {
@@ -68,4 +87,11 @@
         synchronizationPeriod: Number(CCDB_SYNCHRONIZATION_PERIOD) || 24 * 60 * 60 * 1000, // 1d in milliseconds
         runInfoUrl: CCDB_RUN_INFO_URL,
     },
+
+    gaq: {
+        enableRecalculation: process.env?.GAQ_ENABLE_RECALCULATION?.toLowerCase() === 'true',
+        recalculationPeriod: parsePositiveInt(process.env?.GAQ_RECALCULATION_PERIOD, 30 * 1000, 'GAQ_RECALCULATION_PERIOD'), // 30s default
+        minBatchSize: parsePositiveInt(process.env?.GAQ_RECALCULATION_MIN_BATCH_SIZE, 1, 'GAQ_RECALCULATION_MIN_BATCH_SIZE'),
+        maxBatchSize: parsePositiveInt(process.env?.GAQ_RECALCULATION_MAX_BATCH_SIZE, 100, 'GAQ_RECALCULATION_MAX_BATCH_SIZE'),
+    },
 };

lib/database/adapters/GaqSummaryAdapter.js

@@ -0,0 +1,62 @@
+/**
+ * @license
+ * Copyright CERN and copyright holders of ALICE O2. This software is
+ * distributed under the terms of the GNU General Public License v3 (GPL
+ * Version 3), copied verbatim in the file "COPYING".
+ *
+ * See http://alice-o2.web.cern.ch/license for full licensing information.
+ *
+ * In applying this license CERN does not waive the privileges and immunities
+ * granted to it by virtue of its status as an Intergovernmental Organization
+ * or submit itself to any jurisdiction.
+ */
+
+/**
+ * GaqSummaryAdapter
+ */
+class GaqSummaryAdapter {
+    /**
+     * Constructor
+     */
+    constructor() {
+        this.toEntity = this.toEntity.bind(this);
+    }
+
+    /**
+     * Converts the given database object to an entity object.
+     *
+     * @param {SequelizeGaqSummary} databaseObject Object to convert.
+     * @returns {GaqSummary} Converted entity object.
+     */
+    toEntity(databaseObject) {
+        const {
+            dataPassId,
+            runNumber,
+            badRunCoverage,
+            explicitlyNotBadRunCoverage,
+            mcReproducibleCoverage,
+            missingVerificationsCount,
+            undefinedQualityPeriodsCount,
+            notComputable,
+            invalidatedAt,
+            createdAt,
+            updatedAt,
+        } = databaseObject;
+
+        return {
+            dataPassId,
+            runNumber,
+            badRunCoverage,
+            explicitlyNotBadRunCoverage,
+            mcReproducibleCoverage,
+            missingVerificationsCount,
+            undefinedQualityPeriodsCount,
+            notComputable,
+            invalidatedAt,
+            createdAt,
+            updatedAt,
+        };
+    }
+}
+
+module.exports = { GaqSummaryAdapter };

lib/database/adapters/index.js

@@ -27,6 +27,7 @@ const EorReasonAdapter = require('./EorReasonAdapter');
 const FlpRoleAdapter = require('./FlpRoleAdapter');
 const { HostAdapter } = require('./HostAdapter.js');
 const { GaqDetectorAdapter } = require('./GaqDetectorAdapter.js');
+const { GaqSummaryAdapter } = require('./GaqSummaryAdapter.js');
 const { LhcFillAdapter } = require('./LhcFillAdapter.js');
 const { LhcFillStatisticsAdapter } = require('./LhcFillStatisticsAdapter.js');
 const LhcPeriodAdapter = require('./LhcPeriodAdapter');
@@ -63,6 +64,7 @@ const environmentHistoryItemAdapter = new EnvironmentHistoryItemAdapter();
 const eorReasonAdapter = new EorReasonAdapter();
 const flpRoleAdapter = new FlpRoleAdapter();
 const gaqDetectorAdapter = new GaqDetectorAdapter();
+const gaqSummaryAdapter = new GaqSummaryAdapter();
 const hostAdapter = new HostAdapter();
 const lhcFillAdapter = new LhcFillAdapter();
 const lhcFillStatisticsAdapter = new LhcFillStatisticsAdapter();
@@ -159,6 +161,7 @@ module.exports = {
     eorReasonAdapter,
     flpRoleAdapter,
     gaqDetectorAdapter,
+    gaqSummaryAdapter,
     hostAdapter,
     lhcFillAdapter,
     lhcFillStatisticsAdapter,

lib/database/migrations/v1/20260223120000-create-gaq-summary-tables.js

@@ -0,0 +1,71 @@
+'use strict';
+
+/** @type {import('sequelize-cli').Migration} */
+module.exports = {
+    up: async (queryInterface, Sequelize) => queryInterface.sequelize.transaction(async (transaction) => {
+        await queryInterface.createTable('gaq_summaries', {
+            data_pass_id: {
+                type: Sequelize.INTEGER,
+                primaryKey: true,
+                allowNull: false,
+                references: {
+                    model: 'data_passes',
+                    key: 'id',
+                },
+            },
+            run_number: {
+                type: Sequelize.INTEGER,
+                primaryKey: true,
+                allowNull: false,
+                references: {
+                    model: 'runs',
+                    key: 'run_number',
+                },
+            },
+            bad_run_coverage: {
+                type: Sequelize.FLOAT,
+            },
+            explicitly_not_bad_run_coverage: {
+                type: Sequelize.FLOAT,
+            },
+            mc_reproducible_coverage: {
+                type: Sequelize.FLOAT,
+            },
+            missing_verifications_count: {
+                type: Sequelize.INTEGER,
+            },
+            undefined_quality_periods_count: {
+                type: Sequelize.INTEGER,
+            },
+            not_computable: {
+                type: Sequelize.BOOLEAN,
+                allowNull: false,
+                defaultValue: false,
+            },
+            invalidated_at: {
+                type: Sequelize.DATE(3),
+                allowNull: true,
+                defaultValue: null,
+            },
+            created_at: {
+                type: Sequelize.DATE(3),
+                allowNull: false,
+                defaultValue: Sequelize.literal('CURRENT_TIMESTAMP(3)'),
+            },
+            updated_at: {
+                type: Sequelize.DATE(3),
+                allowNull: false,
+                defaultValue: Sequelize.literal('CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3)'),
+            },
+        }, { transaction });
+
+        await queryInterface.addIndex('gaq_summaries', {
+            name: 'gaq_summaries_invalidated_at_idx',
+            fields: ['invalidated_at'],
+        }, { transaction });
+    }),
+
+    down: async (queryInterface) => queryInterface.sequelize.transaction(async (transaction) => {
+        await queryInterface.dropTable('gaq_summaries', { transaction });
+    }),
+};

lib/database/models/gaqSummary.js

@@ -0,0 +1,59 @@
+/**
+ * @license
+ * Copyright CERN and copyright holders of ALICE O2. This software is
+ * distributed under the terms of the GNU General Public License v3 (GPL
+ * Version 3), copied verbatim in the file "COPYING".
+ *
+ * See http://alice-o2.web.cern.ch/license for full licensing information.
+ *
+ * In applying this license CERN does not waive the privileges and immunities
+ * granted to it by virtue of its status as an Intergovernmental Organization
+ * or submit itself to any jurisdiction.
+ */
+
+module.exports = (sequelize) => {
+    const Sequelize = require('sequelize');
+
+    const GaqSummary = sequelize.define('GaqSummary', {
+        dataPassId: {
+            type: Sequelize.INTEGER,
+            primaryKey: true,
+        },
+        runNumber: {
+            type: Sequelize.INTEGER,
+            primaryKey: true,
+        },
+        badRunCoverage: {
+            type: Sequelize.FLOAT,
+        },
+        explicitlyNotBadRunCoverage: {
+            type: Sequelize.FLOAT,
+        },
+        mcReproducibleCoverage: {
+            type: Sequelize.FLOAT,
+        },
+        missingVerificationsCount: {
+            type: Sequelize.INTEGER,
+        },
+        undefinedQualityPeriodsCount: {
+            type: Sequelize.INTEGER,
+        },
+        notComputable: {
+            type: Sequelize.BOOLEAN,
+            allowNull: false,
+            defaultValue: false,
+        },
+        invalidatedAt: {
+            type: Sequelize.DATE(3),
+        },
+    }, { tableName: 'gaq_summaries' });
+
+    GaqSummary.removeAttribute('id');
+
+    GaqSummary.associate = (models) => {
+        GaqSummary.belongsTo(models.Run, { foreignKey: 'runNumber', as: 'run' });
+        GaqSummary.belongsTo(models.DataPass, { foreignKey: 'dataPassId', as: 'dataPass' });
+    };
+
+    return GaqSummary;
+};

lib/database/models/index.js

@@ -27,6 +27,7 @@ const EorReason = require('./eorreason');
 const EpnRoleSession = require('./epnrolesession');
 const FlpRole = require('./flprole');
 const GaqDetector = require('./gaqDetector.js');
+const GaqSummary = require('./gaqSummary.js');
 const Host = require('./host.js');
 const LhcFill = require('./lhcFill');
 const LhcFillStatistics = require('./lhcFillStatistics.js');
@@ -66,6 +67,7 @@ module.exports = (sequelize) => {
         EpnRoleSessionkey: EpnRoleSession(sequelize),
         FlpRole: FlpRole(sequelize),
         GaqDetector: GaqDetector(sequelize),
+        GaqSummary: GaqSummary(sequelize),
         Host: Host(sequelize),
         LhcFill: LhcFill(sequelize),
         LhcFillStatistics: LhcFillStatistics(sequelize),

lib/database/repositories/GaqSummaryRepository.js

@@ -0,0 +1,55 @@
+/**
+ * @license
+ * Copyright CERN and copyright holders of ALICE O2. This software is
+ * distributed under the terms of the GNU General Public License v3 (GPL
+ * Version 3), copied verbatim in the file "COPYING".
+ *
+ * See http://alice-o2.web.cern.ch/license for full licensing information.
+ *
+ * In applying this license CERN does not waive the privileges and immunities
+ * granted to it by virtue of its status as an Intergovernmental Organization
+ * or submit itself to any jurisdiction.
+ */
+
+const {
+    models: {
+        GaqSummary,
+    },
+} = require('..');
+const Repository = require('./Repository');
+
+/**
+ * GaqSummary repository
+ */
+class GaqSummaryRepository extends Repository {
+    /**
+     * Creates a new `GaqSummaryRepository` instance.
+     */
+    constructor() {
+        super(GaqSummary);
+    }
+
+    /**
+     * Mark the summary for a given (dataPassId, runNumber) as invalidated, creating the row if it does not yet exist
+     *
+     * @param {number} dataPassId data pass id
+     * @param {number} runNumber run number
+     * @return {Promise<void>} resolves once the summary is invalidated
+     */
+    async invalidate(dataPassId, runNumber) {
+        await this.upsert({ dataPassId, runNumber, invalidatedAt: new Date() });
+    }
+
+    /**
+     * Mark a list of summaries as invalidated in parallel
+     *
+     * @param {{ dataPassId: number, runNumber: number }[]} pairs the (dataPassId, runNumber) pairs to invalidate
+     * @return {Promise<void>} resolves once all summaries are invalidated
+     */
+    async invalidateMany(pairs) {
+        const invalidatedAt = new Date();
+        await Promise.all(pairs.map(({ dataPassId, runNumber }) => this.upsert({ dataPassId, runNumber, invalidatedAt })));
+    }
+}
+
+module.exports = new GaqSummaryRepository();