Handle multiple modifications to distribution metrics correctly.

infra_libs/ts_mon/common/metric_store.py

 # Copyright 2015 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 import collections
 import logging
-import operator
 import threading
 import time
 
 from infra_libs.ts_mon.common import errors
 
 
 """A light-weight representation of a set or an incr.
 
 Args:
   name: The metric name.
   fields: The normalized field tuple.
   mod_type: Either 'set' or 'incr'.  Other values will raise
             UnknownModificationTypeError when it's used.
   args: (value, enforce_ge) for 'set' or (delta, modify_fn) for 'incr'.
 """  # pylint: disable=pointless-string-statement
 Modification = collections.namedtuple(
     'Modification', ['name', 'fields', 'mod_type', 'args'])
 
 
-def combine_modifications(old, new):
-  """Combines two modifications into one.
-
-  The returned modification will be the result as if the second modification had
-  been applied after the first.
-  """
-
-  if old is None or new.mod_type == 'set':
-    # A 'set' will override any previous value.
-    return new
-  elif new.mod_type == 'incr':
-    # For two 'incr's sum their delta args, for an 'incr' on top of a 'set' add
-    # the delta to the set value.
-    return Modification(
-        old.name, old.fields, old.mod_type,
-        (old.args[0] + new.args[0], old.args[1]))
-  else:
-    raise errors.UnknownModificationTypeError(new.mod_type)
+def default_modify_fn(name):
+  def _modify_fn(value, delta):
+    if delta < 0:
+      raise errors.MonitoringDecreasingValueError(name, None, delta)
+    return value + delta
+  return _modify_fn
 
 
 class MetricStore(object):
   """A place to store values for each metric.
 
   Several methods take "a normalized field tuple".  This is a tuple of
   (key, value) tuples sorted by key.  (The reason this is given as a tuple
   instead of a dict is because tuples are hashable and can be used as dict keys,
   dicts can not).
 
@@ skipping 48 matching lines @@
       value: the new value for the metric.
       enforce_ge: if this is True, raise an exception if the new value is
           less than the old value.
 
     Raises:
       MonitoringDecreasingValueError: if enforce_ge is True and the new value is
           smaller than the old value.
     """
     raise NotImplementedError
 
-  def incr(self, name, fields, delta, modify_fn=operator.add):
+  def incr(self, name, fields, delta, modify_fn=None):
     """Increments the metric's value.
 
     Args:
       name: the metric's name.
       fields: a normalized field tuple.
       delta: how much to increment the value by.
       modify_fn: this function is called with the original value and the delta
           as its arguments and is expected to return the new value.  The
           function must be idempotent as it may be called multiple times.
     """
@@ skipping 59 matching lines @@
 
   def set(self, name, fields, value, enforce_ge=False):
     with self._thread_lock:
       if enforce_ge:
         old_value = self._entry(name)[1].get(fields, 0)
         if value < old_value:
           raise errors.MonitoringDecreasingValueError(name, old_value, value)
 
       self._entry(name)[1][fields] = value
 
-  def incr(self, name, fields, delta, modify_fn=operator.add):
+  def incr(self, name, fields, delta, modify_fn=None):
     if delta < 0:
       raise errors.MonitoringDecreasingValueError(name, None, delta)
 
+    if modify_fn is None:
+      modify_fn = default_modify_fn(name)
+
     with self._thread_lock:
       self._entry(name)[1][fields] = modify_fn(self.get(name, fields, 0), delta)
 
   def modify_multi(self, modifications):
     # This is only used by DeferredMetricStore on top of MemcacheMetricStore,
     # but could be implemented here if required in the future.
     raise NotImplementedError
 
   def reset_for_unittest(self, name=None):
     if name is not None:
       self._reset(name)
     else:
       for name in self._values.keys():
         self._reset(name)
 
   def _reset(self, name):
     self._values[name] = (self._start_time(name), {})