AMBARI-11678 Perf: Add GZIP support for Ambari Server API (dsen)

ambari-agent/src/main/python/ambari_agent/Controller.py

@@ -393,7 +393,8 @@ class Controller(threading.Thread):
     try:
       if self.cachedconnect is None: # Lazy initialization
         self.cachedconnect = security.CachedHTTPSConnection(self.config)
-      req = urllib2.Request(url, data, {'Content-Type': 'application/json'})
+      req = urllib2.Request(url, data, {'Content-Type': 'application/json',
+                                        'Accept-encoding': 'gzip'})
       response = self.cachedconnect.request(req)
       return json.loads(response)
     except Exception, exception:

ambari-agent/src/main/python/ambari_agent/security.py

@@ -15,7 +15,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-
+from StringIO import StringIO
+import gzip
 import httplib
 import urllib2
 import socket
@@ -31,8 +32,9 @@ import platform
 
 logger = logging.getLogger(__name__)
 
-GEN_AGENT_KEY = 'openssl req -new -newkey rsa:1024 -nodes -keyout "%(keysdir)s'+os.sep+'%(hostname)s.key" '\
-	'-subj /OU=%(hostname)s/ -out "%(keysdir)s'+os.sep+'%(hostname)s.csr"'
+GEN_AGENT_KEY = 'openssl req -new -newkey rsa:1024 -nodes -keyout "%(keysdir)s' \
+                + os.sep + '%(hostname)s.key" -subj /OU=%(hostname)s/ ' \
+                '-out "%(keysdir)s' + os.sep + '%(hostname)s.csr"'
 
 
 class VerifiedHTTPSConnection(httplib.HTTPSConnection):
@@ -44,9 +46,11 @@ class VerifiedHTTPSConnection(httplib.HTTPSConnection):
 
   def connect(self):
     self.two_way_ssl_required = self.config.isTwoWaySSLConnection()
-    logger.debug("Server two-way SSL authentication required: %s" % str(self.two_way_ssl_required))
+    logger.debug("Server two-way SSL authentication required: %s" % str(
+      self.two_way_ssl_required))
     if self.two_way_ssl_required is True:
-      logger.info('Server require two-way SSL authentication. Use it instead of one-way...')
+      logger.info(
+        'Server require two-way SSL authentication. Use it instead of one-way...')
 
     if not self.two_way_ssl_required:
       try:
@@ -56,8 +60,9 @@ class VerifiedHTTPSConnection(httplib.HTTPSConnection):
                     'turned off on the server.')
       except (ssl.SSLError, AttributeError):
         self.two_way_ssl_required = True
-        logger.info('Insecure connection to https://' + self.host + ':' + self.port +
-                    '/ failed. Reconnecting using two-way SSL authentication..')
+        logger.info(
+          'Insecure connection to https://' + self.host + ':' + self.port +
+          '/ failed. Reconnecting using two-way SSL authentication..')
 
     if self.two_way_ssl_required:
       self.certMan = CertificateManager(self.config)
@@ -70,21 +75,21 @@ class VerifiedHTTPSConnection(httplib.HTTPSConnection):
 
       try:
         self.sock = ssl.wrap_socket(sock,
-                                keyfile=agent_key,
-                                certfile=agent_crt,
-                                cert_reqs=ssl.CERT_REQUIRED,
-                                ca_certs=server_crt)
+                                    keyfile=agent_key,
+                                    certfile=agent_crt,
+                                    cert_reqs=ssl.CERT_REQUIRED,
+                                    ca_certs=server_crt)
         logger.info('SSL connection established. Two-way SSL authentication '
                     'completed successfully.')
       except ssl.SSLError as err:
         logger.error('Two-way SSL authentication failed. Ensure that '
-                    'server and agent certificates were signed by the same CA '
-                    'and restart the agent. '
-                    '\nIn order to receive a new agent certificate, remove '
-                    'existing certificate file from keys directory. As a '
-                    'workaround you can turn off two-way SSL authentication in '
-                    'server configuration(ambari.properties) '
-                    '\nExiting..')
+                     'server and agent certificates were signed by the same CA '
+                     'and restart the agent. '
+                     '\nIn order to receive a new agent certificate, remove '
+                     'existing certificate file from keys directory. As a '
+                     'workaround you can turn off two-way SSL authentication in '
+                     'server configuration(ambari.properties) '
+                     '\nExiting..')
         raise err
 
   def create_connection(self):
@@ -112,13 +117,15 @@ class CachedHTTPSConnection:
 
   def connect(self):
     if not self.connected:
-      self.httpsconn = VerifiedHTTPSConnection(self.server, self.port, self.config)
+      self.httpsconn = VerifiedHTTPSConnection(self.server, self.port,
+                                               self.config)
       self.httpsconn.connect()
       self.connected = True
     # possible exceptions are caught and processed in Controller
 
   def forceClear(self):
