<commit_before># -*- encoding: utf-8 -*-
#
# Copyright © 2014-2015 eNovance
#
# Authors: <NAME> <<EMAIL>>
#
# 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 logging
import multiprocessing
import uuid

from concurrent import futures
import iso8601
from oslo_config import cfg
from oslo_serialization import msgpackutils
from oslo_utils import timeutils
from tooz import coordination

from gnocchi import carbonara
from gnocchi import storage

OPTS = [
    cfg.IntOpt('aggregation_workers_number',
               default=None,
               help='Number of workers to run during adding new measures for '
                    'pre-aggregation needs.'),
    cfg.StrOpt('coordination_url',
               secret=True,
               help='Coordination driver URL',
               default="file:///var/lib/gnocchi/locks"),

]

LOG = logging.getLogger(__name__)


class CarbonaraBasedStorageToozLock(object):
    def __init__(self, conf):
        self.coord = coordination.get_coordinator(
            conf.coordination_url,
            str(uuid.uuid4()).encode('ascii'))
        self.coord.start()

    def stop(self):
        self.coord.stop()

    def __call__(self, metric):
        lock_name = b"gnocchi-" + str(metric.id).encode('ascii')
        return self.coord.get_lock(lock_name)


class CarbonaraBasedStorage(storage.StorageDriver):
    MEASURE_PREFIX = "measure"

    def __init__(self, conf):
        super(CarbonaraBasedStorage, self).__init__(conf)
        self.executor = futures.ThreadPoolExecutor(
            max_workers=(conf.aggregation_workers_number or
                         multiprocessing.cpu_count()))

    @staticmethod
    def _lock(metric):
        raise NotImplementedError

    @staticmethod
    def _get_measures(metric, aggregation):
        raise NotImplementedError

    @staticmethod
    def _store_metric_measures(metric, aggregation, data):
        raise NotImplementedError

    def get_measures(self, metric, from_timestamp=None, to_timestamp=None,
                     aggregation='mean', granularity=None):
        super(CarbonaraBasedStorage, self).get_measures(
            metric, from_timestamp, to_timestamp, aggregation)
        archive = self._get_measures_archive(metric, aggregation)
        return [(timestamp.replace(tzinfo=iso8601.iso8601.UTC), r, v)
                for timestamp, r, v
                in archive.fetch(from_timestamp, to_timestamp)
                if granularity is None or r == granularity]

    @staticmethod
    def _log_data_corruption(metric, aggregation):
        LOG.error("Data are corrupted for metric %(metric)s and aggregation "
                  "%(aggregation)s, recreating an empty timeserie." %
                  dict(metric=metric.id, aggregation=aggregation))

    def _get_measures_archive(self, metric, aggregation):
        try:
            contents = self._get_measures(metric, aggregation)
        except (storage.MetricDoesNotExist, storage.AggregationDoesNotExist):
            ts = None
        else:
            try:
                ts = carbonara.TimeSerieArchive.unserialize(contents)
            except ValueError:
                self._log_data_corruption(metric, aggregation)
                ts = None

        if ts is None:
            ts = carbonara.TimeSerieArchive.from_definitions(
                [(v.granularity, v.points)
                 for v in metric.archive_policy.definition],
                aggregation_method=aggregation)
        return ts

    def _add_measures(self, aggregation, metric, timeserie):
        archive = self._get_measures_archive(metric, aggregation)
        archive.update(timeserie)
        self._store_metric_measures(metric, aggregation,
                                    archive.serialize())

    def add_measures(self, metric, measures):
        self._store_measures(metric, msgpackutils.dumps(
            list(map(tuple, measures))))

    @staticmethod
    def _store_measures(metric, data):
        raise NotImplementedError

    @staticmethod
    def _delete_metric(metric):
        raise NotImplementedError

    @staticmethod
    def _list_metric_with_measures_to_process(metric_id):
        raise NotImplementedError

    @staticmethod
    def _pending_measures_to_process_count(metric_id):
        raise NotImplementedError

    def delete_metric(self, metric):
        with self._lock(metric):
            self._delete_metric(metric)

    @staticmethod
    def _unserialize_measures(data):
        return msgpackutils.loads(data)

    def measures_report(self):
        metrics_to_process = self._list_metric_with_measures_to_process()
        return dict(
            (metric_id, self._pending_measures_to_process_count(metric_id))
            for metric_id in metrics_to_process)

    def process_measures(self, indexer):
        metrics_to_process = self._list_metric_with_measures_to_process()
        metrics = indexer.get_metrics(metrics_to_process)
        # This build the list of deleted metrics, i.e. the metrics we have
        # measures to process for but that are not in the indexer anymore.
        deleted_metrics_id = (set(map(uuid.UUID, metrics_to_process))
                              - set(m.id for m in metrics))
        for metric_id in deleted_metrics_id:
            self._delete_unprocessed_measures_for_metric_id(metric_id)
        for metric in metrics:
            lock = self._lock(metric)
            agg_methods = list(metric.archive_policy.aggregation_methods)
            # Do not block if we cannot acquire the lock, that means some other
            # worker is doing the job. We'll just ignore this metric and may
            # get back later to it if needed.
            if lock.acquire(blocking=False):
                try:
                    LOG.debug("Processing measures for %s" % metric)
                    with self._process_measure_for_metric(metric) as measures:
                        try:
                            with timeutils.StopWatch() as sw:
                                raw_measures = self._get_measures(metric,
                                                                  'none')
                                LOG.debug(
                                    "Retrieve unaggregated measures "
                                    "for %s in %.2fs"
                                    % (metric.id, sw.elapsed()))
                        except storage.MetricDoesNotExist:
                            try:
                                self._create_metric(metric)
                            except storage.MetricDoesNotExist:
                                # Created in the mean time, do not worry
                                pass
                            ts = None
                        except storage.AggregationDoesNotExist:<commit_msg>carbonara: fix exception when creating metric<commit_after># -*- encoding: utf-8 -*-
