Merge branch 'develop'

Author: bmwilli

streams-metric-exporter/README.md

@@ -289,6 +289,7 @@ Note: the streams-metric-exporter resolves operator input and output ports to na
 |**instancename**|name of streams instance (instance is reserved by prometheus)|
 |**jobname**| name of streams job (job is reserved by prometheus)|
 |**operatorname**| name of operator|
+|**operatorkind**| kind of operator (e.g. spl.relational::Functor)
 |**inputportname**| name of input port|
 |**outputportname**| name of output port|
 |**resource**| name of streams resource|
@@ -313,7 +314,7 @@ streams_job_pecount{domainname="StreamsDomain",instancename="StreamsInstance",jo
 
 # HELP streams_operator_ip_nTuplesProcessed Streams operator input port metric: nTuplesProcessed
 # TYPE streams_operator_ip_nTuplesProcessed gauge
-streams_operator_ip_nTuplesProcessed{domainname="StreamsDomain",instancename="StreamsInstance",jobname="MultiPEJob",resource="streamsqse",peid="1",operatorname="FilterStream",inputportname="BeaconStream",} 675632.0
+streams_operator_ip_nTuplesProcessed{domainname="StreamsDomain",instancename="StreamsInstance",jobname="MultiPEJob",resource="streamsqse",peid="1",operatorname="FilterStream",operatorkind="spl.relational.Filter",inputportname="BeaconStream",} 675632.0
 
 # HELP streams_pe_op_connection_congestionFactor Streams pe output port connection metric: congestionFactor
 # TYPE streams_pe_op_connection_congestionFactor gauge

streams-metric-exporter/src/main/java/streams/metric/exporter/metrics/MetricsExporter.java

@@ -65,9 +65,9 @@ public enum StreamsObjectType {
 		PE_INPUTPORT("streams_pe_ip_",new String[] {"domainname","instancename", "jobname", "resource", "peid","index"}),
 		PE_OUTPUTPORT("streams_pe_op_",new String[] {"domainname","instancename","jobname", "resource", "peid","index"}),
 		PE_OUTPUTPORT_CONNECTION("streams_pe_op_connection_",new String[] {"domainname","instancename","jobname","resource","peid","index","connectionid"}),
-		OPERATOR("streams_operator_",new String[] { "domainname","instancename", "jobname", "resource", "peid", "operatorname" }), 
-		OPERATOR_INPUTPORT("streams_operator_ip_",new String[] { "domainname","instancename", "jobname", "resource", "peid", "operatorname", "inputportname" }), 
-		OPERATOR_OUTPUTPORT("streams_operator_op_",new String[] { "domainname","instancename", "jobname", "resource", "peid", "operatorname","outputportname" });
+		OPERATOR("streams_operator_",new String[] { "domainname","instancename", "jobname", "resource", "peid", "operatorname", "operatorkind" }), 
+		OPERATOR_INPUTPORT("streams_operator_ip_",new String[] { "domainname","instancename", "jobname", "resource", "peid", "operatorname", "operatorkind", "inputportname" }), 
+		OPERATOR_OUTPUTPORT("streams_operator_op_",new String[] { "domainname","instancename", "jobname", "resource", "peid", "operatorname", "operatorkind", "outputportname" });
 
 		private final String metric_prefix;
 		private final String[] labels;

streams-metric-exporter/src/main/java/streams/metric/exporter/streamstracker/job/JobDetails.java

@@ -99,6 +99,7 @@ public class JobDetails implements NotificationListener {
 	private boolean jobTopologyRefreshRequired = false;
 
 	private final Map<BigInteger, String> peResourceMap = new HashMap<BigInteger, String>();
+	private final Map<String, String> operatorKindMap = new HashMap<String, String>();
 	private final Map<String, Map<Integer, String>> operatorInputPortNames = new HashMap<String, Map<Integer, String>>();
 	private final Map<String, Map<Integer, String>> operatorOutputPortNames = new HashMap<String, Map<Integer, String>>();
 
@@ -666,6 +667,7 @@ private void mapPortNames(MXBeanSource beanSource) {
 		for (String operatorName : operators) {
 			OperatorMXBean operatorBean = beanSource.getOperatorMXBean(getDomain(), getInstance(), getJobid(),
 					operatorName);
+			operatorKindMap.put(operatorName,operatorBean.getOperatorKind());
 			mapOperatorInputPortNames(beanSource, operatorName, operatorBean.getInputPorts());
 			mapOperatorOutputPortNames(beanSource, operatorName, operatorBean.getOutputPorts());
 		}
@@ -967,6 +969,7 @@ private void updateExportedMetrics() {
 						JSONObject operator = (JSONObject) operatorArray.get(op);
 						//System.out.println(operator.toString());
 						String operatorName = (String)operator.get("name");
+						String operatorKind = this.operatorKindMap.get(operatorName);
 //						System.out.println("OPERATOR NAME: " + operatorName);
 						JSONArray opMetricsArray = (JSONArray) operator.get("metrics");
 						/* Operator Metrics Loop, these are non-standard metrics */
@@ -988,7 +991,8 @@ private void updateExportedMetrics() {
 										name,
 										resource,
 										peid,
-										operatorName).set((long)metric.get("value"));
+										operatorName,
+										operatorKind).set((long)metric.get("value"));
 								break;
 							}
 						}	// End Operator Metrics Loop		
@@ -1014,6 +1018,7 @@ private void updateExportedMetrics() {
 											resource,
 											peid,
 											operatorName,
+											operatorKind,
 											inputPortName).set((long)metric.get("value"));
 									break;
 								}
@@ -1041,6 +1046,7 @@ private void updateExportedMetrics() {
 											resource,
 											peid,
 											operatorName,
+											operatorKind,
 											outputPortName).set((long)metric.get("value"));
 									break;
 								}