<commit_before># vim: tabstop=4 shiftwidth=4 softtabstop=4

#    Copyright 2013 Rackspace
#
#    Licensed 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.

import calendar
import datetime
from operator import attrgetter
import time

from novaclient import exceptions
from oslo.config import cfg

from qonos.common import exception as exc
from qonos.common import timeutils
from qonos.openstack.common.gettextutils import _
from qonos.openstack.common import importutils
import qonos.openstack.common.log as logging
import qonos.qonosclient.exception as qonos_ex
from qonos.worker import worker


LOG = logging.getLogger(__name__)

snapshot_worker_opts = [
    cfg.StrOpt('nova_client_factory_class',
               default='qonos.worker.snapshot.simple_nova_client_factory.'
                       'NovaClientFactory'),
    cfg.IntOpt('image_poll_interval_sec', default=30,
               help=_('How often to poll Nova for the image status')),
    cfg.IntOpt('job_update_interval_sec', default=300,
               help=_('How often to update the job status, in seconds')),
    cfg.IntOpt('job_timeout_update_interval_min', default=60,
               help=_('How often to update the job timeout, in minutes')),
    cfg.IntOpt('job_timeout_update_increment_min', default=60,
               help=_('How much to increment the timeout, in minutes')),
    cfg.IntOpt('job_timeout_max_updates', default=3,
               help=_('How many times to update the timeout before '
                      'considering the job to be failed')),
    cfg.IntOpt('job_timeout_backoff_factor', default=2,
               help=_('Timeout multiplier to use when an error occurs')),
]

CONF = cfg.CONF
CONF.register_opts(snapshot_worker_opts, group='snapshot_worker')


class SnapshotProcessor(worker.JobProcessor):
    def __init__(self):
        super(SnapshotProcessor, self).__init__()
        self.status_map = {
            "QUEUED": "PROCESSING",
            "SAVING": "PROCESSING",
            "ACTIVE": "DONE",
            "KILLED": "ERROR",
            "DELETED": "ERROR",
            "PENDING_DELETE": "ERROR",
            "ERROR": "ERROR"
        }

    def init_processor(self, worker, nova_client_factory=None):
        super(SnapshotProcessor, self).init_processor(worker)
        self.current_job = None
        self.timeout_count = 0
        self.timeout_max_updates = CONF.snapshot_worker.job_timeout_max_updates
        self.next_timeout = None
        self.update_interval = datetime.timedelta(
            seconds=CONF.snapshot_worker.job_update_interval_sec)
        self.timeout_increment = datetime.timedelta(
            minutes=CONF.snapshot_worker.job_timeout_update_increment_min)
        self.image_poll_interval = CONF.snapshot_worker.image_poll_interval_sec
        self.job_timeout_backoff_factor = (CONF.snapshot_worker
                                           .job_timeout_backoff_factor)

        if not nova_client_factory:
            nova_client_factory = importutils.import_object(
                CONF.snapshot_worker.nova_client_factory_class)
        self.nova_client_factory = nova_client_factory

    def process_job(self, job):
        LOG.info(_("Worker %(worker_id)s Processing job: %(job)s") %
                    {'worker_id': self.worker.worker_id,
                     'job': job['id']})
        LOG.debug(_("Worker %(worker_id)s Processing job: %(job)s") %
                    {'worker_id': self.worker.worker_id,
                     'job': str(job)})

        payload = {'job': job}
        if job['status'] == 'QUEUED':
            self.send_notification_start(payload)
        else:
            self.send_notification_retry(payload)

        job_id = job['id']

        if not self._check_schedule_exists(job):
            msg = ('Schedule %(schedule_id)s deleted for job %(job_id)s' %
                   {'schedule_id': job['schedule_id'], 'job_id': job_id})
            self._job_cancelled(job_id, msg)

            LOG.info(_('Worker %(worker_id)s Job cancelled: %(msg)s') %
                      {'worker_id': self.worker.worker_id,
                       'msg': msg})
            return

        self.current_job = job

        now = self._get_utcnow()
        self.next_timeout = now + self.timeout_increment
        self.update_job(job_id, 'PROCESSING', timeout=self.next_timeout)
        self.next_update = self._get_utcnow() + self.update_interval

        instance_id = self._get_instance_id(job)
        if not instance_id:
            msg = ('Job %s does not specify an instance_id in its metadata.'
                   % job_id)
            self._job_cancelled(job_id, msg)
            return

        should_create = True
        if ('image_id' in job['metadata']):
            image_id = job['metadata']['image_id']
            if (job['status'] in ['PROCESSING', 'TIMED_OUT']):
                should_create = False
                LOG.info(_("Worker %(worker_id)s Resuming image: %(image_id)s")
                         % {'worker_id': self.worker.worker_id,
                            'image_id': image_id})
            elif (job['status'] == 'ERROR'):
                status = self._get_image_status(image_id)
                if (status['job_status'] in ['PROCESSING', 'DONE']):
                    LOG.info(_("Worker %(worker_id)s "
                               "Previous image %(image_id)s failed with status"
                               " %(image_status)s. Retrying.")
                         % {'worker_id': self.worker.worker_id,
                            'image_id': image_id,
                            'image_status': status['image_status']})
                    should_create = False

        if should_create:
            image_id = self._create_image(job_id, instance_id,
                                          job['schedule_id'])
            if image_id is None:
                return

        image_status = None
        active = False
        retry = True

        status = None
        while retry and not active:
            status = self._get_image_status(image_id)
            if status['job_status'] == 'ERROR':
                instance_id = self._get_instance_id(self.current_job)
                msg = _("Error occured while polling snapshot (status ERROR). "
                        "Instance: %s, image_id: %s threw error msg: %s."
                        % (instance_id, image_id, status['error_msg']))
                timeout = self._get_updated_job_timeout(self.current_job['id'])
                self.update_job(job['id'], 'ERROR', timeout=timeout,
                                error_message=msg)
                raise exc.PollingException(msg)

            active = self._is_active_status(status)
            if active:
                self._job_succeeded(job_id)
            else:
                retry = self._try_update(job_id, status['job_status'])

            if not active:
                time.sleep(self.image_poll_interval)

        if (not active) and (not retry):
            self._job_timed_out(job_id)

        if active:
            self._process_retention(instance_id, job['schedule_id'])
            self.send_notification_end(payload)<commit_msg>Sends latest payload into job-end notification<commit_after># vim: tabstop=4 shiftwidth=4 softtabstop=4

