如何用Splunk模块化输入Python代码格式化含无效标签的XML数据

Absolutely! Splunk modular inputs are ideal for tackling this exact scenario—cleaning up messy, invalid XML and getting it into a standardized format for Splunk. As someone who’s worked with plenty of wonky XML in Splunk, let me break down how to do this with Python step by step.

Splunk modular inputs let you run custom Python code to ingest, process, and normalize data before it hits Splunk’s index. This is perfect for your use case: you can pull the messy XML (whether it’s from a Splunk output, file, API, or another source), clean up invalid tags/formatting, and then send the structured, pretty-printed XML into Splunk as events.

• xml.etree.ElementTree: Python’s built-in XML parser for standard cases.
• lxml (optional but recommended): A more robust library that handles unstructured/invalid XML way better than the standard library (think unclosed tags, malformed attributes, etc.). Note: You’ll need to install it in your Splunk environment or package it with your app if deploying across instances.
• Regex/string utilities: To scrub out obviously invalid tags before parsing.

3.1 Build the XML Cleaning Logic #

First, write functions to clean and format the messy XML. Here’s an example that handles common issues:

import re
import xml.etree.ElementTree as ET
from xml.dom import minidom
# If using lxml, uncomment this:
# from lxml import html

def scrub_invalid_xml(raw_xml):
    # Remove tags with invalid characters in their names
    cleaned = re.sub(r'<[^a-zA-Z/_][^>]*>', '', raw_xml)
    # Auto-close unclosed self-contained tags (e.g., <br> → <br/>)
    cleaned = re.sub(r'<([a-zA-Z0-9_]+)([^>]*?)(?<!/)>', r'<\1\2/>', cleaned)
    # Remove any leftover malformed tag fragments
    cleaned = re.sub(r'<[^>]*?(>|$)', lambda m: m.group(0) if '>' in m.group(0) else '', cleaned)
    return cleaned

def pretty_print_xml(xml_string):
    try:
        # Use standard library first
        root = ET.fromstring(xml_string)
        rough_string = ET.tostring(root, 'utf-8')
        reparsed = minidom.parseString(rough_string)
        return reparsed.toprettyxml(indent="  ")
    except ET.ParseError:
        # Fallback to lxml if standard parser fails (for really messy XML)
        # tree = html.fromstring(xml_string)
        # return html.tostring(tree, pretty_print=True, encoding='utf-8').decode('utf-8')
        print("Standard XML parser failed. Consider using lxml for more tolerance.")
        return None

3.2 Wrap It in a Splunk Modular Input #

Next, integrate this logic into a Splunk modular input script. This script will handle Splunk’s input configuration, fetch the raw XML, process it, and write the cleaned events to Splunk.

import sys
import splunklib.modularinput as smi

class XMLCleanerInput(smi.Script):
    def get_scheme(self):
        # Define your input's configuration options
        scheme = smi.Scheme("Messy XML Cleaner")
        scheme.description = "Cleans and formats invalid/messy XML for Splunk ingestion"
        
        # Add a parameter for where to fetch the XML (file path, API URL, etc.)
        scheme.add_argument(
            smi.Argument(
                name="xml_source",
                description="Path to XML file or API endpoint to retrieve messy XML",
                required_on_create=True
            )
        )
        return scheme

    def stream_events(self, inputs, ew):
        # Process each input stanza configured in Splunk
        for input_name, input_config in inputs.items():
            xml_source = input_config["xml_source"]
            
            # Fetch raw XML (adjust this based on your source: file, API, etc.)
            try:
                if xml_source.startswith("http"):
                    # Handle API call example
                    import requests
                    response = requests.get(xml_source)
                    raw_xml = response.text
                else:
                    # Handle file input
                    with open(xml_source, 'r') as f:
                        raw_xml = f.read()
            except Exception as e:
                ew.log(smi.LogLevel.ERROR, f"Failed to retrieve XML from {xml_source}: {str(e)}")
                continue
            
            # Clean and format the XML
            cleaned_xml = scrub_invalid_xml(raw_xml)
            if not cleaned_xml:
                ew.log(smi.LogLevel.WARN, f"No valid XML left after scrubbing for input {input_name}")
                continue
            
            formatted_xml = pretty_print_xml(cleaned_xml)
            if not formatted_xml:
                ew.log(smi.LogLevel.ERROR, f"Failed to format XML for input {input_name}")
                continue
            
            # Write the cleaned XML as a Splunk event
            event = smi.Event()
            event.data = formatted_xml
            event.stanza = input_name
            ew.write_event(event)

if __name__ == "__main__":
    sys.exit(XMLCleanerInput().run(sys.argv))

• Test Locally First: Run your cleaning functions on sample messy XML before integrating with Splunk to iron out edge cases.
• Error Logging: The ew.log() calls are crucial for debugging—check Splunk’s internal logs ($SPLUNK_HOME/var/log/splunk) if your input isn’t working.
• Package as a Splunk App: To deploy this across your Splunk environment, package the script, inputs.conf, and default.meta into a Splunk app. This makes configuration and distribution much easier.
• Lxml Installation: If you use lxml, install it in Splunk’s Python environment (usually $SPLUNK_HOME/bin/python -m pip install lxml) or include it in your app’s bin directory if deploying to air-gapped systems.

内容的提问来源于stack exchange，提问作者sai