Merge lp:~tjoneslo/akiban-server/add-user-monitor into lp:~akiban-technologies/akiban-server/trunk

[Thomas Jones-Low]

Add a UserMonitor with IS.SERVER_USERS to show how much work each user is doing.

This creates a UserMonitor which is registered with the MontitorService and attached to the SessionMonitor when the user logs in. Currently this only works on the Postgres and InternalJDBC connections. If not using password or SSL security (i.e, security is none) there are no UserMonitors created.

This is mostly infrastructure. The only thing the UserMonitor counts currently is number of SQL statements executed. This can easily be extended, and will be at a later time.

This includes a new I_S.SERVER_USERS table to display the data gathered by the UserMonitor to the user. This table is sensitive to the restricted access security that the SERVER_SESSIONS (and other) tables are.

[Nathan Williams (nwilliams)]

Tiny nit. Diff line 170 can just remove, instead of contains and remove. The string based contract being looser (i.e. not asserting) is fine though.

I'm not sure we'll want to blow away the monitor, diff line 322, if a user is deleted in general. Long term random guess: they still need billed. But I'm hand waving and we can always adjust as we figure out how we want to use it.

review: Needs Fixing

[Thomas Jones-Low (tjoneslo)]

There's a whole chunk of how we capture the session monitor (including the user monitor) data when the user logs out that needs to be captured before the logout is completed. Or the server is restarted. These are known issues that need to be addressed.

I've made the deregister cleaner per your request.

review: Needs Resubmitting

[Nathan Williams (nwilliams)]

Looks good.

review: Approve

=== modified file 'src/main/java/com/akiban/server/service/is/ServerSchemaTablesServiceImpl.java'
--- src/main/java/com/akiban/server/service/is/ServerSchemaTablesServiceImpl.java	2013-04-23 22:45:16 +0000
+++ src/main/java/com/akiban/server/service/is/ServerSchemaTablesServiceImpl.java	2013-04-26 00:32:25 +0000
@@ -44,6 +44,7 @@
 import com.akiban.server.service.monitor.PreparedStatementMonitor;
 import com.akiban.server.service.monitor.ServerMonitor;
 import com.akiban.server.service.monitor.SessionMonitor;
+import com.akiban.server.service.monitor.UserMonitor;
 import com.akiban.server.service.security.SecurityService;
 import com.akiban.server.service.session.Session;
 import com.akiban.server.store.SchemaManager;
@@ -67,6 +68,7 @@
     static final TableName SERVER_TAPS = new TableName (SCHEMA_NAME, "server_taps");
     static final TableName SERVER_PREPARED_STATEMENTS = new TableName (SCHEMA_NAME, "server_prepared_statements");
     static final TableName SERVER_CURSORS = new TableName (SCHEMA_NAME, "server_cursors");
+    static final TableName SERVER_USERS = new TableName (SCHEMA_NAME, "server_users");
 
     private final MonitorService monitor;
     private final ConfigurationService configService;
@@ -109,6 +111,8 @@
         attach (ais, true, SERVER_PREPARED_STATEMENTS, PreparedStatements.class);
         //SERVER_CURSORS
         attach (ais, true, SERVER_CURSORS, Cursors.class);
+        //SERVER_USERS
+        attach(ais, true, SERVER_USERS, Users.class);
     }
 
     @Override
@@ -136,6 +140,18 @@
         }
     }
 
+    protected Collection<UserMonitor> getAccessibleUsers (Session session) {
+        if (securityService.hasRestrictedAccess(session)) {
+            return monitor.getUserMonitors();
+        } else {
+            UserMonitor um = monitor.getUserMonitor(session);
+            if (um == null) {
+                return Collections.emptyList();
+            } else {
+                return Collections.singletonList(um);
+            }
+        }
+    }
     private class InstanceSummary extends BasicFactoryBase {
 
         public InstanceSummary(TableName sourceTable) {
@@ -565,6 +581,46 @@
         }
     }
 
