#!/usr/bin/env ambari-python-wrap
"""
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.
"""
from resource_management.core.logger import Logger
import json
import math
import re
from resource_management.libraries.functions import format


class ADH15StackAdvisor(ADH14StackAdvisor):
  def __init__(self):
      super(ADH15StackAdvisor, self).__init__()
      Logger.initialize_logger()

  def getServiceConfigurationRecommenderDict(self):
      parentRecommendConfDict = super(ADH15StackAdvisor, self).getServiceConfigurationRecommenderDict()
      childRecommendConfDict = {
        "DRUID": self.recommendDruidConfigurations,
        "SUPERSET": self.recommendSupersetConfigurations,
        "ATLAS": self.recommendAtlasConfigurations,
        "TEZ": self.recommendTezConfigurations,
        "RANGER": self.recommendRangerConfigurations,
        "RANGER_KMS": self.recommendRangerKMSConfigurations,
        "HDFS": self.recommendHDFSConfigurations,
        "HIVE": self.recommendHIVEConfigurations,
        "HBASE": self.recommendHBASEConfigurations,
        "YARN": self.recommendYARNConfigurations,
        "KAFKA": self.recommendKAFKAConfigurations,
        "STORM": self.recommendSTORMConfigurations,
        "SPARK2": self.recommendSPARK2Configurations,
        "ZEPPELIN": self.recommendZEPPELINConfigurations
      }
      parentRecommendConfDict.update(childRecommendConfDict)
      return parentRecommendConfDict

  def recommendSTORMConfigurations(self, configurations, clusterData, services, hosts):
    """
    In HDF-2.6.1 we introduced a new way of doing Auto Credentials with services such as
    HDFS, HIVE, HBASE. This method will update the required configs for autocreds if the users installs
    STREAMLINE service.
    """
    super(ADH15StackAdvisor, self).recommendStormConfigurations(configurations, clusterData, services, hosts)
    storm_site = self.getServicesSiteProperties(services, "storm-site")
    storm_env = self.getServicesSiteProperties(services, "storm-env")
    putStormSiteProperty = self.putProperty(configurations, "storm-site", services)
    putStormSiteAttributes = self.putPropertyAttribute(configurations, "storm-site")
    security_enabled = self.isSecurityEnabled(services)
    servicesList = [service["StackServices"]["service_name"] for service in services["services"]]

    if storm_env and storm_site and security_enabled and 'STREAMLINE' in servicesList:
      storm_nimbus_impersonation_acl = storm_site["nimbus.impersonation.acl"] if "nimbus.impersonation.acl" in storm_site else None
      streamline_env = self.getServicesSiteProperties(services, "streamline-env")
      if streamline_env:
        _streamline_principal_name = streamline_env['streamline_principal_name'] if 'streamline_principal_name' in streamline_env else None
        if _streamline_principal_name is not None and storm_nimbus_impersonation_acl is not None:
          streamline_bare_principal = get_bare_principal(_streamline_principal_name)
          storm_nimbus_impersonation_acl.replace('{{streamline_bare_principal}}', streamline_bare_principal)
          putStormSiteProperty('nimbus.impersonation.acl', storm_nimbus_impersonation_acl)

      storm_nimbus_autocred_plugin_classes = storm_site["nimbus.autocredential.plugins.classes"] if "nimbus.autocredential.plugins.classes" in storm_site else None
      if storm_nimbus_autocred_plugin_classes is not None:
        new_storm_nimbus_autocred_plugin_classes = ['org.apache.storm.hdfs.security.AutoHDFS',
                                                    'org.apache.storm.hbase.security.AutoHBase',
                                                    'org.apache.storm.hive.security.AutoHive']
        new_conf = DefaultStackAdvisor.appendToYamlString(storm_nimbus_autocred_plugin_classes,
                                      new_storm_nimbus_autocred_plugin_classes)

        putStormSiteProperty("nimbus.autocredential.plugins.classes", new_conf)
      else:
        putStormSiteProperty("nimbus.autocredential.plugins.classes", "['org.apache.storm.hdfs.security.AutoHDFS', 'org.apache.storm.hbase.security.AutoHBase', 'org.apache.storm.hive.security.AutoHive']")


      storm_nimbus_credential_renewer_classes = storm_site["nimbus.credential.renewers.classes"] if "nimbus.credential.renewers.classes" in storm_site else None
      if storm_nimbus_credential_renewer_classes is not None:
        new_storm_nimbus_credential_renewer_classes_array = ['org.apache.storm.hdfs.security.AutoHDFS',
                                                             'org.apache.storm.hbase.security.AutoHBase',
                                                             'org.apache.storm.hive.security.AutoHive']
        new_conf = DefaultStackAdvisor.appendToYamlString(storm_nimbus_credential_renewer_classes,
                                      new_storm_nimbus_credential_renewer_classes_array)
        putStormSiteProperty("nimbus.autocredential.plugins.classes", new_conf)
      else:
        putStormSiteProperty("nimbus.credential.renewers.classes", "['org.apache.storm.hdfs.security.AutoHDFS', 'org.apache.storm.hbase.security.AutoHBase', 'org.apache.storm.hive.security.AutoHive']")
      putStormSiteProperty("nimbus.credential.renewers.freq.secs", "82800")
    pass

  def recommendSPARK2Configurations(self, configurations, clusterData, services, hosts):
    """
    :type configurations dict
    :type clusterData dict
    :type services dict
    :type hosts dict
    """
    super(ADH15StackAdvisor, self).recommendSpark2Configurations(configurations, clusterData, services, hosts)
    self.__addZeppelinToLivy2SuperUsers(configurations, services)

  def recommendZEPPELINConfigurations(self, configurations, clusterData, services, hosts):
    """
    :type configurations dict
    :type clusterData dict
    :type services dict
    :type hosts dict
    """
    super(ADH15StackAdvisor, self).recommendZeppelinConfigurations(configurations, clusterData, services, hosts)
    self.__addZeppelinToLivy2SuperUsers(configurations, services)

  def recommendAtlasConfigurations(self, configurations, clusterData, services, hosts):
    super(ADH15StackAdvisor, self).recommendAtlasConfigurations(configurations, clusterData, services, hosts)
    servicesList = [service["StackServices"]["service_name"] for service in services["services"]]
    putAtlasApplicationProperty = self.putProperty(configurations, "application-properties", services)

    knox_host = 'localhost'
    knox_port = '8443'
    if 'KNOX' in servicesList:
      knox_hosts = self.getComponentHostNames(services, "KNOX", "KNOX_GATEWAY")
      if len(knox_hosts) > 0:
        knox_hosts.sort()
        knox_host = knox_hosts[0]
      if 'gateway-site' in services['configurations'] and 'gateway.port' in services['configurations']["gateway-site"]["properties"]:
        knox_port = services['configurations']["gateway-site"]["properties"]['gateway.port']
      putAtlasApplicationProperty('atlas.sso.knox.providerurl', 'https://{0}:{1}/gateway/knoxsso/api/v1/websso'.format(knox_host, knox_port))

    knox_service_user = ''
    if 'KNOX' in servicesList and 'knox-env' in services['configurations']:
      knox_service_user = services['configurations']['knox-env']['properties']['knox_user']
    else:
      knox_service_user = 'knox'
    putAtlasApplicationProperty('atlas.proxyusers',knox_service_user)

  def recommendDruidConfigurations(self, configurations, clusterData, services, hosts):

      # druid is not in list of services to be installed
      if 'druid-common' not in services['configurations']:
        return

      componentsListList = [service["components"] for service in services["services"]]
      componentsList = [item["StackServiceComponents"] for sublist in componentsListList for item in sublist]
      servicesList = [service["StackServices"]["service_name"] for service in services["services"]]
      putCommonProperty = self.putProperty(configurations, "druid-common", services)

      putCommonProperty('druid.zk.service.host', self.getZKHostPortString(services))
      self.recommendDruidMaxMemoryLimitConfigurations(configurations, clusterData, services, hosts)

      # recommending the metadata storage uri
      database_name = services['configurations']["druid-common"]["properties"]["database_name"]
      metastore_hostname = services['configurations']["druid-common"]["properties"]["metastore_hostname"]
      database_type = services['configurations']["druid-common"]["properties"]["druid.metadata.storage.type"]
      metadata_storage_port = "1527"
      mysql_module_name = "mysql-metadata-storage"
      postgres_module_name = "postgresql-metadata-storage"
      extensions_load_list = services['configurations']['druid-common']['properties']['druid.extensions.loadList']
      putDruidCommonProperty = self.putProperty(configurations, "druid-common", services)

      extensions_load_list = self.removeFromList(extensions_load_list, mysql_module_name)
      extensions_load_list = self.removeFromList(extensions_load_list, postgres_module_name)

      if database_type == 'mysql':
          metadata_storage_port = "3306"
          extensions_load_list = self.addToList(extensions_load_list, mysql_module_name)

      if database_type == 'postgresql':
          extensions_load_list = self.addToList(extensions_load_list, postgres_module_name)
          metadata_storage_port = "5432"

      putDruidCommonProperty('druid.metadata.storage.connector.port', metadata_storage_port)
      putDruidCommonProperty('druid.metadata.storage.connector.connectURI',
                             self.getMetadataConnectionString(database_type).format(metastore_hostname, database_name,
                                                                                    metadata_storage_port))
      # HDFS is installed
      if "HDFS" in servicesList and "hdfs-site" in services["configurations"]:
          # recommend HDFS as default deep storage
          extensions_load_list = self.addToList(extensions_load_list, "druid-hdfs-storage")
          putCommonProperty("druid.storage.type", "hdfs")
          putCommonProperty("druid.storage.storageDirectory", "/apps/druid/warehouse")
          # configure indexer logs configs
          putCommonProperty("druid.indexer.logs.type", "hdfs")
          putCommonProperty("druid.indexer.logs.directory", "/user/druid/logs")

      if "KAFKA" in servicesList:
          extensions_load_list = self.addToList(extensions_load_list, "druid-kafka-indexing-service")

      if 'AMBARI_METRICS' in servicesList:
        extensions_load_list = self.addToList(extensions_load_list, "ambari-metrics-emitter")

      putCommonProperty('druid.extensions.loadList', extensions_load_list)

      # JVM Configs go to env properties
      putEnvProperty = self.putProperty(configurations, "druid-env", services)

      # processing thread pool and memory configs
      for component in ['DRUID_HISTORICAL', 'DRUID_BROKER']:
          component_hosts = self.getHostsWithComponent("DRUID", component, services, hosts)
          nodeType = self.DRUID_COMPONENT_NODE_TYPE_MAP[component]
          putComponentProperty = self.putProperty(configurations, format("druid-{nodeType}"), services)
          if (component_hosts is not None and len(component_hosts) > 0):
              totalAvailableCpu = self.getMinCpu(component_hosts)
              processingThreads = 1
              if totalAvailableCpu > 1:
                  processingThreads = totalAvailableCpu - 1
              numMergeBuffers = max(2, processingThreads/4)
              putComponentProperty('druid.processing.numThreads', processingThreads)
              putComponentProperty('druid.server.http.numThreads', max(10, (totalAvailableCpu * 17) / 16 + 2) + 30)
              putComponentProperty('druid.processing.numMergeBuffers', numMergeBuffers)
              totalAvailableMemInMb = self.getMinMemory(component_hosts) / 1024
              maxAvailableBufferSizeInMb = totalAvailableMemInMb/(processingThreads + numMergeBuffers)
              putComponentProperty('druid.processing.buffer.sizeBytes', self.getDruidProcessingBufferSizeInMb(maxAvailableBufferSizeInMb) * 1024 * 1024)


  # returns the recommended druid processing buffer size in Mb.
  # the recommended buffer size is kept lower then the max available memory to have enough free memory to load druid data.
  # for low memory nodes, the actual allocated buffer size is small to keep some free memory for memory mapping of segments
  # If user installs all druid processes on a single node, memory available for loading segments will be further decreased.
  def getDruidProcessingBufferSizeInMb(self, maxAvailableBufferSizeInMb):
      if maxAvailableBufferSizeInMb <= 256:
          return min(64, maxAvailableBufferSizeInMb)
      elif maxAvailableBufferSizeInMb <= 1024:
          return 128
      elif maxAvailableBufferSizeInMb <= 2048:
          return 256
      elif maxAvailableBufferSizeInMb <= 6144:
          return 512
      # High Memory nodes below
      else :
          return 1024

  def recommendSupersetConfigurations(self, configurations, clusterData, services, hosts):
      # superset is in list of services to be installed
      if 'superset' in services['configurations']:
        # Recommendations for Superset
        superset_database_type = services['configurations']["superset"]["properties"]["SUPERSET_DATABASE_TYPE"]
        putSupersetProperty = self.putProperty(configurations, "superset", services)

        if superset_database_type == "mysql":
            putSupersetProperty("SUPERSET_DATABASE_PORT", "3306")
        elif superset_database_type == "postgresql":
            putSupersetProperty("SUPERSET_DATABASE_PORT", "5432")

  def recommendYARNConfigurations(self, configurations, clusterData, services, hosts):
    super(ADH15StackAdvisor, self).recommendYARNConfigurations(configurations, clusterData, services, hosts)
    putYarnSiteProperty = self.putProperty(configurations, "yarn-site", services)
    putYarnEnvProperty = self.putProperty(configurations, "yarn-env", services)
    servicesList = [service["StackServices"]["service_name"] for service in services["services"]]
    hsi_env_poperties = self.getServicesSiteProperties(services, "hive-interactive-env")

    if "tez-site" not in services["configurations"]:
      putYarnSiteProperty('yarn.timeline-service.entity-group-fs-store.group-id-plugin-classes', '')
    else:
      putYarnSiteProperty('yarn.timeline-service.entity-group-fs-store.group-id-plugin-classes', 'org.apache.tez.dag.history.logging.ats.TimelineCachePluginImpl')

    if 'HIVE' in servicesList and "yarn-site" in services["configurations"] and "yarn.nodemanager.kill-escape.user" in \
                services["configurations"]["yarn-site"]["properties"] and 'hive-env' in services['configurations'] and \
                'hive_user' in services['configurations']['hive-env']['properties']:
      hive_user_name = services['configurations']['hive-env']['properties']['hive_user']
      old_hive_user_name = getOldValue(self, services, "hive-env", "hive_user")
      yarn_nm_kill_escape_user = services["configurations"]["yarn-site"]["properties"]["yarn.nodemanager.kill-escape.user"]
      if not hive_user_name in yarn_nm_kill_escape_user:
        if not yarn_nm_kill_escape_user or yarn_nm_kill_escape_user.strip() == "":
          yarn_nm_kill_escape_user = hive_user_name
        else:
          escape_user_names = yarn_nm_kill_escape_user.split(",")
          if old_hive_user_name in escape_user_names:
            escape_user_names.remove(old_hive_user_name)
          escape_user_names.append(hive_user_name)
          yarn_nm_kill_escape_user = ",".join(escape_user_names)

        putYarnSiteProperty("yarn.nodemanager.kill-escape.user", yarn_nm_kill_escape_user)


    if "yarn-site" in services["configurations"] and \
                    "yarn.resourcemanager.scheduler.monitor.enable" in services["configurations"]["yarn-site"]["properties"]:
      scheduler_monitor_enabled = services["configurations"]["yarn-site"]["properties"]["yarn.resourcemanager.scheduler.monitor.enable"]
      if scheduler_monitor_enabled.lower() == 'true':
        putYarnSiteProperty('yarn.scheduler.capacity.ordering-policy.priority-utilization.underutilized-preemption.enabled', "true")
      else:
        putYarnSiteProperty('yarn.scheduler.capacity.ordering-policy.priority-utilization.underutilized-preemption.enabled', "false")

    # calculate total_preemption_per_round
    total_preemption_per_round = str(round(max(float(1)/len(hosts['items']), 0.1),2))
    putYarnSiteProperty('yarn.resourcemanager.monitor.capacity.preemption.total_preemption_per_round', total_preemption_per_round)


    if 'yarn-env' in services['configurations'] and 'yarn_user' in services['configurations']['yarn-env']['properties']:
      yarn_user = services['configurations']['yarn-env']['properties']['yarn_user']
    else:
      yarn_user = 'yarn'
    if 'ranger-yarn-plugin-properties' in configurations and 'ranger-yarn-plugin-enabled' in configurations['ranger-yarn-plugin-properties']['properties']:
      ranger_yarn_plugin_enabled = (configurations['ranger-yarn-plugin-properties']['properties']['ranger-yarn-plugin-enabled'].lower() == 'Yes'.lower())
    elif 'ranger-yarn-plugin-properties' in services['configurations'] and 'ranger-yarn-plugin-enabled' in services['configurations']['ranger-yarn-plugin-properties']['properties']:
      ranger_yarn_plugin_enabled = (services['configurations']['ranger-yarn-plugin-properties']['properties']['ranger-yarn-plugin-enabled'].lower() == 'Yes'.lower())
    else:
      ranger_yarn_plugin_enabled = False

    if ranger_yarn_plugin_enabled and 'ranger-yarn-plugin-properties' in services['configurations'] and 'REPOSITORY_CONFIG_USERNAME' in services['configurations']['ranger-yarn-plugin-properties']['properties']:
      Logger.info("Setting Yarn Repo user for Ranger.")
      putRangerYarnPluginProperty = self.putProperty(configurations, "ranger-yarn-plugin-properties", services)
      putRangerYarnPluginProperty("REPOSITORY_CONFIG_USERNAME",yarn_user)
    else:
      Logger.info("Not setting Yarn Repo user for Ranger.")


    yarn_timeline_app_cache_size = None
    host_mem = None
    for host in hosts["items"]:
      host_mem = host["Hosts"]["total_mem"]
      break
    # Check if 'yarn.timeline-service.entity-group-fs-store.app-cache-size' in changed configs.
    changed_configs_has_ats_cache_size = self.isConfigPropertiesChanged(
      services, "yarn-site", ['yarn.timeline-service.entity-group-fs-store.app-cache-size'], False)
    # Check if it's : 1. 'apptimelineserver_heapsize' changed detected in changed-configurations)
    # OR 2. cluster initialization (services['changed-configurations'] should be empty in this case)
    if changed_configs_has_ats_cache_size:
      yarn_timeline_app_cache_size = self.read_yarn_apptimelineserver_cache_size(services)
    elif 0 == len(services['changed-configurations']):
      # Fetch host memory from 1st host, to be used for ATS config calculations below.
      if host_mem is not None:
        yarn_timeline_app_cache_size = self.calculate_yarn_apptimelineserver_cache_size(host_mem)
        putYarnSiteProperty('yarn.timeline-service.entity-group-fs-store.app-cache-size', yarn_timeline_app_cache_size)
        Logger.info("Updated YARN config 'yarn.timeline-service.entity-group-fs-store.app-cache-size' as : {0}, "
                    "using 'host_mem' = {1}".format(yarn_timeline_app_cache_size, host_mem))
      else:
        Logger.info("Couldn't update YARN config 'yarn.timeline-service.entity-group-fs-store.app-cache-size' as "
                    "'host_mem' read = {0}".format(host_mem))

    if yarn_timeline_app_cache_size is not None:
      # Calculation for 'ats_heapsize' is in MB.
      ats_heapsize = self.calculate_yarn_apptimelineserver_heapsize(host_mem, yarn_timeline_app_cache_size)
      putYarnEnvProperty('apptimelineserver_heapsize', ats_heapsize) # Value in MB
      Logger.info("Updated YARN config 'apptimelineserver_heapsize' as : {0}, ".format(ats_heapsize))

    # Queue 'llap' creation/removal logic (Used by Hive Interactive server and associated LLAP)
    if hsi_env_poperties and 'enable_hive_interactive' in hsi_env_poperties:
      enable_hive_interactive = hsi_env_poperties['enable_hive_interactive']
      LLAP_QUEUE_NAME = 'llap'

      # Hive Server interactive is already added or getting added
      if enable_hive_interactive == 'true':
        self.updateLlapConfigs(configurations, services, hosts, LLAP_QUEUE_NAME)
      else:  # When Hive Interactive Server is in 'off/removed' state.
        self.checkAndStopLlapQueue(services, configurations, LLAP_QUEUE_NAME)

    putYarnSiteProperty = self.putProperty(configurations, "yarn-site", services)

    timeline_plugin_classes_values = []
    timeline_plugin_classpath_values = []

    if self.__isServiceDeployed(services, "TEZ"):
      timeline_plugin_classes_values.append('org.apache.tez.dag.history.logging.ats.TimelineCachePluginImpl')

    if self.__isServiceDeployed(services, "SPARK"):
      timeline_plugin_classes_values.append('org.apache.spark.deploy.history.yarn.plugin.SparkATSPlugin')
      timeline_plugin_classpath_values.append(stack_root + "/spark/jars/*")

    putYarnSiteProperty('yarn.timeline-service.entity-group-fs-store.group-id-plugin-classes', ",".join(timeline_plugin_classes_values))
    putYarnSiteProperty('yarn.timeline-service.entity-group-fs-store.group-id-plugin-classpath', ":".join(timeline_plugin_classpath_values))

  """
  Calculate YARN config 'apptimelineserver_heapsize' in MB.
  """
  def calculate_yarn_apptimelineserver_heapsize(self, host_mem, yarn_timeline_app_cache_size):
    ats_heapsize = None
    if host_mem < 4096:
      ats_heapsize = 1024
    else:
      ats_heapsize = long(min(math.floor(host_mem/2), long(yarn_timeline_app_cache_size) * 500 + 3072))
    return ats_heapsize

  """
  Calculates for YARN config 'yarn.timeline-service.entity-group-fs-store.app-cache-size', based on YARN's NodeManager size.
  """
  def calculate_yarn_apptimelineserver_cache_size(self, host_mem):
    yarn_timeline_app_cache_size = None
    if host_mem < 4096:
      yarn_timeline_app_cache_size = 3
    elif host_mem >= 4096 and host_mem < 8192:
      yarn_timeline_app_cache_size = 7
    elif host_mem >= 8192:
      yarn_timeline_app_cache_size = 10
    Logger.info("Calculated and returning 'yarn_timeline_app_cache_size' : {0}".format(yarn_timeline_app_cache_size))
    return yarn_timeline_app_cache_size


  """
  Reads YARN config 'yarn.timeline-service.entity-group-fs-store.app-cache-size'.
  """
  def read_yarn_apptimelineserver_cache_size(self, services):
    """
    :type services dict
    :rtype str
    """
    yarn_ats_app_cache_size = None
    yarn_ats_app_cache_size_config = "yarn.timeline-service.entity-group-fs-store.app-cache-size"
    yarn_site_in_services = self.getServicesSiteProperties(services, "yarn-site")

    if yarn_site_in_services and yarn_ats_app_cache_size_config in yarn_site_in_services: