Add performance degradation SLA plugin

This adds SLA plugin that finds minimum and maximum duration of
iterations completed without errors during Rally task execution.
Assuming that minimum duration is 100%, it calculates
performance degradation against maximum duration.

Example config:
  sla:
    performance_degradation:
      max_degradation: 75

Spec: sla_pd_plugin
Change-Id: Ieedba7be72364f5599a3c0cf79f5f494a7391ea0

rally-jobs/rally.yaml

@@ -404,6 +404,8 @@
           max: 1
           min_iterations: 10
           sigmas: 10
+        performance_degradation:
+          max_degradation: 50
 
     -
       args:

rally/common/streaming_algorithms.py

@@ -13,6 +13,8 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from __future__ import division
+
 import abc
 import math
 
@@ -204,3 +206,36 @@ class IncrementComputation(StreamingAlgorithm):
 
     def result(self):
         return self._count
+
+
+class DegradationComputation(StreamingAlgorithm):
+    """Calculates degradation from a stream of numbers
+
+    Finds min and max values from a stream and then calculates
+    ratio between them in percentage. Works only with positive numbers.
+    """
+
+    def __init__(self):
+        self.min_value = MinComputation()
+        self.max_value = MaxComputation()
+
+    def add(self, value):
+        if value <= 0.0:
+            raise ValueError("Unexpected value: %s" % value)
+        self.min_value.add(value)
+        self.max_value.add(value)
+
+    def merge(self, other):
+        min_result = other.min_value.result()
+        if min_result is not None:
+            self.min_value.add(min_result)
+        max_result = other.max_value.result()
+        if max_result is not None:
+            self.max_value.add(max_result)
+
+    def result(self):
+        min_result = self.min_value.result()
+        max_result = self.max_value.result()
+        if min_result is None or max_result is None:
+            return 0.0
+        return (max_result / min_result - 1) * 100.0

rally/plugins/common/sla/performance_degradation.py

@@ -0,0 +1,74 @@
+# Copyright 2016: Mirantis Inc.
+# All Rights Reserved.
+#
+#    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.
+
+
+"""
+SLA (Service-level agreement) is set of details for determining compliance
+with contracted values such as maximum error rate or minimum response time.
+"""
+
+from __future__ import division
+
+from rally.common.i18n import _
+from rally.common import streaming_algorithms
+from rally.common import utils
+from rally import consts
+from rally.task import sla
+
+
+@sla.configure(name="performance_degradation")
+class PerformanceDegradation(sla.SLA):
+    """Calculates perfomance degradation based on iteration time
+
+    This SLA plugin finds minimum and maximum duration of
+    iterations completed without errors during Rally task execution.
+    Assuming that minimum duration is 100%, it calculates
+    performance degradation against maximum duration.
+    """
+    CONFIG_SCHEMA = {
+        "type": "object",
+        "$schema": consts.JSON_SCHEMA,
+        "properties": {
+            "max_degradation": {
+                "type": "number",
+                "minimum": 0.0,
+            },
+        },
+        "required": [
+            "max_degradation",
+        ],
+        "additionalProperties": False,
+    }
+
+    def __init__(self, criterion_value):
+        super(PerformanceDegradation, self).__init__(criterion_value)
+        self.max_degradation = self.criterion_value["max_degradation"]
+        self.degradation = streaming_algorithms.DegradationComputation()
+
+    def add_iteration(self, iteration):
+        if not iteration.get("error"):
+            self.degradation.add(iteration["duration"])
+        self.success = self.degradation.result() <= self.max_degradation
+        return self.success
+
+    def merge(self, other):
+        self.degradation.merge(other.degradation)
+        self.success = self.degradation.result() <= self.max_degradation
+        return self.success
+
+    def details(self):
+        return (_("Current degradation: %s%% - %s") %
+                 (utils.format_float_to_str(self.degradation.result() or 0.0),
+                  self.status()))

tests/functional/test_cli_task.py

@@ -980,3 +980,50 @@ class SLAExtraFlagsTestCase(unittest.TestCase):
                                    "times": 5,
                                    "rps": 3,
                                    "timeout": 6})
+
+
+class SLAPerfDegrTestCase(unittest.TestCase):
+
+    def _get_sample_task_config(self, max_degradation=500):
+        return {
+            "Dummy.dummy_random_action": [
+                {
+                    "args": {
+                        "actions_num": 5,
+                        "sleep_min": 0.5,
+                        "sleep_max": 2
+                    },
+                    "runner": {
+                        "type": "constant",
+                        "times": 10,
+                        "concurrency": 5
+                    },
+                    "sla": {
+                        "performance_degradation": {
+                            "max_degradation": max_degradation
+                        }
+                    }
+                }
+            ]
+        }
+
+    def test_sla_fail(self):
+        rally = utils.Rally()
+        cfg = self._get_sample_task_config(max_degradation=1)
+        config = utils.TaskConfig(cfg)
+        rally("task start --task %s" % config.filename)
+        self.assertRaises(utils.RallyCliError, rally, "task sla_check")
+
+    def test_sla_success(self):
+        rally = utils.Rally()
+        config = utils.TaskConfig(self._get_sample_task_config())
+        rally("task start --task %s" % config.filename)
+        rally("task sla_check")
+        expected = [
+            {"benchmark": "Dummy.dummy_random_action",
+             "criterion": "performance_degradation",
+             "detail": mock.ANY,
+             "pos": 0, "status": "PASS"},
+        ]
+        data = rally("task sla_check --json", getjson=True)
+        self.assertEqual(expected, data)

tests/unit/common/test_streaming_algorithms.py