-    self.httpsconn = VerifiedHTTPSConnection(self.server, self.port, self.config)
+    self.httpsconn = VerifiedHTTPSConnection(self.server, self.port,
+                                             self.config)
     self.connect()
 
   def request(self, req):
@@ -127,6 +134,10 @@ class CachedHTTPSConnection:
       self.httpsconn.request(req.get_method(), req.get_full_url(),
                              req.get_data(), req.headers)
       response = self.httpsconn.getresponse()
+      # Ungzip if gzipped
+      if response.getheader('Content-Encoding') == 'gzip':
+        buf = StringIO(response.read())
+        response = gzip.GzipFile(fileobj=buf)
       readResponse = response.read()
     except Exception as ex:
       # This exception is caught later in Controller
@@ -144,7 +155,7 @@ class CertificateManager():
     self.keysdir = os.path.abspath(self.config.get('security', 'keysdir'))
     self.server_crt = self.config.get('security', 'server_crt')
     self.server_url = 'https://' + hostname.server_hostname(config) + ':' \
-       + self.config.get('server', 'url_port')
+                      + self.config.get('server', 'url_port')
 
   def getAgentKeyName(self):
     keysdir = os.path.abspath(self.config.get('security', 'keysdir'))
@@ -164,7 +175,8 @@ class CertificateManager():
 
   def checkCertExists(self):
 
-    s = os.path.abspath(self.config.get('security', 'keysdir')) + os.sep + "ca.crt"
+    s = os.path.abspath(
+      self.config.get('security', 'keysdir')) + os.sep + "ca.crt"
 
     server_crt_exists = os.path.exists(s)
 
@@ -202,18 +214,20 @@ class CertificateManager():
     srvr_crt_f.write(response)
 
   def reqSignCrt(self):
-    sign_crt_req_url = self.server_url + '/certs/' + hostname.hostname(self.config)
+    sign_crt_req_url = self.server_url + '/certs/' + hostname.hostname(
+      self.config)
     agent_crt_req_f = open(self.getAgentCrtReqName())
     agent_crt_req_content = agent_crt_req_f.read()
     passphrase_env_var = self.config.get('security', 'passphrase_env_var_name')
     passphrase = os.environ[passphrase_env_var]
     register_data = {'csr': agent_crt_req_content,
-                    'passphrase': passphrase}
+                     'passphrase': passphrase}
     data = json.dumps(register_data)
     proxy_handler = urllib2.ProxyHandler({})
     opener = urllib2.build_opener(proxy_handler)
     urllib2.install_opener(opener)
-    req = urllib2.Request(sign_crt_req_url, data, {'Content-Type': 'application/json'})
+    req = urllib2.Request(sign_crt_req_url, data,
+                          {'Content-Type': 'application/json'})
     f = urllib2.urlopen(req)
     response = f.read()
     f.close()
@@ -239,14 +253,16 @@ class CertificateManager():
       raise ssl.SSLError
 
   def genAgentCrtReq(self):
-    generate_script = GEN_AGENT_KEY % {'hostname': hostname.hostname(self.config),
-                                     'keysdir' : os.path.abspath(self.config.get('security', 'keysdir'))}
+    generate_script = GEN_AGENT_KEY % {
+      'hostname': hostname.hostname(self.config),
+      'keysdir': os.path.abspath(self.config.get('security', 'keysdir'))}
     logger.info(generate_script)
     if platform.system() == 'Windows':
       p = subprocess.Popen(generate_script, stdout=subprocess.PIPE)
       p.communicate()
     else:
-      p = subprocess.Popen([generate_script], shell=True, stdout=subprocess.PIPE)
+      p = subprocess.Popen([generate_script], shell=True,
+                           stdout=subprocess.PIPE)
       p.communicate()
 
   def initSecurity(self):

ambari-project/pom.xml

@@ -246,6 +246,11 @@
         <artifactId>jetty-servlet</artifactId>
         <version>8.1.17.v20150415</version>
       </dependency>
+      <dependency>
+        <groupId>org.eclipse.jetty</groupId>
+        <artifactId>jetty-servlets</artifactId>
+        <version>8.1.17.v20150415</version>
+      </dependency>
       <dependency>
         <groupId>org.eclipse.jetty</groupId>
         <artifactId>jetty-webapp</artifactId>

ambari-server/pom.xml

@@ -1592,6 +1592,10 @@
       <groupId>org.eclipse.jetty</groupId>
       <artifactId>jetty-servlet</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.eclipse.jetty</groupId>
+      <artifactId>jetty-servlets</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.eclipse.jetty</groupId>
       <artifactId>jetty-webapp</artifactId>

