feat(reporter): aggregate reports from multi-distributed runners

Zooz · Mar 3, 2019 · 0d7b132 · 0d7b132
1 parent 67b9d22
commit 0d7b132
Show file tree

Hide file tree

Showing 16 changed files with 251 additions and 53 deletions.
diff --git a/docs/swagger.yaml b/docs/swagger.yaml
@@ -948,6 +948,7 @@ definitions:
     - test_name
     - test_description
     - start_time
+    - runner_id
     properties:
       report_id:
         type: string
@@ -976,6 +977,9 @@ definitions:
       notes:
         type: string
         description: notes about the test
+      runner_id:
+        type: string
+        description: the id of the runner that created the report
 
   post_report_response:
     required:

diff --git a/src/database/cassandra-handler/init-scripts/3__create_reports_stats_table.cql b/src/database/cassandra-handler/init-scripts/3__create_reports_stats_table.cql
@@ -1,5 +1,5 @@
 CREATE TABLE IF NOT EXISTS reports_stats(
-container_id text,
+runner_id uuid,
 test_id uuid,
 report_id text,
 stats_id uuid,

diff --git a/src/database/cassandra-handler/init-scripts/4__create_reports_subscribers_table.cql b/src/database/cassandra-handler/init-scripts/4__create_reports_subscribers_table.cql
@@ -0,0 +1,6 @@
+CREATE TABLE IF NOT EXISTS report_subscribers(
+test_id uuid,
+report_id text,
+runner_id uuid,
+stage text,
+PRIMARY KEY (test_id, report_id, runner_id));
diff --git a/...t-scripts/4__create_last_reports_view.cql → ...t-scripts/5__create_last_reports_view.cql b/...t-scripts/4__create_last_reports_view.cql → ...t-scripts/5__create_last_reports_view.cql
diff --git a/...it-scripts/5__create_token_activities.cql → ...it-scripts/6__create_token_activities.cql b/...it-scripts/5__create_token_activities.cql → ...it-scripts/6__create_token_activities.cql
diff --git a/...dra-handler/init-scripts/6__dsl_table.cql → ...dra-handler/init-scripts/7__dsl_table.cql b/...dra-handler/init-scripts/6__dsl_table.cql → ...dra-handler/init-scripts/7__dsl_table.cql
diff --git a/src/reports/models/database/cassandra/cassandraConnector.js b/src/reports/models/database/cassandra/cassandraConnector.js
@@ -9,8 +9,11 @@ const UPDATE_REPORT_SUMMARY = 'UPDATE reports_summary SET status=?, phase=?, las
 const GET_REPORT_SUMMARY = 'SELECT * FROM reports_summary WHERE test_id=? AND report_id=? AND report_type=?';
 const GET_REPORTS_SUMMARIES = 'SELECT * FROM reports_summary WHERE test_id=? AND report_type=?';
 const GET_LAST_SUMMARIES = 'SELECT * FROM last_reports LIMIT ?';
-const INSERT_REPORT_STATS = 'INSERT INTO reports_stats(container_id, test_id, report_id, stats_id, stats_time, phase_index, phase_status, data) values(?,?,?,?,?,?,?,?)';
+const INSERT_REPORT_STATS = 'INSERT INTO reports_stats(runner_id, test_id, report_id, stats_id, stats_time, phase_index, phase_status, data) values(?,?,?,?,?,?,?,?)';
 const GET_REPORT_STATS = 'SELECT * FROM reports_stats WHERE test_id=? AND report_id=?';
+const SUBSCRIBE_RUNNER = 'INSERT INTO report_subscribers(test_id, report_id, runner_id, stage) values(?,?,?,?)';
+const UPDATE_SUBSCRIBERS = 'UPDATE report_subscribers SET stage=? WHERE test_id=? AND report_id=? AND runner_id=?';
+const GET_REPORT_SUBSCRIBERS = 'SELECT * FROM report_subscribers WHERE test_id=? AND report_id=?';
 
 module.exports = {
     init,
@@ -20,7 +23,9 @@ module.exports = {
     getReports,
     getLastReports,
     insertStats,
-    getStats
+    getStats,
+    subscribeRunner,
+    updateSubscribers
 };
 
 let queryOptions = {
@@ -48,24 +53,24 @@ function updateReport(testId, reportId, status, phaseIndex, lastStats, endTime)
 function getReport(testId, reportId) {
     let params;
     params = [testId, reportId, 'basic'];
-    return executeQuery(GET_REPORT_SUMMARY, params, queryOptions);
+    return getReportsAndParse(GET_REPORT_SUMMARY, params, queryOptions);
 }
 
 function getReports(testId) {
     let params;
     params = [testId, 'basic'];
-    return executeQuery(GET_REPORTS_SUMMARIES, params, queryOptions);
+    return getReportsAndParse(GET_REPORTS_SUMMARIES, params, queryOptions);
 }
 
 function getLastReports(limit) {
     let params;
     params = [limit];
-    return executeQuery(GET_LAST_SUMMARIES, params, queryOptions);
+    return getReportsAndParse(GET_LAST_SUMMARIES, params, queryOptions);
 }
 
-function insertStats(containerId, testId, reportId, statId, statsTime, phaseIndex, phaseStatus, data) {
+function insertStats(runnerId, testId, reportId, statId, statsTime, phaseIndex, phaseStatus, data) {
     let params;
-    params = [containerId, testId, reportId, statId, statsTime, phaseIndex, phaseStatus, data];
+    params = [runnerId, testId, reportId, statId, statsTime, phaseIndex, phaseStatus, data];
     return executeQuery(INSERT_REPORT_STATS, params, queryOptions);
 }
 
@@ -75,6 +80,24 @@ function getStats(testId, reportId) {
     return executeQuery(GET_REPORT_STATS, params, queryOptions);
 }
 
+function subscribeRunner(testId, reportId, runnerId) {
+    let params;
+    params = [testId, reportId, runnerId, 'initializing'];
+    return executeQuery(SUBSCRIBE_RUNNER, params, queryOptions);
+}
+
+function updateSubscribers(testId, reportId, runnerId, stage) {
+    let params;
+    params = [stage, testId, reportId, runnerId];
+    return executeQuery(UPDATE_SUBSCRIBERS, params, queryOptions);
+}
+
+function getReportSubscribers(testId, reportId) {
+    let params;
+    params = [testId, reportId];
+    return executeQuery(GET_REPORT_SUBSCRIBERS, params, queryOptions);
+}
+
 function executeQuery(query, params, queryOptions) {
     return client.execute(query, params, queryOptions).then((result) => {
         logger.trace('Query result', {
@@ -88,3 +111,22 @@ function executeQuery(query, params, queryOptions) {
         return Promise.reject(new Error('Error occurred in communication with cassandra'));
     });
 }
+
+async function getReportsAndParse(query, params, queryOptions) {
+    let subscribers, report;
+    const reports = await executeQuery(query, params, queryOptions);
+
+    for (let reportIndex = 0; reportIndex < reports.length; reportIndex++) {
+        report = reports[reportIndex];
+        subscribers = await getReportSubscribers(report.test_id, report.report_id);
+        subscribers = subscribers.map((subscriber) => {
+            return {
+                'runner_id': subscriber.runner_id,
+                'stage': subscriber.stage
+            };
+        });
+        report.subscribers = subscribers;
+    }
+
+    return reports;
+}
diff --git a/src/reports/models/database/sequelize/sequelizeConnector.js b/src/reports/models/database/sequelize/sequelizeConnector.js
@@ -11,7 +11,9 @@ module.exports = {
     getReport,
     getReports,
     getLastReports,
-    getStats
+    getStats,
+    subscribeRunner,
+    updateSubscribers
 };
 
 async function init(sequlizeClient) {
@@ -35,16 +37,17 @@ async function insertReport(testId, revisionId, reportId, jobId, testType, start
         notes: notes || '',
         phase: '0',
         status: 'initialized',
-        test_configuration: testConfiguration
+        test_configuration: testConfiguration,
+        runners_subscribed: []
     };
 
     return report.findOrCreate({ where: { report_id: reportId }, defaults: params });
 }
 
-async function insertStats(containerId, testId, reportId, statsId, statsTime, phaseIndex, phaseStatus, data) {
+async function insertStats(runnerId, testId, reportId, statsId, statsTime, phaseIndex, phaseStatus, data) {
     const stats = client.model('stats');
     const params = {
-        container_id: containerId,
+        runner_id: runnerId,
         report_id: reportId,
         test_id: testId,
         stats_id: statsId,
@@ -74,18 +77,72 @@ async function updateReport(testId, reportId, status, phaseIndex, lastStats, end
     }, options);
 }
 
+async function subscribeRunner(testId, reportId, runnerId) {
+    try {
+        const newSubscriber = {
+            runner_id: runnerId,
+            stage: 'initializing'
+        };
+
+        const reportModel = client.model('report');
+        const options = {
+            where: {
+                test_id: testId,
+                report_id: reportId
+            }
+        };
+
+        let report = await reportModel.findAll(options);
+        report = report[0];
+
+        await report.createSubscriber(newSubscriber);
+    } catch (e) {
+        console.log(e, `Failed to subscribe runner ${runnerId}`);
+    }
+}
+
+async function updateSubscribers(testId, reportId, runnerId, stage) {
+    const reportModel = client.model('report');
+    const getReportOptions = {
+        where: {
+            test_id: testId,
+            report_id: reportId
+        }
+    };
+
+    let report = await reportModel.findAll(getReportOptions);
+    report = report[0];
+
+    const subscribers = await report.getSubscribers();
+    const subscriberToUpdate = await subscribers.find((subscriber) => {
+        return subscriber.dataValues.runner_id === runnerId;
+    });
+
+    await subscriberToUpdate.set('stage', stage);
+    return subscriberToUpdate.save();
+}
+
 async function getReportsAndParse(query) {
     const report = client.model('report');
 
     let options = {
-        attributes: { exclude: ['updated_at', 'created_at'] }
+        attributes: { exclude: ['updated_at', 'created_at'] },
+        include: [report.subscriber]
     };
 
     Object.assign(options, query);
 
     const allReportsRawResponse = await report.findAll(options);
     let allReports = allReportsRawResponse.map(rawReport => rawReport.dataValues);
 
+    allReports.forEach(report => {
+        report.subscribers = report.subscribers.map((sqlJob) => {
+            return {
+                runner_id: sqlJob.dataValues.runner_id,
+                stage: sqlJob.dataValues.stage
+            };
+        });
+    });
     return allReports;
 }
 
@@ -139,8 +196,8 @@ async function initSchemas() {
         report_id: {
             type: Sequelize.DataTypes.STRING
         },
-        container_id: {
-            type: Sequelize.DataTypes.STRING
+        runner_id: {
+            type: Sequelize.DataTypes.UUID
         },
         stats_time: {
             type: Sequelize.DataTypes.DATE
@@ -156,6 +213,16 @@ async function initSchemas() {
         }
     });
 
+    const subscriber = client.define('subscriber', {
+        runner_id: {
+            type: Sequelize.DataTypes.STRING,
+            primaryKey: true
+        },
+        stage: {
+            type: Sequelize.DataTypes.STRING
+        }
+    });
+
     const report = client.define('report', {
         report_id: {
             type: Sequelize.DataTypes.STRING,
@@ -205,6 +272,8 @@ async function initSchemas() {
         }
     });
 
+    report.subscriber = report.hasMany(subscriber);
     await report.sync();
     await stats.sync();
+    await subscriber.sync();
 }
diff --git a/src/reports/models/databaseConnector.js b/src/reports/models/databaseConnector.js
@@ -12,15 +12,17 @@ module.exports = {
     getReport,
     getReports,
     getLastReports,
-    getStats
+    getStats,
+    subscribeRunner,
+    updateSubscribers
 };
 
 function insertReport(testId, revisionId, reportId, jobId, testType, startTime, testName, testDescription, testConfiguration, notes) {
     return databaseConnector.insertReport(testId, revisionId, reportId, jobId, testType, startTime, testName, testDescription, testConfiguration, notes);
 }
 
-function insertStats(containerId, testId, reportId, statId, statsTime, phaseIndex, phaseStatus, data) {
-    return databaseConnector.insertStats(containerId, testId, reportId, statId, statsTime, phaseIndex, phaseStatus, data);
+function insertStats(runnerId, testId, reportId, statId, statsTime, phaseIndex, phaseStatus, data) {
+    return databaseConnector.insertStats(runnerId, testId, reportId, statId, statsTime, phaseIndex, phaseStatus, data);
 }
 
 function updateReport(testId, reportId, status, phaseIndex, lastStats, endTime) {
@@ -41,4 +43,12 @@ function getReport(testId, reportId) {
 
 function getStats(testId, reportId) {
     return databaseConnector.getStats(testId, reportId);
+}
+
+function subscribeRunner(testId, reportId, runnerId) {
+    return databaseConnector.subscribeRunner(testId, reportId, runnerId);
+}
+
+function updateSubscribers(testId, reportId, runnerId, stage) {
+    return databaseConnector.updateSubscribers(testId, reportId, runnerId, stage);
 }