Migrate from flask server to gunicorn

[italovalcy]

We should provide instructions and eventually make adjustments to allow running the Dashboard in production with gunicorn web server, rather than Flask web server (flask run), which is very suitable for development but not delivers good performance and robustness.

See more discussion on this topic on Flask's documentation:

• https://flask.palletsprojects.com/en/stable/tutorial/deploy/#run-with-a-production-server
• https://flask.palletsprojects.com/en/stable/deploying/

I suggest we go with gunicorn, which is also a WSGI server and this seems to be a more smooth transition.

Initial tests showed that web sockets were broken when using gunicorn though. Maybe something to do with proxy_allow_ips

[italovalcy]

It seems that running with gunicorn has to be configured with only one worker and must define the worker-class (matching with SocketIO async_mode):

gunicorn  'apps:create_app()' --bind $HOST:$PORT --worker-class eventlet --access-logfile - -w 1 --threads 10 --proxy-allow-from "127.0.0.1,::1,172.17.0.1"

For this to work I'd to change the config loading to:

diff --git a/apps/__init__.py b/apps/__init__.py
index bf134ad..ae747d8 100644
--- a/apps/__init__.py
+++ b/apps/__init__.py
@@ -17,6 +17,7 @@ from importlib import import_module
 from authlib.integrations.flask_client import OAuth
 from flask_mail import Mail
 from flask_caching import Cache
+from apps.config import app_config


 db = SQLAlchemy()
@@ -74,9 +77,9 @@ def configure_log(app):
     for handler in app.logger.handlers:
         handler.setFormatter(logging.Formatter(fmt=app.config["LOG_FMT"]))

-def create_app(config):
+def create_app():
     app = Flask(__name__)
-    app.config.from_object(config)
+    app.config.from_object(app_config)
     register_extensions(app)
     register_blueprints(app)
     #configure_database(app)

However even with the setup above I saw two problems:

1. server side logs were being set to client web socket (this is probably related to the way we run terminals, by creating a subprocess and redirecting stdout/stderr to the web socket)
2. terminal was pretty slow (specially during session opening.. much more slow than using standard flask server)

One other option seems to be uwsgi (https://flask-socketio.readthedocs.io/en/latest/deployment.html#uwsgi-web-server)

[italovalcy]

when using proxy like nginx with web sockets:

proxy_read_timeout applies to WebSocket connections, and the default 60 seconds can cause premature disconnections if no data is exchanged for that long. To fix this, either increase the proxy_read_timeout in your NGINX configuration (e.g., to a few minutes) or, preferably, implement periodic WebSocket ping/pong frames from the server to the client to keep the connection alive and prevent the timeout from triggering.
Why

[italovalcy]

I was finally able to integrate gunicorn with the HackInSDN Dashboard with support for Flask-SocketIO and multiple workers (via ip_hash nginx strategy to provide sticky sessions for gunicorn and redis as message queue -- it can also be other message queue systems).

Here are the changes I made on the dashboard app:

diff --git a/apps/__init__.py b/apps/__init__.py
index be7ee67..d8886b3 100644
--- a/apps/__init__.py
+++ b/apps/__init__.py
@@ -17,12 +17,13 @@ from importlib import import_module
 from authlib.integrations.flask_client import OAuth
 from flask_mail import Mail
 from flask_caching import Cache
+from apps.config import app_config


 db = SQLAlchemy()
 login_manager = LoginManager()
 oauth = OAuth()
-socketio = SocketIO(cors_allowed_origins="*")
+socketio = SocketIO()
 mail = Mail()
 cache = Cache()
 migrate = Migrate()
@@ -32,7 +33,12 @@ def register_extensions(app):
     db.init_app(app)
     login_manager.init_app(app)
     oauth.init_app(app)
-    socketio.init_app(app)
+    socketio.init_app(
+        app,
+        cors_allowed_origins="*",
+        async_mode='gevent',
+        message_queue=app.config.get("MESSAGE_QUEUE", None),
+    )
     mail.init_app(app)
     cache.init_app(app)
     migrate.init_app(app, db)
@@ -76,9 +82,9 @@ def configure_log(app):
     for handler in app.logger.handlers:
         handler.setFormatter(logging.Formatter(fmt=app.config["LOG_FMT"]))

-def create_app(config):
+def create_app():
     app = Flask(__name__)
-    app.config.from_object(config)
+    app.config.from_object(app_config)
     register_extensions(app)
     register_blueprints(app)
     #configure_database(app)
diff --git a/apps/controllers/kubernetes.py b/apps/controllers/kubernetes.py
index 8ffcd05..34b5b67 100644
--- a/apps/controllers/kubernetes.py
+++ b/apps/controllers/kubernetes.py
@@ -667,3 +667,17 @@ class K8sController():
             "total_pods": total_pods,
             "total_nodes": total_nodes,
         }