ambari-server/src/main/java/org/apache/ambari/server/configuration/Configuration.java

@@ -84,6 +84,9 @@ public class Configuration {
   public static final String API_AUTHENTICATE = "api.authenticate";
   public static final String API_USE_SSL = "api.ssl";
   public static final String API_CSRF_PREVENTION_KEY = "api.csrfPrevention.enabled";
+  public static final String API_GZIP_COMPRESSION_ENABLED_KEY = "api.gzip.compression.enabled";
+  public static final String API_GZIP_MIN_COMPRESSION_SIZE_KEY = "api.gzip.compression.min.size";
+  public static final String AGENT_API_GZIP_COMPRESSION_ENABLED_KEY = "agent.api.gzip.compression.enabled";
   public static final String SRVR_TWO_WAY_SSL_KEY = "security.server.two_way_ssl";
   public static final String SRVR_TWO_WAY_SSL_PORT_KEY = "security.server.two_way_ssl.port";
   public static final String SRVR_ONE_WAY_SSL_PORT_KEY = "security.server.one_way_ssl.port";
@@ -293,6 +296,8 @@ public class Configuration {
   private static final String SRVR_TWO_WAY_SSL_DEFAULT = "false";
   private static final String SRVR_KSTR_DIR_DEFAULT = ".";
   private static final String API_CSRF_PREVENTION_DEFAULT = "true";
+  private static final String API_GZIP_COMPRESSION_ENABLED_DEFAULT = "true";
+  private static final String API_GZIP_MIN_COMPRESSION_SIZE_DEFAULT = "10240";
   private static final String SRVR_CRT_PASS_FILE_DEFAULT = "pass.txt";
   private static final String SRVR_CRT_PASS_LEN_DEFAULT = "50";
   private static final String SRVR_DISABLED_CIPHERS_DEFAULT = "";
@@ -892,7 +897,8 @@ public class Configuration {
    * @return int
    */
   public int getClientSSLApiPort() {
-    return Integer.parseInt(properties.getProperty(CLIENT_API_SSL_PORT_KEY, String.valueOf(CLIENT_API_SSL_PORT_DEFAULT)));
+    return Integer.parseInt(properties.getProperty(CLIENT_API_SSL_PORT_KEY,
+        String.valueOf(CLIENT_API_SSL_PORT_DEFAULT)));
   }
 
   /**
@@ -914,6 +920,37 @@ public class Configuration {
       SRVR_TWO_WAY_SSL_DEFAULT)));
   }
 
+  /**
+   * Check to see if the API responses should be compressed via gzip or not
+   * @return false if not, true if gzip compression needs to be used.
+   */
+  public boolean isApiGzipped() {
+    return "true".equalsIgnoreCase(properties.getProperty(
+        API_GZIP_COMPRESSION_ENABLED_KEY,
+        API_GZIP_COMPRESSION_ENABLED_DEFAULT));
+  }
+
+  /**
+   * Check to see if the agent API responses should be compressed via gzip or not
+   * @return false if not, true if gzip compression needs to be used.
+   */
+  public boolean isAgentApiGzipped() {
+    return "true".equalsIgnoreCase(properties.getProperty(
+        AGENT_API_GZIP_COMPRESSION_ENABLED_KEY,
+        API_GZIP_COMPRESSION_ENABLED_DEFAULT));
+  }
+
+  /**
+   * Check to see if the API responses should be compressed via gzip or not
+   * Content will only be compressed if content length is either unknown or
+   * greater this value
+   * @return false if not, true if ssl needs to be used.
+   */
+  public String getApiGzipMinSize() {
+    return properties.getProperty(API_GZIP_MIN_COMPRESSION_SIZE_KEY,
+        API_GZIP_MIN_COMPRESSION_SIZE_DEFAULT);
+  }
+
   /**
    * Check persistence type Ambari Server should use. Possible values:
    * in-memory - use in-memory Derby database to store data

ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariServer.java

@@ -107,6 +107,7 @@ import org.eclipse.jetty.server.ssl.SslSelectChannelConnector;
 import org.eclipse.jetty.servlet.DefaultServlet;
 import org.eclipse.jetty.servlet.FilterHolder;
 import org.eclipse.jetty.servlet.ServletContextHandler;
+import org.eclipse.jetty.servlets.GzipFilter;
 import org.eclipse.jetty.servlet.ServletHolder;
 import org.eclipse.jetty.util.ssl.SslContextFactory;
 import org.eclipse.jetty.util.thread.QueuedThreadPool;
@@ -274,6 +275,9 @@ public class AmbariServer {
       // and does not use sessions.
       ServletContextHandler agentroot = new ServletContextHandler(
           serverForAgent, "/", ServletContextHandler.NO_SESSIONS);
+      if (configs.isAgentApiGzipped()) {
+        configureHandlerCompression(agentroot);
+      }
 
       ServletHolder rootServlet = root.addServlet(DefaultServlet.class, "/");
       rootServlet.setInitParameter("dirAllowed", "false");
@@ -529,12 +533,13 @@ public class AmbariServer {
   }
 
   /**
-   * Performs basic configuration of root handler with static values and values from
-   * configuration file.
+   * Performs basic configuration of root handler with static values and values
+   * from configuration file.
    *
    * @param root root handler
    */
   protected void configureRootHandler(ServletContextHandler root) {
+    configureHandlerCompression(root);
     root.setContextPath(CONTEXT_PATH);
     root.setErrorHandler(injector.getInstance(AmbariErrorHandler.class));
     root.setMaxFormContentSize(-1);
@@ -543,6 +548,26 @@ public class AmbariServer {
     root.setResourceBase(configs.getWebAppDir());
   }
 
+  /**
+   * Performs GZIP compression configuration of the context handler
+   * with static values and values from configuration file
+   *
+   * @param context handler
+   */
+  protected void configureHandlerCompression(ServletContextHandler context) {
+    if (configs.isApiGzipped()) {
+      FilterHolder gzipFilter = context.addFilter(GzipFilter.class, "/*",
+          EnumSet.of(DispatcherType.REQUEST));
+
+      gzipFilter.setInitParameter("methods","GET,POST,PUT,DELETE");
+      gzipFilter.setInitParameter("mimeTypes",
+          "text/html,text/plain,text/xml,text/css,application/x-javascript," +
+          "application/xml,application/x-www-form-urlencoded," +
+          "application/javascript,application/json");
+      gzipFilter.setInitParameter("minGzipSize", configs.getApiGzipMinSize());
+    }
+  }
+
   /**
    * Performs basic configuration of session manager with static values and values from
    * configuration file.

ambari-server/src/test/java/org/apache/ambari/server/controller/AmbariServerTest.java

@@ -18,6 +18,7 @@
 
 package org.apache.ambari.server.controller;
 
+import static org.easymock.EasyMock.anyObject;
 import static org.easymock.EasyMock.createNiceMock;
 import static org.easymock.EasyMock.expect;
 import static org.easymock.EasyMock.replay;
@@ -26,6 +27,7 @@ import static org.easymock.EasyMock.verify;
 import java.net.Authenticator;
 import java.net.InetAddress;
 import java.net.PasswordAuthentication;
+import java.util.EnumSet;
 
 import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.configuration.Configuration;
@@ -34,7 +36,9 @@ import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
 import org.apache.velocity.app.Velocity;
 import org.easymock.EasyMock;
 import org.eclipse.jetty.server.SessionManager;
+import org.eclipse.jetty.servlet.FilterHolder;
 import org.eclipse.jetty.servlet.ServletContextHandler;
+import org.eclipse.jetty.servlets.GzipFilter;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -43,6 +47,7 @@ import org.junit.Test;
 import com.google.inject.Guice;
 import com.google.inject.Injector;
 
+import javax.servlet.DispatcherType;
 import javax.servlet.SessionCookieConfig;
 
 public class AmbariServerTest {
@@ -119,15 +124,36 @@ public class AmbariServerTest {
 
   @Test
   public void testConfigureRootHandler() throws Exception {
-    final ServletContextHandler handler = EasyMock.createNiceMock(ServletContextHandler.class);
+    final ServletContextHandler handler =
+        EasyMock.createNiceMock(ServletContextHandler.class);
+    final FilterHolder filter = EasyMock.createNiceMock(FilterHolder.class);
 
     handler.setMaxFormContentSize(-1);
     EasyMock.expectLastCall().once();
-    replay(handler);
+    EasyMock.expect(handler.addFilter(GzipFilter.class, "/*",
+        EnumSet.of(DispatcherType.REQUEST))).andReturn(filter).once();
+    replay(handler, filter);
 
     injector.getInstance(AmbariServer.class).configureRootHandler(handler);
 
     EasyMock.verify(handler);
   }
 
+  @Test
+  public void testConfigureCompression() throws Exception {
+    final ServletContextHandler handler =
+        EasyMock.createNiceMock(ServletContextHandler.class);
+    final FilterHolder filter = EasyMock.createNiceMock(FilterHolder.class);
+
+    EasyMock.expect(handler.addFilter(GzipFilter.class, "/*",
+        EnumSet.of(DispatcherType.REQUEST))).andReturn(filter).once();
+    filter.setInitParameter(anyObject(String.class),anyObject(String.class));
+    EasyMock.expectLastCall().times(3);
+    replay(handler, filter);
+
+    injector.getInstance(AmbariServer.class).configureHandlerCompression(handler);
+
+    EasyMock.verify(handler);
+  }
+
 }