Merge branch 'master' into correct-upsert-redis

cache/models/samples.js

@@ -517,12 +517,12 @@ module.exports = {
    */
   findSamples(logObject, method, params) {
     const opts = getOptionsFromReq(params);
-    const commands = [];
     const response = [];
 
     // get all Samples sorted lexicographically
     return redisClient.sortAsync(constants.indexKey.sample, 'alpha')
     .then((allSampKeys) => {
+      const commands = [];
       const filteredSampKeys = applyFiltersOnSampKeys(allSampKeys, opts);
 
       // add to commands

cache/sampleStoreTimeout.js

@@ -0,0 +1,141 @@
+/**
+ * Copyright (c) 2017, salesforce.com, inc.
+ * All rights reserved.
+ * Licensed under the BSD 3-Clause license.
+ * For full license text, see LICENSE.txt file in the repo root or
+ * https://opensource.org/licenses/BSD-3-Clause
+ */
+
+/**
+ * ./cache/sampleStoreTimeout.js
+ *
+ * Timeout samples
+ */
+'use strict'; // eslint-disable-line strict
+const sampleStore = require('./sampleStore');
+const redisClient = require('./redisCache').client.sampleStore;
+const isTimedOut = require('../db/helpers/sampleUtils').isTimedOut;
+const constants = require('../api/v1/constants');
+const redisErrors = require('./redisErrors');
+const ONE = 1;
+
+/**
+ * Create sample array from responses using the samples count
+ * calculated before. The remaining are aspects. Create a hash of
+ * aspects, aspname -> asp object. When analyzing samples, get aspect
+ * from this hash using sample aspect name.
+ * @param  {Array} samples - Samples to be timedout
+ * @param  {Hash} aspects - Aspects hash, name-> object
+ * @param  {Date} curr - Current datetime
+ * @returns {Array} Commands needed to timeout samples
+ */
+function getSampleTimeoutCommands(samples, aspects, curr) {
+  const sampCmds = [];
+
+  for (let num = 0; num < samples.length; num++) {
+    const samp = samples[num];
+    const aspName = samp.name.split('|')[ONE];
+
+    if (!aspName) {
+      throw new redisErrors.ResourceNotFoundError({
+        explanation: 'Aspect not found.',
+      });
+    }
+
+    const asp = aspects[aspName.toLowerCase()];
+    const sampUpdDateTime = new Date(samp.updatedAt);
+
+    /* Update sample if aspect exists, sample status is other than TimeOut and
+    sample is timed out.*/
+    if (asp && isTimedOut(asp.timeout, curr, sampUpdDateTime)) {
+      const objToUpdate = {
+        value: constants.statuses.Timeout,
+        status: constants.statuses.Timeout,
+        previousStatus: samp.status,
+        statusChangedAt: new Date().toString(),
+        updatedAt: new Date().toString(),
+      };
+      sampCmds.push([
+        'hmset',
+        sampleStore.toKey(sampleStore.constants.objectType.sample, samp.name),
+        objToUpdate,
+      ]);
+    }
+  }
+
+  return sampCmds;
+}
+
+module.exports = {
+
+  /**
+   * Invalidates samples which were last updated before the "timeout" specified
+   * by the aspect. Get all samples and corresponding aspects. If the sample
+   * should be timed out, set sample value, status, previous status, status
+   * changed at and updated at fields.
+   * @param  {Date} now - Date object
+   * @returns {Promise} - Resolves to the number of evaluated and timed out
+   * samples
+   */
+  doTimeout(now) {
+    const curr = now || new Date();
+    let numberTimedOut = 0;
+    let numberEvaluated = 0;
+    let samplesCount = 0;
+
+    return redisClient.smembersAsync(sampleStore.constants.indexKey.sample)
+    .then((allSamples) => {
+      const commands = [];
+      const aspectsSet = new Set();
+      const aspectType = sampleStore.constants.objectType.aspect;
+
+      /**
+       * Add commands to get all samples first, create an aspect set with
+       * unique aspect names. Add commands to get the aspects then.
+       */
+      samplesCount = allSamples.length;
+      allSamples.forEach((sampKey) => {
+        const aspectName = sampKey.split('|')[ONE];
+        aspectsSet.add(aspectName);
+        commands.push(['hgetall', sampKey]);
+      });
+
+      aspectsSet.forEach((aspName) => {
+        commands.push(
+          ['hgetall', sampleStore.toKey(aspectType, aspName)]
+        );
+      });
+
+      return redisClient.batch(commands).execAsync();
+    })
+    .then((redisResponses) => {
+      const aspects = {};
+      const samples = [];
+      for (let num = 0; num < samplesCount; num++) {
+        const samp = redisResponses[num];
+        if (samp.status !== constants.statuses.Timeout) {
+          samples.push(samp);
+        }
+      }
+
+      for (let num = samplesCount; num < redisResponses.length; num++) {
+        const aspect = redisResponses[num];
+        if (aspect && aspect.name) {
+          aspects[aspect.name.toLowerCase()] = aspect;
+        }
+      }
+
+      const sampCmds = getSampleTimeoutCommands(samples, aspects, curr);
+      numberEvaluated = samples.length;
+      numberTimedOut = sampCmds.length;
+      return redisClient.batch(sampCmds).execAsync();
+    })
+    .then(() => {
+      const res = { numberEvaluated, numberTimedOut };
+      return res;
+    })
+    .catch((err) => {
+      throw err;
+    });
+  },
+};

clock/scheduledJobs/sampleTimeoutJob.js

@@ -14,13 +14,18 @@
  */
 const featureToggles = require('feature-toggles');
 const dbSample = require('../../db/index').Sample;
+const sampleStoreTimeout = require('../../cache/sampleStoreTimeout');
 
 /**
  * Execute the call to check for sample timeouts.
  *
  * @returns {Promise}
  */
 function execute() {
+  if (featureToggles.isFeatureEnabled('enableRedisSampleStore')) {
+    return sampleStoreTimeout.doTimeout();
+  }
+
   return dbSample.doTimeout();
 } // execute