chore(async): Making create app configurable (apache#25346)

.gitignore

@@ -59,6 +59,7 @@ superset/bin/supersetc
 tmp
 rat-results.txt
 superset/app/
+superset-websocket/config.json
 
 # Node.js, webpack artifacts, storybook
 *.entry.js

docker-compose.yml

@@ -29,6 +29,14 @@ x-superset-volumes: &superset-volumes
 
 version: "3.7"
 services:
+  nginx:
+    image: nginx:latest
+    container_name: superset_nginx
+    restart: unless-stopped
+    ports:
+      - "80:80"
+    volumes:
+      - ./docker/nginx/nginx.conf:/etc/nginx/nginx.conf:ro
   redis:
     image: redis:7
     container_name: superset_cache

docker/nginx/nginx.conf

@@ -0,0 +1,127 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+user  nginx;
+worker_processes  1;
+
+error_log  /var/log/nginx/error.log warn;
+pid        /var/run/nginx.pid;
+
+
+events {
+    worker_connections  1024;
+}
+
+
+http {
+    include       /etc/nginx/mime.types;
+    default_type  application/octet-stream;
+
+    log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
+                      '$status $body_bytes_sent [$connection_requests] "$http_referer" '
+                      '"$http_user_agent" "$http_x_forwarded_for"';
+
+    access_log  /var/log/nginx/access.log  main;
+
+    sendfile        on;
+    #tcp_nopush     on;
+
+    keepalive_timeout  30;
+    keepalive_requests 2;
+
+    ###### Compression Stuff
+
+    # Enable Gzip compressed.
+    gzip on;
+
+    # Compression level (1-9).
+    # 5 is a perfect compromise between size and cpu usage, offering about
+    # 75% reduction for most ascii files (almost identical to level 9).
+    gzip_comp_level    5;
+
+    # Don't compress anything that's already small and unlikely to shrink much
+    # if at all (the default is 20 bytes, which is bad as that usually leads to
+    # larger files after gzipping).
+    gzip_min_length    256;
+
+    # Compress data even for clients that are connecting to us via proxies,
+    # identified by the "Via" header (required for CloudFront).
+    gzip_proxied       any;
+
+    # Tell proxies to cache both the gzipped and regular version of a resource
+    # whenever the client's Accept-Encoding capabilities header varies;
+    # Avoids the issue where a non-gzip capable client (which is extremely rare
+    # today) would display gibberish if their proxy gave them the gzipped version.
+    gzip_vary          on;
+
+    # Compress all output labeled with one of the following MIME-types.
+    gzip_types
+        application/atom+xml
+        application/javascript
+        application/json
+        application/rss+xml
+        application/vnd.ms-fontobject
+        application/x-font-ttf
+        application/x-web-app-manifest+json
+        application/xhtml+xml
+        application/xml
+        font/opentype
+        image/svg+xml
+        image/x-icon
+        text/css
+        text/plain
+        text/x-component;
+    # text/html is always compressed by HttpGzipModule
+
+    output_buffers 20 10m;
+
+    client_max_body_size 10m;
+
+    upstream superset_app {
+        server host.docker.internal:8088;
+        keepalive 100;
+    }
+
+    upstream superset_websocket {
+        server host.docker.internal:8080;
+        keepalive 100;
+    }
+
+    server {
+        listen 80 default_server;
+        server_name  _;
+
+        location /ws {
+            proxy_pass http://superset_websocket;
+            proxy_http_version 1.1;
+            proxy_set_header Upgrade $http_upgrade;
+            proxy_set_header Connection "Upgrade";
+            proxy_set_header Host $host;
+        }
+
+        location / {
+            proxy_pass http://superset_app;
+            proxy_set_header    Host                $host;
+            proxy_set_header    X-Real-IP           $remote_addr;
+            proxy_set_header    X-Forwarded-For     $remote_addr;
+            proxy_set_header    X-Forwarded-Host    $host;
+            proxy_set_header    X-Forwarded-Proto   $scheme;
+            proxy_http_version 1.1;
+            port_in_redirect off;
+            proxy_connect_timeout 300;
+        }
+    }
+}

superset/app.py

@@ -17,6 +17,7 @@
 
 import logging
 import os
+from typing import Optional
 
 from flask import Flask
 
@@ -25,12 +26,14 @@
 logger = logging.getLogger(__name__)
 
 
-def create_app() -> Flask:
+def create_app(superset_config_module: Optional[str] = None) -> Flask:
     app = SupersetApp(__name__)
 
     try:
         # Allow user to override our config completely
-        config_module = os.environ.get("SUPERSET_CONFIG", "superset.config")
+        config_module = superset_config_module or os.environ.get(
+            "SUPERSET_CONFIG", "superset.config"
+        )
         app.config.from_object(config_module)
 
         app_initializer = app.config.get("APP_INITIALIZER", SupersetAppInitializer)(app)

superset/async_events/api.py

@@ -21,8 +21,8 @@
 from flask_appbuilder.api import safe
 from flask_appbuilder.security.decorators import permission_name, protect
 
+from superset.async_events.async_query_manager import AsyncQueryTokenException
 from superset.extensions import async_query_manager, event_logger
-from superset.utils.async_query_manager import AsyncQueryTokenException
 from superset.views.base_api import BaseSupersetApi
 
 logger = logging.getLogger(__name__)

superset/async_events/async_query_manager_factory.py

@@ -0,0 +1,35 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from flask import Flask
+
+from superset.async_events.async_query_manager import AsyncQueryManager
+from superset.utils.class_utils import load_class_from_name
+
+
+class AsyncQueryManagerFactory:
+    def __init__(self) -> None:
+        self._async_query_manager: AsyncQueryManager = None  # type: ignore
+
+    def init_app(self, app: Flask) -> None:
+        self._async_query_manager = load_class_from_name(
+            app.config["GLOBAL_ASYNC_QUERY_MANAGER_CLASS"]
+        )()
+        self._async_query_manager.init_app(app)
+
+    def instance(self) -> AsyncQueryManager:
+        return self._async_query_manager

superset/charts/data/api.py

@@ -28,6 +28,7 @@
 from marshmallow import ValidationError
 
 from superset import is_feature_enabled, security_manager
+from superset.async_events.async_query_manager import AsyncQueryTokenException
 from superset.charts.api import ChartRestApi
 from superset.charts.commands.exceptions import (
     ChartDataCacheLoadError,
@@ -46,7 +47,6 @@
 from superset.exceptions import QueryObjectValidationError
 from superset.extensions import event_logger
 from superset.models.sql_lab import Query
-from superset.utils.async_query_manager import AsyncQueryTokenException
 from superset.utils.core import create_zip, get_user_id, json_int_dttm_ser
 from superset.views.base import CsvResponse, generate_download_headers, XlsxResponse
 from superset.views.base_api import statsd_metrics

superset/config.py

@@ -1503,6 +1503,9 @@ def EMAIL_HEADER_MUTATOR(  # pylint: disable=invalid-name,unused-argument
 
 # Global async query config options.
 # Requires GLOBAL_ASYNC_QUERIES feature flag to be enabled.
+GLOBAL_ASYNC_QUERY_MANAGER_CLASS = (
+    "superset.async_events.async_query_manager.AsyncQueryManager"
+)
 GLOBAL_ASYNC_QUERIES_REDIS_CONFIG = {
     "port": 6379,
     "host": "127.0.0.1",

superset/extensions/__init__.py

@@ -27,9 +27,10 @@
 from flask_wtf.csrf import CSRFProtect
 from werkzeug.local import LocalProxy
 
+from superset.async_events.async_query_manager import AsyncQueryManager
+from superset.async_events.async_query_manager_factory import AsyncQueryManagerFactory
 from superset.extensions.ssh import SSHManagerFactory
 from superset.extensions.stats_logger import BaseStatsLoggerManager
-from superset.utils.async_query_manager import AsyncQueryManager
 from superset.utils.cache_manager import CacheManager
 from superset.utils.encrypt import EncryptedFieldFactory
 from superset.utils.feature_flag_manager import FeatureFlagManager
@@ -114,7 +115,10 @@ def init_app(self, app: Flask) -> None:
 
 APP_DIR = os.path.join(os.path.dirname(__file__), os.path.pardir)
 appbuilder = AppBuilder(update_perms=False)
-async_query_manager = AsyncQueryManager()
+async_query_manager_factory = AsyncQueryManagerFactory()
+async_query_manager: AsyncQueryManager = LocalProxy(
+    async_query_manager_factory.instance
+)
 cache_manager = CacheManager()
 celery_app = celery.Celery()
 csrf = CSRFProtect()

superset/extensions/ssh.py

@@ -15,7 +15,6 @@
 # specific language governing permissions and limitations
 # under the License.
 
-import importlib
 import logging
 from io import StringIO
 from typing import TYPE_CHECKING
@@ -25,6 +24,7 @@
 from paramiko import RSAKey
 
 from superset.databases.utils import make_url_safe
+from superset.utils.class_utils import load_class_from_name
 
 if TYPE_CHECKING:
     from superset.databases.ssh_tunnel.models import SSHTunnel
@@ -78,18 +78,9 @@ def __init__(self) -> None:
         self._ssh_manager = None
 
     def init_app(self, app: Flask) -> None:
-        ssh_manager_fqclass = app.config["SSH_TUNNEL_MANAGER_CLASS"]
-        ssh_manager_classname = ssh_manager_fqclass[
-            ssh_manager_fqclass.rfind(".") + 1 :
-        ]
-        ssh_manager_module_name = ssh_manager_fqclass[
-            0 : ssh_manager_fqclass.rfind(".")
-        ]
-        ssh_manager_class = getattr(
-            importlib.import_module(ssh_manager_module_name), ssh_manager_classname
-        )
-
-        self._ssh_manager = ssh_manager_class(app)
+        self._ssh_manager = load_class_from_name(
+            app.config["SSH_TUNNEL_MANAGER_CLASS"]
+        )(app)
 
     @property
     def instance(self) -> SSHManager:

superset/initialization/__init__.py

@@ -35,7 +35,7 @@
     _event_logger,
     APP_DIR,
     appbuilder,
-    async_query_manager,
+    async_query_manager_factory,
     cache_manager,
     celery_app,
     csrf,
@@ -665,7 +665,7 @@ def configure_wtf(self) -> None:
 
     def configure_async_queries(self) -> None:
         if feature_flag_manager.is_feature_enabled("GLOBAL_ASYNC_QUERIES"):
-            async_query_manager.init_app(self.superset_app)
+            async_query_manager_factory.init_app(self.superset_app)
 
     def register_blueprints(self) -> None:
         for bp in self.config["BLUEPRINTS"]:

superset/utils/class_utils.py

@@ -0,0 +1,39 @@
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#  http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing,
+#  software distributed under the License is distributed on an
+#  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+#  KIND, either express or implied.  See the License for the
+#  specific language governing permissions and limitations
+#  under the License.
+
+from importlib import import_module
+from typing import Any
+
+
+def load_class_from_name(fq_class_name: str) -> Any:
+    """
+    Given a string representing a fully qualified class name, attempts to load
+    the class and return it.
+
+    :param fq_class_name: The fully qualified name of the class to load
+    :return: The class object
+    :raises Exception: if the class cannot be loaded
+    """
+    if not fq_class_name:
+        raise ValueError(f"Invalid class name {fq_class_name}")
+
+    parts = fq_class_name.split(".")
+    module_name = ".".join(parts[:-1])
+    class_name = parts[-1]
+
+    module = import_module(module_name)
+    return getattr(module, class_name)