more wip

add more prometheus metrics

app/__init__.py

@@ -1,19 +1,23 @@
+import time
 import os
 import random
 import string
 import uuid
 
-from flask import _request_ctx_stack, request, g, jsonify, make_response
+from celery import current_task
+from flask import _request_ctx_stack, request, g, jsonify, make_response, current_app, has_request_context
 from flask_sqlalchemy import SQLAlchemy as _SQLAlchemy
 from flask_marshmallow import Marshmallow
 from flask_migrate import Migrate
 from gds_metrics import GDSMetrics
+from gds_metrics.metrics import Gauge, Histogram
 from time import monotonic
 from notifications_utils.clients.zendesk.zendesk_client import ZendeskClient
 from notifications_utils.clients.statsd.statsd_client import StatsdClient
 from notifications_utils.clients.redis.redis_client import RedisClient
 from notifications_utils.clients.encryption.encryption_client import Encryption
 from notifications_utils import logging, request_helper
+from sqlalchemy import event
 from werkzeug.exceptions import HTTPException as WerkzeugHTTPException
 from werkzeug.local import LocalProxy
 
@@ -64,6 +68,11 @@ clients = Clients()
 api_user = LocalProxy(lambda: _request_ctx_stack.top.api_user)
 authenticated_service = LocalProxy(lambda: _request_ctx_stack.top.authenticated_service)
 
+CONCURRENT_REQUESTS = Gauge(
+    'concurrent_web_request_count',
+    'How many concurrent requests are currently being served',
+)
+
 
 def create_app(application):
     from app.config import configs
@@ -108,6 +117,9 @@ def create_app(application):
     from app.commands import setup_commands
     setup_commands(application)
 
+    # set up sqlalchemy events
+    setup_sqlalchemy_events(application)
+
     return application
 
 
@@ -255,17 +267,22 @@ def register_v2_blueprints(application):
 
 
 def init_app(app):
+
     @app.before_request
     def record_user_agent():
         statsd_client.incr("user-agent.{}".format(process_user_agent(request.headers.get('User-Agent', None))))
 
     @app.before_request
     def record_request_details():
+        CONCURRENT_REQUESTS.inc()
+
         g.start = monotonic()
         g.endpoint = request.endpoint
 
     @app.after_request
     def after_request(response):
+        CONCURRENT_REQUESTS.dec()
+
         response.headers.add('Access-Control-Allow-Origin', '*')
         response.headers.add('Access-Control-Allow-Headers', 'Content-Type,Authorization')
         response.headers.add('Access-Control-Allow-Methods', 'GET,PUT,POST,DELETE')
@@ -311,3 +328,83 @@ def process_user_agent(user_agent_string):
         return "non-notify-user-agent"
     else:
         return "unknown"
+
+
+def setup_sqlalchemy_events(app):
+
+    TOTAL_DB_CONNECTIONS = Gauge(
+        'db_connection_total_connected',
+        'How many db connections are currently held (potentially idle) by the server',
+    )
+
+    TOTAL_CHECKED_OUT_DB_CONNECTIONS = Gauge(
+        'db_connection_total_checked_out',
+        'How many db connections are currently checked out by web requests',
+    )
+
+    DB_CONNECTION_OPEN_DURATION_SECONDS = Histogram(
+        'db_connection_open_duration_seconds',
+        'How long db connections are held open for in seconds',
+        ['method', 'host', 'path']
+    )
+
+    # need this or db.engine isn't accessible
+    with app.app_context():
+        @event.listens_for(db.engine, 'connect')
+        def connect(dbapi_connection, connection_record):
+            # connection first opened with db
+            TOTAL_DB_CONNECTIONS.inc()
+
+        @event.listens_for(db.engine, 'close')
+        def close(dbapi_connection, connection_record):
+            # connection closed (probably only happens with overflow connections)
+            TOTAL_DB_CONNECTIONS.dec()
+
+        @event.listens_for(db.engine, 'checkout')
+        def checkout(dbapi_connection, connection_record, connection_proxy):
+            # connection given to a web worker
+            TOTAL_CHECKED_OUT_DB_CONNECTIONS.inc()
+
+            # this will overwrite any previous checkout_at timestamp
+            connection_record.info['checkout_at'] = time.monotonic()
+
+            # checkin runs after the request is already torn down, therefore we add the request_data onto the
+            # connection_record as otherwise it won't have that information when checkin actually runs.
+            # Note: this is not a problem for checkouts as the checkout always happens within a web request or task
+
+            # web requests
+            if has_request_context():
+                connection_record.info['request_data'] = {
+                    'method': request.method,
+                    'host': request.host,
+                    'url_rule': request.url_rule.rule if request.url_rule else 'No endpoint'
+                }
+            # celery apps
+            elif current_task:
+                connection_record.info['request_data'] = {
+                    'method': 'celery',
+                    'host': current_app.config['NOTIFY_APP_NAME'],  # worker name
+                    'url_rule': current_task.name,  # task name
+                }
+            # anything else. migrations possibly.
+            else:
+                current_app.logger.warning('Checked out sqlalchemy connection from outside of request/task')
+                connection_record.info['request_data'] = {
+                    'method': 'unknown',
+                    'host': 'unknown',
+                    'url_rule': 'unknown',
+                }
+
+        @event.listens_for(db.engine, 'checkin')
+        def checkin(dbapi_connection, connection_record):
+            # connection returned by a web worker
+            TOTAL_CHECKED_OUT_DB_CONNECTIONS.dec()
+
+            # duration that connection was held by a single web request
+            duration = time.monotonic() - connection_record.info['checkout_at']
+
+            DB_CONNECTION_OPEN_DURATION_SECONDS.labels(
+                connection_record.info['request_data']['method'],
+                connection_record.info['request_data']['host'],
+                connection_record.info['request_data']['url_rule']
+            ).observe(duration)

