<commit_before>#!/usr/bin/env python
""" The agent bundle, contains all metric classes and agent running code """
import subprocess
import logging
import os
import sys
import signal
import ConfigParser
import json
import threading
import time

from optparse import OptionParser
import Queue as q



logger = logging.getLogger("agent")
collector_logger = logging.getLogger("telegraf")


def signal_handler(sig, frame):
    """ required for non-tty python runs to interrupt """
    raise KeyboardInterrupt()

signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)


class DataReader(object):
    """generator reads from source line-by-line"""
    def __init__(self, filename, pipe=False):
        self.pipe = pipe
        if not self.pipe:
            self.monout = open(filename, 'r')
        else:
            self.monout = filename
        self.buffer = ""
        self.closed = False

    def __iter__(self):
        while not self.closed:
            data = self.monout.readline()
            if data:
                parts = data.rsplit('\n', 1)
                if len(parts) > 1:
                    ready_chunk = self.buffer + parts[0] + '\n'
                    self.buffer = parts[1]
                    yield ready_chunk
                else:
                    self.buffer += parts[0]
            else:
                time.sleep(1)
        if not self.pipe:
            self.monout.close()

    def close(self):
        self.closed = True


class Consolidator(object):
    """generator consolidates data from source, cache it by timestamp"""
    def __init__(self, source):
        self.source = source
        self.results = {}

    def __iter__(self):
        for chunk in self.source:
            #logger.info('Consolidator got %s', chunk)
            if chunk:
                try:
                    data = json.loads(chunk)
                except ValueError:
                    logger.error('unable to decode chunk %s', chunk, exc_info=True)
                else:
                    try:
                        ts = str(data['timestamp'])
                        self.results.setdefault(ts, {})
                        for key, value in data['fields'].iteritems():
                            key = data['name']+"_"+key
                            self.results[ts][key] = value
                    except KeyError:
                        logger.error('Malformed json from source: %s', chunk, exc_info=True)
                    except:
                        logger.error('Something nasty happend in consolidator work', exc_info=True)
            #logger.info('Consolidator results lendth %s', len(self.results))
            if len(self.results) > 5:
                ready_to_go_index = min(self.results)
                yield json.dumps(
                    {
                        ready_to_go_index: self.results.pop(ready_to_go_index, None)
                    }
                )



class Drain(threading.Thread):
    """
    Drain a generator to a destination that answers to put(), in a thread
    """

    def __init__(self, source, destination):
        super(Drain, self).__init__()
        self.source = source
        self.destination = destination
        self._finished = threading.Event()
        self._interrupted = threading.Event()

    def run(self):
        for item in self.source:
            self.destination.put(item)
            if self._interrupted.is_set():
                break
        self._finished.set()

    def wait(self, timeout=None):
        self._finished.wait(timeout=timeout)

    def close(self):
        self._interrupted.set()


class AgentWorker(threading.Thread):
    def __init__(self, telegraf_path):
        super(AgentWorker, self).__init__()
        self.working_dir = os.path.dirname(__file__)
        self.startups = []
        self.startup_processes = []
        self.shutdowns = []
        self.daemon = True  # Thread auto-shutdown
        self.finished = False
        self.drain = None
        self.drain_stdout = None
        self.drain_err = None
        self.data_reader = None
        self.telegraf_path = telegraf_path
        self.results = q.Queue()
        self.results_stdout = q.Queue()
        self.results_err = q.Queue()

    @staticmethod
    def popen(cmnd):
        return subprocess.Popen(
            cmnd,
            bufsize=0,
            preexec_fn=os.setsid,
            close_fds=True,
            shell=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )

    def read_startup_config(self, cfg_file='agent_startup.cfg'):
        try:
            config = ConfigParser.ConfigParser()
            with open(os.path.join(self.working_dir, cfg_file), 'rb') as f:
                config.readfp(f)

            if config.has_section('startup'):
                for option in config.options('startup'):
                    if option.startswith('cmd'):
                        self.startups.append(config.get('startup', option))

            if config.has_section('shutdown'):
                for option in config.options('shutdown'):
                    if option.startswith('cmd'):
                        self.shutdowns.append(config.get('shutdown', option))
            logger.info(
                'Successfully loaded startup config.\n'
                'Startups: %s\n'
                'Shutdowns: %s\n', self.startups, self.shutdowns
            )
        except:<commit_msg>try fixing telegraf inside of Docker<commit_after>#!/usr/bin/env python
""" The agent bundle, contains all metric classes and agent running code """
import subprocess
import logging
import os
import sys
import signal
import ConfigParser
import json
import threading
import time

