AMBARI-3853. Poll Ganglia metrics from rrd.py with POST instead of GET(odiachenko)

ambari-agent/src/main/puppet/modules/hdp-ganglia/files/rrd.py

@@ -24,6 +24,7 @@ import rrdtool
 import sys
 import time
 import re
+import urlparse
 
 # place this script in /var/www/cgi-bin of the Ganglia collector
 # requires 'yum install rrdtool-python' on the Ganglia collector
@@ -94,7 +95,14 @@ sys.stdout.write("Content-type: text/plain\n\n")
 sys.stdout.write(str(time.mktime(time.gmtime())))
 sys.stdout.write("\n")
 
-queryString = dict(cgi.parse_qsl(os.environ['QUERY_STRING']));
+requestMethod = os.environ['REQUEST_METHOD']
+
+if requestMethod == 'POST':
+  postData = sys.stdin.read()
+  queryString = urlparse.parse_qs(postData)
+  queryString = dict((k, v[0]) for k, v in queryString.items())
+elif requestMethod == 'GET':
+  queryString = dict(cgi.parse_qsl(os.environ['QUERY_STRING']));
 
 if "m" in queryString:
   metricParts = queryString["m"].split(",")

ambari-server/src/main/java/org/apache/ambari/server/actionmanager/ActionScheduler.java