app/authentication/auth.py

@@ -6,12 +6,18 @@ from notifications_python_client.errors import (
 from notifications_utils import request_helper
 from sqlalchemy.exc import DataError
 from sqlalchemy.orm.exc import NoResultFound
+from gds_metrics import Histogram
 
 from app.dao.services_dao import dao_fetch_service_by_id_with_api_keys
 
 
 GENERAL_TOKEN_ERROR_MESSAGE = 'Invalid token: make sure your API token matches the example at https://docs.notifications.service.gov.uk/rest-api.html#authorisation-header'  # noqa
 
+AUTH_DB_CONNECTION_DURATION_SECONDS = Histogram(
+    'auth_db_connection_duration_seconds',
+    'Time taken to get DB connection and fetch service from database',
+)
+
 
 class AuthError(Exception):
     def __init__(self, message, code, service_id=None, api_key_id=None):
@@ -87,7 +93,8 @@ def requires_auth():
     issuer = __get_token_issuer(auth_token)  # ie the `iss` claim which should be a service ID
 
     try:
-        service = dao_fetch_service_by_id_with_api_keys(issuer)
+        with AUTH_DB_CONNECTION_DURATION_SECONDS.time():
+            service = dao_fetch_service_by_id_with_api_keys(issuer)
     except DataError:
         raise AuthError("Invalid token: service id is not the right data type", 403)
     except NoResultFound:

app/celery/celery.py

@@ -1,5 +1,6 @@
 import time
 
+from gds_metrics.metrics import Histogram
 from celery import Celery, Task
 from celery.signals import worker_process_shutdown
 from flask import g, request
@@ -19,6 +20,12 @@ def log_on_worker_shutdown(sender, signal, pid, exitcode, **kwargs):
 
 
 def make_task(app):
+    SQS_APPLY_ASYNC_DURATION_SECONDS = Histogram(
+        'sqs_apply_async_duration_seconds',
+        'Time taken to put task on queue',
+        ['task_name']
+    )
+
     class NotifyTask(Task):
         abstract = True
         start = None
@@ -52,7 +59,8 @@ def make_task(app):
             if has_request_context() and hasattr(request, 'request_id'):
                 kwargs['request_id'] = request.request_id
 
-            return super().apply_async(args, kwargs, task_id, producer, link, link_error, **options)
+            with SQS_APPLY_ASYNC_DURATION_SECONDS.labels(self.name).time():
+                return super().apply_async(args, kwargs, task_id, producer, link, link_error, **options)
 
     return NotifyTask
 

app/dao/notifications_dao.py

@@ -677,6 +677,12 @@ def dao_get_notifications_by_references(references):
     ).all()
 
 
