AMBARI-13336 Flume agent specific metrics not showing up as graphs (dsen)

ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/cache/TimelineMetricsCache.java

@@ -78,7 +78,7 @@ public class TimelineMetricsCache {
   }
 
   // TODO: Change to ConcurentHashMap with weighted eviction
-  class TimelineMetricHolder extends LinkedHashMap<String, TimelineMetricWrapper> {
+  class TimelineMetricHolder extends LinkedHashMap<String, TimelineMetricWrapper> {//
     private static final long serialVersionUID = 1L;
     private boolean gotOverflow = false;
     // To avoid duplication at the end of the buffer and beginning of the next

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

@@ -36,6 +36,7 @@ import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -46,7 +47,10 @@ import java.util.concurrent.TimeUnit;
 
 public class FlumeTimelineMetricsSink extends AbstractTimelineMetricsSink implements MonitorService {
   private String collectorUri;
-  private TimelineMetricsCache metricsCache;
+  // Key - component(instance_id)
+  private Map<String, TimelineMetricsCache> metricsCaches;
+  private int maxRowCacheSize;
+  private int metricsSendInterval;
   private ScheduledExecutorService scheduledExecutorService;
   private long pollFrequency;
   private String hostname;
@@ -83,11 +87,11 @@ public class FlumeTimelineMetricsSink extends AbstractTimelineMetricsSink implem
     Configuration configuration = new Configuration("/flume-metrics2.properties");
     timeoutSeconds = Integer.parseInt(configuration.getProperty(METRICS_POST_TIMEOUT_SECONDS,
         String.valueOf(DEFAULT_POST_TIMEOUT_SECONDS)));
-    int maxRowCacheSize = Integer.parseInt(configuration.getProperty(MAX_METRIC_ROW_CACHE_SIZE,
+    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,
+    metricsSendInterval = Integer.parseInt(configuration.getProperty(METRICS_SEND_INTERVAL,
         String.valueOf(TimelineMetricsCache.MAX_EVICTION_TIME_MILLIS)));
-    metricsCache = new TimelineMetricsCache(maxRowCacheSize, metricsSendInterval);
+    metricsCaches = new HashMap<String, TimelineMetricsCache>();
     String collectorHostname = configuration.getProperty(COLLECTOR_HOST_PROPERTY);
     String port = configuration.getProperty(COLLECTOR_PORT_PROPERTY);
     collectorUri = "http://" + collectorHostname + ":" + port + "/ws/v1/timeline/metrics";
@@ -111,8 +115,9 @@ public class FlumeTimelineMetricsSink extends AbstractTimelineMetricsSink implem
     this.pollFrequency = pollFrequency;
   }
 
-  public void setMetricsCache(TimelineMetricsCache metricsCache) {
-    this.metricsCache = metricsCache;
+  //Test hepler method
+  protected void setMetricsCaches(Map<String, TimelineMetricsCache> metricsCaches) {
+    this.metricsCaches = metricsCaches;
   }
 
   /**
@@ -144,6 +149,10 @@ public class FlumeTimelineMetricsSink extends AbstractTimelineMetricsSink implem
 
     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)) {

ambari-metrics/ambari-metrics-flume-sink/src/test/java/org/apache/hadoop/metrics2/sink/flume/FlumeTimelineMetricsSinkTest.java

@@ -65,7 +65,7 @@ public class FlumeTimelineMetricsSinkTest {
 
   private TimelineMetricsCache getTimelineMetricsCache(FlumeTimelineMetricsSink flumeTimelineMetricsSink) {
     TimelineMetricsCache timelineMetricsCache = EasyMock.createNiceMock(TimelineMetricsCache.class);
-    flumeTimelineMetricsSink.setMetricsCache(timelineMetricsCache);
+    flumeTimelineMetricsSink.setMetricsCaches(Collections.singletonMap("SINK",timelineMetricsCache));
     EasyMock.expect(timelineMetricsCache.getTimelineMetric("key1"))
         .andReturn(new TimelineMetric()).once();
     timelineMetricsCache.putTimelineMetric(EasyMock.anyObject(TimelineMetric.class));

ambari-server/src/main/java/org/apache/ambari/server/controller/internal/AbstractPropertyProvider.java

@@ -181,6 +181,8 @@ public abstract class AbstractPropertyProvider extends BaseProvider implements P
     // find the argument in the property id
     int argStart = propertyId.indexOf(argName);
 
+    String value = val == null ? "" : val;
+
     if (argStart > -1) {
       // get the string segment starting with the given argument
       String argSegment = propertyId.substring(argStart);
@@ -212,15 +214,18 @@ public abstract class AbstractPropertyProvider extends BaseProvider implements P
           }
 
           try {
-            val = invokeArgumentMethod(val, methodName, argList, paramTypes);
+            value = invokeArgumentMethod(value, methodName, argList, paramTypes);
           } catch (Exception e) {
             throw new IllegalArgumentException("Can't apply method " + methodName + " for argument " +
                 argName + " in " + propertyId, e);
           }
         }
+        if (value.equals(val)) {
+          return propertyId;
+        }
       }
       // Do the substitution
-      return propertyId.replace(argName, val);
+      return propertyId.replace(argName, value);
     }
     throw new IllegalArgumentException("Can't substitute " + val + "  for argument " + argName + " in " + propertyId);
   }
@@ -368,7 +373,7 @@ public abstract class AbstractPropertyProvider extends BaseProvider implements P
    * Get value from the given metric.
    *
    * @param metric      the metric
-   * @param isTemporal  indicates whether or not this a temporal metric
+   * @param temporalInfo  indicates whether or not this a temporal metric
    *
    * @return a range of temporal data or a point in time value if not temporal
    */

ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/AMSPropertyProvider.java

@@ -399,13 +399,19 @@ public abstract class AMSPropertyProvider extends MetricsPropertyProvider {
               if (metricsMap.containsKey(propertyId)){
                 if (containsArguments(propertyId)) {
                   int i = 1;
-                  for (String param : parameterList) {
-                    propertyId = substituteArgument(propertyId, "$" + i, param);
-                    ++i;
+                  //if nothing to substitute in metric name, then
+                  //substitute $1 with an instanceId
+                  if (!parameterList.isEmpty())  {
+                    for (String param : parameterList) {
+                      propertyId = substituteArgument(propertyId, "$" + i, param);
+                      ++i;
+                    }
+                  } else {
+                    propertyId = substituteArgument(propertyId, "$1", metric.getInstanceId());
                   }
                 }
                 Object value = getValue(metric, temporalInfo);
-                if (value != null) {
+                if (value != null && !containsArguments(propertyId)) {
                   resource.setProperty(propertyId, value);
                 }
               }

ambari-server/src/main/resources/common-services/FLUME/1.4.0.2.0/metrics.json

@@ -218,205 +218,204 @@
         "type": "ganglia",
         "metrics": {
           "default": {
-            "metrics/flume/flume/CHANNEL/ChannelCapacity": {
+              "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/ChannelCapacity": {
               "metric": "ChannelCapacity",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/CHANNEL/StartTime": {
+            "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/StartTime": {
               "metric": "StartTime",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/CHANNEL/EventTakeAttemptCount": {
+            "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/EventTakeAttemptCount": {
               "metric": "EventTakeAttemptCount",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/CHANNEL/EventTakeSuccessCount": {
+            "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/EventTakeSuccessCount": {
               "metric": "EventTakeSuccessCount",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/CHANNEL/EventPutAttemptCount": {
+            "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/EventPutAttemptCount": {
               "metric": "EventPutAttemptCount",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/CHANNEL/StopTime": {
+            "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/StopTime": {
               "metric": "StopTime",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/CHANNEL/ChannelFillPercentage": {
+            "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/ChannelFillPercentage": {
               "metric": "ChannelFillPercentage",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/CHANNEL/ChannelSize": {
+            "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/ChannelSize": {
               "metric": "ChannelSize",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/CHANNEL/EventPutSuccessCount": {
+            "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/EventPutSuccessCount": {
               "metric": "EventPutSuccessCount",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/SINK/ConnectionCreatedCount": {
+            "metrics/flume/flume/SINK/$1.replaceAll(\"^SINK\.\",\"\")/ConnectionCreatedCount": {
               "metric": "ConnectionCreatedCount",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/SINK/BatchCompleteCount": {
+            "metrics/flume/flume/SINK/$1.replaceAll(\"^SINK\.\",\"\")/BatchCompleteCount": {
               "metric": "BatchCompleteCount",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/SINK/EventDrainSuccessCount": {
+            "metrics/flume/flume/SINK/$1.replaceAll(\"^SINK\.\",\"\")/EventDrainSuccessCount": {
               "metric": "EventDrainSuccessCount",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/SINK/StartTime": {
+            "metrics/flume/flume/SINK/$1.replaceAll(\"^SINK\.\",\"\")/StartTime": {
               "metric": "StartTime",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/SINK/EventDrainAttemptCount": {
+            "metrics/flume/flume/SINK/$1.replaceAll(\"^SINK\.\",\"\")/EventDrainAttemptCount": {
               "metric": "EventDrainAttemptCount",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/SINK/ConnectionFailedCount": {
+            "metrics/flume/flume/SINK/$1.replaceAll(\"^SINK\.\",\"\")/ConnectionFailedCount": {
               "metric": "ConnectionFailedCount",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/SINK/BatchUnderflowCount": {
+            "metrics/flume/flume/SINK/$1.replaceAll(\"^SINK\.\",\"\")/BatchUnderflowCount": {
               "metric": "BatchUnderflowCount",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/SINK/ConnectionClosedCount": {
+            "metrics/flume/flume/SINK/$1.replaceAll(\"^SINK\.\",\"\")/ConnectionClosedCount": {
               "metric": "ConnectionClosedCount",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/SINK/StopTime": {
+            "metrics/flume/flume/SINK/$1.replaceAll(\"^SINK\.\",\"\")/StopTime": {
               "metric": "StopTime",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/SINK/BatchEmptyCount": {
+            "metrics/flume/flume/SINK/$1.replaceAll(\"^SINK\.\",\"\")/BatchEmptyCount": {
               "metric": "BatchEmptyCount",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/SOURCE/AppendBatchReceivedCount": {
+            "metrics/flume/flume/SOURCE/$1.replaceAll(\"^SOURCE\.\",\"\")/AppendBatchReceivedCount": {
               "metric": "AppendBatchReceivedCount",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/SOURCE/AppendAcceptedCount": {
+            "metrics/flume/flume/SOURCE/$1.replaceAll(\"^SOURCE\.\",\"\")/AppendAcceptedCount": {
               "metric": "AppendAcceptedCount",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/SOURCE/StartTime": {
+            "metrics/flume/flume/SOURCE/$1.replaceAll(\"^SOURCE\.\",\"\")/StartTime": {
               "metric": "StartTime",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/SOURCE/OpenConnectionCount": {
+            "metrics/flume/flume/SOURCE/$1.replaceAll(\"^SOURCE\.\",\"\")/OpenConnectionCount": {
               "metric": "OpenConnectionCount",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/SOURCE/AppendBatchAcceptedCount": {
+            "metrics/flume/flume/SOURCE/$1.replaceAll(\"^SOURCE\.\",\"\")/AppendBatchAcceptedCount": {
               "metric": "AppendBatchAcceptedCount",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/SOURCE/AppendReceivedCount": {
+            "metrics/flume/flume/SOURCE/$1.replaceAll(\"^SOURCE\.\",\"\")/AppendReceivedCount": {
               "metric": "AppendReceivedCount",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/SOURCE/EventReceivedCount": {
+            "metrics/flume/flume/SOURCE/$1.replaceAll(\"^SOURCE\.\",\"\")/EventReceivedCount": {
               "metric": "EventReceivedCount",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/SOURCE/StopTime": {
+            "metrics/flume/flume/SOURCE/$1.replaceAll(\"^SOURCE\.\",\"\")/StopTime": {
               "metric": "StopTime",
               "pointInTime": true,
               "temporal": true
             },
-            "metrics/flume/flume/SOURCE/EventAcceptedCount": {
+            "metrics/flume/flume/SOURCE/$1.replaceAll(\"^SOURCE\.\",\"\")/EventAcceptedCount": {
               "metric": "EventAcceptedCount",
               "pointInTime": true,
               "temporal": true
             },
 
-            "metrics/flume/flume/CHANNEL/EventTakeSuccessCount/rate/avg": {
+            "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/EventTakeSuccessCount/rate/avg": {
               "metric": "EventTakeSuccessCount._rate._avg",
               "pointInTime": false,
               "temporal": true
             },
-            "metrics/flume/flume/CHANNEL/EventTakeSuccessCount/rate/max": {
+            "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/EventTakeSuccessCount/rate/max": {
               "metric": "EventTakeSuccessCount._rate._max",
               "pointInTime": false,
               "temporal": true
             },
-            "metrics/flume/flume/CHANNEL/EventTakeSuccessCount/rate/min": {
+            "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/EventTakeSuccessCount/rate/min": {
               "metric": "EventTakeSuccessCount._rate._min",
               "pointInTime": false,
               "temporal": true
             },
-            "metrics/flume/flume/CHANNEL/EventTakeSuccessCount/rate/sum": {
+            "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/EventTakeSuccessCount/rate/sum": {
               "metric": "EventTakeSuccessCount._rate._sum",
               "pointInTime": false,
               "temporal": true
             },
-
-            "metrics/flume/flume/CHANNEL/EventPutSuccessCount/rate/avg": {
+            "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/EventPutSuccessCount/rate/avg": {
               "metric": "EventPutSuccessCount._rate._avg",
               "pointInTime": false,
               "temporal": true
             },
-            "metrics/flume/flume/CHANNEL/EventPutSuccessCount/rate/max": {
+            "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/EventPutSuccessCount/rate/max": {
               "metric": "EventPutSuccessCount._rate._max",
               "pointInTime": false,
               "temporal": true
             },
-            "metrics/flume/flume/CHANNEL/EventPutSuccessCount/rate/min": {
+            "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/EventPutSuccessCount/rate/min": {
               "metric": "EventPutSuccessCount._rate._min",
               "pointInTime": false,
               "temporal": true
             },
-            "metrics/flume/flume/CHANNEL/EventPutSuccessCount/rate/sum": {
+            "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/EventPutSuccessCount/rate/sum": {
               "metric": "EventPutSuccessCount._rate._sum",
               "pointInTime": false,
               "temporal": true
             },
 
-            "metrics/flume/flume/CHANNEL/ChannelSize/rate/avg": {
+            "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/ChannelSize/rate/avg": {
               "metric": "ChannelSize._rate._avg",
               "pointInTime": false,
               "temporal": true
             },
-            "metrics/flume/flume/CHANNEL/ChannelSize/rate/max": {
+            "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/ChannelSize/rate/max": {
               "metric": "ChannelSize._rate._max",
               "pointInTime": false,
               "temporal": true
             },
-            "metrics/flume/flume/CHANNEL/ChannelSize/rate/min": {
+            "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/ChannelSize/rate/min": {
               "metric": "ChannelSize._rate._min",
               "pointInTime": false,
               "temporal": true
             },
-            "metrics/flume/flume/CHANNEL/ChannelSize/rate/sum": {
+            "metrics/flume/flume/CHANNEL/$1.replaceAll(\"^CHANNEL\.\",\"\")/ChannelSize/rate/sum": {
               "metric": "ChannelSize._rate._sum",
               "pointInTime": false,
               "temporal": true