#    Copyright 2013 Rackspace
#
#    Licensed 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.

import calendar
import datetime
from operator import attrgetter
import time

from novaclient import exceptions
from oslo.config import cfg

from qonos.common import exception as exc
from qonos.common import timeutils
from qonos.openstack.common.gettextutils import _
from qonos.openstack.common import importutils
import qonos.openstack.common.log as logging
import qonos.qonosclient.exception as qonos_ex
from qonos.worker import worker


LOG = logging.getLogger(__name__)

snapshot_worker_opts = [
    cfg.StrOpt('nova_client_factory_class',
               default='qonos.worker.snapshot.simple_nova_client_factory.'
                       'NovaClientFactory'),
    cfg.IntOpt('image_poll_interval_sec', default=30,
               help=_('How often to poll Nova for the image status')),
    cfg.IntOpt('job_update_interval_sec', default=300,
               help=_('How often to update the job status, in seconds')),
    cfg.IntOpt('job_timeout_update_interval_min', default=60,
               help=_('How often to update the job timeout, in minutes')),
    cfg.IntOpt('job_timeout_update_increment_min', default=60,
               help=_('How much to increment the timeout, in minutes')),
    cfg.IntOpt('job_timeout_max_updates', default=3,
               help=_('How many times to update the timeout before '
                      'considering the job to be failed')),
    cfg.IntOpt('job_timeout_backoff_factor', default=2,
               help=_('Timeout multiplier to use when an error occurs')),
]

CONF = cfg.CONF
CONF.register_opts(snapshot_worker_opts, group='snapshot_worker')


