NULL_VALUE = -4294967296.0 def log(event):
    event['system'] = event.get('type', 'console')
    observer(event) class Settings(dict):
class OrderedConfigParser(ConfigParser):
class RurouniOptions(usage.Options):
    def __init__(self):
    def readFrom(self, path, section):
    def read(self, path):
    def postOptions(self):
    def _normpath(path):
    def parseArgs(self, *action):
    def handleAction(self):
def get_parser(usage="%prog [options] <start|stop|status>"):
def read_config(program, options):
def _process_alive(pid): def getFilePath(schema_name, file_idx):
    return join(settings.LOCAL_DATA_DIR, settings['instance'],
                schema_name, '%d.hs' % file_idx) def createLink(metric, file_path):
    metric_path = getMetricPath(metric)
    try:
        _createLinkHelper(metric_path, file_path)
    except OSError as exc:
        if exc.errno == errno.ENAMETOOLONG:
            pass
        else:
            raise class StorageSchemas(object):
    def __init__(self, conf_file):
        self.schemas = loadStorageSchemas(conf_file)

    def getSchemaByMetric(self, metric):
        for schema in self.schemas:
            if schema.match(metric):
                return schema
        return defaultSchema

    def getSchemaByName(self, schema_name):
        for schema in self.schemas:
            if schema.name == schema_name:
                return schema
        return None def rebuildIndex(instance_data_dir, instance_index_file):
    """
    Rebuild index file from data file, if a data file has no valid metric,
    we will remove it.
    """
    out = open(instance_index_file, 'w')
    for schema_name in os.listdir(instance_data_dir):
        hs_file_pat = os.path.join(instance_data_dir, schema_name, '*.hs')
        for fp in glob.glob(hs_file_pat):
            with open(fp) as f:
                empty_flag = True
                header = kenshin.header(f)
                metric_list = header['tag_list']
                file_id = splitext(basename(fp))[0]
                for i, metric in enumerate(metric_list):
                    if metric != '':
                        empty_flag = False
                        out.write('%s %s %s %s\n' %
                                  (metric, schema_name, file_id, i))
            if empty_flag:
                os.remove(fp)
    out.close() def rebuildLink(instance_data_dir, instance_link_dir):
    for schema_name in os.listdir(instance_data_dir):
        hs_file_pat = os.path.join(instance_data_dir, schema_name, '*.hs')
        for fp in glob.glob(hs_file_pat):
            with open(fp) as f:
                header = kenshin.header(f)
                metric_list = header['tag_list']
                for metric in metric_list:
                    if metric != '':
                        link_path = getMetricPathByInstanceDir(instance_link_dir, metric)
                        try:
                            _createLinkHelper(link_path, fp)
                        except OSError as exc:
                            if exc.errno == errno.ENAMETOOLONG:
                                pass
                            else:
                                raise class TokenBucket(object):
    ''' Token Bucket algorithm for rate-limiting.
    URL: https://en.wikipedia.org/wiki/Token_bucket

    >>> bucket = TokenBucket(60, 1)
    >>> print bucket.consume(6)
    True
    >>> print bucket.consume(54)
    True
    >>> print bucket.consume(1)
    False
    >>> import time
    >>> time.sleep(1)
    >>> print bucket.consume(1)
    True
    '''
    def __init__(self, capacity, fill_rate):
        '''
        @capacity: total number of tokens in the bucket.
        @fill_rate: the rate in tokens/second that the bucket will be refilled.
        '''
        self.capacity = float(capacity)
        self._tokens = float(capacity)
        self.fill_rate = float(fill_rate)
        self.timestamp = time()

    def consume(self, tokens):
        ''' Consume tokens from the bucket.

        Return True if there were sufficient tokens otherwise False.
        '''
        if tokens <= self.tokens:
            self._tokens -= tokens
            return True
        else:
            return False

    @property
    def tokens(self):
        ''' Return the current number of tokens in the bucket. '''
        if self._tokens < self.capacity:
            now = time()
            delta = self.fill_rate * (now - self.timestamp)
            self._tokens = min(self.capacity, self._tokens + delta)
            self.timestamp = now
        return self._tokens

    def __repr__(self):
        return '<%s %.2f %.2f>' % (
            self.__class__.__name__, self.capacity, self.fill_rate) def get_instance_of_metric(metric, num_all_instance):
    return get_int32_hash(metric) % num_all_instance class TokenBucketFull(RurouniException):
    pass class UnexpectedMetric(RurouniException):
    pass
import os
import time
import kenshin
from threading import Lock
from kenshin.consts import NULL_VALUE
from rurouni import log
from rurouni.conf import settings
from rurouni.storage import (
    getFilePath, createLink, StorageSchemas, rebuildIndex, rebuildLink
)
from rurouni.utils import TokenBucket, get_instance_of_metric
from rurouni.exceptions import TokenBucketFull, UnexpectedMetric
                from rurouni.state import instrumentation
# coding: utf-8

"""
Copyright 2015-Present Douban Inc.
Copyright 2009 Chris Davis
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.
"""




class MetricCache(object):
    """
    (schema, file_idx, pos_idx)
    """
    def __init__(self):
        self.lock = Lock()
        self.metric_idxs = {}
        self.schema_caches = {}
        self.metrics_fh = None
        self.storage_schemas = None
        self.token_bucket = None

    def __del__(self):
        if self.metrics_fh is not None:
            self.metrics_fh.close()

    def init(self):
        with self.lock:
            index_file = settings.INDEX_FILE
            instance_data_dir = os.path.join(
                settings.LOCAL_DATA_DIR, settings.instance)
            instance_link_dir = os.path.join(
                settings.LOCAL_LINK_DIR, settings.instance)

            if os.path.exists(instance_data_dir):
                if not os.path.exists(index_file):
                    rebuildIndex(instance_data_dir, index_file)
                if not os.path.exists(instance_link_dir):
                    rebuildLink(instance_data_dir, instance_link_dir)

            # init token bucket
            capacity = settings.MAX_CREATES_PER_MINUTE
            fill_rate = float(capacity) / 60
            self.token_bucket = TokenBucket(capacity, fill_rate)

            self._initCache(index_file)

    def _initCache(self, index_file):
        # avoid repeated call
        if self.metrics_fh is not None:
            return

        self._initStorageSchemas()
        if os.path.exists(index_file):
            MAX_ALLOW_ERR_LINE = 1
            err_line_cnt = 0
            with open(index_file) as f:
                for line in f:
                    line = line.strip('\n')
                    try:
                        metric, schema_name, file_idx, file_pos = line.split(" ")
                        file_idx = int(file_idx)
                        file_pos = int(file_pos)
                    except Exception as e:
                        if err_line_cnt < MAX_ALLOW_ERR_LINE:
                            err_line_cnt += 1
                            continue
                        else:
                            raise Exception('Index file has many error: %s' % e)

                    schema = self.storage_schemas.getSchemaByName(schema_name)
                    schema_cache = self.getSchemaCache(schema)
                    schema_cache.add(schema, file_idx, file_pos)
                    self.metric_idxs[metric] = (schema.name, file_idx, file_pos)

        self.metrics_fh = open(index_file, 'a')

    def _initStorageSchemas(self):
        if self.storage_schemas is None:
            conf_file = os.path.join(settings.CONF_DIR, 'storage-schemas.conf')

            self.storage_schemas = StorageSchemas(conf_file)