Make log send async

CHANGELOG.md

@@ -28,6 +28,11 @@ The types of changes are:
 * Fix analytics opt out environment variable name [#1170](https://github.com/ethyca/fidesops/pull/1170)
 * Added how to view a subject request history and reprocess a subject request [#1164](https://github.com/ethyca/fidesops/pull/1164)
 
+### Fixed
+
+* Fix issue with fideslog event loop errors [#1174](https://github.com/ethyca/fidesops/pull/1174)
+
+
 ## [1.7.2](https://github.com/ethyca/fidesops/compare/1.7.1...1.7.2)
 
 ### Added

dev-requirements.txt

@@ -8,6 +8,7 @@ packaging==21.3
 pdbpp
 pre-commit==2.20.0
 pylint==2.14.5
+pytest-asyncio==0.19.0
 pytest-cov==3.0.0
 pytest-env==0.6.2
 pytest==7.1.2

pyproject.toml

@@ -137,3 +137,4 @@ addopts = ["--cov-report=term-missing",
             "-vv",
             "--no-cov-on-fail",
             "--disable-pytest-warnings"]
+asyncio_mode = "auto"

src/fidesops/main.py

@@ -1,3 +1,4 @@
+import asyncio
 import logging
 import os
 import re
@@ -92,13 +93,13 @@ async def dispatch_log_request(request: Request, call_next: Callable) -> Respons
         return response
 
     except Exception as e:
-        prepare_and_log_request(
+        await prepare_and_log_request(
             endpoint, request.url.hostname, 500, now, fides_source, e.__class__.__name__
         )
         raise
 
 
-def prepare_and_log_request(
+async def prepare_and_log_request(
     endpoint: str,
     hostname: Optional[str],
     status_code: int,
@@ -113,7 +114,7 @@ def prepare_and_log_request(
     # this check prevents AnalyticsEvent from being called with invalid endpoint during unit tests
     if config.root_user.analytics_opt_out:
         return
-    send_analytics_event(
+    await send_analytics_event(
         AnalyticsEvent(
             docker=in_docker_container(),
             event=Event.endpoint_call.value,
@@ -255,11 +256,13 @@ def start_webserver() -> None:
         logger.info("Starting scheduled request intake...")
         initiate_scheduled_request_intake()
 
-    send_analytics_event(
-        AnalyticsEvent(
-            docker=in_docker_container(),
-            event=Event.server_start.value,
-            event_created_at=datetime.now(tz=timezone.utc),
+    asyncio.run(
+        send_analytics_event(
+            AnalyticsEvent(
+                docker=in_docker_container(),
+                event=Event.server_start.value,
+                event_created_at=datetime.now(tz=timezone.utc),
+            )
         )
     )
 

src/fidesops/ops/analytics.py

@@ -35,11 +35,13 @@ def accessed_through_local_host(hostname: Optional[str]) -> bool:
 )
 
 
-def send_analytics_event(event: AnalyticsEvent) -> None:
+async def send_analytics_event(event: AnalyticsEvent) -> None:
     if config.root_user.analytics_opt_out:
         return
     try:
-        analytics_client.send(event)
+        await analytics_client._AnalyticsClient__send(  # pylint: disable=protected-access
+            event
+        )
     except AnalyticsError as err:
         logger.warning("Error sending analytics event: %s", err)
     else:

src/fidesops/ops/api/v1/endpoints/drp_endpoints.py

@@ -55,7 +55,7 @@
     status_code=HTTP_200_OK,
     response_model=PrivacyRequestDRPStatusResponse,
 )
-def create_drp_privacy_request(
+async def create_drp_privacy_request(
     *,
     cache: FidesopsRedis = Depends(deps.get_cache),
     db: Session = Depends(deps.get_db),

src/fidesops/ops/api/v1/endpoints/privacy_request_endpoints.py

@@ -133,7 +133,7 @@ def get_privacy_request_or_error(
     status_code=HTTP_200_OK,
     response_model=BulkPostPrivacyRequests,
 )
-def create_privacy_request(
+async def create_privacy_request(
     *,
     db: Session = Depends(deps.get_db),
     data: conlist(PrivacyRequestCreate, max_items=50) = Body(...),  # type: ignore
@@ -719,7 +719,7 @@ def get_request_preview_queries(
     status_code=HTTP_200_OK,
     response_model=PrivacyRequestResponse,
 )
-def resume_privacy_request(
+async def resume_privacy_request(
     privacy_request_id: str,
     *,
     db: Session = Depends(deps.get_db),
@@ -777,7 +777,7 @@ def validate_manual_input(
                 )
 
 
-def resume_privacy_request_with_manual_input(
+async def resume_privacy_request_with_manual_input(
     privacy_request_id: str,
     db: Session,
     expected_paused_step: PausedStep,
@@ -868,7 +868,7 @@ def resume_privacy_request_with_manual_input(
         Security(verify_oauth_client, scopes=[PRIVACY_REQUEST_CALLBACK_RESUME])
     ],
 )
-def resume_with_manual_input(
+async def resume_with_manual_input(
     privacy_request_id: str,
     *,
     db: Session = Depends(deps.get_db),
@@ -879,7 +879,7 @@ def resume_with_manual_input(
 
     If there's no manual data to submit, pass in an empty list to resume the privacy request.
     """
-    return resume_privacy_request_with_manual_input(
+    return await resume_privacy_request_with_manual_input(
         privacy_request_id=privacy_request_id,
         db=db,
         expected_paused_step=PausedStep.access,
@@ -895,7 +895,7 @@ def resume_with_manual_input(
         Security(verify_oauth_client, scopes=[PRIVACY_REQUEST_CALLBACK_RESUME])
     ],
 )
-def resume_with_erasure_confirmation(
+async def resume_with_erasure_confirmation(
     privacy_request_id: str,
     *,
     db: Session = Depends(deps.get_db),
@@ -906,7 +906,7 @@ def resume_with_erasure_confirmation(
 
     If no rows were masked, pass in a 0 to resume the privacy request.
     """
-    return resume_privacy_request_with_manual_input(
+    return await resume_privacy_request_with_manual_input(
         privacy_request_id=privacy_request_id,
         db=db,
         expected_paused_step=PausedStep.erasure,
@@ -922,7 +922,7 @@ def resume_with_erasure_confirmation(
         Security(verify_oauth_client, scopes=[PRIVACY_REQUEST_CALLBACK_RESUME])
     ],
 )
-def restart_privacy_request_from_failure(
+async def restart_privacy_request_from_failure(
     privacy_request_id: str,
     *,
     db: Session = Depends(deps.get_db),
@@ -1020,7 +1020,7 @@ def review_privacy_request(
     status_code=HTTP_200_OK,
     response_model=PrivacyRequestResponse,
 )
-def verify_identification_code(
+async def verify_identification_code(
     privacy_request_id: str,
     *,
     db: Session = Depends(deps.get_db),

src/fidesops/ops/graph/analytics_events.py

@@ -21,20 +21,20 @@
     from fidesops.ops.task.graph_task import GraphTask
 
 
-def fideslog_graph_failure(event: Optional[AnalyticsEvent]) -> None:
+async def fideslog_graph_failure(event: Optional[AnalyticsEvent]) -> None:
     """Send an Analytics Event if privacy request execution has failed"""
     if config.root_user.analytics_opt_out or not event:
         return
 
-    send_analytics_event(event)
+    await send_analytics_event(event)
 
 
-def fideslog_graph_rerun(event: Optional[AnalyticsEvent]) -> None:
+async def fideslog_graph_rerun(event: Optional[AnalyticsEvent]) -> None:
     """Send an Analytics Event if a privacy request has been reprocessed, comparing its graph to the previous graph"""
     if config.root_user.analytics_opt_out or not event:
         return
 
-    send_analytics_event(event)
+    await send_analytics_event(event)
 
 
 def prepare_rerun_graph_analytics_event(

src/fidesops/ops/service/privacy_request/request_runner_service.py

@@ -49,6 +49,7 @@
 )
 from fidesops.ops.util.collection_util import Row
 from fidesops.ops.util.logger import Pii, _log_exception, _log_warning
+from fidesops.ops.util.wrappers import sync
 
 logger = get_task_logger(__name__)
 
@@ -189,7 +190,8 @@ def session(self) -> ContextManager[Session]:
 
 
 @celery_app.task(base=DatabaseTask, bind=True)
-def run_privacy_request(
+@sync
+async def run_privacy_request(
     self: DatabaseTask,
     privacy_request_id: str,
     from_webhook_id: Optional[str] = None,
@@ -202,6 +204,9 @@ def run_privacy_request(
         2. Take the provided identity data
         3. Start the access request / erasure request execution
         4. When finished, upload the results to the configured storage destination if applicable
+
+    Celery does not like for the function to be async so the @sync decorator runs the
+    coroutine for it.
     """
     if from_step is not None:
         # Re-cast `from_step` into an Enum to enforce the validation since unserializable objects
@@ -248,7 +253,7 @@ def run_privacy_request(
             if (
                 from_step != PausedStep.erasure
             ):  # Skip if we're resuming from erasure step
-                access_result: Dict[str, List[Row]] = run_access_request(
+                access_result: Dict[str, List[Row]] = await run_access_request(
                     privacy_request=privacy_request,
                     policy=policy,
                     graph=dataset_graph,
@@ -267,7 +272,7 @@ def run_privacy_request(
 
             if policy.get_rules_for_action(action_type=ActionType.erasure):
                 # We only need to run the erasure once until masking strategies are handled
-                run_erasure(
+                await run_erasure(
                     privacy_request=privacy_request,
                     policy=policy,
                     graph=dataset_graph,
@@ -287,7 +292,9 @@ def run_privacy_request(
         except BaseException as exc:  # pylint: disable=broad-except
             privacy_request.error_processing(db=session)
             # If dev mode, log traceback
-            fideslog_graph_failure(failed_graph_analytics_event(privacy_request, exc))
+            await fideslog_graph_failure(
+                failed_graph_analytics_event(privacy_request, exc)
+            )
             _log_exception(exc, config.dev_mode)
             return
 

src/fidesops/ops/task/graph_task.py

@@ -584,7 +584,7 @@ def g() -> List[Dict[str, Any]]:
     return g
 
 
-def run_access_request(
+async def run_access_request(
     privacy_request: PrivacyRequest,
     policy: Policy,
     graph: DatasetGraph,
@@ -624,7 +624,7 @@ def termination_fn(
         dsk[TERMINATOR_ADDRESS] = (termination_fn, *end_nodes)
         update_mapping_from_cache(dsk, resources, start_function)
 
-        fideslog_graph_rerun(
+        await fideslog_graph_rerun(
             prepare_rerun_graph_analytics_event(
                 privacy_request, env, end_nodes, resources, ActionType.access
             )
@@ -668,7 +668,7 @@ def update_erasure_mapping_from_cache(
         )
 
 
-def run_erasure(  # pylint: disable = too-many-arguments, too-many-locals
+async def run_erasure(  # pylint: disable = too-many-arguments, too-many-locals
     privacy_request: PrivacyRequest,
     policy: Policy,
     graph: DatasetGraph,
@@ -707,7 +707,7 @@ def termination_fn(*dependent_values: int) -> Tuple[int, ...]:
         # terminator function waits for all keys
         dsk[TERMINATOR_ADDRESS] = (termination_fn, *env.keys())
         update_erasure_mapping_from_cache(dsk, resources, start_function)
-        fideslog_graph_rerun(
+        await fideslog_graph_rerun(
             prepare_rerun_graph_analytics_event(
                 privacy_request, env, end_nodes, resources, ActionType.erasure
             )

src/fidesops/ops/util/wrappers.py

@@ -0,0 +1,17 @@
+import asyncio
+from functools import wraps
+from typing import Any, Callable
+
+
+def sync(func: Callable) -> Any:
+    """Converts an async function into a sync function."""
+
+    @wraps(func)
+    def wrap(*args: Any, **kwargs: Any) -> Any:
+        try:
+            loop = asyncio.get_running_loop()
+            return loop.run_until_complete(func(*args, **kwargs))
+        except RuntimeError:
+            return asyncio.run(func(*args, **kwargs))
+
+    return wrap

tests/ops/conftest.py

@@ -1,5 +1,6 @@
 # pylint: disable=unused-wildcard-import, wildcard-import
 
+import asyncio
 import json
 import logging
 from typing import Any, Callable, Dict, Generator, List
@@ -250,3 +251,13 @@ def subject_identity_verification_not_required():
     config.execution.subject_identity_verification_required = False
     yield
     config.execution.subject_identity_verification_required = original_value
+
+
+@pytest.fixture(scope="session", autouse=True)
+def event_loop():
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        loop = asyncio.new_event_loop()
+    yield loop
+    loop.close()

tests/ops/integration_tests/saas/request_override/test_mailchimp_override_task.py

@@ -29,7 +29,7 @@
 
 @pytest.mark.integration_saas
 @pytest.mark.integration_saas_override
-def test_mailchimp_override_access_request_task(
+async def test_mailchimp_override_access_request_task(
     db,
     policy,
     mailchimp_override_connection_config,
@@ -48,7 +48,7 @@ def test_mailchimp_override_access_request_task(
     merged_graph = mailchimp_override_dataset_config.get_graph()
     graph = DatasetGraph(merged_graph)
 
-    v = graph_task.run_access_request(
+    v = await graph_task.run_access_request(
         privacy_request,
         policy,
         graph,
@@ -140,7 +140,7 @@ def test_mailchimp_override_access_request_task(
 
 @pytest.mark.integration_saas
 @pytest.mark.integration_saas_override
-def test_mailchimp_erasure_request_task(
+async def test_mailchimp_erasure_request_task(
     db,
     policy,
     erasure_policy_string_rewrite,
@@ -161,7 +161,7 @@ def test_mailchimp_erasure_request_task(
     merged_graph = mailchimp_override_dataset_config.get_graph()
     graph = DatasetGraph(merged_graph)
 
-    graph_task.run_access_request(
+    await graph_task.run_access_request(
         privacy_request,
         policy,
         graph,
@@ -170,7 +170,7 @@ def test_mailchimp_erasure_request_task(
         db,
     )
 
-    v = graph_task.run_erasure(
+    v = await graph_task.run_erasure(
         privacy_request,
         erasure_policy_string_rewrite,
         graph,