<commit_before>from __future__ import absolute_import

import functools
import itertools
import logging
import random
import time
from contextlib import contextmanager

from django.conf import settings
from rb import Cluster
from redis.client import Script
from redis.exceptions import (
    ResponseError,
    WatchError,
)

from sentry.digests import (
    Record,
    ScheduleEntry,
)
from sentry.digests.backends.base import Backend
from sentry.utils.cache import Lock


logger = logging.getLogger('sentry.digests')

SCHEDULE_PATH_COMPONENT = 's'
SCHEDULE_STATE_WAITING = 'w'
SCHEDULE_STATE_READY = 'r'

TIMELINE_PATH_COMPONENT = 't'
TIMELINE_ITERATION_PATH_COMPONENT = 'i'
TIMELINE_DIGEST_PATH_COMPONENT = 'd'
TIMELINE_RECORD_PATH_COMPONENT = 'r'


def ilen(iterator):
    i = 0
    for i, _ in enumerate(iterator):
        pass
    return i


def make_schedule_key(namespace, state):
    return '{0}:{1}:{2}'.format(namespace, SCHEDULE_PATH_COMPONENT, state)


def make_timeline_key(namespace, key):
    return '{0}:{1}:{2}'.format(namespace, TIMELINE_PATH_COMPONENT, key)


def make_iteration_key(timeline_key):
    return '{0}:{1}'.format(timeline_key, TIMELINE_ITERATION_PATH_COMPONENT)


def make_digest_key(timeline_key):
    return '{0}:{1}'.format(timeline_key, TIMELINE_DIGEST_PATH_COMPONENT)


def make_record_key(timeline_key, record):
    return '{0}:{1}:{2}'.format(timeline_key, TIMELINE_RECORD_PATH_COMPONENT, record)


# Ensures an timeline is scheduled to be digested.
# KEYS: {WATING, READY}
# ARGV: {TIMELINE, TIMESTAMP}
ENSURE_TIMELINE_SCHEDULED_SCRIPT = """\
-- Check to see if the timeline exists in the "waiting" set (heuristics tell us
-- that this should be more likely than it's presence in the "ready" set.)
local waiting = redis.call('ZSCORE', KEYS[1], ARGV[1])

if waiting ~= false then
    -- If the item already exists, update the score if the provided timestamp
    -- is less than the current score.
    if tonumber(waiting) > tonumber(ARGV[2]) then
        redis.call('ZADD', KEYS[1], ARGV[2], ARGV[1])
    end
    return
end

-- Otherwise, check to see if the timeline already exists in the "ready" set.
-- If it doesn't, it needs to be added to the "waiting" set to be scheduled.
if redis.call('ZSCORE', KEYS[2], ARGV[1]) == false then
    redis.call('ZADD', KEYS[1], ARGV[2], ARGV[1])
    return
end
"""


# Trims a timeline to a maximum number of records.
# Returns the number of keys that were deleted.
# KEYS: {TIMELINE}
# ARGV: {LIMIT}
TRUNCATE_TIMELINE_SCRIPT = """\
local keys = redis.call('ZREVRANGE', KEYS[1], ARGV[1], -1)
for i, record in pairs(keys) do
    redis.call('DEL', KEYS[1] .. ':{TIMELINE_RECORD_PATH_COMPONENT}:' .. record)
    redis.call('ZREM', KEYS[1], record)
end
return table.getn(keys)
""".format(TIMELINE_RECORD_PATH_COMPONENT=TIMELINE_RECORD_PATH_COMPONENT)


# XXX: Passing `None` as the first argument is a dirty hack to allow us to use
# this more easily with the cluster
ensure_timeline_scheduled = Script(None, ENSURE_TIMELINE_SCHEDULED_SCRIPT)
truncate_timeline = Script(None, TRUNCATE_TIMELINE_SCRIPT)


class RedisBackend(Backend):
    def __init__(self, **options):
        super(RedisBackend, self).__init__(**options)

        self.cluster = Cluster(**options.pop('cluster', settings.SENTRY_REDIS_OPTIONS))
        self.namespace = options.pop('namespace', 'd')
        self.record_ttl = options.pop('record_ttl', 60 * 60)

        if options:
            logger.warning('Discarding invalid options: %r', options)<commit_msg>Add documentation to `RedisBackend`.<commit_after>from __future__ import absolute_import

import functools
import itertools
import logging
import random
import time
from contextlib import contextmanager

from django.conf import settings
from rb import Cluster
from redis.client import Script
from redis.exceptions import (
    ResponseError,
    WatchError,
)

from sentry.digests import (
    Record,
    ScheduleEntry,
)
from sentry.digests.backends.base import Backend
from sentry.utils.cache import Lock


logger = logging.getLogger('sentry.digests')

SCHEDULE_PATH_COMPONENT = 's'
SCHEDULE_STATE_WAITING = 'w'
SCHEDULE_STATE_READY = 'r'

TIMELINE_PATH_COMPONENT = 't'
TIMELINE_ITERATION_PATH_COMPONENT = 'i'
TIMELINE_DIGEST_PATH_COMPONENT = 'd'
TIMELINE_RECORD_PATH_COMPONENT = 'r'


def ilen(iterator):
    i = 0
    for i, _ in enumerate(iterator):
        pass
    return i


def make_schedule_key(namespace, state):
    return '{0}:{1}:{2}'.format(namespace, SCHEDULE_PATH_COMPONENT, state)


