<commit_before>"""streamparse.ext.invoke

invoke tasks that are imported into a project's task.py.

Should be used like this::

    from invoke import task, run
    from streamparse.ext.invoke import *

    # your other tasks
"""
from __future__ import absolute_import, print_function
import os
import shutil
import sys
import json
from glob import glob
import subprocess

from invoke import run, task

from .util import get_config, die
from ..contextmanagers import ssh_tunnel


__all__ = ["run_local_topology", "deploy_topology"]


def stormdeps(topology=None):
    print("Storm dependencies (via lein):")
    run("lein deps :tree")
    print("Python dependencies (via pip):")
    run("cat virtualenvs/{}".format(topology))


def _get_topology(name=None):
    """Fetch a topology name and definition file.  If the name is None, we'll
    select the first .clj file in the topologies/ directory.

    :returns: a `tuple` containing (topology_name, topology_file)
    """
    config = get_config()
    topology_path = config["topology_specs"]
    if name is None:
        topology_files = glob("{}/*.clj".format(topology_path))
        if not topology_files:
            die("No topology definitions are defined in {}."
                .format(topology_path))
        topology_file = topology_files[0]
        name = topology_file.rstrip(".clj").lstrip(topology_path)
    else:
        topology_file = "{}.clj".format(os.path.join(topology_path, name))
        if not os.path.exists(topology_file):
            die("Topology definition file not found {}. You need to "
                "create a topology definition file first."
                .format(topology_file))

    return (name, topology_file)


@task
def uberjar_for_deploy():
    print("Creating jar for topology...")
    res = run("lein with-profile deploy uberjar", hide="stdout")
    if not res.ok:
        raise Exception("Unable to uberjar!\nSTDOUT:\n{}"
                        "\nSTDERR:\n{}".format(res.stdout, res.stderr))

    return res.stdout.split("\n")[0]


@task
def run_local_topology(name=None, time=5, debug=False):
    name, topology_file = _get_topology(name)
    print("Running {} topology...".format(name))
    if os.path.isdir("_resources/resources"):
        shutil.rmtree("_resources/resources")
    shutil.copytree("src", "_resources/resources")
    cmd = ["lein run -m streamparse.commands.run/-main", topology_file,
           "-t", str(time)]
    if debug:
        cmd.append("--debug")
    print(" ".join(cmd))
    run(" ".join(cmd))


@task
def deploy_topology(name=None, env="prod", debug=False):
    config = get_config()
    env = config["envs"][env]
    name, topology_file = _get_topology(name)
    if ":" in env["nimbus"]:
        host, port = env["nimbus"].split(":", 1)
        port = int(port)
    else:
        host = env["nimbus"]
        port = 6627

    # Prepare a JAR that doesn't have Storm dependencies packaged
    topology_jar = uberjar_for_deploy()

    print("Deploying {} topology...".format(name))
    with ssh_tunnel(env["user"], host, 6627, port):
        print("ssh tunnel to Nimbus established.")
        os.environ["JVM_OPTS"] = "-Dstorm.jar={}".format(topology_jar)
        cmd = ["lein",
               "with-profile deploy",
               "run -m streamparse.commands.submit_topology/-main",
               topology_file]
        print(" ".join(cmd))
        run(" ".join(cmd))

<commit_msg>Fix bug in fetching the path to the uberjar for submission to remote cluster. #1<commit_after>"""streamparse.ext.invoke

invoke tasks that are imported into a project's task.py.

Should be used like this::

    from invoke import task, run
    from streamparse.ext.invoke import *

    # your other tasks
"""
from __future__ import absolute_import, print_function
import os
import shutil
import sys
import json
from glob import glob
import subprocess

from invoke import run, task

from .util import get_config, die
from ..contextmanagers import ssh_tunnel


__all__ = ["run_local_topology", "deploy_topology"]


def stormdeps(topology=None):
    print("Storm dependencies (via lein):")
    run("lein deps :tree")
    print("Python dependencies (via pip):")
    run("cat virtualenvs/{}".format(topology))


def _get_topology(name=None):
    """Fetch a topology name and definition file.  If the name is None, we'll
    select the first .clj file in the topologies/ directory.

    :returns: a `tuple` containing (topology_name, topology_file)
    """
    config = get_config()
    topology_path = config["topology_specs"]
    if name is None:
        topology_files = glob("{}/*.clj".format(topology_path))
        if not topology_files:
            die("No topology definitions are defined in {}."
                .format(topology_path))
        topology_file = topology_files[0]
        name = topology_file.rstrip(".clj").lstrip(topology_path)
    else:
        topology_file = "{}.clj".format(os.path.join(topology_path, name))
        if not os.path.exists(topology_file):
            die("Topology definition file not found {}. You need to "
                "create a topology definition file first."
                .format(topology_file))

    return (name, topology_file)


@task
def uberjar_for_deploy():
    print("Creating topology uber-JAR...")
    res = run("lein uberjar", hide="stdout")
    if not res.ok:
        raise Exception("Unable to uberjar!\nSTDOUT:\n{}"
                        "\nSTDERR:\n{}".format(res.stdout, res.stderr))

    # TODO: This is a lil dicey, should get the uberjar based on settings or
    # something
    lines = [l.strip().lstrip("Created ") for l in res.stdout.split()
             if "standalone.jar" in l]
    return lines[0]


@task
def run_local_topology(name=None, time=5, debug=False):
    """Run a topology locally using Storm's LocalCluster class."""
    name, topology_file = _get_topology(name)
    print("Running {} topology...".format(name))
    if os.path.isdir("_resources/resources"):
        shutil.rmtree("_resources/resources")
    shutil.copytree("src", "_resources/resources")
    cmd = ["lein run -m streamparse.commands.run/-main", topology_file,
           "-t", str(time)]
    if debug:
        cmd.append("--debug")
    run(" ".join(cmd))


@task
def deploy_topology(name=None, env="prod", debug=False):
    """Deploy a topology to a remote Storm cluster."""
    config = get_config()

    # TODO: Update virtualenvs on Storm workers

    env = config["envs"][env]
    name, topology_file = _get_topology(name)
    if ":" in env["nimbus"]:
        host, port = env["nimbus"].split(":", 1)
        port = int(port)
    else:
        host = env["nimbus"]
        port = 6627

    # Prepare a JAR that doesn't have Storm dependencies packaged
    topology_jar = uberjar_for_deploy()

    print('Deploying "{}" topology...'.format(name))
    with ssh_tunnel(env["user"], host, 6627, port):
        print("ssh tunnel to Nimbus {}:{} established."
              .format(host, port))
        jvm_opts = [
            "-Dstorm.jar={}".format(topology_jar),
            "-Dstorm.options=",
            "-Dstorm.conf.file=",
        ]
        os.environ["JVM_OPTS"] = " ".join(jvm_opts)
        cmd = ["lein",
               "run -m streamparse.commands.submit_topology/-main",
               topology_file]
        run(" ".join(cmd))


{'fraction_non_alphanumeric': 0.13033369803063458, 'fraction_numerical': 0.0034190371991247265, 'mean_word_length': 2.89818763326226, 'pattern_counts': {'":': 4, '<': 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': '7359633', 'n_tokens_mistral': 2340, 'n_tokens_neox': 2222, 'n_words': 604}