Repository: vantage6/vantage6
Issue: Nodes appear to be offline

If the server reboots, the nodes status field is not set properly when using multiple workers. It seems likely that the `start()` method is triggered after some of the clients (=nodes) reconnected. 

Reported by @anjavangestel 

Changed files:
vantage6-server/vantage6/server/globals.py, vantage6-server/vantage6/server/__init__.py, vantage6-node/vantage6/node/socket.py, vantage6-server/vantage6/server/model/node.py

diff --git a/vantage6-node/vantage6/node/socket.py b/vantage6-node/vantage6/node/socket.py
index 51a1cc54..e5c58c34 100644
--- a/vantage6-node/vantage6/node/socket.py
+++ b/vantage6-node/vantage6/node/socket.py
@@ -41,6 +41,17 @@ class NodeTaskNamespace(ClientNamespace):
         # self.node_worker_ref.socketIO.disconnect()
         self.log.info('Disconnected from the server')
 
+    def on_ping(self) -> int:
+        """
+        Actions to be taken when server pings the nodes
+
+        Returns
+        -------
+        int
+            The id of this node so that server knows it is online
+        """
+        return self.node_worker_ref.server_io.whoami.id_
+
     def on_new_task(self, task_id: int):
         """
         Actions to be taken when node is notified of new task by server
diff --git a/vantage6-server/vantage6/server/__init__.py b/vantage6-server/vantage6/server/__init__.py
index 4f182286..c27c7b6d 100644
--- a/vantage6-server/vantage6/server/__init__.py
+++ b/vantage6-server/vantage6/server/__init__.py
@@ -9,6 +9,8 @@ import logging
 import os
 import uuid
 import json
+import time
+import datetime as dt
 import traceback
 
 from http import HTTPStatus
@@ -24,6 +26,7 @@ from flask_restful import Api
 from flask_mail import Mail
 from flask_principal import Principal, Identity, identity_changed
 from flask_socketio import SocketIO
+from threading import Thread
 
 from vantage6.server import db
 from vantage6.cli.context import ServerContext
@@ -38,7 +41,8 @@ from vantage6.server.globals import (
     RESOURCES,
     SUPER_USER_INFO,
     REFRESH_TOKENS_EXPIRE,
-    DEFAULT_SUPPORT_EMAIL_ADDRESS
+    DEFAULT_SUPPORT_EMAIL_ADDRESS,
+    MAX_RESPONSE_TIME_PING
 )
 from vantage6.server.resource.common.swagger_templates import swagger_template
 from vantage6.server._version import __version__
@@ -97,8 +101,15 @@ class ServerApp:
         # make specific log settings (muting etc)
         self.configure_logging()
 
-        # set the serv
+        # set the server version
         self.__version__ = __version__
+
+        # set up socket ping/pong
+        log.debug(
+            "Starting thread for socket ping/pong between server and nodes")
+        t = Thread(target=self.__socket_pingpong_worker, daemon=True)
+        t.start()
+
         log.info("Initialization done")
 
     def setup_socket_connection(self):
@@ -453,16 +464,54 @@ class ServerApp:
                            failed_login_attempts=0,
                            last_login_attempt=None)
             user.save()
+        return self
 
-        # set all nodes to offline
-        # TODO: this is *not* the way
-        nodes = db.Node.get()
-        for node in nodes:
-            node.status = 'offline'
-            node.save()
-        # session.commit()
+    def __socket_pingpong_worker(self) -> None:
+        """
+        Send ping messages periodically to nodes over the socketIO connection
+        and set node status online/offline depending on whether they respond
+        or not.
+        """
+        # when starting up the server, wait a few seconds to allow nodes that
+        # are already online to connect back to the server (otherwise they
+        # would be incorrectly set to offline for one period)
+        time.sleep(5)
+
+        # start periodic check if nodes are responsive
+        while True:
+            # Send ping event
+            try:
+                self.__pong_node_ids = []
+                self.socketio.emit(
+                    'ping', namespace='/tasks', room='all_nodes',
+                    callback=self.__pong_response
+                )
 
-        return self
+                # Wait a while to give nodes opportunity to pong
+                time.sleep(MAX_RESPONSE_TIME_PING)
+
+                # Check for each node that is online if they have responded.
+                # Otherwise set them to offline.
+                online_status_nodes = db.Node.get_online_nodes()
+                for node in online_status_nodes:
+                    if node.id not in self.__pong_node_ids:
+                        node.status = 'offline'
+                        node.save()
+
+                # we need to sleep here for a bit to make sure that there is a
+                # delay between setting nodes offline and pinging again - this
+                # prevents a racing condition in setting status
+                time.sleep(5)
+            except Exception:
+                log.exception('Pingpong thread had an exception')
+                time.sleep(MAX_RESPONSE_TIME_PING)
+
+    def __pong_response(self, node_id) -> None:
+        node = db.Node.get(node_id)
+        node.status = 'online'
+        node.last_seen = dt.datetime.utcnow()
+        node.save()
+        self.__pong_node_ids.append(node_id)
 
 
 def run_server(config: str, environment: str = 'prod',
diff --git a/vantage6-server/vantage6/server/globals.py b/vantage6-server/vantage6/server/globals.py
index c3295400..adeb70cd 100644
--- a/vantage6-server/vantage6/server/globals.py
+++ b/vantage6-server/vantage6/server/globals.py
@@ -42,4 +42,7 @@ REFRESH_TOKENS_EXPIRE = False
 DEFAULT_SUPPORT_EMAIL_ADDRESS = 'support@vantage6.ai'
 
 # default time that token is valid in minutes
-DEFAULT_EMAILED_TOKEN_VALIDITY_MINUTES = 60
\\ No newline at end of file
+DEFAULT_EMAILED_TOKEN_VALIDITY_MINUTES = 60
+
+# maximum time given to nodes to respond to ping in seconds
+MAX_RESPONSE_TIME_PING = 60
diff --git a/vantage6-server/vantage6/server/model/node.py b/vantage6-server/vantage6/server/model/node.py
index e7edafc0..1a90387b 100644
--- a/vantage6-server/vantage6/server/model/node.py
+++ b/vantage6-server/vantage6/server/model/node.py
@@ -54,6 +54,20 @@ class Node(Authenticatable):
         # no node found with matching API key
         return None
 
+    @classmethod
+    def get_online_nodes(cls):
+        """
+        Return nodes that currently have status 'online'
+
+        Returns
+        -------
+        List[Node]
+            List of node models that are currently online
+        """
+        session = DatabaseSessionManager.get_session()
+
+        return session.query(cls).filter_by(status='online').all()
+
     @classmethod
     def exists(cls, organization_id, collaboration_id):
         session = DatabaseSessionManager.get_session()