diff --git a/oslo_utils/tests/test_timeutils.py b/oslo_utils/tests/test_timeutils.py
index 1178f1d0..cacb0ef4 100644
--- a/oslo_utils/tests/test_timeutils.py
+++ b/oslo_utils/tests/test_timeutils.py
@@ -25,6 +25,12 @@ from testtools import matchers
 from oslo_utils import timeutils
 
 
+def monotonic_iter(start=0, incr=0.05):
+    while True:
+        yield start
+        start += incr
+
+
 class TimeUtilsTest(test_base.BaseTestCase):
 
     def setUp(self):
@@ -344,3 +350,159 @@ class TestIso8601Time(test_base.BaseTestCase):
         dtn = datetime.datetime(2011, 2, 14, 19, 53, 7)
         naive = timeutils.normalize_time(dtn)
         self.assertTrue(naive < dt)
+
+
+class StopWatchTest(test_base.BaseTestCase):
+    def test_leftover_no_duration(self):
+        watch = timeutils.StopWatch()
+        watch.start()
+        self.assertRaises(RuntimeError, watch.leftover)
+        self.assertRaises(RuntimeError, watch.leftover, return_none=False)
+        self.assertIsNone(watch.leftover(return_none=True))
+
+    def test_no_states(self):
+        watch = timeutils.StopWatch()
+        self.assertRaises(RuntimeError, watch.stop)
+        self.assertRaises(RuntimeError, watch.resume)
+
+    def test_bad_expiry(self):
+        self.assertRaises(ValueError, timeutils.StopWatch, -1)
+
+    @mock.patch('oslo_utils.timeutils.now')
+    def test_backwards(self, mock_now):
+        mock_now.side_effect = [0, 0.5, -1.0, -1.0]
+        watch = timeutils.StopWatch(0.1)
+        watch.start()
+        self.assertTrue(watch.expired())
+        self.assertFalse(watch.expired())
+        self.assertEqual(0.0, watch.elapsed())
+
+    @mock.patch('oslo_utils.timeutils.now')
+    def test_expiry(self, mock_now):
+        mock_now.side_effect = monotonic_iter(incr=0.2)
+        watch = timeutils.StopWatch(0.1)
+        watch.start()
+        self.assertTrue(watch.expired())
+
+    @mock.patch('oslo_utils.timeutils.now')
+    def test_not_expired(self, mock_now):
+        mock_now.side_effect = monotonic_iter()
+        watch = timeutils.StopWatch(0.1)
+        watch.start()
+        self.assertFalse(watch.expired())
+
+    def test_has_started_stopped(self):
+        watch = timeutils.StopWatch()
+        self.assertFalse(watch.has_started())
+        self.assertFalse(watch.has_stopped())
+        watch.start()
+
+        self.assertTrue(watch.has_started())
+        self.assertFalse(watch.has_stopped())
+
+        watch.stop()
+        self.assertTrue(watch.has_stopped())
+        self.assertFalse(watch.has_started())
+
+    def test_no_expiry(self):
+        watch = timeutils.StopWatch(0.1)
+        self.assertRaises(RuntimeError, watch.expired)
+
+    @mock.patch('oslo_utils.timeutils.now')
+    def test_elapsed(self, mock_now):
+        mock_now.side_effect = monotonic_iter(incr=0.2)
+        watch = timeutils.StopWatch()
+        watch.start()
+        matcher = matchers.GreaterThan(0.19)
+        self.assertThat(watch.elapsed(), matcher)
+
+    def test_no_elapsed(self):
+        watch = timeutils.StopWatch()
+        self.assertRaises(RuntimeError, watch.elapsed)
+
+    def test_no_leftover(self):
+        watch = timeutils.StopWatch()
+        self.assertRaises(RuntimeError, watch.leftover)
+        watch = timeutils.StopWatch(1)
+        self.assertRaises(RuntimeError, watch.leftover)
+
+    @mock.patch('oslo_utils.timeutils.now')
+    def test_pause_resume(self, mock_now):
+        mock_now.side_effect = monotonic_iter()
+        watch = timeutils.StopWatch()
+        watch.start()
+        watch.stop()
+        elapsed = watch.elapsed()
+        self.assertAlmostEqual(elapsed, watch.elapsed())
+        watch.resume()
+        self.assertNotEqual(elapsed, watch.elapsed())
+
+    @mock.patch('oslo_utils.timeutils.now')
+    def test_context_manager(self, mock_now):
+        mock_now.side_effect = monotonic_iter()
+        with timeutils.StopWatch() as watch:
+            pass
+        matcher = matchers.GreaterThan(0.04)
+        self.assertThat(watch.elapsed(), matcher)
+
+    @mock.patch('oslo_utils.timeutils.now')
+    def test_context_manager_splits(self, mock_now):
+        mock_now.side_effect = monotonic_iter()
+        with timeutils.StopWatch() as watch:
+            time.sleep(0.01)
+            watch.split()
+        self.assertRaises(RuntimeError, watch.split)
+        self.assertEqual(1, len(watch.splits))
+
+    def test_splits_stopped(self):
+        watch = timeutils.StopWatch()
+        watch.start()
+        watch.split()
+        watch.stop()
+        self.assertRaises(RuntimeError, watch.split)
+
+    def test_splits_never_started(self):
+        watch = timeutils.StopWatch()
+        self.assertRaises(RuntimeError, watch.split)
+
+    @mock.patch('oslo_utils.timeutils.now')
+    def test_splits(self, mock_now):
+        mock_now.side_effect = monotonic_iter()
+
+        watch = timeutils.StopWatch()
+        watch.start()
+        self.assertEqual(0, len(watch.splits))
+
+        watch.split()
+        self.assertEqual(1, len(watch.splits))
+        self.assertEqual(watch.splits[0].elapsed,
+                         watch.splits[0].length)
+
+        watch.split()
+        splits = watch.splits
+        self.assertEqual(2, len(splits))
+        self.assertNotEqual(splits[0].elapsed, splits[1].elapsed)
+        self.assertEqual(splits[1].length,
+                         splits[1].elapsed - splits[0].elapsed)
+
+        watch.stop()
+        self.assertEqual(2, len(watch.splits))
+
+        watch.start()
+        self.assertEqual(0, len(watch.splits))
+
+    @mock.patch('oslo_utils.timeutils.now')
+    def test_elapsed_maximum(self, mock_now):
+        mock_now.side_effect = [0, 1] + ([11] * 4)
+
+        watch = timeutils.StopWatch()
+        watch.start()
+        self.assertEqual(1, watch.elapsed())
+
+        self.assertEqual(11, watch.elapsed())
+        self.assertEqual(1, watch.elapsed(maximum=1))
+
+        watch.stop()
+        self.assertEqual(11, watch.elapsed())
+        self.assertEqual(11, watch.elapsed())
+        self.assertEqual(0, watch.elapsed(maximum=-1))
diff --git a/oslo_utils/timeutils.py b/oslo_utils/timeutils.py
index c7c3d622..437b729b 100644
--- a/oslo_utils/timeutils.py
+++ b/oslo_utils/timeutils.py
@@ -24,12 +24,26 @@ import time
 import iso8601
 import six
 
