<commit_before># Copyright 2016 Twitter. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

#!/usr/bin/env python2.7
""" The Heron executor is a process that runs on a container and is responsible for starting and
monitoring the processes of the topology and it's support services."""
import argparse
import atexit
import base64
import json
import os
import random
import signal
import string
import subprocess
import sys
import stat
import threading
import time
import yaml
import socket

from functools import partial



from heron.common.src.python.utils import log
# pylint: disable=unused-import
from heron.proto.packing_plan_pb2 import PackingPlan
from heron.statemgrs.src.python import statemanagerfactory
from heron.statemgrs.src.python import configloader
from heron.statemgrs.src.python.config import Config as StateMgrConfig

Log = log.Log

def print_usage():
  print (
      "Usage: ./heron-executor <shardid> <topname> <topid> <topdefnfile>"
      " <zknode> <zkroot> <tmaster_binary> <stmgr_binary>"
      " <metricsmgr_classpath> <instance_jvm_opts_in_base64> <classpath>"
      " <master_port> <tmaster_controller_port> <tmaster_stats_port> <heron_internals_config_file>"
      " <component_rammap> <component_jvm_opts_in_base64> <pkg_type> <topology_bin_file>"
      " <heron_java_home> <shell-port> <heron_shell_binary> <metricsmgr_port>"
      " <cluster> <role> <environ> <instance_classpath> <metrics_sinks_config_file>"
      " <scheduler_classpath> <scheduler_port> <python_instance_binary>"
      " <metricscachemgr_classpath> <metricscachemgr_masterport> <metricscachemgr_statsport>")

def id_map(prefix, container_plans, add_zero_id=False):
  ids = {}
  if add_zero_id:
    ids[0] = "%s-0" % prefix

  for container_plan in container_plans:
    ids[container_plan.id] = "%s-%d" % (prefix, container_plan.id)
  return ids

def stmgr_map(container_plans):
  return id_map("stmgr", container_plans)

def metricsmgr_map(container_plans):
  return id_map("metricsmgr", container_plans, True)

def heron_shell_map(container_plans):
  return id_map("heron-shell", container_plans, True)

def get_heron_executor_process_name(shard_id):
  return 'heron-executor-%d' % shard_id

def get_process_pid_filename(process_name):
  return '%s.pid' % process_name

def get_tmp_filename():
  return '%s.heron.tmp' % (''.join(random.choice(string.ascii_uppercase) for i in range(12)))

def atomic_write_file(path, content):
  """
  file.write(...) is not atomic.
  We write to a tmp file and then rename to target path since rename is atomic.
  We do this to avoid the content of file is dirty read/partially read by others.
  """
  # Write to a randomly tmp file
  tmp_file = get_tmp_filename()
  with open(tmp_file, 'w') as f:
    f.write(content)
    # make sure that all data is on disk
    f.flush()
    os.fsync(f.fileno())

  # Rename the tmp file
  os.rename(tmp_file, path)

def log_pid_for_process(process_name, pid):
  filename = get_process_pid_filename(process_name)
  Log.info('Logging pid %d to file %s' %(pid, filename))
  atomic_write_file(filename, str(pid))

class ProcessInfo(object):
  def __init__(self, process, name, command, attempts=1):
    """
    Container for info related to a running process
    :param process: the process POpen object
    :param name: the logical (i.e., unique) name of the process
    :param command: an array of strings comprising the command and it's args
    :param attempts: how many times the command has been run (defaults to 1)
    """
    self.process = process
    self.pid = process.pid
    self.name = name
    self.command = command
    self.command_str = ' '.join(command) # convenience for unit tests
    self.attempts = attempts

  def increment_attempts(self):
    self.attempts += 1
    return self

