Add staleness evaluation to conditions. (#3110)

* Add staleness evaluation to conditions.
Add supporting tests
Resolves #3109

* Fix broken test

Co-authored-by: Deep Tailor <deep.j.tailor@nasa.gov>

src/plugins/condition/Condition.js

@@ -150,6 +150,7 @@ export default class Condition extends EventEmitter {
             criterion = new TelemetryCriterion(criterionConfigurationWithId, this.openmct);
         }
         criterion.on('criterionUpdated', (obj) => this.handleCriterionUpdated(obj));
+        criterion.on('telemetryIsStale', (obj) => this.handleStaleCriterion(obj));
         if (!this.criteria) {
             this.criteria = [];
         }
@@ -178,10 +179,12 @@ export default class Condition extends EventEmitter {
             const newCriterionConfiguration = this.generateCriterion(criterionConfiguration);
             let newCriterion = new TelemetryCriterion(newCriterionConfiguration, this.openmct);
             newCriterion.on('criterionUpdated', (obj) => this.handleCriterionUpdated(obj));
+            newCriterion.on('telemetryIsStale', (obj) => this.handleStaleCriterion(obj));
 
             let criterion = found.item;
             criterion.unsubscribe();
             criterion.off('criterionUpdated', (obj) => this.handleCriterionUpdated(obj));
+            criterion.off('telemetryIsStale', (obj) => this.handleStaleCriterion(obj));
             this.criteria.splice(found.index, 1, newCriterion);
             this.updateDescription();
         }
@@ -194,6 +197,9 @@ export default class Condition extends EventEmitter {
             criterion.off('criterionUpdated', (obj) => {
                 this.handleCriterionUpdated(obj);
             });
+            criterion.off('telemetryIsStale', (obj) => {
+                this.handleStaleCriterion(obj);
+            });
             criterion.destroy();
             this.criteria.splice(found.index, 1);
             this.updateDescription();
@@ -211,6 +217,18 @@ export default class Condition extends EventEmitter {
         }
     }
 
+    handleStaleCriterion(updatedCriterion) {
+        this.result = evaluateResults(this.criteria.map(criterion => criterion.result), this.trigger);
+        let latestTimestamp = {};
+        latestTimestamp = getLatestTimestamp(
+            latestTimestamp,
+            updatedCriterion.data,
+            this.timeSystems,
+            this.openmct.time.timeSystem()
+        );
+        this.conditionManager.updateCurrentCondition(latestTimestamp);
+    }
+
     updateDescription() {
         const triggerDescription = this.getTriggerDescription();
         let description = '';

src/plugins/condition/ConditionManager.js

@@ -315,6 +315,10 @@ export default class ConditionManager extends EventEmitter {
             condition.getResult(normalizedDatum);
         });
 
+        this.updateCurrentCondition(timestamp);
+    }
+
+    updateCurrentCondition(timestamp) {
         const currentCondition = this.getCurrentCondition();
 
         this.emit('conditionSetResultUpdated',

src/plugins/condition/components/Criterion.vue

@@ -55,6 +55,7 @@
                 >
                     {{ option.name }}
                 </option>
+                <option value="dataReceived">any data received</option>
             </select>
         </span>
         <span v-if="criterion.telemetry && criterion.metadata"
@@ -83,6 +84,7 @@
                     >
                     <span v-if="inputIndex < inputCount-1">and</span>
                 </span>
+                <span v-if="criterion.metadata === 'dataReceived'">seconds</span>
             </template>
             <span v-else>
                 <span v-if="inputCount && criterion.operation"
@@ -148,7 +150,11 @@ export default {
             return (this.index !== 0 ? operator : '') + ' when';
         },
         filteredOps: function () {
-            return this.operations.filter(op => op.appliesTo.indexOf(this.operationFormat) !== -1);
+            if (this.criterion.metadata === 'dataReceived') {
+                return this.operations.filter(op => op.name === 'isStale');
+            } else {
+                return this.operations.filter(op => op.appliesTo.indexOf(this.operationFormat) !== -1);
+            }
         },
         setInputType: function () {
             let type = '';
@@ -214,6 +220,8 @@ export default {
                 } else {
                     this.operationFormat = 'number';
                 }
+            } else if (this.criterion.metadata === 'dataReceived') {
+                this.operationFormat = 'number';
             }
             this.updateInputVisibilityAndValues();
         },

src/plugins/condition/criterion/AllTelemetryCriterion.js

@@ -22,7 +22,7 @@
 
 import TelemetryCriterion from './TelemetryCriterion';
 import { evaluateResults } from "../utils/evaluator";
-import { getLatestTimestamp } from '../utils/time';
+import {getLatestTimestamp, subscribeForStaleness} from '../utils/time';
 import { getOperatorText } from "@/plugins/condition/utils/operations";
 
 export default class AllTelemetryCriterion extends TelemetryCriterion {
@@ -41,6 +41,32 @@ export default class AllTelemetryCriterion extends TelemetryCriterion {
     initialize() {
         this.telemetryObjects = { ...this.telemetryDomainObjectDefinition.telemetryObjects };
         this.telemetryDataCache = {};
+        if (this.isValid() && this.isStalenessCheck() && this.isValidInput()) {
+            this.subscribeForStaleData(this.telemetryObjects || {});
+        }
+    }
+
+    subscribeForStaleData(telemetryObjects) {
+
+        if (!this.stalenessSubscription) {
+            this.stalenessSubscription = {};
+        }
+        Object.values(telemetryObjects).forEach((telemetryObject) => {
+            const id = this.openmct.objects.makeKeyString(telemetryObject.identifier);
+            if (!this.stalenessSubscription[id]) {
+                this.stalenessSubscription[id] = subscribeForStaleness((data) => {
+                    this.handleStaleTelemetry(id, data);
+                }, this.input[0]*1000);
+            }
+        })
+    }
+
+    handleStaleTelemetry(id, data) {
+        if (this.telemetryDataCache) {
+            this.telemetryDataCache[id] = true;
+            this.result = evaluateResults(Object.values(this.telemetryDataCache), this.telemetry);
+        }
+        this.emitEvent('telemetryIsStale', data);
     }
 
     isValid() {
@@ -50,6 +76,9 @@ export default class AllTelemetryCriterion extends TelemetryCriterion {
     updateTelemetryObjects(telemetryObjects) {
         this.telemetryObjects = { ...telemetryObjects };
         this.removeTelemetryDataCache();
+        if (this.isValid() && this.isStalenessCheck() && this.isValidInput()) {
+            this.subscribeForStaleData(this.telemetryObjects || {});
+        }
     }
 
     removeTelemetryDataCache() {
@@ -63,6 +92,7 @@ export default class AllTelemetryCriterion extends TelemetryCriterion {
         });
         telemetryCacheIds.forEach(id => {
             delete (this.telemetryDataCache[id]);
+            delete (this.stalenessSubscription[id]);
         });
     }
 
@@ -96,7 +126,14 @@ export default class AllTelemetryCriterion extends TelemetryCriterion {
         const validatedData = this.isValid() ? data : {};
 
         if (validatedData) {
-            this.telemetryDataCache[validatedData.id] = this.computeResult(validatedData);
+            if (this.isStalenessCheck()) {
+                if (this.stalenessSubscription[validatedData.id]) {
+                    this.stalenessSubscription[validatedData.id].update(validatedData);
+                }
+                this.telemetryDataCache[validatedData.id] = false;
+            } else {
+                this.telemetryDataCache[validatedData.id] = this.computeResult(validatedData);
+            }
         }
 
         Object.values(telemetryObjects).forEach(telemetryObject => {
@@ -162,7 +199,7 @@ export default class AllTelemetryCriterion extends TelemetryCriterion {
 
     getDescription() {
         const telemetryDescription = this.telemetry === 'all' ? 'all telemetry' : 'any telemetry';
-        let metadataValue = this.metadata;
+        let metadataValue = (this.metadata === 'dataReceived' ? '' : this.metadata);
         let inputValue = this.input;
         if (this.metadata) {
             const telemetryObjects = Object.values(this.telemetryObjects);
@@ -182,5 +219,9 @@ export default class AllTelemetryCriterion extends TelemetryCriterion {
     destroy() {
         delete this.telemetryObjects;
         delete this.telemetryDataCache;
+        if (this.stalenessSubscription) {
+            Object.values(this.stalenessSubscription).forEach((subscription) => subscription.clear);
+            delete this.stalenessSubscription;
+        }
     }
 }

src/plugins/condition/criterion/TelemetryCriterion.js

@@ -22,6 +22,7 @@
 
 import EventEmitter from 'EventEmitter';
 import { OPERATIONS, getOperatorText } from '../utils/operations';
+import { subscribeForStaleness } from "../utils/time";
 
 export default class TelemetryCriterion extends EventEmitter {
 
@@ -43,6 +44,7 @@ export default class TelemetryCriterion extends EventEmitter {
         this.input = telemetryDomainObjectDefinition.input;
         this.metadata = telemetryDomainObjectDefinition.metadata;
         this.result = undefined;
+        this.stalenessSubscription = undefined;
 
         this.initialize();
         this.emitEvent('criterionUpdated', this);
@@ -51,14 +53,40 @@ export default class TelemetryCriterion extends EventEmitter {
     initialize() {
         this.telemetryObjectIdAsString = this.openmct.objects.makeKeyString(this.telemetryDomainObjectDefinition.telemetry);
         this.updateTelemetryObjects(this.telemetryDomainObjectDefinition.telemetryObjects);
+        if (this.isValid() && this.isStalenessCheck() && this.isValidInput()) {
+            this.subscribeForStaleData()
+        }
+    }
+
+    subscribeForStaleData() {
+        if (this.stalenessSubscription) {
+            this.stalenessSubscription.clear();
+        }
+        this.stalenessSubscription = subscribeForStaleness(this.handleStaleTelemetry.bind(this), this.input[0]*1000);
+    }
+
+    handleStaleTelemetry(data) {
+        this.result = true;
+        this.emitEvent('telemetryIsStale', data);
     }
 
     isValid() {
         return this.telemetryObject && this.metadata && this.operation;
     }
 
+    isStalenessCheck() {
+        return this.metadata && this.metadata === 'dataReceived';
+    }
+
+    isValidInput() {
+        return this.input instanceof Array && this.input.length;
+    }
+
     updateTelemetryObjects(telemetryObjects) {
         this.telemetryObject = telemetryObjects[this.telemetryObjectIdAsString];
+        if (this.isValid() && this.isStalenessCheck() && this.isValidInput()) {
+            this.subscribeForStaleData()
+        }
     }
 
     createNormalizedDatum(telemetryDatum, endpoint) {
@@ -91,7 +119,14 @@ export default class TelemetryCriterion extends EventEmitter {
 
     getResult(data) {
         const validatedData = this.isValid() ? data : {};
-        this.result = this.computeResult(validatedData);
+        if (this.isStalenessCheck()) {
+            if (this.stalenessSubscription) {
+                this.stalenessSubscription.update(validatedData);
+            }
+            this.result = false;
+        } else {
+            this.result = this.computeResult(validatedData);
+        }
     }
 
     requestLAD() {
@@ -136,7 +171,7 @@ export default class TelemetryCriterion extends EventEmitter {
             let comparator = this.findOperation(this.operation);
             let params = [];
             params.push(data[this.metadata]);
-            if (this.input instanceof Array && this.input.length) {
+            if (this.isValidInput()) {
                 this.input.forEach(input => params.push(input));
             }
             if (typeof comparator === 'function') {
@@ -191,7 +226,7 @@ export default class TelemetryCriterion extends EventEmitter {
             description = `Unknown ${this.metadata} ${getOperatorText(this.operation, this.input)}`;
         } else {
             const metadataObject = this.getMetaDataObject(this.telemetryObject, this.metadata);
-            const metadataValue = this.getMetadataValueFromMetaData(metadataObject) || this.metadata;
+            const metadataValue = this.getMetadataValueFromMetaData(metadataObject) || (this.metadata === 'dataReceived' ? '' : this.metadata);
             const inputValue = this.getInputValueFromMetaData(metadataObject, this.input) || this.input;
             description = `${this.telemetryObject.name} ${metadataValue} ${getOperatorText(this.operation, inputValue)}`;
         }
@@ -202,5 +237,8 @@ export default class TelemetryCriterion extends EventEmitter {
     destroy() {
         delete this.telemetryObject;
         delete this.telemetryObjectIdAsString;
+        if (this.stalenessSubscription) {
+            delete this.stalenessSubscription;
+        }
     }
 }

src/plugins/condition/pluginSpec.js

@@ -25,19 +25,50 @@ import ConditionPlugin from "./plugin";
 import StylesView from "./components/inspector/StylesView.vue";
 import Vue from 'vue';
 import {getApplicableStylesForItem} from "./utils/styleUtils";
+import ConditionManager from "@/plugins/condition/ConditionManager";
 
 describe('the plugin', function () {
     let conditionSetDefinition;
     let mockConditionSetDomainObject;
+    let mockListener;
     let element;
     let child;
     let openmct;
+    let testTelemetryObject;
 
     beforeAll(() => {
         resetApplicationState(openmct);
     });
 
     beforeEach((done) => {
+        testTelemetryObject = {
+            identifier:{ namespace: "", key: "test-object"},
+            type: "test-object",
+            name: "Test Object",
+            telemetry: {
+                valueMetadatas: [{
+                    key: "some-key",
+                    name: "Some attribute",
+                    hints: {
+                        range: 2
+                    }
+                },
+                {
+                    key: "utc",
+                    name: "Time",
+                    format: "utc",
+                    hints: {
+                        domain: 1
+                    }
+                }, {
+                    key: "testSource",
+                    source: "value",
+                    name: "Test",
+                    format: "string"
+                }]
+            }
+        };
+
         openmct = createOpenMct();
         openmct.install(new ConditionPlugin());
 
@@ -55,6 +86,8 @@ describe('the plugin', function () {
             type: 'conditionSet'
         };
 
+        mockListener = jasmine.createSpy('mockListener');
+
         conditionSetDefinition.initialize(mockConditionSetDomainObject);
 
         openmct.on('start', done);
@@ -356,4 +389,113 @@ describe('the plugin', function () {
         });
 
     });
+
+    describe('the condition check for staleness', () => {
+        let conditionSetDomainObject;
+
+        beforeEach(()=>{
+            conditionSetDomainObject = {
+                "configuration":{
+                    "conditionTestData":[
+                        {
+                            "telemetry":"",
+                            "metadata":"",
+                            "input":""
+                        }
+                    ],
+                    "conditionCollection":[
+                        {
+                            "id":"39584410-cbf9-499e-96dc-76f27e69885d",
+                            "configuration":{
+                                "name":"Unnamed Condition",
+                                "output":"Any stale telemetry",
+                                "trigger":"all",
+                                "criteria":[
+                                    {
+                                        "id":"35400132-63b0-425c-ac30-8197df7d5862",
+                                        "telemetry":"any",
+                                        "operation":"isStale",
+                                        "input":[
+                                            "1"
+                                        ],
+                                        "metadata":"dataReceived"
+                                    }
+                                ]
+                            },
+                            "summary":"Match if all criteria are met: Any telemetry is stale after 5 seconds"
+                        },
+                        {
+                            "isDefault":true,
+                            "id":"2532d90a-e0d6-4935-b546-3123522da2de",
+                            "configuration":{
+                                "name":"Default",
+                                "output":"Default",
+                                "trigger":"all",
+                                "criteria":[
+                                ]
+                            },
+                            "summary":""
+                        }
+                    ]
+                },
+                "composition":[
+                    {
+                        "namespace":"",
+                        "key":"test-object"
+                    }
+                ],
+                "telemetry":{
+                },
+                "name":"Condition Set",
+                "type":"conditionSet",
+                "identifier":{
+                    "namespace":"",
+                    "key":"cf4456a9-296a-4e6b-b182-62ed29cd15b9"
+                }
+
+            };
+        });
+
+        it('should evaluate as stale when telemetry is not received in the allotted time', (done) => {
+
+            let conditionMgr = new ConditionManager(conditionSetDomainObject, openmct);
+            conditionMgr.on('conditionSetResultUpdated', mockListener);
+            conditionMgr.telemetryObjects = {
+                "test-object": testTelemetryObject
+            };
+            conditionMgr.updateConditionTelemetryObjects();
+            setTimeout(() => {
+                expect(mockListener).toHaveBeenCalledWith({
+                    output: 'Any stale telemetry',
+                    id: { namespace: '', key: 'cf4456a9-296a-4e6b-b182-62ed29cd15b9' },
+                    conditionId: '39584410-cbf9-499e-96dc-76f27e69885d',
+                    utc: undefined
+                });
+                done();
+            }, 1500);
+        });
+
+        it('should not evaluate as stale when telemetry is received in the allotted time', (done) => {
+            const date = Date.now();
+            conditionSetDomainObject.configuration.conditionCollection[0].configuration.criteria[0].input = ["2"];
+            let conditionMgr = new ConditionManager(conditionSetDomainObject, openmct);
+            conditionMgr.on('conditionSetResultUpdated', mockListener);
+            conditionMgr.telemetryObjects = {
+                "test-object": testTelemetryObject
+            };
+            conditionMgr.updateConditionTelemetryObjects();
+            conditionMgr.telemetryReceived(testTelemetryObject, {
+                utc: date
+            });
+            setTimeout(() => {
+                expect(mockListener).toHaveBeenCalledWith({
+                    output: 'Default',
+                    id: { namespace: '', key: 'cf4456a9-296a-4e6b-b182-62ed29cd15b9' },
+                    conditionId: '2532d90a-e0d6-4935-b546-3123522da2de',
+                    utc: undefined
+                });
+                done();
+            }, 1500);
+        });
+    });
 });

src/plugins/condition/utils/operations.js

@@ -283,6 +283,18 @@ export const OPERATIONS = [
         getDescription: function (values) {
             return ' is not one of ' + values[0];
         }
+    },
+    {
+        name: 'isStale',
+        operation: function () {
+            return false;
+        },
+        text: 'is older than',
+        appliesTo: ["number"],
+        inputCount: 1,
+        getDescription: function (values) {
+            return ` is older than ${values[0] || ''} seconds`;
+        }
     }
 ];
 

src/plugins/condition/utils/time.js

@@ -50,3 +50,26 @@ function updateLatestTimeStamp(timestamp, timeSystems) {
 
     return latest;
 }
+
+export const subscribeForStaleness = (callback, timeout) => {
+    let stalenessTimer = setTimeout(() => {
+        clearTimeout(stalenessTimer);
+        callback();
+    }, timeout);
+    return {
+        update: (data) => {
+            if (stalenessTimer) {
+                clearTimeout(stalenessTimer);
+            }
+            stalenessTimer = setTimeout(() => {
+                clearTimeout(stalenessTimer);
+                callback(data);
+            }, timeout);
+        },
+        clear: () => {
+            if (stalenessTimer) {
+                clearTimeout(stalenessTimer);
+            }
+        }
+    }
+};

src/plugins/condition/utils/timeSpec.js

@@ -0,0 +1,64 @@
+/*****************************************************************************
+ * Open MCT, Copyright (c) 2014-2020, United States Government
+ * as represented by the Administrator of the National Aeronautics and Space
+ * Administration. All rights reserved.
+ *
+ * Open MCT is licensed under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ *
+ * Open MCT includes source code licensed under additional open source
+ * licenses. See the Open Source Licenses file (LICENSES.md) included with
+ * this source code distribution or the Licensing information page available
+ * at runtime from the About dialog for additional information.
+ *****************************************************************************/
+import { subscribeForStaleness } from "./time";
+
+describe('time related utils', () => {
+    let subscription;
+    let mockListener;
+
+    beforeEach(() => {
+        mockListener = jasmine.createSpy('listener');
+        subscription = subscribeForStaleness(mockListener, 100);
+    });
+
+    describe('subscribe for staleness', () => {
+        it('should call listeners when stale', (done) => {
+            setTimeout(() => {
+                expect(mockListener).toHaveBeenCalled();
+                done();
+            }, 200);
+        });
+
+        it('should update the subscription', (done) => {
+            function updated() {
+                setTimeout(() => {
+                    expect(mockListener).not.toHaveBeenCalled();
+                    done();
+                }, 50);
+            }
+            setTimeout(() => {
+                subscription.update();
+                updated();
+            }, 50);
+        });
+
+        it('should clear the subscription', (done) => {
+            subscription.clear();
+
+            setTimeout(() => {
+                expect(mockListener).not.toHaveBeenCalled();
+                done();
+            }, 200);
+        });
+    });
+
+});