<commit_before>from . daemon import ServiceDaemon
from kafka import KafkaClient, SimpleProducer
from kafka.protocol import CODEC_SNAPPY, CODEC_NONE
from systemd.journal import Reader
from threading import Thread, Lock
import errno
import json
import kafka.common
import logging
import os
import systemd.journal
import time
import types
import uuid

try:
    import snappy
except ImportError:
    snappy = None


KAFKA_COMPRESSED_MESSAGE_OVERHEAD = 30
MAX_KAFKA_MESSAGE_SIZE = 1024 ** 2


def _convert_uuid(s):
    return str(uuid.UUID(s.decode()))


def convert_mon(s):  # pylint: disable=unused-argument
    return None


def convert_realtime(t):
    return int(t) / 1000000.0  # Stock systemd transforms these into datetimes


converters = {
    "MESSAGE_ID": _convert_uuid,
    "_MACHINE_ID": _convert_uuid,
    "_BOOT_ID": _convert_uuid,
    "_SOURCE_REALTIME_TIMESTAMP": convert_realtime,
    "__REALTIME_TIMESTAMP": convert_realtime,
    "_SOURCE_MONOTONIC_TIMESTAMP": convert_mon,
    "__MONOTONIC_TIMESTAMP": convert_mon,
    "COREDUMP_TIMESTAMP": convert_realtime
}

systemd.journal.DEFAULT_CONVERTERS.update(converters)


def _convert_field(self, key, value):
    convert = self.converters.get(key, bytes.decode)
    try:
        return convert(value)
    except ValueError:
        # Leave in default bytes
        try:
            return bytes.decode(value)
        except:  # pylint: disable=bare-except
            return value


def get_next(self, skip=1):
    """Own get_next implementation that doesn't store the cursor
       since we don't want it"""
    if super(Reader, self)._next(skip):  # pylint: disable=protected-access
        entry = super(Reader, self)._get_all()  # pylint: disable=protected-access
        if entry:
            entry["__REALTIME_TIMESTAMP"] = self._get_realtime()  # pylint: disable=protected-access
            cursor = self._get_cursor()  # pylint: disable=protected-access
            return self._convert_entry(entry), cursor  # pylint: disable=protected-access
    return dict(), None


# KafkaSender exists mostly because kafka-python's async handling is still broken (11.5.2015)
# and we need to preserve the cursor position
class KafkaSender(Thread):
    def __init__(self, config, msg_buffer, kafka_address, max_send_interval=0.3):
        Thread.__init__(self)
        self.log = logging.getLogger("KafkaSender")
        self.config = config
        if not isinstance(self.config["kafka_topic"], bytes):
            topic = self.config["kafka_topic"].encode("utf8")
        self.topic = topic
        self.cursor = None
        self.kafka_address = kafka_address
        self.kafka = None
        self.kafka_producer = None
        self.last_send_time = time.time()
        self.last_state_save_time = time.time()
        self.msg_buffer = msg_buffer
        self.max_send_interval = max_send_interval
        self.start_time = time.time()
        self.previous_state = None
        self.running = True
        self._init_kafka()
        self.log.info("Initialized KafkaJournalPump")

    def _init_kafka(self):
        self.log.info("Initializing Kafka client, address: %r", self.kafka_address)
        while self.running:
            try:
                if self.kafka_producer:
                    self.kafka_producer.stop()
                if self.kafka:
                    self.kafka.close()

                self.kafka = KafkaClient(
                    self.kafka_address,
                    ssl=self.config.get("ssl", False),
                    certfile=self.config.get("certfile"),
                    keyfile=self.config.get("keyfile"),
                    ca=self.config.get("ca")
                )
                self.kafka_producer = SimpleProducer(self.kafka, codec=CODEC_SNAPPY
                                                     if snappy else CODEC_NONE)
                self.log.info("Initialized Kafka Client, address: %r", self.kafka_address)
                break
            except (kafka.common.KafkaUnavailableError,
                    kafka.common.LeaderNotAvailableError):<commit_msg>pylint: Make travis's pylint happy since it has a non ssl using python-kafka<commit_after>from . daemon import ServiceDaemon