# pylint: disable=too-many-instance-attributes
class HeronExecutor(object):
  """ Heron executor is a class that is responsible for running each of the process on a given
  container. Based on the container id and the instance distribution, it determines if the container
  is a master node or a worker node and it starts processes accordingly."""
  def init_parsed_args(self, args):
    """ initialize from parsed arguments """
    parsed_args = self.parse_args(args)

    self.shard = parsed_args.shard
    self.topology_name = parsed_args.topology_name
    self.topology_id = parsed_args.topology_id
    self.topology_defn_file = parsed_args.topology_defn_file
    self.zknode = parsed_args.zknode
    self.zkroot = parsed_args.zkroot
    self.tmaster_binary = parsed_args.tmaster_binary
    self.stmgr_binary = parsed_args.stmgr_binary
    self.metricsmgr_classpath = parsed_args.metricsmgr_classpath
    self.metricscachemgr_classpath = parsed_args.metricscachemgr_classpath
    self.instance_jvm_opts =\
        base64.b64decode(parsed_args.instance_jvm_opts.lstrip('"').
                         rstrip('"').replace('&equals;', '='))
    self.classpath = parsed_args.classpath
    self.master_host = os.environ.get('HOST') if 'HOST' in os.environ else socket.gethostname()
    self.master_port = parsed_args.master_port
    self.tmaster_controller_port = parsed_args.tmaster_controller_port
    self.tmaster_stats_port = parsed_args.tmaster_stats_port
    self.heron_internals_config_file = parsed_args.heron_internals_config_file
    self.component_rammap =\
        map(lambda x: {x.split(':')[0]:
                           int(x.split(':')[1])}, parsed_args.component_rammap.split(','))
    self.component_rammap =\
        reduce(lambda x, y: dict(x.items() + y.items()), self.component_rammap)

    # component_jvm_opts_in_base64 itself is a base64-encoding-json-map, which is appended with
    # " at the start and end. It also escapes "=" to "&equals" due to aurora limitation
    # And the json is a map from base64-encoding-component-name to base64-encoding-jvm-options<commit_msg>Performing a check to determine if in docker environment<commit_after># Copyright 2016 Twitter. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

#!/usr/bin/env python2.7
""" The Heron executor is a process that runs on a container and is responsible for starting and
monitoring the processes of the topology and it's support services."""
import argparse
import atexit
import base64
import json
import os
import random
import signal
import string
import subprocess
import sys
import stat
import threading
import time
import yaml
import socket

from functools import partial



from heron.common.src.python.utils import log
# pylint: disable=unused-import
from heron.proto.packing_plan_pb2 import PackingPlan
from heron.statemgrs.src.python import statemanagerfactory
from heron.statemgrs.src.python import configloader
from heron.statemgrs.src.python.config import Config as StateMgrConfig

Log = log.Log

def print_usage():
  print (
      "Usage: ./heron-executor <shardid> <topname> <topid> <topdefnfile>"
      " <zknode> <zkroot> <tmaster_binary> <stmgr_binary>"
      " <metricsmgr_classpath> <instance_jvm_opts_in_base64> <classpath>"
      " <master_port> <tmaster_controller_port> <tmaster_stats_port> <heron_internals_config_file>"
      " <component_rammap> <component_jvm_opts_in_base64> <pkg_type> <topology_bin_file>"
      " <heron_java_home> <shell-port> <heron_shell_binary> <metricsmgr_port>"
      " <cluster> <role> <environ> <instance_classpath> <metrics_sinks_config_file>"
      " <scheduler_classpath> <scheduler_port> <python_instance_binary>"
      " <metricscachemgr_classpath> <metricscachemgr_masterport> <metricscachemgr_statsport>")

def id_map(prefix, container_plans, add_zero_id=False):
  ids = {}
  if add_zero_id:
    ids[0] = "%s-0" % prefix

  for container_plan in container_plans:
    ids[container_plan.id] = "%s-%d" % (prefix, container_plan.id)
  return ids

def stmgr_map(container_plans):
  return id_map("stmgr", container_plans)

def metricsmgr_map(container_plans):
  return id_map("metricsmgr", container_plans, True)

def heron_shell_map(container_plans):
  return id_map("heron-shell", container_plans, True)

def get_heron_executor_process_name(shard_id):
  return 'heron-executor-%d' % shard_id

def get_process_pid_filename(process_name):
  return '%s.pid' % process_name

def get_tmp_filename():
  return '%s.heron.tmp' % (''.join(random.choice(string.ascii_uppercase) for i in range(12)))

