create a log handler for log analytics

[TordAreStromsnes]

We need to setup resources that can be used for creating logs in log analytics workspace for our jobs in databricks. We will probably need some sort of custom table. The schema should be discussed between teams, so we find something that could be used by all.

setup is not that complex. azure sdk comes with some helpers. Example from google search:

Install
pip install azure-monitor-ingestion

Setup

import logging
from azure.monitor.ingestion import LogsIngestionClient
from azure.identity import DefaultAzureCredential

workspace_id = "YOUR_WORKSPACE_ID"
rule_id = "YOUR_DATA_COLLECTION_RULE_ID"
stream_name = "YOUR_STREAM_NAME" # e.g., "Custom-MyApplicationLogs"

credential = DefaultAzureCredential()
client = LogsIngestionClient(endpoint=f"https://{workspace_id}.ods.opinsights.azure.com", credential=credential)


class AzureLogAnalyticsHandler(logging.Handler):
        def __init__(self, client, rule_id, stream_name):
            super().__init__()
            self.client = client
            self.rule_id = rule_id
            self.stream_name = stream_name

        def emit(self, record):
            try:
                log_entry = self.format(record)
                self.client.upload(rule_id=self.rule_id, stream_name=self.stream_name, logs=[log_entry])
            except Exception:
                self.handleError(record)

        def format(self, record):
            # Customize the log format as needed for your Log Analytics table schema
            log_data = {
                "timestamp": self.formatTime(record, datefmt="%Y-%m-%dT%H:%M:%SZ"),
                "level": record.levelname,
                "message": record.getMessage(),
                "loggerName": record.name,
                "processId": record.process,
                "threadId": record.thread,
                # Add other relevant record attributes
            }
            return log_data

Then we should be able to use this in our pipeline quite simple with a setup like this:

logger = logging.getLogger("my_app_logger")
logger.setLevel(logging.INFO)

azure_handler = AzureLogAnalyticsHandler(client, rule_id, stream_name)
logger.addHandler(azure_handler)

# Example usage
logger.info("This is an informational message.")
logger.warning("A warning occurred in the application.")
logger.error("An error happened!", exc_info=True)