/*
 * 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
 * <p/>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p/>
 * 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.
 */
package org.apache.ambari.server.metrics.system.impl;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;

import org.apache.ambari.server.configuration.ComponentSSLConfiguration;
import org.apache.ambari.server.controller.AmbariManagementController;
import org.apache.ambari.server.controller.internal.AbstractControllerResourceProvider;
import org.apache.ambari.server.controller.internal.ServiceConfigVersionResourceProvider;
import org.apache.ambari.server.controller.spi.Predicate;
import org.apache.ambari.server.controller.spi.Request;
import org.apache.ambari.server.controller.spi.Resource;
import org.apache.ambari.server.controller.spi.ResourceProvider;
import org.apache.ambari.server.controller.utilities.PredicateBuilder;
import org.apache.ambari.server.controller.utilities.PropertyHelper;
import org.apache.ambari.server.metrics.system.MetricsSink;
import org.apache.ambari.server.metrics.system.SingleMetric;
import org.apache.ambari.server.security.authorization.internal.InternalAuthenticationToken;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.Config;
import org.apache.ambari.server.state.ConfigHelper;
import org.apache.ambari.server.state.Service;
import org.apache.ambari.server.state.ServiceComponent;
import org.apache.ambari.server.state.ServiceComponentHost;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.metrics2.sink.timeline.AbstractTimelineMetricsSink;
import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
import org.apache.hadoop.metrics2.sink.timeline.cache.TimelineMetricsCache;
import org.springframework.security.core.context.SecurityContextHolder;

/**
 * Ambari Server Metrics Sink implementation to push collected metrics to AMS.
 */
public class AmbariMetricSinkImpl extends AbstractTimelineMetricsSink implements MetricsSink {
  private static final String AMBARI_SERVER_APP_ID = "ambari_server";
  private Collection<String> collectorHosts;

  private String collectorUri;
  private String port;
  private String protocol;
  private String hostName;
  private AmbariManagementController ambariManagementController;
  private TimelineMetricsCache timelineMetricsCache;
  private boolean isInitialized = false;
  private boolean setInstanceId = false;
  private String instanceId;

  public AmbariMetricSinkImpl(AmbariManagementController amc) {
    this.ambariManagementController = amc;
  }

