Add a canary app for testing

using lots of CPU by tracing the application to gather some data

Alternatively we could take a stab in the dark, which is what this
commit is doing.

I have the hypothesis that we are not re-using TCP connections using
HTTP keepalive

Refer to https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Keep-Alive

This means we are renegotiating the TLS connection every time we want to
send an SMS.  When we are sending lots of SMS messages then this will do
a lot of crypto handshaking which is expensive (in terms of CPU)

ie send_sms calls request() which creates a new tcp connection and a
new TLS handshake

When you use request.Session() to create a session it uses urllib3's
connection pooling, I've arbitrarily chosen 32 connections per pool
(with a default number of pools = 10)

ie init_app creates a session which has underlying connection pools
send_sms claims a connection from the pool and uses it to create or
re-use an existing TLS connection

Sessions are usually not great because they share data like cookies, but
when calling an API this is fine, or at least it is probably worth
canarying

Another way of re-using connections is by running HAProxy or similar as
a side-car proxy, which proxies to the API. send_sms would make a
local TCP connection to HAProxy which proxies to the MMG or Firetext
API via TLS, adding the Connection: keep-alive header

This command can be used to see how many TLS handshakes your computer is
doing, with some false positives:

tcpdump -n "tcp port 443 and (tcp[((tcp[12] & 0xf0) >> 2)] = 0x16)"

(alternatively we could just instrument the code /shrug)

Signed-off-by: toby lorne <toby@toby.codes>

app/__init__.py

@@ -19,7 +19,6 @@ 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.exceptions import RequestEntityTooLarge as WerkzeugRequestEntityTooLarge
 from werkzeug.local import LocalProxy
 
 from app.celery.celery import NotifyCelery
@@ -282,15 +281,6 @@ def init_app(app):
         g.start = monotonic()
         g.endpoint = request.endpoint
 
-    @app.before_request
-    def check_content_length():
-        if (
-            request.content_length is not None
-            and current_app.config['MAX_CONTENT_LENGTH'] is not None
-            and request.content_length > current_app.config['MAX_CONTENT_LENGTH']
-        ):
-            raise WerkzeugRequestEntityTooLarge()
-
     @app.after_request
     def after_request(response):
         CONCURRENT_REQUESTS.dec()

app/broadcast_message/rest.py

@@ -121,7 +121,6 @@ def create_broadcast_message(service_id):
         created_by_id=user.id,
         content=content,
         reference=reference,
-        stubbed=service.restricted
     )
 
     dao_save_object(broadcast_message)
@@ -215,8 +214,7 @@ def _create_broadcast_event(broadcast_message):
 
     dao_save_object(event)
 
-    if not broadcast_message.stubbed:
-        send_broadcast_event.apply_async(
-            kwargs={'broadcast_event_id': str(event.id)},
-            queue=QueueNames.BROADCASTS
-        )
+    send_broadcast_event.apply_async(
+        kwargs={'broadcast_event_id': str(event.id)},
+        queue=QueueNames.BROADCASTS
+    )

app/clients/cbc_proxy.py

@@ -174,8 +174,8 @@ class CBCProxyCanary(CBCProxyClientBase):
 
 
 class CBCProxyEE(CBCProxyClientBase):
-    lambda_name = 'ee-1-proxy'
-    failover_lambda_name = 'ee-2-proxy'
+    lambda_name = 'bt-ee-1-proxy'
+    failover_lambda_name = 'bt-ee-2-proxy'
 
     LANGUAGE_ENGLISH = 'en-GB'
     LANGUAGE_WELSH = 'cy-GB'

app/clients/email/aws_ses.py

@@ -62,6 +62,7 @@ class AwsSesClient(EmailClient):
         # before-call, after-call, after-call-error, request-created, response-received
         self._client.meta.events.register('request-created.ses.SendEmail', self.ses_request_created_hook)
         self._client.meta.events.register('response-received.ses.SendEmail', self.ses_response_received_hook)
+        self._client.meta.events.register('before-call.ses', self.ses_inject_connection_header)
 
     def ses_request_created_hook(self, **kwargs):
         # request created may be called multiple times if the request auto-retries. We want to count all these as the