@@ -450,9 +450,6 @@ class ActionScheduler implements Runnable {
 
     //Try to get clusterHostInfo from cache
     Map<String, List<String>> clusterHostInfo = clusterHostInfoCache.getIfPresent(s.getStageId());
-    
-    LOG.info("Cluster host info cache size: " + clusterHostInfoCache.size());
-    
 
     if (clusterHostInfo == null) {
       Type type = new TypeToken<Map<String, List<String>>>() {}.getType();

ambari-server/src/main/java/org/apache/ambari/server/controller/ganglia/GangliaPropertyProvider.java

@@ -263,7 +263,7 @@ public abstract class GangliaPropertyProvider extends AbstractPropertyProvider {
    * @param metricSet     the set of metric names
    * @param temporalInfo  the temporal information
    *
-   * @return the spec
+   * @return the spec, like http://example.com/path?param1=val1&paramn=valn
    *
    * @throws SystemException if unable to get the Ganglia Collector host name
    */
@@ -272,11 +272,11 @@ public abstract class GangliaPropertyProvider extends AbstractPropertyProvider {
                          Set<String> hostSet,
                          Set<String> metricSet,
                          TemporalInfo temporalInfo) throws SystemException {
-
+    
     String clusters = getSetString(clusterSet, -1);
-    String hosts    = getSetString(hostSet, 100);
-    String metrics  = getSetString(metricSet, 60);
-
+    String hosts = getSetString(hostSet, -1);
+    String metrics = getSetString(metricSet, -1);
+    
     URIBuilder uriBuilder = new URIBuilder();
 
     if (configuration.isGangliaSSL()) {
@@ -286,11 +286,10 @@ public abstract class GangliaPropertyProvider extends AbstractPropertyProvider {
     }
 
     uriBuilder.setHost(hostProvider.getGangliaCollectorHostName(clusterName));
-    
+
     uriBuilder.setPath("/cgi-bin/rrd.py");
     
     uriBuilder.setParameter("c", clusters);
-    
 
     if (hosts.length() > 0) {
       uriBuilder.setParameter("h", hosts);
@@ -318,14 +317,14 @@ public abstract class GangliaPropertyProvider extends AbstractPropertyProvider {
       if (step != -1) {
         uriBuilder.setParameter("r", String.valueOf(step));
       }
-    }
-    else {
+    } else {
       uriBuilder.setParameter("e", "now");
       uriBuilder.setParameter("pt", "true");
     }
 
     return uriBuilder.toString();
   }
+  
 
   /**
    * Get value from the given metric.
@@ -425,7 +424,23 @@ public abstract class GangliaPropertyProvider extends AbstractPropertyProvider {
      */
     public Collection<Resource> populateResources() throws SystemException {
 
-      String spec = getSpec(clusterName, clusterSet, hostSet, metrics.keySet(), temporalInfo);
+      //Get full url with parameters
+      String specWithParams = getSpec(clusterName, clusterSet, hostSet, metrics.keySet(), temporalInfo);
+      
+      //URL
+      String spec = null;
+      //Parameters
+      String params = null;
+      
+      String[] tokens = specWithParams.split("\\?", 2);
+
+      try {
+        spec = tokens[0];
+        params = tokens[1];
+      } catch (ArrayIndexOutOfBoundsException e) {
+        LOG.info(e.toString());
+      }
+      
 
       BufferedReader reader = null;
       try {
@@ -443,7 +458,7 @@ public abstract class GangliaPropertyProvider extends AbstractPropertyProvider {
         }
 
         reader = new BufferedReader(new InputStreamReader(
-            getStreamProvider().readFrom(spec)));
+            getStreamProvider().readFrom(spec, "POST", params)));
 
         String feedStart = reader.readLine();
         if (feedStart == null || feedStart.isEmpty()) {

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

@@ -74,9 +74,9 @@ public class URLStreamProvider implements StreamProvider {
     appCookieManager = new AppCookieManager();
   }
 
+  
   @Override
-  public InputStream readFrom(String spec) throws IOException {
-    
+  public InputStream readFrom(String spec, String requestMethod, String params) throws IOException {
     if (LOG.isDebugEnabled()) {
       LOG.debug("readFrom spec:" + spec);
     }
@@ -93,7 +93,11 @@ public class URLStreamProvider implements StreamProvider {
     connection.setConnectTimeout(connTimeout);
     connection.setReadTimeout(readTimeout);
     connection.setDoOutput(true);
-
+    connection.setRequestMethod(requestMethod);
+    
+    if (params != null)
+      connection.getOutputStream().write(params.getBytes());
+    
     int statusCode = connection.getResponseCode();
     if (statusCode == HttpStatus.SC_UNAUTHORIZED ) {
       String wwwAuthHeader = connection.getHeaderField(WWW_AUTHENTICATE);
@@ -111,6 +115,7 @@ public class URLStreamProvider implements StreamProvider {
         connection.setConnectTimeout(connTimeout);
         connection.setReadTimeout(readTimeout);
         connection.setDoOutput(true);
+        
         return connection.getInputStream();
       } else {
         // no supported authentication type found
@@ -123,9 +128,15 @@ public class URLStreamProvider implements StreamProvider {
       // we would let the original response propogate
       return connection.getInputStream();
     }
-
   }
-
+  
+  @Override
+  public InputStream readFrom(String spec) throws IOException {
+    
+    return readFrom(spec, "GET", null);
+    
+  }
+  
   // ----- helper methods ----------------------------------------------------
 
   // Get a connection

ambari-server/src/main/java/org/apache/ambari/server/controller/utilities/StreamProvider.java

@@ -26,4 +26,5 @@ import java.io.InputStream;
  */
 public interface StreamProvider {
   public InputStream readFrom(String spec) throws IOException;
+  public InputStream readFrom(String spec, String requestMethod, String params) throws IOException;
 }

ambari-server/src/test/java/org/apache/ambari/server/controller/ganglia/GangliaPropertyProviderTest.java

@@ -319,10 +319,17 @@ public class GangliaPropertyProviderTest {
 
     Set<Resource> resources = new HashSet<Resource>();
 
+    StringBuilder hostsList = new StringBuilder();
+    
     for (int i = 0; i < 150; ++i) {
       Resource resource = new ResourceImpl(Resource.Type.Host);
       resource.setProperty(HOST_NAME_PROPERTY_ID, "host" + i);
       resources.add(resource);
+      
+      if (hostsList.length() != 0)
+        hostsList.append("," + "host" + i );
+      else
+        hostsList.append("host" + i); 
     }
 
     // only ask for one property
@@ -333,21 +340,26 @@ public class GangliaPropertyProviderTest {
     Assert.assertEquals(150, propertyProvider.populateResources(resources, request, null).size());
 
     
-    URIBuilder uriBuilder = new URIBuilder();
+    URIBuilder expectedUri = new URIBuilder();
     
-    uriBuilder.setScheme((configuration.isGangliaSSL() ? "https" : "http"));
-    uriBuilder.setHost("domU-12-31-39-0E-34-E1.compute-1.internal");
-    uriBuilder.setPath("/cgi-bin/rrd.py");
-    uriBuilder.setParameter("c", "HDPJobTracker,HDPHBaseMaster,HDPResourceManager,HDPSlaves,HDPHistoryServer,HDPNameNode");
-    uriBuilder.setParameter("m", "jvm.metrics.gcCount");
-    uriBuilder.setParameter("s", "10");
-    uriBuilder.setParameter("e", "20");
-    uriBuilder.setParameter("r", "1");
+    expectedUri.setScheme((configuration.isGangliaSSL() ? "https" : "http"));
+    expectedUri.setHost("domU-12-31-39-0E-34-E1.compute-1.internal");
+    expectedUri.setPath("/cgi-bin/rrd.py");
+    expectedUri.setParameter("c", "HDPJobTracker,HDPHBaseMaster,HDPResourceManager,HDPSlaves,HDPHistoryServer,HDPNameNode");
+   
+    expectedUri.setParameter("h", hostsList.toString());
+    expectedUri.setParameter("m", "jvm.metrics.gcCount");
+    expectedUri.setParameter("s", "10");
+    expectedUri.setParameter("e", "20");
+    expectedUri.setParameter("r", "1");
     
-    String expected = uriBuilder.toString();
+    URIBuilder actualUri = new URIBuilder(streamProvider.getLastSpec());
     
-    Assert.assertEquals(expected, streamProvider.getLastSpec());
-
+    Assert.assertEquals(expectedUri.getScheme(), actualUri.getScheme());
+    Assert.assertEquals(expectedUri.getHost(), actualUri.getHost());
+    Assert.assertEquals(expectedUri.getPath(), actualUri.getPath());
+    
+    Assert.assertTrue(isUrlParamsEquals(actualUri, expectedUri));
   }
   
   class PopulateResourceManagerResourcesTest{

ambari-server/src/test/java/org/apache/ambari/server/controller/ganglia/TestStreamProvider.java

@@ -27,6 +27,7 @@ public class TestStreamProvider implements StreamProvider {
 
   private final String fileName;
   private String lastSpec;
+  private boolean isLastSpecUpdated;
 
   public TestStreamProvider(String fileName) {
     this.fileName = fileName;
@@ -34,11 +35,22 @@ public class TestStreamProvider implements StreamProvider {
 
   @Override
   public InputStream readFrom(String spec) throws IOException {
-    lastSpec = spec;
+    if (!isLastSpecUpdated)
+      lastSpec = spec;
+    
+    isLastSpecUpdated = false;
+    
     return ClassLoader.getSystemResourceAsStream(fileName);
   }
 
   public String getLastSpec() {
     return lastSpec;
   }
+
+  @Override
+  public InputStream readFrom(String spec, String requestMethod, String params) throws IOException {
+    lastSpec = spec + "?" + params;
+    isLastSpecUpdated = true;
+    return readFrom(spec);
+  }
 }

ambari-server/src/test/java/org/apache/ambari/server/controller/internal/HttpPropertyProviderTest.java

@@ -193,6 +193,7 @@ public class HttpPropertyProviderTest {
   private static class TestStreamProvider implements StreamProvider {
     private boolean throwError = false;
     private String lastSpec = null;
+    private boolean isLastSpecUpdated;
 
     private TestStreamProvider(boolean throwErr) {
       throwError = throwErr;
@@ -200,7 +201,10 @@ public class HttpPropertyProviderTest {
     
     @Override
     public InputStream readFrom(String spec) throws IOException {
-      lastSpec = spec;
+      if (!isLastSpecUpdated)
+        lastSpec = spec;
+      
+      isLastSpecUpdated = false;
 
       if (throwError) {
         throw new IOException("Fake error");
@@ -214,6 +218,13 @@ public class HttpPropertyProviderTest {
     public String getLastSpec() {
       return lastSpec;
     }
+
+    @Override
+    public InputStream readFrom(String spec, String requestMethod, String params) throws IOException {
+      lastSpec = spec + "?" + params;
+      isLastSpecUpdated = true;
+      return readFrom(spec);
+    }
   }
   
 }

ambari-server/src/test/java/org/apache/ambari/server/controller/internal/StackDefinedPropertyProviderTest.java

@@ -167,6 +167,12 @@ public class StackDefinedPropertyProviderTest {
         return ClassLoader.getSystemResourceAsStream("temporal_ganglia_data.txt");
       }
     }
+
+    @Override
+    public InputStream readFrom(String spec, String requestMethod, String params)
+        throws IOException {
+      return readFrom(spec);
+    }
   }
   
   private static class EmptyPropertyProvider implements PropertyProvider {

ambari-server/src/test/java/org/apache/ambari/server/controller/jmx/TestStreamProvider.java

@@ -47,6 +47,8 @@ public class TestStreamProvider implements StreamProvider {
 
   private String lastSpec;
 
+  private boolean isLastSpecUpdated;
+
   public TestStreamProvider() {
     delay = 0;
   }
@@ -57,7 +59,10 @@ public class TestStreamProvider implements StreamProvider {
 
   @Override
   public InputStream readFrom(String spec) throws IOException {
-    lastSpec = spec;
+    if (!isLastSpecUpdated)
+      lastSpec = spec;
+    
+    isLastSpecUpdated = false;
     String filename = FILE_MAPPING.get(getPort(spec));
     if (filename == null) {
       throw new IOException("Can't find JMX source for " + spec);
@@ -83,4 +88,11 @@ public class TestStreamProvider implements StreamProvider {
 
     return spec.substring(colonIndex + 1, slashIndex);
   }
+
+  @Override
+  public InputStream readFrom(String spec, String requestMethod, String params) throws IOException {
+    lastSpec = spec + "?" + params;
+    isLastSpecUpdated = true;
+    return readFrom(spec);
+  }
 }