class SnapshotProcessor(worker.JobProcessor):
    def __init__(self):
        super(SnapshotProcessor, self).__init__()
        self.status_map = {
            "QUEUED": "PROCESSING",
            "SAVING": "PROCESSING",
            "ACTIVE": "DONE",
            "KILLED": "ERROR",
            "DELETED": "ERROR",
            "PENDING_DELETE": "ERROR",
            "ERROR": "ERROR"
        }

    def init_processor(self, worker, nova_client_factory=None):
        super(SnapshotProcessor, self).init_processor(worker)
        self.current_job = None
        self.timeout_count = 0
        self.timeout_max_updates = CONF.snapshot_worker.job_timeout_max_updates
        self.next_timeout = None
        self.update_interval = datetime.timedelta(
            seconds=CONF.snapshot_worker.job_update_interval_sec)
        self.timeout_increment = datetime.timedelta(
            minutes=CONF.snapshot_worker.job_timeout_update_increment_min)
        self.image_poll_interval = CONF.snapshot_worker.image_poll_interval_sec
        self.job_timeout_backoff_factor = (CONF.snapshot_worker
                                           .job_timeout_backoff_factor)

        if not nova_client_factory:
            nova_client_factory = importutils.import_object(
                CONF.snapshot_worker.nova_client_factory_class)
        self.nova_client_factory = nova_client_factory

    def process_job(self, job):
        LOG.info(_("Worker %(worker_id)s Processing job: %(job)s") %
                    {'worker_id': self.worker.worker_id,
                     'job': job['id']})
        LOG.debug(_("Worker %(worker_id)s Processing job: %(job)s") %
                    {'worker_id': self.worker.worker_id,
                     'job': str(job)})

        payload = {'job': job}
        if job['status'] == 'QUEUED':
            self.send_notification_start(payload)
        else:
            self.send_notification_retry(payload)

        job_id = job['id']

        if not self._check_schedule_exists(job):
            msg = ('Schedule %(schedule_id)s deleted for job %(job_id)s' %
                   {'schedule_id': job['schedule_id'], 'job_id': job_id})
            self._job_cancelled(job_id, msg)

            LOG.info(_('Worker %(worker_id)s Job cancelled: %(msg)s') %
                      {'worker_id': self.worker.worker_id,
                       'msg': msg})
            return

        self.current_job = job

        now = self._get_utcnow()
        self.next_timeout = now + self.timeout_increment
        self.update_job(job_id, 'PROCESSING', timeout=self.next_timeout)
        self.next_update = self._get_utcnow() + self.update_interval

        instance_id = self._get_instance_id(job)
        if not instance_id:
            msg = ('Job %s does not specify an instance_id in its metadata.'
                   % job_id)
            self._job_cancelled(job_id, msg)
            return

        should_create = True
        if ('image_id' in job['metadata']):
            image_id = job['metadata']['image_id']
            if (job['status'] in ['PROCESSING', 'TIMED_OUT']):
                should_create = False
                LOG.info(_("Worker %(worker_id)s Resuming image: %(image_id)s")
                         % {'worker_id': self.worker.worker_id,
                            'image_id': image_id})
            elif (job['status'] == 'ERROR'):
                status = self._get_image_status(image_id)
                if (status['job_status'] in ['PROCESSING', 'DONE']):
                    LOG.info(_("Worker %(worker_id)s "
                               "Previous image %(image_id)s failed with status"
                               " %(image_status)s. Retrying.")
                         % {'worker_id': self.worker.worker_id,
                            'image_id': image_id,
                            'image_status': status['image_status']})
                    should_create = False

        if should_create:
            image_id = self._create_image(job_id, instance_id,
                                          job['schedule_id'])
            if image_id is None:
                return

        image_status = None
        active = False
        retry = True

        status = None
        while retry and not active:
            status = self._get_image_status(image_id)
            if status['job_status'] == 'ERROR':
                instance_id = self._get_instance_id(self.current_job)
                msg = _("Error occured while polling snapshot (status ERROR). "
                        "Instance: %s, image_id: %s threw error msg: %s."
                        % (instance_id, image_id, status['error_msg']))
                timeout = self._get_updated_job_timeout(self.current_job['id'])
                self.update_job(job['id'], 'ERROR', timeout=timeout,
                                error_message=msg)
                raise exc.PollingException(msg)

            active = self._is_active_status(status)
            if active:
                self._job_succeeded(job_id)
            else:
                retry = self._try_update(job_id, status['job_status'])

            if not active:
                time.sleep(self.image_poll_interval)

        if (not active) and (not retry):
            self._job_timed_out(job_id)

        if active:
            self._process_retention(instance_id, job['schedule_id'])
            payload['job'] = job
            self.send_notification_end(payload)
{'fraction_non_alphanumeric': 0.09762430221991432, 'fraction_numerical': 0.0029858496689601456, 'mean_word_length': 2.1584665846658466, 'pattern_counts': {'":': 14, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '648873', 'n_tokens_mistral': 4367, 'n_tokens_neox': 4271, 'n_words': 990}