+    private class Users extends BasicFactoryBase {
+
+        public Users(TableName sourceTable) {
+            super(sourceTable);
+        }
+
+        @Override
+        public GroupScan getGroupScan(MemoryAdapter adapter) {
+            return new Scan (adapter.getSession(), getRowType(adapter));
+        }
+
+        @Override
+        public long rowCount() {
+            return monitor.getUserMonitors().size();
+        }
+        
+        private class Scan extends BaseScan {
+            final Iterator<UserMonitor> users;
+
+            public Scan(Session session, RowType rowType) {
+                super(rowType);
+                users = getAccessibleUsers(session).iterator();
+            }
+
+            @Override
+            public Row next() {
+                if (!users.hasNext()) {
+                    return null;
+                }
+                UserMonitor user = users.next();
+                ValuesRow row = new ValuesRow (rowType,
+                                            user.getUserName(),
+                                            user.getStatementCount(),
+                                            ++rowCounter);
+                return row;
+            }
+        }
+    }
+
+    
     static AkibanInformationSchema createTablesToRegister() {
         NewAISBuilder builder = AISBBasedBuilder.create();
         
@@ -635,6 +691,10 @@
             .colTimestamp("creation_time", true)
             .colBigInt("row_count", true);
 
+        builder.userTable(SERVER_USERS)
+            .colString("user_name", IDENT_MAX, false)
+            .colBigInt("statement_count", false);
+            
         return builder.ais(false);
     }
 }

=== modified file 'src/main/java/com/akiban/server/service/monitor/MonitorService.java'
--- src/main/java/com/akiban/server/service/monitor/MonitorService.java	2013-04-23 22:45:16 +0000
+++ src/main/java/com/akiban/server/service/monitor/MonitorService.java	2013-04-26 00:32:25 +0000
@@ -58,4 +58,22 @@
 
     /** Log last statement from given monitor. */
     void logQuery(SessionMonitor sessionMonitor);
+    
+    /** Register the given User monitor. */
+    void registerUserMonitor (UserMonitor userMonitor);
+    
+    /** Deregister the given user montitor. */
+    void deregisterUserMonitor (UserMonitor userMonitor);
+    
+    /** Deregister the montor for the given user */
+    void deregisterUserMonitor (String userName);
+    
+    /** Get the user monitor for the given user name. */
+    UserMonitor getUserMonitor(String userName);
+    
+    /** Get the user monitor for the session user */
+    UserMonitor getUserMonitor(Session session);
+    
+    /** Get all the user monitors. */
+    Collection<UserMonitor> getUserMonitors();
 }

=== modified file 'src/main/java/com/akiban/server/service/monitor/MonitorServiceImpl.java'
--- src/main/java/com/akiban/server/service/monitor/MonitorServiceImpl.java	2013-04-23 22:45:16 +0000
+++ src/main/java/com/akiban/server/service/monitor/MonitorServiceImpl.java	2013-04-26 00:32:25 +0000
@@ -62,6 +62,8 @@
     private BufferedWriter queryOut;
 
     private long execTimeThreshold;
+    
+    private Map<String, UserMonitor> users;
 
     @Inject
     public MonitorServiceImpl(ConfigurationService config) {
@@ -76,6 +78,7 @@
 
         sessionAllocator = new AtomicInteger();
         sessions = new ConcurrentHashMap<>();
+        users = new ConcurrentHashMap<>();
 
         String enableLog = config.getProperty(QUERY_LOG_PROPERTY);
         this.queryLogEnabled = new AtomicBoolean(Boolean.parseBoolean(enableLog));
@@ -310,5 +313,42 @@
         logger.debug("Query log file ready for writing.");
         return true;
     }
-
+    
+    /** Register the given User monitor. */
+    @Override
+    public void registerUserMonitor (UserMonitor userMonitor) {
+        UserMonitor monitor = users.put(userMonitor.getUserName(), userMonitor);
+        assert (monitor == null || monitor == userMonitor);
+    }
+
+    /** Deregister the monitor for the given user */
+    @Override
+    public void deregisterUserMonitor (String userName) {
+        users.remove(userName);
+    }
+    
+    /** Deregister the given user monitor. */
+    @Override
+    public void deregisterUserMonitor (UserMonitor userMonitor) {
+        UserMonitor monitor = users.remove(userMonitor.getUserName());
+        assert (monitor== null || monitor == userMonitor);
+    }
+    
+    /** Get the user monitor for the given user name. */
+    @Override 
+    public UserMonitor getUserMonitor(String userName) {
+        return users.get(userName); 
+    }
+    
+    /** Get the user monitor for the session user */
+    @Override
+    public UserMonitor getUserMonitor(Session session) {
+       return session.get(SESSION_KEY).getUserMonitor();
+    }
+
+    /** Get all the user monitors. */
+    @Override
+    public Collection<UserMonitor> getUserMonitors() {
+        return users.values();
+    }
 }