from kafka import KafkaClient, SimpleProducer
from kafka.protocol import CODEC_SNAPPY, CODEC_NONE
from systemd.journal import Reader
from threading import Thread, Lock
import errno
import json
import kafka.common
import logging
import os
import systemd.journal
import time
import types
import uuid

try:
    import snappy
except ImportError:
    snappy = None


KAFKA_COMPRESSED_MESSAGE_OVERHEAD = 30
MAX_KAFKA_MESSAGE_SIZE = 1024 ** 2


def _convert_uuid(s):
    return str(uuid.UUID(s.decode()))


def convert_mon(s):  # pylint: disable=unused-argument
    return None


def convert_realtime(t):
    return int(t) / 1000000.0  # Stock systemd transforms these into datetimes


converters = {
    "MESSAGE_ID": _convert_uuid,
    "_MACHINE_ID": _convert_uuid,
    "_BOOT_ID": _convert_uuid,
    "_SOURCE_REALTIME_TIMESTAMP": convert_realtime,
    "__REALTIME_TIMESTAMP": convert_realtime,
    "_SOURCE_MONOTONIC_TIMESTAMP": convert_mon,
    "__MONOTONIC_TIMESTAMP": convert_mon,
    "COREDUMP_TIMESTAMP": convert_realtime
}

systemd.journal.DEFAULT_CONVERTERS.update(converters)


def _convert_field(self, key, value):
    convert = self.converters.get(key, bytes.decode)
    try:
        return convert(value)
    except ValueError:
        # Leave in default bytes
        try:
            return bytes.decode(value)
        except:  # pylint: disable=bare-except
            return value


def get_next(self, skip=1):
    """Own get_next implementation that doesn't store the cursor
       since we don't want it"""
    if super(Reader, self)._next(skip):  # pylint: disable=protected-access
        entry = super(Reader, self)._get_all()  # pylint: disable=protected-access
        if entry:
            entry["__REALTIME_TIMESTAMP"] = self._get_realtime()  # pylint: disable=protected-access
            cursor = self._get_cursor()  # pylint: disable=protected-access
            return self._convert_entry(entry), cursor  # pylint: disable=protected-access
    return dict(), None


# KafkaSender exists mostly because kafka-python's async handling is still broken (11.5.2015)
# and we need to preserve the cursor position
class KafkaSender(Thread):
    def __init__(self, config, msg_buffer, kafka_address, max_send_interval=0.3):
        Thread.__init__(self)
        self.log = logging.getLogger("KafkaSender")
        self.config = config
        if not isinstance(self.config["kafka_topic"], bytes):
            topic = self.config["kafka_topic"].encode("utf8")
        self.topic = topic
        self.cursor = None
        self.kafka_address = kafka_address
        self.kafka = None
        self.kafka_producer = None
        self.last_send_time = time.time()
        self.last_state_save_time = time.time()
        self.msg_buffer = msg_buffer
        self.max_send_interval = max_send_interval
        self.start_time = time.time()
        self.previous_state = None
        self.running = True
        self._init_kafka()
        self.log.info("Initialized KafkaJournalPump")

    def _init_kafka(self):
        self.log.info("Initializing Kafka client, address: %r", self.kafka_address)
        while self.running:
            try:
                if self.kafka_producer:
                    self.kafka_producer.stop()
                if self.kafka:
                    self.kafka.close()

                self.kafka = KafkaClient(  # pylint: disable=unexpected-keyword-arg
                    self.kafka_address,
                    ssl=self.config.get("ssl", False),
                    certfile=self.config.get("certfile"),
                    keyfile=self.config.get("keyfile"),
                    ca=self.config.get("ca")
                )
                self.kafka_producer = SimpleProducer(self.kafka, codec=CODEC_SNAPPY
                                                     if snappy else CODEC_NONE)
                self.log.info("Initialized Kafka Client, address: %r", self.kafka_address)
                break
            except (kafka.common.KafkaUnavailableError,
                    kafka.common.LeaderNotAvailableError):
{'fraction_non_alphanumeric': 0.08893735611292863, 'fraction_numerical': 0.006300739125166606, 'mean_word_length': 2.5824652777777777, 'pattern_counts': {'":': 16, '<': 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': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6184416', 'n_tokens_mistral': 2520, 'n_tokens_neox': 2411, 'n_words': 547}