Merge pull request #72 from authorizon/fix_logs

Fix logging as JSON to centralized drain - some log messages were not JSONable

Author: asafc

opal_client/client.py

@@ -184,7 +184,7 @@ async def launch_policy_store_dependent_tasks(self):
             for task in asyncio.as_completed([self.launch_policy_updater(), self.launch_data_updater()]):
                 await task
         except websockets.exceptions.InvalidStatusCode as err:
-            logger.error("Failed to launch background task -- {err}", err=err)
+            logger.error("Failed to launch background task -- {err}", err=repr(err))
             self._trigger_shutdown()
 
     async def maybe_init_healthcheck_policy(self):
@@ -208,13 +208,13 @@ async def maybe_init_healthcheck_policy(self):
         try:
             healthcheck_policy_code = open(healthcheck_policy_path, 'r').read()
         except IOError as err:
-            logger.error("Critical: Cannot read healthcheck policy: {err}", err=err)
+            logger.error("Critical: Cannot read healthcheck policy: {err}", err=repr(err))
             raise
 
         try:
             await self.policy_store.init_healthcheck_policy(policy_id=healthcheck_policy_relpath, policy_code=healthcheck_policy_code)
         except aiohttp.ClientError as err:
-            logger.error("Failed to connect to OPA agent while init healthcheck policy -- {err}", err=err)
+            logger.error("Failed to connect to OPA agent while init healthcheck policy -- {err}", err=repr(err))
             raise
 
     def _trigger_shutdown(self):

opal_client/data/updater.py

@@ -101,7 +101,7 @@ def trigger_data_update(self, update: DataUpdate):
         # make sure the id has a unique id for tracking
         if update.id is None:
             update.id = uuid.uuid4().hex
-        logger.info("Triggering data update with id: {id}", update=update, id=update.id)
+        logger.info("Triggering data update with id: {id}", id=update.id)
         asyncio.create_task(self.update_policy_data(
             update, policy_store=self._policy_store, data_fetcher=self._data_fetcher))
 
