<commit_before># This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

import pika
from Queue import (
    Queue,
    Empty
)

from configman import (
    Namespace,
    class_converter
)
from socorro.external.rabbitmq.connection_context import (
    ConnectionContextPooled
)
from socorro.external.crashstorage_base import (
    CrashStorageBase,
)


#==============================================================================
class RabbitMQCrashStorage(CrashStorageBase):
    """This class is an implementation of a Socorro Crash Storage system.
    It is used as a crash queing methanism for raw crashes.  It implements
    the save_raw_crash method as a queue submission function, and the
    new_crashes generator as a queue consumption function.  Please note: as
    it only queues the crash_id and not the whole raw crash, it is not suitable
    to actually save a crash.  It is a very lossy container.  This class
    should be used in conjuction with a more persistant storage mechanism.

    The implementations CrashStorage classes can use arbitrarly high or low
    level semantics to talk to their underlying resource.  In the RabbitMQ,
    implementation, queing through the 'save_raw_crash' method is given full
    transactional semantics using the TransactorExecutor classes.  The
    'new_crashes' generator has a lower level relationship with the
    underlying connection object"""

    required_config = Namespace()
    required_config.add_option(
        'rabbitmq_class',
        default=ConnectionContextPooled,  # we choose a pooled connection
                                          # because we need thread safe
                                          # connection behaviors
        doc='the class responsible for connecting to RabbitMQ'
    )
    required_config.add_option(
        'transaction_executor_class',
        default="socorro.database.transaction_executor."
                "TransactionExecutorWithInfiniteBackoff",
        doc='a class that will manage transactions',
        from_string_converter=class_converter
    )
    required_config.add_option(
        'routing_key',
        default='socorro.normal',
        doc='the name of the queue to recieve crashes'
    )
    required_config.add_option(
        'filter_on_legacy_processing',
        default=True,
        doc='toggle for using or ignoring the throttling flag'
    )

    #--------------------------------------------------------------------------
    def __init__(self, config, quit_check_callback=None):
        super(RabbitMQCrashStorage, self).__init__(
            config,
            quit_check_callback=quit_check_callback
        )

        self.config = config

        # Note: this may continue to grow if we aren't acking certain UUIDs.
        # We should find a way to time out UUIDs after a certain time.
        self.acknowledgement_token_cache = {}
        self.acknowledgment_queue = Queue()

        self.rabbitmq = config.rabbitmq_class(config)
        self.transaction = config.transaction_executor_class(
            config,
            self.rabbitmq,
            quit_check_callback=quit_check_callback
        )

        # cache this object so we don't have to remake it for every transaction
        self._basic_properties = pika.BasicProperties(
            delivery_mode = 2, # make message persistent
        )

    #--------------------------------------------------------------------------
    def save_raw_crash(self, raw_crash, dumps, crash_id):
        try:
            this_crash_should_be_queued = (
                (not self.config.filter_on_legacy_processing)
                or
                raw_crash.legacy_processing == 0
            )
        except KeyError:
            self.config.logger.debug(
                'RabbitMQCrashStorage legacy_processing key absent in crash '
                '%s', crash_id
            )
            return

        if this_crash_should_be_queued:
            self.config.logger.debug(
                'RabbitMQCrashStorage saving crash %s', crash_id
            )
            self.transaction(self._save_raw_crash_transaction, crash_id)
        else:
            self.config.logger.debug(
                'RabbitMQCrashStorage not saving crash %s, legacy processing '
                'flag is %s', crash_id, raw_crash.legacy_processing
            )

    #--------------------------------------------------------------------------
    def _save_raw_crash_transaction(self, connection, crash_id):
        connection.channel.basic_publish(
            exchange='',
            routing_key=self.config.routing_key,
            body=crash_id,
            properties=self._basic_properties
        )

    #--------------------------------------------------------------------------
    def new_crashes(self):
        """This generator fetches crash_ids from RabbitMQ."""

        # We've set up RabbitMQ to require acknowledgement of processing of a
        # crash_id from this generator.  It is the responsibility of the
        # consumer of the crash_id to tell this instance of the class when has
        # completed its work on the crash_id.  That is done with the call to
        # 'ack_crash' below.  Because RabbitMQ connections are not thread safe,
        # only the thread that read the crash may acknowledge it.  'ack_crash'
        # queues the crash_id. The '_consume_acknowledgement_queue' function
        # is run to send acknowledgments back to RabbitMQ
        self._consume_acknowledgement_queue()
        conn = self.rabbitmq.connection()
        while True:
            for queue in ['socorro.priority', 'socorro.normal']:
                method_frame, header_frame, body = conn.channel.basic_get(
                    queue=queue
                )
                if method_frame:
                    break
            if not method_frame:
                return
            self._consume_acknowledgement_queue()
            self.acknowledgement_token_cache[body] = method_frame
            yield body

    #--------------------------------------------------------------------------
    def ack_crash(self, crash_id):
        self.acknowledgment_queue.put(crash_id)

    #--------------------------------------------------------------------------
    def _consume_acknowledgement_queue(self):
        """The acknowledgement of the processing of each crash_id yielded
        from the 'new_crashes' method must take place on the same connection
        that the crash_id came from.  The crash_ids are queued in the
        'acknowledgment_queue'.  That queue is consumed by the QueuingThread"""
        try:
            while True:
                crash_id_to_be_acknowledged = \
                    self.acknowledgment_queue.get_nowait()<commit_msg>Fixes bug 945736 - use configuration for RabbitMQ in crashstorage.py<commit_after># This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

import pika
from Queue import (
    Queue,
    Empty
)

