Merge "Allow max_avg_sla per atomic actions"

rally-jobs/rally.yaml

@@ -625,6 +625,23 @@
         times: 10
         concurrency: 5
 
+  Dummy.dummy_timed_atomic_actions:
+    -
+      args:
+        number_of_actions: 5
+        sleep_factor: 1
+      runner:
+        type: "constant"
+        times: 3
+        concurrency: 3
+      sla:
+        max_avg_duration_per_atomic:
+          action_0: 1.0
+          action_1: 2.0
+          action_2: 3.0
+          action_3: 4.0
+          action_4: 5.0
+
   FakePlugin.testplugin:
     -
       runner:

rally/plugins/common/scenarios/dummy/dummy.py

@@ -196,3 +196,14 @@ class Dummy(scenario.Scenario):
             duration = random.uniform(sleep_min, sleep_max)
             with atomic.ActionTimer(self, "action_%d" % idx):
                 utils.interruptable_sleep(duration)
+
+    @scenario.configure()
+    def dummy_timed_atomic_actions(self, number_of_actions=5, sleep_factor=1):
+        """Run some sleepy atomic actions for SLA atomic action tests.
+
+        :param number_of_actions: int number of atomic actions to create
+        :param sleep_factor: int multiplier for number of seconds to sleep
+        """
+        for sleeptime in range(number_of_actions):
+            with atomic.ActionTimer(self, "action_%d" % sleeptime):
+                utils.interruptable_sleep(sleeptime * sleep_factor)

rally/plugins/common/sla/max_average_duration_per_atomic.py