+
+    def get_pod_exec_stream(self, pod, container):
+        return stream(
+            self.v1_api.connect_get_namespaced_pod_exec,
+            pod,
+            self.namespace,
+            command=['sh', '-c', 'if [ -x /bin/bash ]; then exec /bin/bash; else exec /bin/sh; fi'],
+            container=container,
+            stderr=True,
+            stdin=True,
+            stdout=True,
+            tty=True,
+            _preload_content=False,
+        )
diff --git a/apps/events.py b/apps/events.py
index d257b8e..c243400 100644
--- a/apps/events.py
+++ b/apps/events.py
@@ -15,6 +15,8 @@ from apps import socketio
 from apps.controllers import k8s
 from flask_login import login_required, current_user

+from apps.controllers import k8s
+
 xterm_clients = {}


@@ -27,6 +29,10 @@ def set_winsize(fd, row, col, xpix=0, ypix=0):
     fcntl.ioctl(fd, termios.TIOCSWINSZ, winsize)


+def resize_terminal(client_stream, rows, cols):
+    client_stream.write_channel(4, '{"Width":%d,"Height":%d}' % (cols, rows))
+
+
 def read_and_forward_pty_output(session_id, fd):
     max_read_bytes = 1024 * 20
     while True:
@@ -48,6 +54,22 @@ def read_and_forward_pty_output(session_id, fd):
     socketio.emit("server-disconnected", namespace="/pty", to=session_id)


+def read_and_forward_k8s_stream_output(session_id, client_stream):
+    max_read_bytes = 1024 * 20
+    while client_stream.is_open():
+        socketio.sleep(0.01)
+        try:
+            data = client_stream.read_stdout(max_read_bytes)
+        except Exception as exc:
+            break
+        if client_stream.is_open():
+            if len(data or "") > 0:
+                socketio.emit("pty-output", {"output": data}, namespace="/pty", to=session_id)
+        else:
+            break
+    socketio.emit("server-disconnected", namespace="/pty", to=session_id)
+
+
 @socketio.on("pty-input", namespace="/pty")
 @login_required
 def pty_input(data):
@@ -57,9 +79,9 @@ def pty_input(data):
     if not session_id or session_id not in xterm_clients:
         current_app.logger.info(f"Host not connected {session_id=}")
         return False
-    (fd, _) = xterm_clients[session_id]
-    if fd:
-        os.write(fd, data["input"].encode())
+    client_stream = xterm_clients[session_id]
+    if client_stream.is_open():
+        client_stream.write_stdin(data["input"].encode())


 @socketio.on("connect", namespace="/pty")
@@ -81,22 +103,11 @@ def pty_connect(auth):
         current_app.logger.info(f"session already connected")
         return False
     current_app.logger.info(f"connecting to {pod=} {container=} {session_id=}")
