#!/usr/bin/env python
# -*- encoding: utf-8 -*-

#  Licensed to the Apache Software Foundation (ASF) under one
#  or more contributor license agreements.  See the NOTICE file
#  distributed with this work for additional information
#  regarding copyright ownership.  The ASF licenses this file
#  to you 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.

'''heron executor unittest'''
import os
import socket
import unittest2 as unittest
import json

from heron.executor.src.python.heron_executor import ProcessInfo
from heron.executor.src.python.heron_executor import HeronExecutor
from heron.proto.packing_plan_pb2 import PackingPlan

# pylint: disable=unused-argument
# pylint: disable=missing-docstring

def get_test_heron_internal_yaml():
  """Get the path to test_heron_internal.yaml

  For example, __file__ would be
  /tmp/_bazel_heron/randgen_dir/heron/heron/executor/tests/python/heron_executor_unittest.py
  """
  heron_dir = '/'.join(__file__.split('/')[:-5])
  yaml_path = os.path.join(heron_dir, 'heron/config/src/yaml/conf/test/test_heron_internals.yaml')
  override_path = os.path.join(
    heron_dir, 'heron/config/src/yaml/conf/test/test_override.yaml')

  return yaml_path, override_path

INTERNAL_CONF_PATH, OVERRIDE_PATH = get_test_heron_internal_yaml()
HOSTNAME = socket.gethostname()

class CommandEncoder(json.JSONEncoder):
  """Customized JSONEncoder that works with Command object"""
  def default(self, o):
    return o.cmd

class MockPOpen(object):
  """fake subprocess.Popen object that we can use to mock processes and pids"""
  next_pid = 0

  def __init__(self):
    self.pid = MockPOpen.next_pid
    MockPOpen.next_pid += 1

  @staticmethod
  def set_next_pid(next_pid):
    MockPOpen.next_pid = next_pid

class MockExecutor(HeronExecutor):
  """mock executor that overrides methods that don't apply to unit tests, like running processes"""
  def __init__(self, args):
    self.processes = []
    super(MockExecutor, self).__init__(args, None)

  # pylint: disable=no-self-use
  def _load_logging_dir(self, heron_internals_config_file):
    return "fake_dir"

  def _run_process(self, name, cmd, env=None):
    popen = MockPOpen()
    self.processes.append(ProcessInfo(popen, name, cmd))
    return popen

  def _get_jvm_version(self):
    return "1.8.y.x"