  @Override
  public void init(MetricsConfiguration configuration) {

    if (ambariManagementController == null) {
      return;
    }

    InternalAuthenticationToken authenticationToken = new InternalAuthenticationToken("admin");
    authenticationToken.setAuthenticated(true);
    SecurityContextHolder.getContext().setAuthentication(authenticationToken);
    Clusters clusters = ambariManagementController.getClusters();

    if (clusters == null || clusters.getClusters().isEmpty()) {
      LOG.info("No clusters configured.");
      return;
    }

    String ambariMetricsServiceName = "AMBARI_METRICS";
    collectorHosts = new HashSet<>();

    for (Map.Entry<String, Cluster> kv : clusters.getClusters().entrySet()) {
      String clusterName = kv.getKey();
      instanceId = clusterName;
      Cluster c = kv.getValue();
      Resource.Type type = Resource.Type.ServiceConfigVersion;

      //If Metrics Collector VIP settings are configured, use that.
      boolean externalHostConfigPresent = false;
      boolean externalPortConfigPresent = false;
      Config clusterEnv = c.getDesiredConfigByType(ConfigHelper.CLUSTER_ENV);
      if (clusterEnv != null) {
        Map<String, String> configs = clusterEnv.getProperties();

        String metricsCollectorExternalHosts = configs.get("metrics_collector_external_hosts");
        if (StringUtils.isNotEmpty(metricsCollectorExternalHosts)) {
          LOG.info("Setting Metrics Collector External Host : " + metricsCollectorExternalHosts);
          collectorHosts.addAll(Arrays.asList(metricsCollectorExternalHosts.split(",")));
          externalHostConfigPresent = true;
          setInstanceId = true;
        }

        String metricsCollectorExternalPort = configs.get("metrics_collector_external_port");
        if (StringUtils.isNotEmpty(metricsCollectorExternalPort)) {
          LOG.info("Setting Metrics Collector External Port : " + metricsCollectorExternalPort);
          port = metricsCollectorExternalPort;
          externalPortConfigPresent = true;
        }
      }

      Set<String> propertyIds = new HashSet<>();
      propertyIds.add(ServiceConfigVersionResourceProvider.CONFIGURATIONS_PROPERTY_ID);

      Predicate predicate = new PredicateBuilder().property(
        ServiceConfigVersionResourceProvider.CLUSTER_NAME_PROPERTY_ID).equals(clusterName).and().property(
        ServiceConfigVersionResourceProvider.SERVICE_NAME_PROPERTY_ID).equals(ambariMetricsServiceName).and().property(
        ServiceConfigVersionResourceProvider.IS_CURRENT_PROPERTY_ID).equals("true").toPredicate();

      Request request = PropertyHelper.getReadRequest(propertyIds);

      ResourceProvider provider = AbstractControllerResourceProvider.getResourceProvider(
        type,
        ambariManagementController);

      try {
        if ( !externalHostConfigPresent ) {
          //get collector host(s)
          Service service = c.getService(ambariMetricsServiceName);
          if (service != null) {
            for (String component : service.getServiceComponents().keySet()) {
              ServiceComponent sc = service.getServiceComponents().get(component);
              for (ServiceComponentHost serviceComponentHost : sc.getServiceComponentHosts().values()) {
                if (serviceComponentHost.getServiceComponentName().equals("METRICS_COLLECTOR")) {
                  collectorHosts.add(serviceComponentHost.getHostName());
                }
              }
            }
          }
        }

        // get collector port and protocol
        Set<Resource> resources = provider.getResources(request, predicate);

        for (Resource resource : resources) {
          if (resource != null) {
            ArrayList<LinkedHashMap<Object, Object>> configs = (ArrayList<LinkedHashMap<Object, Object>>)
              resource.getPropertyValue(ServiceConfigVersionResourceProvider.CONFIGURATIONS_PROPERTY_ID);
            for (LinkedHashMap<Object, Object> config : configs) {
              if (config != null && config.get("type").equals("ams-site")) {
                TreeMap<Object, Object> properties = (TreeMap<Object, Object>) config.get("properties");
                String timelineWebappAddress = (String) properties.get("timeline.metrics.service.webapp.address");
                if (!externalPortConfigPresent && StringUtils.isNotEmpty(timelineWebappAddress) && timelineWebappAddress.contains(":")) {
                  port = timelineWebappAddress.split(":")[1];
                }
                String httpPolicy = (String) properties.get("timeline.metrics.service.http.policy");
                protocol = httpPolicy.equals("HTTP_ONLY") ? "http" : "https";
                break;
              }
            }
          }
        }
      } catch (Exception e) {
        LOG.info("Exception caught when retrieving Collector URI", e);
      }
    }

    hostName = configuration.getProperty("ambariserver.hostname.override", getDefaultLocalHostName());
    LOG.info("Hostname used for ambari server metrics : " + hostName);

    if (protocol.contains("https")) {
      ComponentSSLConfiguration sslConfiguration = ComponentSSLConfiguration.instance();
      String trustStorePath = sslConfiguration.getTruststorePath();
      String trustStoreType = sslConfiguration.getTruststoreType();
      String trustStorePwd = sslConfiguration.getTruststorePassword();
      loadTruststore(trustStorePath, trustStoreType, trustStorePwd);
    }

    collectorUri = getCollectorUri(findPreferredCollectHost());

    int maxRowCacheSize = Integer.parseInt(configuration.getProperty(MAX_METRIC_ROW_CACHE_SIZE,
      String.valueOf(TimelineMetricsCache.MAX_RECS_PER_NAME_DEFAULT)));
    int metricsSendInterval = Integer.parseInt(configuration.getProperty(METRICS_SEND_INTERVAL,
      String.valueOf(TimelineMetricsCache.MAX_EVICTION_TIME_MILLIS)));

    timelineMetricsCache = new TimelineMetricsCache(maxRowCacheSize, metricsSendInterval);

    if (CollectionUtils.isNotEmpty(collectorHosts)) {
      LOG.info("Metric Sink initialized with collectorHosts : " + collectorHosts.toString());
      isInitialized = true;
    }
  }

