ccplugin.py 3.9 KB
Newer Older
Philippe Pepiot's avatar
Philippe Pepiot committed
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# -*- coding: utf-8 -*-
# copyright 2016 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
# contact http://www.logilab.fr -- mailto:contact@logilab.fr
#
# This program is free software: you can redistribute it and/or modify it under
# the terms of the GNU Lesser General Public License as published by the Free
# Software Foundation, either version 2.1 of the License, or (at your option)
# any later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
# details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

18
import json
19
20
21
22
import logging
import time

import redis.exceptions
Philippe Pepiot's avatar
Philippe Pepiot committed
23
24
25
26
27

from cubicweb.toolsutils import Command
from cubicweb.server.serverconfig import ServerConfiguration
from cubicweb.cwctl import CWCTL

28
29
30
from cw_celerytask_helpers.monitor import MONITOR_KEY
from cw_celerytask_helpers.utils import get_redis_client

31
32
logger = logging.getLogger(__name__)

Philippe Pepiot's avatar
Philippe Pepiot committed
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

class CeleryMonitorCommand(Command):
    """Synchronize celery task statuses"""

    name = 'celery-monitor'
    arguments = '<instance>'
    min_args = max_args = 1
    options = (
        ('loglevel',
         {'short': 'l', 'type': 'choice', 'metavar': '<log level>',
          'default': 'info', 'choices': ('debug', 'info', 'warning', 'error')},
         ),
    )

    def run(self, args):
        from cubicweb import repoapi
        from cubicweb.cwctl import init_cmdline_log_threshold
        config = ServerConfiguration.config_for(args[0])
        config.global_set_option('log-file', None)
        config.log_format = '%(levelname)s %(name)s: %(message)s'
        init_cmdline_log_threshold(config, self['loglevel'])
        repo = repoapi.get_repository(config=config)
        repo.hm.call_hooks('server_maintenance', repo=repo)
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
        source = repo.system_source
        while True:
            try:
                with repo.internal_cnx() as cnx:
                    while True:
                        try:
                            self.loop(cnx)
                        except redis.exceptions.ConnectionError:
                            logger.error('redis ConnectionError, retry in 5s')
                            time.sleep(5)
            except (source.OperationalError, source.InterfaceError):
                while True:
                    source.exception('connection lost, retry in 5s')
                    time.sleep(5)
                    try:
                        for cnxset in repo.cnxsets:
                            cnxset.reconnect()
                    except (source.OperationalError, source.InterfaceError):
                        continue
                    else:
                        break
Philippe Pepiot's avatar
Philippe Pepiot committed
77
78

    @staticmethod
79
80
    def loop(cnx, timeout=None):
        client = get_redis_client()
81
82
        client.ping()
        logger.info('Connected to redis')
83
        test = (cnx.repo.config.mode == "test")
Philippe Pepiot's avatar
Philippe Pepiot committed
84
        while True:
85
86
            data = client.brpop(MONITOR_KEY, timeout=timeout)
            if data is None:
Philippe Pepiot's avatar
Philippe Pepiot committed
87
                break
88
89
90
91
92
93
94
95
96
97
98
99
100
101
            data = json.loads(data[1].decode())
            task_id, task_name = data['task_id'], data['task_name']
            for adapter in cnx.vreg['adapters']['ICeleryTask']:
                try:
                    adapter.sync_task_state(cnx, task_id,
                                            task_name=task_name)
                except Exception:
                    adapter.exception('Unhandled exception while syncing '
                                      'task <Task %s (%s)>', task_id,
                                      task_name)
                    cnx.rollback()
                    if test:
                        # we should not hide exceptions in tests
                        raise
Philippe Pepiot's avatar
Philippe Pepiot committed
102
103
104


CWCTL.register(CeleryMonitorCommand)