Merge pull request #1 from jesushernandez/retries

Tweak timer logic to support retries

.circle/lint-all.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+
+set -x
+set -e
+
+# Flake the entire project using setup.cfg with specific settings.
+flake8 --config=../setup.cfg --count celery_statsd/ tests/

.circle/py_tests.sh

@@ -0,0 +1,5 @@
+#!/bin/bash
+set -x
+set -e
+
+tox tests/

.gitignore

@@ -1,3 +1,8 @@
-/build/
-/dist/
-*.egg-info
+dist/
+.tox
+*.egg-info/
+*.pyc
+.idea
+.venv
+.cache/
+.eggs

celery_statsd/__about__.py

@@ -7,9 +7,9 @@
 __summary__ = (
     "Send various info to statsd for each Celery task"
 )
-__uri__ = "https://github.com/ssaw/celery-statsd"
+__uri__ = "https://github.com/lyst/celery-statsd/"
 
-__version__ = "0.1.2"
+__version__ = "0.1.3"
 
 __author__ = "Lyst LTD"
 __email__ = "[email protected]"

celery_statsd/__init__.py

@@ -18,9 +18,9 @@ def task_key(task):
                      "CELERY_STATSD_PREFIX", "celery.")
 
     if isinstance(task, six.string_types):
-        return prefix + task
+        return '{}{}'.format(prefix, task)
     else:
-        return prefix + task.name
+        return '{}{}'.format(prefix, task.name)
 
 
 def get_client(celery_app):
@@ -46,14 +46,26 @@ def get_client(celery_app):
 
 def start_timer(name, group, instance):
     try:
-        assert (name, group, instance) not in _state.timers
         _state.timers[(name, group, instance)] = time.time()
     except AttributeError:
         _state.timers = {(name, group, instance): time.time()}
 
 
+def _get_timer(name, group, instance):
+    try:
+        return _state.timers.pop((name, group, instance))
+    except (AttributeError, KeyError):
+        return
+
+
 def stop_timer(name, group, instance):
-    total = time.time() - _state.timers.pop((name, group, instance))
+
+    start = _get_timer(name, group, instance)
+
+    if start is None:
+        return
+
+    total = time.time() - start
 
     get_client(celery.current_app).timing(
         "{0}.{1}".format(group, name),

circle.yml

@@ -0,0 +1,24 @@
+checkout:
+    post:
+    - |
+        git clean -fdx
+        git fetch origin master
+        git branch
+        # Merge with master if this commit is not merged yet
+        if ! git merge-base --is-ancestor ${CIRCLE_SHA1} origin/master ; then
+            git config --global user.email "[email protected]"
+            git config --global user.name "CircleCI"
+            git checkout -f origin/master
+            echo Merging ${CIRCLE_SHA1} into origin/master at $(git rev-parse origin/master)
+            git merge -m Auto-merge --no-ff $CIRCLE_SHA1
+        fi
+
+dependencies:
+    override:
+        - pip install coverage flake8 flake8-import-order
+        - pip install -r requirements.txt
+
+test:
+  override:
+    - ./.circle/lint-all.sh
+    - ./.circle/py_tests.sh

requirements.txt

@@ -0,0 +1,6 @@
+celery==3.1.17
+mock==1.1.3
+pytest==2.7.2
+six==1.9.0
+statsd==3.0
+tox==2.2

setup.cfg

@@ -0,0 +1,7 @@
+[flake8]
+max-line-length = 100
+select = E,W,F,I
+application-import-names = celery_statsd
+
+[bdist_wheel]
+universal = 1

setup.py

@@ -21,9 +21,4 @@
 
     packages=find_packages(),
     zip_safe=False,
-
-    install_requires=[
-        'six',
-        'statsd>=2.0.0'
-    ]
 )

tests/test_celery_statsd.py

@@ -1,9 +1,9 @@
 from __future__ import absolute_import
 
-import mock
-
 import celery
 
+import mock
+
 import pytest
 
 import celery_statsd
@@ -17,6 +17,11 @@ def _stub_task(arg):
     return arg
 
 
+@celery.task(bind=True, max_retries=2)
+def _stub_task_with_retries(self, arg):
+    self.retry()
+
+
 @pytest.fixture
 def mock_client(monkeypatch):
     mock_client = mock.Mock()
@@ -36,3 +41,24 @@ def test_run(mock_client):
         "celery.tests.test_celery_statsd._stub_task.run", mock.ANY)
     mock_client.incr.assert_called_with(
         "celery.tests.test_celery_statsd._stub_task.success")
+
+
+def test_run_with_retry(mock_client, monkeypatch):
+    task = _stub_task_with_retries
+    task.apply_async = mock.Mock(wraps=_stub_task_with_retries.apply_async)
+
+    get_timer_mock = mock.Mock(wraps=celery_statsd._get_timer)
+
+    monkeypatch.setattr('celery_statsd._get_timer', get_timer_mock)
+
+    task.delay(1)
+
+    # Only one call to the timing (the last task)
+    assert mock_client.timing.call_count == 1
+
+    # The name of the metric does not change among retried tasks
+    mock_client.timing.assert_called_with(
+        "celery.tests.test_celery_statsd._stub_task_with_retries.run", mock.ANY)
+
+    # We get the timer 3 times
+    assert get_timer_mock.call_count == 3

tox.ini

@@ -0,0 +1,6 @@
+[tox]
+envlist = py27,py34
+
+[testenv]
+deps=-rrequirements.txt
+commands= py.test {posargs}