Merge lp:~tjoneslo/akiban-server/add-http-session-table into lp:~akiban-technologies/akiban-server/trunk

[Thomas Jones-Low]

Per the demo today, add the ability for the HttpConnections to be seen in the IS.SERVER_SESSIONS table. This allows both the REST and Postgres server connections to be tracked and some accounting performed.

This is still mostly infrastructure. There are some basic statistics capture, but a further review of the business requirements should drive design of more statistics capture.

Also add test to verify this.

[Mike McMahon (mmcm)]

This gives the HTTP server connection a Session, but it isn't otherwise used. Would it be feasible for REST resources, many of which create new sessions each request today, to get access to this one?

review: Needs Information

[Thomas Jones-Low (tjoneslo)]

It is possible to allow the REST processing to use the session created for the HTTP connection.

It requires a chunk of refactoring that I'd prefer to do in a separate commit.

[Thomas Jones-Low (tjoneslo)]

It's possible, but is it ok to use the same Session across multiple threads?

The interaction of the HTTP connection (jetty) and the Servlet processing (Jersey) is via a (theoretical) thread pool. As each request comes across the wire into the connection, it gets packaged into the Request/Response and given to the servlet to process on its own thread. When the Servlet finishes, the the Response it picked up by the connection thread and sent back across the wire.

So the safe thing to do is to create a new session for each Servlet request (i.e. as we do now). Nathan and I had a discussion on the wiki page regarding the REST processing not mapping directly to the Postgres processing, and this is one of those points.

[Mike McMahon (mmcm)]

Okay, seems that at the least more throught is required to do anything there.

review: Approve

=== modified file 'src/main/java/com/akiban/http/HttpConductorImpl.java'
--- src/main/java/com/akiban/http/HttpConductorImpl.java	2013-04-22 22:58:39 +0000
+++ src/main/java/com/akiban/http/HttpConductorImpl.java	2013-04-26 23:40:34 +0000
@@ -22,8 +22,15 @@
 import com.akiban.server.service.monitor.MonitorService;
 import com.akiban.server.service.monitor.ServerMonitor;
 import com.akiban.server.service.security.SecurityService;
-import com.akiban.sql.embedded.EmbeddedJDBCService;
+import com.akiban.server.service.session.Session;
+import com.akiban.server.service.session.SessionService;
+import com.akiban.sql.server.ServerSessionMonitor;
 import com.google.inject.Inject;
+
+import org.eclipse.jetty.io.AsyncEndPoint;
+import org.eclipse.jetty.io.Connection;
+import org.eclipse.jetty.io.nio.AsyncConnection;
+import org.eclipse.jetty.io.nio.SslConnection;
 import org.eclipse.jetty.servlets.CrossOriginFilter;
 import org.eclipse.jetty.util.security.Constraint;
 import org.eclipse.jetty.security.Authenticator;
@@ -32,6 +39,7 @@
 import org.eclipse.jetty.security.LoginService;
 import org.eclipse.jetty.security.authentication.BasicAuthenticator;
 import org.eclipse.jetty.security.authentication.DigestAuthenticator;
+import org.eclipse.jetty.server.AsyncHttpConnection;
 import org.eclipse.jetty.server.Connector;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.server.handler.ContextHandler;
@@ -44,6 +52,8 @@
 
 import javax.servlet.FilterRegistration;
 import javax.servlet.ServletException;
+
+import java.nio.channels.SocketChannel;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Set;
@@ -66,10 +76,12 @@
     private static final String CONFIG_XORIGIN_CREDENTIALS = CONFIG_XORIGIN_PREFIX + "allow_credentials";
 
     private static final String REST_ROLE = "rest-user";
+    public  static final String SERVER_TYPE = "REST";
 
     private final ConfigurationService configurationService;
     private final SecurityService securityService;
     private final MonitorService monitorService;
+    private final SessionService sessionService;
 
     private final Object lock = new Object();
     private SimpleHandlerList handlerList;
