ambari-metrics-flume-sink/src/main/java/org/apache/hadoop/metrics2/sink/flume/FlumeTimelineMetricsSink.java - ambari-metrics - Git at Google

 /**
  * 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.
  */

 package org.apache.hadoop.metrics2.sink.flume;

 import org.apache.commons.lang.math.NumberUtils;
 import org.apache.flume.Context;
 import org.apache.flume.FlumeException;
 import org.apache.flume.instrumentation.MonitorService;
 import org.apache.flume.instrumentation.util.JMXPollUtil;
 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.UnableToConnectException;
 import org.apache.hadoop.metrics2.sink.timeline.cache.TimelineMetricsCache;
 import org.apache.hadoop.metrics2.sink.timeline.configuration.Configuration;

 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;

 public class FlumeTimelineMetricsSink extends AbstractTimelineMetricsSink implements MonitorService {
   private String collectorUri;
   private String protocol;
   // Key - component(instance_id)
   private Map<String, TimelineMetricsCache> metricsCaches;
   private int maxRowCacheSize;
   private int metricsSendInterval;
   private ScheduledExecutorService scheduledExecutorService;
   private long pollFrequency;
   private String hostname;
   private String port;
   private Collection<String> collectorHosts;
   private String zookeeperQuorum;
   private final static String COUNTER_METRICS_PROPERTY = "counters";
   private final Set<String> counterMetrics = new HashSet<String>();
   private int timeoutSeconds = 10;
   private boolean setInstanceId;
   private String instanceId;
   private boolean hostInMemoryAggregationEnabled;
   private int hostInMemoryAggregationPort;
   private String hostInMemoryAggregationProtocol;


   @Override
   public void start() {
     LOG.info("Starting Flume Metrics Sink");
     TimelineMetricsCollector timelineMetricsCollector = new TimelineMetricsCollector();
     if (scheduledExecutorService == null || scheduledExecutorService.isShutdown() || scheduledExecutorService.isTerminated()) {
       scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
     }
     scheduledExecutorService.scheduleWithFixedDelay(timelineMetricsCollector, 0,
         pollFrequency, TimeUnit.MILLISECONDS);
   }

   @Override
   public void stop() {
     LOG.info("Stopping Flume Metrics Sink");
     scheduledExecutorService.shutdown();
   }

   @Override
   public void configure(Context context) {
     LOG.info("Context parameters " + context);
     try {
       hostname = InetAddress.getLocalHost().getHostName();
       //If not FQDN , call  DNS
       if ((hostname == null) || (!hostname.contains("."))) {
         hostname = InetAddress.getLocalHost().getCanonicalHostName();
       }
       hostname = hostname.toLowerCase();

     } catch (UnknownHostException e) {
       LOG.error("Could not identify hostname.");
       throw new FlumeException("Could not identify hostname.", e);
     }
     Configuration configuration = new Configuration("/flume-metrics2.properties");
     timeoutSeconds = Integer.parseInt(configuration.getProperty(METRICS_POST_TIMEOUT_SECONDS,
         String.valueOf(DEFAULT_POST_TIMEOUT_SECONDS)));
     maxRowCacheSize = Integer.parseInt(configuration.getProperty(MAX_METRIC_ROW_CACHE_SIZE,
         String.valueOf(TimelineMetricsCache.MAX_RECS_PER_NAME_DEFAULT)));
     metricsSendInterval = Integer.parseInt(configuration.getProperty(METRICS_SEND_INTERVAL,
         String.valueOf(TimelineMetricsCache.MAX_EVICTION_TIME_MILLIS)));
     metricsCaches = new HashMap<String, TimelineMetricsCache>();
     collectorHosts = parseHostsStringIntoCollection(configuration.getProperty(COLLECTOR_HOSTS_PROPERTY));
     zookeeperQuorum = configuration.getProperty("zookeeper.quorum");
     protocol = configuration.getProperty(COLLECTOR_PROTOCOL, "http");
     port = configuration.getProperty(COLLECTOR_PORT, "6188");
     setInstanceId = Boolean.valueOf(configuration.getProperty(SET_INSTANCE_ID_PROPERTY, "false"));
     instanceId = configuration.getProperty(INSTANCE_ID_PROPERTY, "");

     hostInMemoryAggregationEnabled = Boolean.getBoolean(configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_ENABLED_PROPERTY, "false"));
     hostInMemoryAggregationPort = Integer.valueOf(configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_PORT_PROPERTY, "61888"));
     hostInMemoryAggregationProtocol = configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_PROTOCOL_PROPERTY, "http");
     // Initialize the collector write strategy
     super.init();

     if (protocol.contains("https") || hostInMemoryAggregationProtocol.contains("https")) {
       String trustStorePath = configuration.getProperty(SSL_KEYSTORE_PATH_PROPERTY).trim();
       String trustStoreType = configuration.getProperty(SSL_KEYSTORE_TYPE_PROPERTY).trim();
       String trustStorePwd = configuration.getProperty(SSL_KEYSTORE_PASSWORD_PROPERTY).trim();
       loadTruststore(trustStorePath, trustStoreType, trustStorePwd);
     }
     collectorUri = constructTimelineMetricUri(protocol, findPreferredCollectHost(), port);

     pollFrequency = Long.parseLong(configuration.getProperty("collectionFrequency"));

     String[] metrics = configuration.getProperty(COUNTER_METRICS_PROPERTY).trim().split(",");
     Collections.addAll(counterMetrics, metrics);
   }

   @Override
   public 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 timeoutSeconds;
   }

   @Override
   protected String getZookeeperQuorum() {
     return zookeeperQuorum;
   }

   @Override
   protected Collection<String> getConfiguredCollectorHosts() {
     return collectorHosts;
   }

   @Override
   protected String getHostname() {
     return hostname;
   }

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

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

   @Override
   protected String getHostInMemoryAggregationProtocol() {
     return hostInMemoryAggregationProtocol;
   }

   public void setPollFrequency(long pollFrequency) {
     this.pollFrequency = pollFrequency;
   }

   //Test hepler method
   protected void setMetricsCaches(Map<String, TimelineMetricsCache> metricsCaches) {
     this.metricsCaches = metricsCaches;
   }

   /**
    * Worker which polls JMX for all mbeans with
    * {@link javax.management.ObjectName} within the flume namespace:
    * org.apache.flume. All attributes of such beans are sent
    * to the metrics collector service.
    */
   class TimelineMetricsCollector implements Runnable {
     @Override
     public void run() {
       LOG.debug("Collecting Metrics for Flume");
       try {
         Map<String, Map<String, String>> metricsMap = JMXPollUtil.getAllMBeans();
         long currentTimeMillis = System.currentTimeMillis();
         for (String component : metricsMap.keySet()) {
           Map<String, String> attributeMap = metricsMap.get(component);
           LOG.debug("Attributes for component " + component);
           processComponentAttributes(currentTimeMillis, component, attributeMap);
         }
       } catch (UnableToConnectException uce) {
         LOG.warn("Unable to send metrics to collector by address:" + uce.getConnectUrl());
       } catch (Exception e) {
         LOG.error("Unexpected error", e);
       }
       LOG.debug("Finished collecting Metrics for Flume");
     }

     private void processComponentAttributes(long currentTimeMillis, String component, Map<String, String> attributeMap) throws IOException {
       List<TimelineMetric> metricList = new ArrayList<TimelineMetric>();
       if (!metricsCaches.containsKey(component)) {
         metricsCaches.put(component, new TimelineMetricsCache(maxRowCacheSize, metricsSendInterval));
       }
       TimelineMetricsCache metricsCache = metricsCaches.get(component);
       for (String attributeName : attributeMap.keySet()) {
         String attributeValue = attributeMap.get(attributeName);
         if (NumberUtils.isNumber(attributeValue)) {
           LOG.info(attributeName + " = " + attributeValue);
           TimelineMetric timelineMetric = createTimelineMetric(currentTimeMillis,
               component, attributeName, attributeValue);
           // Put intermediate values into the cache until it is time to send
           metricsCache.putTimelineMetric(timelineMetric, isCounterMetric(attributeName));

           TimelineMetric cachedMetric = metricsCache.getTimelineMetric(attributeName);

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

       if (!metricList.isEmpty()) {
         TimelineMetrics timelineMetrics = new TimelineMetrics();
         timelineMetrics.setMetrics(metricList);
         emitMetrics(timelineMetrics);
       }
     }

     private TimelineMetric createTimelineMetric(long currentTimeMillis, String component, String attributeName, String attributeValue) {
       TimelineMetric timelineMetric = new TimelineMetric();
       timelineMetric.setMetricName(attributeName);
       timelineMetric.setHostName(hostname);
       if (setInstanceId) {
         timelineMetric.setInstanceId(instanceId + component);
       } else {
         timelineMetric.setInstanceId(component);
       }
       timelineMetric.setAppId("FLUME_HANDLER");
       timelineMetric.setStartTime(currentTimeMillis);
       timelineMetric.getMetricValues().put(currentTimeMillis, Double.parseDouble(attributeValue));
       return timelineMetric;
     }
   }

   private boolean isCounterMetric(String attributeName) {
     return counterMetrics.contains(attributeName);
   }
 }
	/**
	* 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.
	*/

	package org.apache.hadoop.metrics2.sink.flume;

	import org.apache.commons.lang.math.NumberUtils;
	import org.apache.flume.Context;
	import org.apache.flume.FlumeException;
	import org.apache.flume.instrumentation.MonitorService;
	import org.apache.flume.instrumentation.util.JMXPollUtil;
	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.UnableToConnectException;
	import org.apache.hadoop.metrics2.sink.timeline.cache.TimelineMetricsCache;
	import org.apache.hadoop.metrics2.sink.timeline.configuration.Configuration;

	import java.io.IOException;
	import java.net.InetAddress;
	import java.net.UnknownHostException;
	import java.util.ArrayList;
	import java.util.Collection;
	import java.util.Collections;
	import java.util.HashMap;
	import java.util.HashSet;
	import java.util.List;
	import java.util.Map;
	import java.util.Set;
	import java.util.concurrent.Executors;
	import java.util.concurrent.ScheduledExecutorService;
	import java.util.concurrent.TimeUnit;

	public class FlumeTimelineMetricsSink extends AbstractTimelineMetricsSink implements MonitorService {
	private String collectorUri;
	private String protocol;
	// Key - component(instance_id)
	private Map<String, TimelineMetricsCache> metricsCaches;
	private int maxRowCacheSize;
	private int metricsSendInterval;
	private ScheduledExecutorService scheduledExecutorService;
	private long pollFrequency;
	private String hostname;
	private String port;
	private Collection<String> collectorHosts;
	private String zookeeperQuorum;
	private final static String COUNTER_METRICS_PROPERTY = "counters";
	private final Set<String> counterMetrics = new HashSet<String>();
	private int timeoutSeconds = 10;
	private boolean setInstanceId;
	private String instanceId;
	private boolean hostInMemoryAggregationEnabled;
	private int hostInMemoryAggregationPort;
	private String hostInMemoryAggregationProtocol;


	@Override
	public void start() {
	LOG.info("Starting Flume Metrics Sink");
	TimelineMetricsCollector timelineMetricsCollector = new TimelineMetricsCollector();
	if (scheduledExecutorService == null \|\| scheduledExecutorService.isShutdown() \|\| scheduledExecutorService.isTerminated()) {
	scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
	}
	scheduledExecutorService.scheduleWithFixedDelay(timelineMetricsCollector, 0,
	pollFrequency, TimeUnit.MILLISECONDS);
	}

	@Override
	public void stop() {
	LOG.info("Stopping Flume Metrics Sink");
	scheduledExecutorService.shutdown();
	}

	@Override
	public void configure(Context context) {
	LOG.info("Context parameters " + context);
	try {
	hostname = InetAddress.getLocalHost().getHostName();
	//If not FQDN , call DNS
	if ((hostname == null) \|\| (!hostname.contains("."))) {
	hostname = InetAddress.getLocalHost().getCanonicalHostName();
	}
	hostname = hostname.toLowerCase();

	} catch (UnknownHostException e) {
	LOG.error("Could not identify hostname.");
	throw new FlumeException("Could not identify hostname.", e);
	}
	Configuration configuration = new Configuration("/flume-metrics2.properties");
	timeoutSeconds = Integer.parseInt(configuration.getProperty(METRICS_POST_TIMEOUT_SECONDS,
	String.valueOf(DEFAULT_POST_TIMEOUT_SECONDS)));
	maxRowCacheSize = Integer.parseInt(configuration.getProperty(MAX_METRIC_ROW_CACHE_SIZE,
	String.valueOf(TimelineMetricsCache.MAX_RECS_PER_NAME_DEFAULT)));
	metricsSendInterval = Integer.parseInt(configuration.getProperty(METRICS_SEND_INTERVAL,
	String.valueOf(TimelineMetricsCache.MAX_EVICTION_TIME_MILLIS)));
	metricsCaches = new HashMap<String, TimelineMetricsCache>();
	collectorHosts = parseHostsStringIntoCollection(configuration.getProperty(COLLECTOR_HOSTS_PROPERTY));
	zookeeperQuorum = configuration.getProperty("zookeeper.quorum");
	protocol = configuration.getProperty(COLLECTOR_PROTOCOL, "http");
	port = configuration.getProperty(COLLECTOR_PORT, "6188");
	setInstanceId = Boolean.valueOf(configuration.getProperty(SET_INSTANCE_ID_PROPERTY, "false"));
	instanceId = configuration.getProperty(INSTANCE_ID_PROPERTY, "");

	hostInMemoryAggregationEnabled = Boolean.getBoolean(configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_ENABLED_PROPERTY, "false"));
	hostInMemoryAggregationPort = Integer.valueOf(configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_PORT_PROPERTY, "61888"));
	hostInMemoryAggregationProtocol = configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_PROTOCOL_PROPERTY, "http");
	// Initialize the collector write strategy
	super.init();

	if (protocol.contains("https") \|\| hostInMemoryAggregationProtocol.contains("https")) {
	String trustStorePath = configuration.getProperty(SSL_KEYSTORE_PATH_PROPERTY).trim();
	String trustStoreType = configuration.getProperty(SSL_KEYSTORE_TYPE_PROPERTY).trim();
	String trustStorePwd = configuration.getProperty(SSL_KEYSTORE_PASSWORD_PROPERTY).trim();
	loadTruststore(trustStorePath, trustStoreType, trustStorePwd);
	}
	collectorUri = constructTimelineMetricUri(protocol, findPreferredCollectHost(), port);

	pollFrequency = Long.parseLong(configuration.getProperty("collectionFrequency"));

	String[] metrics = configuration.getProperty(COUNTER_METRICS_PROPERTY).trim().split(",");
	Collections.addAll(counterMetrics, metrics);
	}

	@Override
	public 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 timeoutSeconds;
	}

	@Override
	protected String getZookeeperQuorum() {
	return zookeeperQuorum;
	}

	@Override
	protected Collection<String> getConfiguredCollectorHosts() {
	return collectorHosts;
	}

	@Override
	protected String getHostname() {
	return hostname;
	}

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

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

	@Override
	protected String getHostInMemoryAggregationProtocol() {
	return hostInMemoryAggregationProtocol;
	}

	public void setPollFrequency(long pollFrequency) {
	this.pollFrequency = pollFrequency;
	}

	//Test hepler method
	protected void setMetricsCaches(Map<String, TimelineMetricsCache> metricsCaches) {
	this.metricsCaches = metricsCaches;
	}

	/**
	* Worker which polls JMX for all mbeans with
	* {@link javax.management.ObjectName} within the flume namespace:
	* org.apache.flume. All attributes of such beans are sent
	* to the metrics collector service.
	*/
	class TimelineMetricsCollector implements Runnable {
	@Override
	public void run() {
	LOG.debug("Collecting Metrics for Flume");
	try {
	Map<String, Map<String, String>> metricsMap = JMXPollUtil.getAllMBeans();
	long currentTimeMillis = System.currentTimeMillis();
	for (String component : metricsMap.keySet()) {
	Map<String, String> attributeMap = metricsMap.get(component);
	LOG.debug("Attributes for component " + component);
	processComponentAttributes(currentTimeMillis, component, attributeMap);
	}
	} catch (UnableToConnectException uce) {
	LOG.warn("Unable to send metrics to collector by address:" + uce.getConnectUrl());
	} catch (Exception e) {
	LOG.error("Unexpected error", e);
	}
	LOG.debug("Finished collecting Metrics for Flume");
	}

	private void processComponentAttributes(long currentTimeMillis, String component, Map<String, String> attributeMap) throws IOException {
	List<TimelineMetric> metricList = new ArrayList<TimelineMetric>();
	if (!metricsCaches.containsKey(component)) {
	metricsCaches.put(component, new TimelineMetricsCache(maxRowCacheSize, metricsSendInterval));
	}
	TimelineMetricsCache metricsCache = metricsCaches.get(component);
	for (String attributeName : attributeMap.keySet()) {
	String attributeValue = attributeMap.get(attributeName);
	if (NumberUtils.isNumber(attributeValue)) {
	LOG.info(attributeName + " = " + attributeValue);
	TimelineMetric timelineMetric = createTimelineMetric(currentTimeMillis,
	component, attributeName, attributeValue);
	// Put intermediate values into the cache until it is time to send
	metricsCache.putTimelineMetric(timelineMetric, isCounterMetric(attributeName));

	TimelineMetric cachedMetric = metricsCache.getTimelineMetric(attributeName);

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

	if (!metricList.isEmpty()) {
	TimelineMetrics timelineMetrics = new TimelineMetrics();
	timelineMetrics.setMetrics(metricList);
	emitMetrics(timelineMetrics);
	}
	}

	private TimelineMetric createTimelineMetric(long currentTimeMillis, String component, String attributeName, String attributeValue) {
	TimelineMetric timelineMetric = new TimelineMetric();
	timelineMetric.setMetricName(attributeName);
	timelineMetric.setHostName(hostname);
	if (setInstanceId) {
	timelineMetric.setInstanceId(instanceId + component);
	} else {
	timelineMetric.setInstanceId(component);
	}
	timelineMetric.setAppId("FLUME_HANDLER");
	timelineMetric.setStartTime(currentTimeMillis);
	timelineMetric.getMetricValues().put(currentTimeMillis, Double.parseDouble(attributeValue));
	return timelineMetric;
	}
	}

	private boolean isCounterMetric(String attributeName) {
	return counterMetrics.contains(attributeName);
	}
	}