-    # create child process attached to a pty we can read from and write to
-    (child_pid, fd) = pty.fork()
-    if child_pid == 0:
-        # this is the child process fork.
-        # change default env vars to avoid massive logs when DEBUG is enabled
-        myenv = dict(os.environ)
-        myenv.update({"DEBUG": ""})
-        subprocess.run(['kubectl', 'exec', '-it', f"{kind}/{pod}", '--container', container, '--', 'sh', '-c', 'if [ -x /bin/bash ]; then exec /bin/bash; else exec /bin/sh; fi'], env=myenv)
-        os._exit(os.EX_OK)
-    else:
-        # this is the parent process fork.
-        # store child fd and pid
-        xterm_clients[session_id] = (fd, child_pid)
-        set_winsize(fd, 50, 50)
-        socketio.start_background_task(read_and_forward_pty_output, session_id, fd)
-        current_app.logger.info(f"client added to xterm_clients {session_id=} {fd=} {child_pid=}")
+
+    client_stream = k8s.get_pod_exec_stream(pod, container)
+    xterm_clients[session_id] = client_stream
+    socketio.start_background_task(read_and_forward_k8s_stream_output, session_id, client_stream)
+    current_app.logger.info(f"client added to xterm_clients {session_id=}")


 @socketio.on("resize", namespace="/pty")
@@ -108,24 +119,24 @@ def resize(data):
     if not session_id or session_id not in xterm_clients:
         current_app.logger.info(f"Host not connected {session_id=}")
         return False
-    (fd, _) = xterm_clients[session_id]
-    if fd:
-        set_winsize(fd, data["dims"]["rows"], data["dims"]["cols"])
+    client_stream = xterm_clients[session_id]
+    resize_terminal(client_stream, data["dims"]["rows"], data["dims"]["cols"])


 @socketio.on("disconnect", namespace="/pty")
 @login_required
-def pty_disconnect():
+def pty_disconnect(message):
     """client disconnected."""
     global xterm_clients
     session_id = request.sid
     if not session_id or session_id not in xterm_clients:
-        current_app.logger.info(f"Host not connected {session_id=} {xterm_clients=}")
+        current_app.logger.warning(f"Host not connected {session_id=} {xterm_clients=}")
         return False
-    (_, pid) = xterm_clients[session_id]
+    current_app.logger.info(f"pty_disconnect client={request.sid} message={message}")
+    client_stream = xterm_clients[session_id]
     try:
-        os.kill(pid, signal.SIGTERM)
+        client_stream.close()
     except Exception as exc:
-        print(f"Error terminating xterm child process for {session_id}: {exc}")
-    print(f"Client disconnected {session_id}")
+        current_app.logger.error(f"Error terminating xterm child process for {session_id}: {exc}")
+    current_app.logger.info(f"Client disconnected {session_id}")
     xterm_clients.pop(session_id, None)
diff --git a/apps/config.py b/apps/config.py
index b39ac91..2edded5 100644
--- a/apps/config.py
+++ b/apps/config.py
@@ -111,6 +111,9 @@ class Config(object):
     MAP_ZOOM_LEVEL = int(os.getenv('MAP_ZOOM_LEVEL', "1"))
     MAP_POINTS = json.loads(os.getenv('MAP_POINTS', '[]'))

+    # message queue for socketio
+    MESSAGE_QUEUE = os.getenv("MESSAGE_QUEUE", None)
+

 class ProductionConfig(Config):
     DEBUG = False

One advantage of the approach above is that we can finally get rid of kubectl to open a shell console to the Pod! Now we will leverage Kubernetes API and websockets. It took me a while to figure out one simple detail: resizing the terminal. Out of curiosity, I was able to learn how the RESIZE_CHANNEL can be used by setting up a kubectl proxy and sniffing the traffic while opening a exec and forcing terminal to be resized.

For nginx config, I followed the guidelines described in https://atchison.dev/load-balancing-gunicorn-flask-socketio-workers/

Command to run gunicorn was:

python3 -m pip install gevent

gunicorn "apps:create_app()" --bind 0.0.0.0:3000 --worker-class gevent --access-logfile - -w 1 --proxy-allow-from "127.0.0.1,::1,172.17.0.1" --log-level debug --access-logformat '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" "Origin: %({Origin}i)s"' --threads 10 &

gunicorn "apps:create_app()" --bind 0.0.0.0:3001 --worker-class gevent --access-logfile - -w 1 --proxy-allow-from "127.0.0.1,::1,172.17.0.1" --log-level debug --access-logformat '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" "Origin: %({Origin}i)s"' --threads 10 &

Great documentation is available here: https://flask-socketio.readthedocs.io/en/latest/deployment.html#using-multiple-workers