+@statsd(namespace="dao")
+def dao_created_scheduled_notification(scheduled_notification):
+    db.session.add(scheduled_notification)
+    db.session.commit()
+
+
 def dao_get_total_notifications_sent_per_day_for_performance_platform(start_date, end_date):
     """
     SELECT

app/delivery/send_to_providers.py

@@ -72,8 +72,8 @@ def send_sms_to_provider(notification):
                 notification.billable_units = template.fragment_count
                 update_notification_to_sending(notification, provider)
 
-        delta_milliseconds = (datetime.utcnow() - notification.created_at).total_seconds() * 1000
-        statsd_client.timing("sms.total-time", delta_milliseconds)
+        delta_seconds = (datetime.utcnow() - notification.created_at).total_seconds()
+        statsd_client.timing("sms.total-time", delta_seconds)
 
 
 def send_email_to_provider(notification):
@@ -118,8 +118,8 @@ def send_email_to_provider(notification):
             notification.reference = reference
             update_notification_to_sending(notification, provider)
 
-        delta_milliseconds = (datetime.utcnow() - notification.created_at).total_seconds() * 1000
-        statsd_client.timing("email.total-time", delta_milliseconds)
+        delta_seconds = (datetime.utcnow() - notification.created_at).total_seconds()
+        statsd_client.timing("email.total-time", delta_seconds)
 
 
 def update_notification_to_sending(notification, provider):

app/models.py

@@ -27,7 +27,7 @@ from notifications_utils.template import (
     SMSMessageTemplate,
     LetterPrintTemplate,
 )
-from notifications_utils.timezones import convert_utc_to_bst
+from notifications_utils.timezones import convert_bst_to_utc, convert_utc_to_bst
 
 from app.hashing import (
     hashpw,
@@ -291,6 +291,7 @@ service_letter_branding = db.Table(
 
 INTERNATIONAL_SMS_TYPE = 'international_sms'
 INBOUND_SMS_TYPE = 'inbound_sms'
+SCHEDULE_NOTIFICATIONS = 'schedule_notifications'
 EMAIL_AUTH = 'email_auth'
 LETTERS_AS_PDF = 'letters_as_pdf'
 PRECOMPILED_LETTER = 'precompiled_letter'
@@ -305,6 +306,7 @@ SERVICE_PERMISSION_TYPES = [
     LETTER_TYPE,
     INTERNATIONAL_SMS_TYPE,
     INBOUND_SMS_TYPE,
+    SCHEDULE_NOTIFICATIONS,
     EMAIL_AUTH,
     LETTERS_AS_PDF,
     UPLOAD_DOCUMENT,
@@ -1413,6 +1415,8 @@ class Notification(db.Model):
     client_reference = db.Column(db.String, index=True, nullable=True)
     _personalisation = db.Column(db.String, nullable=True)
 
+    scheduled_notification = db.relationship('ScheduledNotification', uselist=False)
+
     client_reference = db.Column(db.String, index=True, nullable=True)
 
     international = db.Column(db.Boolean, nullable=False, default=False)
@@ -1628,7 +1632,13 @@ class Notification(db.Model):
             "created_by_name": self.get_created_by_name(),
             "sent_at": self.sent_at.strftime(DATETIME_FORMAT) if self.sent_at else None,
             "completed_at": self.completed_at(),
-            "scheduled_for": None,
+            "scheduled_for": (
+                convert_bst_to_utc(
+                    self.scheduled_notification.scheduled_for
+                ).strftime(DATETIME_FORMAT)
+                if self.scheduled_notification
+                else None
+            ),
             "postage": self.postage
         }
 

app/notifications/process_notifications.py

@@ -9,6 +9,7 @@ from notifications_utils.recipients import (
     validate_and_format_phone_number,
     format_email_address
 )
+from notifications_utils.timezones import convert_bst_to_utc
 
 from app import redis_store
 from app.celery import provider_tasks
@@ -21,16 +22,27 @@ from app.models import (
     SMS_TYPE,
     LETTER_TYPE,
     NOTIFICATION_CREATED,
-    Notification
+    Notification,
+    ScheduledNotification
 )
 from app.dao.notifications_dao import (
     dao_create_notification,
-    dao_delete_notifications_by_id
+    dao_delete_notifications_by_id,
+    dao_created_scheduled_notification
 )
 
 from app.v2.errors import BadRequestError
 
 
+from gds_metrics import Histogram
+
+
+REDIS_GET_AND_INCR_DAILY_LIMIT_DURATION_SECONDS = Histogram(
+    'redis_get_and_incr_daily_limit_duration_seconds',
+    'Time taken to get and possibly incremement the daily limit cache key',
+)
+
+
 def create_content_for_notification(template, personalisation):
     template_object = template._as_utils_template_with_personalisation(personalisation)
     check_placeholders(template_object)
@@ -112,8 +124,9 @@ def persist_notification(
     if not simulated:
         dao_create_notification(notification)
         if key_type != KEY_TYPE_TEST:
-            if redis_store.get(redis.daily_limit_cache_key(service.id)):
-                redis_store.incr(redis.daily_limit_cache_key(service.id))
+            with REDIS_GET_AND_INCR_DAILY_LIMIT_DURATION_SECONDS.time():
+                if redis_store.get(redis.daily_limit_cache_key(service.id)):
+                    redis_store.incr(redis.daily_limit_cache_key(service.id))
 
         current_app.logger.info(
             "{} {} created at {}".format(notification_type, notification_id, notification_created_at)
@@ -158,3 +171,10 @@ def simulated_recipient(to_address, notification_type):
         return to_address in formatted_simulated_numbers
     else:
         return to_address in current_app.config['SIMULATED_EMAIL_ADDRESSES']
+
+
+def persist_scheduled_notification(notification_id, scheduled_for):
+    scheduled_datetime = convert_bst_to_utc(datetime.strptime(scheduled_for, "%Y-%m-%d %H:%M"))
+    scheduled_notification = ScheduledNotification(notification_id=notification_id,
+                                                   scheduled_for=scheduled_datetime)
+    dao_created_scheduled_notification(scheduled_notification)

app/notifications/validators.py

@@ -12,7 +12,7 @@ from app.dao import services_dao, templates_dao
 from app.dao.service_sms_sender_dao import dao_get_service_sms_senders_by_id
 from app.models import (
     INTERNATIONAL_SMS_TYPE, SMS_TYPE, EMAIL_TYPE, LETTER_TYPE,
-    KEY_TYPE_TEST, KEY_TYPE_TEAM
+    KEY_TYPE_TEST, KEY_TYPE_TEAM, SCHEDULE_NOTIFICATIONS
 )
 from app.service.utils import service_allowed_to_send_to
 from app.v2.errors import TooManyRequestsError, BadRequestError, RateLimitError
@@ -22,15 +22,24 @@ from app.utils import get_public_notify_type_text
 from app.dao.service_email_reply_to_dao import dao_get_reply_to_by_id
 from app.dao.service_letter_contact_dao import dao_get_letter_contact_by_id
 
+from gds_metrics.metrics import Histogram
+
+
+REDIS_EXCEEDED_RATE_LIMIT_DURATION_SECONDS = Histogram(
+    'redis_exceeded_rate_limit_duration_seconds',
+    'Time taken to check rate limit',
+)
+
 
 def check_service_over_api_rate_limit(service, api_key):
     if current_app.config['API_RATE_LIMIT_ENABLED'] and current_app.config['REDIS_ENABLED']:
         cache_key = rate_limit_cache_key(service.id, api_key.key_type)
         rate_limit = service.rate_limit
         interval = 60
-        if redis_store.exceeded_rate_limit(cache_key, rate_limit, interval):
-            current_app.logger.info("service {} has been rate limited for throughput".format(service.id))
-            raise RateLimitError(rate_limit, interval, api_key.key_type)
+        with REDIS_EXCEEDED_RATE_LIMIT_DURATION_SECONDS.time():
+            if redis_store.exceeded_rate_limit(cache_key, rate_limit, interval):
+                current_app.logger.info("service {} has been rate limited for throughput".format(service.id))
+                raise RateLimitError(rate_limit, interval, api_key.key_type)
 
 
 def check_service_over_daily_message_limit(key_type, service):
@@ -98,6 +107,12 @@ def check_if_service_can_send_files_by_email(service_contact_link, service_id):
         )
 
 
+def check_service_can_schedule_notification(permissions, scheduled_for):
+    if scheduled_for:
+        if not service_has_permission(SCHEDULE_NOTIFICATIONS, permissions):
+            raise BadRequestError(message="Cannot schedule notifications (this feature is invite-only)")
+
+
 def validate_and_format_recipient(send_to, key_type, service, notification_type, allow_whitelisted_recipients=True):
     if send_to is None:
         raise BadRequestError(message="Recipient can't be empty")

app/v2/notifications/notification_schemas.py

@@ -136,6 +136,7 @@ post_sms_request = {
         "phone_number": {"type": "string", "format": "phone_number"},
         "template_id": uuid,
         "personalisation": personalisation,
+        "scheduled_for": {"type": ["string", "null"], "format": "datetime_within_next_day"},
         "sms_sender_id": uuid
     },
     "required": ["phone_number", "template_id"],
@@ -181,6 +182,7 @@ post_email_request = {
         "email_address": {"type": "string", "format": "email_address"},
         "template_id": uuid,
         "personalisation": personalisation,
+        "scheduled_for": {"type": ["string", "null"], "format": "datetime_within_next_day"},
         "email_reply_to_id": uuid
     },
     "required": ["email_address", "template_id"],

app/v2/notifications/post_notifications.py

@@ -7,6 +7,7 @@ from boto.exception import SQSError
 from flask import request, jsonify, current_app, abort
 from notifications_utils.postal_address import PostalAddress
 from notifications_utils.recipients import try_validate_and_format_phone_number
+from gds_metrics import Histogram
 
 from app import (
     api_user,
@@ -43,12 +44,14 @@ from app.notifications.process_letter_notifications import (
 )
 from app.notifications.process_notifications import (
     persist_notification,
+    persist_scheduled_notification,
     send_notification_to_queue,
     simulated_recipient
 )
 from app.notifications.validators import (
     check_if_service_can_send_files_by_email,
     check_rate_limiting,
+    check_service_can_schedule_notification,
     check_service_email_reply_to_id,
     check_service_has_permission,
     check_service_sms_sender_id,
@@ -72,6 +75,12 @@ from app.v2.notifications.notification_schemas import (
 from app.v2.utils import get_valid_json
 
 
+POST_NOTIFICATION_JSON_PARSE_DURATION_SECONDS = Histogram(
+    'post_notification_json_parse_duration_seconds',
+    'Time taken to parse and validate post request json',
+)
+
+
 @v2_notification_blueprint.route('/{}'.format(LETTER_TYPE), methods=['POST'])
 def post_precompiled_letter_notification():
     request_json = get_valid_json()
@@ -114,19 +123,24 @@ def post_precompiled_letter_notification():
 
 @v2_notification_blueprint.route('/<notification_type>', methods=['POST'])
 def post_notification(notification_type):
-    request_json = get_valid_json()
+    with POST_NOTIFICATION_JSON_PARSE_DURATION_SECONDS.time():
+        request_json = get_valid_json()
 
-    if notification_type == EMAIL_TYPE:
-        form = validate(request_json, post_email_request)
-    elif notification_type == SMS_TYPE:
-        form = validate(request_json, post_sms_request)
-    elif notification_type == LETTER_TYPE:
-        form = validate(request_json, post_letter_request)
-    else:
-        abort(404)
+        if notification_type == EMAIL_TYPE:
+            form = validate(request_json, post_email_request)
+        elif notification_type == SMS_TYPE:
+            form = validate(request_json, post_sms_request)
+        elif notification_type == LETTER_TYPE:
+            form = validate(request_json, post_letter_request)
+        else:
+            abort(404)
 
     check_service_has_permission(notification_type, authenticated_service.permissions)
 
+    scheduled_for = form.get("scheduled_for", None)
+
+    check_service_can_schedule_notification(authenticated_service.permissions, scheduled_for)
+
     check_rate_limiting(authenticated_service, api_user)
 
     template, template_with_content = validate_template(
@@ -177,7 +191,7 @@ def post_notification(notification_type):
     resp = create_resp_partial(
         notification=notification,
         url_root=request.url_root,
-        scheduled_for=None,
+        scheduled_for=scheduled_for,
         content=template_with_content.content_with_placeholders_filled_in,
     )
     return jsonify(resp), 201
@@ -244,15 +258,19 @@ def process_sms_or_email_notification(*, form, notification_type, api_key, templ
         document_download_count=document_download_count
     )
 
-    if not simulated:
-        queue_name = QueueNames.PRIORITY if template.process_type == PRIORITY else None
-        send_notification_to_queue(
-            notification=notification,
-            research_mode=service.research_mode,
-            queue=queue_name
-        )
+    scheduled_for = form.get("scheduled_for", None)
+    if scheduled_for:
+        persist_scheduled_notification(notification.id, form["scheduled_for"])
     else:
-        current_app.logger.debug("POST simulated notification for id: {}".format(notification.id))
+        if not simulated:
+            queue_name = QueueNames.PRIORITY if template.process_type == PRIORITY else None
+            send_notification_to_queue(
+                notification=notification,
+                research_mode=service.research_mode,
+                queue=queue_name
+            )
+        else:
+            current_app.logger.debug("POST simulated notification for id: {}".format(notification.id))
 
     return notification
 

application.py

@@ -2,9 +2,12 @@
 from __future__ import print_function
 
 from flask import Flask
+import psycogreen.eventlet
 
 from app import create_app
 
+psycogreen.eventlet.patch_psycopg()
+
 application = Flask('app')
 
 create_app(application)

gunicorn_config.py

@@ -6,7 +6,7 @@ import gunicorn
 from gds_metrics.gunicorn import child_exit  # noqa
 
 workers = 4
-worker_class = "eventlet"
+worker_class = "gevent"
 worker_connections = 256
 errorlog = "/home/vcap/logs/gunicorn_error.log"
 bind = "0.0.0.0:{}".format(os.getenv("PORT"))

manifest.yml.j2

@@ -1,6 +1,6 @@
 {%- set app_vars = {
-  'notify-api': {
-    'NOTIFY_APP_NAME': 'api',
+  'notify-api-canary1': {
+    'NOTIFY_APP_NAME': 'api-canary1',
     'disk_quota': '2G',
     'sqlalchemy_pool_size': 30,
     'routes': {
@@ -12,8 +12,8 @@
     'health-check-invocation-timeout': 3,
     'instances': {
       'preview': None,
-      'staging': None,
-      'production': 25
+      'staging': 1,
+      'production': 1
     },
   },
   'notify-api-db-migration': {

requirements-app.txt

@@ -10,13 +10,14 @@ Flask-Migrate==2.5.3
 git+https://github.com/mitsuhiko/flask-sqlalchemy.git@500e732dd1b975a56ab06a46bd1a20a21e682262#egg=Flask-SQLAlchemy==2.3.2.dev20190108
 Flask==1.1.2
 click-datetime==0.2
-eventlet==0.25.2
+gevent==20.6.1
 gunicorn==20.0.4
 iso8601==0.1.12
 itsdangerous==1.1.0
 jsonschema==3.2.0
 marshmallow-sqlalchemy==0.23.0
 marshmallow==2.21.0 # pyup: <3 # v3 throws errors
+psycogreen==1.0.2
 psycopg2-binary==2.8.5
 PyJWT==1.7.1
 SQLAlchemy==1.3.17
@@ -26,6 +27,8 @@ notifications-python-client==5.5.1
 # PaaS
 awscli-cwlogs>=1.4,<1.5
 
-git+https://github.com/alphagov/notifications-utils.git@39.4.3#egg=notifications-utils==39.4.3
+git+https://github.com/alphagov/notifications-utils.git@39.4.4#egg=notifications-utils==39.4.4
 
+# gds-metrics requires prometheseus 0.2.0, override that requirement as 0.7.1 brings significant performance gains
+prometheus-client==0.7.1
 gds-metrics==0.2.0

requirements.txt

@@ -12,13 +12,14 @@ Flask-Migrate==2.5.3
 git+https://github.com/mitsuhiko/flask-sqlalchemy.git@500e732dd1b975a56ab06a46bd1a20a21e682262#egg=Flask-SQLAlchemy==2.3.2.dev20190108
 Flask==1.1.2
 click-datetime==0.2
-eventlet==0.25.2
+gevent==20.6.1
 gunicorn==20.0.4
 iso8601==0.1.12
 itsdangerous==1.1.0
 jsonschema==3.2.0
 marshmallow-sqlalchemy==0.23.0
 marshmallow==2.21.0 # pyup: <3 # v3 throws errors
+psycogreen==1.0.2
 psycopg2-binary==2.8.5
 PyJWT==1.7.1
 SQLAlchemy==1.3.17
@@ -28,8 +29,10 @@ notifications-python-client==5.5.1
 # PaaS
 awscli-cwlogs>=1.4,<1.5
 
-git+https://github.com/alphagov/notifications-utils.git@39.4.3#egg=notifications-utils==39.4.3
+git+https://github.com/alphagov/notifications-utils.git@39.4.4#egg=notifications-utils==39.4.4
 
+# gds-metrics requires prometheseus 0.2.0, override that requirement as 0.7.1 brings significant performance gains
+prometheus-client==0.7.1
 gds-metrics==0.2.0
 
 ## The following requirements were added by pip freeze:
@@ -37,19 +40,18 @@ alembic==1.4.2
 amqp==1.4.9
 anyjson==0.3.3
 attrs==19.3.0
-awscli==1.18.75
+awscli==1.18.80
 bcrypt==3.1.7
 billiard==3.3.0.23
 bleach==3.1.4
 blinker==1.4
 boto==2.49.0
 boto3==1.10.38
-botocore==1.16.25
+botocore==1.17.3
 certifi==2020.4.5.2
 chardet==3.0.4
 click==7.1.2
 colorama==0.4.3
-dnspython==1.16.0
 docutils==0.15.2
 flask-redis==0.4.0
 future==0.18.2
@@ -66,7 +68,6 @@ mistune==0.8.4
 monotonic==1.5
 orderedset==2.0.1
 phonenumbers==8.11.2
-prometheus-client==0.2.0
 pyasn1==0.4.8
 pycparser==2.20
 PyPDF2==1.26.0
@@ -87,3 +88,5 @@ urllib3==1.25.9
 webencodings==0.5.1
 Werkzeug==1.0.1
 zipp==3.1.0
+zope.event==4.4
+zope.interface==5.1.0

scripts/paas_app_wrapper.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
 case $NOTIFY_APP_NAME in
-  api)
+  api|api-canary1)
     unset GUNICORN_CMD_ARGS
     exec scripts/run_app_paas.sh gunicorn -c /home/vcap/app/gunicorn_config.py application
     ;;

tests/app/dao/notification_dao/test_notification_dao.py

@@ -10,6 +10,7 @@ from sqlalchemy.orm.exc import NoResultFound
 
 from app.dao.notifications_dao import (
     dao_create_notification,
+    dao_created_scheduled_notification,
     dao_delete_notifications_by_id,
     dao_get_last_notification_added_for_job_id,
     dao_get_notifications_by_recipient_or_reference,
@@ -35,6 +36,7 @@ from app.models import (
     Job,
     Notification,
     NotificationHistory,
+    ScheduledNotification,
     NOTIFICATION_STATUS_TYPES,
     NOTIFICATION_STATUS_TYPES_FAILED,
     NOTIFICATION_TEMPORARY_FAILURE,
@@ -447,6 +449,7 @@ def test_save_notification_with_no_job(sample_template, mmg_provider):
 
 def test_get_notification_with_personalisation_by_id(sample_template):
     notification = create_notification(template=sample_template,
+                                       scheduled_for='2017-05-05 14:15',
                                        status='created')
     notification_from_db = get_notification_with_personalisation(
         sample_template.service.id,
@@ -454,6 +457,7 @@ def test_get_notification_with_personalisation_by_id(sample_template):
         key_type=None
     )
     assert notification == notification_from_db
+    assert notification_from_db.scheduled_notification.scheduled_for == datetime(2017, 5, 5, 14, 15)
 
 
 def test_get_notification_by_id_when_notification_exists(sample_notification):
@@ -1388,6 +1392,18 @@ def test_dao_get_notifications_by_reference(
     assert results.items[0].id == letter.id
 
 
+def test_dao_created_scheduled_notification(sample_notification):
+
+    scheduled_notification = ScheduledNotification(notification_id=sample_notification.id,
+                                                   scheduled_for=datetime.strptime("2017-01-05 14:15",
+                                                                                   "%Y-%m-%d %H:%M"))
+    dao_created_scheduled_notification(scheduled_notification)
+    saved_notification = ScheduledNotification.query.all()
+    assert len(saved_notification) == 1
+    assert saved_notification[0].notification_id == sample_notification.id
+    assert saved_notification[0].scheduled_for == datetime(2017, 1, 5, 14, 15)
+
+
 def test_dao_get_notifications_by_to_field_filters_status(sample_template):
     notification = create_notification(
         template=sample_template, to_field='+447700900855',

tests/app/db.py

@@ -9,7 +9,8 @@ from app.dao.invited_org_user_dao import save_invited_org_user
 from app.dao.invited_user_dao import save_invited_user
 from app.dao.jobs_dao import dao_create_job
 from app.dao.notifications_dao import (
-    dao_create_notification
+    dao_create_notification,
+    dao_created_scheduled_notification
 )
 from app.dao.organisation_dao import dao_create_organisation, dao_add_service_to_organisation
 from app.dao.permissions_dao import permission_dao
@@ -38,6 +39,7 @@ from app.models import (
     ServiceInboundApi,
     ServiceCallbackApi,
     ServiceLetterContact,
+    ScheduledNotification,
     ServicePermission,
     ServiceSmsSender,
     ServiceWhitelist,
@@ -292,6 +294,14 @@ def create_notification(
     }
     notification = Notification(**data)
     dao_create_notification(notification)
+    if scheduled_for:
+        scheduled_notification = ScheduledNotification(id=uuid.uuid4(),
+                                                       notification_id=notification.id,
+                                                       scheduled_for=datetime.strptime(scheduled_for,
+                                                                                       "%Y-%m-%d %H:%M"))
+        if status != 'created':
+            scheduled_notification.pending = False
+        dao_created_scheduled_notification(scheduled_notification)
 
     return notification
 

tests/app/notifications/test_process_notification.py

@@ -10,12 +10,14 @@ from collections import namedtuple
 from app.models import (
     Notification,
     NotificationHistory,
+    ScheduledNotification,
     Template,
     LETTER_TYPE
 )
 from app.notifications.process_notifications import (
     create_content_for_notification,
     persist_notification,
+    persist_scheduled_notification,
     send_notification_to_queue,
     simulated_recipient
 )
@@ -383,6 +385,14 @@ def test_persist_notification_with_international_info_does_not_store_for_email(
     assert persisted_notification.rate_multiplier is None
 
 
+def test_persist_scheduled_notification(sample_notification):
+    persist_scheduled_notification(sample_notification.id, '2017-05-12 14:15')
+    scheduled_notification = ScheduledNotification.query.all()
+    assert len(scheduled_notification) == 1
+    assert scheduled_notification[0].notification_id == sample_notification.id
+    assert scheduled_notification[0].scheduled_for == datetime.datetime(2017, 5, 12, 13, 15)
+
+
 @pytest.mark.parametrize('recipient, expected_recipient_normalised', [
     ('7900900123', '447900900123'),
     ('+447900   900 123', '447900900123'),

tests/app/v2/notifications/test_get_notifications.py

@@ -21,14 +21,16 @@ def test_get_notification_by_id_returns_200(
     sample_notification = create_notification(
         template=sample_template,
         billable_units=billable_units,
-        sent_by=provider
+        sent_by=provider,
+        scheduled_for="2017-05-12 15:15"
     )
 
     # another
     create_notification(
         template=sample_template,
         billable_units=billable_units,
-        sent_by=provider
+        sent_by=provider,
+        scheduled_for="2017-06-12 15:15"
     )
 
     auth_header = create_authorization_header(service_id=sample_notification.service_id)
@@ -68,7 +70,7 @@ def test_get_notification_by_id_returns_200(
         "subject": None,
         'sent_at': sample_notification.sent_at,
         'completed_at': sample_notification.completed_at(),
-        'scheduled_for': None,
+        'scheduled_for': '2017-05-12T14:15:00.000000Z',
         'postage': None,
     }
 
@@ -164,7 +166,7 @@ def test_get_notification_by_id_returns_created_by_name_if_notification_created_
     assert json_response['created_by_name'] == 'Test User'
 
 
-def test_get_notifications_returns_none_for_scheduled_for(client, sample_template):
+def test_get_notifications_returns_scheduled_for(client, sample_template):
     sample_notification_with_reference = create_notification(template=sample_template,
                                                              client_reference='some-client-reference',
                                                              scheduled_for='2017-05-23 17:15')
@@ -181,7 +183,7 @@ def test_get_notifications_returns_none_for_scheduled_for(client, sample_templat
     assert len(json_response['notifications']) == 1
 
     assert json_response['notifications'][0]['id'] == str(sample_notification_with_reference.id)
-    assert not json_response['notifications'][0]['scheduled_for']
+    assert json_response['notifications'][0]['scheduled_for'] == "2017-05-23T16:15:00.000000Z"
 
 
 def test_get_notification_by_reference_nonexistent_reference_returns_no_notifications(client, sample_service):

tests/app/v2/notifications/test_notification_schemas.py

@@ -2,6 +2,7 @@ import uuid
 
 import pytest
 from flask import json
+from freezegun import freeze_time
 from jsonschema import ValidationError
 
 from app.models import NOTIFICATION_CREATED, EMAIL_TYPE
@@ -256,3 +257,66 @@ def valid_email_response():
         },
         "scheduled_for": ""
     }
+
+
+@pytest.mark.parametrize("schema",
+                         [post_email_request_schema, post_sms_request_schema])
+@freeze_time("2017-05-12 13:00:00")
+def test_post_schema_valid_scheduled_for(schema):
+    j = {"template_id": str(uuid.uuid4()),
+         "scheduled_for": "2017-05-12 13:15"}
+    if schema == post_email_request_schema:
+        j.update({"email_address": "joe@gmail.com"})
+    else:
+        j.update({"phone_number": "07515111111"})
+    assert validate(j, schema) == j
+
+
+@pytest.mark.parametrize("invalid_datetime",
+                         ["13:00:00 2017-01-01",
+                          "2017-31-12 13:00:00",
+                          "01-01-2017T14:00:00.0000Z"
+                          ])
+@pytest.mark.parametrize("schema",
+                         [post_email_request_schema, post_sms_request_schema])
+def test_post_email_schema_invalid_scheduled_for(invalid_datetime, schema):
+    j = {"template_id": str(uuid.uuid4()),
+         "scheduled_for": invalid_datetime}
+    if schema == post_email_request_schema:
+        j.update({"email_address": "joe@gmail.com"})
+    else:
+        j.update({"phone_number": "07515111111"})
+    with pytest.raises(ValidationError) as e:
+        validate(j, schema)
+    error = json.loads(str(e.value))
+    assert error['status_code'] == 400
+    assert error['errors'] == [{'error': 'ValidationError',
+                                'message': "scheduled_for datetime format is invalid. "
+                                           "It must be a valid ISO8601 date time format, "
+                                           "https://en.wikipedia.org/wiki/ISO_8601"}]
+
+
+@freeze_time("2017-05-12 13:00:00")
+def test_scheduled_for_raises_validation_error_when_in_the_past():
+    j = {"phone_number": "07515111111",
+         "template_id": str(uuid.uuid4()),
+         "scheduled_for": "2017-05-12 10:00"}
+    with pytest.raises(ValidationError) as e:
+        validate(j, post_sms_request_schema)
+    error = json.loads(str(e.value))
+    assert error['status_code'] == 400
+    assert error['errors'] == [{'error': 'ValidationError',
+                                'message': "scheduled_for datetime can not be in the past"}]
+
+
+@freeze_time("2017-05-12 13:00:00")
+def test_scheduled_for_raises_validation_error_when_more_than_24_hours_in_the_future():
+    j = {"phone_number": "07515111111",
+         "template_id": str(uuid.uuid4()),
+         "scheduled_for": "2017-05-13 14:00"}
+    with pytest.raises(ValidationError) as e:
+        validate(j, post_sms_request_schema)
+    error = json.loads(str(e.value))
+    assert error['status_code'] == 400
+    assert error['errors'] == [{'error': 'ValidationError',
+                                'message': "scheduled_for datetime can only be 24 hours in the future"}]

tests/app/v2/notifications/test_post_notifications.py

@@ -3,12 +3,15 @@ from unittest import mock
 from unittest.mock import call
 
 import pytest
+from freezegun import freeze_time
 from boto.exception import SQSError
 
 from app.dao.service_sms_sender_dao import dao_update_service_sms_sender
 from app.models import (
+    ScheduledNotification,
     EMAIL_TYPE,
     NOTIFICATION_CREATED,
+    SCHEDULE_NOTIFICATIONS,
     SMS_TYPE,
     INTERNATIONAL_SMS_TYPE
 )
@@ -606,6 +609,56 @@ def test_post_sms_should_persist_supplied_sms_number(client, sample_template_wit
     assert mocked.called
 
 
+@pytest.mark.parametrize("notification_type, key_send_to, send_to",
+                         [("sms", "phone_number", "07700 900 855"),
+                          ("email", "email_address", "sample@email.com")])
+@freeze_time("2017-05-14 14:00:00")
+def test_post_notification_with_scheduled_for(
+        client, notify_db_session, notification_type, key_send_to, send_to
+):
+    service = create_service(service_name=str(uuid.uuid4()),
+                             service_permissions=[EMAIL_TYPE, SMS_TYPE, SCHEDULE_NOTIFICATIONS])
+    template = create_template(service=service, template_type=notification_type)
+    data = {
+        key_send_to: send_to,
+        'template_id': str(template.id) if notification_type == EMAIL_TYPE else str(template.id),
+        'scheduled_for': '2017-05-14 14:15'
+    }
+    auth_header = create_authorization_header(service_id=service.id)
+
+    response = client.post('/v2/notifications/{}'.format(notification_type),
+                           data=json.dumps(data),
+                           headers=[('Content-Type', 'application/json'), auth_header])
+    assert response.status_code == 201
+    resp_json = json.loads(response.get_data(as_text=True))
+    scheduled_notification = ScheduledNotification.query.filter_by(notification_id=resp_json["id"]).all()
+    assert len(scheduled_notification) == 1
+    assert resp_json["id"] == str(scheduled_notification[0].notification_id)
+    assert resp_json["scheduled_for"] == '2017-05-14 14:15'
+
+
+@pytest.mark.parametrize("notification_type, key_send_to, send_to",
+                         [("sms", "phone_number", "07700 900 855"),
+                          ("email", "email_address", "sample@email.com")])
+@freeze_time("2017-05-14 14:00:00")
+def test_post_notification_raises_bad_request_if_service_not_invited_to_schedule(
+        client, sample_template, sample_email_template, notification_type, key_send_to, send_to):
+    data = {
+        key_send_to: send_to,
+        'template_id': str(sample_email_template.id) if notification_type == EMAIL_TYPE else str(sample_template.id),
+        'scheduled_for': '2017-05-14 14:15'
+    }
+    auth_header = create_authorization_header(service_id=sample_template.service_id)
+
+    response = client.post('/v2/notifications/{}'.format(notification_type),
+                           data=json.dumps(data),
+                           headers=[('Content-Type', 'application/json'), auth_header])
+    assert response.status_code == 400
+    error_json = json.loads(response.get_data(as_text=True))
+    assert error_json['errors'] == [
+        {"error": "BadRequestError", "message": 'Cannot schedule notifications (this feature is invite-only)'}]
+
+
 def test_post_notification_raises_bad_request_if_not_valid_notification_type(client, sample_service):
     auth_header = create_authorization_header(service_id=sample_service.id)
     response = client.post(

tests/app/v2/test_errors.py

@@ -4,7 +4,7 @@ from sqlalchemy.exc import DataError
 
 
 @pytest.fixture(scope='function')
-def app_for_test(mocker):
+def app_for_test():
     import flask
     from flask import Blueprint
     from app.authentication.auth import AuthError