Sending Log Data to Splunk using Python

Question

I have an app that detects file changes, backups, and syncs files to Azure.

I currently have a logger setup writes log events to a file called log.log. I also have event data streaming to stdout. This is my current working code.

I’d like to send log data to Splunk via requests.post() or logging.handlers.HTTPHandler.

Question: How do I set up an HTTP Handler in Python logging?

(I need to become more familiar with the advanced features of logging in Python.)

import logging


def setup_logger(logger_name:str=__name__, logfile:str='log.log'):
    """ Standard Logging: std out and log file.

    Args:
        logger_name (str, optional): Logger Name. Defaults to __name__.
        logfile (str, optional): Log File Name. Defaults to 'app.log'.

    Returns:
        logging: function to set logging as a object.
    """
    logger = logging.getLogger(logger_name)
    logger.setLevel(logging.INFO)
    
    fh = logging.FileHandler(logfile)
    fh.setLevel(logging.INFO)
    
    ch = logging.StreamHandler()
    ch.setLevel(logging.INFO)
    
    formatter = logging.Formatter(
        '%(asctime)s | %(name)s | %(levelname)s | %(message)s', 
        '%m-%d-%Y %H:%M:%S')    
    
    fh.setFormatter(formatter)
    ch.setFormatter(formatter)

    logger.addHandler(fh)
    logger.addHandler(ch)

    

    return logger

if __name__ == "__main__":
    logger=setup_logger('logger','log-sample.log') # Creates a test file vs default log.log
    logger.info("My Logger has been initialized")

Currently I’m trying to send test data to Splunk via this code example (before I figure out the logging issue):

import requests

# Set up the Splunk HEC URL and token
splunk_url = "http://127.0.0.1:8088/services/collector/event"
splunk_token = "57489f00-605e-4f2a-8df3-123456789abcdef="

# Set up the log event data
log_data = {
    "event": "This is a test log event",
    "sourcetype": "my_sourcetype",
    "index": "test_index"
}

# Send the log event to Splunk
response = requests.post(splunk_url, json=log_data, headers={
    "Authorization": f"Splunk {splunk_token}"
})

# Check the response status code to make sure the request was successful
if response.status_code == 200:
    print("Log event sent to Splunk successfully")
else:
    print(f"Error sending log event to Splunk: {response.text}")

Answer 1

I found the solution myself.

import logging
import requests
import urllib3

urllib3.disable_warnings() # using default cert.

url = "https://127.0.0.1:8088/services/collector/event"
headers = {"Authorization": "Splunk 09584dbe-183b-4d14-9ee9-be66a37b331a"}
index = 'test_index'

class CustomHttpHandler(logging.Handler):
    
    def __init__(self, url:str, headers:dict, index:str) -> None:
    
        self.url = url
        self.headers = headers
        self.index = index
        super().__init__()
    
    
    def emit(self, record:str) -> exec:
        '''
        This function gets called when a log event gets emitted. It receives a
        record, formats it and sends it to the url
        Parameters:
            record: a log record (created by logging module)
        '''
        log_entry = self.format(record)
        response = requests.post(
            url=self.url, headers=self.headers, 
            json={"index": self.index, "event": log_entry}, 
            verify=False)


def setup_logger(logger_name:str=__name__, logfile:str='log.log'):
    """ Standard Logging: std out and log file.
        1.creates file handler which logs even debug messages: fh
        2.creates console handler with a higher log level: ch
        3.creates formatter and add it to the handlers: formatter, setFormatter
        4.adds the handlers to the logger: addHandler

    Args:
        logger_name (str, optional): Logger Name. Defaults to __name__.
        logfile (str, optional): Log File Name. Defaults to 'app.log'.

    Returns:
        logging: function to set logging as a object.
    """
    logger = logging.getLogger(logger_name)
    logger.setLevel(logging.INFO)
    
    fh = logging.FileHandler(logfile)
    fh.setLevel(logging.INFO)
    
    ch = logging.StreamHandler()
    ch.setLevel(logging.INFO)
    
    formatter = logging.Formatter(
        '%(asctime)s | %(name)s | %(levelname)s | %(message)s', 
        '%m-%d-%Y %H:%M:%S')    
    
    splunk_handler = CustomHttpHandler(url=url, headers=headers, index=index)
    
    fh.setFormatter(formatter)
    ch.setFormatter(formatter)
    splunk_handler.setFormatter(formatter)

    logger.addHandler(fh)
    logger.addHandler(ch)
    logger.addHandler(splunk_handler)
    
    return logger

if __name__ == "__main__":
    logger=setup_logger('logger','app.log')
    logger.info("My Logger has been initialized")