Add console session cleanup task (#7132)

api/src/main/java/org/apache/cloudstack/consoleproxy/ConsoleAccessManager.java

@@ -18,8 +18,24 @@ package org.apache.cloudstack.consoleproxy;
 
 import com.cloud.utils.component.Manager;
 import org.apache.cloudstack.api.command.user.consoleproxy.ConsoleEndpoint;
+import org.apache.cloudstack.framework.config.ConfigKey;
+import org.apache.cloudstack.framework.config.Configurable;
 
-public interface ConsoleAccessManager extends Manager {
+public interface ConsoleAccessManager extends Manager, Configurable {
+
+    ConfigKey<Integer> ConsoleSessionCleanupRetentionHours = new ConfigKey<>("Advanced", Integer.class,
+            "console.session.cleanup.retention.hours",
+            "240",
+            "Determines the hours to keep removed console session records before expunging them",
+            false,
+            ConfigKey.Scope.Global);
+
+    ConfigKey<Integer> ConsoleSessionCleanupInterval = new ConfigKey<>("Advanced", Integer.class,
+            "console.session.cleanup.interval",
+            "180",
+            "Determines the interval (in hours) to wait between the console session cleanup tasks",
+            false,
+            ConfigKey.Scope.Global);
 
     ConsoleEndpoint generateConsoleEndpoint(Long vmId, String extraSecurityToken, String clientAddress);
 
@@ -27,5 +43,5 @@ public interface ConsoleAccessManager extends Manager {
 
     void removeSessions(String[] sessionUuids);
 
-    void removeSession(String sessionUuid);
+    void acquireSession(String sessionUuid);
 }

engine/schema/src/main/java/com/cloud/vm/ConsoleSessionVO.java

@@ -54,6 +54,9 @@ public class ConsoleSessionVO {
     @Column(name = "host_id")
     private long hostId;
 
+    @Column(name = "acquired")
+    private boolean acquired;
+
     @Column(name = "removed")
     private Date removed;
 
@@ -120,4 +123,12 @@ public class ConsoleSessionVO {
     public void setRemoved(Date removed) {
         this.removed = removed;
     }
+
+    public boolean isAcquired() {
+        return acquired;
+    }
+
+    public void setAcquired(boolean acquired) {
+        this.acquired = acquired;
+    }
 }

engine/schema/src/main/java/com/cloud/vm/dao/ConsoleSessionDao.java

@@ -22,10 +22,15 @@ package com.cloud.vm.dao;
 import com.cloud.vm.ConsoleSessionVO;
 import com.cloud.utils.db.GenericDao;
 
+import java.util.Date;
+
 public interface ConsoleSessionDao extends GenericDao<ConsoleSessionVO, Long> {
 
     void removeSession(String sessionUuid);
 
     boolean isSessionAllowed(String sessionUuid);
 
+    int expungeSessionsOlderThanDate(Date date);
+
+    void acquireSession(String sessionUuid);
 }

engine/schema/src/main/java/com/cloud/vm/dao/ConsoleSessionDaoImpl.java

@@ -19,11 +19,23 @@
 
 package com.cloud.vm.dao;
 
+import com.cloud.utils.db.SearchBuilder;
+import com.cloud.utils.db.SearchCriteria;
 import com.cloud.vm.ConsoleSessionVO;
 import com.cloud.utils.db.GenericDaoBase;
 
+import java.util.Date;
+
 public class ConsoleSessionDaoImpl extends GenericDaoBase<ConsoleSessionVO, Long> implements ConsoleSessionDao {
 
+    private final SearchBuilder<ConsoleSessionVO> searchByRemovedDate;
+
+    public ConsoleSessionDaoImpl() {
+        searchByRemovedDate = createSearchBuilder();
+        searchByRemovedDate.and("removedNotNull", searchByRemovedDate.entity().getRemoved(), SearchCriteria.Op.NNULL);
+        searchByRemovedDate.and("removed", searchByRemovedDate.entity().getRemoved(), SearchCriteria.Op.LTEQ);
+    }
+
     @Override
     public void removeSession(String sessionUuid) {
         ConsoleSessionVO session = findByUuid(sessionUuid);
@@ -32,6 +44,26 @@ public class ConsoleSessionDaoImpl extends GenericDaoBase<ConsoleSessionVO, Long
 
     @Override
     public boolean isSessionAllowed(String sessionUuid) {
-        return findByUuid(sessionUuid) != null;
+        ConsoleSessionVO consoleSessionVO = findByUuid(sessionUuid);
+        if (consoleSessionVO == null) {
+            return false;
+        }
+        return !consoleSessionVO.isAcquired();
     }
+
+    @Override
+    public int expungeSessionsOlderThanDate(Date date) {
+        SearchCriteria<ConsoleSessionVO> searchCriteria = searchByRemovedDate.create();
+        searchCriteria.setParameters("removed", date);
+        return expunge(searchCriteria);
+    }
+
+    @Override
+    public void acquireSession(String sessionUuid) {
+        ConsoleSessionVO consoleSessionVO = findByUuid(sessionUuid);
+        consoleSessionVO.setAcquired(true);
+        update(consoleSessionVO.getId(), consoleSessionVO);
+    }
+
+
 }

engine/schema/src/main/resources/META-INF/db/schema-41720to41800.sql

@@ -1480,6 +1480,7 @@ CREATE TABLE IF NOT EXISTS `cloud`.`console_session` (
     `user_id` bigint(20) unsigned NOT NULL COMMENT 'User who generated the session',
     `instance_id` bigint(20) unsigned NOT NULL COMMENT 'VM for which the session was generated',
     `host_id` bigint(20) unsigned NOT NULL COMMENT 'Host where the VM was when the session was generated',
+    `acquired` int(1) NOT NULL DEFAULT 0 COMMENT 'True if the session was already used',
     `removed` datetime COMMENT 'When the session was removed/used',
     CONSTRAINT `fk_consolesession__account_id` FOREIGN KEY(`account_id`) REFERENCES `cloud`.`account` (`id`),
     CONSTRAINT `fk_consolesession__user_id` FOREIGN KEY(`user_id`) REFERENCES `cloud`.`user`(`id`),

server/src/main/java/com/cloud/consoleproxy/AgentHookBase.java

@@ -110,8 +110,8 @@ public abstract class AgentHookBase implements AgentHook {
                 return new ConsoleAccessAuthenticationAnswer(cmd, false);
             }
 
-            s_logger.debug(String.format("Removing session [%s] as it was just used.", sessionUuid));
+            s_logger.debug(String.format("Acquiring session [%s] as it was just used.", sessionUuid));
-            consoleAccessManager.removeSession(sessionUuid);
+            consoleAccessManager.acquireSession(sessionUuid);
 
             if (!ticket.equals(ticketInUrl)) {
                 Date now = new Date();

server/src/main/java/org/apache/cloudstack/consoleproxy/ConsoleAccessManagerImpl.java

@@ -59,7 +59,9 @@ import com.cloud.uservm.UserVm;
 import com.cloud.utils.Pair;
 import com.cloud.utils.Ternary;
 import com.cloud.utils.component.ManagerBase;
+import com.cloud.utils.concurrency.NamedThreadFactory;
 import com.cloud.utils.db.EntityManager;
+import com.cloud.utils.db.GlobalLock;
 import com.cloud.utils.exception.CloudRuntimeException;
 import com.cloud.vm.ConsoleSessionVO;
 import com.cloud.vm.UserVmDetailVO;
@@ -70,6 +72,13 @@ import com.cloud.vm.dao.ConsoleSessionDao;
 import com.cloud.vm.dao.UserVmDetailsDao;
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
+import org.apache.cloudstack.framework.config.ConfigKey;
+import org.apache.cloudstack.managed.context.ManagedContextRunnable;
+import org.joda.time.DateTime;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 public class ConsoleAccessManagerImpl extends ManagerBase implements ConsoleAccessManager {
 
@@ -94,6 +103,8 @@ public class ConsoleAccessManagerImpl extends ManagerBase implements ConsoleAcce
     @Inject
     private ConsoleSessionDao consoleSessionDao;
 
+    private ScheduledExecutorService executorService = null;
+
     private static KeysManager secretKeysManager;
     private final Gson gson = new GsonBuilder().create();
 
@@ -106,9 +117,69 @@ public class ConsoleAccessManagerImpl extends ManagerBase implements ConsoleAcce
     @Override
     public boolean configure(String name, Map<String, Object> params) throws ConfigurationException {
         ConsoleAccessManagerImpl.secretKeysManager = keysManager;
+        executorService = Executors.newScheduledThreadPool(1, new NamedThreadFactory("ConsoleSession-Scavenger"));
         return super.configure(name, params);
     }
 
+    @Override
+    public boolean start() {
+        int consoleCleanupInterval = ConsoleAccessManager.ConsoleSessionCleanupInterval.value();
+        if (consoleCleanupInterval > 0) {
+            s_logger.info(String.format("The ConsoleSessionCleanupTask will run every %s hours", consoleCleanupInterval));
+            executorService.scheduleWithFixedDelay(new ConsoleSessionCleanupTask(), consoleCleanupInterval, consoleCleanupInterval, TimeUnit.HOURS);
+        }
+        return true;
+    }
+
+    @Override
+    public String getConfigComponentName() {
+        return ConsoleAccessManager.class.getName();
+    }
+
+    @Override
+    public ConfigKey<?>[] getConfigKeys() {
+        return new ConfigKey[] {
+                ConsoleAccessManager.ConsoleSessionCleanupInterval,
+                ConsoleAccessManager.ConsoleSessionCleanupRetentionHours
+        };
+    }
+
+    public class ConsoleSessionCleanupTask extends ManagedContextRunnable {
+        @Override
+        protected void runInContext() {
+            final GlobalLock gcLock = GlobalLock.getInternLock("ConsoleSession.Cleanup.Lock");
+            try {
+                if (gcLock.lock(3)) {
+                    try {
+                        reallyRun();
+                    } finally {
+                        gcLock.unlock();
+                    }
+                }
+            } finally {
+                gcLock.releaseRef();
+            }
+        }
+
+        private void reallyRun() {
+            if (s_logger.isDebugEnabled()) {
+                s_logger.debug("Starting ConsoleSessionCleanupTask...");
+            }
+            Integer retentionHours = ConsoleAccessManager.ConsoleSessionCleanupRetentionHours.value();
+            Date dateBefore = DateTime.now().minusHours(retentionHours).toDate();
+            if (s_logger.isDebugEnabled()) {
+                s_logger.debug(String.format("Retention hours: %s, checking for removed console session " +
+                        "records to expunge older than: %s", retentionHours, dateBefore));
+            }
+            int sessionsExpunged = consoleSessionDao.expungeSessionsOlderThanDate(dateBefore);
+            if (s_logger.isDebugEnabled()) {
+                s_logger.debug(sessionsExpunged > 0 ?
+                        String.format("Expunged %s removed console session records", sessionsExpunged) :
+                        "No removed console session records expunged on this cleanup task run");
+            }
+        }
+    }
+
     @Override
     public ConsoleEndpoint generateConsoleEndpoint(Long vmId, String extraSecurityToken, String clientAddress) {
         try {
@@ -171,11 +242,15 @@ public class ConsoleAccessManagerImpl extends ManagerBase implements ConsoleAcce
         }
     }
 
-    @Override
+    protected void removeSession(String sessionUuid) {
-    public void removeSession(String sessionUuid) {
         consoleSessionDao.removeSession(sessionUuid);
     }
 
+    @Override
+    public void acquireSession(String sessionUuid) {
+        consoleSessionDao.acquireSession(sessionUuid);
+    }
+
     protected boolean checkSessionPermission(VirtualMachine vm, Account account) {
         if (accountManager.isRootAdmin(account.getId())) {
             return true;

services/console-proxy/server/src/main/java/com/cloud/consoleproxy/ConsoleProxy.java

@@ -31,6 +31,7 @@ import java.util.Hashtable;
 import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
 
 import org.apache.commons.lang3.ArrayUtils;
@@ -68,6 +69,7 @@ public class ConsoleProxy {
     public static Method ensureRouteMethod;
 
     static Hashtable<String, ConsoleProxyClient> connectionMap = new Hashtable<String, ConsoleProxyClient>();
+    static Set<String> removedSessionsSet = ConcurrentHashMap.newKeySet();
     static int httpListenPort = 80;
     static int httpCmdListenPort = 8001;
     static int reconnectMaxRetry = 5;
@@ -372,7 +374,7 @@ public class ConsoleProxy {
             s_logger.info("HTTP command port is disabled");
         }
 
-        ConsoleProxyGCThread cthread = new ConsoleProxyGCThread(connectionMap);
+        ConsoleProxyGCThread cthread = new ConsoleProxyGCThread(connectionMap, removedSessionsSet);
         cthread.setName("Console Proxy GC Thread");
         cthread.start();
     }
@@ -540,6 +542,7 @@ public class ConsoleProxy {
             for (Map.Entry<String, ConsoleProxyClient> entry : connectionMap.entrySet()) {
                 if (entry.getValue() == viewer) {
                     connectionMap.remove(entry.getKey());
+                    removedSessionsSet.add(viewer.getSessionUuid());
                     return;
                 }
             }

services/console-proxy/server/src/main/java/com/cloud/consoleproxy/ConsoleProxyClientStatsCollector.java

@@ -18,9 +18,10 @@ package com.cloud.consoleproxy;
 
 import java.io.OutputStreamWriter;
 import java.util.ArrayList;
-import java.util.Enumeration;
+import java.util.Iterator;
-import java.util.Hashtable;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
@@ -42,7 +43,7 @@ public class ConsoleProxyClientStatsCollector {
         removedSessions.addAll(removed);
     }
 
-    public ConsoleProxyClientStatsCollector(Hashtable<String, ConsoleProxyClient> connMap) {
+    public ConsoleProxyClientStatsCollector(Map<String, ConsoleProxyClient> connMap) {
         setConnections(connMap);
     }
 
@@ -56,13 +57,14 @@ public class ConsoleProxyClientStatsCollector {
         gson.toJson(this, os);
     }
 
-    private void setConnections(Hashtable<String, ConsoleProxyClient> connMap) {
+    private void setConnections(Map<String, ConsoleProxyClient> connMap) {
 
         ArrayList<ConsoleProxyConnection> conns = new ArrayList<ConsoleProxyConnection>();
-        Enumeration<String> e = connMap.keys();
+        Set<String> e = connMap.keySet();
-        while (e.hasMoreElements()) {
+        Iterator<String> iterator = e.iterator();
+        while (iterator.hasNext()) {
             synchronized (connMap) {
-                String key = e.nextElement();
+                String key = iterator.next();
                 ConsoleProxyClient client = connMap.get(key);
 
                 ConsoleProxyConnection conn = new ConsoleProxyConnection();

services/console-proxy/server/src/main/java/com/cloud/consoleproxy/ConsoleProxyGCThread.java

@@ -18,9 +18,9 @@ package com.cloud.consoleproxy;
 
 import java.io.File;
 import java.util.ArrayList;
-import java.util.Enumeration;
+import java.util.Iterator;
-import java.util.Hashtable;
+import java.util.Map;
-import java.util.List;
+import java.util.Set;
 
 import org.apache.log4j.Logger;
 
@@ -35,11 +35,13 @@ public class ConsoleProxyGCThread extends Thread {
 
     private final static int MAX_SESSION_IDLE_SECONDS = 180;
 
-    private final Hashtable<String, ConsoleProxyClient> connMap;
+    private final Map<String, ConsoleProxyClient> connMap;
+    private final Set<String> removedSessionsSet;
     private long lastLogScan = 0;
 
-    public ConsoleProxyGCThread(Hashtable<String, ConsoleProxyClient> connMap) {
+    public ConsoleProxyGCThread(Map<String, ConsoleProxyClient> connMap, Set<String> removedSet) {
         this.connMap = connMap;
+        this.removedSessionsSet = removedSet;
     }
 
     private void cleanupLogging() {
@@ -69,22 +71,22 @@ public class ConsoleProxyGCThread extends Thread {
 
         boolean bReportLoad = false;
         long lastReportTick = System.currentTimeMillis();
-        List<String> removedSessions = new ArrayList<>();
 
         while (true) {
             cleanupLogging();
             bReportLoad = false;
-            removedSessions.clear();
 
-            if (s_logger.isDebugEnabled())
+            if (s_logger.isDebugEnabled()) {
-                s_logger.debug("connMap=" + connMap);
+                s_logger.debug(String.format("connMap=%s, removedSessions=%s", connMap, removedSessionsSet));
-            Enumeration<String> e = connMap.keys();
+            }
-            while (e.hasMoreElements()) {
+            Set<String> e = connMap.keySet();
+            Iterator<String> iterator = e.iterator();
+            while (iterator.hasNext()) {
                 String key;
                 ConsoleProxyClient client;
 
                 synchronized (connMap) {
-                    key = e.nextElement();
+                    key = iterator.next();
                     client = connMap.get(key);
                 }
 
@@ -94,7 +96,6 @@ public class ConsoleProxyGCThread extends Thread {
                 }
 
                 synchronized (connMap) {
-                    removedSessions.add(client.getSessionUuid());
                     connMap.remove(key);
                     bReportLoad = true;
                 }
@@ -107,13 +108,17 @@ public class ConsoleProxyGCThread extends Thread {
             if (bReportLoad || System.currentTimeMillis() - lastReportTick > 5000) {
                 // report load changes
                 ConsoleProxyClientStatsCollector collector = new ConsoleProxyClientStatsCollector(connMap);
-                collector.setRemovedSessions(removedSessions);
+                collector.setRemovedSessions(new ArrayList<>(removedSessionsSet));
                 String loadInfo = collector.getStatsReport();
                 ConsoleProxy.reportLoadInfo(loadInfo);
                 lastReportTick = System.currentTimeMillis();
+                synchronized (removedSessionsSet) {
+                    removedSessionsSet.clear();
+                }
 
-                if (s_logger.isDebugEnabled())
+                if (s_logger.isDebugEnabled()) {
                     s_logger.debug("Report load change : " + loadInfo);
+                }
             }
 
             try {

services/console-proxy/server/src/main/java/com/cloud/consoleproxy/ConsoleProxyNoVncClient.java

@@ -19,6 +19,7 @@ package com.cloud.consoleproxy;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.log4j.Logger;
 import org.eclipse.jetty.websocket.api.Session;
+import org.eclipse.jetty.websocket.api.WebSocketException;
 import org.eclipse.jetty.websocket.api.extensions.Frame;
 
 import java.awt.Image;
@@ -125,12 +126,8 @@ public class ConsoleProxyNoVncClient implements ConsoleProxyClient {
                         } else {
                             b = new byte[100];
                             readBytes = client.read(b);
-                            if (readBytes == -1) {
+                            if (readBytes == -1 || (readBytes > 0 && !sendReadBytesToNoVNC(b, readBytes))) {
-                                break;
+                                connectionAlive = false;
-                            }
-                            if (readBytes > 0) {
-                                session.getRemote().sendBytes(ByteBuffer.wrap(b, 0, readBytes));
-                                updateFrontEndActivityTime();
                             }
                         }
                     }
@@ -144,6 +141,17 @@ public class ConsoleProxyNoVncClient implements ConsoleProxyClient {
         worker.start();
     }
 
+    private boolean sendReadBytesToNoVNC(byte[] b, int readBytes) {
+        try {
+            session.getRemote().sendBytes(ByteBuffer.wrap(b, 0, readBytes));
+            updateFrontEndActivityTime();
+        } catch (WebSocketException | IOException e) {
+            s_logger.debug("Connection exception", e);
+            return false;
+        }
+        return true;
+    }
+
     /**
      * Authenticate to VNC server when not using websockets
      *