@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;
    }
  }
Ambari Server Metrics Sink implementation to push collected metrics to AMS.
ambari-server/src/main/java/org/apache/ambari/server/metrics/system/impl/AmbariMetricSinkImpl.java