feat: add middleware to insert frontend moniroting script to response

edx_django_utils/monitoring/__init__.py

@@ -14,6 +14,7 @@
     CachedCustomMonitoringMiddleware,
     CookieMonitoringMiddleware,
     DeploymentMonitoringMiddleware,
+    FrontendMonitoringMiddleware,
     MonitoringMemoryMiddleware
 )
 from .internal.transactions import (

edx_django_utils/monitoring/internal/middleware.py

@@ -8,6 +8,7 @@
 import math
 import platform
 import random
+import re
 import warnings
 from uuid import uuid4
 
@@ -28,6 +29,9 @@
 _DEFAULT_NAMESPACE = 'edx_django_utils.monitoring'
 _REQUEST_CACHE_NAMESPACE = f'{_DEFAULT_NAMESPACE}.custom_attributes'
 
+_HTML_HEAD_REGEX = br"<\/head[^>]*>"
+_HTML_BODY_REGEX = br"<body[^>]*>"
+
 
 class DeploymentMonitoringMiddleware:
     """
@@ -462,6 +466,61 @@ def log_corrupt_cookie_headers(self, request, corrupt_cookie_count):
             log.info(piece)
 
 
+class FrontendMonitoringMiddleware:
+    """
+    Middleware for adding the forntend monitoring scripts to the response
+    """
+    def __init__(self, get_response):
+        self.get_response = get_response
+
+
+    def __call__(self, request):
+        response = self.get_response(request)
+
+        if response.status_code != 200 or not response['Content-Type'].startswith('text/html'):
+            return response
+
+        # .. setting_name: OPENEDX_TELEMETRY_FRONTEND_SCRIPTS
+        # .. setting_default: None
+        # .. setting_description: Scripts to inject to response for frontend monitoring, this can
+        #    have multiple scripts as we support multiple telemetry backends at once, so we can
+        #    provide multiple frontend scripts in a multiline string for multiple platforms tracking.
+        #    Best is to have one at a time for better performance.
+        frontend_scripts = getattr(settings, 'OPENEDX_TELEMETRY_FRONTEND_SCRIPTS', None)
+
+        if not frontend_scripts:
+            return response
+
+        if not isinstance(frontend_scripts, str):
+            # Prevent a certain kind of easy mistake.
+            raise Exception("OPENEDX_TELEMETRY_FRONTEND_SCRIPTS must be a string.")
+
+        response.content = self.inject_script(response.content, frontend_scripts)
+        return response
+
+    def inject_script(self, content, script):
+        """
+        Add script to the response, if body tag is present. 
+        """
+        body = re.search(_HTML_BODY_REGEX, content, re.IGNORECASE)
+
+        # If there is no body tag in html, don't add monitoring scripts
+        if not body:
+            return content
+
+        def insert_html_at_index(index):
+            return content[:index] + script.encode() + content[index:]
+
+        head_closing_tag = re.search(_HTML_HEAD_REGEX, content, re.IGNORECASE)
+
+        # If head tag is present, insert the monitoring scripts just before the closing of head tag
+        if head_closing_tag:
+            return insert_html_at_index(head_closing_tag.start())
+
+        # If not head tag, add scripts before the start of body tag
+        return insert_html_at_index(body.start())
+
+
 # This function should be cleaned up and made into a general logging utility, but it will first
 # need some work to make it able to handle multibyte characters.
 #

edx_django_utils/monitoring/tests/test_middleware.py

@@ -7,6 +7,7 @@
 from unittest.mock import Mock, call, patch
 
 import ddt
+from django.http import HttpRequest, HttpResponse
 from django.test import TestCase
 from django.test.client import RequestFactory
 from django.test.utils import override_settings
@@ -16,6 +17,7 @@
 from edx_django_utils.monitoring import (
     CookieMonitoringMiddleware,
     DeploymentMonitoringMiddleware,
+    FrontendMonitoringMiddleware,
     MonitoringMemoryMiddleware
 )
 
@@ -322,3 +324,58 @@ def get_mock_request(self, cookies_dict):
         for name, value in cookies_dict.items():
             factory.cookies[name] = value
         return factory.request()
+
+@ddt.ddt
+class FrontendMonitoringMiddlewareTestCase(TestCase):
+    """
+    Tests for FrontendMonitoringMiddleware.
+    """
+    def setUp(self):
+        super().setUp()
+        self.script = "<script>test script</script>"
+
+    @patch("edx_django_utils.monitoring.internal.middleware.FrontendMonitoringMiddleware.inject_script")
+    def test_frontend_middleware_without_setting_variable(self, mock_inject_script):
+        """
+        Test that middleware behaves correctly when setting variable is not defined.
+        """
+        response_html = '<html><head></head><body></body><html>'
+        middleware = FrontendMonitoringMiddleware(lambda r: HttpResponse(response_html, content_type='text/html'))
+        response = middleware(HttpRequest())
+        # Assert that the response content remains unchanged if settings not defined
+        assert response.content == response_html.encode()
+
+        mock_inject_script.assert_not_called()
+
+    @ddt.data(
+        ('<html><body></body><html>', '<body>'),
+        ('<html><head></head><body></body><html>', '</head>'),
+        ('<head></head><body></body>', '</head>'),
+        ('<body></body>', '<body>'),
+    )
+    @ddt.unpack
+    def test_frontend_middleware_script_with_body_tag(self, response_html, expected_tag):
+        """
+        Test that script is inserted at the right place.
+        """
+        with override_settings(OPENEDX_TELEMETRY_FRONTEND_SCRIPTS=self.script):
+            middleware = FrontendMonitoringMiddleware(lambda r: HttpResponse(response_html, content_type='text/html'))
+            response = middleware(HttpRequest())
+
+        # Assert that the script is inserted at the right place
+        assert f"{self.script}{expected_tag}".encode() in response.content
+
+    @ddt.data(
+        '<html></html>',
+        '<html><head></head></html>',
+        '<head></head>',
+    )
+    def test_frontend_middleware_without_body_tag(self, response_html):
+        """
+        Test that middleware behavior is correct when no body tag is found in the response.
+        """
+        with override_settings(OPENEDX_TELEMETRY_FRONTEND_SCRIPTS=self.script):
+            middleware = FrontendMonitoringMiddleware(lambda r: HttpResponse(response_html, content_type='text/html'))
+            response = middleware(HttpRequest())
+        # Assert that the response content remains unchanged if no body tag is found
+        assert response.content == response_html.encode()