=== modified file 'src/main/java/com/akiban/server/service/monitor/SessionMonitor.java'
--- src/main/java/com/akiban/server/service/monitor/SessionMonitor.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/server/service/monitor/SessionMonitor.java	2013-04-26 00:32:25 +0000
@@ -76,4 +76,7 @@
 
     /** Get any prepared statements. */
     List<PreparedStatementMonitor> getPreparedStatements();
+    
+    /** Get the user monitor for this session */
+    UserMonitor getUserMonitor();
 }

=== modified file 'src/main/java/com/akiban/server/service/monitor/SessionMonitorBase.java'
--- src/main/java/com/akiban/server/service/monitor/SessionMonitorBase.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/server/service/monitor/SessionMonitorBase.java	2013-04-26 00:32:25 +0000
@@ -32,6 +32,7 @@
     private long currentStatementEndTime = -1;
     private int rowsProcessed = 0;
     private int statementCount = 0;
+    private UserMonitor user = null; 
 
     protected SessionMonitorBase(int sessionID) {
         this.sessionID = sessionID;
@@ -65,6 +66,9 @@
     public void endStatement(int rowsProcessed) {
         currentStatementEndTime = System.currentTimeMillis();
         this.rowsProcessed = rowsProcessed;
+        if (user != null) {
+            user.statementRun();
+        }
     }
 
     // Caller can sequence all stages and avoid any gaps at the cost of more complicated
@@ -167,5 +171,13 @@
     public List<PreparedStatementMonitor> getPreparedStatements() {
         return Collections.emptyList();
     }
+    
+    public void setUserMonitor(UserMonitor monitor) {
+        this.user = monitor;
+    }
+    
+    public UserMonitor getUserMonitor() {
+        return this.user;
+    }
 
 }

=== added file 'src/main/java/com/akiban/server/service/monitor/UserMonitor.java'
--- src/main/java/com/akiban/server/service/monitor/UserMonitor.java	1970-01-01 00:00:00 +0000
+++ src/main/java/com/akiban/server/service/monitor/UserMonitor.java	2013-04-26 00:32:25 +0000
@@ -0,0 +1,32 @@
+/**
+ * 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.server.service.monitor;
+
+public interface UserMonitor {
+
+    /** User name of the user being monitored */
+    String getUserName();
+    
+    /** The number of queries executed. */
+    long getStatementCount();
+
+    /** Get total time in nanoseconds not spent idle. */
+    long getNonIdleTimeNanos();
+    
+    /** Flag a new statment has been processed */
+    void statementRun();
+}

=== modified file 'src/main/java/com/akiban/server/service/security/SecurityServiceImpl.java'
--- src/main/java/com/akiban/server/service/security/SecurityServiceImpl.java	2013-04-23 22:45:16 +0000
+++ src/main/java/com/akiban/server/service/security/SecurityServiceImpl.java	2013-04-26 00:32:25 +0000
@@ -27,6 +27,8 @@
 import com.akiban.server.error.SecurityException;
 import com.akiban.server.service.Service;
 import com.akiban.server.service.config.ConfigurationService;
+import com.akiban.server.service.monitor.MonitorService;
+import com.akiban.server.service.monitor.UserMonitor;
 import com.akiban.server.service.session.Session;
 import com.akiban.server.store.SchemaManager;
 import com.akiban.sql.server.ServerCallContextStack;
@@ -81,6 +83,7 @@
 
     private final ConfigurationService configService;
     private final SchemaManager schemaManager;
+    private final MonitorService monitor;
 
     private boolean restrictUserSchema;
 
@@ -88,9 +91,11 @@
 
     @Inject
     public SecurityServiceImpl(ConfigurationService configService,
-                               SchemaManager schemaManager) {
+                               SchemaManager schemaManager,
+                               MonitorService monitor) {
         this.configService = configService;
         this.schemaManager = schemaManager;
+        this.monitor = monitor;
     }
 
     // Connections are not thread safe, and prepared statements remember a Session,
