From ed1be70b15223f8a6ca1bd547470b542b550a3a2 Mon Sep 17 00:00:00 2001 From: Kevin Lu <6320810+kevinlul@users.noreply.github.com> Date: Tue, 10 May 2022 14:43:34 -0400 Subject: [PATCH 01/16] draft: poll_slurm for job status --- flask/app/slurm.py | 43 +++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 41 insertions(+), 2 deletions(-) diff --git a/flask/app/slurm.py b/flask/app/slurm.py index 7910c3cf..83d0bf6b 100644 --- a/flask/app/slurm.py +++ b/flask/app/slurm.py @@ -1,5 +1,5 @@ import json -from typing import Optional +from typing import List, Optional from flask import current_app as app from slurm_rest import ApiClient, ApiException @@ -8,9 +8,10 @@ V0037JobSubmission, V0037JobProperties, V0037JobSubmissionResponse, + V0037JobsResponse, ) -from .models import Analysis +from .models import Analysis, AnalysisState # Slurm notes: @@ -67,3 +68,41 @@ def run_crg2_on_family(analysis: Analysis) -> Optional[V0037JobSubmissionRespons f"Exception when calling slurmctld_submit_job for analysis {analysis.analysis_id}", exc_info=e, ) + + +def poll_slurm() -> None: + with ApiClient(app.config["slurm"]) as api_client: + api_instance = SlurmApi(api_client) + running_analyses = Analysis.query.filter( + Analysis.analysis_state == AnalysisState.Running, + Analysis.scheduler_id != None, + ).all() + if len(running_analyses): + analyses = { analysis.scheduler_id: analysis for analysis in running_analyses } + requests = [ + api_instance.slurmctld_get_job(analysis.scheduler_id, async_req=True) + for analysis in running_analyses + ] + jobs: List[V0037JobsResponse] = [ + async_result.get() for async_result in requests + ] + for job in jobs: + # V0037JobResponseProperties + # https://slurm.schedmd.com/squeue.html#SECTION_JOB-STATE-CODES + if job.jobs[0].job_state in [ + "BOOT_FAIL", + "CANCELLED", + "DEADLINE", + "FAILED", + "NODE_FAIL", + "OUT_OF_MEMORY", + "PREEMPTED", + "TIMEOUT", + ]: + analyses[job.jobs[0].job_id].analysis_state = AnalysisState.Error + analyses[job.jobs[0].job_id].result_path = job.jobs[0].job_state + analyses[job.jobs[0].job_id].finished = job.jobs[0].end_time # int64, convert to datetime? + elif job.jobs[0].job_state == "COMPLETED": + analyses[job.jobs[0].job_id].analysis_state = AnalysisState.Done + analyses[job.jobs[0].job_id].finished = job.jobs[0].end_time # int64, convert to datetime? + db.session.commit() From e3bee40eac9f340b183ed930affb951c31c4a192 Mon Sep 17 00:00:00 2001 From: Kevin Lu <6320810+kevinlul@users.noreply.github.com> Date: Tue, 10 May 2022 15:09:18 -0400 Subject: [PATCH 02/16] half-draft: gunicorn hooks setup --- flask/gunicorn.conf.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/flask/gunicorn.conf.py b/flask/gunicorn.conf.py index d0f43fd8..ed1d0f17 100644 --- a/flask/gunicorn.conf.py +++ b/flask/gunicorn.conf.py @@ -1,9 +1,24 @@ +from apscheduler.schedulers.background import BackgroundScheduler from prometheus_flask_exporter.multiprocess import GunicornPrometheusMetrics +from flask.app.slurm import poll_slurm + + +scheduler = BackgroundScheduler(timezone="America/Toronto") + def when_ready(server): GunicornPrometheusMetrics.start_http_server_when_ready(8080) + scheduler.add_job( + send_email_notification, "cron", [app], day_of_week="mon-fri", hour="9" + ) + scheduler.add_job(poll_slurm, "interval", minutes=5) # match retention interval of Slurm + scheduler.start() def child_exit(server, worker): GunicornPrometheusMetrics.mark_process_dead_on_child_exit(worker.pid) + + +def on_exit(server): + scheduler.shutdown() From 81d080642e0ce1cbfc579aca19bb75fc1deccad9 Mon Sep 17 00:00:00 2001 From: Kevin Lu <6320810+kevinlul@users.noreply.github.com> Date: Wed, 8 Jun 2022 10:40:10 -0400 Subject: [PATCH 03/16] Set containers to Toronto time --- Dockerfile | 1 + docker-compose.ccm.yaml | 2 ++ docker-compose.test.yaml | 2 ++ docker-compose.yaml | 2 ++ flask/Dockerfile | 1 + 5 files changed, 8 insertions(+) diff --git a/Dockerfile b/Dockerfile index 42f7857b..3ffd1818 100644 --- a/Dockerfile +++ b/Dockerfile @@ -15,6 +15,7 @@ RUN pip3 install --no-cache-dir -r requirements.txt COPY --from=mc /usr/bin/mc /usr/bin/mc COPY . . ENV FLASK_ENV production +ENV TZ America/Toronto ENV PROMETHEUS_MULTIPROC_DIR /tmp EXPOSE 5000 8080 # Prevent accidentally using this image for development by adding the prod server arguments in the entrypoint diff --git a/docker-compose.ccm.yaml b/docker-compose.ccm.yaml index e8049d2c..e1d6bcfe 100644 --- a/docker-compose.ccm.yaml +++ b/docker-compose.ccm.yaml @@ -130,6 +130,8 @@ services: # - app-hawkins proxy: image: traefik:2.6 + environment: + TZ: America/Toronto volumes: - /var/run/docker.sock:/var/run/docker.sock - "${PROJECT_ROOT:-.}/traefik:/etc/traefik" diff --git a/docker-compose.test.yaml b/docker-compose.test.yaml index e932ef95..e2e5c81a 100644 --- a/docker-compose.test.yaml +++ b/docker-compose.test.yaml @@ -4,6 +4,7 @@ services: mysql: image: mysql:8.0 environment: + TZ: America/Toronto MYSQL_DATABASE: "${TEST_MYSQL_DATABASE}" MYSQL_USER: "${TEST_MYSQL_USER}" MYSQL_PASSWORD: "${TEST_MYSQL_PASSWORD}" @@ -11,6 +12,7 @@ services: minio: image: minio/minio:RELEASE.2021-11-03T03-36-36Z environment: + TZ: America/Toronto MINIO_ROOT_USER: "${TEST_MINIO_ACCESS_KEY}" MINIO_ROOT_PASSWORD: "${TEST_MINIO_SECRET_KEY}" MINIO_REGION_NAME: diff --git a/docker-compose.yaml b/docker-compose.yaml index be429756..80fe7c9b 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -17,6 +17,7 @@ services: image: mysql:8.0 restart: on-failure environment: + TZ: America/Toronto MYSQL_DATABASE: MYSQL_USER: MYSQL_PASSWORD: @@ -29,6 +30,7 @@ services: image: minio/minio:RELEASE.2021-11-03T03-36-36Z restart: on-failure environment: + TZ: America/Toronto MINIO_ROOT_USER: "${MINIO_ACCESS_KEY}" MINIO_ROOT_PASSWORD: "${MINIO_SECRET_KEY}" MINIO_REGION_NAME: diff --git a/flask/Dockerfile b/flask/Dockerfile index c3f6bde7..3682739f 100644 --- a/flask/Dockerfile +++ b/flask/Dockerfile @@ -11,6 +11,7 @@ COPY requirements-dev.txt . RUN pip3 install --no-cache-dir -r requirements-dev.txt COPY --from=mc /usr/bin/mc /usr/bin/mc ENV FLASK_ENV development +ENV TZ America/Toronto # Prevent accidentally running this image in production. # Must mount the source code in the working directory to run this image. EXPOSE 5000 From e9dae12f92bcac9c1eb68d97a05dc3537af42db7 Mon Sep 17 00:00:00 2001 From: Kevin Lu <6320810+kevinlul@users.noreply.github.com> Date: Wed, 8 Jun 2022 10:58:44 -0400 Subject: [PATCH 04/16] Remove timezone references from code --- flask/app/__init__.py | 2 +- flask/app/email.py | 18 ++++++------------ flask/gunicorn.conf.py | 2 +- 3 files changed, 8 insertions(+), 14 deletions(-) diff --git a/flask/app/__init__.py b/flask/app/__init__.py index 03e6cab2..f2854c1e 100644 --- a/flask/app/__init__.py +++ b/flask/app/__init__.py @@ -63,7 +63,7 @@ def create_app(config): def register_schedulers(app): - scheduler = BackgroundScheduler(timezone="America/Toronto") + scheduler = BackgroundScheduler() scheduler.add_job( send_email_notification, "cron", [app], day_of_week="mon-fri", hour="9" ) diff --git a/flask/app/email.py b/flask/app/email.py index 7c4f46d8..e546197e 100644 --- a/flask/app/email.py +++ b/flask/app/email.py @@ -1,23 +1,17 @@ -# using SendGrid's Python Library -# https://github.com/sendgrid/sendgrid-python import json import math import os from datetime import datetime, timedelta -from typing import Any - -from pytz import timezone -from sendgrid import SendGridAPIClient -from sendgrid.helpers.mail import From, Mail, ReplyTo, SendAt, To +from typing import Any, Dict from flask import current_app as app +from sendgrid import SendGridAPIClient +from sendgrid.helpers.mail import From, Mail, SendAt, To -from typing import Any, Dict +# https://github.com/sendgrid/sendgrid-python sg = SendGridAPIClient(os.getenv("SENDGRID_API_KEY")) -tz = timezone("EST") - def send_email( from_email: str, to_emails: str, dynamic_template_object: Dict[str, Any] @@ -52,7 +46,7 @@ def send_email( def get_daily_stats(): - today = datetime.now(tz).strftime("%Y-%m-%d") + today = datetime.now().strftime("%Y-%m-%d") params = { "aggregated_by": "day", "start_date": today, @@ -70,7 +64,7 @@ def get_daily_stats(): def get_send_time(stats): limit_per_day = 100 - send_at = datetime.now(tz) + send_at = datetime.now() for stat in stats: diff --git a/flask/gunicorn.conf.py b/flask/gunicorn.conf.py index ed1d0f17..a0ff8084 100644 --- a/flask/gunicorn.conf.py +++ b/flask/gunicorn.conf.py @@ -4,7 +4,7 @@ from flask.app.slurm import poll_slurm -scheduler = BackgroundScheduler(timezone="America/Toronto") +scheduler = BackgroundScheduler() def when_ready(server): From 56fc4661fcf229580aaabb7c09937d51b9dfb934 Mon Sep 17 00:00:00 2001 From: Kevin Lu <6320810+kevinlul@users.noreply.github.com> Date: Wed, 8 Jun 2022 13:12:06 -0400 Subject: [PATCH 05/16] Move some configuration to gunicorn.conf.py as a crucial part of the application --- Dockerfile | 2 +- docker-compose.ccm.yaml | 2 +- docker-compose.cheo.yaml | 2 +- flask/gunicorn.conf.py | 18 +++++++++++++----- 4 files changed, 16 insertions(+), 8 deletions(-) diff --git a/Dockerfile b/Dockerfile index 3ffd1818..0f5e476f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -20,4 +20,4 @@ ENV PROMETHEUS_MULTIPROC_DIR /tmp EXPOSE 5000 8080 # Prevent accidentally using this image for development by adding the prod server arguments in the entrypoint # Automatically run migrations on startup -ENTRYPOINT ["./utils/run.sh", "prod", "--bind", "0.0.0.0:5000", "--access-logfile", "-", "--log-file", "-"] +ENTRYPOINT ["./utils/run.sh", "prod", "--bind", "0.0.0.0:5000"] diff --git a/docker-compose.ccm.yaml b/docker-compose.ccm.yaml index e1d6bcfe..57112496 100644 --- a/docker-compose.ccm.yaml +++ b/docker-compose.ccm.yaml @@ -8,7 +8,7 @@ x-common: &common x-app: &app image: "ghcr.io/ccmbioinfo/stager:${ST_VERSION}" user: www-data - command: --preload --workers ${GUNICORN_WORKERS:-1} + command: --workers ${GUNICORN_WORKERS:-1} tmpfs: - /tmp healthcheck: diff --git a/docker-compose.cheo.yaml b/docker-compose.cheo.yaml index 24d056e0..76acc1f7 100644 --- a/docker-compose.cheo.yaml +++ b/docker-compose.cheo.yaml @@ -26,7 +26,7 @@ services: ports: - "5000:5000" - "9121:8080" - command: --preload --workers ${GUNICORN_WORKERS:-1} + command: --workers ${GUNICORN_WORKERS:-1} tmpfs: - /tmp volumes: diff --git a/flask/gunicorn.conf.py b/flask/gunicorn.conf.py index a0ff8084..80535f22 100644 --- a/flask/gunicorn.conf.py +++ b/flask/gunicorn.conf.py @@ -1,18 +1,26 @@ from apscheduler.schedulers.background import BackgroundScheduler from prometheus_flask_exporter.multiprocess import GunicornPrometheusMetrics -from flask.app.slurm import poll_slurm +from app.slurm import poll_slurm +# --preload https://docs.gunicorn.org/en/stable/settings.html#preload-app +# Absolutely required to initialize the app instance and threads in the master process +preload_app = True +# --access-logfile - (stdout) https://docs.gunicorn.org/en/stable/settings.html#accesslog +accesslog = "-" +# --log-file - (stderr) https://docs.gunicorn.org/en/stable/settings.html#errorlog +errorlog = "-" + scheduler = BackgroundScheduler() def when_ready(server): GunicornPrometheusMetrics.start_http_server_when_ready(8080) - scheduler.add_job( - send_email_notification, "cron", [app], day_of_week="mon-fri", hour="9" - ) - scheduler.add_job(poll_slurm, "interval", minutes=5) # match retention interval of Slurm + # scheduler.add_job( + # send_email_notification, "cron", [app], day_of_week="mon-fri", hour="9" + # ) + scheduler.add_job(poll_slurm, "interval", seconds=15) # match retention interval of Slurm scheduler.start() From 7ad01744ccb7e554dd5ea925bbc814fcd6716f50 Mon Sep 17 00:00:00 2001 From: Kevin Lu <6320810+kevinlul@users.noreply.github.com> Date: Wed, 8 Jun 2022 14:11:40 -0400 Subject: [PATCH 06/16] Refactor: move blueprints to directory, routes -> misc, buckets -> unlinked --- flask/app/__init__.py | 38 ++----------------- flask/app/blueprints/__init__.py | 31 +++++++++++++++ flask/app/{ => blueprints}/analyses.py | 10 ++--- flask/app/{ => blueprints}/datasets.py | 8 ++-- flask/app/{ => blueprints}/error_handler.py | 3 +- flask/app/{ => blueprints}/families.py | 9 +++-- flask/app/{ => blueprints}/genes.py | 6 ++- flask/app/{ => blueprints}/groups.py | 11 +++--- flask/app/{routes.py => blueprints/misc.py} | 6 +-- flask/app/{ => blueprints}/participants.py | 9 ++--- flask/app/{ => blueprints}/tissue_samples.py | 9 +++-- .../{buckets.py => blueprints/unlinked.py} | 4 +- flask/app/{ => blueprints}/users.py | 10 ++--- flask/app/{ => blueprints}/variants.py | 6 +-- 14 files changed, 81 insertions(+), 79 deletions(-) create mode 100644 flask/app/blueprints/__init__.py rename flask/app/{ => blueprints}/analyses.py (99%) rename flask/app/{ => blueprints}/datasets.py (99%) rename flask/app/{ => blueprints}/error_handler.py (99%) rename flask/app/{ => blueprints}/families.py (98%) rename flask/app/{ => blueprints}/genes.py (94%) rename flask/app/{ => blueprints}/groups.py (98%) rename flask/app/{routes.py => blueprints/misc.py} (99%) rename flask/app/{ => blueprints}/participants.py (99%) rename flask/app/{ => blueprints}/tissue_samples.py (98%) rename flask/app/{buckets.py => blueprints/unlinked.py} (98%) rename flask/app/{ => blueprints}/users.py (99%) rename flask/app/{ => blueprints}/variants.py (99%) diff --git a/flask/app/__init__.py b/flask/app/__init__.py index f2854c1e..a7a2456b 100644 --- a/flask/app/__init__.py +++ b/flask/app/__init__.py @@ -7,22 +7,9 @@ from flask import Flask, logging as flask_logging from slurm_rest import Configuration -from app import ( - analyses, - buckets, - datasets, - error_handler, - families, - genes, - groups, - manage, - participants, - routes, - tissue_samples, - users, - variants, -) +from .blueprints import register_blueprints from .extensions import db, login, ma, metrics, migrate, oauth +from .manage import register_commands from .tasks import send_email_notification from .utils import DateTimeEncoder @@ -54,7 +41,7 @@ def create_app(config): app.config["slurm"] = None register_extensions(app) - manage.register_commands(app) + register_commands(app) register_blueprints(app) if os.getenv("SENDGRID_API_KEY"): register_schedulers(app) @@ -74,25 +61,6 @@ def register_schedulers(app): atexit.register(scheduler.shutdown) -def register_blueprints(app): - - app.register_blueprint(routes.routes) - - app.register_blueprint(families.family_blueprint) - app.register_blueprint(datasets.datasets_blueprint) - app.register_blueprint(participants.participants_blueprint) - app.register_blueprint(tissue_samples.tissue_blueprint) - app.register_blueprint(analyses.analyses_blueprint) - app.register_blueprint(genes.genes_blueprint) - app.register_blueprint(variants.variants_blueprint) - - app.register_blueprint(buckets.bucket_blueprint) - app.register_blueprint(groups.groups_blueprint) - app.register_blueprint(users.users_blueprint) - - app.register_blueprint(error_handler.error_blueprint) - - def register_extensions(app): db.init_app(app) ma.init_app(app) diff --git a/flask/app/blueprints/__init__.py b/flask/app/blueprints/__init__.py new file mode 100644 index 00000000..29996c12 --- /dev/null +++ b/flask/app/blueprints/__init__.py @@ -0,0 +1,31 @@ +from flask import Flask + +from .misc import routes +from .families import family_blueprint +from .datasets import datasets_blueprint +from .participants import participants_blueprint +from .tissue_samples import tissue_blueprint +from .analyses import analyses_blueprint +from .genes import genes_blueprint +from .variants import variants_blueprint +from .unlinked import bucket_blueprint +from .groups import groups_blueprint +from .users import users_blueprint +from .error_handler import error_blueprint + +def register_blueprints(app: Flask) -> None: + app.register_blueprint(routes) + + app.register_blueprint(family_blueprint) + app.register_blueprint(datasets_blueprint) + app.register_blueprint(participants_blueprint) + app.register_blueprint(tissue_blueprint) + app.register_blueprint(analyses_blueprint) + app.register_blueprint(genes_blueprint) + app.register_blueprint(variants_blueprint) + + app.register_blueprint(bucket_blueprint) + app.register_blueprint(groups_blueprint) + app.register_blueprint(users_blueprint) + + app.register_blueprint(error_blueprint) diff --git a/flask/app/analyses.py b/flask/app/blueprints/analyses.py similarity index 99% rename from flask/app/analyses.py rename to flask/app/blueprints/analyses.py index 2b633804..b549d0a8 100644 --- a/flask/app/analyses.py +++ b/flask/app/blueprints/analyses.py @@ -8,11 +8,11 @@ from sqlalchemy.orm import aliased, joinedload, selectinload from sqlalchemy.sql.expression import cast -from . import models -from .extensions import db -from .schemas import AnalysisSchema -from .slurm import run_crg2_on_family -from .utils import ( +from .. import models +from ..extensions import db +from ..schemas import AnalysisSchema +from ..slurm import run_crg2_on_family +from ..utils import ( check_admin, clone_entity, csv_response, diff --git a/flask/app/datasets.py b/flask/app/blueprints/datasets.py similarity index 99% rename from flask/app/datasets.py rename to flask/app/blueprints/datasets.py index 75ba4111..376e4fae 100644 --- a/flask/app/datasets.py +++ b/flask/app/blueprints/datasets.py @@ -13,10 +13,10 @@ from sqlalchemy import distinct, func, select from sqlalchemy.orm import contains_eager, joinedload, selectinload -from . import models -from .extensions import db -from .schemas import RNASeqDatasetSchema -from .utils import ( +from .. import models +from ..extensions import db +from ..schemas import RNASeqDatasetSchema +from ..utils import ( check_admin, csv_response, expects_csv, diff --git a/flask/app/error_handler.py b/flask/app/blueprints/error_handler.py similarity index 99% rename from flask/app/error_handler.py rename to flask/app/blueprints/error_handler.py index 9b2815a7..a230d423 100644 --- a/flask/app/error_handler.py +++ b/flask/app/blueprints/error_handler.py @@ -1,6 +1,7 @@ -import requests import traceback + from flask import Blueprint, json, jsonify, request, current_app as app +import requests from werkzeug.exceptions import HTTPException diff --git a/flask/app/families.py b/flask/app/blueprints/families.py similarity index 98% rename from flask/app/families.py rename to flask/app/blueprints/families.py index ba20dc12..f2d3ac36 100644 --- a/flask/app/families.py +++ b/flask/app/blueprints/families.py @@ -3,10 +3,11 @@ from flask import abort, jsonify, request, Blueprint, current_app as app from flask_login import current_user, login_required from sqlalchemy.orm import joinedload -from .extensions import db -from . import models -from .schemas import FamilySchema -from .utils import ( + +from .. import models +from ..extensions import db +from ..schemas import FamilySchema +from ..utils import ( check_admin, filter_datasets_by_user_groups, get_current_user, diff --git a/flask/app/genes.py b/flask/app/blueprints/genes.py similarity index 94% rename from flask/app/genes.py rename to flask/app/blueprints/genes.py index a66c5b07..2f1c1d38 100644 --- a/flask/app/genes.py +++ b/flask/app/blueprints/genes.py @@ -1,11 +1,13 @@ from dataclasses import asdict from typing import Any, Dict + from flask import abort, jsonify, request, Blueprint from flask_login import login_required from sqlalchemy import func from sqlalchemy.orm import contains_eager, joinedload -from .models import Gene, GeneAlias -from .utils import csv_response, expects_csv, expects_json, paged, paginated_response + +from ..models import Gene, GeneAlias +from ..utils import csv_response, expects_csv, expects_json, paged, paginated_response genes_blueprint = Blueprint( "genes", diff --git a/flask/app/groups.py b/flask/app/blueprints/groups.py similarity index 98% rename from flask/app/groups.py rename to flask/app/blueprints/groups.py index 3245d79f..a2e27ce0 100644 --- a/flask/app/groups.py +++ b/flask/app/blueprints/groups.py @@ -2,13 +2,12 @@ from flask import abort, jsonify, request, Response, Blueprint, current_app as app from flask_login import login_required -from minio import Minio -from . import models -from .extensions import db -from .madmin import stager_buckets_policy -from .schemas import GroupSchema -from .utils import ( +from .. import models +from ..extensions import db +from ..madmin import stager_buckets_policy +from ..schemas import GroupSchema +from ..utils import ( check_admin, get_current_user, get_minio_admin, diff --git a/flask/app/routes.py b/flask/app/blueprints/misc.py similarity index 99% rename from flask/app/routes.py rename to flask/app/blueprints/misc.py index cbcb2c73..ccfc5418 100644 --- a/flask/app/routes.py +++ b/flask/app/blueprints/misc.py @@ -10,9 +10,9 @@ import numpy as np import pandas as pd from sqlalchemy.orm import joinedload -from . import models, schemas -from .extensions import db, oauth -from .utils import ( +from .. import models, schemas +from ..extensions import db, oauth +from ..utils import ( get_current_user, transaction_or_abort, validate_json, diff --git a/flask/app/participants.py b/flask/app/blueprints/participants.py similarity index 99% rename from flask/app/participants.py rename to flask/app/blueprints/participants.py index 9e138ca2..fa85d22f 100644 --- a/flask/app/participants.py +++ b/flask/app/blueprints/participants.py @@ -5,12 +5,11 @@ from flask_login import current_user, login_required from sqlalchemy import distinct, func, select from sqlalchemy.orm import contains_eager, joinedload -from sqlalchemy.orm.exc import NoResultFound -from . import models -from .extensions import db -from .schemas import ParticipantSchema -from .utils import ( +from .. import models +from ..extensions import db +from ..schemas import ParticipantSchema +from ..utils import ( check_admin, csv_response, expects_csv, diff --git a/flask/app/tissue_samples.py b/flask/app/blueprints/tissue_samples.py similarity index 98% rename from flask/app/tissue_samples.py rename to flask/app/blueprints/tissue_samples.py index 74141baf..e3fee8fd 100644 --- a/flask/app/tissue_samples.py +++ b/flask/app/blueprints/tissue_samples.py @@ -2,11 +2,12 @@ from flask import abort, jsonify, request, Blueprint, current_app as app from flask_login import current_user, login_required -from .extensions import db -from . import models + +from .. import models +from ..extensions import db from sqlalchemy.orm import contains_eager, joinedload -from .schemas import TissueSampleSchema -from .utils import ( +from ..schemas import TissueSampleSchema +from ..utils import ( check_admin, filter_datasets_by_user_groups, get_current_user, diff --git a/flask/app/buckets.py b/flask/app/blueprints/unlinked.py similarity index 98% rename from flask/app/buckets.py rename to flask/app/blueprints/unlinked.py index 0e5f7caa..b73186b2 100644 --- a/flask/app/buckets.py +++ b/flask/app/blueprints/unlinked.py @@ -3,8 +3,8 @@ from sqlalchemy.orm import joinedload from sqlalchemy import or_ -from . import models -from .utils import get_minio_client +from .. import models +from ..utils import get_minio_client bucket_blueprint = Blueprint( diff --git a/flask/app/users.py b/flask/app/blueprints/users.py similarity index 99% rename from flask/app/users.py rename to flask/app/blueprints/users.py index c6da6d9c..52850b24 100644 --- a/flask/app/users.py +++ b/flask/app/blueprints/users.py @@ -6,11 +6,11 @@ from flask_login import current_user, login_required from sqlalchemy.orm import joinedload -from . import models -from .extensions import db -from .madmin import MinioAdmin -from .schemas import UserSchema -from .utils import ( +from .. import models +from ..extensions import db +from ..madmin import MinioAdmin +from ..schemas import UserSchema +from ..utils import ( check_admin, get_minio_admin, transaction_or_abort, diff --git a/flask/app/variants.py b/flask/app/blueprints/variants.py similarity index 99% rename from flask/app/variants.py rename to flask/app/blueprints/variants.py index 9535bc62..eee049d3 100644 --- a/flask/app/variants.py +++ b/flask/app/blueprints/variants.py @@ -8,9 +8,9 @@ from sqlalchemy.orm import aliased, contains_eager from sqlalchemy.sql import and_, or_ -from . import models -from .extensions import db -from .utils import ( +from .. import models +from ..extensions import db +from ..utils import ( expects_csv, expects_json, filter_datasets_by_user_groups, From 39f96444ac036b04a68b10fc21085ce9ffb89bea Mon Sep 17 00:00:00 2001 From: Kevin Lu <6320810+kevinlul@users.noreply.github.com> Date: Thu, 9 Jun 2022 10:37:23 -0400 Subject: [PATCH 07/16] Create StagerLoginManager --- flask/app/blueprints/misc.py | 8 ++--- flask/app/login.py | 62 ++++++++++++++++++++++++++++++++++++ flask/app/models.py | 44 +------------------------ 3 files changed, 67 insertions(+), 47 deletions(-) create mode 100644 flask/app/login.py diff --git a/flask/app/blueprints/misc.py b/flask/app/blueprints/misc.py index ccfc5418..9517d7e4 100644 --- a/flask/app/blueprints/misc.py +++ b/flask/app/blueprints/misc.py @@ -11,7 +11,7 @@ import pandas as pd from sqlalchemy.orm import joinedload from .. import models, schemas -from ..extensions import db, oauth +from ..extensions import db from ..utils import ( get_current_user, transaction_or_abort, @@ -92,7 +92,7 @@ def oidc_login(): abort(405) provider = app.config.get("OIDC_PROVIDER") app.logger.debug(f"Creating client with {provider}...") - client = oauth.create_client(provider) + client = app.oauth.create_client(provider) app.logger.debug("Building redirect url...") # It's safe to take redirect_uris from the client since # the OAuth provider maintains a list of valid redirect_uris @@ -114,7 +114,7 @@ def authorize(): """ if not app.config.get("ENABLE_OIDC"): abort(404) - client = oauth.create_client(app.config.get("OIDC_PROVIDER")) + client = app.oauth.create_client(app.config.get("OIDC_PROVIDER")) # Exchange authorization code for token token = client.authorize_access_token() userinfo = client.parse_id_token(token) @@ -161,7 +161,7 @@ def logout(): if app.config.get("ENABLE_OIDC"): # Log out of OAuth session as well as Stager session provider = app.config.get("OIDC_PROVIDER") - client = oauth.create_client(provider) + client = app.oauth.create_client(provider) client_id = app.config.get("OIDC_CLIENT_ID") metadata = client.load_server_metadata() # .well-known/openid-configuration if username: diff --git a/flask/app/login.py b/flask/app/login.py new file mode 100644 index 00000000..c64a3587 --- /dev/null +++ b/flask/app/login.py @@ -0,0 +1,62 @@ +from typing import Any, Dict + +from authlib.integrations.flask_client import OAuth +from flask import current_app as app, Request +from flask_login import LoginManager +import requests +from werkzeug.exceptions import Unauthorized + +from .models import User + + +def load_user(uid: int) -> User: + return User.query.get(uid) + + +def fetch_userinfo(token: str) -> Dict[str, Any]: + """get roles and other information from the userinfo endpoint""" + provider = app.config.get("OIDC_PROVIDER") + client = app.oauth.create_client(provider) + userinfo_endpoint = client.load_server_metadata().get("userinfo_endpoint") + userinfo_response = requests.get( + userinfo_endpoint, headers={"Authorization": f"Bearer {token}"} + ) + + if userinfo_response.status_code == 401: + raise Unauthorized + + return userinfo_response.json() + + +def get_user_identity_from_userinfo(user_info: Dict[str, Any]) -> User: + """find token user based on subject identifier""" + sub = user_info["sub"] + return User.query.filter(User.subject == sub).first() + + +def load_user_from_request(request: Request) -> User: + """if user session can't be found, this function will be called to look for it elsewhere""" + auth_header = request.headers.get("Authorization") + if not auth_header or not app.config.get("ENABLE_OIDC"): + return None + token = auth_header[7:] + try: + user_info = fetch_userinfo(token) + except Unauthorized: + app.logger.info("Invalid Token!") + return None + return get_user_identity_from_userinfo(user_info) + + +class StagerLoginManager(LoginManager): + """ + Flask-Login manager. Must be initialized with a Stager Flask instance as it depends on app.oauth + """ + session_protection = "strong" + + def __init__(self, stager): + if not (hasattr(stager, "oauth") and isinstance(stager.oauth, OAuth)): + raise TypeError("Flask application is missing oauth attribute") + super().__init__(stager) + self.user_loader(load_user) + self.request_loader(load_user_from_request) diff --git a/flask/app/models.py b/flask/app/models.py index eb08c8e3..75396ab8 100644 --- a/flask/app/models.py +++ b/flask/app/models.py @@ -9,7 +9,7 @@ from werkzeug.security import check_password_hash, generate_password_hash from werkzeug.exceptions import Unauthorized -from .extensions import db, login, oauth +from .extensions import db users_groups_table = db.Table( "users_groups", @@ -50,48 +50,6 @@ def set_oidc_fields(self, issuer: str, subject: str): self.issuer = issuer self.subject = subject - -@login.user_loader -def load_user(uid: int): - return User.query.get(uid) - - -@login.request_loader -def load_user_from_request(request: Request): - """if user session can't be found, this function will be called to look for it elsewhere""" - auth_header = request.headers.get("Authorization") - if not auth_header or not app.config.get("ENABLE_OIDC"): - return None - token = auth_header[7:] - try: - user_info = fetch_userinfo(token) - except Unauthorized: - app.logger.info("Invalid Token!") - return None - return get_user_identity_from_userinfo(user_info) - - -def fetch_userinfo(token: str): - """get roles and other information from the userinfo endpoint""" - provider = app.config.get("OIDC_PROVIDER") - client = oauth.create_client(provider) - userinfo_endpoint = client.load_server_metadata().get("userinfo_endpoint") - userinfo_response = get( - userinfo_endpoint, headers={"Authorization": f"Bearer {token}"} - ) - - if userinfo_response.status_code == 401: - raise Unauthorized - - return userinfo_response.json() - - -def get_user_identity_from_userinfo(user_info: dict): - """find token user based on subject identifier""" - sub = user_info["sub"] - return User.query.filter(User.subject == sub).first() - - @dataclass class Group(db.Model): group_id: int = db.Column(db.Integer, primary_key=True) From 8ea4cd1c5a03fffaa16043b7c329b1057588bf51 Mon Sep 17 00:00:00 2001 From: Kevin Lu <6320810+kevinlul@users.noreply.github.com> Date: Thu, 9 Jun 2022 10:38:06 -0400 Subject: [PATCH 08/16] Create Stager Flask app subclass --- flask/app/__init__.py | 46 +++-------------------- flask/app/extensions.py | 22 +++++------ flask/app/madmin.py | 1 - flask/app/manage.py | 15 +++----- flask/app/mapping_utils.py | 11 +++--- flask/app/schemas.py | 3 ++ flask/app/stager.py | 75 ++++++++++++++++++++++++++++++++++++++ 7 files changed, 104 insertions(+), 69 deletions(-) create mode 100644 flask/app/stager.py diff --git a/flask/app/__init__.py b/flask/app/__init__.py index a7a2456b..275ef9bb 100644 --- a/flask/app/__init__.py +++ b/flask/app/__init__.py @@ -1,17 +1,15 @@ -import atexit import logging import os from stat import S_ISFIFO -from apscheduler.schedulers.background import BackgroundScheduler -from flask import Flask, logging as flask_logging +from flask import logging as flask_logging from slurm_rest import Configuration + from .blueprints import register_blueprints -from .extensions import db, login, ma, metrics, migrate, oauth +from .extensions import db from .manage import register_commands -from .tasks import send_email_notification -from .utils import DateTimeEncoder +from .stager import Stager def create_app(config): @@ -19,9 +17,7 @@ def create_app(config): The application factory. Returns an instance of the app. """ # Create the application object - app = Flask(__name__) - app.config.from_object(config) - app.json_encoder = DateTimeEncoder + app = Stager(config, db, __name__) config_logger(app) @@ -40,44 +36,12 @@ def create_app(config): else: app.config["slurm"] = None - register_extensions(app) register_commands(app) register_blueprints(app) - if os.getenv("SENDGRID_API_KEY"): - register_schedulers(app) return app -def register_schedulers(app): - scheduler = BackgroundScheduler() - scheduler.add_job( - send_email_notification, "cron", [app], day_of_week="mon-fri", hour="9" - ) - - scheduler.start() - - # Shut down the scheduler when exiting the app - atexit.register(scheduler.shutdown) - - -def register_extensions(app): - db.init_app(app) - ma.init_app(app) - migrate.init_app(app, db) - login.init_app(app) - oauth.init_app(app) - oauth.register( - name=app.config["OIDC_PROVIDER"], - client_id=app.config["OIDC_CLIENT_ID"], - client_secret=app.config["OIDC_CLIENT_SECRET"], - server_metadata_url=app.config["OIDC_WELL_KNOWN"], - client_kwargs={"scope": "openid"}, - ) - metrics.init_app(app) - metrics.info("stager", "Stager process info", revision=app.config.get("GIT_SHA")) - - def config_logger(app): """ Configure the main loggers: Flask application, SQLAlchemy, and werkzeug diff --git a/flask/app/extensions.py b/flask/app/extensions.py index fdaabde6..ea54c388 100644 --- a/flask/app/extensions.py +++ b/flask/app/extensions.py @@ -8,15 +8,15 @@ from prometheus_flask_exporter.multiprocess import GunicornPrometheusMetrics db = SQLAlchemy() -login = LoginManager() -migrate = Migrate(compare_type=True) -oauth = OAuth() -ma = Marshmallow() -login.session_protection = "strong" +# login = LoginManager() +# migrate = Migrate(compare_type=True) +# oauth = OAuth() +# ma = Marshmallow() +# login.session_protection = "strong" -if getenv("FLASK_ENV") == "development": - # Note that /metrics will not be live without DEBUG_METRICS set - metrics = PrometheusMetrics(None) -else: # production - # Must additionally set PROMETHEUS_MULTIPROC_DIR - metrics = GunicornPrometheusMetrics(None, path=None) +# if getenv("FLASK_ENV") == "development": +# # Note that /metrics will not be live without DEBUG_METRICS set +# metrics = PrometheusMetrics(None) +# else: # production +# # Must additionally set PROMETHEUS_MULTIPROC_DIR +# metrics = GunicornPrometheusMetrics(None, path=None) diff --git a/flask/app/madmin.py b/flask/app/madmin.py index cea7a325..365e0e3a 100644 --- a/flask/app/madmin.py +++ b/flask/app/madmin.py @@ -1,4 +1,3 @@ -from flask import abort import json, os, subprocess from itertools import chain from typing import Any, Dict, List, Optional, Union diff --git a/flask/app/manage.py b/flask/app/manage.py index 1f606fcc..7cc569ca 100644 --- a/flask/app/manage.py +++ b/flask/app/manage.py @@ -1,23 +1,20 @@ +import gzip, pickle, random from datetime import datetime, date - from pprint import pprint -import gzip, pickle, random -from sqlalchemy import exc import click from click.exceptions import ClickException - from flask import Flask, current_app as app from flask.cli import with_appcontext import pandas as pd -from sqlalchemy import and_ +from sqlalchemy import exc -from app import models # duplicated - how to best account for this +from . import models +from .models import * from .extensions import db from .madmin import stager_buckets_policy +from .utils import get_minio_admin, get_minio_client, stager_is_keycloak_admin from .manage_keycloak import * - -# for report mapping and insertion from .mapping_utils import ( get_report_paths, preprocess_report, @@ -25,8 +22,6 @@ check_result_paths, try_int, ) -from .models import * -from .utils import get_minio_admin, get_minio_client, stager_is_keycloak_admin def register_commands(app: Flask) -> None: diff --git a/flask/app/mapping_utils.py b/flask/app/mapping_utils.py index 8ba21825..e6ac1af3 100644 --- a/flask/app/mapping_utils.py +++ b/flask/app/mapping_utils.py @@ -3,17 +3,16 @@ """ from glob import glob -import pandas as pd import os +from typing import List +import numpy as np +import pandas as pd from sqlalchemy.orm.exc import MultipleResultsFound from sqlalchemy import or_ -from app import models -from app.extensions import db - -from typing import List -import numpy as np +from . import models +from .extensions import db def try_int(value: str): diff --git a/flask/app/schemas.py b/flask/app/schemas.py index 8739b6eb..8af30503 100644 --- a/flask/app/schemas.py +++ b/flask/app/schemas.py @@ -4,6 +4,9 @@ from .models import * +# TODO: use flask_marshmallow's instance if we use these to serialize +# https://flask-marshmallow.readthedocs.io/en/latest/ + class FamilySchema(SQLAlchemyAutoSchema): """ POST /api/families diff --git a/flask/app/stager.py b/flask/app/stager.py new file mode 100644 index 00000000..25cc12a5 --- /dev/null +++ b/flask/app/stager.py @@ -0,0 +1,75 @@ +import atexit +from os import getenv +import os + +from apscheduler.schedulers.base import BaseScheduler +from apscheduler.schedulers.background import BackgroundScheduler +from authlib.integrations.flask_client import OAuth +from flask import Flask +from flask_migrate import Migrate +from flask_sqlalchemy import SQLAlchemy +from prometheus_flask_exporter import PrometheusMetrics +from prometheus_flask_exporter.multiprocess import GunicornPrometheusMetrics + +from .login import StagerLoginManager +from .tasks import send_email_notification +from .utils import DateTimeEncoder + + +class Stager(Flask): + """ + Main application subclass, instead of cluttering __init__ and create_app. + """ + + json_encoder = DateTimeEncoder + # Only available in master process + scheduler: BaseScheduler + + def __init__(self, config, db: SQLAlchemy, *args, **kwargs): + super().__init__(*args, **kwargs) + self.config.from_object(config) + db.init_app(self) + self.migrate = Migrate(self, db, compare_type=True) + + # The rest are not required for Click commands, but required for routes, shell, tests, etc. + self.oauth = OAuth(self) + self.oauth.register( + name=self.config["OIDC_PROVIDER"], + client_id=self.config["OIDC_CLIENT_ID"], + client_secret=self.config["OIDC_CLIENT_SECRET"], + server_metadata_url=self.config["OIDC_WELL_KNOWN"], + client_kwargs={"scope": "openid"}, + ) + self.login = StagerLoginManager(self) + + if self.env == "development": + # Note that /metrics will not be live without DEBUG_METRICS set + self.metrics = PrometheusMetrics(self) + else: # production + # Must additionally set PROMETHEUS_MULTIPROC_DIR + self.metrics = GunicornPrometheusMetrics(self, path=None) + self.metrics.info( + "stager", "Stager process info", revision=self.config.get("GIT_SHA") + ) + + if self.env == "development": + # avoid starting additional threads for Click commands or + # duplicating the scheduler in the dev server master process + self.before_first_request(self.start_scheduler) + else: # production, start in master process + self.start_scheduler() + + def start_scheduler(self): + # If this setup of when the scheduler can be started becomes too confusing + # or cumbersome, it can be separated to be started by a completely different + # entrypoint in the same codebase and deployed as a separate container. + self.scheduler = BackgroundScheduler() + self.scheduler.add_job( + lambda: self.logger.info(os.getpid()), "interval", seconds=5 + ) + if getenv("SENDGRID_API_KEY"): + self.scheduler.add_job( + send_email_notification, "cron", [self], day_of_week="mon-fri", hour="9" + ) + self.scheduler.start() + atexit.register(self.scheduler.shutdown) From 3dee1c81f472d7d4c8a55a6dfc8089d4fe87cf2f Mon Sep 17 00:00:00 2001 From: Kevin Lu <6320810+kevinlul@users.noreply.github.com> Date: Thu, 9 Jun 2022 10:49:11 -0400 Subject: [PATCH 09/16] Remove extensions, move db global to models --- flask/app/__init__.py | 2 +- flask/app/blueprints/analyses.py | 2 +- flask/app/blueprints/datasets.py | 2 +- flask/app/blueprints/families.py | 2 +- flask/app/blueprints/groups.py | 2 +- flask/app/blueprints/misc.py | 2 +- flask/app/blueprints/participants.py | 2 +- flask/app/blueprints/tissue_samples.py | 2 +- flask/app/blueprints/users.py | 2 +- flask/app/blueprints/variants.py | 2 +- flask/app/extensions.py | 22 ------------------- flask/app/manage.py | 22 +++++++++---------- flask/app/manage_keycloak.py | 3 +-- flask/app/mapping_utils.py | 2 +- flask/app/models.py | 8 +++---- flask/app/utils.py | 4 +--- ...6f448fc94a2d_many_to_many_dataset_files.py | 3 ++- ...7e6ad115b13_drop_old_dataset_file_table.py | 4 +++- 18 files changed, 32 insertions(+), 56 deletions(-) delete mode 100644 flask/app/extensions.py diff --git a/flask/app/__init__.py b/flask/app/__init__.py index 275ef9bb..c3a6ae51 100644 --- a/flask/app/__init__.py +++ b/flask/app/__init__.py @@ -7,8 +7,8 @@ from .blueprints import register_blueprints -from .extensions import db from .manage import register_commands +from .models import db from .stager import Stager diff --git a/flask/app/blueprints/analyses.py b/flask/app/blueprints/analyses.py index b549d0a8..9e771703 100644 --- a/flask/app/blueprints/analyses.py +++ b/flask/app/blueprints/analyses.py @@ -9,7 +9,7 @@ from sqlalchemy.sql.expression import cast from .. import models -from ..extensions import db +from ..models import db from ..schemas import AnalysisSchema from ..slurm import run_crg2_on_family from ..utils import ( diff --git a/flask/app/blueprints/datasets.py b/flask/app/blueprints/datasets.py index 376e4fae..6b9b055d 100644 --- a/flask/app/blueprints/datasets.py +++ b/flask/app/blueprints/datasets.py @@ -14,7 +14,7 @@ from sqlalchemy.orm import contains_eager, joinedload, selectinload from .. import models -from ..extensions import db +from ..models import db from ..schemas import RNASeqDatasetSchema from ..utils import ( check_admin, diff --git a/flask/app/blueprints/families.py b/flask/app/blueprints/families.py index f2d3ac36..48306581 100644 --- a/flask/app/blueprints/families.py +++ b/flask/app/blueprints/families.py @@ -5,7 +5,7 @@ from sqlalchemy.orm import joinedload from .. import models -from ..extensions import db +from ..models import db from ..schemas import FamilySchema from ..utils import ( check_admin, diff --git a/flask/app/blueprints/groups.py b/flask/app/blueprints/groups.py index a2e27ce0..072c7e6d 100644 --- a/flask/app/blueprints/groups.py +++ b/flask/app/blueprints/groups.py @@ -4,7 +4,7 @@ from flask_login import login_required from .. import models -from ..extensions import db +from ..models import db from ..madmin import stager_buckets_policy from ..schemas import GroupSchema from ..utils import ( diff --git a/flask/app/blueprints/misc.py b/flask/app/blueprints/misc.py index 9517d7e4..1ee0792e 100644 --- a/flask/app/blueprints/misc.py +++ b/flask/app/blueprints/misc.py @@ -11,7 +11,7 @@ import pandas as pd from sqlalchemy.orm import joinedload from .. import models, schemas -from ..extensions import db +from ..models import db from ..utils import ( get_current_user, transaction_or_abort, diff --git a/flask/app/blueprints/participants.py b/flask/app/blueprints/participants.py index fa85d22f..247f32b2 100644 --- a/flask/app/blueprints/participants.py +++ b/flask/app/blueprints/participants.py @@ -7,7 +7,7 @@ from sqlalchemy.orm import contains_eager, joinedload from .. import models -from ..extensions import db +from ..models import db from ..schemas import ParticipantSchema from ..utils import ( check_admin, diff --git a/flask/app/blueprints/tissue_samples.py b/flask/app/blueprints/tissue_samples.py index e3fee8fd..9622cc23 100644 --- a/flask/app/blueprints/tissue_samples.py +++ b/flask/app/blueprints/tissue_samples.py @@ -4,7 +4,7 @@ from flask_login import current_user, login_required from .. import models -from ..extensions import db +from ..models import db from sqlalchemy.orm import contains_eager, joinedload from ..schemas import TissueSampleSchema from ..utils import ( diff --git a/flask/app/blueprints/users.py b/flask/app/blueprints/users.py index 52850b24..f69cfbea 100644 --- a/flask/app/blueprints/users.py +++ b/flask/app/blueprints/users.py @@ -7,7 +7,7 @@ from sqlalchemy.orm import joinedload from .. import models -from ..extensions import db +from ..models import db from ..madmin import MinioAdmin from ..schemas import UserSchema from ..utils import ( diff --git a/flask/app/blueprints/variants.py b/flask/app/blueprints/variants.py index eee049d3..c9d17984 100644 --- a/flask/app/blueprints/variants.py +++ b/flask/app/blueprints/variants.py @@ -9,7 +9,7 @@ from sqlalchemy.sql import and_, or_ from .. import models -from ..extensions import db +from ..models import db from ..utils import ( expects_csv, expects_json, diff --git a/flask/app/extensions.py b/flask/app/extensions.py deleted file mode 100644 index ea54c388..00000000 --- a/flask/app/extensions.py +++ /dev/null @@ -1,22 +0,0 @@ -from os import getenv -from flask_login import LoginManager -from flask_marshmallow import Marshmallow -from flask_migrate import Migrate -from flask_sqlalchemy import SQLAlchemy -from authlib.integrations.flask_client import OAuth -from prometheus_flask_exporter import PrometheusMetrics -from prometheus_flask_exporter.multiprocess import GunicornPrometheusMetrics - -db = SQLAlchemy() -# login = LoginManager() -# migrate = Migrate(compare_type=True) -# oauth = OAuth() -# ma = Marshmallow() -# login.session_protection = "strong" - -# if getenv("FLASK_ENV") == "development": -# # Note that /metrics will not be live without DEBUG_METRICS set -# metrics = PrometheusMetrics(None) -# else: # production -# # Must additionally set PROMETHEUS_MULTIPROC_DIR -# metrics = GunicornPrometheusMetrics(None, path=None) diff --git a/flask/app/manage.py b/flask/app/manage.py index 7cc569ca..5eda70f0 100644 --- a/flask/app/manage.py +++ b/flask/app/manage.py @@ -9,9 +9,7 @@ import pandas as pd from sqlalchemy import exc -from . import models from .models import * -from .extensions import db from .madmin import stager_buckets_policy from .utils import get_minio_admin, get_minio_client, stager_is_keycloak_admin from .manage_keycloak import * @@ -43,7 +41,7 @@ def register_commands(app: Flask) -> None: def migrate_minio_policies() -> None: minio_client = get_minio_client() minio_admin = get_minio_admin() - groups = models.Group.query.all() + groups = Group.query.all() for group in groups: policy = stager_buckets_policy(group.group_code) # All adds are upserts @@ -101,12 +99,12 @@ def map_insert_c4r_reports(report_root_path) -> None: conn = engine.connect() app.logger.info("Deleting Genotype table..") - models.Genotype.query.delete() + Genotype.query.delete() db.session.commit() app.logger.info("Done") app.logger.info("Deleting Variant table..") - models.Variant.query.delete() + Variant.query.delete() db.session.commit() app.logger.info("Done") @@ -164,11 +162,11 @@ def map_insert_c4r_reports(report_root_path) -> None: analysis_ptp_id = fam_dict[family_codename][ptp][0] update_stmt = ( - models.datasets_analyses_table.update() + datasets_analyses_table.update() .where( - (models.datasets_analyses_table.c.dataset_id == dataset_ptp_id) + (datasets_analyses_table.c.dataset_id == dataset_ptp_id) & ( - models.datasets_analyses_table.c.analysis_id + datasets_analyses_table.c.analysis_id == analysis_ptp_id ) ) @@ -181,8 +179,8 @@ def map_insert_c4r_reports(report_root_path) -> None: # ---- updating the result path ----- # only update if the paths don't already end in the family folder eg. .../2x/216/ if not all(ends_in_fam_folder): - analysis_query = models.Analysis.query.filter( - models.Analysis.analysis_id == family_analyses[0] + analysis_query = Analysis.query.filter( + Analysis.analysis_id == family_analyses[0] ).first() if analysis_query.result_path is not None: pprint(analysis_query.result_path) @@ -205,7 +203,7 @@ def map_insert_c4r_reports(report_root_path) -> None: # ---- variant logic ---- - variant_obj = models.Variant( + variant_obj = Variant( analysis_id=family_analyses[0], chromosome=row.get("chromosome"), position=row.get("position"), @@ -290,7 +288,7 @@ def map_insert_c4r_reports(report_root_path) -> None: row.get(col.lower()) for col in gt_cols ] - analyzed_variant_dataset = models.Genotype( + analyzed_variant_dataset = Genotype( variant_id=variant_obj.variant_id, analysis_id=family_analyses[0], dataset_id=dataset_ptp_id, diff --git a/flask/app/manage_keycloak.py b/flask/app/manage_keycloak.py index 8ec2603a..6ed2cf67 100644 --- a/flask/app/manage_keycloak.py +++ b/flask/app/manage_keycloak.py @@ -14,8 +14,7 @@ from flask import current_app as app, g from flask.cli import with_appcontext -from .extensions import db -from .models import User +from .models import db, User from .utils import stager_is_keycloak_admin keycloak_host = os.getenv("KEYCLOAK_HOST", "http://keycloak:8080") diff --git a/flask/app/mapping_utils.py b/flask/app/mapping_utils.py index e6ac1af3..f7cc767a 100644 --- a/flask/app/mapping_utils.py +++ b/flask/app/mapping_utils.py @@ -12,7 +12,7 @@ from sqlalchemy import or_ from . import models -from .extensions import db +from .models import db def try_int(value: str): diff --git a/flask/app/models.py b/flask/app/models.py index 75396ab8..012f1a93 100644 --- a/flask/app/models.py +++ b/flask/app/models.py @@ -1,15 +1,15 @@ from dataclasses import dataclass from datetime import date, datetime from enum import Enum -from requests import get from flask_login import UserMixin -from flask import current_app as app, Request +from flask_sqlalchemy import SQLAlchemy from sqlalchemy import CheckConstraint from werkzeug.security import check_password_hash, generate_password_hash -from werkzeug.exceptions import Unauthorized -from .extensions import db + +db = SQLAlchemy() + users_groups_table = db.Table( "users_groups", diff --git a/flask/app/utils.py b/flask/app/utils.py index fb4d067b..4f3dd9a1 100644 --- a/flask/app/utils.py +++ b/flask/app/utils.py @@ -15,7 +15,6 @@ Request, send_file, ) -from flask.globals import current_app from flask.json import JSONEncoder from flask_login import current_user from flask_sqlalchemy import Model @@ -26,9 +25,8 @@ from sqlalchemy.sql.sqltypes import Enum as SqlAlchemyEnum from werkzeug.exceptions import HTTPException -from .extensions import db from .madmin import MinioAdmin -from .models import User, Group, Dataset +from .models import db, User, Group, Dataset def str_to_bool(param: str) -> bool: diff --git a/flask/migrations/versions/6f448fc94a2d_many_to_many_dataset_files.py b/flask/migrations/versions/6f448fc94a2d_many_to_many_dataset_files.py index 89b18792..e85cd853 100644 --- a/flask/migrations/versions/6f448fc94a2d_many_to_many_dataset_files.py +++ b/flask/migrations/versions/6f448fc94a2d_many_to_many_dataset_files.py @@ -8,7 +8,8 @@ from alembic import op import sqlalchemy as sa -from app.extensions import db +# NOTE: DO NOT DO THIS, STATIC MIGRATIONS CANNOT DEPEND ON A CHANGING MODEL FILE +from app import db # revision identifiers, used by Alembic. diff --git a/flask/migrations/versions/b7e6ad115b13_drop_old_dataset_file_table.py b/flask/migrations/versions/b7e6ad115b13_drop_old_dataset_file_table.py index 7cbe7252..ccd5b2c3 100644 --- a/flask/migrations/versions/b7e6ad115b13_drop_old_dataset_file_table.py +++ b/flask/migrations/versions/b7e6ad115b13_drop_old_dataset_file_table.py @@ -7,8 +7,10 @@ """ from alembic import op import sqlalchemy as sa + +# NOTE: DO NOT DO THIS, STATIC MIGRATIONS CANNOT DEPEND ON A CHANGING MODEL FILE from app.models import File -from app.extensions import db +from app import db # revision identifiers, used by Alembic. revision = "b7e6ad115b13" From bc4a951d600bac5e89ab23de0aa67e8748b74b06 Mon Sep 17 00:00:00 2001 From: Kevin Lu <6320810+kevinlul@users.noreply.github.com> Date: Thu, 9 Jun 2022 14:49:53 -0400 Subject: [PATCH 10/16] Start scheduler in master --- flask/app/__init__.py | 8 -------- flask/app/stager.py | 19 ++++++++++++------- flask/gunicorn.conf.py | 19 ++++++++----------- 3 files changed, 20 insertions(+), 26 deletions(-) diff --git a/flask/app/__init__.py b/flask/app/__init__.py index c3a6ae51..f5b25dd5 100644 --- a/flask/app/__init__.py +++ b/flask/app/__init__.py @@ -44,8 +44,6 @@ def create_app(config): def config_logger(app): """ - Configure the main loggers: Flask application, SQLAlchemy, and werkzeug - SQLAlchemy logs can be very noisy and hard to filter out with grep because the logged queries can span multiple lines. However, they are still useful for auditing query efficiency and performance. Therefore, instead of using @@ -81,9 +79,3 @@ def config_logger(app): logging.getLogger("sqlalchemy").addHandler(handler) logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO) logging.getLogger("sqlalchemy.pool").setLevel(logging.INFO) - # This configures Flask's logger and then we can customize it after - flask_logging.create_logger(app) - # %(asctime)s may be useful in development but redundant in production with journald - flask_logging.default_handler.setFormatter( - logging.Formatter("%(levelname)s [%(funcName)s, line %(lineno)s]: %(message)s") - ) diff --git a/flask/app/stager.py b/flask/app/stager.py index 25cc12a5..223db83c 100644 --- a/flask/app/stager.py +++ b/flask/app/stager.py @@ -1,11 +1,11 @@ import atexit from os import getenv -import os +from logging import INFO, Formatter from apscheduler.schedulers.base import BaseScheduler from apscheduler.schedulers.background import BackgroundScheduler from authlib.integrations.flask_client import OAuth -from flask import Flask +from flask import Flask, logging as flask_logging from flask_migrate import Migrate from flask_sqlalchemy import SQLAlchemy from prometheus_flask_exporter import PrometheusMetrics @@ -28,9 +28,15 @@ class Stager(Flask): def __init__(self, config, db: SQLAlchemy, *args, **kwargs): super().__init__(*args, **kwargs) self.config.from_object(config) + # Configures Flask's logger and customize. N.B.: logging modules conflict in name + flask_logging.create_logger(self) + # %(asctime)s may be useful is redundant with Docker timestamps and production journald + flask_logging.default_handler.setFormatter( + Formatter("%(levelname)s [%(funcName)s, line %(lineno)s]: %(message)s") + ) + # Initialize extensions db.init_app(self) self.migrate = Migrate(self, db, compare_type=True) - # The rest are not required for Click commands, but required for routes, shell, tests, etc. self.oauth = OAuth(self) self.oauth.register( @@ -64,12 +70,11 @@ def start_scheduler(self): # or cumbersome, it can be separated to be started by a completely different # entrypoint in the same codebase and deployed as a separate container. self.scheduler = BackgroundScheduler() - self.scheduler.add_job( - lambda: self.logger.info(os.getpid()), "interval", seconds=5 - ) if getenv("SENDGRID_API_KEY"): self.scheduler.add_job( send_email_notification, "cron", [self], day_of_week="mon-fri", hour="9" ) self.scheduler.start() - atexit.register(self.scheduler.shutdown) + if self.env == "development": + # in production, a gunicorn exit hook will take care of this + atexit.register(self.scheduler.shutdown) diff --git a/flask/gunicorn.conf.py b/flask/gunicorn.conf.py index 80535f22..e81bc1e6 100644 --- a/flask/gunicorn.conf.py +++ b/flask/gunicorn.conf.py @@ -1,27 +1,24 @@ -from apscheduler.schedulers.background import BackgroundScheduler from prometheus_flask_exporter.multiprocess import GunicornPrometheusMetrics -from app.slurm import poll_slurm +from wsgi import app # --preload https://docs.gunicorn.org/en/stable/settings.html#preload-app -# Absolutely required to initialize the app instance and threads in the master process +# Initialize the app instance and threads in the master process. Workers are then +# forked from the master process, which does not preserve any threads. This is +# not specifically required to initialize a copy of the app instance in the master, +# as the import above will create one regardless, but in the absence of the import +# and the preload configuration, there would be no instance in the master. preload_app = True # --access-logfile - (stdout) https://docs.gunicorn.org/en/stable/settings.html#accesslog accesslog = "-" # --log-file - (stderr) https://docs.gunicorn.org/en/stable/settings.html#errorlog errorlog = "-" -scheduler = BackgroundScheduler() - def when_ready(server): GunicornPrometheusMetrics.start_http_server_when_ready(8080) - # scheduler.add_job( - # send_email_notification, "cron", [app], day_of_week="mon-fri", hour="9" - # ) - scheduler.add_job(poll_slurm, "interval", seconds=15) # match retention interval of Slurm - scheduler.start() + app.start_scheduler() def child_exit(server, worker): @@ -29,4 +26,4 @@ def child_exit(server, worker): def on_exit(server): - scheduler.shutdown() + app.scheduler.shutdown() From 737cf9a465a2e24144f6ec08fdd571094e0ad317 Mon Sep 17 00:00:00 2001 From: Kevin Lu <6320810+kevinlul@users.noreply.github.com> Date: Thu, 9 Jun 2022 15:09:58 -0400 Subject: [PATCH 11/16] Refactor email functionality into a Mailer class --- flask/app/blueprints/__init__.py | 1 + flask/app/config.py | 4 + flask/app/email.py | 215 ++++++++++++++++++++----------- flask/app/login.py | 1 + flask/app/manage.py | 5 +- flask/app/models.py | 1 + flask/app/schemas.py | 1 + flask/app/slurm.py | 12 +- flask/app/stager.py | 21 ++- flask/app/tasks.py | 67 ---------- 10 files changed, 175 insertions(+), 153 deletions(-) delete mode 100644 flask/app/tasks.py diff --git a/flask/app/blueprints/__init__.py b/flask/app/blueprints/__init__.py index 29996c12..b6f1f355 100644 --- a/flask/app/blueprints/__init__.py +++ b/flask/app/blueprints/__init__.py @@ -13,6 +13,7 @@ from .users import users_blueprint from .error_handler import error_blueprint + def register_blueprints(app: Flask) -> None: app.register_blueprint(routes) diff --git a/flask/app/config.py b/flask/app/config.py index a9e99738..7618ec15 100644 --- a/flask/app/config.py +++ b/flask/app/config.py @@ -41,4 +41,8 @@ class Config(object): "http://keycloak:8080/auth/realms/ccm/.well-known/openid-configuration", ) MSTEAMS_WEBHOOK_URL = os.getenv("MSTEAMS_WEBHOOK_URL") + SENDGRID_API_KEY = os.getenv("SENDGRID_API_KEY") + SENDGRID_EMAIL_TEMPLATE_ID = os.getenv("SENDGRID_EMAIL_TEMPLATE_ID") + SENDGRID_TO_EMAIL = os.getenv("SENDGRID_TO_EMAIL") + SENDGRID_FROM_EMAIL = os.getenv("SENDGRID_FROM_EMAIL") # LOGIN_DISABLED = True diff --git a/flask/app/email.py b/flask/app/email.py index e546197e..e81023c1 100644 --- a/flask/app/email.py +++ b/flask/app/email.py @@ -1,81 +1,148 @@ import json import math -import os from datetime import datetime, timedelta from typing import Any, Dict -from flask import current_app as app +from flask import Flask from sendgrid import SendGridAPIClient from sendgrid.helpers.mail import From, Mail, SendAt, To - - -# https://github.com/sendgrid/sendgrid-python -sg = SendGridAPIClient(os.getenv("SENDGRID_API_KEY")) - - -def send_email( - from_email: str, to_emails: str, dynamic_template_object: Dict[str, Any] -) -> None: - """Sends an email based on a template stored in SendGrid - :param dynamic_template_object: Data for a transactional template. - :type dynamic_template_object: A JSON-serializable structure - :param from: Sender of the email. The sender's email domain needs to have been authenticated and added to SendGrid Dashboard. - :type from: string - :param from: Email of the recipient - :type from: string - """ - - emails_stats = get_daily_stats() - - scheduled_time = get_send_time(emails_stats) - - message = Mail() - - message.to = To(to_emails) - message.from_email = From(from_email, "Stager Team") - message.send_at = SendAt(math.ceil(scheduled_time)) - message.dynamic_template_data = dynamic_template_object - message.template_id = os.getenv("SENDGRID_EMAIL_TEMPLATE_ID") - - try: - sg.send(message) - app.logger.debug(f"Email successfully sent from {from_email} to {to_emails}") - except Exception as e: - app.logger.error("Failed to send email...") - app.logger.error(e) - - -def get_daily_stats(): - today = datetime.now().strftime("%Y-%m-%d") - params = { - "aggregated_by": "day", - "start_date": today, - "end_date": today, - "offset": 1, - } - try: - response = sg.client.stats.get(query_params=params) - return json.loads(response.body.decode("utf-8")) - except Exception as e: - app.logger.error("Failed to get daily email stats") - app.logger.debug(e) - return [] - - -def get_send_time(stats): - limit_per_day = 100 - send_at = datetime.now() - - for stat in stats: - - requests_count = sum( - [r.get("metrics").get("requests") for r in stat.get("stats")] - ) - - if stat.get("date") == datetime.strftime(send_at, "%Y-%m-%d"): - if requests_count < limit_per_day: - send_at = send_at + timedelta(seconds=15) - else: - send_at = send_at + timedelta(days=1, seconds=15) - - return send_at.timestamp() +from sqlalchemy.orm import joinedload + + +from . import models + + +class Mailer: + def __init__(self, app: Flask): + self.app = app + # https://github.com/sendgrid/sendgrid-python + self.sg = SendGridAPIClient(app.config["SENDGRID_API_KEY"]) + + def send( + self, from_email: str, to_emails: str, dynamic_template_object: Dict[str, Any] + ) -> None: + """Sends an email based on a template stored in SendGrid + :param dynamic_template_object: Data for a transactional template. + :type dynamic_template_object: A JSON-serializable structure + :param from: Sender of the email. The sender's email domain needs to have been authenticated and added to SendGrid Dashboard. + :type from: string + :param from: Email of the recipient + :type from: string + """ + + emails_stats = self._get_daily_stats() + + scheduled_time = self._get_send_time(emails_stats) + + message = Mail() + + message.to = To(to_emails) + message.from_email = From(from_email, "Stager Team") + message.send_at = SendAt(math.ceil(scheduled_time)) + message.dynamic_template_data = dynamic_template_object + message.template_id = self.app.config["SENDGRID_EMAIL_TEMPLATE_ID"] + + try: + self.sg.send(message) + self.app.logger.debug( + f"Email successfully sent from {from_email} to {to_emails}" + ) + except Exception as e: + self.app.logger.error("Failed to send email...") + self.app.logger.error(e) + + def _get_daily_stats(self): + today = datetime.now().strftime("%Y-%m-%d") + params = { + "aggregated_by": "day", + "start_date": today, + "end_date": today, + "offset": 1, + } + try: + response = self.sg.client.stats.get(query_params=params) + return json.loads(response.body.decode("utf-8")) + except Exception as e: + self.app.logger.error("Failed to get daily email stats") + self.app.logger.debug(e) + return [] + + def _get_send_time(self, stats): + limit_per_day = 100 + send_at = datetime.now() + + for stat in stats: + + requests_count = sum( + [r.get("metrics").get("requests") for r in stat.get("stats")] + ) + + if stat.get("date") == datetime.strftime(send_at, "%Y-%m-%d"): + if requests_count < limit_per_day: + send_at = send_at + timedelta(seconds=15) + else: + send_at = send_at + timedelta(days=1, seconds=15) + + return send_at.timestamp() + + def send_notification(self): + """ + This is a scheduled background task. Because it runs in a separate thread, + it pushes an app context for itself. + """ + with self.app.app_context(): + yesterday = (datetime.now() - timedelta(1)).strftime("%Y-%m-%d") + analyses = ( + models.Analysis.query.options( + joinedload(models.Analysis.datasets).joinedload( + models.Dataset.linked_files + ), + joinedload(models.Analysis.requester), + joinedload(models.Analysis.datasets) + .joinedload(models.Dataset.tissue_sample) + .joinedload(models.TissueSample.participant) + .joinedload(models.Participant.family), + ) + .filter(models.Analysis.requested >= yesterday) + .all() + ) + + email_analyses = [] + + for analysis in analyses: + email_analyses.append( + { + "analysis_id": analysis.analysis_id, + "requested": analysis.requested.strftime("%Y-%m-%d"), + "requester": analysis.requester.username, + "pipeline": analysis.kind, + "priority": analysis.priority.value, + "datasets": [ + { + "dataset_id": dataset.dataset_id, + "notes": dataset.notes or "", + "linked_files": ", ".join( + [file.path for file in dataset.linked_files] + ), + "participant_codename": dataset.tissue_sample.participant.participant_codename, + "participant_aliases": dataset.tissue_sample.participant.participant_aliases + or "", + "family_codename": dataset.tissue_sample.participant.family.family_codename, + "participant_notes": dataset.tissue_sample.participant.notes + or "", + } + for dataset in analysis.datasets + ], + } + ) + + if len(email_analyses) > 0: + self.send( + to_emails=self.app.config["SENDGRID_TO_EMAIL"], + from_email=self.app.config["SENDGRID_FROM_EMAIL"], + dynamic_template_object={"analyses": email_analyses}, + ) + + self.app.logger.debug( + f"{len(email_analyses)} analysis requests found... {json.dumps(email_analyses)}" + ) diff --git a/flask/app/login.py b/flask/app/login.py index c64a3587..fab7828f 100644 --- a/flask/app/login.py +++ b/flask/app/login.py @@ -52,6 +52,7 @@ class StagerLoginManager(LoginManager): """ Flask-Login manager. Must be initialized with a Stager Flask instance as it depends on app.oauth """ + session_protection = "strong" def __init__(self, stager): diff --git a/flask/app/manage.py b/flask/app/manage.py index 5eda70f0..699ad099 100644 --- a/flask/app/manage.py +++ b/flask/app/manage.py @@ -165,10 +165,7 @@ def map_insert_c4r_reports(report_root_path) -> None: datasets_analyses_table.update() .where( (datasets_analyses_table.c.dataset_id == dataset_ptp_id) - & ( - datasets_analyses_table.c.analysis_id - == analysis_ptp_id - ) + & (datasets_analyses_table.c.analysis_id == analysis_ptp_id) ) .values(analysis_id=family_analyses[0]) ) diff --git a/flask/app/models.py b/flask/app/models.py index 012f1a93..a3b39ce5 100644 --- a/flask/app/models.py +++ b/flask/app/models.py @@ -50,6 +50,7 @@ def set_oidc_fields(self, issuer: str, subject: str): self.issuer = issuer self.subject = subject + @dataclass class Group(db.Model): group_id: int = db.Column(db.Integer, primary_key=True) diff --git a/flask/app/schemas.py b/flask/app/schemas.py index 8af30503..2c18a43c 100644 --- a/flask/app/schemas.py +++ b/flask/app/schemas.py @@ -7,6 +7,7 @@ # TODO: use flask_marshmallow's instance if we use these to serialize # https://flask-marshmallow.readthedocs.io/en/latest/ + class FamilySchema(SQLAlchemyAutoSchema): """ POST /api/families diff --git a/flask/app/slurm.py b/flask/app/slurm.py index 83d0bf6b..71601127 100644 --- a/flask/app/slurm.py +++ b/flask/app/slurm.py @@ -78,7 +78,9 @@ def poll_slurm() -> None: Analysis.scheduler_id != None, ).all() if len(running_analyses): - analyses = { analysis.scheduler_id: analysis for analysis in running_analyses } + analyses = { + analysis.scheduler_id: analysis for analysis in running_analyses + } requests = [ api_instance.slurmctld_get_job(analysis.scheduler_id, async_req=True) for analysis in running_analyses @@ -101,8 +103,12 @@ def poll_slurm() -> None: ]: analyses[job.jobs[0].job_id].analysis_state = AnalysisState.Error analyses[job.jobs[0].job_id].result_path = job.jobs[0].job_state - analyses[job.jobs[0].job_id].finished = job.jobs[0].end_time # int64, convert to datetime? + analyses[job.jobs[0].job_id].finished = job.jobs[ + 0 + ].end_time # int64, convert to datetime? elif job.jobs[0].job_state == "COMPLETED": analyses[job.jobs[0].job_id].analysis_state = AnalysisState.Done - analyses[job.jobs[0].job_id].finished = job.jobs[0].end_time # int64, convert to datetime? + analyses[job.jobs[0].job_id].finished = job.jobs[ + 0 + ].end_time # int64, convert to datetime? db.session.commit() diff --git a/flask/app/stager.py b/flask/app/stager.py index 223db83c..7dae861a 100644 --- a/flask/app/stager.py +++ b/flask/app/stager.py @@ -1,6 +1,5 @@ import atexit -from os import getenv -from logging import INFO, Formatter +from logging import Formatter from apscheduler.schedulers.base import BaseScheduler from apscheduler.schedulers.background import BackgroundScheduler @@ -11,8 +10,8 @@ from prometheus_flask_exporter import PrometheusMetrics from prometheus_flask_exporter.multiprocess import GunicornPrometheusMetrics +from .email import Mailer from .login import StagerLoginManager -from .tasks import send_email_notification from .utils import DateTimeEncoder @@ -23,6 +22,7 @@ class Stager(Flask): json_encoder = DateTimeEncoder # Only available in master process + mailer: Mailer # Only available if configured scheduler: BaseScheduler def __init__(self, config, db: SQLAlchemy, *args, **kwargs): @@ -70,9 +70,20 @@ def start_scheduler(self): # or cumbersome, it can be separated to be started by a completely different # entrypoint in the same codebase and deployed as a separate container. self.scheduler = BackgroundScheduler() - if getenv("SENDGRID_API_KEY"): + if self.config.get("SENDGRID_API_KEY"): + self.logger.info( + "Configuring with SendGrid [%s] (from: %s) (to: %s)", + self.config["SENDGRID_EMAIL_TEMPLATE_ID"], + self.config["SENDGRID_FROM_EMAIL"], + self.config["SENDGRID_TO_EMAIL"], + ) + self.mailer = Mailer(self) self.scheduler.add_job( - send_email_notification, "cron", [self], day_of_week="mon-fri", hour="9" + self.mailer.send_notification, + "cron", + [self], + day_of_week="mon-fri", + hour="9", ) self.scheduler.start() if self.env == "development": diff --git a/flask/app/tasks.py b/flask/app/tasks.py deleted file mode 100644 index 2520359a..00000000 --- a/flask/app/tasks.py +++ /dev/null @@ -1,67 +0,0 @@ -import json -import os -from datetime import datetime, timedelta - -from sqlalchemy.orm import joinedload - -from . import models -from .email import send_email - - -def send_email_notification(app): - with app.app_context(): - yesterday = (datetime.now() - timedelta(1)).strftime("%Y-%m-%d") - analyses = ( - models.Analysis.query.options( - joinedload(models.Analysis.datasets).joinedload( - models.Dataset.linked_files - ), - joinedload(models.Analysis.requester), - joinedload(models.Analysis.datasets) - .joinedload(models.Dataset.tissue_sample) - .joinedload(models.TissueSample.participant) - .joinedload(models.Participant.family), - ) - .filter(models.Analysis.requested >= yesterday) - .all() - ) - - email_analyses = [] - - for analysis in analyses: - email_analyses.append( - { - "analysis_id": analysis.analysis_id, - "requested": analysis.requested.strftime("%Y-%m-%d"), - "requester": analysis.requester.username, - "pipeline": analysis.kind, - "priority": analysis.priority.value, - "datasets": [ - { - "dataset_id": dataset.dataset_id, - "notes": dataset.notes or "", - "linked_files": ", ".join( - [file.path for file in dataset.linked_files] - ), - "participant_codename": dataset.tissue_sample.participant.participant_codename, - "participant_aliases": dataset.tissue_sample.participant.participant_aliases - or "", - "family_codename": dataset.tissue_sample.participant.family.family_codename, - "participant_notes": dataset.tissue_sample.participant.notes - or "", - } - for dataset in analysis.datasets - ], - } - ) - - if len(email_analyses) > 0: - send_email( - to_emails=os.getenv("SENDGRID_TO_EMAIL"), - from_email=os.getenv("SENDGRID_FROM_EMAIL"), - dynamic_template_object={"analyses": email_analyses}, - ) - - app.logger.debug( - f"{len(email_analyses)} analysis requests found... {json.dumps(email_analyses)}" - ) From 5e55a35b762746a1c73c89fb7ced245ebb9153ac Mon Sep 17 00:00:00 2001 From: Kevin Lu <6320810+kevinlul@users.noreply.github.com> Date: Thu, 9 Jun 2022 15:17:40 -0400 Subject: [PATCH 12/16] Fix test imports as a result of blueprints refactor --- flask/tests/test_datasets.py | 2 +- flask/tests/test_misc.py | 3 +-- flask/tests/test_users.py | 4 ++-- 3 files changed, 4 insertions(+), 5 deletions(-) diff --git a/flask/tests/test_datasets.py b/flask/tests/test_datasets.py index 41058a8e..1f432b15 100644 --- a/flask/tests/test_datasets.py +++ b/flask/tests/test_datasets.py @@ -3,7 +3,7 @@ from sqlalchemy.orm import joinedload from app import db, models -from app.datasets import update_dataset_linked_files +from app.blueprints.datasets import update_dataset_linked_files # TODO: some tests do not precisely verify response structure diff --git a/flask/tests/test_misc.py b/flask/tests/test_misc.py index e12c0e25..90e9eb4a 100644 --- a/flask/tests/test_misc.py +++ b/flask/tests/test_misc.py @@ -3,12 +3,11 @@ from io import StringIO from pytest import raises from flask.wrappers import Request -from flask import request from sqlalchemy.orm import joinedload from werkzeug.exceptions import BadRequest from app import models, db -from app.routes import link_files_to_dataset +from app.blueprints.misc import link_files_to_dataset from app.utils import filter_datasets_by_user_groups, get_current_user diff --git a/flask/tests/test_users.py b/flask/tests/test_users.py index 6073face..c74f5346 100644 --- a/flask/tests/test_users.py +++ b/flask/tests/test_users.py @@ -1,13 +1,13 @@ -from app.users import reset_minio_credentials import json from io import BytesIO from minio import Minio - import pytest + from app import db from app.madmin import MinioAdmin, stager_buckets_policy from app.models import User, Group +from app.blueprints.users import reset_minio_credentials from conftest import TestConfig # Common response values between list and individual get endpoints From 5a6c22ce9543acf20bf1c574b95eb76014e39e7d Mon Sep 17 00:00:00 2001 From: Kevin Lu <6320810+kevinlul@users.noreply.github.com> Date: Thu, 9 Jun 2022 15:39:05 -0400 Subject: [PATCH 13/16] Initialize Slurm clients at top level --- flask/app/__init__.py | 22 -------------- flask/app/slurm.py | 69 ++++++++++++++++++++++--------------------- flask/app/stager.py | 27 +++++++++++++++-- 3 files changed, 61 insertions(+), 57 deletions(-) diff --git a/flask/app/__init__.py b/flask/app/__init__.py index f5b25dd5..0457e7a6 100644 --- a/flask/app/__init__.py +++ b/flask/app/__init__.py @@ -2,10 +2,6 @@ import os from stat import S_ISFIFO -from flask import logging as flask_logging -from slurm_rest import Configuration - - from .blueprints import register_blueprints from .manage import register_commands from .models import db @@ -18,27 +14,9 @@ def create_app(config): """ # Create the application object app = Stager(config, db, __name__) - config_logger(app) - - if app.config["SLURM_ENDPOINT"]: - app.logger.info( - "Configuring with Slurm REST API %s", app.config["SLURM_ENDPOINT"] - ) - # Could instead use one environment variable and urllib.parse.urlsplit for this - app.config["slurm"] = Configuration( - host=app.config["SLURM_ENDPOINT"], - api_key={ - "user": app.config["SLURM_USER"], - "token": app.config["SLURM_JWT"], - }, - ) - else: - app.config["slurm"] = None - register_commands(app) register_blueprints(app) - return app diff --git a/flask/app/slurm.py b/flask/app/slurm.py index 71601127..ccc6d979 100644 --- a/flask/app/slurm.py +++ b/flask/app/slurm.py @@ -1,8 +1,8 @@ import json from typing import List, Optional -from flask import current_app as app -from slurm_rest import ApiClient, ApiException +from flask import current_app as app, Flask +from slurm_rest import ApiException from slurm_rest.apis import SlurmApi from slurm_rest.models import ( V0037JobSubmission, @@ -11,7 +11,7 @@ V0037JobsResponse, ) -from .models import Analysis, AnalysisState +from .models import db, Analysis, AnalysisState # Slurm notes: @@ -38,41 +38,44 @@ def run_crg2_on_family(analysis: Analysis) -> Optional[V0037JobSubmissionRespons } # Will only be used for capturing stdout/stderr instead of explicitly for each cwd = app.config["SLURM_PWD"] - with ApiClient(app.config["slurm"]) as api_client: - api_instance = SlurmApi(api_client) - try: - # This should already be safely shell-escaped so there's no arbitrary code execution - # but if there are further issues then pass the user inputs in through the environment - submitted_job = api_instance.slurmctld_submit_job( - V0037JobSubmission( - script=f"""#!/bin/bash + api_instance: SlurmApi = app.extensions["slurm"] + try: + # This should already be safely shell-escaped so there's no arbitrary code execution + # but if there are further issues then pass the user inputs in through the environment + submitted_job = api_instance.slurmctld_submit_job( + V0037JobSubmission( + script=f"""#!/bin/bash exec '{app.config["CRG2_ENTRYPOINT"]}' {analysis.analysis_id} '{family_codename}' '{json.dumps(files)}' """, - job=V0037JobProperties( - environment={"STAGER": True}, - current_working_directory=cwd, - name=f"Stager-CRG2 (analysis {analysis.analysis_id}, family {family_codename})", - standard_output=f"stager-crg2-{analysis.analysis_id}.out", - memory_per_node=4096, # MB, equivalent to --mem and SBATCH_MEM_PER_NODE - time_limit=3000, # minutes, 50 hours, equivalent to --time and SBATCH_TIMELIMIT - # partition, nodes, and CPUs are left implied - ), - ) - ) - app.logger.info( - f"Submitted analysis {analysis.analysis_id} to scheduler: {submitted_job}" - ) - return submitted_job - except ApiException as e: - app.logger.warn( - f"Exception when calling slurmctld_submit_job for analysis {analysis.analysis_id}", - exc_info=e, + job=V0037JobProperties( + environment={"STAGER": True}, + current_working_directory=cwd, + name=f"Stager-CRG2 (analysis {analysis.analysis_id}, family {family_codename})", + standard_output=f"stager-crg2-{analysis.analysis_id}.out", + memory_per_node=4096, # MB, equivalent to --mem and SBATCH_MEM_PER_NODE + time_limit=3000, # minutes, 50 hours, equivalent to --time and SBATCH_TIMELIMIT + # partition, nodes, and CPUs are left implied + ), ) + ) + app.logger.info( + f"Submitted analysis {analysis.analysis_id} to scheduler: {submitted_job}" + ) + return submitted_job + except ApiException as e: + app.logger.warn( + f"Exception when calling slurmctld_submit_job for analysis {analysis.analysis_id}", + exc_info=e, + ) -def poll_slurm() -> None: - with ApiClient(app.config["slurm"]) as api_client: - api_instance = SlurmApi(api_client) +def poll_slurm(app: Flask) -> None: + """ + This is a scheduled background task. Because it runs in a separate thread, + it pushes an app context for itself. + """ + with app.app_context(): + api_instance: SlurmApi = app.extensions["slurm"] running_analyses = Analysis.query.filter( Analysis.analysis_state == AnalysisState.Running, Analysis.scheduler_id != None, diff --git a/flask/app/stager.py b/flask/app/stager.py index 7dae861a..392f322f 100644 --- a/flask/app/stager.py +++ b/flask/app/stager.py @@ -9,10 +9,13 @@ from flask_sqlalchemy import SQLAlchemy from prometheus_flask_exporter import PrometheusMetrics from prometheus_flask_exporter.multiprocess import GunicornPrometheusMetrics +from slurm_rest import Configuration, ApiClient +from slurm_rest.apis import SlurmApi from .email import Mailer from .login import StagerLoginManager from .utils import DateTimeEncoder +from .slurm import poll_slurm class Stager(Flask): @@ -48,6 +51,25 @@ def __init__(self, config, db: SQLAlchemy, *args, **kwargs): ) self.login = StagerLoginManager(self) + if self.config["SLURM_ENDPOINT"]: + self.logger.info( + "Configuring with Slurm REST API %s", self.config["SLURM_ENDPOINT"] + ) + # Could instead use one environment variable and urllib.parse.urlsplit for this + self.config["slurm"] = Configuration( + host=self.config["SLURM_ENDPOINT"], + api_key={ + "user": self.config["SLURM_USER"], + "token": self.config["SLURM_JWT"], + }, + ) + # The thread pool used for requests is not instantiated until first use, so it is + # safe to construct this object in the master process pre-fork. + self.extensions["slurm"] = SlurmApi(ApiClient(self.config["slurm"])) + # Access ApiClient if needed with .api_client + else: + self.config["slurm"] = self.extensions["slurm"] = None + if self.env == "development": # Note that /metrics will not be live without DEBUG_METRICS set self.metrics = PrometheusMetrics(self) @@ -70,7 +92,7 @@ def start_scheduler(self): # or cumbersome, it can be separated to be started by a completely different # entrypoint in the same codebase and deployed as a separate container. self.scheduler = BackgroundScheduler() - if self.config.get("SENDGRID_API_KEY"): + if self.config["SENDGRID_API_KEY"]: self.logger.info( "Configuring with SendGrid [%s] (from: %s) (to: %s)", self.config["SENDGRID_EMAIL_TEMPLATE_ID"], @@ -81,10 +103,11 @@ def start_scheduler(self): self.scheduler.add_job( self.mailer.send_notification, "cron", - [self], day_of_week="mon-fri", hour="9", ) + if self.config["SLURM_JWT"]: + self.scheduler.add_job(poll_slurm, "interval", [self], minutes=5) self.scheduler.start() if self.env == "development": # in production, a gunicorn exit hook will take care of this From 19e7826e6b75c258b1fae699fe048601afa39a43 Mon Sep 17 00:00:00 2001 From: Kevin Lu <6320810+kevinlul@users.noreply.github.com> Date: Thu, 9 Jun 2022 17:20:23 -0400 Subject: [PATCH 14/16] Poll with requests due to Slurm not respecting own schema --- flask/app/slurm.py | 55 +++++++++++++++++++++++++++------------------ flask/app/stager.py | 6 ++++- 2 files changed, 38 insertions(+), 23 deletions(-) diff --git a/flask/app/slurm.py b/flask/app/slurm.py index ccc6d979..df7bd689 100644 --- a/flask/app/slurm.py +++ b/flask/app/slurm.py @@ -1,14 +1,15 @@ +from datetime import datetime import json -from typing import List, Optional +from typing import Optional from flask import current_app as app, Flask +from requests import Session from slurm_rest import ApiException from slurm_rest.apis import SlurmApi from slurm_rest.models import ( V0037JobSubmission, V0037JobProperties, V0037JobSubmissionResponse, - V0037JobsResponse, ) from .models import db, Analysis, AnalysisState @@ -75,26 +76,38 @@ def poll_slurm(app: Flask) -> None: it pushes an app context for itself. """ with app.app_context(): - api_instance: SlurmApi = app.extensions["slurm"] running_analyses = Analysis.query.filter( Analysis.analysis_state == AnalysisState.Running, Analysis.scheduler_id != None, ).all() + app.logger.info(f"Found {len(running_analyses)} running analyses to poll.") if len(running_analyses): analyses = { analysis.scheduler_id: analysis for analysis in running_analyses } - requests = [ - api_instance.slurmctld_get_job(analysis.scheduler_id, async_req=True) - for analysis in running_analyses - ] - jobs: List[V0037JobsResponse] = [ - async_result.get() for async_result in requests - ] - for job in jobs: + # api_instance.slurmctld_get_job and api_instance.slurmctld_get_jobs + # are broken due to Slurm not respecting its own OpenAPI schema on the + # typing of .array_job_id (returns an int 0 for a string type) + # The autogenerated V0037JobsResponse will reject this + # api_instance: SlurmApi = app.extensions["slurm"] + session: Session = app.extensions["slurm-requests"] + response = session.get(app.config["SLURM_ENDPOINT"] + "/slurm/v0.0.37/jobs", headers={ + "X-SLURM-USER-NAME": app.config["SLURM_USER"], + "X-SLURM-USER-TOKEN": app.config["SLURM_JWT"] + }) + response.raise_for_status() + result = response.json() + if len(result["errors"]) > 0: + app.logger.warning(result["errors"]) + for job in result["jobs"]: + job_id = job["job_id"] + if job_id not in analyses: + continue + job_state = job["job_state"] + end_time = datetime.fromtimestamp(job["end_time"]) # V0037JobResponseProperties # https://slurm.schedmd.com/squeue.html#SECTION_JOB-STATE-CODES - if job.jobs[0].job_state in [ + if job_state in [ "BOOT_FAIL", "CANCELLED", "DEADLINE", @@ -104,14 +117,12 @@ def poll_slurm(app: Flask) -> None: "PREEMPTED", "TIMEOUT", ]: - analyses[job.jobs[0].job_id].analysis_state = AnalysisState.Error - analyses[job.jobs[0].job_id].result_path = job.jobs[0].job_state - analyses[job.jobs[0].job_id].finished = job.jobs[ - 0 - ].end_time # int64, convert to datetime? - elif job.jobs[0].job_state == "COMPLETED": - analyses[job.jobs[0].job_id].analysis_state = AnalysisState.Done - analyses[job.jobs[0].job_id].finished = job.jobs[ - 0 - ].end_time # int64, convert to datetime? + analyses[job_id].analysis_state = AnalysisState.Error + analyses[job_id].result_path = f"{job_state} {analyses[job_id].result_path}" + analyses[job_id].finished = end_time + app.logger.warning(f"Slurm {job_id} (analysis {analyses[job_id].analysis_id}): {job_state}") + elif job_state == "COMPLETED": + analyses[job_id].analysis_state = AnalysisState.Done + analyses[job_id].finished = end_time + app.logger.info(f"Slurm {job_id} (analysis {analyses[job_id].analysis_id}): {job_state}") db.session.commit() diff --git a/flask/app/stager.py b/flask/app/stager.py index 392f322f..1fa0e8bd 100644 --- a/flask/app/stager.py +++ b/flask/app/stager.py @@ -9,6 +9,7 @@ from flask_sqlalchemy import SQLAlchemy from prometheus_flask_exporter import PrometheusMetrics from prometheus_flask_exporter.multiprocess import GunicornPrometheusMetrics +from requests import Session from slurm_rest import Configuration, ApiClient from slurm_rest.apis import SlurmApi @@ -107,7 +108,10 @@ def start_scheduler(self): hour="9", ) if self.config["SLURM_JWT"]: - self.scheduler.add_job(poll_slurm, "interval", [self], minutes=5) + # requests session used to bypass the SDK when Slurm doesn't respect + # its own API Schema (e.g. job response properties .array_job_id) + self.extensions["slurm-requests"] = Session() + self.scheduler.add_job(poll_slurm, "interval", [self], seconds=10) self.scheduler.start() if self.env == "development": # in production, a gunicorn exit hook will take care of this From 17068a619a8ef35bd680e97568ee04f45cc70df9 Mon Sep 17 00:00:00 2001 From: Kevin Lu <6320810+kevinlul@users.noreply.github.com> Date: Thu, 9 Jun 2022 17:27:12 -0400 Subject: [PATCH 15/16] blacken --- flask/app/slurm.py | 20 +++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/flask/app/slurm.py b/flask/app/slurm.py index df7bd689..ae4fa29c 100644 --- a/flask/app/slurm.py +++ b/flask/app/slurm.py @@ -91,10 +91,13 @@ def poll_slurm(app: Flask) -> None: # The autogenerated V0037JobsResponse will reject this # api_instance: SlurmApi = app.extensions["slurm"] session: Session = app.extensions["slurm-requests"] - response = session.get(app.config["SLURM_ENDPOINT"] + "/slurm/v0.0.37/jobs", headers={ - "X-SLURM-USER-NAME": app.config["SLURM_USER"], - "X-SLURM-USER-TOKEN": app.config["SLURM_JWT"] - }) + response = session.get( + app.config["SLURM_ENDPOINT"] + "/slurm/v0.0.37/jobs", + headers={ + "X-SLURM-USER-NAME": app.config["SLURM_USER"], + "X-SLURM-USER-TOKEN": app.config["SLURM_JWT"], + }, + ) response.raise_for_status() result = response.json() if len(result["errors"]) > 0: @@ -118,11 +121,14 @@ def poll_slurm(app: Flask) -> None: "TIMEOUT", ]: analyses[job_id].analysis_state = AnalysisState.Error - analyses[job_id].result_path = f"{job_state} {analyses[job_id].result_path}" analyses[job_id].finished = end_time - app.logger.warning(f"Slurm {job_id} (analysis {analyses[job_id].analysis_id}): {job_state}") + app.logger.warning( + f"Slurm {job_id} (analysis {analyses[job_id].analysis_id}): {job_state}" + ) elif job_state == "COMPLETED": analyses[job_id].analysis_state = AnalysisState.Done analyses[job_id].finished = end_time - app.logger.info(f"Slurm {job_id} (analysis {analyses[job_id].analysis_id}): {job_state}") + app.logger.info( + f"Slurm {job_id} (analysis {analyses[job_id].analysis_id}): {job_state}" + ) db.session.commit() From 610c951bb4fc282eccc53b71414619d930ba0004 Mon Sep 17 00:00:00 2001 From: Kevin Lu <6320810+kevinlul@users.noreply.github.com> Date: Thu, 9 Jun 2022 17:32:03 -0400 Subject: [PATCH 16/16] Change poll interval from development frequency --- flask/app/stager.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/flask/app/stager.py b/flask/app/stager.py index 1fa0e8bd..9f4df238 100644 --- a/flask/app/stager.py +++ b/flask/app/stager.py @@ -111,7 +111,7 @@ def start_scheduler(self): # requests session used to bypass the SDK when Slurm doesn't respect # its own API Schema (e.g. job response properties .array_job_id) self.extensions["slurm-requests"] = Session() - self.scheduler.add_job(poll_slurm, "interval", [self], seconds=10) + self.scheduler.add_job(poll_slurm, "interval", [self], minutes=2) self.scheduler.start() if self.env == "development": # in production, a gunicorn exit hook will take care of this