@@ -77,6 +78,11 @@ class AwsSesClient(EmailClient):
     def get_name(self):
         return self.name
 
+    def ses_inject_connection_header(self, params, **kwargs):
+        # keep underlying TLS connection open, so we do not spend lots of CPU
+        # and network time renegotiating TLS
+        params['headers']['Connection'] = 'Keep-Alive'
+
     def send_email(self,
                    source,
                    to_addresses,
@@ -148,3 +154,4 @@ def punycode_encode_email(email_address):
     # only the hostname should ever be punycode encoded.
     local, hostname = email_address.split('@')
     return '{}@{}'.format(local, hostname.encode('idna').decode('utf-8'))
+

app/clients/sms/firetext.py

@@ -2,7 +2,8 @@ import json
 import logging
 
 from time import monotonic
-from requests import request, RequestException
+from requests import request, RequestException, Session
+from requests.adapters import HTTPAdapter
 
 from app.clients.sms import (SmsClient, SmsClientResponseException)
 
@@ -69,6 +70,9 @@ class FiretextClient(SmsClient):
         self.name = 'firetext'
         self.url = current_app.config.get('FIRETEXT_URL')
         self.statsd_client = statsd_client
+        # this uses urllib3 under the hood to create a connection pool
+        self.session = Session()
+        self.session.mount('https://', HTTPAdapter(pool_maxsize=32))
 
     def get_name(self):
         return self.name
@@ -103,8 +107,7 @@ class FiretextClient(SmsClient):
         response = None
         start_time = monotonic()
         try:
-            response = request(
-                "POST",
+            response = self.session.post(
                 self.url,
                 data=data,
                 timeout=60

app/clients/sms/mmg.py

@@ -1,6 +1,8 @@
 import json
 from time import monotonic
-from requests import (request, RequestException)
+from requests import (request, RequestException, Session)
+from requests.adapters import HTTPAdapter
+
 from app.clients.sms import (SmsClient, SmsClientResponseException)
 
 mmg_response_map = {
@@ -75,6 +77,9 @@ class MMGClient(SmsClient):
         self.name = 'mmg'
         self.statsd_client = statsd_client
         self.mmg_url = current_app.config.get('MMG_URL')
+        # this uses urllib3 under the hood to create a connection pool
+        self.session = Session()
+        self.session.mount('https://', HTTPAdapter(pool_maxsize=32))
 
     def record_outcome(self, success, response):
         status_code = response.status_code if response else 503
@@ -108,8 +113,8 @@ class MMGClient(SmsClient):
         response = None
         start_time = monotonic()
         try:
-            response = request(
-                "POST",
+
+            response = self.session.post(
                 self.mmg_url,
                 data=json.dumps(data),
                 headers={

app/config.py

@@ -381,7 +381,6 @@ class Config(object):
     CBC_PROXY_ENABLED = bool(CBC_PROXY_AWS_ACCESS_KEY_ID)
 
     ENABLED_CBCS = {BroadcastProvider.EE, BroadcastProvider.THREE, BroadcastProvider.O2, BroadcastProvider.VODAFONE}
-    MAX_CONTENT_LENGTH = 5 * 1024 * 1024  # 5MB
 
 
 ######################
@@ -509,8 +508,6 @@ class Staging(Config):
     API_RATE_LIMIT_ENABLED = True
     CHECK_PROXY_HEADER = True
 
-    MAX_CONTENT_LENGTH = None
-
 
 class Live(Config):
     NOTIFY_EMAIL_DOMAIN = 'notifications.service.gov.uk'
@@ -532,8 +529,6 @@ class Live(Config):
 
     CRONITOR_ENABLED = True
 
-    MAX_CONTENT_LENGTH = None
-
 
 class CloudFoundryConfig(Config):
     pass

app/models.py

@@ -488,10 +488,6 @@ class Service(db.Model, Versioned):
     organisation = db.relationship('Organisation', backref='services')
 
     notes = db.Column(db.Text, nullable=True)
-    purchase_order_number = db.Column(db.String(255), nullable=True)
-    billing_contact_names = db.Column(db.Text, nullable=True)
-    billing_contact_email_addresses = db.Column(db.Text, nullable=True)
-    billing_reference = db.Column(db.String(255), nullable=True)
 
     email_branding = db.relationship(
         'EmailBranding',
@@ -2269,8 +2265,6 @@ class BroadcastMessage(db.Model):
 
     reference = db.Column(db.String(255), nullable=True)
 
-    stubbed = db.Column(db.Boolean, nullable=True)
-
     CheckConstraint("created_by_id is not null or api_key_id is not null")
 
     @property

manifest.yml.j2

@@ -53,6 +53,11 @@
   'notify-delivery-worker-jobs': {},
   'notify-delivery-worker-research': {},
   'notify-delivery-worker-sender': {'disk_quota': '2G', 'memory': '4G'},
+  'notify-delivery-worker-sender-canary': {'disk_quota': '2G', 'memory': '4G',  'instances': {
+      'preview': 0,
+      'staging': 1,
+      'production': 0
+    },},
   'notify-delivery-worker-periodic': {},
   'notify-delivery-worker-reporting': {
     'additional_env_vars': {

migrations/versions/0339_service_billing_details.py

@@ -1,39 +0,0 @@
-"""
-
-Revision ID: 0339_service_billing_details
-Revises: 0338_add_notes_to_service
-Create Date: 2021-01-20 17:55:46.555460
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-revision = '0339_service_billing_details'
-down_revision = '0338_add_notes_to_service'
-
-
-def upgrade():
-    # ### commands auto generated by Alembic - please adjust! ###
-    op.add_column('services', sa.Column('billing_contact_email_addresses', sa.Text(), nullable=True))
-    op.add_column('services', sa.Column('billing_contact_names', sa.Text(), nullable=True))
-    op.add_column('services', sa.Column('billing_reference', sa.String(length=255), nullable=True))
-    op.add_column('services', sa.Column('purchase_order_number', sa.String(length=255), nullable=True))
-    op.add_column('services_history', sa.Column('billing_contact_email_addresses', sa.Text(), nullable=True))
-    op.add_column('services_history', sa.Column('billing_contact_names', sa.Text(), nullable=True))
-    op.add_column('services_history', sa.Column('billing_reference', sa.String(length=255), nullable=True))
-    op.add_column('services_history', sa.Column('purchase_order_number', sa.String(length=255), nullable=True))
-    # ### end Alembic commands ###
-
-
-def downgrade():
-    # ### commands auto generated by Alembic - please adjust! ###
-    op.drop_column('services_history', 'purchase_order_number')
-    op.drop_column('services_history', 'billing_reference')
-    op.drop_column('services_history', 'billing_contact_names')
-    op.drop_column('services_history', 'billing_contact_email_addresses')
-    op.drop_column('services', 'purchase_order_number')
-    op.drop_column('services', 'billing_reference')
-    op.drop_column('services', 'billing_contact_names')
-    op.drop_column('services', 'billing_contact_email_addresses')
-    # ### end Alembic commands ###

migrations/versions/0340_stub_training_broadcasts.py

@@ -1,25 +0,0 @@
-"""
-
-Revision ID: 0340_stub_training_broadcasts
-Revises: 0339_service_billing_details
-Create Date: 2021-01-26 16:48:44.921065
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-revision = '0340_stub_training_broadcasts'
-down_revision = '0339_service_billing_details'
-
-
-def upgrade():
-    # ### commands auto generated by Alembic - please adjust! ###
-    op.add_column('broadcast_message', sa.Column('stubbed', sa.Boolean(), nullable=True))
-    # ### end Alembic commands ###
-
-
-def downgrade():
-    # ### commands auto generated by Alembic - please adjust! ###
-    op.drop_column('broadcast_message', 'stubbed')
-    # ### end Alembic commands ###

scripts/paas_app_wrapper.sh

@@ -24,6 +24,10 @@ case $NOTIFY_APP_NAME in
     exec scripts/run_multi_worker_app_paas.sh celery multi start 3 -c 10 -A run_celery.notify_celery --loglevel=INFO \
     --logfile=/dev/null --pidfile=/tmp/celery%N.pid -Q send-sms-tasks,send-email-tasks
     ;;
+    delivery-worker-sender-canary)
+    exec scripts/run_multi_worker_app_paas.sh celery multi start 3 -c 10 -A run_celery.notify_celery --loglevel=INFO \
+    --logfile=/dev/null --pidfile=/tmp/celery%N.pid -Q send-sms-tasks,send-email-tasks
+    ;;
   delivery-worker-periodic)
     exec scripts/run_app_paas.sh celery -A run_celery.notify_celery worker --loglevel=INFO --concurrency=2 \
     -Q periodic-tasks 2> /dev/null

tests/app/broadcast_message/test_rest.py

@@ -4,7 +4,6 @@ import uuid
 from freezegun import freeze_time
 import pytest
 
-from app.dao.broadcast_message_dao import dao_get_broadcast_message_by_id_and_service_id
 from app.models import BROADCAST_TYPE, BroadcastStatusType, BroadcastEventMessageType
 
 from tests.app.db import create_broadcast_message, create_template, create_service, create_user
@@ -118,9 +117,7 @@ def test_get_broadcast_messages_for_service(admin_request, sample_broadcast_serv
 
 
 @freeze_time('2020-01-01')
-@pytest.mark.parametrize('training_mode_service', [True, False])
-def test_create_broadcast_message(admin_request, sample_broadcast_service, training_mode_service):
-    sample_broadcast_service.restricted = training_mode_service
+def test_create_broadcast_message(admin_request, sample_broadcast_service):
     t = create_template(sample_broadcast_service, BROADCAST_TYPE)
 
     response = admin_request.post(
@@ -141,8 +138,6 @@ def test_create_broadcast_message(admin_request, sample_broadcast_service, train
     assert response['personalisation'] == {}
     assert response['areas'] == []
 
-    broadcast_message = dao_get_broadcast_message_by_id_and_service_id(response["id"], sample_broadcast_service.id)
-    assert broadcast_message.stubbed == training_mode_service
 
 @pytest.mark.parametrize('data, expected_errors', [
     (
@@ -551,47 +546,6 @@ def test_update_broadcast_message_status_stores_approved_by_and_approved_at_and_
     assert alert_event.transmitted_content == {"body": "emergency broadcast"}
 
 
-def test_update_broadcast_message_status_updates_details_but_does_not_queue_task_for_stubbed_broadcast_message(
-    admin_request,
-    sample_broadcast_service,
-    mocker
-):
-    sample_broadcast_service.restricted = True
-    t = create_template(sample_broadcast_service, BROADCAST_TYPE, content='emergency broadcast')
-    bm = create_broadcast_message(
-        t,
-        status=BroadcastStatusType.PENDING_APPROVAL,
-        areas={"areas": ["london"], "simple_polygons": [[[51.30, 0.7], [51.28, 0.8], [51.25, -0.7]]]},
-        stubbed=True
-    )
-    approver = create_user(email='approver@gov.uk')
-    sample_broadcast_service.users.append(approver)
-    mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_event.apply_async')
-
-    response = admin_request.post(
-        'broadcast_message.update_broadcast_message_status',
-        _data={'status': BroadcastStatusType.BROADCASTING, 'created_by': str(approver.id)},
-        service_id=t.service_id,
-        broadcast_message_id=bm.id,
-        _expected_status=200
-    )
-
-    assert response['status'] == BroadcastStatusType.BROADCASTING
-    assert response['approved_at'] is not None
-    assert response['approved_by_id'] == str(approver.id)
-
-    assert len(bm.events) == 1
-    alert_event = bm.events[0]
-
-    assert len(mock_task.mock_calls) == 0
-
-    assert alert_event.service_id == sample_broadcast_service.id
-    assert alert_event.transmitted_areas == bm.areas
-    assert alert_event.message_type == BroadcastEventMessageType.ALERT
-    assert alert_event.transmitted_finishes_at == bm.finishes_at
-    assert alert_event.transmitted_content == {"body": "emergency broadcast"}
-
-
 def test_update_broadcast_message_status_creates_event_with_correct_content_if_broadcast_has_no_template(
     admin_request,
     sample_broadcast_service,

tests/app/clients/test_cbc_proxy.py

@@ -78,7 +78,7 @@ def test_cbc_proxy_lambda_client_has_correct_keys(cbc_proxy_ee):
     ('my-description', 'en-GB'),
     ('mŷ-description', 'cy-GB'),
 ))
-@pytest.mark.parametrize('cbc', ['ee', 'three', 'o2'])
+@pytest.mark.parametrize('cbc', ['bt-ee', 'three', 'o2'])
 def test_cbc_proxy_one_2_many_create_and_send_invokes_function(
     mocker,
     cbc_proxy_client,
@@ -86,7 +86,7 @@ def test_cbc_proxy_one_2_many_create_and_send_invokes_function(
     cbc,
     expected_language,
 ):
-    cbc_proxy = cbc_proxy_client.get_proxy(cbc)
+    cbc_proxy = cbc_proxy_client.get_proxy('ee') if cbc == 'bt-ee' else cbc_proxy_client.get_proxy(cbc)
 
     identifier = 'my-identifier'
     headline = 'my-headline'
@@ -135,9 +135,9 @@ def test_cbc_proxy_one_2_many_create_and_send_invokes_function(
     assert payload['language'] == expected_language
 
 
-@pytest.mark.parametrize('cbc', ['ee', 'three', 'o2'])
+@pytest.mark.parametrize('cbc', ['bt-ee', 'three', 'o2'])
 def test_cbc_proxy_one_2_many_cancel_invokes_function(mocker, cbc_proxy_client, cbc):
-    cbc_proxy = cbc_proxy_client.get_proxy(cbc)
+    cbc_proxy = cbc_proxy_client.get_proxy('ee') if cbc == 'bt-ee' else cbc_proxy_client.get_proxy(cbc)
 
     identifier = 'my-identifier'
     MockProviderMessage = namedtuple(
@@ -310,13 +310,13 @@ def test_cbc_proxy_vodafone_cancel_invokes_function(mocker, cbc_proxy_vodafone):
     assert payload['sent'] == sent
 
 
-@pytest.mark.parametrize('cbc', ['ee', 'vodafone', 'three', 'o2'])
+@pytest.mark.parametrize('cbc', ['bt-ee', 'vodafone', 'three', 'o2'])
 def test_cbc_proxy_will_failover_to_second_lambda_if_function_error(
     mocker,
     cbc_proxy_client,
     cbc
 ):
-    cbc_proxy = cbc_proxy_client.get_proxy(cbc)
+    cbc_proxy = cbc_proxy_client.get_proxy('ee') if cbc == 'bt-ee' else cbc_proxy_client.get_proxy(cbc)
 
     ld_client_mock = mocker.patch.object(
         cbc_proxy,
@@ -362,13 +362,13 @@ def test_cbc_proxy_will_failover_to_second_lambda_if_function_error(
     ]
 
 
-@pytest.mark.parametrize('cbc', ['ee', 'vodafone', 'three', 'o2'])
+@pytest.mark.parametrize('cbc', ['bt-ee', 'vodafone', 'three', 'o2'])
 def test_cbc_proxy_will_failover_to_second_lambda_if_invoke_error(
     mocker,
     cbc_proxy_client,
     cbc
 ):
-    cbc_proxy = cbc_proxy_client.get_proxy(cbc)
+    cbc_proxy = cbc_proxy_client.get_proxy('ee') if cbc == 'bt-ee' else cbc_proxy_client.get_proxy(cbc)
 
     ld_client_mock = mocker.patch.object(
         cbc_proxy,
@@ -409,11 +409,11 @@ def test_cbc_proxy_will_failover_to_second_lambda_if_invoke_error(
     ]
 
 
-@pytest.mark.parametrize('cbc', ['ee', 'vodafone', 'three', 'o2'])
+@pytest.mark.parametrize('cbc', ['bt-ee', 'vodafone', 'three', 'o2'])
 def test_cbc_proxy_create_and_send_tries_failover_lambda_on_invoke_error_and_raises_if_both_invoke_error(
     mocker, cbc_proxy_client, cbc
 ):
-    cbc_proxy = cbc_proxy_client.get_proxy(cbc)
+    cbc_proxy = cbc_proxy_client.get_proxy('ee') if cbc == 'bt-ee' else cbc_proxy_client.get_proxy(cbc)
 
     ld_client_mock = mocker.patch.object(
         cbc_proxy,
@@ -452,11 +452,11 @@ def test_cbc_proxy_create_and_send_tries_failover_lambda_on_invoke_error_and_rai
     ]
 
 
-@pytest.mark.parametrize('cbc', ['ee', 'vodafone', 'three', 'o2'])
+@pytest.mark.parametrize('cbc', ['bt-ee', 'vodafone', 'three', 'o2'])
 def test_cbc_proxy_create_and_send_tries_failover_lambda_on_function_error_and_raises_if_both_function_error(
     mocker, cbc_proxy_client, cbc
 ):
-    cbc_proxy = cbc_proxy_client.get_proxy(cbc)
+    cbc_proxy = cbc_proxy_client.get_proxy('ee') if cbc == 'bt-ee' else cbc_proxy_client.get_proxy(cbc)
 
     ld_client_mock = mocker.patch.object(
         cbc_proxy,
@@ -532,9 +532,9 @@ def test_cbc_proxy_send_canary_invokes_function(mocker, cbc_proxy_client):
     assert payload['identifier'] == identifier
 
 
-@pytest.mark.parametrize('cbc', ['ee', 'three', 'o2'])
+@pytest.mark.parametrize('cbc', ['bt-ee', 'three', 'o2'])
 def test_cbc_proxy_one_2_many_send_link_test_invokes_function(mocker, cbc_proxy_client, cbc):
-    cbc_proxy = cbc_proxy_client.get_proxy(cbc)
+    cbc_proxy = cbc_proxy_client.get_proxy('ee') if cbc == 'bt-ee' else cbc_proxy_client.get_proxy(cbc)
 
     identifier = str(uuid.uuid4())
 

tests/app/db.py

@@ -1022,7 +1022,6 @@ def create_broadcast_message(
     starts_at=None,
     finishes_at=None,
     areas=None,
-    stubbed=False
 ):
     if template:
         service = template.service
@@ -1050,8 +1049,7 @@ def create_broadcast_message(
         finishes_at=finishes_at,
         created_by_id=created_by.id if created_by else service.created_by_id,
         areas=areas or {},
-        content=content,
-        stubbed=stubbed
+        content=content
     )
     db.session.add(broadcast_message)
     db.session.commit()

tests/app/service/test_rest.py

@@ -248,9 +248,6 @@ def test_get_service_by_id(admin_request, sample_service):
     assert set(json_resp['data'].keys()) == {
         'active',
         'allowed_broadcast_provider',
-        'billing_contact_email_addresses',
-        'billing_contact_names',
-        'billing_reference',
         'consent_to_research',
         'contact_link',
         'count_as_live',
@@ -269,7 +266,6 @@ def test_get_service_by_id(admin_request, sample_service):
         'organisation_type',
         'permissions',
         'prefix_sms',
-        'purchase_order_number',
         'rate_limit',
         'research_mode',
         'restricted',

tests/app/test_request_size.py

@@ -1,36 +0,0 @@
-import pytest
-import json
-
-@pytest.mark.parametrize('endpoint, max_content_length, expected_status_code', [
-    ("/_status", 5*1024*1024, 413),
-    ("/provider-details", 5*1024*1024, 413),
-    ("/v2/notifications/email", 5*1024*1024, 413),
-
-    ("/_status", None, 200),
-    ("/provider-details", None, 405),
-    ("/v2/notifications/email", None, 401),
-])
-def test_request_status_when_content_length_is_set(
-    notify_api,
-    sample_email_template_with_placeholders,
-    mocker,
-    endpoint,
-    max_content_length,
-    expected_status_code):
-
-    notify_api.config['MAX_CONTENT_LENGTH'] = max_content_length
-    large_name = "J" * (max_content_length or 1 + 1)
-    data = {
-        'email_address': 'ok@ok.com',
-        'template_id': str(sample_email_template_with_placeholders.id),
-        'personalisation': {
-            'name': large_name
-        }
-    }
-    with notify_api.test_client() as client:
-        response = client.post(
-            path=endpoint,
-            data=json.dumps(data),
-            headers=[('Content-Type', 'application/json')])
-
-        assert response.status_code == expected_status_code