@@ -84,10 +96,12 @@
     @Inject
     public HttpConductorImpl(ConfigurationService configurationService,
                              SecurityService securityService,
-                             MonitorService monitor) {
+                             MonitorService monitor,
+                             SessionService session) {
         this.configurationService = configurationService;
         this.securityService = securityService;
         this.monitorService = monitor;
+        this.sessionService = session;
 
         jerseyLogging = java.util.logging.Logger.getLogger("com.sun.jersey");
         jerseyLogging.setLevel(java.util.logging.Level.OFF);
@@ -182,14 +196,14 @@
         Server localServer = new Server();
         SelectChannelConnector connector;
         if (!ssl) {
-            connector = new SelectChannelConnector();
+            connector = new SelectChannelConnectorExtended();
         }
         else {
             // Share keystore configuration with PSQL.
             SslContextFactory sslFactory = new SslContextFactory();
             sslFactory.setKeyStorePath(System.getProperty("javax.net.ssl.keyStore"));
             sslFactory.setKeyStorePassword(System.getProperty("javax.net.ssl.keyStorePassword"));
-            connector = new SslSelectChannelConnector(sslFactory);
+            connector = new SslSelectChannelConnectorExtended(sslFactory);
         }
         connector.setPort(portLocal);
         connector.setThreadPool(new QueuedThreadPool(200));
@@ -262,7 +276,7 @@
     @Override
     public void stop() {
         Server localServer;
-        monitorService.deregisterServerMonitor(monitorService.getServerMonitors().get(ConnectionMonitor.SERVER_TYPE));
+        monitorService.deregisterServerMonitor(monitorService.getServerMonitors().get(SERVER_TYPE));
         synchronized (lock) {
             xOriginFilterEnabled = false;
             localServer = server;
@@ -313,7 +327,6 @@
     }
 
     private class ConnectionMonitor implements ServerMonitor {
-        public static final String SERVER_TYPE = "REST";
         private final SelectChannelConnector connector;
         private final AtomicLong _statsStartedAt = new AtomicLong(System.currentTimeMillis());
         
@@ -341,4 +354,66 @@
             return connector.getConnections();
         }
     }
+    
+    private class SelectChannelConnectorExtended extends SelectChannelConnector {
+        private Session session;
+        @Override
+        protected AsyncConnection newConnection(SocketChannel channel,final AsyncEndPoint endpoint)
+        {
+            AsyncHttpConnection conn = (AsyncHttpConnection)super.newConnection(channel, endpoint);
+            ServerSessionMonitor sessionMonitor = new ServerSessionMonitor(SERVER_TYPE,
+                    monitorService.allocateSessionId());
+
+            conn.setAssociatedObject(sessionMonitor);
+            this.session = sessionService.createSession();
+            monitorService.registerSessionMonitor(sessionMonitor, session);
+            return conn;
+        }
+        
+        @Override  
+        protected void connectionClosed(Connection connection) {
+            if (connection instanceof AsyncHttpConnection) {
+                AsyncHttpConnection conn = (AsyncHttpConnection)connection;
+                ServerSessionMonitor monitor = (ServerSessionMonitor)conn.getAssociatedObject();
+                if (monitor != null) {
+                    monitorService.deregisterSessionMonitor(monitor, session);
+                    conn.setAssociatedObject(null);
+                }
+            }
+            super.connectionClosed(connection);
+        }
+    }
+    
+    private class SslSelectChannelConnectorExtended extends SslSelectChannelConnector {
+        private Session session;
+        public SslSelectChannelConnectorExtended(SslContextFactory sslFactory) {
+            super(sslFactory);
+        }
+
+        @Override
+        protected AsyncConnection newConnection(SocketChannel channel,final AsyncEndPoint endpoint)
+        {
+            AsyncHttpConnection conn = (AsyncHttpConnection)((SslConnection)super.newConnection(channel, endpoint)).getSslEndPoint().getConnection();
+            ServerSessionMonitor sessionMonitor = new ServerSessionMonitor(SERVER_TYPE,
+                    monitorService.allocateSessionId());
+
+            conn.setAssociatedObject(sessionMonitor);
+            this.session = sessionService.createSession();
+            monitorService.registerSessionMonitor(sessionMonitor, session);
+            return conn;
+        }
+        
+        @Override  
+        protected void connectionClosed (Connection connection) {
+            if (connection instanceof SslConnection) {
+                AsyncHttpConnection conn = (AsyncHttpConnection)((SslConnection) connection).getSslEndPoint().getConnection();
+                ServerSessionMonitor monitor = (ServerSessionMonitor)conn.getAssociatedObject();
+                if (monitor != null) {
+                    monitorService.deregisterSessionMonitor(monitor, session);
+                    conn.setAssociatedObject(null);
+                }
+            }
+            super.connectionClosed(connection);
+        }
+    }
 }

