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

import os
import sys
import time
import select
import logging
import argparse
import threading
import subprocess

ETCD_IMAGE = "quay.io/coreos/etcd:v3.3.5"

LOG_LEVELS = {
    "critical": logging.CRITICAL,
    "error": logging.ERROR,
    "warning": logging.WARNING,
    "info": logging.INFO,
    "debug": logging.DEBUG,
}

LOG_COLORS = {
    "critical": "\x1b[31;1m",
    "error": "\x1b[31;1m",
    "warning": "\x1b[33;1m",
}

log = logging.getLogger(__name__)
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(levelname)s:%(message)s"))
log.addHandler(handler)

class ExcThread(threading.Thread):
    def __init__(self, target):
        super().__init__(target=target)
        self.error = None

    def run(self):
        try:
            self._target()
        except Exception as e:
            self.error = e

def join(*targets):
    threads = []

    for target in targets:
        t = ExcThread(target)
        t.start()
        threads.append(t)

    for t in threads:
        t.join()
    for t in threads:
        if t.error is not None:
            raise Exception("Thread error") from t.error

class Output:
    def __init__(self, pipe, level):
        self.pipe = pipe
        self.level = level
        self.lines = []

class ProcessResult:
    def __init__(self, rc, stdout, stderr):
        self.rc = rc
        self.stdout = stdout
        self.stderr = stderr

class DefaultDriver:
    def available(self):
        return True

    def clear(self):
        if run("yes | pachctl delete all", shell=True, raise_on_error=False).rc != 0:
            log.error("could not call `pachctl delete all`; most likely this just means that a pachyderm cluster hasn't been setup, but may indicate a bad state")

    def start(self):
        pass

    def inspect(self):
        pass

    def push_images(self, deploy_version, dash_image):
        pass

    def wait(self):
        while suppress("pachctl", "version") != 0:
            log.info("Waiting for pachyderm to come up...")
            time.sleep(1)

class MinikubeDriver(DefaultDriver):
    def available(self):
        return run("which", "minikube", raise_on_error=False).rc == 0

    def clear(self):
        run("minikube", "delete")

    def start(self):
        run("minikube", "start")

        while suppress("minikube", "status") != 0:
            log.info("Waiting for minikube to come up...")
            time.sleep(1)

    def push_images(self, deploy_version, dash_image):
        run("./etc/kube/push-to-minikube.sh", "pachyderm/pachd:{}".format(deploy_version))
        run("./etc/kube/push-to-minikube.sh", "pachyderm/worker:{}".format(deploy_version))
        run("./etc/kube/push-to-minikube.sh", ETCD_IMAGE)
        run("./etc/kube/push-to-minikube.sh", dash_image)

def parse_log_level(s):
    try:
        return LOG_LEVELS[s]
    except KeyError:
        raise Exception("Unknown log level: {}".format(s))

def redirect_to_logger(stdout, stderr):
    for io in select.select([stdout.pipe, stderr.pipe], [], [], 1000)[0]:
        line = io.readline().decode().rstrip()

        if line == "":
            continue

        dest = stdout if io == stdout.pipe else stderr
        log.log(LOG_LEVELS[dest.level], "{}{}\x1b[0m".format(LOG_COLORS.get(dest.level, ""), line))
        dest.lines.append(line)

