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

## NB: not all commands work ##

"""Cascaded queue administration.

londiste.py INI pause [NODE [CONS]]

setadm.py INI pause NODE [CONS]

"""

import optparse
import os.path
import Queue
import sys
import threading
import time

import skytools
from skytools import UsageError, DBError
from pgq.cascade.nodeinfo import *

__all__ = ['CascadeAdmin']

RESURRECT_DUMP_FILE = "resurrect-lost-events.json"

command_usage = """\
%prog [options] INI CMD [subcmd args]

Node Initialization:
  create-root   NAME [PUBLIC_CONNSTR]
  create-branch NAME [PUBLIC_CONNSTR] --provider=<public_connstr>
  create-leaf   NAME [PUBLIC_CONNSTR] --provider=<public_connstr>
    All of the above initialize a node

Node Administration:
  pause                 Pause node worker
  resume                Resume node worker
  wait-root             Wait until node has caught up with root
  wait-provider         Wait until node has caught up with provider
  status                Show cascade state
  node-status           Show status of local node
  members               Show members in set

Cascade layout change:
  change-provider --provider NEW_NODE
    Change where worker reads from

  takeover FROM_NODE [--all] [--dead]
    Take other node position

  drop-node NAME
    Remove node from cascade

  tag-dead NODE ..
    Tag node as dead

  tag-alive NODE ..
    Tag node as alive
"""

standalone_usage = """
setadm extra switches:

  pause/resume/change-provider:
    --node=NODE_NAME | --consumer=CONSUMER_NAME

  create-root/create-branch/create-leaf:
    --worker=WORKER_NAME
"""


class CascadeAdmin(skytools.AdminScript):
    """Cascaded PgQ administration."""
    queue_name = None
    queue_info = None
    extra_objs = []
    local_node = None
    root_node_name = None

    commands_without_pidfile = ['status', 'node-status', 'node-info']

    def __init__(self, svc_name, dbname, args, worker_setup = False):
        skytools.AdminScript.__init__(self, svc_name, args)
        self.initial_db_name = dbname
        if worker_setup:
            self.options.worker = self.job_name
            self.options.consumer = self.job_name

    def init_optparse(self, parser = None):
        """Add SetAdmin switches to parser."""
        p = skytools.AdminScript.init_optparse(self, parser)

        usage = command_usage + standalone_usage
        p.set_usage(usage.strip())

        g = optparse.OptionGroup(p, "actual queue admin options")
        g.add_option("--connstr", action="store_true",
                     help = "initial connect string")
        g.add_option("--provider",
                     help = "init: connect string for provider")
        g.add_option("--queue",
                     help = "specify queue name")
        g.add_option("--worker",
                     help = "create: specify worker name")
        g.add_option("--node",
                     help = "specify node name")
        g.add_option("--consumer",
                     help = "specify consumer name")
        g.add_option("--target",
                    help = "takeover: specify node to take over")
        g.add_option("--merge",
                    help = "create-node: combined queue name")
        g.add_option("--dead", action="append",
                    help = "tag some node as dead")
        g.add_option("--dead-root", action="store_true",
                    help = "tag some node as dead")
        g.add_option("--dead-branch", action="store_true",
                    help = "tag some node as dead")
        g.add_option("--sync-watermark",
                    help = "list of node names to sync with")
        p.add_option_group(g)
        return p

    def reload(self):
        """Reload config."""
        skytools.AdminScript.reload(self)
        if self.options.queue:
            self.queue_name = self.options.queue
        else:
            self.queue_name = self.cf.get('queue_name', '')
            if not self.queue_name:
                self.queue_name = self.cf.get('pgq_queue_name', '')
                if not self.queue_name:
                    raise Exception('"queue_name" not specified in config')

    #
    # Node initialization.
    #

    def cmd_install(self):
        db = self.get_database(self.initial_db_name)
        self.install_code(db)

    def cmd_create_root(self, *args):
        return self.create_node('root', args)

    def cmd_create_branch(self, node_name, *args):
        return self.create_node('branch', args)

    def cmd_create_leaf(self, node_name, *args):
        return self.create_node('leaf', args)

    def create_node(self, node_type, args):
        """Generic node init."""

        if node_type not in ('root', 'branch', 'leaf'):
            raise Exception('unknown node type')

        # load node name
        if len(args) > 0:
            node_name = args[0]
        else:
            node_name = self.cf.get('node_name', '')
        if not node_name:
            raise UsageError('Node public location must be given either in command line or config')

        # load node public location
        if len(args) > 1:<commit_msg>cascadeadmin: forgot to make node_name optional everywhere<commit_after>#! /usr/bin/env python

## NB: not all commands work ##

