#!/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.
"""
import os
import traceback

import imp

SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
STACKS_DIR = os.path.join(SCRIPT_DIR, '../../../../../stacks/')
PARENT_FILE = os.path.join(STACKS_DIR, 'service_advisor.py')

try:
    with open(PARENT_FILE, 'rb') as fp:
        service_advisor = imp.load_module('service_advisor', fp, PARENT_FILE, ('.py', 'rb', imp.PY_SOURCE))
except Exception as e:
    traceback.print_exc()
    print "Failed to load parent"

class METRON${metron.short.version}ServiceAdvisor(service_advisor.ServiceAdvisor):

    def getServiceComponentLayoutValidations(self, services, hosts):

        componentsListList = [service["components"] for service in services["services"]]
        componentsList = [item["StackServiceComponents"] for sublist in componentsListList for item in sublist]

        metronParsersHost = self.getHosts(componentsList, "METRON_PARSERS")[0]
        metronEnrichmentMaster = self.getHosts(componentsList, "METRON_ENRICHMENT_MASTER")[0]
        metronProfilerHost = self.getHosts(componentsList, "METRON_PROFILER")[0]
        metronPcapHost = self.getHosts(componentsList, "METRON_PCAP")[0]
        metronIndexingHost = self.getHosts(componentsList, "METRON_INDEXING")[0]
        metronRESTHost = self.getHosts(componentsList, "METRON_REST")[0]
        metronManagementUIHost = self.getHosts(componentsList, "METRON_MANAGEMENT_UI")[0]
        metronAlertsUIHost = self.getHosts(componentsList, "METRON_ALERTS_UI")[0]

        hbaseClientHosts = self.getHosts(componentsList, "HBASE_CLIENT")
        hdfsClientHosts = self.getHosts(componentsList, "HDFS_CLIENT")
        zookeeperClientHosts = self.getHosts(componentsList, "ZOOKEEPER_CLIENT")

        kafkaBrokers = self.getHosts(componentsList, "KAFKA_BROKER")
        stormSupervisors = self.getHosts(componentsList, "SUPERVISOR")

        items = []

        # Metron Must Co-locate with KAFKA_BROKER and STORM_SUPERVISOR
        if metronParsersHost not in kafkaBrokers:
            message = "Metron must be colocated with an instance of KAFKA BROKER"
            items.append({ "type": 'host-component', "level": 'ERROR', "message": message, "component-name": 'METRON_PARSERS', "host": metronParsersHost })

        if metronParsersHost not in stormSupervisors:
            message = "Metron must be colocated with an instance of STORM SUPERVISOR"
            items.append({ "type": 'host-component', "level": 'WARN', "message": message, "component-name": 'METRON_PARSERS', "host": metronParsersHost })

        if metronRESTHost not in stormSupervisors:
            message = "Metron REST must be colocated with an instance of STORM SUPERVISOR"
            items.append({ "type": 'host-component', "level": 'WARN', "message": message, "component-name": 'METRON_REST', "host": metronRESTHost })

        if metronParsersHost !=  metronRESTHost:
            message = "Metron REST must be co-located with Metron Parsers on {0}".format(metronParsersHost)
            items.append({ "type": 'host-component', "level": 'WARN', "message": message, "component-name": 'METRON_REST', "host": metronRESTHost })

        if metronParsersHost != metronEnrichmentMaster:
            message = "Metron Enrichment Master must be co-located with Metron Parsers on {0}".format(metronParsersHost)
            items.append({ "type": 'host-component', "level": 'ERROR', "message": message, "component-name": 'METRON_ENRICHMENT_MASTER', "host": metronEnrichmentMaster })

        if metronParsersHost != metronIndexingHost:
            message = "Metron Indexing must be co-located with Metron Parsers on {0}".format(metronParsersHost)
            items.append({ "type": 'host-component', "level": 'ERROR', "message": message, "component-name": 'METRON_INDEXING', "host": metronIndexingHost })

        if metronParsersHost != metronPcapHost:
            message = "Metron PCAP must be co-located with Metron Parsers on {0}".format(metronParsersHost)
            items.append({ "type": 'host-component', "level": 'ERROR', "message": message, "component-name": 'METRON_PCAP', "host": metronPcapHost })

        if metronParsersHost != metronProfilerHost:
            message = "Metron Profiler must be co-located with Metron Parsers on {0}".format(metronParsersHost)
            items.append({ "type": 'host-component', "level": 'ERROR', "message": message, "component-name": 'METRON_PROFILER', "host": metronProfilerHost })

        # Enrichment Master also needs ZK Client, but this is already guaranteed by being colocated with Parsers Master
        if metronParsersHost not in zookeeperClientHosts:
            message = "Metron must be co-located with an instance of Zookeeper Client"
            items.append({ "type": 'host-component', "level": 'WARN', "message": message, "component-name": 'METRON_PARSERS', "host": metronParsersHost })

            # Enrichment Master also needs HDFS clients, but this is already guaranteed by being colocated with Parsers Master
        if metronParsersHost not in hdfsClientHosts:
            message = "Metron must be co-located with an instance of HDFS Client"
            items.append({ "type": 'host-component', "level": 'WARN', "message": message, "component-name": 'METRON_PARSERS', "host": metronParsersHost })

        if metronEnrichmentMaster not in hbaseClientHosts:
            message = "Metron Enrichment Master must be co-located with an instance of HBase Client"
            items.append({ "type": 'host-component', "level": 'WARN', "message": message, "component-name": 'METRON_ENRICHMENT_MASTER', "host": metronEnrichmentMaster })

        if metronManagementUIHost != metronAlertsUIHost:
            message = "Metron Alerts UI must be co-located with Metron Management UI on {0}".format(metronManagementUIHost)
            items.append({ "type": 'host-component', "level": 'ERROR', "message": message, "component-name": 'METRON_ALERTS_UI', "host": metronAlertsUIHost })

        return items

    def getServiceConfigurationsValidationItems(self, configurations, recommendedDefaults, services, hosts):

        # validate recommended properties in storm-site
        siteName = "storm-site"
        method = self.validateSTORMSiteConfigurations
        items = self.validateConfigurationsForSite(configurations, recommendedDefaults, services, hosts, siteName, method)

        return items

    def getServiceConfigurationRecommendations(self, configurations, clusterData, services, hosts):
        is_secured = self.isSecurityEnabled(services)

        #Suggest Storm Rest URL
        if "storm-site" in services["configurations"]:
            stormUIServerHost = self.getComponentHostNames(services, "STORM", "STORM_UI_SERVER")[0]
            stormUIServerPort = services["configurations"]["storm-site"]["properties"]["ui.port"]
            stormUIProtocol = "http://"
            if "ui.https.port" in services["configurations"]["storm-site"]["properties"]:
                stormUIServerPort = services["configurations"]["storm-site"]["properties"]["ui.https.port"]
                stormUIProtocol = "https://"
            stormUIServerURL = stormUIProtocol + stormUIServerHost + ":" + stormUIServerPort
            putMetronEnvProperty = self.putProperty(configurations, "metron-env", services)
            putMetronEnvProperty("storm_rest_addr",stormUIServerURL)

            storm_site = services["configurations"]["storm-site"]["properties"]
            putStormSiteProperty = self.putProperty(configurations, "storm-site", services)

            for property, desired_value in self.getSTORMSiteDesiredValues(is_secured).iteritems():
                if property not in storm_site:
                    putStormSiteProperty(property, desired_value)
                elif  property == "topology.classpath" and storm_site[property] != desired_value:
                    topologyClasspath = storm_site[property]
                    #check that desired values exist in topology.classpath. append them if they do not
                    for path in desired_value.split(':'):
                        if path not in topologyClasspath:
                            topologyClasspath += ":" + path
                    putStormSiteProperty(property,topologyClasspath)

        #Suggest Zeppelin Server URL
        if "zeppelin-config" in services["configurations"]:
            zeppelinServerHost = self.getComponentHostNames(services, "ZEPPELIN", "ZEPPELIN_MASTER")[0]
            zeppelinServerPort = services["configurations"]["zeppelin-config"]["properties"]["zeppelin.server.port"]
            zeppelinServerUrl = zeppelinServerHost + ":" + zeppelinServerPort
            putMetronEnvProperty = self.putProperty(configurations, "metron-env", services)
            putMetronEnvProperty("zeppelin_server_url", zeppelinServerUrl)

        #Suggest Zookeeper quorum
        if "solr-cloud" in services["configurations"]:
            zookeeperHost = self.getComponentHostNames(services, "ZOOKEEPER", "ZOOKEEPER_SERVER")[0]
            zookeeperClientPort = services["configurations"]["zoo.cfg"]["properties"]["clientPort"]
            solrZkDir = services["configurations"]["solr-cloud"]["properties"]["solr_cloud_zk_directory"]
            solrZookeeperUrl = zookeeperHost + ":" + zookeeperClientPort + solrZkDir
            putMetronEnvProperty = self.putProperty(configurations, "metron-env", services)
            putMetronEnvProperty("solr_zookeeper_url", solrZookeeperUrl)


    def validateSTORMSiteConfigurations(self, properties, recommendedDefaults, configurations, services, hosts):
        # Determine if the cluster is secured
        is_secured = self.isSecurityEnabled(services)

        storm_site = properties
        validationItems = []

        for property, desired_value in self.getSTORMSiteDesiredValues(is_secured).iteritems():
            if property not in storm_site :
                message = "Metron requires this property to be set to the recommended value of " + desired_value
                item = self.getErrorItem(message) if property == "topology.classpath" else self.getWarnItem(message)
                validationItems.append({"config-name": property, "item": item})
            elif  storm_site[property] != desired_value:
                topologyClasspath = storm_site[property]
                for path in desired_value.split(':'):
                    if path not in topologyClasspath:
                        message = "Metron requires this property to contain " + desired_value
                        item = self.getErrorItem(message)
                        validationItems.append({"config-name": property, "item": item})

        return self.toConfigurationValidationProblems(validationItems, "storm-site")

    def getSTORMSiteDesiredValues(self, is_secured):

        storm_site_desired_values = {
            "topology.classpath" : "/etc/hbase/conf:/etc/hadoop/conf"
        }

        return storm_site_desired_values
