<commit_before>import time
import json
import datetime

from twisted.internet import reactor, defer, task
from twisted.python import log

try:
    from OpenSSL import SSL
    from twisted.internet import ssl
except:
    SSL=None

from duct.protocol import elasticsearch

from duct.objects import Output

class ElasticSearch(Output):
    """ElasticSearch HTTP API output

    **Configuration arguments:**

    :param url: Elasticsearch URL (default: http://localhost:9200)
    :type url: str
    :param maxsize: Maximum queue backlog size (default: 250000, 0 disables)
    :type maxsize: int
    :param maxrate: Maximum rate of documents added to index (default: 100)
    :type maxrate: int
    :param interval: Queue check interval in seconds (default: 1.0)
    :type interval: int
    :param user: Optional basic auth username
    :type user: str
    :param password: <PASSWORD>
    :type password: <PASSWORD>
    :param index: Index name format to store documents in Elastic
                  (default: duct-%Y.%m.%d)
    :type index: str
    """
    def __init__(self, *a):
        Output.__init__(self, *a)
        self.events = []
        self.t = task.LoopingCall(self.tick)

        self.inter = float(self.config.get('interval', 1.0))  # tick interval
        self.maxsize = int(self.config.get('maxsize', 250000))

        self.user = self.config.get('user')
        self.password = self.config.get('password')

        self.url = self.config.get('url', 'http://localhost:9200')

        maxrate = int(self.config.get('maxrate', 100))

        self.index = self.config.get('index', 'duct-%Y.%m.%d')

        if maxrate > 0:
            self.queueDepth = int(maxrate * self.inter)
        else:
            self.queueDepth = None

    def createClient(self):
        """Sets up HTTP connector and starts queue timer
        """

        self.client = elasticsearch.ElasticSearch(self.url, self.user,
            self.password, self.index)

        self.t.start(self.inter)

    def stop(self):
        """Stop this client.
        """
        self.t.stop()

    def transformEvent(self, e):
        d = dict(e)
        t = datetime.datetime.utcfromtimestamp(e.time)
        d['metric'] = float(e.metric)
        d['@timestamp'] = t.isoformat()

        if 'ttl' in d:
            # Useless field to Elasticsearch
            del d['ttl']

        return d

    def sendEvents(self, events):
        return self.client.bulkIndex([self.transformEvent(e) for e in events])

    @defer.inlineCallbacks
    def tick(self):
        """Clock tick called every self.inter
        """
        if self.events:
            if self.queueDepth and (len(self.events) > self.queueDepth):
                # Remove maximum of self.queueDepth items from queue
                events = self.events[:self.queueDepth]
                self.events = self.events[self.queueDepth:]
            else:
                events = self.events
                self.events = []

            try:
                result = yield self.sendEvents(events)
                if result.get('errors', False):
                    log.msg(repr(result))
                    self.events.extend(events)

            except Exception as e:
                log.msg('Could not connect to elasticsearch ' + str(e))
                self.events.extend(events)

    def eventsReceived(self, events):
        """Receives a list of events and queues them

        Arguments:
        events -- list of `duct.objects.Event`
        """
        # Make sure queue isn't oversized
        if (self.maxsize < 1) or (len(self.events) < self.maxsize):
            self.events.extend(events)

# Backward compatibility stub
ElasticSearchLog = ElasticSearch<commit_msg>If we successfully get errors then the re-queing won't help anyone<commit_after>import time
import json
import datetime

from twisted.internet import reactor, defer, task
from twisted.python import log

try:
    from OpenSSL import SSL
    from twisted.internet import ssl
except:
    SSL=None

from duct.protocol import elasticsearch

from duct.objects import Output

class ElasticSearch(Output):
    """ElasticSearch HTTP API output

    **Configuration arguments:**

    :param url: Elasticsearch URL (default: http://localhost:9200)
    :type url: str
    :param maxsize: Maximum queue backlog size (default: 250000, 0 disables)
    :type maxsize: int
    :param maxrate: Maximum rate of documents added to index (default: 100)
    :type maxrate: int
    :param interval: Queue check interval in seconds (default: 1.0)
    :type interval: int
    :param user: Optional basic auth username
    :type user: str
    :param password: Optional basic auth password
    :type password: str
    :param index: Index name format to store documents in Elastic
                  (default: duct-%Y.%m.%d)
    :type index: str
    """
    def __init__(self, *a):
        Output.__init__(self, *a)
        self.events = []
        self.t = task.LoopingCall(self.tick)

        self.inter = float(self.config.get('interval', 1.0))  # tick interval
        self.maxsize = int(self.config.get('maxsize', 250000))

        self.user = self.config.get('user')
        self.password = self.config.get('password')

        self.url = self.config.get('url', 'http://localhost:9200')

        maxrate = int(self.config.get('maxrate', 100))

        self.index = self.config.get('index', 'duct-%Y.%m.%d')

        if maxrate > 0:
            self.queueDepth = int(maxrate * self.inter)
        else:
            self.queueDepth = None

    def createClient(self):
        """Sets up HTTP connector and starts queue timer
        """

        self.client = elasticsearch.ElasticSearch(self.url, self.user,
            self.password, self.index)

        self.t.start(self.inter)

    def stop(self):
        """Stop this client.
        """
        self.t.stop()

    def transformEvent(self, e):
        d = dict(e)
        t = datetime.datetime.utcfromtimestamp(e.time)
        d['metric'] = float(e.metric)
        d['@timestamp'] = t.isoformat()

        if 'ttl' in d:
            # Useless field to Elasticsearch
            del d['ttl']

        return d

    def sendEvents(self, events):
        return self.client.bulkIndex([self.transformEvent(e) for e in events])

    @defer.inlineCallbacks
    def tick(self):
        """Clock tick called every self.inter
        """
        if self.events:
            if self.queueDepth and (len(self.events) > self.queueDepth):
                # Remove maximum of self.queueDepth items from queue
                events = self.events[:self.queueDepth]
                self.events = self.events[self.queueDepth:]
            else:
                events = self.events
                self.events = []

            try:
                result = yield self.sendEvents(events)
                if result.get('errors', False):
                    log.msg(repr(result))

            except Exception as e:
                log.msg('Could not connect to elasticsearch ' + str(e))
                self.events.extend(events)

    def eventsReceived(self, events):
        """Receives a list of events and queues them

        Arguments:
        events -- list of `duct.objects.Event`
        """
        # Make sure queue isn't oversized
        if (self.maxsize < 1) or (len(self.events) < self.maxsize):
            self.events.extend(events)

# Backward compatibility stub
ElasticSearchLog = ElasticSearch
{'fraction_non_alphanumeric': 0.10840618695359784, 'fraction_numerical': 0.008876933422999328, 'mean_word_length': 2.175064047822374, 'pattern_counts': {'":': 0, '<': 9, '<?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': '6861240', 'n_tokens_mistral': 2200, 'n_tokens_neox': 2057, 'n_words': 603}