"""Cascaded queue administration.

londiste.py INI pause [NODE [CONS]]

setadm.py INI pause NODE [CONS]

"""

import optparse
import os.path
import Queue
import sys
import threading
import time

import skytools
from skytools import UsageError, DBError
from pgq.cascade.nodeinfo import *

__all__ = ['CascadeAdmin']

RESURRECT_DUMP_FILE = "resurrect-lost-events.json"

command_usage = """\
%prog [options] INI CMD [subcmd args]

Node Initialization:
  create-root   NAME [PUBLIC_CONNSTR]
  create-branch NAME [PUBLIC_CONNSTR] --provider=<public_connstr>
  create-leaf   NAME [PUBLIC_CONNSTR] --provider=<public_connstr>
    All of the above initialize a node

Node Administration:
  pause                 Pause node worker
  resume                Resume node worker
  wait-root             Wait until node has caught up with root
  wait-provider         Wait until node has caught up with provider
  status                Show cascade state
  node-status           Show status of local node
  members               Show members in set

Cascade layout change:
  change-provider --provider NEW_NODE
    Change where worker reads from

  takeover FROM_NODE [--all] [--dead]
    Take other node position

  drop-node NAME
    Remove node from cascade

  tag-dead NODE ..
    Tag node as dead

  tag-alive NODE ..
    Tag node as alive
"""

standalone_usage = """
setadm extra switches:

  pause/resume/change-provider:
    --node=NODE_NAME | --consumer=CONSUMER_NAME

  create-root/create-branch/create-leaf:
    --worker=WORKER_NAME
"""


class CascadeAdmin(skytools.AdminScript):
    """Cascaded PgQ administration."""
    queue_name = None
    queue_info = None
    extra_objs = []
    local_node = None
    root_node_name = None

    commands_without_pidfile = ['status', 'node-status', 'node-info']

    def __init__(self, svc_name, dbname, args, worker_setup = False):
        skytools.AdminScript.__init__(self, svc_name, args)
        self.initial_db_name = dbname
        if worker_setup:
            self.options.worker = self.job_name
            self.options.consumer = self.job_name

    def init_optparse(self, parser = None):
        """Add SetAdmin switches to parser."""
        p = skytools.AdminScript.init_optparse(self, parser)

        usage = command_usage + standalone_usage
        p.set_usage(usage.strip())

        g = optparse.OptionGroup(p, "actual queue admin options")
        g.add_option("--connstr", action="store_true",
                     help = "initial connect string")
        g.add_option("--provider",
                     help = "init: connect string for provider")
        g.add_option("--queue",
                     help = "specify queue name")
        g.add_option("--worker",
                     help = "create: specify worker name")
        g.add_option("--node",
                     help = "specify node name")
        g.add_option("--consumer",
                     help = "specify consumer name")
        g.add_option("--target",
                    help = "takeover: specify node to take over")
        g.add_option("--merge",
                    help = "create-node: combined queue name")
        g.add_option("--dead", action="append",
                    help = "tag some node as dead")
        g.add_option("--dead-root", action="store_true",
                    help = "tag some node as dead")
        g.add_option("--dead-branch", action="store_true",
                    help = "tag some node as dead")
        g.add_option("--sync-watermark",
                    help = "list of node names to sync with")
        p.add_option_group(g)
        return p

    def reload(self):
        """Reload config."""
        skytools.AdminScript.reload(self)
        if self.options.queue:
            self.queue_name = self.options.queue
        else:
            self.queue_name = self.cf.get('queue_name', '')
            if not self.queue_name:
                self.queue_name = self.cf.get('pgq_queue_name', '')
                if not self.queue_name:
                    raise Exception('"queue_name" not specified in config')

    #
    # Node initialization.
    #

    def cmd_install(self):
        db = self.get_database(self.initial_db_name)
        self.install_code(db)

    def cmd_create_root(self, *args):
        return self.create_node('root', args)

    def cmd_create_branch(self, *args):
        return self.create_node('branch', args)

    def cmd_create_leaf(self, *args):
        return self.create_node('leaf', args)

    def create_node(self, node_type, args):
        """Generic node init."""

        if node_type not in ('root', 'branch', 'leaf'):
            raise Exception('unknown node type')

        # load node name
        if len(args) > 0:
            node_name = args[0]
        else:
            node_name = self.cf.get('node_name', '')
        if not node_name:
            raise UsageError('Node name must be given either in command line or config')

        # load node public location
        if len(args) > 1:
{'fraction_non_alphanumeric': 0.10450819672131148, 'fraction_numerical': 0.000585480093676815, 'mean_word_length': 2.2148682559598494, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 11, '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': '3280014', 'n_tokens_mistral': 3051, 'n_tokens_neox': 2920, 'n_words': 822}