def run(cmd, *args, raise_on_error=True, shell=False, stdout_log_level="info", stderr_log_level="error"):
    log.debug("Running `%s %s`", cmd, " ".join(args))

    proc = subprocess.Popen([cmd, *args], shell=shell, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    stdout = Output(proc.stdout, stdout_log_level)
    stderr = Output(proc.stderr, stderr_log_level)

    while proc.poll() is None:
        redirect_to_logger(stdout, stderr)
    redirect_to_logger(stdout, stderr)<commit_msg>Increase polling timeout<commit_after>#!/usr/bin/env python3

import os
import sys
import time
import select
import logging
import argparse
import threading
import subprocess

ETCD_IMAGE = "quay.io/coreos/etcd:v3.3.5"

LOG_LEVELS = {
    "critical": logging.CRITICAL,
    "error": logging.ERROR,
    "warning": logging.WARNING,
    "info": logging.INFO,
    "debug": logging.DEBUG,
}

LOG_COLORS = {
    "critical": "\x1b[31;1m",
    "error": "\x1b[31;1m",
    "warning": "\x1b[33;1m",
}

log = logging.getLogger(__name__)
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(levelname)s:%(message)s"))
log.addHandler(handler)

class ExcThread(threading.Thread):
    def __init__(self, target):
        super().__init__(target=target)
        self.error = None

    def run(self):
        try:
            self._target()
        except Exception as e:
            self.error = e

def join(*targets):
    threads = []

    for target in targets:
        t = ExcThread(target)
        t.start()
        threads.append(t)

    for t in threads:
        t.join()
    for t in threads:
        if t.error is not None:
            raise Exception("Thread error") from t.error

class Output:
    def __init__(self, pipe, level):
        self.pipe = pipe
        self.level = level
        self.lines = []

class ProcessResult:
    def __init__(self, rc, stdout, stderr):
        self.rc = rc
        self.stdout = stdout
        self.stderr = stderr

class DefaultDriver:
    def available(self):
        return True

    def clear(self):
        if run("yes | pachctl delete all", shell=True, raise_on_error=False).rc != 0:
            log.error("could not call `pachctl delete all`; most likely this just means that a pachyderm cluster hasn't been setup, but may indicate a bad state")

    def start(self):
        pass

    def inspect(self):
        pass

    def push_images(self, deploy_version, dash_image):
        pass

    def wait(self):
        while suppress("pachctl", "version") != 0:
            log.info("Waiting for pachyderm to come up...")
            time.sleep(1)

class MinikubeDriver(DefaultDriver):
    def available(self):
        return run("which", "minikube", raise_on_error=False).rc == 0

    def clear(self):
        run("minikube", "delete")

    def start(self):
        run("minikube", "start")

        while suppress("minikube", "status") != 0:
            log.info("Waiting for minikube to come up...")
            time.sleep(1)

    def push_images(self, deploy_version, dash_image):
        run("./etc/kube/push-to-minikube.sh", "pachyderm/pachd:{}".format(deploy_version))
        run("./etc/kube/push-to-minikube.sh", "pachyderm/worker:{}".format(deploy_version))
        run("./etc/kube/push-to-minikube.sh", ETCD_IMAGE)
        run("./etc/kube/push-to-minikube.sh", dash_image)

def parse_log_level(s):
    try:
        return LOG_LEVELS[s]
    except KeyError:
        raise Exception("Unknown log level: {}".format(s))

def redirect_to_logger(stdout, stderr):
    for io in select.select([stdout.pipe, stderr.pipe], [], [], 5000)[0]:
        line = io.readline().decode().rstrip()

        if line == "":
            continue

        dest = stdout if io == stdout.pipe else stderr
        log.log(LOG_LEVELS[dest.level], "{}{}\x1b[0m".format(LOG_COLORS.get(dest.level, ""), line))
        dest.lines.append(line)

def run(cmd, *args, raise_on_error=True, shell=False, stdout_log_level="info", stderr_log_level="error"):
    log.debug("Running `%s %s`", cmd, " ".join(args))

    proc = subprocess.Popen([cmd, *args], shell=shell, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    stdout = Output(proc.stdout, stdout_log_level)
    stderr = Output(proc.stderr, stderr_log_level)

    while proc.poll() is None:
        redirect_to_logger(stdout, stderr)
    redirect_to_logger(stdout, stderr)
{'fraction_non_alphanumeric': 0.13890691938205893, 'fraction_numerical': 0.007529533947812541, 'mean_word_length': 2.8233250620347397, 'pattern_counts': {'":': 18, '<': 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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '48059', 'n_tokens_mistral': 2594, 'n_tokens_neox': 2512, 'n_words': 579}