  private String getDefaultLocalHostName() {
    try {
      return InetAddress.getLocalHost().getCanonicalHostName();
    } catch (UnknownHostException e) {
      LOG.info("Error getting host address");
    }
    return null;
  }

  /**
   * Publish metrics to AMS.
   * @param metrics Set of metrics
   */
  @Override
  public void publish(List<SingleMetric> metrics) {

    //If Sink not yet initialized, drop the metrics on the floor.
    if (isInitialized) {
      List<TimelineMetric> metricList = getFilteredMetricList(metrics);
      if (!metricList.isEmpty()) {
        TimelineMetrics timelineMetrics = new TimelineMetrics();
        timelineMetrics.setMetrics(metricList);
        emitMetrics(timelineMetrics);
      }
    } else {
      LOG.debug("Metric Sink not yet initialized. Discarding metrics.");
    }
  }

  @Override
  public boolean isInitialized() {
    return isInitialized;
  }


  /**
   * Get a pre-formatted URI for the collector
   *
   * @param host
   */
  @Override
  protected String getCollectorUri(String host) {
    return constructTimelineMetricUri(protocol, host, port);
  }

  @Override
  protected String getCollectorProtocol() {
    return protocol;
  }

  @Override
  protected String getCollectorPort() {
    return port;
  }

  @Override
  protected int getTimeoutSeconds() {
    return 10;
  }

  /**
   * Get the zookeeper quorum for the cluster used to find collector
   *
   * @return String "host1:port1,host2:port2"
   */
  @Override
  protected String getZookeeperQuorum() {
    //Ignoring Zk Fallback.
    return null;
  }

  /**
   * Get pre-configured list of collectors available
   *
   * @return Collection<String> host1,host2
   */
  @Override
  protected Collection<String> getConfiguredCollectorHosts() {
    return collectorHosts;
  }

  /**
   * Get hostname used for calculating write shard.
   *
   * @return String "host1"
   */
  @Override
  protected String getHostname() {
    return hostName;
  }

  @Override
  protected boolean isHostInMemoryAggregationEnabled() {
    return false;
  }

  @Override
  protected int getHostInMemoryAggregationPort() {
    return 0;
  }

  @Override
  protected String getHostInMemoryAggregationProtocol() {
    return "http";
  }

  private List<TimelineMetric> getFilteredMetricList(List<SingleMetric> metrics) {
    final List<TimelineMetric> metricList = new ArrayList<>();
    for (SingleMetric metric : metrics) {

      String metricName = metric.getMetricName();
      Double value = metric.getValue();

      TimelineMetric timelineMetric = createTimelineMetric(metric.getTimestamp(), AMBARI_SERVER_APP_ID, metricName, value);
      timelineMetricsCache.putTimelineMetric(timelineMetric, false);
      TimelineMetric cachedMetric = timelineMetricsCache.getTimelineMetric(metricName);

      if (cachedMetric != null) {
        metricList.add(cachedMetric);
      }
    }
    return metricList;
  }

  private TimelineMetric createTimelineMetric(long currentTimeMillis, String component, String attributeName,
                                              Number attributeValue) {
    TimelineMetric timelineMetric = new TimelineMetric();
    timelineMetric.setMetricName(attributeName);
    timelineMetric.setHostName(hostName);
    if (setInstanceId) {
      timelineMetric.setInstanceId(instanceId);
    }
    timelineMetric.setAppId(component);
    timelineMetric.setStartTime(currentTimeMillis);

    timelineMetric.getMetricValues().put(currentTimeMillis, attributeValue.doubleValue());
    return timelineMetric;
  }
}