from optparse import OptionParser
import Queue as q



logger = logging.getLogger("agent")
collector_logger = logging.getLogger("telegraf")


def signal_handler(sig, frame):
    """ required for non-tty python runs to interrupt """
    raise KeyboardInterrupt()

signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)


class DataReader(object):
    """generator reads from source line-by-line"""
    def __init__(self, filename, pipe=False):
        self.pipe = pipe
        if not self.pipe:
            self.monout = open(filename, 'r')
        else:
            self.monout = filename
        self.buffer = ""
        self.closed = False

    def __iter__(self):
        while not self.closed:
            data = self.monout.readline()
            if data:
                parts = data.rsplit('\n', 1)
                if len(parts) > 1:
                    ready_chunk = self.buffer + parts[0] + '\n'
                    self.buffer = parts[1]
                    yield ready_chunk
                else:
                    self.buffer += parts[0]
            else:
                time.sleep(1)
        if not self.pipe:
            self.monout.close()

    def close(self):
        self.closed = True


class Consolidator(object):
    """generator consolidates data from source, cache it by timestamp"""
    def __init__(self, source):
        self.source = source
        self.results = {}

    def __iter__(self):
        for chunk in self.source:
            #logger.info('Consolidator got %s', chunk)
            if chunk:
                try:
                    data = json.loads(chunk)
                except ValueError:
                    logger.error('unable to decode chunk %s', chunk, exc_info=True)
                else:
                    try:
                        ts = str(data['timestamp'])
                        self.results.setdefault(ts, {})
                        for key, value in data['fields'].iteritems():
                            key = data['name']+"_"+key
                            self.results[ts][key] = value
                    except KeyError:
                        logger.error('Malformed json from source: %s', chunk, exc_info=True)
                    except:
                        logger.error('Something nasty happend in consolidator work', exc_info=True)
            #logger.info('Consolidator results lendth %s', len(self.results))
            if len(self.results) > 5:
                ready_to_go_index = min(self.results)
                yield json.dumps(
                    {
                        ready_to_go_index: self.results.pop(ready_to_go_index, None)
                    }
                )



class Drain(threading.Thread):
    """
    Drain a generator to a destination that answers to put(), in a thread
    """

    def __init__(self, source, destination):
        super(Drain, self).__init__()
        self.source = source
        self.destination = destination
        self._finished = threading.Event()
        self._interrupted = threading.Event()

    def run(self):
        for item in self.source:
            self.destination.put(item)
            if self._interrupted.is_set():
                break
        self._finished.set()

    def wait(self, timeout=None):
        self._finished.wait(timeout=timeout)

    def close(self):
        self._interrupted.set()


class AgentWorker(threading.Thread):
    def __init__(self, telegraf_path):
        super(AgentWorker, self).__init__()
        self.working_dir = os.path.dirname(__file__)
        self.startups = []
        self.startup_processes = []
        self.shutdowns = []
        self.daemon = True  # Thread auto-shutdown
        self.finished = False
        self.drain = None
        self.drain_stdout = None
        self.drain_err = None
        self.data_reader = None
        self.telegraf_path = telegraf_path
        self.results = q.Queue()
        self.results_stdout = q.Queue()
        self.results_err = q.Queue()

    @staticmethod
    def popen(cmnd):
        return subprocess.Popen(
            cmnd,
            bufsize=0,
            preexec_fn=os.setsid,
            close_fds=True,
            shell=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            stdin=subprocess.PIPE,
        )

    def read_startup_config(self, cfg_file='agent_startup.cfg'):
        try:
            config = ConfigParser.ConfigParser()
            with open(os.path.join(self.working_dir, cfg_file), 'rb') as f:
                config.readfp(f)

            if config.has_section('startup'):
                for option in config.options('startup'):
                    if option.startswith('cmd'):
                        self.startups.append(config.get('startup', option))

            if config.has_section('shutdown'):
                for option in config.options('shutdown'):
                    if option.startswith('cmd'):
                        self.shutdowns.append(config.get('shutdown', option))
            logger.info(
                'Successfully loaded startup config.\n'
                'Startups: %s\n'
                'Shutdowns: %s\n', self.startups, self.shutdowns
            )
        except:
{'fraction_non_alphanumeric': 0.1001733418483715, 'fraction_numerical': 0.00145972082839157, 'mean_word_length': 1.6920432220039292, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '2355880', 'n_tokens_mistral': 3031, 'n_tokens_neox': 2907, 'n_words': 695}