ambari-metrics-storm-sink/src/main/java/org/apache/hadoop/metrics2/sink/storm/StormTimelineMetricsReporter.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.storm;

 import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;

 import org.apache.commons.lang3.ClassUtils;
 import org.apache.commons.lang3.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.UnableToConnectException;
 import org.apache.hadoop.metrics2.sink.timeline.configuration.Configuration;
 import org.apache.storm.metric.api.DataPoint;
 import org.apache.storm.metric.api.IClusterMetricsConsumer;

 public class StormTimelineMetricsReporter extends AbstractTimelineMetricsSink
     implements IClusterMetricsConsumer {

   public static final String CLUSTER_REPORTER_APP_ID = "clusterReporterAppId";
   public static final String DEFAULT_CLUSTER_REPORTER_APP_ID = "nimbus";

   private String hostname;
   private String port;
   private Collection<String> collectorHosts;
   private String zkQuorum;
   private String protocol;
   private boolean setInstanceId;
   private String instanceId;
   private String applicationId;
   private int timeoutSeconds;
   private boolean hostInMemoryAggregationEnabled;
   private int hostInMemoryAggregationPort;
   private String hostInMemoryAggregationProtocol;

   public StormTimelineMetricsReporter() {

   }

   @Override
   protected String getCollectorUri(String host) {
     return constructTimelineMetricUri(protocol, host, port);
   }

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

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

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

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

   @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;
   }

   @Override
   public void prepare(Object registrationArgument) {
     LOG.info("Preparing Storm Metrics Reporter");
     try {
       try {
         hostname = InetAddress.getLocalHost().getHostName();
         // If not FQDN , call  DNS
         if ((hostname == null) || (!hostname.contains("."))) {
           hostname = InetAddress.getLocalHost().getCanonicalHostName();
         }
       } catch (UnknownHostException e) {
         LOG.error("Could not identify hostname.");
         throw new RuntimeException("Could not identify hostname.", e);
       }

       Configuration configuration = new Configuration("/storm-metrics2.properties");
       collectorHosts = parseHostsStringIntoCollection(configuration.getProperty(COLLECTOR_HOSTS_PROPERTY));
       protocol = configuration.getProperty(COLLECTOR_PROTOCOL, "http");
       port = configuration.getProperty(COLLECTOR_PORT, "6188");

       zkQuorum = StringUtils.isEmpty(configuration.getProperty(COLLECTOR_ZOOKEEPER_QUORUM)) ?
         configuration.getProperty(ZOOKEEPER_QUORUM) : configuration.getProperty(COLLECTOR_ZOOKEEPER_QUORUM);

       timeoutSeconds = configuration.getProperty(METRICS_POST_TIMEOUT_SECONDS) != null ?
           Integer.parseInt(configuration.getProperty(METRICS_POST_TIMEOUT_SECONDS)) :
           DEFAULT_POST_TIMEOUT_SECONDS;
       applicationId = configuration.getProperty(CLUSTER_REPORTER_APP_ID, DEFAULT_CLUSTER_REPORTER_APP_ID);
       setInstanceId = Boolean.valueOf(configuration.getProperty(SET_INSTANCE_ID_PROPERTY));
       instanceId = configuration.getProperty(INSTANCE_ID_PROPERTY);

       hostInMemoryAggregationEnabled = Boolean.valueOf(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");

       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);
       }
     } catch (Exception e) {
       LOG.warn("Could not initialize metrics collector, please specify " +
           "protocol, host, port, appId, zkQuorum under $STORM_HOME/conf/storm-metrics2.properties ", e);
     }
     // Initialize the collector write strategy
     super.init();
   }

   @Override
   public void handleDataPoints(ClusterInfo clusterInfo, Collection<DataPoint> dataPoints) {
     long timestamp = clusterInfo.getTimestamp();
     List<TimelineMetric> totalMetrics = new ArrayList<>();

     for (DataPoint dataPoint : dataPoints) {
       LOG.debug(dataPoint.getName() + " = " + dataPoint.getValue());
       List<DataPoint> populatedDataPoints = populateDataPoints(dataPoint);

       for (DataPoint populatedDataPoint : populatedDataPoints) {
         LOG.debug("Populated datapoint: " + dataPoint.getName() + " = " + dataPoint.getValue());

         try {
           StormAmbariMappedMetric mappedMetric = StormAmbariMappedMetric
               .valueOf(populatedDataPoint.getName());
           TimelineMetric timelineMetric = createTimelineMetric(timestamp * 1000, applicationId,
               mappedMetric.getAmbariMetricName(),
               Double.valueOf(populatedDataPoint.getValue().toString()));

           totalMetrics.add(timelineMetric);
         } catch (IllegalArgumentException e) {
           // not interested metrics on Ambari, skip
           LOG.debug("Not interested metrics, skip: " + populatedDataPoint.getName());
         }
       }
     }

     if (totalMetrics.size() <= 0) {
       return;
     }

     TimelineMetrics timelineMetrics = new TimelineMetrics();
     timelineMetrics.setMetrics(totalMetrics);

     try {
       emitMetrics(timelineMetrics);
     } catch (UnableToConnectException e) {
       LOG.warn("Unable to connect to Metrics Collector " + e.getConnectUrl() + ". " + e.getMessage());
     }
   }

   @Override
   public void handleDataPoints(SupervisorInfo supervisorInfo, Collection<DataPoint> dataPoints) {
     // Ambari is not interested on metrics on each supervisor
   }

   @Override
   public void cleanup() {
     LOG.info("Stopping Storm Metrics Reporter");
   }

   private List<DataPoint> populateDataPoints(DataPoint dataPoint) {
     List<DataPoint> dataPoints = new ArrayList<>();

     if (dataPoint.getValue() == null) {
       LOG.warn("Data point with name " + dataPoint.getName() + " is null. Discarding." + dataPoint
           .getName());
     } else if (dataPoint.getValue() instanceof Map) {
       Map<String, Object> dataMap = (Map<String, Object>) dataPoint.getValue();

       for (Map.Entry<String, Object> entry : dataMap.entrySet()) {
         Double value = convertValueToDouble(entry.getKey(), entry.getValue());
         if (value != null) {
           dataPoints.add(new DataPoint(dataPoint.getName() + "." + entry.getKey(), value));
         }
       }
     } else {
       Double value = convertValueToDouble(dataPoint.getName(), dataPoint.getValue());
       if (value != null) {
         dataPoints.add(new DataPoint(dataPoint.getName(), value));
       }
     }

     return dataPoints;
   }

   private Double convertValueToDouble(String metricName, Object value) {
     try {
       Double converted = NumberUtil.convertValueToDouble(value);
       if (converted == null) {
         LOG.warn("Data point with name " + metricName + " has value " + value +
             " which is not supported. Discarding.");
       }
       return converted;
     } catch (NumberFormatException e) {
       LOG.warn("Data point with name " + metricName + " doesn't have number format value " +
           value + ". Discarding.");
       return null;
     }
   }

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

   enum StormAmbariMappedMetric {
     supervisors("Supervisors"),
     topologies("Topologies"),
     slotsTotal("Total Slots"),
     slotsUsed("Used Slots"),
     slotsFree("Free Slots"),
     executorsTotal("Total Executors"),
     tasksTotal("Total Tasks");

     private String ambariMetricName;

     StormAmbariMappedMetric(String ambariMetricName) {
       this.ambariMetricName = ambariMetricName;
     }

     public String getAmbariMetricName() {
       return ambariMetricName;
     }
   }
 }
	/**
	* 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.storm;

	import java.net.InetAddress;
	import java.net.UnknownHostException;
	import java.util.ArrayList;
	import java.util.Collection;
	import java.util.List;
	import java.util.Map;

	import org.apache.commons.lang3.ClassUtils;
	import org.apache.commons.lang3.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.UnableToConnectException;
	import org.apache.hadoop.metrics2.sink.timeline.configuration.Configuration;
	import org.apache.storm.metric.api.DataPoint;
	import org.apache.storm.metric.api.IClusterMetricsConsumer;

	public class StormTimelineMetricsReporter extends AbstractTimelineMetricsSink
	implements IClusterMetricsConsumer {

	public static final String CLUSTER_REPORTER_APP_ID = "clusterReporterAppId";
	public static final String DEFAULT_CLUSTER_REPORTER_APP_ID = "nimbus";

	private String hostname;
	private String port;
	private Collection<String> collectorHosts;
	private String zkQuorum;
	private String protocol;
	private boolean setInstanceId;
	private String instanceId;
	private String applicationId;
	private int timeoutSeconds;
	private boolean hostInMemoryAggregationEnabled;
	private int hostInMemoryAggregationPort;
	private String hostInMemoryAggregationProtocol;

	public StormTimelineMetricsReporter() {

	}

	@Override
	protected String getCollectorUri(String host) {
	return constructTimelineMetricUri(protocol, host, port);
	}

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

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

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

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

	@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;
	}

	@Override
	public void prepare(Object registrationArgument) {
	LOG.info("Preparing Storm Metrics Reporter");
	try {
	try {
	hostname = InetAddress.getLocalHost().getHostName();
	// If not FQDN , call DNS
	if ((hostname == null) \|\| (!hostname.contains("."))) {
	hostname = InetAddress.getLocalHost().getCanonicalHostName();
	}
	} catch (UnknownHostException e) {
	LOG.error("Could not identify hostname.");
	throw new RuntimeException("Could not identify hostname.", e);
	}

	Configuration configuration = new Configuration("/storm-metrics2.properties");
	collectorHosts = parseHostsStringIntoCollection(configuration.getProperty(COLLECTOR_HOSTS_PROPERTY));
	protocol = configuration.getProperty(COLLECTOR_PROTOCOL, "http");
	port = configuration.getProperty(COLLECTOR_PORT, "6188");

	zkQuorum = StringUtils.isEmpty(configuration.getProperty(COLLECTOR_ZOOKEEPER_QUORUM)) ?
	configuration.getProperty(ZOOKEEPER_QUORUM) : configuration.getProperty(COLLECTOR_ZOOKEEPER_QUORUM);

	timeoutSeconds = configuration.getProperty(METRICS_POST_TIMEOUT_SECONDS) != null ?
	Integer.parseInt(configuration.getProperty(METRICS_POST_TIMEOUT_SECONDS)) :
	DEFAULT_POST_TIMEOUT_SECONDS;
	applicationId = configuration.getProperty(CLUSTER_REPORTER_APP_ID, DEFAULT_CLUSTER_REPORTER_APP_ID);
	setInstanceId = Boolean.valueOf(configuration.getProperty(SET_INSTANCE_ID_PROPERTY));
	instanceId = configuration.getProperty(INSTANCE_ID_PROPERTY);

	hostInMemoryAggregationEnabled = Boolean.valueOf(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");

	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);
	}
	} catch (Exception e) {
	LOG.warn("Could not initialize metrics collector, please specify " +
	"protocol, host, port, appId, zkQuorum under $STORM_HOME/conf/storm-metrics2.properties ", e);
	}
	// Initialize the collector write strategy
	super.init();
	}

	@Override
	public void handleDataPoints(ClusterInfo clusterInfo, Collection<DataPoint> dataPoints) {
	long timestamp = clusterInfo.getTimestamp();
	List<TimelineMetric> totalMetrics = new ArrayList<>();

	for (DataPoint dataPoint : dataPoints) {
	LOG.debug(dataPoint.getName() + " = " + dataPoint.getValue());
	List<DataPoint> populatedDataPoints = populateDataPoints(dataPoint);

	for (DataPoint populatedDataPoint : populatedDataPoints) {
	LOG.debug("Populated datapoint: " + dataPoint.getName() + " = " + dataPoint.getValue());

	try {
	StormAmbariMappedMetric mappedMetric = StormAmbariMappedMetric
	.valueOf(populatedDataPoint.getName());
	TimelineMetric timelineMetric = createTimelineMetric(timestamp * 1000, applicationId,
	mappedMetric.getAmbariMetricName(),
	Double.valueOf(populatedDataPoint.getValue().toString()));

	totalMetrics.add(timelineMetric);
	} catch (IllegalArgumentException e) {
	// not interested metrics on Ambari, skip
	LOG.debug("Not interested metrics, skip: " + populatedDataPoint.getName());
	}
	}
	}

	if (totalMetrics.size() <= 0) {
	return;
	}

	TimelineMetrics timelineMetrics = new TimelineMetrics();
	timelineMetrics.setMetrics(totalMetrics);

	try {
	emitMetrics(timelineMetrics);
	} catch (UnableToConnectException e) {
	LOG.warn("Unable to connect to Metrics Collector " + e.getConnectUrl() + ". " + e.getMessage());
	}
	}

	@Override
	public void handleDataPoints(SupervisorInfo supervisorInfo, Collection<DataPoint> dataPoints) {
	// Ambari is not interested on metrics on each supervisor
	}

	@Override
	public void cleanup() {
	LOG.info("Stopping Storm Metrics Reporter");
	}

	private List<DataPoint> populateDataPoints(DataPoint dataPoint) {
	List<DataPoint> dataPoints = new ArrayList<>();

	if (dataPoint.getValue() == null) {
	LOG.warn("Data point with name " + dataPoint.getName() + " is null. Discarding." + dataPoint
	.getName());
	} else if (dataPoint.getValue() instanceof Map) {
	Map<String, Object> dataMap = (Map<String, Object>) dataPoint.getValue();

	for (Map.Entry<String, Object> entry : dataMap.entrySet()) {
	Double value = convertValueToDouble(entry.getKey(), entry.getValue());
	if (value != null) {
	dataPoints.add(new DataPoint(dataPoint.getName() + "." + entry.getKey(), value));
	}
	}
	} else {
	Double value = convertValueToDouble(dataPoint.getName(), dataPoint.getValue());
	if (value != null) {
	dataPoints.add(new DataPoint(dataPoint.getName(), value));
	}
	}

	return dataPoints;
	}

	private Double convertValueToDouble(String metricName, Object value) {
	try {
	Double converted = NumberUtil.convertValueToDouble(value);
	if (converted == null) {
	LOG.warn("Data point with name " + metricName + " has value " + value +
	" which is not supported. Discarding.");
	}
	return converted;
	} catch (NumberFormatException e) {
	LOG.warn("Data point with name " + metricName + " doesn't have number format value " +
	value + ". Discarding.");
	return null;
	}
	}

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

	enum StormAmbariMappedMetric {
	supervisors("Supervisors"),
	topologies("Topologies"),
	slotsTotal("Total Slots"),
	slotsUsed("Used Slots"),
	slotsFree("Free Slots"),
	executorsTotal("Total Executors"),
	tasksTotal("Total Tasks");

	private String ambariMetricName;

	StormAmbariMappedMetric(String ambariMetricName) {
	this.ambariMetricName = ambariMetricName;
	}

	public String getAmbariMetricName() {
	return ambariMetricName;
	}
	}
	}