sendlog

Plugin DSL

See Plugin Architecture prior to reading this section.

Plugin Modules

At the highest level, all plugins are Python modules.

There are two types of plugin module:

Plugin modules should be placed in their respective parent directories within the root plugins directory. For example:

sendlog # repository root
└── sendlog
    └── plugins # root plugins directory
        ├── channels # channel plugin modules go in here
        │   ├── file.py
        │   ├── smtp.py
        │   ├── stdout.py
        │   ├── telegram.py
        │   └── twilio_sms.py
        └── logs # log plugin modules go in here
            ├── auth.py
            ├── sendlog.py
            └── systemd.py

While modules provide a layer of convenience when managing plugins, the plugin system does not recognise or enforce them beyond the import constraints shown above.

Plugin Class Basics

Plugin modules contain one or more plugin class. These classes represent nodes on a worktree: processes that convert a log to a payload or send it to an endpoint.

Generally speaking, plugin classes are created by extending the relevant superclass and implementing the __call__ method. If plugins do not inherit from the expected base, they will be rejected.

Plugin superclasses can be imported from the plugin module. However, plugin superclasses will vary depending on which module you are writing for.

Channel Module

Channel modules should only define one class: that extending the Channel superclass, which defines how an alert gets to an endpoint.

Channel Class

All custom channel classes must extend the Channel superclass:

class Channel(ABC, metaclass=_ChannelMeta):
    __slots__ = ["name"]
    def __init__(self, name, **kwargs):
        self.name = name

        for key, value in kwargs.items():
            setattr(self, key, value)

    @abstractmethod
    def __call__(self, msg):
        pass

Explanation:

Constraints:

Example

Below is an example of a Channel plugin module, which contains a Channel plugin class:

from plugin import Channel # import Channel superclass

import requests

class Telegram(Channel): # must inherit from Channel superclass
    __slots__ = ["chat_id", "token"] # Channel variables

    # entrypoint method
    def __call__(self, msg): # must take an additional parameter
        url = f"https://api.telegram.org/bot{self.token}/sendMessage"
        params = {"chat_id": self.chat_id, "text": msg}
        requests.post(url, params=params,)

Explanation

Log Module

Log modules define how logs are matched, filtered and transformed by extending the LogType, Rule, and Transformer superclasses.

The DSL allows you to create modular, hierarchical definitions that map directly to components of log processing:

The hierarchical constraints of these components are maintained by nesting:

from plugin import LogType, Rule, Transformer

class MyLogType(LogType):                 # Top-level log source plugin

    class MyRule(Rule):                   # A specific type of event

        class MyTransformer(Transformer): # Transforms matched event into payload
            pass

Currently, sendlog only permits workflows that consist of a LogType, Rule, and Transformer. If we think of the structure as a tree:

Like all plugins, the __call__ method defines how the class performs its’ function. The output (return value) of one plugin is chained to the subnodes in the tree. However, some of these plugins have default implementations that can simplify the plugin definition.

LogType Class

The LogType superclass defines how logs are matched to a generic structure. It is best used to extract common fields from semi-structured log sources.

class LogType(ABC, metaclass=_LogTypeMeta):
    """
    Base class for a LogType plugin.
    """

    regex = None

    def __call__(self, log_line: str):
        """
        Convert log_line to structured JSON using regex.
        
        Must return a dictionary with the "message" key.
        """
        if self.__class__.regex:
            match = re.match(self.__class__.regex, log_line)
            if match:
                log_parts = match.groupdict()
                return log_parts
            else:
                raise TypeError("Log line did not match the expected format.")
        else:
            return {"message": log_line}

Explanation:

Constraints:

Rule Class

Rule plugins are similar to LogType plugins, but they do not raise errors when the message doesn’t match. They extend the Rule superclass:

class Rule(ABC, metaclass=_RuleMeta):

    regex = None
    
    def __call__(self, log_parts: dict):
        """Convert log message to structured JSON."""
        if self.__class__.regex:
            match = re.match(self.__class__.regex, log_parts["message"])
            if match:
                message_parts = match.groupdict()
                return {**log_parts, "context": {**message_parts}}
        return False

Explanation

Constraints:

Transformer Class

class Transformer(ABC, metaclass=_TransformerMeta):

    def __call__(self, log_parts: dict):
        return log_parts["message"]

Explanation

Constraints

Example

from plugin import LogType, Rule, Transformer
from datetime import datetime

class Pacman(LogType):
    regex = r"\[(?P<timestamp>.*?)\] \[(?P<application>.*?)\] (?P<message>.*)"

    class RunCommand(Rule):
        regex = r"Running\s+'(?P<command>[^']+)'"

        class HumanReadable(Transformer):
            def __call__(self, parts):
                context = parts["context"]
                timestamp = datetime.strptime(parts["timestamp"], "%Y-%m-%dT%H:%M:%S%z")
                return f"Command '{context["command"]}' detected at {timestamp.strftime("%Y-%m-%d %H:%M")}."
        
        class JSONL(Transformer):
            def __call__(self, parts):
                return parts

Explanation:

This site is open source. Improve this page.