@@ -274,3 +274,47 @@ class IncrementComputationTestCase(test.TestCase):
 
         self.assertEqual(single_inc._count, merged_inc._count)
         self.assertEqual(single_inc.result(), merged_inc.result())
+
+
+@ddt.ddt
+class DegradationComputationTestCase(test.TestCase):
+
+    @ddt.data(
+        ([], None, None, 0.0),
+        ([30.0, 30.0, 30.0, 30.0], 30.0, 30.0, 0.0),
+        ([45.0, 45.0, 45.0, 30.0], 30.0, 45.0, 50.0),
+        ([15.0, 10.0, 20.0, 19.0], 10.0, 20.0, 100.0),
+        ([30.0, 56.0, 90.0, 73.0], 30.0, 90.0, 200.0))
+    @ddt.unpack
+    def test_add(self, stream, min_value, max_value, result):
+        comp = algo.DegradationComputation()
+        for value in stream:
+            comp.add(value)
+        self.assertEqual(min_value, comp.min_value.result())
+        self.assertEqual(max_value, comp.max_value.result())
+        self.assertEqual(result, comp.result())
+
+    @ddt.data(-10.0, -1.0, -1, 0.0, 0)
+    def test_add_raise(self, value):
+        comp = algo.DegradationComputation()
+        self.assertRaises(ValueError, comp.add, value)
+
+    @ddt.data(([39.0, 30.0, 32.0], [49.0, 40.0, 51.0], 30.0, 51.0, 70.0),
+              ([31.0, 30.0, 32.0], [39.0, 45.0, 43.0], 30.0, 45.0, 50.0),
+              ([], [31.0, 30.0, 45.0], 30.0, 45.0, 50.0),
+              ([31.0, 30.0, 45.0], [], 30.0, 45.0, 50.0),
+              ([], [], None, None, 0.0))
+    @ddt.unpack
+    def test_merge(self, stream1, stream2, min_value, max_value, result):
+        comp1 = algo.DegradationComputation()
+        for value in stream1:
+            comp1.add(value)
+
+        comp2 = algo.DegradationComputation()
+        for value in stream2:
+            comp2.add(value)
+
+        comp1.merge(comp2)
+        self.assertEqual(min_value, comp1.min_value.result())
+        self.assertEqual(max_value, comp1.max_value.result())
+        self.assertEqual(result, comp1.result())

tests/unit/plugins/common/sla/test_performance_degradation.py

@@ -0,0 +1,92 @@
+# Copyright 2016: Mirantis Inc.
+# All Rights Reserved.
+#
+#    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.
+
+import ddt
+import jsonschema
+
+from rally.plugins.common.sla import performance_degradation as perfdegr
+from tests.unit import test
+
+
+@ddt.ddt
+class PerformanceDegradationTestCase(test.TestCase):
+
+    def setUp(self):
+        super(PerformanceDegradationTestCase, self).setUp()
+        self.sla = perfdegr.PerformanceDegradation({"max_degradation": 50})
+
+    def test_config_schema(self):
+        properties = {
+            "performance_degradation": {}
+        }
+        self.assertRaises(
+            jsonschema.ValidationError,
+            perfdegr.PerformanceDegradation.validate,
+            properties)
+        properties["performance_degradation"]["max_degradation"] = -1
+        self.assertRaises(
+            jsonschema.ValidationError,
+            perfdegr.PerformanceDegradation.validate,
+            properties)
+        properties["performance_degradation"]["max_degradation"] = 1000.0
+        perfdegr.PerformanceDegradation.validate(properties)
+
+    @ddt.data(([39.0, 30.0, 32.0, 49.0, 47.0, 43.0], False, "Failed"),
+              ([31.0, 30.0, 32.0, 39.0, 45.0, 43.0], True, "Passed"),
+              ([], True, "Passed"))
+    @ddt.unpack
+    def test_iterations(self, durations, result, status):
+        for duration in durations:
+            self.sla.add_iteration({"duration": duration})
+        self.assertIs(self.sla.success, result)
+        self.assertIs(self.sla.result()["success"], result)
+        self.assertEqual(status, self.sla.status())
+
+    @ddt.data(([39.0, 30.0, 32.0], [49.0, 40.0, 51.0], False, "Failed"),
+              ([31.0, 30.0, 32.0], [39.0, 45.0, 43.0], True, "Passed"),
+              ([31.0, 30.0, 32.0], [32.0, 49.0, 30.0], False, "Failed"),
+              ([], [31.0, 30.0, 32.0], True, "Passed"),
+              ([31.0, 30.0, 32.0], [], True, "Passed"),
+              ([], [], True, "Passed"),
+              ([35.0, 30.0, 49.0], [], False, "Failed"),
+              ([], [35.0, 30.0, 49.0], False, "Failed"))
+    @ddt.unpack
+    def test_merge(self, durations1, durations2, result, status):
+        for duration in durations1:
+            self.sla.add_iteration({"duration": duration})
+
+        sla2 = perfdegr.PerformanceDegradation({"max_degradation": 50})
+        for duration in durations2:
+            sla2.add_iteration({"duration": duration})
+
+        self.sla.merge(sla2)
+        self.assertIs(self.sla.success, result)
+        self.assertIs(self.sla.result()["success"], result)
+        self.assertEqual(status, self.sla.status())
+
+    def test_details(self):
+        self.assertEqual("Current degradation: 0.0% - Passed",
+                         self.sla.details())
+
+        for duration in [39.0, 30.0, 32.0]:
+            self.sla.add_iteration({"duration": duration})
+
+        self.assertEqual("Current degradation: 30.0% - Passed",
+                         self.sla.details())
+
+        self.sla.add_iteration({"duration": 75.0})
+
+        self.assertEqual("Current degradation: 150.0% - Failed",
+                         self.sla.details())