<commit_before>#!/usr/bin/env python

"""Process that subscribes to ALL events and persists them efficiently in bulk into the events datastore"""

import pprint
from gevent.queue import Queue
from gevent.event import Event

from pyon.ion.event import EventSubscriber
from pyon.ion.process import SimpleProcess
from pyon.util.async import spawn
from pyon.util.containers import named_any
from pyon.public import log

PROCESS_PLUGINS = []


class EventPersister(SimpleProcess):

    def on_init(self):
        # Time in between event persists
        self.persist_interval = float(self.CFG.get_safe("process.event_persister.persist_interval", 1.0))

        self.persist_blacklist = self.CFG.get_safe("process.event_persister.persist_blacklist", {})

        self._event_type_blacklist = [entry['event_type'] for entry in self.persist_blacklist if entry.get('event_type', None) and len(entry) == 1]
        self._complex_blacklist = [entry for entry in self.persist_blacklist if not (entry.get('event_type', None) and len(entry) == 1)]
        if self._complex_blacklist:
            log.warn("EventPersister does not yet support complex blacklist expressions: %s", self._complex_blacklist)

        # Holds received events FIFO in synchronized queue
        self.event_queue = Queue()

        # Temporarily holds list of events to persist while datastore operation are not yet completed
        # This is where events to persist will remain if datastore operation fails occasionally.
        self.events_to_persist = None

        # Number of unsuccessful consecutive attempts to persist during loop
        self.failure_count = 0

        # bookkeeping for greenlet
        self._persist_greenlet = None
        self._terminate_persist = Event() # when set, exits the persister greenlet

        # The event subscriber
        self.event_sub = None

        # Registered event process plugins
        self.process_plugins = {}
        for plugin_name, plugin_cls, plugin_args in PROCESS_PLUGINS:
            try:
                plugin = named_any(plugin_cls)(**plugin_args)
                self.process_plugins[plugin_name]= plugin
                log.info("Loaded event processing plugin %s (%s)", plugin_name, plugin_cls)
            except Exception as ex:
                log.error("Cannot instantiate event processing plugin %s (%s): %s", plugin_name, plugin_cls, ex)


    def on_start(self):
        # Persister thread
        self._persist_greenlet = spawn(self._persister_loop, self.persist_interval)
        log.debug('EventPersister persist greenlet started in "%s" (interval %s)', self.__class__.__name__, self.persist_interval)

        # Event subscription
        self.event_sub = EventSubscriber(pattern=EventSubscriber.ALL_EVENTS,
                                         callback=self._on_event,
                                         queue_name="event_persister",
                                         auto_delete=False)

        self.event_sub.start()

    def on_quit(self):
        # Stop event subscriber<commit_msg>Event persister plugins can be configured<commit_after>#!/usr/bin/env python

"""Process that subscribes to ALL events and persists them efficiently in bulk into the events datastore"""

import pprint
from gevent.queue import Queue
from gevent.event import Event

from pyon.ion.event import EventSubscriber
from pyon.ion.process import SimpleProcess
from pyon.util.async import spawn
from pyon.util.containers import named_any
from pyon.public import log


class EventPersister(SimpleProcess):

    def on_init(self):
        # Time in between event persists
        self.persist_interval = float(self.CFG.get_safe("process.event_persister.persist_interval", 1.0))

        self.persist_blacklist = self.CFG.get_safe("process.event_persister.persist_blacklist", {})

        self._event_type_blacklist = [entry['event_type'] for entry in self.persist_blacklist if entry.get('event_type', None) and len(entry) == 1]
        self._complex_blacklist = [entry for entry in self.persist_blacklist if not (entry.get('event_type', None) and len(entry) == 1)]
        if self._complex_blacklist:
            log.warn("EventPersister does not yet support complex blacklist expressions: %s", self._complex_blacklist)

        # Holds received events FIFO in synchronized queue
        self.event_queue = Queue()

        # Temporarily holds list of events to persist while datastore operation are not yet completed
        # This is where events to persist will remain if datastore operation fails occasionally.
        self.events_to_persist = None

        # Number of unsuccessful consecutive attempts to persist during loop
        self.failure_count = 0

        # bookkeeping for greenlet
        self._persist_greenlet = None
        self._terminate_persist = Event() # when set, exits the persister greenlet

        # The event subscriber
        self.event_sub = None

        process_plugin_defs = self.CFG.get_safe("process.event_persister.process_plugins", {}) or {}

        # Registered event process plugins
        self.process_plugins = {}
        for plugin_name, plugin_cls, plugin_args in process_plugin_defs:
            try:
                plugin = named_any(plugin_cls)(**plugin_args)
                self.process_plugins[plugin_name]= plugin
                log.info("Loaded event processing plugin %s (%s)", plugin_name, plugin_cls)
            except Exception as ex:
                log.error("Cannot instantiate event processing plugin %s (%s): %s", plugin_name, plugin_cls, ex)


    def on_start(self):
        # Persister thread
        self._persist_greenlet = spawn(self._persister_loop, self.persist_interval)
        log.debug('EventPersister persist greenlet started in "%s" (interval %s)', self.__class__.__name__, self.persist_interval)

        # Event subscription
        self.event_sub = EventSubscriber(pattern=EventSubscriber.ALL_EVENTS,
                                         callback=self._on_event,
                                         queue_name="event_persister",
                                         auto_delete=False)

        self.event_sub.start()

    def on_quit(self):
        # Stop event subscriber
{'fraction_non_alphanumeric': 0.07624113475177305, 'fraction_numerical': 0.0016118633139909735, 'mean_word_length': 2.8137676705593115, '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': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2636942', 'n_tokens_mistral': 1681, 'n_tokens_neox': 1606, 'n_words': 478}