ship-logs 0.1.0

A pluggable log shipper and alerting agent.

Markdown

Log Shipper & Alerting Agent

A highly pluggable, pattern-driven logging agent built in Python that dynamically routes log events to multiple third-party providers (like Slack, Email, and more) based on log severity thresholds.

This project serves as a practical blueprint for implementing clean, decoupled software architectures using classic Gang of Four (GoF) design patterns inside a modern, production-ready Python layout.

---

Architecture & Design Patterns

The core philosophy of this tool is Open-Closed Principle: the engine shouldn't know how notifications are sent, and individual alerting channels shouldn't know when or why a log event occurs.

[ Application Core ] │ ▼ (Triggers log events) [ LogShipEngine ] ──(Broadcasts)──► [ FilteringAlertObserver ] │ │ │ (Dynamic registration) ▼ (Matches Level?) ├─► SlackAlertObserver ─────────────► Calls SlackStrategy └─► EmailAlertObserver ─────────────► Calls EmailStrategy

Applied Patterns

• Strategy Pattern: Encapsulates the network payload execution details for each provider (SlackNotificationStrategy, EmailNotificationStrategy) behind a uniform NotificationStrategy abstract base interface.
• Observer Pattern: Implements pub-sub routing logic. The LogShipperEngine acts as the Subject, passing all logs down to registered LogObserver instances, which filter entries before executing notification routines.
• Factory Method Pattern: Isolates creation logic inside NotificationFactory. It parses raw dictionary runtime profiles into functional code blocks, abstracting implementation dependencies away from client endpoints.
• Singleton Pattern: Employs a custom SingletonMeta metaclass configured with a thread-safe execution lock (threading.Lock) to manage global configuration states safely across concurrent processing environments.

---

Getting Started

This repository leverages uv, an incredibly fast Python package installer and workspace manager.

Prerequisites

Make sure you have uv installed on your local environment:

pip install uv

Installation

1. Clone this repository to your computer:

git clone https://github.com/khansohil530/ship-logs.git
cd ship-logs

2. Sync the environment lockfile and compile requirements:

uv sync

Running the Test Suite

The codebase is built strictly around Test-Driven Development (TDD) principles. Run the unit test suite to verify your configuration structure matches:

uv run pytest

Usage Example

1. Define a Configuration File: Create a profile containing an alerts declaration block. You can mix and match log filters across target endpoints. Save this file as config.json:

{
  "alerts": [
    {
      "type": "slack",
      "level": "CRITICAL",
      "webhook_url": "[https://hooks.slack.com/services/mock-endpoint-string](https://hooks.slack.com/services/mock-endpoint-string)"
    },
    {
      "type": "email",
      "level": "ERROR",
      "smtp": {
        "host": "smtp.internal.net",
        "port": 587,
        "sender": "[e-mail]"
      }
    }
  ]
}

2. "Execute via the CLI Engine":Trigger the shipper instantly from your terminal environment using the registered executable command:

uv run ship-logs --config config.json --level CRITICAL --message "Production database partition out of memory space!"

How to Extend (Add a New Provider)

Adding a brand new delivery provider (like Discord or PagerDuty) requires zero modifications to the core logging pipeline engine.

1: Write a New Strategy Subclass

Open src/ship_logs/strategies.py and extend NotificationStrategy:

class DiscordNotificationStrategy(NotificationStrategy):
    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url

    def send(self, message: str, context: dict) -> bool:
        # Custom HTTP execution logic for Discord payloads goes here...
        print(f"Dispatched to Discord: {message}")
        return True

2: Register in the Factory

Update NotificationFactory.create_strategy inside src/ship_logs/factory.py to recognize the entry mapping string:

elif normalized_type == "discord":
    return DiscordNotificationStrategy(webhook_url=config["webhook_url"])

Now, anyone can use your updated strategy simply by specifying "type": "discord" inside their config.json profile!