Max test duration

Add a maximum time limit for any one given test so
that silly mistakes like block size 512, queue depth 1
does not end up running for 2 days

JIRA: STORPERF-53

Change-Id: I1f3e5046e1dd346fc5b28a7b935df2ee9aa6ba3d
Signed-off-by: Mark Beierl <mark.beierl@emc.com>

3 files changed, 48 insertions, 7 deletions

diff --git a/rest_server.py b/rest_server.py
index b628a7e..b7d32d9 100644
--- a/rest_server.py
+++ b/rest_server.py
@@ -220,6 +220,7 @@ class WorkloadModel:
         'target': fields.String,
         'nossd': fields.String,
         'nowarm': fields.String,
+        'deadline': fields.Integer,
         'workload': fields.String,
         'queue_depths': fields.String,
         'block_sizes': fields.String
@@ -291,13 +292,22 @@ class Job(Resource):
         parameters=[
             {
                 "name": "body",
-                "description": 'Start execution of a workload with the '
-                'following parameters: "target": The target device to '
-                'profile", "nossd": Do not fill the target with random '
-                'data prior to running the test, "nowarm": Do not '
-                'refill the target with data '
-                'prior to running any further tests, "workload":if specified, '
-                'the workload to run. Defaults to all.',
+                "description": """Start execution of a workload with the
+                following parameters:
+
+                "target": The target device to profile",
+
+                "deadline": if specified, the maximum duration in minutes
+                for any single test iteration.
+
+                "nossd": Do not fill the target with random
+                data prior to running the test,
+
+                "nowarm": Do not refill the target with data
+                prior to running any further tests,
+
+                "workload":if specified, the workload to run. Defaults to all.
+                """,
                 "required": True,
                 "type": "WorkloadModel",
                 "paramType": "body"
@@ -324,6 +334,8 @@ class Job(Resource):
         try:
             if ('target' in request.json):
                 storperf.filename = request.json['target']
+            if ('deadline' in request.json):
+                storperf.deadline = request.json['deadline']
             if ('nossd' in request.json):
                 storperf.precondition = False
             if ('nowarm' in request.json):
diff --git a/storperf/storperf_master.py b/storperf/storperf_master.py
index 33f0819..0f86a95 100644
--- a/storperf/storperf_master.py
+++ b/storperf/storperf_master.py
@@ -177,6 +177,14 @@ class StorPerfMaster(object):
         self._test_executor.precondition = value
 
     @property
+    def deadline(self):
+        return self._test_executor.deadline
+
+    @deadline.setter
+    def deadline(self, value):
+        self._test_executor.deadline = value
+
+    @property
     def warm_up(self):
         return self._test_executor.warm
 
diff --git a/storperf/test_executor.py b/storperf/test_executor.py
index d1ad3ca..8230174 100644
--- a/storperf/test_executor.py
+++ b/storperf/test_executor.py
@@ -18,6 +18,8 @@ import copy
 import imp
 import logging
 import os
+import sched
+import time
 
 
 class UnknownWorkload(Exception):
@@ -31,6 +33,7 @@ class TestExecutor(object):
         self.workload_modules = []
         self.filename = None
         self.precondition = True
+        self.deadline = None
         self.warm = True
         self._queue_depths = [1, 4, 8]
         self._block_sizes = [512, 4096, 16384]
@@ -143,6 +146,10 @@ class TestExecutor(object):
 
     def terminate(self):
         self._terminated = True
+        return self.terminate_current_run()
+
+    def terminate_current_run(self):
+        self.logger.info("Terminating current run")
         terminated_hosts = []
         for workload in self._workload_executors:
             workload.terminate()
@@ -170,9 +177,17 @@ class TestExecutor(object):
             for blocksize in blocksizes:
                 for iodepth in iodepths:
 
+                    scheduler = sched.scheduler(time.time, time.sleep)
                     if self._terminated:
                         return
 
+                    if self.deadline is not None \
+                            and not workload_name.startswith("_"):
+                        event = scheduler.enter(self.deadline * 60, 1,
+                                                self.terminate_current_run, ())
+                        t = Thread(target=scheduler.run, args=())
+                        t.start()
+
                     workload.options['iodepth'] = str(iodepth)
                     workload.options['bs'] = str(blocksize)
 
@@ -192,6 +207,12 @@ class TestExecutor(object):
                     for slave_thread in slave_threads:
                         slave_thread.join()
 
+                    if not scheduler.empty():
+                        try:
+                            scheduler.cancel(event)
+                        except:
+                            pass
+
                     self._workload_executors = []
 
     def execute_on_node(self, workload):