| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you 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. |
| |
| from __future__ import unicode_literals |
| from __future__ import absolute_import |
| import os |
| import time |
| import signal |
| import socket |
| import subprocess |
| from ming.orm.ormsession import ThreadLocalORMSession |
| import six |
| |
| from allura import model as M |
| from . import base |
| from six.moves import range |
| |
| |
| class TaskdCleanupCommand(base.Command): |
| summary = 'Tasks cleanup command. Determines which taskd processes are handling tasks, and what has been dropped or got hung.' |
| parser = base.Command.standard_parser(verbose=True) |
| parser.add_option('-k', '--kill-stuck-taskd', |
| dest='kill', action='store_true', |
| help='automatically kill stuck taskd processes. Be careful with this, a taskd process ' |
| 'may just be very busy on certain operations and not able to respond to our status request') |
| parser.add_option('-n', '--num-retry-status-check', |
| dest='num_retry', type='int', default=5, |
| help='number of retries to read taskd status log after sending USR1 signal (5 by default)') |
| usage = '<ini file> [-k] <taskd status log file>' |
| min_args = 2 |
| max_args = 2 |
| |
| def command(self): |
| self.basic_setup() |
| self.hostname = socket.gethostname() |
| self.taskd_status_log = self.args[1] |
| self.stuck_pids = [] |
| self.error_tasks = [] |
| self.suspicious_tasks = [] |
| |
| taskd_pids = self._taskd_pids() |
| base.log.info('Taskd processes on %s: %s' % |
| (self.hostname, taskd_pids)) |
| |
| # find stuck taskd processes |
| base.log.info('Seeking for stuck taskd processes') |
| for pid in taskd_pids: |
| base.log.info('...sending USR1 to %s and watching status log' % |
| (pid)) |
| status = self._check_taskd_status(int(pid)) |
| if status != 'OK': |
| base.log.info('...taskd pid %s has stuck' % pid) |
| self.stuck_pids.append(pid) |
| if self.options.kill: |
| base.log.info('...-k is set. Killing %s' % pid) |
| self._kill_stuck_taskd(pid) |
| else: |
| base.log.info('...%s' % status) |
| |
| # find 'forsaken' tasks |
| base.log.info('Seeking for forsaken busy tasks') |
| tasks = [t for t in self._busy_tasks() |
| if t not in self.error_tasks] # skip seen tasks |
| base.log.info('Found %s busy tasks on %s' % |
| (len(tasks), self.hostname)) |
| for task in tasks: |
| base.log.info('Verifying task %s' % task) |
| pid = task.process.split()[-1] |
| if pid not in taskd_pids: |
| # 'forsaken' task |
| base.log.info('Task is forsaken ' |
| '(can\'t find taskd with given pid). ' |
| 'Setting state to \'error\'') |
| task.state = 'error' |
| task.result = 'Can\'t find taskd with given pid' |
| self.error_tasks.append(task) |
| else: |
| # check if taskd with given pid really processing this task |
| # now: |
| base.log.info( |
| 'Checking that taskd pid %s is really processing task %s' % |
| (pid, task._id)) |
| status = self._check_task(pid, task) |
| if status != 'OK': |
| # maybe task moved quickly and now is complete |
| # so we need to check such tasks later |
| # and mark incomplete ones as 'error' |
| self.suspicious_tasks.append(task) |
| base.log.info('...NO. Adding task to suspisious list') |
| else: |
| base.log.info('...OK') |
| |
| # check suspicious task and mark incomplete as error |
| base.log.info('Checking suspicious list for incomplete tasks') |
| self._check_suspicious_tasks() |
| ThreadLocalORMSession.flush_all() |
| self.print_summary() |
| |
| def print_summary(self): |
| base.log.info('-' * 80) |
| if self.stuck_pids: |
| base.log.info('Found stuck taskd: %s' % self.stuck_pids) |
| if self.options.kill: |
| base.log.info('...stuck taskd processes were killed') |
| else: |
| base.log.info( |
| '...to kill these processes run command with -k flag if you are sure they are really stuck') |
| if self.error_tasks: |
| base.log.info('Tasks marked as \'error\': %s' % self.error_tasks) |
| |
| def _busy_tasks(self, pid=None): |
| regex = '^%s ' % self.hostname |
| if pid is not None: |
| regex = '^%s pid %s' % (self.hostname, pid) |
| return M.MonQTask.query.find({ |
| 'state': 'busy', |
| 'process': {'$regex': regex} |
| }) |
| |
| def _taskd_pids(self): |
| # space or colon after "taskd" to ensure no match on taskd_cleanup (ourself) |
| p = subprocess.Popen(['pgrep', '-f', '^taskd[ :]'], |
| stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE) |
| stdout, stderr = p.communicate() |
| tasks = [] |
| if p.returncode == 0: |
| tasks = [pid for pid in six.ensure_text(stdout).split('\n') if pid != ''] |
| return tasks |
| |
| def _taskd_status(self, pid, retry=False): |
| if not retry: |
| os.kill(int(pid), signal.SIGUSR1) |
| p = subprocess.Popen(['tail', '-n1', self.taskd_status_log], |
| stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE) |
| stdout, stderr = p.communicate() |
| if p.returncode != 0: |
| base.log.error('Can\'t read taskd status log %s' % |
| self.taskd_status_log) |
| exit(1) |
| return six.ensure_text(stdout) |
| |
| def _check_taskd_status(self, pid): |
| for i in range(self.options.num_retry): |
| retry = False if i == 0 else True |
| status = self._taskd_status(pid, retry) |
| if ('taskd pid %s' % pid) in status: |
| return 'OK' |
| base.log.info('retrying after one second') |
| time.sleep(1) |
| return 'STUCK' |
| |
| def _check_task(self, taskd_pid, task): |
| for i in range(self.options.num_retry): |
| retry = False if i == 0 else True |
| status = self._taskd_status(taskd_pid, retry) |
| line = 'taskd pid %s is currently handling task %s' % ( |
| taskd_pid, task) |
| if line in status: |
| return 'OK' |
| base.log.info('retrying after one second') |
| time.sleep(1) |
| return 'FAIL' |
| |
| def _kill_stuck_taskd(self, pid): |
| os.kill(int(pid), signal.SIGKILL) |
| # find all 'busy' tasks for this pid and mark them as 'error' |
| tasks = list(self._busy_tasks(pid=pid)) |
| base.log.info('...taskd pid %s has assigned tasks: %s. ' |
| 'setting state to \'error\' for all of them' % (pid, tasks)) |
| for task in tasks: |
| task.state = 'error' |
| task.result = 'Taskd has stuck with this task' |
| self.error_tasks.append(task) |
| |
| def _complete_suspicious_tasks(self): |
| complete_tasks = M.MonQTask.query.find({ |
| 'state': 'complete', |
| '_id': {'$in': [t._id for t in self.suspicious_tasks]} |
| }) |
| return [t._id for t in complete_tasks] |
| |
| def _check_suspicious_tasks(self): |
| if not self.suspicious_tasks: |
| return |
| complete_tasks = self._complete_suspicious_tasks() |
| for task in self.suspicious_tasks: |
| base.log.info('Verifying task %s' % task) |
| if task._id not in complete_tasks: |
| base.log.info('...incomplete. Setting status to \'error\'') |
| task.state = 'error' |
| task.result = 'Forsaken task' |
| self.error_tasks.append(task) |
| else: |
| base.log.info('...complete') |
