<commit_before>"""Interface classes to Storm clusters. As things are, they implement
cluster-topologies interactions, but not cluster management.
"""
from __future__ import absolute_import

import os
import signal
import subprocess

from pyleus.exception import ConfigurationError
from pyleus.exception import StormError


TOPOLOGY_BUILDER_CLASS = "com.yelp.pyleus.PyleusTopologyBuilder"
LOCAL_OPTION = "--local"
DEBUG_OPTION = "--debug"
STORM_JAR_JVM_OPTS = "STORM_JAR_JVM_OPTS"


def _watch_over_storm(storm_pid):
    """Ensure that if the pyleus process is killed, also the storm process
    will terminate.
    """
    def _kill_storm_handler(signum, frame):
        # Killing the storm process is enough for killing all python
        # subprocesses
        os.kill(storm_pid, signal.SIGTERM)

    signal.signal(signal.SIGTERM, _kill_storm_handler)
    signal.signal(signal.SIGINT, _kill_storm_handler)


def _get_storm_cmd_env(jvm_opts):
    """Return a copy of os.environ containing JVM options from the user.

    If no JVM options were specified, return None to defer to the default
    behavior of subprocess.Popen.
    """
    if jvm_opts:
        env = os.environ.copy()
        env[STORM_JAR_JVM_OPTS] = jvm_opts
        return env

    return None


class StormCluster(object):
    """Object representing an interface to a Storm cluster.
    All the requests are basically translated into Storm commands.
    """
    def __init__(self, storm_cmd_path, nimbus_ip, verbose, jvm_opts):
        """Create the cluster object."""

        self.storm_cmd_path = storm_cmd_path

        if nimbus_ip is None:
            raise ConfigurationError(
                "You must specify a storm cluster IP address."
                " Use option <storm_cluster_ip> in the configuration file"
                " or the command line option --storm-cluster")

        self.nimbus_ip = nimbus_ip
        self.verbose = verbose
        self.jvm_opts = jvm_opts

    def _exec_storm_cmd(self, cmd, verbose=None):
        """Interface to any Storm command."""
        out_stream = None
        if verbose is None:
            verbose = self.verbose
        if not verbose:
            out_stream = open(os.devnull, "w")

        storm_cmd = [self.storm_cmd_path]
        storm_cmd += cmd
        storm_cmd += ["-c", "nimbus.host={0}".format(self.nimbus_ip)]

        env = _get_storm_cmd_env(self.jvm_opts)

        proc = subprocess.Popen(storm_cmd,
                                stdout=out_stream,
                                stderr=subprocess.STDOUT,
                                env=env)
        out_data, _ = proc.communicate()
        if proc.returncode != 0:
            raise StormError(
                "Storm command failed. Run with --verbose for more info.")
        return out_data

    def submit(self, jar_path):
        """Submit the pyleus topology jar to the Storm cluster."""<commit_msg>Rename nimbus_ip -> nimbus in StormCluster<commit_after>"""Interface classes to Storm clusters. As things are, they implement
cluster-topologies interactions, but not cluster management.
"""
from __future__ import absolute_import

import os
import signal
import subprocess

from pyleus.exception import ConfigurationError
from pyleus.exception import StormError


TOPOLOGY_BUILDER_CLASS = "com.yelp.pyleus.PyleusTopologyBuilder"
LOCAL_OPTION = "--local"
DEBUG_OPTION = "--debug"
STORM_JAR_JVM_OPTS = "STORM_JAR_JVM_OPTS"


def _watch_over_storm(storm_pid):
    """Ensure that if the pyleus process is killed, also the storm process
    will terminate.
    """
    def _kill_storm_handler(signum, frame):
        # Killing the storm process is enough for killing all python
        # subprocesses
        os.kill(storm_pid, signal.SIGTERM)

    signal.signal(signal.SIGTERM, _kill_storm_handler)
    signal.signal(signal.SIGINT, _kill_storm_handler)


def _get_storm_cmd_env(jvm_opts):
    """Return a copy of os.environ containing JVM options from the user.

    If no JVM options were specified, return None to defer to the default
    behavior of subprocess.Popen.
    """
    if jvm_opts:
        env = os.environ.copy()
        env[STORM_JAR_JVM_OPTS] = jvm_opts
        return env

    return None


class StormCluster(object):
    """Object representing an interface to a Storm cluster.
    All the requests are basically translated into Storm commands.
    """

    def __init__(self, storm_cmd_path, nimbus, verbose, jvm_opts):
        """Create the cluster object."""

        self.storm_cmd_path = storm_cmd_path

        if nimbus is None:
            raise ConfigurationError(
                "You must specify a storm cluster IP address."
                " Use option <storm_cluster_ip> in the configuration file"
                " or the command line option --storm-cluster")

        self.nimbus = nimbus
        self.verbose = verbose
        self.jvm_opts = jvm_opts

    def _exec_storm_cmd(self, cmd, verbose=None):
        """Interface to any Storm command."""
        out_stream = None
        if verbose is None:
            verbose = self.verbose
        if not verbose:
            out_stream = open(os.devnull, "w")

        storm_cmd = [self.storm_cmd_path]
        storm_cmd += cmd
        storm_cmd += ["-c", "nimbus.host={0}".format(self.nimbus)]

        env = _get_storm_cmd_env(self.jvm_opts)

        proc = subprocess.Popen(storm_cmd,
                                stdout=out_stream,
                                stderr=subprocess.STDOUT,
                                env=env)
        out_data, _ = proc.communicate()
        if proc.returncode != 0:
            raise StormError(
                "Storm command failed. Run with --verbose for more info.")
        return out_data

    def submit(self, jar_path):
        """Submit the pyleus topology jar to the Storm cluster."""
{'fraction_non_alphanumeric': 0.07928608203192038, 'fraction_numerical': 0.0006864595846919513, 'mean_word_length': 2.489820359281437, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 6, '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': '745951', 'n_tokens_mistral': 1725, 'n_tokens_neox': 1665, 'n_words': 499}