@@ -273,6 +278,7 @@
                 throw new SecurityException("Failed to delete user");
             }
             conn.commit();
+            monitor.deregisterUserMonitor(name);
         }
         catch (SQLException ex) {
             throw new SecurityException("Error deleting user", ex);
@@ -336,6 +342,9 @@
         if (session != null) {
             session.put(SESSION_KEY, user);
         }
+        if (monitor.getUserMonitor(user.getName()) == null) {
+            monitor.registerUserMonitor(new UserMonitorImpl(user.getName()));
+        }
         return user;
     }
 
@@ -367,6 +376,9 @@
         if (session != null) {
             session.put(SESSION_KEY, user);
         }
+        if (monitor.getUserMonitor(user.getName()) == null) {
+            monitor.registerUserMonitor(new UserMonitorImpl(user.getName()));
+        }
         return user;
     }
 

=== added file 'src/main/java/com/akiban/server/service/security/UserMonitorImpl.java'
--- src/main/java/com/akiban/server/service/security/UserMonitorImpl.java	1970-01-01 00:00:00 +0000
+++ src/main/java/com/akiban/server/service/security/UserMonitorImpl.java	2013-04-26 00:32:25 +0000
@@ -0,0 +1,51 @@
+/**
+ * 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.server.service.security;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+import com.akiban.server.service.monitor.UserMonitor;
+
+public class UserMonitorImpl implements UserMonitor {
+    private final String userName;
+    protected final AtomicLong count = new AtomicLong();
+
+    public UserMonitorImpl (String name) {
+        this.userName = name;
+    }
+    @Override
+    public String getUserName() {
+        return userName;
+    }
+
+    @Override
+    public long getStatementCount() {
+        return count.get();
+    }
+
+    @Override
+    public void statementRun() {
+        count.incrementAndGet();
+    }
+    
+    @Override
+    public long getNonIdleTimeNanos() {
+        // TODO Auto-generated method stub
+        return 0;
+    }
+
+}

=== modified file 'src/main/java/com/akiban/sql/pg/PostgresServerConnection.java'
--- src/main/java/com/akiban/sql/pg/PostgresServerConnection.java	2013-04-23 22:45:16 +0000
+++ src/main/java/com/akiban/sql/pg/PostgresServerConnection.java	2013-04-26 00:32:25 +0000
@@ -522,6 +522,7 @@
         }
         logger.debug("Login {}", (principal != null) ? principal : user);
         authenticationOkay(user);
+        sessionMonitor.setUserMonitor(reqs.monitor().getUserMonitor(user));
     }
     
     protected void authenticationOkay(String user) throws IOException {

=== modified file 'src/test/java/com/akiban/server/service/is/SchemaTableServiceIT.java'
--- src/test/java/com/akiban/server/service/is/SchemaTableServiceIT.java	2013-03-22 20:05:57 +0000
+++ src/test/java/com/akiban/server/service/is/SchemaTableServiceIT.java	2013-04-26 00:32:25 +0000
@@ -98,7 +98,7 @@
     
     @Test
     public void serverExamine() {
-        assertEquals ("Table count", 10, ServerSchemaTablesServiceImpl.createTablesToRegister().getUserTables().size());
+        assertEquals ("Table count", 11, ServerSchemaTablesServiceImpl.createTablesToRegister().getUserTables().size());
         assertNotNull (ais.getUserTable(ServerSchemaTablesServiceImpl.ERROR_CODES));
         assertNotNull (ais.getUserTable(ServerSchemaTablesServiceImpl.SERVER_INSTANCE_SUMMARY));
         assertNotNull (ais.getUserTable(ServerSchemaTablesServiceImpl.SERVER_SERVERS));
@@ -109,5 +109,6 @@
         assertNotNull (ais.getUserTable(ServerSchemaTablesServiceImpl.SERVER_TAPS));
         assertNotNull (ais.getUserTable(ServerSchemaTablesServiceImpl.SERVER_PREPARED_STATEMENTS));
         assertNotNull (ais.getUserTable(ServerSchemaTablesServiceImpl.SERVER_CURSORS));
+        assertNotNull (ais.getUserTable(ServerSchemaTablesServiceImpl.SERVER_USERS));
     }
 }