+from oslo_utils import reflection
 
 # ISO 8601 extended time format with microseconds
 _ISO8601_TIME_FORMAT_SUBSECOND = '%Y-%m-%dT%H:%M:%S.%f'
 _ISO8601_TIME_FORMAT = '%Y-%m-%dT%H:%M:%S'
 PERFECT_TIME_FORMAT = _ISO8601_TIME_FORMAT_SUBSECOND
 
+# Use monotonic time in stopwatches if we can get at it...
+#
+# PEP @ https://www.python.org/dev/peps/pep-0418/
+try:
+    now = time.monotonic
+except AttributeError:
+    try:
+        # Try to use the pypi module if it's available (optionally...)
+        from monotonic import monotonic as now
+    except (AttributeError, ImportError):
+        # Ok fallback to the non-monotonic one...
+        now = time.time
+
 
 def isotime(at=None, subsecond=False):
     """Stringify time in ISO 8601 format."""
@@ -239,3 +253,178 @@ def is_soon(dt, window):
     """
     soon = (utcnow() + datetime.timedelta(seconds=window))
     return normalize_time(dt) <= soon
+
+
+class Split(object):
+    """A *immutable* stopwatch split.
+
+    See: http://en.wikipedia.org/wiki/Stopwatch for what this is/represents.
+    """
+
+    __slots__ = ['_elapsed', '_length']
+
+    def __init__(self, elapsed, length):
+        self._elapsed = elapsed
+        self._length = length
+
+    @property
+    def elapsed(self):
+        """Duration from stopwatch start."""
+        return self._elapsed
+
+    @property
+    def length(self):
+        """Seconds from last split (or the elapsed time if no prior split)."""
+        return self._length
+
+    def __repr__(self):
+        r = reflection.get_class_name(self, fully_qualified=False)
+        r += "(elapsed=%s, length=%s)" % (self._elapsed, self._length)
+        return r
+
+
+class StopWatch(object):
+    """A simple timer/stopwatch helper class.
+
+    Inspired by: apache-commons-lang java stopwatch.
+
+    Not thread-safe (when a single watch is mutated by multiple threads at
+    the same time). Thread-safe when used by a single thread (not shared) or
+    when operations are performed in a thread-safe manner on these objects by
+    wrapping those operations with locks.
+    """
+    _STARTED = 'STARTED'
+    _STOPPED = 'STOPPED'
+
+    def __init__(self, duration=None):
+        if duration is not None and duration < 0:
+            raise ValueError("Duration must be greater or equal to"
+                             " zero and not %s" % duration)
+        self._duration = duration
+        self._started_at = None
+        self._stopped_at = None
+        self._state = None
+        self._splits = []
+
+    def start(self):
+        """Starts the watch (if not already started).
+
+        NOTE(harlowja): resets any splits previously captured (if any).
+        """
+        if self._state == self._STARTED:
+            return self
+        self._started_at = now()
+        self._stopped_at = None
+        self._state = self._STARTED
+        self._splits = []
+        return self
+
+    @property
+    def splits(self):
+        """Accessor to all/any splits that have been captured."""
+        return tuple(self._splits)
+
+    def split(self):
+        """Captures a split/elapsed since start time (and doesn't stop)."""
+        if self._state == self._STARTED:
+            elapsed = self.elapsed()
+            if self._splits:
+                length = self._delta_seconds(self._splits[-1].elapsed, elapsed)
+            else:
+                length = elapsed
+            self._splits.append(Split(elapsed, length))
+            return self._splits[-1]
+        else:
+            raise RuntimeError("Can not create a split time of a stopwatch"
+                               " if it has not been started or if it has been"
+                               " stopped")
+
+    def restart(self):
+        """Restarts the watch from a started/stopped state."""
+        if self._state == self._STARTED:
+            self.stop()
+        self.start()
+        return self
+
+    @staticmethod
+    def _delta_seconds(earlier, later):
+        # Uses max to avoid the delta/time going backwards (and thus negative).
+        return max(0.0, later - earlier)
+
+    def elapsed(self, maximum=None):
+        """Returns how many seconds have elapsed."""
+        if self._state not in (self._STARTED, self._STOPPED):
+            raise RuntimeError("Can not get the elapsed time of a stopwatch"
+                               " if it has not been started/stopped")
+        if self._state == self._STOPPED:
+            elapsed = self._delta_seconds(self._started_at, self._stopped_at)
+        else:
+            elapsed = self._delta_seconds(self._started_at, now())
+        if maximum is not None and elapsed > maximum:
+            elapsed = max(0.0, maximum)
+        return elapsed
+
+    def __enter__(self):
+        """Starts the watch."""
+        self.start()
+        return self
+
+    def __exit__(self, type, value, traceback):
+        """Stops the watch (ignoring errors if stop fails)."""
+        try:
+            self.stop()
+        except RuntimeError:
+            pass
+
+    def leftover(self, return_none=False):
+        """Returns how many seconds are left until the watch expires.
+
+        :param return_none: when ``True`` instead of raising a ``RuntimeError``
+                            when no duration has been set this call will
+                            return ``None`` instead.
+        :type return_none: boolean
+        """
+        if self._state != self._STARTED:
+            raise RuntimeError("Can not get the leftover time of a stopwatch"
+                               " that has not been started")
+        if self._duration is None:
+            if not return_none:
+                raise RuntimeError("Can not get the leftover time of a watch"
+                                   " that has no duration")
+            return None
+        return max(0.0, self._duration - self.elapsed())
+
+    def expired(self):
+        """Returns if the watch has expired (ie, duration provided elapsed)."""
+        if self._state not in (self._STARTED, self._STOPPED):
+            raise RuntimeError("Can not check if a stopwatch has expired"
+                               " if it has not been started/stopped")
+        if self._duration is None:
+            return False
+        return self.elapsed() > self._duration
+
+    def has_started(self):
+        return self._state == self._STARTED
+
+    def has_stopped(self):
+        return self._state == self._STOPPED
+
+    def resume(self):
+        """Resumes the watch from a stopped state."""
+        if self._state == self._STOPPED:
+            self._state = self._STARTED
+            return self
+        else:
+            raise RuntimeError("Can not resume a stopwatch that has not been"
+                               " stopped")
+
+    def stop(self):
+        """Stops the watch."""
+        if self._state == self._STOPPED:
+            return self
+        if self._state != self._STARTED:
+            raise RuntimeError("Can not stop a stopwatch that has not been"
+                               " started")
+        self._stopped_at = now()
+        self._state = self._STOPPED
+        return self