@@ -136,7 +136,7 @@ async def get_base_policy_data(self, config_url: str = None, data_fetch_reason="
             config_url (str, optional): URL to retrive data sources config from. Defaults to None ( self._data_sources_config_url).
             data_fetch_reason (str, optional): Reason to log for the update operation. Defaults to "Initial load".
         """
-        logger.info("Performing data configuration, reason: {reason}", reason={data_fetch_reason})
+        logger.info("Performing data configuration, reason: {reason}", reason=data_fetch_reason)
         sources_config = await self.get_policy_data_config(url=config_url)
         # translate config to a data update
         entries = sources_config.entries
@@ -200,7 +200,7 @@ async def stop(self):
             try:
                 await self._subscriber_task
             except asyncio.CancelledError as exc:
-                logger.debug("DataUpdater subscriber task was force-cancelled: {e}", exc=exc)
+                logger.debug("DataUpdater subscriber task was force-cancelled: {exc}", exc=repr(exc))
             self._subscriber_task = None
             logger.debug("DataUpdater subscriber task was cancelled")
 
@@ -243,7 +243,7 @@ async def report_update_results(self, update: DataUpdate, reports: List[DataEntr
                 callback_config.data = whole_report.json()
                 urls.append((url, callback_config))
 
-            logger.info("Reporting the update to requested callbacks", urls=urls)
+            logger.info("Reporting the update to requested callbacks", urls=repr(urls))
             report_results = await data_fetcher.handle_urls(urls)
             # log reports which we failed to send
             for (url, config), result in zip(urls,report_results):
@@ -270,7 +270,7 @@ async def update_policy_data(self, update: DataUpdate = None, policy_store: Base
             urls = [(entry.url, entry.config) for entry in entries]
 
         # get the data for the update
-        logger.info("Fetching policy data", urls=urls)
+        logger.info("Fetching policy data", urls=repr(urls))
         # Urls may be None - handle_urls has a default for None
         policy_data_with_urls = await data_fetcher.handle_urls(urls)
         # Save the data from the update

opal_client/policy/fetcher.py

@@ -16,7 +16,7 @@ def force_valid_bundle(bundle) -> PolicyBundle:
     try:
         return PolicyBundle(**bundle)
     except ValidationError as e:
-        logger.warning("server returned invalid bundle: {err}", bundle=bundle, err=e)
+        logger.warning("server returned invalid bundle: {err}", bundle=bundle, err=repr(e))
         raise
 
 async def throw_if_bad_status_code(response: aiohttp.ClientResponse, expected: List[int]) -> aiohttp.ClientResponse:
@@ -58,7 +58,7 @@ async def fetch_policy_bundle(
         try:
             return await attempter(directories=directories, base_hash=base_hash)
         except Exception as err:
-            logger.warning("Failed all attempts to fetch bundle, got error: {err}", err=err)
+            logger.warning("Failed all attempts to fetch bundle, got error: {err}", err=repr(err))
             return None
 
     async def _fetch_policy_bundle(
@@ -90,6 +90,6 @@ async def _fetch_policy_bundle(
                     bundle = await response.json()
                     return force_valid_bundle(bundle)
             except aiohttp.ClientError as e:
-                logger.warning("server connection error: {err}", err=e)
+                logger.warning("server connection error: {err}", err=repr(e))
                 raise
 

opal_client/policy/updater.py

@@ -142,7 +142,7 @@ async def stop(self):
             try:
                 await self._subscriber_task
             except asyncio.CancelledError as exc:
-                logger.debug("PolicyUpdater subscriber task was force-cancelled: {e}", exc=exc)
+                logger.debug("PolicyUpdater subscriber task was force-cancelled: {exc}", exc=repr(exc))
             self._subscriber_task = None
             logger.debug("PolicyUpdater subscriber task was cancelled")
 

opal_client/policy_store/base_policy_store_client.py

@@ -152,5 +152,5 @@ async def end_transcation(self, exc_type=None, exc=None, tb=None, transaction_id
             # transaction context. If they fail, we do nothing special.
             logger.error("Cannot write to OPAL transaction log, transaction id={id}, error={err}",
                 id=transaction.id,
-                err=e
+                err=repr(e)
             )

opal_client/policy_store/opa_client.py

@@ -165,7 +165,7 @@ async def set_policy(self, policy_id: str, policy_code: str, transaction_id:Opti
                 ) as opa_response:
                     return await proxy_response_unless_invalid(opa_response, accepted_status_codes=[status.HTTP_200_OK])
             except aiohttp.ClientError as e:
-                logger.warning("Opa connection error: {err}", err=e)
+                logger.warning("Opa connection error: {err}", err=repr(e))
                 raise
 
     @fail_silently()
@@ -179,7 +179,7 @@ async def get_policy(self, policy_id: str) -> Optional[str]:
                     result = await opa_response.json()
                     return result.get("result", {}).get("raw", None)
             except aiohttp.ClientError as e:
-                logger.warning("Opa connection error: {err}", err=e)
+                logger.warning("Opa connection error: {err}", err=repr(e))
                 raise
 
     @affects_transaction
@@ -195,7 +195,7 @@ async def delete_policy(self, policy_id: str, transaction_id:Optional[str]=None)
                         status.HTTP_404_NOT_FOUND
                     ])
             except aiohttp.ClientError as e:
-                logger.warning("Opa connection error: {err}", err=e)
+                logger.warning("Opa connection error: {err}", err=repr(e))
                 raise
 
     @fail_silently()
@@ -209,7 +209,7 @@ async def get_policy_module_ids(self) -> List[str]:
                     result = await opa_response.json()
                     return OpaClient._extract_module_ids_from_policies_json(result)
             except aiohttp.ClientError as e:
-                logger.warning("Opa connection error: {err}", err=e)
+                logger.warning("Opa connection error: {err}", err=repr(e))
                 raise
 
     @staticmethod
@@ -291,12 +291,12 @@ async def _set_policy_data_from_bundle_data_module(self, module: DataModule, has
                 path=module_path,
             )
         except aiohttp.ClientError as e:
-            logger.warning("Opa connection error: {err}", err=e)
+            logger.warning("Opa connection error: {err}", err=repr(e))
             raise
         except json.JSONDecodeError as e:
             logger.warning(
                 "bundle contains non-json data module: {module_path}",
-                err=e,
+                err=repr(e),
                 module_path=module_path,
                 bundle_hash=hash
             )
@@ -317,7 +317,7 @@ async def set_policy_data(self, policy_data: JsonableValue, path: str = "", tran
                         status.HTTP_304_NOT_MODIFIED
                     ])
             except aiohttp.ClientError as e:
-                logger.warning("Opa connection error: {err}", err=e)
+                logger.warning("Opa connection error: {err}", err=repr(e))
                 raise
 
     @affects_transaction
@@ -337,7 +337,7 @@ async def delete_policy_data(self, path: str = "", transaction_id:Optional[str]=
                         status.HTTP_404_NOT_FOUND
                     ])
             except aiohttp.ClientError as e:
-                logger.warning("Opa connection error: {err}", err=e)
+                logger.warning("Opa connection error: {err}", err=repr(e))
                 raise
 
     @affects_transaction
@@ -355,7 +355,7 @@ async def patch_data(self, path: str, patch_document: JSONPatchDocument, transac
                 ) as opa_response:
                     return await proxy_response_unless_invalid(opa_response, accepted_status_codes=[status.HTTP_204_NO_CONTENT])
             except aiohttp.ClientError as e:
-                logger.warning("Opa connection error: {err}", err=e)
+                logger.warning("Opa connection error: {err}", err=repr(e))
                 raise
 
     @fail_silently()
@@ -375,7 +375,7 @@ async def get_data(self, path: str) -> Dict:
                 async with session.get(f"{self._opa_url}/data/{path}") as opa_response:
                     return await opa_response.json()
         except aiohttp.ClientError as e:
-            logger.warning("Opa connection error: {err}", err=e)
+            logger.warning("Opa connection error: {err}", err=repr(e))
             raise
 
     @retry(**RETRY_CONFIG)
@@ -404,7 +404,7 @@ async def get_data_with_input(self, path: str, input: BaseModel) -> Dict:
                 ) as opa_response:
                     return await proxy_response(opa_response)
         except aiohttp.ClientError as e:
-            logger.warning("Opa connection error: {err}", err=e)
+            logger.warning("Opa connection error: {err}", err=repr(e))
             raise
 
     @retry(**RETRY_CONFIG)

opal_common/authentication/signer.py

@@ -86,7 +86,7 @@ def _verify_crypto_keys(self):
             try:
                 jwt.decode(token, self._public_key, algorithms=[self._algorithm])
             except jwt.PyJWTError as exc:
-                logger.info("JWT Signer key verification failed with error: {err}", err=exc)
+                logger.info("JWT Signer key verification failed with error: {err}", err=repr(exc))
                 raise InvalidJWTCryptoKeysException("private key and public key do not match!") from exc
             # save jwk
             self._jwk: PyJWK = PyJWK.from_json(self.get_jwk(), algorithm=self._algorithm)

opal_common/fetcher/engine/fetching_engine.py

@@ -64,7 +64,7 @@ async def __aenter__(self):
 
     async def __aexit__(self, exc_type, exc, tb):
         if (exc is not None):
-            logger.error("Error occurred within FetchingEngine context", exc_info=(exc_type, exc, tb))
+            logger.error("Error occurred within FetchingEngine context", exc_info=repr((exc_type, exc, tb)))
         await self.terminate_workers()
 
     async def terminate_workers(self):

opal_server/deps/authentication.py

@@ -29,7 +29,7 @@ def verify_logged_in(signer: JWTSigner, token: Optional[str]) -> UUID:
     forces bearer token authentication with valid JWT or throws 401.
     """
     if not signer.enabled:
-        logger.debug("signer diabled, cannot verify request!")
+        logger.debug("signer diabled, cannot verify requests!")
         return
     if token is None:
         raise Unauthorized(token=token, description="access token was not provided")

opal_server/policy/watcher/task.py

@@ -71,5 +71,5 @@ async def _fail(self, exc: Exception):
         """
         called when the watcher fails, and stops all tasks gracefully
         """
-        logger.error("watcher failed with exception: {err}", err=exc)
+        logger.error("watcher failed with exception: {err}", err=repr(exc))
         await self.stop()