=== modified file 'src/main/java/com/akiban/http/SimpleHandlerList.java'
--- src/main/java/com/akiban/http/SimpleHandlerList.java	2013-04-22 22:37:46 +0000
+++ src/main/java/com/akiban/http/SimpleHandlerList.java	2013-04-26 23:40:34 +0000
@@ -22,6 +22,8 @@
 import org.eclipse.jetty.server.handler.AbstractHandler;
 import org.eclipse.jetty.util.MultiException;
 
+import com.akiban.server.service.monitor.MonitorStage;
+import com.akiban.sql.server.ServerSessionMonitor;
 import com.akiban.util.tap.InOutTap;
 import com.akiban.util.tap.Tap;
 
@@ -53,10 +55,18 @@
                        Request baseRequest,
                        HttpServletRequest request,
                        HttpServletResponse response) throws IOException, ServletException {
+        ServerSessionMonitor monitor = null;
         if(!isStarted()) {
             return;
         }
         REST_TAP.in();
+        Object obj = baseRequest.getConnection().getAssociatedObject();
+        if (obj instanceof ServerSessionMonitor) {
+            monitor = (ServerSessionMonitor)obj;
+            monitor.setRemoteAddress(request.getRemoteAddr());
+            monitor.startStatement(request.getMethod(), request.getRequestURI());
+            monitor.enterStage(MonitorStage.EXECUTE);
+        }
         try {
             for(Handler h : handlers) {
                 h.handle(target,baseRequest, request, response);
@@ -70,6 +80,10 @@
             }
         }finally {
             REST_TAP.out();
+            if (monitor != null) {
+                monitor.leaveStage();
+                monitor.endStatement(1);
+            }
         }
     }
 