class HeronExecutorTest(unittest.TestCase):
  """Unittest for Heron Executor"""

  def get_expected_shell_command(container_id):
    return 'heron_shell_binary --port=shell-port ' \
           '--log_file_prefix=fake_dir/heron-shell-%s.log ' \
           '--secret=topid' % container_id

  def build_packing_plan(self, instance_distribution):
    packing_plan = PackingPlan()
    for container_id in list(instance_distribution.keys()):
      container_plan = packing_plan.container_plans.add()
      container_plan.id = int(container_id)
      for (component_name, global_task_id, component_index) in instance_distribution[container_id]:
        instance_plan = container_plan.instance_plans.add()
        instance_plan.component_name = component_name
        instance_plan.task_id = int(global_task_id)
        instance_plan.component_index = int(component_index)
    return packing_plan

  # pylint: disable=no-self-argument
  def get_expected_metricsmgr_command(container_id):
    return "heron_java_home/bin/java -Xmx1024M -XX:+PrintCommandLineFlags -verbosegc " \
           "-XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+PrintGCDateStamps -XX:+PrintGCCause " \
           "-XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=5 -XX:GCLogFileSize=100M " \
           "-XX:+PrintPromotionFailure -XX:+PrintTenuringDistribution -XX:+PrintHeapAtGC " \
           "-XX:+HeapDumpOnOutOfMemoryError -XX:+UseConcMarkSweepGC -XX:+PrintCommandLineFlags " \
           "-Xloggc:log-files/gc.metricsmgr-%d.log -Djava.net.preferIPv4Stack=true " \
           "-cp metricsmgr_classpath org.apache.heron.metricsmgr.MetricsManager " \
           "--id=metricsmgr-%d --port=metricsmgr_port " \
           "--topology=topname --cluster=cluster --role=role --environment=environ --topology-id=topid " \
           "--system-config-file=%s --override-config-file=%s --sink-config-file=metrics_sinks_config_file" %\
           (container_id, container_id, INTERNAL_CONF_PATH, OVERRIDE_PATH)

  def get_expected_metricscachemgr_command():
      return "heron_java_home/bin/java -Xmx1024M -XX:+PrintCommandLineFlags -verbosegc " \
             "-XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+PrintGCDateStamps -XX:+PrintGCCause " \
             "-XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=5 -XX:GCLogFileSize=100M " \
             "-XX:+PrintPromotionFailure -XX:+PrintTenuringDistribution -XX:+PrintHeapAtGC " \
             "-XX:+HeapDumpOnOutOfMemoryError -XX:+UseConcMarkSweepGC -XX:+PrintCommandLineFlags " \
             "-Xloggc:log-files/gc.metricscache.log -Djava.net.preferIPv4Stack=true " \
             "-cp metricscachemgr_classpath org.apache.heron.metricscachemgr.MetricsCacheManager " \
             "--metricscache_id metricscache-0 --master_port metricscachemgr_masterport " \
             "--stats_port metricscachemgr_statsport --topology_name topname --topology_id topid " \
             "--system_config_file %s --override_config_file %s " \
             "--sink_config_file metrics_sinks_config_file " \
             "--cluster cluster --role role --environment environ" %\
             (INTERNAL_CONF_PATH, OVERRIDE_PATH)

  def get_expected_healthmgr_command():
      return "heron_java_home/bin/java -Xmx1024M -XX:+PrintCommandLineFlags -verbosegc " \
             "-XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+PrintGCDateStamps -XX:+PrintGCCause " \
             "-XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=5 -XX:GCLogFileSize=100M " \
             "-XX:+PrintPromotionFailure -XX:+PrintTenuringDistribution -XX:+PrintHeapAtGC " \
             "-XX:+HeapDumpOnOutOfMemoryError -XX:+UseConcMarkSweepGC -XX:+PrintCommandLineFlags " \
             "-Xloggc:log-files/gc.healthmgr.log -Djava.net.preferIPv4Stack=true " \
             "-cp scheduler_classpath:healthmgr_classpath " \
             "org.apache.heron.healthmgr.HealthManager --cluster cluster --role role " \
             "--environment environ --topology_name topname --metricsmgr_port metricsmgr_port"

  def get_expected_instance_command(component_name, instance_id, container_id):
    instance_name = "container_%d_%s_%d" % (container_id, component_name, instance_id)
    return "heron_java_home/bin/java -Xmx320M -Xms320M -Xmn160M -XX:MaxMetaspaceSize=128M " \
           "-XX:MetaspaceSize=128M -XX:ReservedCodeCacheSize=64M -XX:+CMSScavengeBeforeRemark " \
           "-XX:TargetSurvivorRatio=90 -XX:+PrintCommandLineFlags -verbosegc -XX:+PrintGCDetails " \
           "-XX:+PrintGCTimeStamps -XX:+PrintGCDateStamps -XX:+PrintGCCause " \
           "-XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=5 -XX:GCLogFileSize=100M " \
           "-XX:+PrintPromotionFailure -XX:+PrintTenuringDistribution -XX:+PrintHeapAtGC " \
           "-XX:+HeapDumpOnOutOfMemoryError -XX:+UseConcMarkSweepGC -XX:ParallelGCThreads=4 " \
           "-Xloggc:log-files/gc.%s.log -Djava.net.preferIPv4Stack=true " \
           "-cp instance_classpath:classpath -XX:+HeapDumpOnOutOfMemoryError " \
           "org.apache.heron.instance.HeronInstance -topology_name topname -topology_id topid -instance_id %s -component_name %s -task_id %d -component_index 0 -stmgr_id stmgr-%d " \
           "-stmgr_port tmaster_controller_port -metricsmgr_port metricsmgr_port -system_config_file %s -override_config_file %s" \
           % (instance_name, instance_name, component_name, instance_id,
              container_id, INTERNAL_CONF_PATH, OVERRIDE_PATH)

  MockPOpen.set_next_pid(37)
  expected_processes_container_0 = [
      ProcessInfo(MockPOpen(), 'heron-tmaster',
                  'tmaster_binary --topology_name=topname --topology_id=topid '
                  '--zkhostportlist=zknode --zkroot=zkroot --myhost=%s --master_port=master_port '
                  '--controller_port=tmaster_controller_port --stats_port=tmaster_stats_port '
                  '--config_file=%s --override_config_file=%s '
                  '--metrics_sinks_yaml=metrics_sinks_config_file '
                  '--metricsmgr_port=metricsmgr_port '
                  '--ckptmgr_port=ckptmgr-port' % (HOSTNAME, INTERNAL_CONF_PATH, OVERRIDE_PATH)),
      ProcessInfo(MockPOpen(), 'heron-shell-0', get_expected_shell_command(0)),
      ProcessInfo(MockPOpen(), 'metricsmgr-0', get_expected_metricsmgr_command(0)),
      ProcessInfo(MockPOpen(), 'heron-metricscache', get_expected_metricscachemgr_command()),
      ProcessInfo(MockPOpen(), 'heron-healthmgr', get_expected_healthmgr_command()),
  ]

  MockPOpen.set_next_pid(37)
  expected_processes_container_1 = [
      ProcessInfo(MockPOpen(), 'stmgr-1',
                  'stmgr_binary --topology_name=topname --topology_id=topid '
                  '--topologydefn_file=topdefnfile --zkhostportlist=zknode --zkroot=zkroot '
                  '--stmgr_id=stmgr-1 '
                  '--instance_ids=container_1_word_3,container_1_exclaim1_2,container_1_exclaim1_1 '
                  '--myhost=%s --data_port=master_port '
                  '--local_data_port=tmaster_controller_port --metricsmgr_port=metricsmgr_port '
                  '--shell_port=shell-port --config_file=%s --override_config_file=%s '
                  '--ckptmgr_port=ckptmgr-port --ckptmgr_id=ckptmgr-1 '
                  '--metricscachemgr_mode=cluster'
                  % (HOSTNAME, INTERNAL_CONF_PATH, OVERRIDE_PATH)),
      ProcessInfo(MockPOpen(), 'container_1_word_3', get_expected_instance_command('word', 3, 1)),
      ProcessInfo(MockPOpen(), 'container_1_exclaim1_1',
                  get_expected_instance_command('exclaim1', 1, 1)),
      ProcessInfo(MockPOpen(), 'container_1_exclaim1_2',
                  get_expected_instance_command('exclaim1', 2, 1)),
      ProcessInfo(MockPOpen(), 'heron-shell-1', get_expected_shell_command(1)),
      ProcessInfo(MockPOpen(), 'metricsmgr-1', get_expected_metricsmgr_command(1)),
  ]

  MockPOpen.set_next_pid(37)
  expected_processes_container_7 = [
      ProcessInfo(MockPOpen(), 'container_7_word_11', get_expected_instance_command('word', 11, 7)),
      ProcessInfo(MockPOpen(), 'container_7_exclaim1_210',
                  get_expected_instance_command('exclaim1', 210, 7)),
      ProcessInfo(MockPOpen(), 'stmgr-7',
                  'stmgr_binary --topology_name=topname --topology_id=topid '
                  '--topologydefn_file=topdefnfile --zkhostportlist=zknode --zkroot=zkroot '
                  '--stmgr_id=stmgr-7 '
                  '--instance_ids=container_7_word_11,container_7_exclaim1_210 --myhost=%s '
                  '--data_port=master_port '
                  '--local_data_port=tmaster_controller_port --metricsmgr_port=metricsmgr_port '
                  '--shell_port=shell-port --config_file=%s --override_config_file=%s '
                  '--ckptmgr_port=ckptmgr-port --ckptmgr_id=ckptmgr-7 '
                  '--metricscachemgr_mode=cluster'
                  % (HOSTNAME, INTERNAL_CONF_PATH, OVERRIDE_PATH)),
      ProcessInfo(MockPOpen(), 'metricsmgr-7', get_expected_metricsmgr_command(7)),
      ProcessInfo(MockPOpen(), 'heron-shell-7', get_expected_shell_command(7)),
  ]

  def setUp(self):
    MockPOpen.set_next_pid(37)
    self.maxDiff = None
    self.executor_0 = MockExecutor(self.get_args(0))
    self.executor_1 = MockExecutor(self.get_args(1))
    self.executor_7 = MockExecutor(self.get_args(7))
    self.packing_plan_expected = self.build_packing_plan({
      1:[('word', '3', '0'), ('exclaim1', '2', '0'), ('exclaim1', '1', '0')],
      7:[('word', '11', '0'), ('exclaim1', '210', '0')],
    })

  # ./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>
  # <override_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>
  @staticmethod
  def get_args(shard_id):
    executor_args = [
      ("--shard", shard_id),
      ("--topology-name", "topname"),
      ("--topology-id", "topid"),
      ("--topology-defn-file", "topdefnfile"),
      ("--state-manager-connection", "zknode"),
      ("--state-manager-root", "zkroot"),
      ("--state-manager-config-file", "state_manager_config_file"),
      ("--tmaster-binary", "tmaster_binary"),
      ("--stmgr-binary", "stmgr_binary"),
      ("--metrics-manager-classpath", "metricsmgr_classpath"),
      ("--instance-jvm-opts", "LVhYOitIZWFwRHVtcE9uT3V0T2ZNZW1vcnlFcnJvcg(61)(61)"),
      ("--classpath", "classpath"),
      ("--master-port", "master_port"),
      ("--tmaster-controller-port", "tmaster_controller_port"),
      ("--tmaster-stats-port", "tmaster_stats_port"),
      ("--heron-internals-config-file", INTERNAL_CONF_PATH),
      ("--override-config-file", OVERRIDE_PATH),
      ("--component-ram-map", "exclaim1:536870912,word:536870912"),
      ("--component-jvm-opts", ""),
      ("--pkg-type", "jar"),
      ("--topology-binary-file", "topology_bin_file"),
      ("--heron-java-home", "heron_java_home"),
      ("--shell-port", "shell-port"),
      ("--heron-shell-binary", "heron_shell_binary"),
      ("--metrics-manager-port", "metricsmgr_port"),
      ("--cluster", "cluster"),
      ("--role", "role"),
      ("--environment", "environ"),
      ("--instance-classpath", "instance_classpath"),
      ("--metrics-sinks-config-file", "metrics_sinks_config_file"),
      ("--scheduler-classpath", "scheduler_classpath"),
      ("--scheduler-port", "scheduler_port"),
      ("--python-instance-binary", "python_instance_binary"),
      ("--cpp-instance-binary", "cpp_instance_binary"),
      ("--metricscache-manager-classpath", "metricscachemgr_classpath"),
      ("--metricscache-manager-master-port", "metricscachemgr_masterport"),
      ("--metricscache-manager-stats-port", "metricscachemgr_statsport"),
      ("--is-stateful", "is_stateful_enabled"),
      ("--checkpoint-manager-classpath", "ckptmgr_classpath"),
      ("--checkpoint-manager-port", "ckptmgr-port"),
      ("--checkpoint-manager-ram", "1073741824"),
      ("--stateful-config-file", "stateful_config_file"),
      ("--health-manager-mode", "cluster"),
      ("--health-manager-classpath", "healthmgr_classpath"),
      ("--metricscache-manager-mode", "cluster")
    ]

    args = ("%s=%s" % (arg[0], (str(arg[1]))) for arg in executor_args)
    command = "./heron-executor %s" % (" ".join(args))
    return command.split()


  def test_update_packing_plan(self):
    self.executor_0.update_packing_plan(self.packing_plan_expected)

    self.assertEqual(self.packing_plan_expected, self.executor_0.packing_plan)
    self.assertEqual({1: "stmgr-1", 7: "stmgr-7"}, self.executor_0.stmgr_ids)
    self.assertEqual(
      {0: "metricsmgr-0", 1: "metricsmgr-1", 7: "metricsmgr-7"}, self.executor_0.metricsmgr_ids)
    self.assertEqual(
      {0: "heron-shell-0", 1: "heron-shell-1", 7: "heron-shell-7"}, self.executor_0.heron_shell_ids)

  def test_launch_container_0(self):
    self.do_test_launch(self.executor_0, self.expected_processes_container_0)

  def test_launch_container_1(self):
    self.do_test_launch(self.executor_1, self.expected_processes_container_1)

  def test_launch_container_7(self):
    self.do_test_launch(self.executor_7, self.expected_processes_container_7)

  def do_test_launch(self, executor, expected_processes):
    executor.update_packing_plan(self.packing_plan_expected)
    executor.launch()
    monitored_processes = executor.processes_to_monitor

    # convert to (pid, name, command)
    found_processes = list([(process_info.pid, process_info.name, process_info.command_str) for process_info in executor.processes])
    found_monitored = list([(pinfo[0], pinfo[1].name, pinfo[1].command_str) for pinfo in list(monitored_processes.items())])
    found_processes.sort(key=lambda tuple: tuple[0])
    found_monitored.sort(key=lambda tuple: tuple[0])
    print("do_test_commands - found_processes: %s found_monitored: %s" \
          % (found_processes, found_monitored))
    self.assertEqual(found_processes, found_monitored)

    print("do_test_commands - expected_processes: %s monitored_processes: %s" \
          % (expected_processes, monitored_processes))
    self.assert_processes(expected_processes, monitored_processes)

  def test_change_instance_dist_container_1(self):
    MockPOpen.set_next_pid(37)
    self.executor_1.update_packing_plan(self.packing_plan_expected)
    current_commands = self.executor_1.get_commands_to_run()

    temp_dict = dict(
        list(map((lambda process_info: (process_info.name, process_info.command.split(' '))),
            self.expected_processes_container_1)))

    current_json = json.dumps(current_commands, sort_keys=True, cls=CommandEncoder).split(' ')
    temp_json = json.dumps(temp_dict, sort_keys=True).split(' ')

    print("current_json: %s" % current_json)
    print("temp_json: %s" % temp_json)

    # better test error report
    for (s1, s2) in zip(current_json, temp_json):
      self.assertEqual(s1, s2)

    # update instance distribution
    new_packing_plan = self.build_packing_plan(
      {1:[('word', '3', '0'), ('word', '2', '0'), ('exclaim1', '1', '0')]})
    self.executor_1.update_packing_plan(new_packing_plan)
    updated_commands = self.executor_1.get_commands_to_run()

    # get the commands to kill, keep and start and verify
    commands_to_kill, commands_to_keep, commands_to_start = \
      self.executor_1.get_command_changes(current_commands, updated_commands)

    self.assertEqual(['container_1_exclaim1_2', 'stmgr-1'], sorted(commands_to_kill.keys()))
    self.assertEqual(
        ['container_1_exclaim1_1', 'container_1_word_3', 'heron-shell-1', 'metricsmgr-1'],
        sorted(commands_to_keep.keys()))
    self.assertEqual(['container_1_word_2', 'stmgr-1'], sorted(commands_to_start.keys()))

  def assert_processes(self, expected_processes, found_processes):
    self.assertEqual(len(expected_processes), len(found_processes))
    for expected_process in expected_processes:
      self.assert_process(expected_process, found_processes)

  def assert_process(self, expected_process, found_processes):
    pid = expected_process.pid
    self.assertTrue(found_processes[pid])
    self.assertEqual(expected_process.name, found_processes[pid].name)
    self.assertEqual(expected_process.command, found_processes[pid].command_str)
    self.assertEqual(1, found_processes[pid].attempts)