from configman import (
    Namespace,
    class_converter
)
from socorro.external.rabbitmq.connection_context import (
    ConnectionContextPooled
)
from socorro.external.crashstorage_base import (
    CrashStorageBase,
)


#==============================================================================
class RabbitMQCrashStorage(CrashStorageBase):
    """This class is an implementation of a Socorro Crash Storage system.
    It is used as a crash queing methanism for raw crashes.  It implements
    the save_raw_crash method as a queue submission function, and the
    new_crashes generator as a queue consumption function.  Please note: as
    it only queues the crash_id and not the whole raw crash, it is not suitable
    to actually save a crash.  It is a very lossy container.  This class
    should be used in conjuction with a more persistant storage mechanism.

    The implementations CrashStorage classes can use arbitrarly high or low
    level semantics to talk to their underlying resource.  In the RabbitMQ,
    implementation, queing through the 'save_raw_crash' method is given full
    transactional semantics using the TransactorExecutor classes.  The
    'new_crashes' generator has a lower level relationship with the
    underlying connection object"""

    required_config = Namespace()
    required_config.add_option(
        'rabbitmq_class',
        default=ConnectionContextPooled,  # we choose a pooled connection
                                          # because we need thread safe
                                          # connection behaviors
        doc='the class responsible for connecting to RabbitMQ'
    )
    required_config.add_option(
        'transaction_executor_class',
        default="socorro.database.transaction_executor."
                "TransactionExecutorWithInfiniteBackoff",
        doc='a class that will manage transactions',
        from_string_converter=class_converter
    )
    required_config.add_option(
        'routing_key',
        default='socorro.normal',
        doc='the name of the queue to recieve crashes'
    )
    required_config.add_option(
        'filter_on_legacy_processing',
        default=True,
        doc='toggle for using or ignoring the throttling flag'
    )

    #--------------------------------------------------------------------------
    def __init__(self, config, quit_check_callback=None):
        super(RabbitMQCrashStorage, self).__init__(
            config,
            quit_check_callback=quit_check_callback
        )

        self.config = config

        # Note: this may continue to grow if we aren't acking certain UUIDs.
        # We should find a way to time out UUIDs after a certain time.
        self.acknowledgement_token_cache = {}
        self.acknowledgment_queue = Queue()

        self.rabbitmq = config.rabbitmq_class(config)
        self.transaction = config.transaction_executor_class(
            config,
            self.rabbitmq,
            quit_check_callback=quit_check_callback
        )

        # cache this object so we don't have to remake it for every transaction
        self._basic_properties = pika.BasicProperties(
            delivery_mode = 2, # make message persistent
        )

    #--------------------------------------------------------------------------
    def save_raw_crash(self, raw_crash, dumps, crash_id):
        try:
            this_crash_should_be_queued = (
                (not self.config.filter_on_legacy_processing)
                or
                raw_crash.legacy_processing == 0
            )
        except KeyError:
            self.config.logger.debug(
                'RabbitMQCrashStorage legacy_processing key absent in crash '
                '%s', crash_id
            )
            return

        if this_crash_should_be_queued:
            self.config.logger.debug(
                'RabbitMQCrashStorage saving crash %s', crash_id
            )
            self.transaction(self._save_raw_crash_transaction, crash_id)
        else:
            self.config.logger.debug(
                'RabbitMQCrashStorage not saving crash %s, legacy processing '
                'flag is %s', crash_id, raw_crash.legacy_processing
            )

    #--------------------------------------------------------------------------
    def _save_raw_crash_transaction(self, connection, crash_id):
        connection.channel.basic_publish(
            exchange='',
            routing_key=self.config.routing_key,
            body=crash_id,
            properties=self._basic_properties
        )

    #--------------------------------------------------------------------------
    def new_crashes(self):
        """This generator fetches crash_ids from RabbitMQ."""

        # We've set up RabbitMQ to require acknowledgement of processing of a
        # crash_id from this generator.  It is the responsibility of the
        # consumer of the crash_id to tell this instance of the class when has
        # completed its work on the crash_id.  That is done with the call to
        # 'ack_crash' below.  Because RabbitMQ connections are not thread safe,
        # only the thread that read the crash may acknowledge it.  'ack_crash'
        # queues the crash_id. The '_consume_acknowledgement_queue' function
        # is run to send acknowledgments back to RabbitMQ
        self._consume_acknowledgement_queue()
        conn = self.rabbitmq.connection()
        while True:
            for queue in [self.rabbitmq.config.priority_queue_name,
                self.rabbitmq.config.standard_queue_name]:
                method_frame, header_frame, body = conn.channel.basic_get(
                    queue=queue
                )
                if method_frame:
                    break
            if not method_frame:
                return
            self._consume_acknowledgement_queue()
            self.acknowledgement_token_cache[body] = method_frame
            yield body

    #--------------------------------------------------------------------------
    def ack_crash(self, crash_id):
        self.acknowledgment_queue.put(crash_id)

    #--------------------------------------------------------------------------
    def _consume_acknowledgement_queue(self):
        """The acknowledgement of the processing of each crash_id yielded
        from the 'new_crashes' method must take place on the same connection
        that the crash_id came from.  The crash_ids are queued in the
        'acknowledgment_queue'.  That queue is consumed by the QueuingThread"""
        try:
            while True:
                crash_id_to_be_acknowledged = \
                    self.acknowledgment_queue.get_nowait()
{'fraction_non_alphanumeric': 0.1274516847474675, 'fraction_numerical': 0.0012931963503125224, 'mean_word_length': 1.8819875776397517, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3213049', 'n_tokens_mistral': 3592, 'n_tokens_neox': 3342, 'n_words': 1075}