diff --git a/Dockerfile b/Dockerfile index 0a9db0a..21146c3 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,5 +1,5 @@ # syntax=docker/dockerfile:1 -ARG PYTHON_VERSION=3.14 +ARG PYTHON_VERSION=3.13 FROM python:${PYTHON_VERSION}-slim ENV GIT_SSL_NO_VERIFY=1 diff --git a/pcgl2cache/app/__init__.py b/pcgl2cache/app/__init__.py index 5fbf024..f7399cd 100644 --- a/pcgl2cache/app/__init__.py +++ b/pcgl2cache/app/__init__.py @@ -8,6 +8,7 @@ import numpy as np import redis from flask import Flask +from flask.json.provider import DefaultJSONProvider from flask.logging import default_handler from flask_cors import CORS from rq import Queue @@ -15,8 +16,8 @@ from . import config from .common import bp as l2cache_bp from .v1.routes import bp as l2cache_api_v1 -from ..ingest.cli import init_ingest_cmds -from ..ingest.rq_cli import init_rq_cmds + +# ..ingest.cli and ..ingest.rq_cli are imported lazily in configure_app, see the note there. class CustomJsonEncoder(json.JSONEncoder): @@ -37,6 +38,12 @@ def default(self, obj): return obj.__str__() return json.JSONEncoder.default(self, obj) + +class CustomJSONProvider(DefaultJSONProvider): + def dumps(self, obj, **kwargs): + return super().dumps(obj, default=None, cls=CustomJsonEncoder, **kwargs) + + def get_app_base_path(): return os.path.dirname(os.path.realpath(__file__)) @@ -47,7 +54,9 @@ def get_instance_folder_path(): def create_app(test_config=None): app = Flask(__name__,instance_path=get_instance_folder_path(), instance_relative_config=True) - app.json_encoder = CustomJsonEncoder + # app.json_encoder was removed in Flask 2.3; without the provider numpy + # arrays raise "Object of type ndarray is not JSON serializable" + app.json = CustomJSONProvider(app) CORS(app, expose_headers="WWW-Authenticate") configure_app(app) @@ -76,6 +85,15 @@ def configure_app(app): app.logger.propagate = False if app.config["USE_REDIS_JOBS"]: + # Imported here rather than at module scope: ..ingest.cli pulls in + # pcgl2cache.core.features, and with it sklearn and scipy.ndimage -- about 76 MB + # of interpreter memory, per uwsgi worker. Only the ingest/CLI configurations set + # USE_REDIS_JOBS, so the API deployment (DevelopmentConfig, USE_REDIS_JOBS=False) + # was paying that cost for commands it never registers. The API reads precomputed + # features out of Bigtable and never computes them, so it has no need of either. + from ..ingest.cli import init_ingest_cmds + from ..ingest.rq_cli import init_rq_cmds + app.redis = redis.Redis.from_url(app.config["REDIS_URL"]) app.test_q = Queue("test", connection=app.redis) with app.app_context(): diff --git a/pcgl2cache/app/common.py b/pcgl2cache/app/common.py index bb05e81..109b845 100644 --- a/pcgl2cache/app/common.py +++ b/pcgl2cache/app/common.py @@ -87,7 +87,7 @@ def unhandled_exception(e): status_code = 500 response_time = (time.time() - current_app.request_start_time) * 1000 user_ip = str(request.remote_addr) - tb = traceback.format_exception(etype=type(e), value=e, tb=e.__traceback__) + tb = traceback.format_exception(e) current_app.logger.error( { "message": str(e), @@ -114,7 +114,7 @@ def unhandled_exception(e): def api_exception(e): response_time = (time.time() - current_app.request_start_time) * 1000 user_ip = str(request.remote_addr) - tb = traceback.format_exception(etype=type(e), value=e, tb=e.__traceback__) + tb = traceback.format_exception(e) current_app.logger.error( { "message": str(e), @@ -144,6 +144,12 @@ def handle_attr_metadata(): } +def _attribute_name(key) -> str: + """Column name, whether kvdbclient keys the row by Attribute or raw bytes.""" + key = getattr(key, "key", key) + return key.decode() if isinstance(key, bytes) else str(key) + + def handle_attributes(graph_id: str, is_binary=False): if is_binary: l2ids = np.frombuffer(request.data, np.uint64) @@ -171,25 +177,37 @@ def handle_attributes(graph_id: str, is_binary=False): result[int(l2id)] = {} for k, v in attrs.items(): val = v[0].value + name = _attribute_name(k) try: # if empty list skip from response if len(val) > 0: - result[int(l2id)][k.decode()] = val + result[int(l2id)][name] = val except TypeError: # add all scalar values to response - result[int(l2id)][k.decode()] = val + result[int(l2id)][name] = val except KeyError: result[int(l2id)] = {} missing_l2ids.append(l2id) _add_offset_to_coords(graph_id, l2ids, result) _rescale_volume(graph_id, l2ids, result) update_cache = request.args.get("update_cache", default=True, type=toboolean) - if not update_cache or len(l2ids) == 0: + if not update_cache or len(missing_l2ids) == 0: return result try: - _trigger_cache_update(missing_l2ids, graph_id, cache_client.table_id) + # kvdbclient's Client stores the table id privately and exposes no public property, + # so `.table_id` raised AttributeError here while evaluating the argument -- before + # _trigger_cache_update was ever entered. Silently, because of the handler below: the + # endpoint still returned 200 with the missing ids simply absent, so callers saw + # incomplete data and no recompute was ever queued. + _trigger_cache_update(missing_l2ids, graph_id, cache_client._table_id) except Exception as e: - current_app.logger.error(str(e)) + # exc_info so the next failure here shows a traceback. str(e) alone is what made an + # AttributeError read like an idle queue for days. + current_app.logger.error( + f"Failed to trigger l2cache update for {len(missing_l2ids)} l2 ids " + f"on {graph_id}: {e}", + exc_info=True, + ) return result @@ -236,6 +254,8 @@ def _add_offset_to_coords(graph_id: str, l2ids: Iterable, result: dict): def _trigger_cache_update(l2ids, graph_id: str, l2_cache_id: str) -> None: + if len(l2ids) == 0: + return payload = np.array(l2ids, dtype=np.uint64).tobytes() attributes = { "table_id": graph_id, diff --git a/pcgl2cache/app/utils.py b/pcgl2cache/app/utils.py index 8988aa9..17109a9 100644 --- a/pcgl2cache/app/utils.py +++ b/pcgl2cache/app/utils.py @@ -1,4 +1,5 @@ import os +from functools import lru_cache from typing import Iterable import numpy as np @@ -11,8 +12,6 @@ from ..core import attributes -CACHE = {} - class DoNothingCreds(Credentials): def refresh(self, request): @@ -29,26 +28,53 @@ def get_instance_folder_path(): def jsonify_with_kwargs(data, as_response=True, **kwargs): kwargs.setdefault("separators", (",", ":")) - if current_app.config["JSONIFY_PRETTYPRINT_REGULAR"] or current_app.debug: + # JSONIFY_PRETTYPRINT_REGULAR and JSONIFY_MIMETYPE were removed in Flask 2.3 + if current_app.json.compact == False or current_app.debug: kwargs["indent"] = 2 kwargs["separators"] = (", ", ": ") resp = json.dumps(data, **kwargs) if as_response: return current_app.response_class( - resp + "\n", mimetype=current_app.config["JSONIFY_MIMETYPE"] + resp + "\n", mimetype=current_app.json.mimetype ) else: return resp +@lru_cache(maxsize=32) +def _l2cache_client(l2cache_id: str) -> BigTableClient: + """One BigTableClient per table, reused for the life of the worker. + + Constructing this per request builds a fresh gRPC channel each time. The channels + are dropped immediately but the allocation churn sets the worker's heap high-water, + which glibc never returns: a worker reaches ~68 MB above its post-import baseline + within roughly 20 requests and then stays there. Long-lived clients are also what + gRPC is designed for -- it reconnects internally, so there is nothing to refresh. + + Keyed on l2cache_id rather than graph_id so the cache stays correct if two graphs + resolve to the same table, or if config is rebuilt. + """ + info = get_default_client_info() + return BigTableClient(l2cache_id, config=info.CONFIG) + + +@lru_cache(maxsize=32) +def _l2cache_cv(cv_path: str) -> CloudVolume: + """One CloudVolume per path, reused for the life of the worker. + + Only metadata is read from it here (resolution, bounds, graph_chunk_size, meta), + never voxel data, so sharing one instance across requests is safe. Constructing it + per request re-parses the info document on every call. + """ + return CloudVolume(cv_path) + + def get_l2cache_client(graph_id: str) -> BigTableClient: l2cache_config = current_app.config["L2CACHE_CONFIG"] assert graph_id in l2cache_config, f"Dataset {graph_id} does not have an L2 Cache." - l2cache_id = l2cache_config[graph_id]["l2cache_id"] - info = get_default_client_info() - return BigTableClient(l2cache_id, config=info.CONFIG) + return _l2cache_client(l2cache_config[graph_id]["l2cache_id"]) def get_l2cache_cv(graph_id: str) -> CloudVolume: @@ -57,8 +83,7 @@ def get_l2cache_cv(graph_id: str) -> CloudVolume: graph_id in l2cache_config ), f"Dataset {graph_id} does not have CV graphene path." - cv_path = l2cache_config[graph_id]["cv_path"] - return CloudVolume(cv_path) + return _l2cache_cv(l2cache_config[graph_id]["cv_path"]) def toboolean(value): diff --git a/uwsgi.ini b/uwsgi.ini index 1917bba..a2e8128 100644 --- a/uwsgi.ini +++ b/uwsgi.ini @@ -66,6 +66,19 @@ reload-mercy = 605 harakiri = 1200 +### Stats +# Worker state for the uwsgi-exporter sidecar, which turns it into the +# uwsgi_perc_busy_workers metric the API autoscaler reads. Pod-local only; +# the sidecar shares the pod's network namespace. +# On uWSGI 2.0.31 (this image) stats-http is a boolean that makes the `stats` +# socket speak HTTP -- it is not a second address. Older uWSGI (2.0.21, still +# used by pychunkedgraph) accepted an address there, which is why that repo's +# ini looks like it binds two ports. Keep the port on `stats`. +stats = 127.0.0.1:9191 +stats-http = true +stats-interval = 5 + + ### Misc # Maintain Python thread support enable-threads = true