#68 read process output until it exits

Remove version pins to upgrade Flask

README.md

@@ -73,9 +73,11 @@ Optional environment variables:
 * `EXTERNAL_PROTOCOL` - typically http when running locally, can be https when deployed if the gateway is behind a load balancer or reverse proxy that performs https termination. Default value "http"
 * `GATEWAY_IP` - ip addess of instance gateway is running on, mostly used to display SSH instructions. Defaults to `socket.gethostbyname(socket.gethostname())`
 * `GATEWAY_PORT` - local port that the gateway should bind to, defaults to 5005
+* `GATEWAY_EXPIRE_SECONDS` - time in seconds that a cellxgene process will remain idle before being terminated. Defaults to 3600 (one hour)
 * `GATEWAY_EXTRA_SCRIPTS` - JSON array of script paths, will be embedded into each page and forwarded with `--scripts` to cellxgene server
 * `GATEWAY_ENABLE_ANNOTATIONS` - Set to `true` or to `1` to enable cellxgene annotations. 
 * `GATEWAY_ENABLE_BACKED_MODE` - Set to `true` or to `1` to load AnnData in file-backed mode. This saves memory and speeds up launch time but may reduce overall performance.
+* `GATEWAY_LOG_LEVEL` - default is `INFO`. set to `DEBUG` to increase logging and to `WARNING` to decrease logging.
 * `S3_ENABLE_LISTINGS_CACHE` - Set to `true` or to `1` to cache listings of S3 folders for performance. If the cache becomes stale, set `filecrawl.html?refresh=true` query parameter to refresh the cache.
 
 If any of the following optional variables are set, [ProxyFix](https://werkzeug.palletsprojects.com/en/1.0.x/middleware/proxy_fix/) will be used.

cellxgene_gateway/cache_entry.py

@@ -9,7 +9,6 @@
 import datetime
 import logging
 import re
-import urllib.parse
 from enum import Enum
 
 import psutil
@@ -22,6 +21,8 @@ from cellxgene_gateway.cellxgene_exception import CellxgeneException
 from cellxgene_gateway.flask_util import querystring
 from cellxgene_gateway.util import current_time_stamp
 
+logger = logging.getLogger(__name__)
+
 
 class CacheEntryStatus(Enum):
     loaded = "loaded"
@@ -111,7 +112,7 @@ class CacheEntry:
             except psutil.NoSuchProcess:
                 pass
 
-            logging.getLogger("cellxgene_gateway").info(f"terminated {terminated}")
+            logger.info(f"terminated {terminated}")
         self.status = CacheEntryStatus.terminated
 
     def rewrite_text_content(self, cellxgene_content):
@@ -169,38 +170,43 @@ class CacheEntry:
 
         full_path = self.cellxgene_basepath() + subpath + querystring()
 
-        if request.method in ["GET", "HEAD", "OPTIONS"]:
+        try:
-            cellxgene_response = get(full_path, headers=headers)
+            cellxgene_response = None
-        elif request.method == "PUT":
+            if request.method in ["GET", "HEAD", "OPTIONS"]:
-            cellxgene_response = put(
+                cellxgene_response = get(full_path, headers=headers)
-                full_path,
+            elif request.method == "PUT":
-                headers=headers,
+                cellxgene_response = put(
-                data=request.data,
+                    full_path,
-            )
+                    headers=headers,
-        elif request.method == "POST":
+                    data=request.data,
-            cellxgene_response = post(
+                )
-                full_path,
+            elif request.method == "POST":
-                headers=headers,
+                cellxgene_response = post(
-                data=request.data,
+                    full_path,
-            )
+                    headers=headers,
-        else:
+                    data=request.data,
-            raise CellxgeneException(f"Unexpected method {request.method}", 400)
+                )
-        content_type = cellxgene_response.headers["content-type"]
+            else:
-        if "text" in content_type:
+                raise CellxgeneException(f"Unexpected method {request.method}", 400)
-            gateway_content = self.rewrite_text_content(
+            content_type = cellxgene_response.headers["content-type"]
-                cellxgene_response.content.decode()
+            if "text" in content_type:
-            )
+                gateway_content = self.rewrite_text_content(
-        else:
+                    cellxgene_response.content.decode()
-            gateway_content = cellxgene_response.content
+                )
+            else:
+                gateway_content = cellxgene_response.content
 
-        resp_headers = {}
+            resp_headers = {}
-        for h in copy_headers:
+            for h in copy_headers:
-            if h in cellxgene_response.headers:
+                if h in cellxgene_response.headers:
-                resp_headers[h] = cellxgene_response.headers[h]
+                    resp_headers[h] = cellxgene_response.headers[h]
 
-        gateway_response = make_response(
+            gateway_response = make_response(
-            gateway_content,
+                gateway_content,
-            cellxgene_response.status_code,
+                cellxgene_response.status_code,
-            resp_headers,
+                resp_headers,
-        )
+            )
+        finally:
+            if cellxgene_response is not None:
+                cellxgene_response.close()
         return gateway_response

cellxgene_gateway/env.py

@@ -9,7 +9,6 @@
 
 import logging
 import os
-import socket
 
 cellxgene_location = os.environ.get("CELLXGENE_LOCATION")
 cellxgene_data = os.environ.get("CELLXGENE_DATA", "")
@@ -24,7 +23,9 @@ external_protocol = os.environ.get(
 )
 ip = os.environ.get("GATEWAY_IP")
 extra_scripts = os.environ.get("GATEWAY_EXTRA_SCRIPTS")
-ttl = os.environ.get("GATEWAY_TTL")
+expire_seconds = int(
+    os.environ.get("GATEWAY_EXPIRE_SECONDS", os.environ.get("GATEWAY_TTL", "3600"))
+)
 enable_annotations = os.environ.get("GATEWAY_ENABLE_ANNOTATIONS", "").lower() in [
     "true",
     "1",
@@ -33,6 +34,7 @@ enable_backed_mode = os.environ.get("GATEWAY_ENABLE_BACKED_MODE", "").lower() in
     "true",
     "1",
 ]
+log_level = logging.getLevelName(os.environ.get("GATEWAY_LOG_LEVEL", "INFO"))
 
 env_vars = {
     "CELLXGENE_LOCATION": cellxgene_location,
@@ -50,9 +52,10 @@ optional_env_vars = {
     "GATEWAY_IP": ip,
     "GATEWAY_PORT": gateway_port,
     "GATEWAY_EXTRA_SCRIPTS": extra_scripts,
-    "GATEWAY_TTL": ttl,
+    "GATEWAY_EXPIRE_SECONDS": expire_seconds,
     "GATEWAY_ENABLE_ANNOTATIONS": enable_annotations,
     "GATEWAY_ENABLE_BACKED_MODE": enable_backed_mode,
+    "GATEWAY_LOG_LEVEL": log_level,
     "CELLXGENE_ARGS": cellxgene_args,
     "CELLXGENE_DATA": cellxgene_data,
     "PROXY_FIX_FOR": proxy_fix_for,

cellxgene_gateway/gateway.py

@@ -299,7 +299,7 @@ def launch():
 
 def main():
     logging.basicConfig(
-        level=logging.INFO,
+        level=env.log_level,
         format="%(asctime)s:%(name)s:%(levelname)s:%(message)s",
     )
     cellxgene_data = os.environ.get("CELLXGENE_DATA", None)

cellxgene_gateway/prune_process_cache.py

@@ -18,7 +18,7 @@ logger = logging.getLogger(__name__)
 class PruneProcessCache:
     def __init__(self, cache):
         self.cache = cache
-        self.expire_seconds = 3600 if env.ttl is None else int(env.ttl)
+        self.expire_seconds = env.expire_seconds
 
     def __call__(self):
         while True:

cellxgene_gateway/subprocess_backend.py

@@ -17,6 +17,8 @@ from cellxgene_gateway.dir_util import make_annotations
 from cellxgene_gateway.env import cellxgene_args, enable_annotations, enable_backed_mode
 from cellxgene_gateway.process_exception import ProcessException
 
+logger = logging.getLogger(__name__)
+
 
 class SubprocessBackend:
     def __init__(self):
@@ -55,7 +57,7 @@ class SubprocessBackend:
             scripts,
             cache_entry.key.annotation_file_path,
         )
-        logging.getLogger("cellxgene_gateway").info(f"launching {cmd}")
+        logger.info(f"launching {cmd}")
         process = subprocess.Popen(
             [cmd], stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True
         )
@@ -84,5 +86,6 @@ class SubprocessBackend:
                 cache_entry.append_output(output)
 
         cache_entry.set_loaded(process.pid)
-
+        for output in process.communicate():
-        return
+            logger.debug(f"cellxgene:{output}")
+        logger.info(f"exiting {cmd}")

tests/test_prune_process_cache.py

@@ -15,7 +15,7 @@ key = CacheKey(
 
 class TestPruneProcessCache(unittest.TestCase):
     @patch("cellxgene_gateway.util.current_time_stamp", new=lambda: 0)
-    @patch("cellxgene_gateway.env.ttl", new="10")
+    @patch("cellxgene_gateway.env.expire_seconds", new=10)
     @patch("cellxgene_gateway.cache_entry.CacheEntry")
     @patch("cellxgene_gateway.cache_entry.CacheEntry")
     def test_GIVEN_one_old_one_new_THEN_prune_old(self, old, new):