def atomic_write_file(path, content):
  """
  file.write(...) is not atomic.
  We write to a tmp file and then rename to target path since rename is atomic.
  We do this to avoid the content of file is dirty read/partially read by others.
  """
  # Write to a randomly tmp file
  tmp_file = get_tmp_filename()
  with open(tmp_file, 'w') as f:
    f.write(content)
    # make sure that all data is on disk
    f.flush()
    os.fsync(f.fileno())

  # Rename the tmp file
  os.rename(tmp_file, path)

def log_pid_for_process(process_name, pid):
  filename = get_process_pid_filename(process_name)
  Log.info('Logging pid %d to file %s' %(pid, filename))
  atomic_write_file(filename, str(pid))

def is_docker_environment():
  return os.path.isfile('/.dockerenv')

class ProcessInfo(object):
  def __init__(self, process, name, command, attempts=1):
    """
    Container for info related to a running process
    :param process: the process POpen object
    :param name: the logical (i.e., unique) name of the process
    :param command: an array of strings comprising the command and it's args
    :param attempts: how many times the command has been run (defaults to 1)
    """
    self.process = process
    self.pid = process.pid
    self.name = name
    self.command = command
    self.command_str = ' '.join(command) # convenience for unit tests
    self.attempts = attempts

  def increment_attempts(self):
    self.attempts += 1
    return self

# pylint: disable=too-many-instance-attributes
class HeronExecutor(object):
  """ Heron executor is a class that is responsible for running each of the process on a given
  container. Based on the container id and the instance distribution, it determines if the container
  is a master node or a worker node and it starts processes accordingly."""
  def init_parsed_args(self, args):
    """ initialize from parsed arguments """
    parsed_args = self.parse_args(args)

    self.shard = parsed_args.shard
    self.topology_name = parsed_args.topology_name
    self.topology_id = parsed_args.topology_id
    self.topology_defn_file = parsed_args.topology_defn_file
    self.zknode = parsed_args.zknode
    self.zkroot = parsed_args.zkroot
    self.tmaster_binary = parsed_args.tmaster_binary
    self.stmgr_binary = parsed_args.stmgr_binary
    self.metricsmgr_classpath = parsed_args.metricsmgr_classpath
    self.metricscachemgr_classpath = parsed_args.metricscachemgr_classpath
    self.instance_jvm_opts =\
        base64.b64decode(parsed_args.instance_jvm_opts.lstrip('"').
                         rstrip('"').replace('&equals;', '='))
    self.classpath = parsed_args.classpath
    # Needed for Docker environments since the hostname of a docker container is the container's
    # id within docker, rather than the host's hostname. NOTE: this 'HOST' env variable is not
    # guaranteed to be set in all Docker executor environments (outside of Marathon)
    if is_docker_environment():
      self.master_host = os.environ.get('HOST') if 'HOST' in os.environ else socket.gethostname()
    else:
      self.master_host = socket.gethostname()
    self.master_port = parsed_args.master_port
    self.tmaster_controller_port = parsed_args.tmaster_controller_port
    self.tmaster_stats_port = parsed_args.tmaster_stats_port
    self.heron_internals_config_file = parsed_args.heron_internals_config_file
    self.component_rammap =\
        map(lambda x: {x.split(':')[0]:
                           int(x.split(':')[1])}, parsed_args.component_rammap.split(','))
    self.component_rammap =\
        reduce(lambda x, y: dict(x.items() + y.items()), self.component_rammap)

    # component_jvm_opts_in_base64 itself is a base64-encoding-json-map, which is appended with
    # " at the start and end. It also escapes "=" to "&equals" due to aurora limitation
    # And the json is a map from base64-encoding-component-name to base64-encoding-jvm-options
{'fraction_non_alphanumeric': 0.0924933987174651, 'fraction_numerical': 0.005733685401735194, 'mean_word_length': 4.2960447463044344, 'pattern_counts': {'":': 0, '<': 71, '<?xml version=': 0, '>': 71, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '6334591', 'n_tokens_mistral': 4178, 'n_tokens_neox': 4147, 'n_words': 1209}