def make_timeline_key(namespace, key):
    return '{0}:{1}:{2}'.format(namespace, TIMELINE_PATH_COMPONENT, key)


def make_iteration_key(timeline_key):
    return '{0}:{1}'.format(timeline_key, TIMELINE_ITERATION_PATH_COMPONENT)


def make_digest_key(timeline_key):
    return '{0}:{1}'.format(timeline_key, TIMELINE_DIGEST_PATH_COMPONENT)


def make_record_key(timeline_key, record):
    return '{0}:{1}:{2}'.format(timeline_key, TIMELINE_RECORD_PATH_COMPONENT, record)


# Ensures an timeline is scheduled to be digested.
# KEYS: {WATING, READY}
# ARGV: {TIMELINE, TIMESTAMP}
ENSURE_TIMELINE_SCHEDULED_SCRIPT = """\
-- Check to see if the timeline exists in the "waiting" set (heuristics tell us
-- that this should be more likely than it's presence in the "ready" set.)
local waiting = redis.call('ZSCORE', KEYS[1], ARGV[1])

if waiting ~= false then
    -- If the item already exists, update the score if the provided timestamp
    -- is less than the current score.
    if tonumber(waiting) > tonumber(ARGV[2]) then
        redis.call('ZADD', KEYS[1], ARGV[2], ARGV[1])
    end
    return
end

-- Otherwise, check to see if the timeline already exists in the "ready" set.
-- If it doesn't, it needs to be added to the "waiting" set to be scheduled.
if redis.call('ZSCORE', KEYS[2], ARGV[1]) == false then
    redis.call('ZADD', KEYS[1], ARGV[2], ARGV[1])
    return
end
"""


# Trims a timeline to a maximum number of records.
# Returns the number of keys that were deleted.
# KEYS: {TIMELINE}
# ARGV: {LIMIT}
TRUNCATE_TIMELINE_SCRIPT = """\
local keys = redis.call('ZREVRANGE', KEYS[1], ARGV[1], -1)
for i, record in pairs(keys) do
    redis.call('DEL', KEYS[1] .. ':{TIMELINE_RECORD_PATH_COMPONENT}:' .. record)
    redis.call('ZREM', KEYS[1], record)
end
return table.getn(keys)
""".format(TIMELINE_RECORD_PATH_COMPONENT=TIMELINE_RECORD_PATH_COMPONENT)


# XXX: Passing `None` as the first argument is a dirty hack to allow us to use
# this more easily with the cluster
ensure_timeline_scheduled = Script(None, ENSURE_TIMELINE_SCHEDULED_SCRIPT)
truncate_timeline = Script(None, TRUNCATE_TIMELINE_SCRIPT)


class RedisBackend(Backend):
    """
    Implements the digest backend API, backed by Redis.

    Each timeline is modeled as a modeled as a sorted set, as well as a
    separate key that contains the iteration counter (for implementing backoff
    strategies that require this value as an argument, such as exponential
    backoff.)

    .. code::

        redis:6379> GET "d:t:mail:p:1:i"
        "1"

        redis:6379> ZREVRANGEBYSCORE "d:t:mail:p:1" inf -inf WITHSCORES
        1) "433be20b807c4cd49a132de69c0f6c55"
        2) "1444847625"
        3) "0f9d5fe4b5b3400fab85d9a841aa8467"
        4) "1444847625"
        ...

    In the example above, the timeline ``mail:p:1`` has already digested once,
    as evidenced by the iteration counter, which ends with ``:i``. The timeline
    also contains references to several records, which are stored separately,
    encoded using the codec:

    .. code::

        redis:6379> GET "d:t:mail:p:1:r:433be20b807c4cd49a132de69c0f6c55"
        [ binary content ]

    When the timeline is ready to be digested, the timeline set is renamed,
    creating a digest set (in this case the key would be ``d:t:mail:p:1:d``),
    that represents a snapshot of the timeline contents at that point in time.
    (If the digest set already exists, the timeline contents are instead
    unioned into the digest set and then the timeline is cleared.) This allows
    new records to be added to the timeline that will be processed after the
    next scheduling interval without the risk of data loss due to race
    conditions between the record addition and digest generation and delivery.

    Schedules are modeled as two sorted sets -- one for ``waiting`` items, and
    one for ``ready`` items. Items in the ``waiting`` set are scored by the
    time at which they should be transitioned to the ``ready`` set.  Items in
    the ``ready`` set are scored by the time at which they were scheduled to be
    added to the ``ready`` set. Iterating each set from oldest to newest yields
    the highest priority items for action (moving from the ``waiting`` to
    ``ready`` set, or delivering a digest for the ``waiting`` and ``ready``
    set, respectively.)

    .. code::

        redis:6379> ZREVRANGEBYSCORE "d:s:w" inf -inf WITHSCORES
        1) "mail:p:1"
        2) "1444847638"

    """
    def __init__(self, **options):
        super(RedisBackend, self).__init__(**options)

        self.cluster = Cluster(**options.pop('cluster', settings.SENTRY_REDIS_OPTIONS))
        self.namespace = options.pop('namespace', 'd')
        self.record_ttl = options.pop('record_ttl', 60 * 60)

        if options:
            logger.warning('Discarding invalid options: %r', options)
{'fraction_non_alphanumeric': 0.10401713979646492, 'fraction_numerical': 0.019925013390465988, 'mean_word_length': 4.152317880794702, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6422117', 'n_tokens_mistral': 3233, 'n_tokens_neox': 3092, 'n_words': 954}