Merge pull request #639 from Zooz/runjob-logics

feat: create job experiments on createJob logics

src/chaos-experiments/models/database/databaseConnector.js

@@ -5,8 +5,8 @@ module.exports = {
     getAllChaosExperiments,
     insertChaosExperiment,
     getChaosExperimentById,
-    getChaosExperimentsByIds,
     getChaosExperimentByName,
+    getChaosExperimentsByIds,
     deleteChaosExperiment,
     insertChaosJobExperiment,
     getChaosJobExperimentById,

src/jobs/models/jobExperimentsHandler.js

@@ -0,0 +1,67 @@
+
+const chaosExperimentsDbConnector = require('../../chaos-experiments/models/database/databaseConnector'),
+    { v4: uuid } = require('uuid'),
+    logger = require('../../common/logger');
+
+const SEC_TO_MS = 1000;
+const MIN_TO_MS = 60 * 1000;
+const HOUR_TO_MS = 60 * 1000;
+const DAY_TO_MS = 60 * 1000;
+const jobExperimentsIdToTimeout = new Map();
+
+async function setChaosExperimentsIfExist(jobId, jobExperiments) {
+    if (!jobExperiments) {
+        return;
+    }
+    try {
+        const baseTimestamp = Date.now();
+        const experimentIds = jobExperiments.map(experiment => experiment.experiment_id);
+        const experimentsFromDb = await chaosExperimentsDbConnector.getChaosExperimentsByIds(experimentIds);
+        await Promise.all(jobExperiments.map(async(experimentRequest) =>
+            await setSingleJobExperiment(experimentRequest, experimentsFromDb, baseTimestamp, jobId)
+        ));
+    } catch (error){
+        logger.error(error, `error while setting chaos experiments for job ${jobId}`);
+    }
+};
+
+async function setSingleJobExperiment(experimentRequest, experimentsFromDb, baseTimestamp, jobId) {
+    try {
+        const experiment = experimentsFromDb.find(e => e.id === experimentRequest.experiment_id);
+        const startTime = baseTimestamp + experimentRequest.start_after;
+        const endTime = startTime + convertDurationStringToMillisecond(experiment.kubeObject.spec.duration);
+        const jobExperimentId = uuid();
+        await chaosExperimentsDbConnector.insertChaosJobExperiment(jobExperimentId, jobId, experiment.id, startTime, endTime);
+        const kubeObject = experiment.kubeObject;
+        kubeObject.name = kubeObject.metadata.name.concat(`_${jobExperimentId}`);
+        const timeout = setTimeout(() => runChaosExperiment(kubeObject, jobId, jobExperimentId), experimentRequest.start_after);
+        jobExperimentsIdToTimeout.set(jobExperimentId, timeout);
+    } catch (error){
+        logger.error(error, `error while setting chaos experiment ${experimentRequest.experiment_id} for job ${jobId}`);
+    }
+}
+
+async function runChaosExperiment(kubeObject, jobId, jobExperimentId) {
+
+}
+
+function convertDurationStringToMillisecond(durationString) {
+    if (durationString.endsWith('s')){
+        return durationString.split('s')[0] * SEC_TO_MS;
+    }
+    if (durationString.endsWith('m')){
+        return durationString.split('m')[0] * MIN_TO_MS;
+    }
+    if (durationString.endsWith('h')){
+        return durationString.split('h')[0] * HOUR_TO_MS;
+    }
+    if (durationString.endsWith('d')){
+        return durationString.split('h')[0] * DAY_TO_MS;
+    }
+    return durationString.split('ms')[0];
+}
+
+module.exports = {
+    jobExperimentsIdToTimeout,
+    setChaosExperimentsIfExist
+};

src/jobs/models/jobManager.js

@@ -16,7 +16,8 @@ const logger = require('../../common/logger'),
     { STREAMING_EVENT_TYPES } = require('../../streaming/entities/common'),
     { CONFIG, CONTEXT_ID, JOB_TYPE_FUNCTIONAL_TEST } = require('../../common/consts'),
     generateError = require('../../common/generateError'),
-    { version: PREDATOR_VERSION } = require('../../../package.json');
+    { version: PREDATOR_VERSION } = require('../../../package.json'),
+    jobExperimentHandler = require('./jobExperimentsHandler');
 
 let jobConnector;
 const cronJobs = {};
@@ -363,6 +364,7 @@ async function runJob(job, configData) {
         report = await createReportForJob(test, job);
         const jobSpecificPlatformRequest = await createJobRequest(job.id, report.report_id, job, latestDockerImage, configData);
         await jobConnector.runJob(jobSpecificPlatformRequest, job);
+        await jobExperimentHandler.setChaosExperimentsIfExist(job.id, job.experiments);
     } catch (error) {
         logger.error({ id: job.id, error: error }, 'Unable to run scheduled job.');
         await failReport(report);

tests/unit-tests/chaos-experiments/models/chaosExperimentsManager-test.js

@@ -27,6 +27,9 @@ describe('Chaos experiments manager tests', function () {
         deleteStub = sandbox.stub(database, 'deleteChaosExperiment');
         updatedChaosExperimentStub = sandbox.stub(database, 'updateChaosExperiment');
     });
+    after(() => {
+        sandbox.restore();
+    });
 
     beforeEach(() => {
         sandbox.reset();

tests/unit-tests/jobs/models/jobExperimentsHandler-test.js

@@ -0,0 +1,98 @@
+const sinon = require('sinon'),
+    databaseConnector = require('../../../../src/chaos-experiments/models/database/databaseConnector'),
+    jobExperimentHandler = require('../../../../src/jobs/models/jobExperimentsHandler');
+;
+const { v4: uuid } = require('uuid');
+
+function generateExperiment(id = uuid()){
+    return {
+        id: id,
+        kubeObject: {
+            kind: 'PodChaos',
+            apiVersion: 'chaos-mesh.org/v1alpha1',
+            metadata: {
+                name: 'TestChaos'
+            },
+            spec: {
+                mode: 'all',
+                action: 'pod-kill',
+                duration: '1m'
+            }
+        }
+
+    };
+}
+describe('Job experiments handler tests', function () {
+    let databaseConnectorInsertStub;
+    let databaseConnectorGetStub;
+    let sandbox;
+
+    before(() => {
+        sandbox = sinon.sandbox.create();
+        databaseConnectorInsertStub = sandbox.stub(databaseConnector, 'insertChaosJobExperiment');
+        databaseConnectorGetStub = sandbox.stub(databaseConnector, 'getChaosExperimentsByIds');
+    });
+    beforeEach(async () => {
+        sandbox.reset();
+    });
+
+    after(() => {
+        sandbox.restore();
+    });
+
+    it('set chaos experiments with 2 experiments', async () => {
+        const firstExperiment = generateExperiment();
+        const secondExperiment = generateExperiment();
+        databaseConnectorInsertStub.resolves();
+        databaseConnectorGetStub.resolves([firstExperiment, secondExperiment]);
+        const jobExperiments = [
+            {
+                experiment_id: firstExperiment.id,
+                start_after: 1000
+            },
+            {
+                experiment_id: secondExperiment.id,
+                start_after: 2000
+            }
+        ];
+        const jobId = uuid();
+        await jobExperimentHandler.setChaosExperimentsIfExist(jobId, jobExperiments);
+        databaseConnectorGetStub.callCount.should.eql(1);
+        databaseConnectorInsertStub.callCount.should.eql(2);
+        databaseConnectorInsertStub.args[0][1].should.eql(jobId);
+        databaseConnectorInsertStub.args[0][2].should.eql(firstExperiment.id);
+        (databaseConnectorInsertStub.args[0][4] - databaseConnectorInsertStub.args[0][3]).should.eql(60000);
+        databaseConnectorInsertStub.args[1][1].should.eql(jobId);
+        databaseConnectorInsertStub.args[1][2].should.eql(secondExperiment.id);
+        (databaseConnectorInsertStub.args[1][3] - databaseConnectorInsertStub.args[0][3]).should.eql(1000);
+    });
+
+    it('set chaos experiments with same experiment in different times', async () => {
+        const experiment = generateExperiment();
+        databaseConnectorInsertStub.resolves();
+        databaseConnectorGetStub.resolves([experiment]);
+        const jobExperiments = [
+            {
+                experiment_id: experiment.id,
+                start_after: 1000
+            },
+            {
+                experiment_id: experiment.id,
+                start_after: 2000
+            }
+        ];
+        const jobId = uuid();
+        await jobExperimentHandler.setChaosExperimentsIfExist(jobId, jobExperiments);
+        databaseConnectorGetStub.callCount.should.eql(1);
+        databaseConnectorInsertStub.callCount.should.eql(2);
+        databaseConnectorInsertStub.args[0][1].should.eql(jobId);
+        databaseConnectorInsertStub.args[0][2].should.eql(experiment.id);
+        (databaseConnectorInsertStub.args[0][4] - databaseConnectorInsertStub.args[0][3]).should.eql(60000);
+    });
+
+    it('set chaos experiments with no experiments', async () => {
+        await jobExperimentHandler.setChaosExperimentsIfExist(uuid(), []);
+        databaseConnectorGetStub.callCount.should.eql(1);
+        databaseConnectorInsertStub.callCount.should.eql(0);
+    });
+});