=== added file 'src/test/java/com/akiban/http/HttpMonitorVerifyIT.java'
--- src/test/java/com/akiban/http/HttpMonitorVerifyIT.java	1970-01-01 00:00:00 +0000
+++ src/test/java/com/akiban/http/HttpMonitorVerifyIT.java	2013-04-26 23:40:34 +0000
@@ -0,0 +1,106 @@
+/**
+ * Copyright (C) 2009-2013 Akiban Technologies, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package com.akiban.http;
+
+import static org.junit.Assert.assertEquals;
+
+import java.net.URI;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.http.HttpResponse;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.impl.client.DefaultHttpClient;
+import org.apache.http.util.EntityUtils;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.akiban.rest.RestService;
+import com.akiban.rest.RestServiceImpl;
+import com.akiban.server.service.monitor.MonitorService;
+import com.akiban.server.service.security.SecurityService;
+import com.akiban.server.service.security.SecurityServiceImpl;
+import com.akiban.server.service.servicemanager.GuicedServiceManager;
+import com.akiban.server.test.it.ITBase;
+import com.akiban.sql.embedded.EmbeddedJDBCService;
+import com.akiban.sql.embedded.EmbeddedJDBCServiceImpl;
+
+public class HttpMonitorVerifyIT extends ITBase {
+    
+    private static final Logger LOG = LoggerFactory.getLogger(HttpMonitorVerifyIT.class);
+
+    @Override
+    protected GuicedServiceManager.BindingsConfigurationProvider serviceBindingsProvider() {
+        return super.serviceBindingsProvider()
+            .bindAndRequire(SecurityService.class, SecurityServiceImpl.class)
+            .bindAndRequire(EmbeddedJDBCService.class, EmbeddedJDBCServiceImpl.class)
+            .bindAndRequire(RestService.class, RestServiceImpl.class);
+    }
+
+    @Before
+    public void setUp() {
+        SecurityService securityService = securityService();
+        securityService.addRole("rest-user");
+        securityService.addUser("user1", "password", Arrays.asList("rest-user"));
+    }
+
+    protected SecurityService securityService() {
+        return serviceManager().getServiceByClass(SecurityService.class);
+    }
+    
+    protected HttpConductor httpConductor() {
+        return serviceManager().getServiceByClass(HttpConductor.class);
+    }
+    
+    protected MonitorService monitorService () {
+        return serviceManager().getServiceByClass(MonitorService.class);
+    }
+
+    @Override
+    protected Map<String, String> startupConfigProperties() {
+        Map<String, String> properties = new HashMap<>();
+        properties.put("akserver.http.login", "basic"); // "digest"
+        properties.put("akserver.restrict_user_schema", "true");
+        return properties;
+    }
+
+    private static int openRestURL(HttpClient client, String userInfo, int port, String path) throws Exception {
+        URI uri = new URI("http", userInfo, "localhost", port, path, null, null);
+        HttpGet get = new HttpGet(uri);
+        HttpResponse response = client.execute(get);
+        int code = response.getStatusLine().getStatusCode();
+        EntityUtils.consume(response.getEntity());
+        return code;
+    }
+
+    @Test
+    public void runTest () throws Exception {
+        MonitorService monitor = monitorService();
+        
+        HttpClient client = new DefaultHttpClient();
+        openRestURL(client, "user1:password", httpConductor().getPort(), "/version");
+        
+        assertEquals(monitor.getSessionMonitors().size(), 1);
+        
+        client.getConnectionManager().shutdown();
+    }
+
+}

=== added file 'src/test/java/com/akiban/http/HttpMonitorVerifySSLIT.java'
--- src/test/java/com/akiban/http/HttpMonitorVerifySSLIT.java	1970-01-01 00:00:00 +0000
+++ src/test/java/com/akiban/http/HttpMonitorVerifySSLIT.java	2013-04-26 23:40:34 +0000
@@ -0,0 +1,199 @@
+/**
+ * Copyright (C) 2009-2013 Akiban Technologies, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package com.akiban.http;
+
+import static org.junit.Assert.assertEquals;
+
+import java.net.URI;
+import java.security.cert.CertificateException;
+import java.security.cert.X509Certificate;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.X509TrustManager;
+
+import org.apache.http.HttpResponse;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.conn.ClientConnectionManager;
+import org.apache.http.conn.scheme.Scheme;
+import org.apache.http.conn.scheme.SchemeRegistry;
+import org.apache.http.conn.ssl.SSLSocketFactory;
+import org.apache.http.impl.client.DefaultHttpClient;
+import org.apache.http.util.EntityUtils;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.akiban.rest.RestService;
+import com.akiban.rest.RestServiceImpl;
+import com.akiban.server.service.monitor.MonitorService;
+import com.akiban.server.service.security.SecurityService;
+import com.akiban.server.service.security.SecurityServiceImpl;
+import com.akiban.server.service.servicemanager.GuicedServiceManager;
+import com.akiban.server.test.it.ITBase;
+import com.akiban.sql.embedded.EmbeddedJDBCService;
+import com.akiban.sql.embedded.EmbeddedJDBCServiceImpl;
+
+/**
+ * In order to run this test, you need to generate a key store, then update the
+ * javax.net.ssl.keyStore system property (set below), to the path where you put the 
+ * keystore file.
+ * <pre> 
+ *  $ keytool -keystore keystore -alias akiban -genkey -keyalg RSA 
+ *  Enter keystore password: password  
+ *  Re-enter new password: password 
+ *  What is your first and last name?
+ *    [Unknown]:  akiban.com
+ *  What is the name of your organizational unit?
+ *    [Unknown]:  akiban.com
+ *  What is the name of your organization?
+ *    [Unknown]:  akiban
+ *  What is the name of your City or Locality?
+ *    [Unknown]:  
+ *  What is the name of your State or Province?
+ *    [Unknown]:  
+ *  What is the two-letter country code for this unit?
+ *    [Unknown]:  
+ *  Is CN=akiban.com, OU=akiban.com, O=akiban, L=Unknown, ST=Unknown, C=Unknown correct?
+ *    [no]:  yes
+ * </pre>   
+ * Because we don't want to be shipping a half completed SSL certificate with the source 
+ * code, This is a manual step required for this (otherwise) disabled test. 
+ * @author tjoneslo
+ *
+ */
+public class HttpMonitorVerifySSLIT extends ITBase {
+    private static final Logger LOG = LoggerFactory.getLogger(HttpMonitorVerifySSLIT.class);
+
+    @Override
+    protected GuicedServiceManager.BindingsConfigurationProvider serviceBindingsProvider() {
+        return super.serviceBindingsProvider()
+            .bindAndRequire(SecurityService.class, SecurityServiceImpl.class)
+            .bindAndRequire(EmbeddedJDBCService.class, EmbeddedJDBCServiceImpl.class)
+            .bindAndRequire(RestService.class, RestServiceImpl.class);
+    }
+
+    @Before
+    public void setUp() {
+        SecurityService securityService = securityService();
+        securityService.addRole("rest-user");
+        securityService.addRole("admin");
+        securityService.addUser("user1", "password", Arrays.asList("rest-user"));
+        securityService.addUser("akiban", "topsecret", Arrays.asList("rest-user", "admin"));
+    }
+
+    protected SecurityService securityService() {
+        return serviceManager().getServiceByClass(SecurityService.class);
+    }
+    
+    protected HttpConductor httpConductor() {
+        return serviceManager().getServiceByClass(HttpConductor.class);
+    }
+    
+    protected MonitorService monitorService () {
+        return serviceManager().getServiceByClass(MonitorService.class);
+    }
+
+    @Override
+    protected Map<String, String> startupConfigProperties() {
+        Map<String, String> properties = new HashMap<>();
+        properties.put("akserver.http.login", "digest"); // "digest"
+        properties.put("akserver.http.ssl", "true");
+        properties.put("akserver.restrict_user_schema", "true");
+        
+        Properties p = new Properties(System.getProperties());
+        p.put("javax.net.ssl.keyStore", "./keystore");
+        p.put("javax.net.ssl.keyStorePassword", "password");
+        System.setProperties(p);
+        return properties;
+    }
+
+    private static int openRestURL(HttpClient client, String userInfo, int port, String path) throws Exception {
+        URI uri = new URI("https", userInfo, "localhost", port, path, null, null);
+        HttpGet get = new HttpGet(uri);
+        HttpResponse response = client.execute(get);
+        int code = response.getStatusLine().getStatusCode();
+        EntityUtils.consume(response.getEntity());
+        return code;
+    }
+
+    @Ignore ("need setup")
+    @Test
+    public void runTest () throws Exception {
+        MonitorService monitor = monitorService();
+        
+        HttpClient client = new DefaultHttpClient();
+        client = wrapClient(client);
+        
+        openRestURL(client, "user1:password", httpConductor().getPort(), "/version");
+        
+        assertEquals(monitor.getSessionMonitors().size(), 1);
+        
+        client.getConnectionManager().shutdown();
+    }
+
+ 
+    /**
+     * This code sets up the httpclient to accept any SSL certificate. The 
+     * SSL certificate generated by the instructions above is not correctly
+     * signed, so we need ignore the problem. 
+     * This code should not, under any circumstances, be allowed anywhere 
+     * the production code. 
+     * @param base
+     * @return
+     */
+    private HttpClient wrapClient (HttpClient base) {
+        try {
+            SSLContext ctx = SSLContext.getInstance("TLS");
+    
+            ctx.init(null, new TrustManager[]{getTrustManager()}, null);
+            SSLSocketFactory ssf = new SSLSocketFactory(ctx);
+            ssf.setHostnameVerifier(SSLSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER);
+            ClientConnectionManager ccm = base.getConnectionManager();
+            SchemeRegistry sr = ccm.getSchemeRegistry();
+            sr.register(new Scheme("https", ssf, 8091));
+            return new DefaultHttpClient(ccm, base.getParams());
+        } catch (Exception ex) {
+            ex.printStackTrace();
+            return null;
+        }
+    }
+
+    private X509TrustManager getTrustManager() {
+        return new X509TrustManager() {
+        
+        public X509Certificate[] getAcceptedIssuers() {
+        return null;
+        }
+
+        @Override
+        public void checkClientTrusted(X509Certificate[] arg0, String arg1)
+                throws CertificateException { }
+
+        @Override
+        public void checkServerTrusted(X509Certificate[] arg0, String arg1)
+                throws CertificateException {  }
+        };
+    }
+}

=== modified file 'src/test/java/com/akiban/http/HttpThreadedLoginIT.java'
--- src/test/java/com/akiban/http/HttpThreadedLoginIT.java	2013-04-22 22:51:14 +0000
+++ src/test/java/com/akiban/http/HttpThreadedLoginIT.java	2013-04-26 23:40:34 +0000
@@ -21,8 +21,6 @@
 import com.akiban.rest.RestServiceImpl;
 import com.akiban.server.service.servicemanager.GuicedServiceManager;
 import com.akiban.server.test.it.ITBase;
-import com.akiban.sql.embedded.EmbeddedJDBCService;
-import com.akiban.sql.embedded.EmbeddedJDBCServiceImpl;
 
 import org.apache.http.HttpResponse;
 import org.apache.http.HttpStatus;