#
# Copyright © 2014-2015 eNovance
#
# Authors: <NAME> <<EMAIL>>
#
# 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 logging
import multiprocessing
import uuid

from concurrent import futures
import iso8601
from oslo_config import cfg
from oslo_serialization import msgpackutils
from oslo_utils import timeutils
from tooz import coordination

from gnocchi import carbonara
from gnocchi import storage

OPTS = [
    cfg.IntOpt('aggregation_workers_number',
               default=None,
               help='Number of workers to run during adding new measures for '
                    'pre-aggregation needs.'),
    cfg.StrOpt('coordination_url',
               secret=True,
               help='Coordination driver URL',
               default="file:///var/lib/gnocchi/locks"),

]

LOG = logging.getLogger(__name__)


class CarbonaraBasedStorageToozLock(object):
    def __init__(self, conf):
        self.coord = coordination.get_coordinator(
            conf.coordination_url,
            str(uuid.uuid4()).encode('ascii'))
        self.coord.start()

    def stop(self):
        self.coord.stop()

    def __call__(self, metric):
        lock_name = b"gnocchi-" + str(metric.id).encode('ascii')
        return self.coord.get_lock(lock_name)


class CarbonaraBasedStorage(storage.StorageDriver):
    MEASURE_PREFIX = "measure"

    def __init__(self, conf):
        super(CarbonaraBasedStorage, self).__init__(conf)
        self.executor = futures.ThreadPoolExecutor(
            max_workers=(conf.aggregation_workers_number or
                         multiprocessing.cpu_count()))

    @staticmethod
    def _lock(metric):
        raise NotImplementedError

    @staticmethod
    def _get_measures(metric, aggregation):
        raise NotImplementedError

    @staticmethod
    def _store_metric_measures(metric, aggregation, data):
        raise NotImplementedError

    def get_measures(self, metric, from_timestamp=None, to_timestamp=None,
                     aggregation='mean', granularity=None):
        super(CarbonaraBasedStorage, self).get_measures(
            metric, from_timestamp, to_timestamp, aggregation)
        archive = self._get_measures_archive(metric, aggregation)
        return [(timestamp.replace(tzinfo=iso8601.iso8601.UTC), r, v)
                for timestamp, r, v
                in archive.fetch(from_timestamp, to_timestamp)
                if granularity is None or r == granularity]

    @staticmethod
    def _log_data_corruption(metric, aggregation):
        LOG.error("Data are corrupted for metric %(metric)s and aggregation "
                  "%(aggregation)s, recreating an empty timeserie." %
                  dict(metric=metric.id, aggregation=aggregation))

    def _get_measures_archive(self, metric, aggregation):
        try:
            contents = self._get_measures(metric, aggregation)
        except (storage.MetricDoesNotExist, storage.AggregationDoesNotExist):
            ts = None
        else:
            try:
                ts = carbonara.TimeSerieArchive.unserialize(contents)
            except ValueError:
                self._log_data_corruption(metric, aggregation)
                ts = None

        if ts is None:
            ts = carbonara.TimeSerieArchive.from_definitions(
                [(v.granularity, v.points)
                 for v in metric.archive_policy.definition],
                aggregation_method=aggregation)
        return ts

    def _add_measures(self, aggregation, metric, timeserie):
        archive = self._get_measures_archive(metric, aggregation)
        archive.update(timeserie)
        self._store_metric_measures(metric, aggregation,
                                    archive.serialize())

    def add_measures(self, metric, measures):
        self._store_measures(metric, msgpackutils.dumps(
            list(map(tuple, measures))))

    @staticmethod
    def _store_measures(metric, data):
        raise NotImplementedError

    @staticmethod
    def _delete_metric(metric):
        raise NotImplementedError

    @staticmethod
    def _list_metric_with_measures_to_process(metric_id):
        raise NotImplementedError

    @staticmethod
    def _pending_measures_to_process_count(metric_id):
        raise NotImplementedError

    def delete_metric(self, metric):
        with self._lock(metric):
            self._delete_metric(metric)

    @staticmethod
    def _unserialize_measures(data):
        return msgpackutils.loads(data)

    def measures_report(self):
        metrics_to_process = self._list_metric_with_measures_to_process()
        return dict(
            (metric_id, self._pending_measures_to_process_count(metric_id))
            for metric_id in metrics_to_process)

    def process_measures(self, indexer):
        metrics_to_process = self._list_metric_with_measures_to_process()
        metrics = indexer.get_metrics(metrics_to_process)
        # This build the list of deleted metrics, i.e. the metrics we have
        # measures to process for but that are not in the indexer anymore.
        deleted_metrics_id = (set(map(uuid.UUID, metrics_to_process))
                              - set(m.id for m in metrics))
        for metric_id in deleted_metrics_id:
            self._delete_unprocessed_measures_for_metric_id(metric_id)
        for metric in metrics:
            lock = self._lock(metric)
            agg_methods = list(metric.archive_policy.aggregation_methods)
            # Do not block if we cannot acquire the lock, that means some other
            # worker is doing the job. We'll just ignore this metric and may
            # get back later to it if needed.
            if lock.acquire(blocking=False):
                try:
                    LOG.debug("Processing measures for %s" % metric)
                    with self._process_measure_for_metric(metric) as measures:
                        try:
                            with timeutils.StopWatch() as sw:
                                raw_measures = self._get_measures(metric,
                                                                  'none')
                                LOG.debug(
                                    "Retrieve unaggregated measures "
                                    "for %s in %.2fs"
                                    % (metric.id, sw.elapsed()))
                        except storage.MetricDoesNotExist:
                            try:
                                self._create_metric(metric)
                            except storage.MetricAlreadyExists:
                                # Created in the mean time, do not worry
                                pass
                            ts = None
                        except storage.AggregationDoesNotExist:
{'fraction_non_alphanumeric': 0.0728830990743915, 'fraction_numerical': 0.0037024340075419953, 'mean_word_length': 2.140826873385013, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4125334', 'n_tokens_mistral': 3892, 'n_tokens_neox': 3645, 'n_words': 982}