@@ -0,0 +1,70 @@
+# 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.
+"""
+
+import collections
+
+from rally.common.i18n import _
+from rally.common import streaming_algorithms
+from rally import consts
+from rally.task import sla
+
+
+@sla.configure(name="max_avg_duration_per_atomic")
+class MaxAverageDurationPerAtomic(sla.SLA):
+    """Maximum average duration of one iterations atomic actions in seconds."""
+    CONFIG_SCHEMA = {"type": "object", "$schema": consts.JSON_SCHEMA,
+                     "patternProperties": {".*": {"type": "number"}},
+                     "additionalProperties": False}
+
+    def __init__(self, criterion_value):
+        super(MaxAverageDurationPerAtomic, self).__init__(criterion_value)
+        self.avg_by_action = collections.defaultdict(float)
+        self.avg_comp_by_action = collections.defaultdict(
+            streaming_algorithms.MeanComputation)
+        self.criterion_items = self.criterion_value.items()
+
+    def add_iteration(self, iteration):
+        if not iteration.get("error"):
+            for action, value in iteration["atomic_actions"].items():
+                self.avg_comp_by_action[action].add(value)
+                result = self.avg_comp_by_action[action].result()
+                self.avg_by_action[action] = result
+        self.success = all(self.avg_by_action[atom] <= val
+                           for atom, val in self.criterion_items)
+        return self.success
+
+    def merge(self, other):
+        for atom, comp in self.avg_comp_by_action.items():
+            if atom in other.avg_comp_by_action:
+                comp.merge(other.avg_comp_by_action[atom])
+        self.avg_by_action = {a: comp.result() or 0.0
+                              for a, comp in self.avg_comp_by_action.items()}
+        self.success = all(self.avg_by_action[atom] <= val
+                           for atom, val in self.criterion_items)
+        return self.success
+
+    def details(self):
+        strs = [_("Action: '%s'. %.2fs <= %.2fs") %
+                (atom, self.avg_by_action[atom], val)
+                for atom, val in self.criterion_items]
+        head = _("Average duration of one iteration for atomic actions:")
+        end = _("Status: %s") % self.status()
+        return "\n".join([head] + strs + [end])

samples/tasks/scenarios/dummy/dummy_timed_atomic_actions.json

@@ -0,0 +1,26 @@
+{
+    "Dummy.dummy_timed_atomic_actions": [
+        {
+            "args": {
+                "number_of_actions": 1,
+                "sleep_factor": 1
+            },
+            "runner": {
+                "type": "constant",
+                "times": 5,
+                "concurrency": 5
+            },
+            "sla": {
+                "max_avg_duration_per_atomic": {
+                    "action_0": 1.0
+                }
+            },
+            "context": {
+                "users": {
+                    "tenants": 1,
+                    "users_per_tenant": 1
+                }
+            }
+        }
+    ]
+}

samples/tasks/scenarios/dummy/dummy_timed_atomic_actions.yaml

@@ -0,0 +1,17 @@
+---
+  Dummy.dummy_timed_atomic_actions:
+    -
+      args:
+        number_of_actions: 1
+        sleep_factor: 1
+      runner:
+        type: "constant"
+        times: 5
+        concurrency: 5
+      sla:
+        max_avg_duration_per_atomic:
+          action_0: 1.0
+      context:
+        users:
+          tenants: 1
+          users_per_tenant: 1

tests/unit/plugins/common/scenarios/dummy/test_dummy.py

@@ -161,3 +161,18 @@ class DummyTestCase(test.TestCase):
         for i in range(actions_num):
             self._test_atomic_action_timer(scenario.atomic_actions(),
                                            "action_%d" % i)
+
+    @ddt.data({"number_of_actions": 5, "sleep_factor": 1},
+              {"number_of_actions": 7, "sleep_factor": 2},
+              {"number_of_actions": 1, "sleep_factor": 3})
+    @ddt.unpack
+    @mock.patch(DUMMY + "utils.interruptable_sleep")
+    def test_dummy_timed_atomic_actions(self, mock_interruptable_sleep,
+                                        number_of_actions, sleep_factor):
+        scenario = dummy.Dummy(test.get_test_context())
+        scenario.dummy_random_action(number_of_actions, sleep_factor)
+        scenario.dummy_timed_atomic_actions(number_of_actions, sleep_factor)
+        for i in range(number_of_actions):
+            self._test_atomic_action_timer(scenario.atomic_actions(),
+                                           "action_%d" % i)
+            mock_interruptable_sleep.assert_any_call(i * sleep_factor)

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

@@ -0,0 +1,90 @@
+# 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 max_average_duration_per_atomic as madpa
+from tests.unit import test
+
+
+@ddt.ddt
+class MaxAverageDurationPerAtomicTestCase(test.TestCase):
+    def test_config_schema(self):
+        properties = {
+            "max_avg_duration_per_atomic": {"neutron.list_ports": "elf",
+                                            "neutron.create_port": 1.0}
+        }
+        self.assertRaises(
+            jsonschema.ValidationError,
+            madpa.MaxAverageDurationPerAtomic.validate,
+            properties)
+        properties["max_avg_duration_per_atomic"]["neutron.list_ports"] = 1.0
+        madpa.MaxAverageDurationPerAtomic.validate(properties)
+
+    def test_result(self):
+        cls = madpa.MaxAverageDurationPerAtomic
+        sla1 = cls({"a1": 42, "a2": 42})
+        sla2 = cls({"a1": 42, "a2": 2})
+        for sla in [sla1, sla2]:
+            sla.add_iteration({"atomic_actions": {"a1": 3.14, "a2": 7.77}})
+            sla.add_iteration({"atomic_actions": {"a1": 8.14, "a2": 9.77}})
+        self.assertTrue(sla1.result()["success"])
+        self.assertFalse(sla2.result()["success"])
+        self.assertEqual("Passed", sla1.status())
+        self.assertEqual("Failed", sla2.status())
+
+    def test_result_no_iterations(self):
+        sla = madpa.MaxAverageDurationPerAtomic({"a1": 8.14, "a2": 9.77})
+        self.assertTrue(sla.result()["success"])
+
+    def test_add_iteration(self):
+        sla = madpa.MaxAverageDurationPerAtomic({"a1": 5, "a2": 10})
+        add = sla.add_iteration
+        self.assertTrue(add({"atomic_actions": {"a1": 2.5, "a2": 5.0}}))
+        self.assertTrue(add({"atomic_actions": {"a1": 5.0, "a2": 10.0}}))
+        # the following pushes a2 over the limit
+        self.assertFalse(add({"atomic_actions": {"a1": 5.0, "a2": 20.0}}))
+        # bring a2 back
+        self.assertTrue(add({"atomic_actions": {"a1": 5.0, "a2": 2.0}}))
+        # push a1 over
+        self.assertFalse(add({"atomic_actions": {"a1": 10.0, "a2": 2.0}}))
+        # bring it back
+        self.assertTrue(add({"atomic_actions": {"a1": 1.0, "a2": 2.0}}))
+
+    @ddt.data([[1.0, 2.0, 1.5, 4.3],
+               [2.1, 3.4, 1.2, 6.3, 7.2, 7.0, 1.],
+               [1.1, 1.1, 2.2, 2.2, 3.3, 4.3]])
+    def test_merge(self, durations):
+        init = {"a1": 8.14, "a2": 9.77}
+        single_sla = madpa.MaxAverageDurationPerAtomic(init)
+
+        for dd in durations:
+            for d in dd:
+                single_sla.add_iteration(
+                    {"atomic_actions": {"a1": d, "a2": d * 2}})
+
+        slas = [madpa.MaxAverageDurationPerAtomic(init) for _ in durations]
+
+        for idx, sla in enumerate(slas):
+            for d in durations[idx]:
+                sla.add_iteration({"atomic_actions": {"a1": d, "a2": d * 2}})
+
+        merged_sla = slas[0]
+        for sla in slas[1:]:
+            merged_sla.merge(sla)
+
+        self.assertEqual(single_sla.success, merged_sla.success)
+        self.assertEqual(single_sla.avg_by_action, merged_sla.avg_by_action)