If you manage more than one or two home Linux servers, you already know the routine. Every morning, logwatch wakes up, collects a pile of system data, and drops a flat text file into your inbox. If you have three, four, or five machines running, checking those logs turns into an absolute chore. Half the time you just scan the headers, notice nothing looks on fire, and close the window.
That’s bad security hygiene. But let’s be honest—reading regular operational noise day in and day out is mind-numbing.
So I started thinking. The future happened while we weren’t looking, and now we have incredibly capable local Large Language Models (LLMs) that can run completely offline right on our own hardware. Could I offload the boring task of daily log parsing to an AI running on my home Intel NUC server? Could it act like a virtual, around-the-clock Ubuntu System Administrator, reading the noise and only buzzing my phone when an actual infrastructure anomaly hits?
Yes, yes it can. Partnering with Google AI, which did all the heavy lifting for the coding, I mapped out a complete multi-server extraction framework.
The best part? It costs zero dollars in subscription APIs, and not a single line of my server log data ever leaves my local network. Here is how I made that happen.
The Architectural Philosophy: Why Pull vs. Push?
When designing a multi-node tracking setup, you have two real choices for collecting files: you can have your remote servers push their data to your central node, or have your central node pull the data from them.
I explicitly chose a Single-Pass Pull Architecture.
If a remote server gets compromised, the very first thing an attacker does is look for API tokens or SSH keys to pivot across your network. By implementing a pull architecture, my remote nodes remain completely passive. They do not hold any connection keys or write permissions to my central NUC. The central monitoring node initiates a hardened outbound connection, pulls the data into an isolated staging queue, processes it, and clears the remote space. If a sub-node drops offline or suffers a breach, the rest of the infrastructure remains entirely safe.
The Vision: Section Ingestion Over Raw Chunking
My initial prototype was pretty naive. I just took the flat log text files, combined them, and shoved them at a tiny 1B parameter model. It was fast, but it had a massive problem: it was dumb. It kept hallucinating standard initialization wrappers—like Processing Initiated or basic logrotate notifications—as active security threats. I kept getting woken up by “actionable alerts” that were just my servers doing normal housekeeping.
Worse, when one of my remote nodes generated a massive log file, the raw character chunking cut the file right in half. The model panicked, locked up my NUC’s CPU for 30 minutes, and threw a nasty HTTP 500 error.
To fix this, I had to completely re-architect how the logs are read. Instead of treating a log file like one giant, flat block of text, the script now dynamically parses the report into isolated structural service sections along standard markdown boundaries (e.g., --- SSHD CONNECTIONS ---).
This structural approach changed everything:
1. Benign Section Skipping: We can completely drop sections we don’t care about—like a massive list of successful cron jobs or routine log rotations—before they ever hit the LLM.
2. Context Preservation: The AI reads a complete narrative block for a service. It can easily tell the difference between a standard automated retry and an active brute-force intrusion.
3. CPU-Friendly Handshakes: Processing small, focused sections sequentially drops memory utilization into a safe zone, meaning we can upgrade to a much heavier, smarter model.
Upgrading the Brains: Enter Qwen 2.5 Coder
Once the log context size was small and predictable, I dropped the 1B model and pulled down qwen2.5-coder:7b. Even though it’s technically a coding model, its understanding of Linux command-line syntax, raw tracing logs, and structural JSON schemas is absolutely phenomenal for system administration.
Running a 7 Billion parameter model on an Intel i5 NUC CPU is a heavy lift, but because this is an overnight batch script, I don’t care if it takes 15 to 20 seconds per section. To optimize it, I locked down Ollama’s execution parameters to use exactly 4 physical cores, preventing hyperthreading from thrashing the CPU cache. The result? Smooth, reliable local inference at around 2 tokens per second.
Phase 1: Preparing the Infrastructure Environment
Before diving into the Python processing files, we need to prepare our target machines, install our local execution dependencies, and get the LLM runtime live.
1. Configuring Logwatch to Drop Local Staging Files (The Ubuntu Way)
On modern Ubuntu systems, changing /etc/logwatch/conf/logwatch.conf isn’t enough to stop the daily email reports because Ubuntu handles the default execution sequence via an automated cron hook. To override this behavior and force it to drop a clean, flat text file directly into a local staging directory instead, we have to modify the execution script directly.
Open up your daily cron wrapper:
sudo nano /etc/cron.daily/00logwatchLocate the line that triggers the final execution binary and update the layout arguments to intercept the output before it mails it away:
# Force Logwatch to write straight to a flat file layout instead of mailing it out
/usr/sbin/logwatch --output file --format text --filename /var/log/logwatch_staging/system_report.log2. Installing Ollama and the 7B Model
On your central monitoring node (the Intel NUC), install the local Ollama processing service using the standard deployment string:
curl -fsSL https://ollama.com | shOnce the backend engine is running, pull down qwen2.5-coder:7b:
ollama pull qwen2.5-coder:7b3. Setting Up Your Python Prerequisites
To keep your global system libraries pristine, isolate the Python pipeline inside a dedicated virtual environment. Install requests for handling the local API handshakes and pydantic to enforce structured JSON output boundaries:
# Navigate to your processing workspace directory
cd /path/to/log_processor/
# Generate and activate the virtual environment
python3 -m venv .venv
source .venv/bin/activate
# Install core parsing dependencies
pip install requests pydanticPhase 2: The Core Log Processor Code Base
I broke the Python parsing pipeline into a clean, modular structure. This keeps the configuration layout completely separate from the underlying operational logic.
File 1: The Configuration Engine (config.py)
#!/usr/bin/env python3
import os
# Feature Toggle Routing Flags
ENABLE_NTFY = True
ENABLE_CLEAN_NOTIFICATIONS = True
ENABLE_EMAIL = True
# Ollama Server Configurations
OLLAMA_API_URL = "http://localhost:11434/api/chat"
MODEL_NAME = "qwen2.5-coder:7b"
OLLAMA_TIMEOUT_SECONDS = 180
OLLAMA_NUM_CTX = 4096
# Communication Endpoints & Routing Mappings
EMAIL_DESTINATION = "your_admin_email@domain.com"
NTFY_URL = "https://ntfy.sh"
NTFY_TOPIC = "your_private_topic_here"
# Absolute System Path Layout Boundaries
MAIN_LOG_PATH = "/var/log/reviewlogwatch.log"
DEBUG_DUMP_DIR = "/path/to/log_processor/debug_dump"
# Safe Processing Optimization Parameters
SECTION_CHARACTER_LIMIT = 6000
# Senior Ubuntu SRE Analysis Prompt
SYSTEM_PROMPT = (
"You are an expert Senior Ubuntu Linux Site Reliability Engineer (SRE). "
"Analyze the logs strictly for critical infrastructure threats, brute force, "
"kernel panics, hardware faults, or unexpected service crashes. "
"Ignore normal status messages, routine cron jobs, and tool execution headers. "
"Be conservative: standard log rotation metadata and system uptime metrics "
"are NOT anomalies. If no real threat or critical fault exists, you MUST set "
"'is_anomaly' to false and return an empty 'anomalies' list. "
"Output strictly valid JSON matching the schema."
)
# Case-Insensitive Ingestion Filtering Rule Arrays
EXCLUDED_STRINGS = [
"allowusers",
"not in allowusers",
"mounted filesystem with ordered data mode",
"logrotate: alert",
"modification time rule skipped",
"processing initiated",
"detail level of output",
"type of output/format",
"operating system:",
"logwatch end"
]
# Logwatch Section Exclusions (Dropped before LLM processing network handshakes)
EXCLUDED_SECTIONS = [
"LOGROTATE STATUS",
"LOGWATCH STATUS",
"DISK SPACE",
"CRON"
]File 2: The Structured Schema Layer (models.py)
#!/usr/bin/env python3
from typing import List
from pydantic import BaseModel, Field
class LogAnomaly(BaseModel):
timestamp: str = Field(description="The timestamp or date string extracted from the log line.")
service: str = Field(description="The system daemon or process name (e.g., sshd, kernel, systemd).")
message: str = Field(description="The critical core message explaining the fault or warning.")
class AnalysisResult(BaseModel):
hostname: str = Field(description="The target human-readable server name in uppercase.")
is_anomaly: bool = Field(description="True if real actionable issues exist, False if baseline normal noise.")
anomalies: List[LogAnomaly] = Field(default_factory=list, description="List of validated anomalies found.")File 3: The Communication Channel Interface (alerts.py)
#!/usr/bin/env python3
import logging
import json
import subprocess
import requests
from typing import List, Optional
from config import EMAIL_DESTINATION, NTFY_URL, NTFY_TOPIC
def send_alert_email(report_body: str):
"""
Dispatches severe summary report directly via local mail binary.
"""
subject = "Local Monitoring Agent Alert: Actionable Anomalies Detected"
try:
process = subprocess.Popen(
["mail", "-s", subject, EMAIL_DESTINATION],
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
)
stdout, stderr = process.communicate(input=report_body)
if process.returncode == 0:
logging.info(f"Alert email successfully dispatched to {EMAIL_DESTINATION}")
else:
logging.error(f"Mail binary failed with return code {process.returncode}: {stderr}")
except Exception as e:
logging.error(f"Failed to execute local mail subsystem: {str(e)}")
def send_ntfy_alert(title: str, message: str, priority: str = "default", tags: Optional[List[str]] = None):
"""
Pushes an immediate endpoint alert payload to the centralized ntfy instance topic.
"""
headers = {
"Content-Type": "application/json"
}
payload = {
"topic": NTFY_TOPIC,
"title": title,
"message": message,
"priority": 3 if priority == "default" else (4 if priority == "high" else 5),
"tags": tags if tags else []
}
try:
res = requests.post(NTFY_URL, json=payload, headers=headers, timeout=15)
if res.status_code == 200:
logging.info("Ntfy notification successfully pushed via structured JSON with emoji tags.")
else:
logging.error(f"Ntfy API dropped notification with error code {res.status_code}: {res.text}")
except Exception as e:
logging.error(f"Failed to transmit HTTP request payload to ntfy server: {str(e)}")File 4: The Section Slicing and LLM Processing Engine (core.py)
#!/usr/bin/env python3
import os
import json
import logging
import requests
from typing import Optional, Tuple, Dict, Any, List
from config import (
OLLAMA_API_URL, MODEL_NAME, EXCLUDED_STRINGS, DEBUG_DUMP_DIR,
SYSTEM_PROMPT, OLLAMA_TIMEOUT_SECONDS, EXCLUDED_SECTIONS,
SECTION_CHARACTER_LIMIT, OLLAMA_NUM_CTX
)
from models import AnalysisResult, LogAnomaly
def dump_problematic_log(hostname: str, section: str, content: str, reason: str):
"""Saves a copy of the raw processed section buffer to a debug directory for offline triage."""
try:
os.makedirs(DEBUG_DUMP_DIR, exist_ok=True)
safe_section = section.lower().replace(" ", "_")
dump_path = os.path.join(DEBUG_DUMP_DIR, f"{hostname.lower()}_{safe_section}_failed.log")
with open(dump_path, "w", encoding="utf-8") as f:
f.write(f"# DEBUG REASON: {reason}\n")
f.write(content)
logging.warning(f" -> Preserved problematic section for offline triage at: {dump_path}")
except Exception as e:
logging.error(f"Failed to write debug log dump file: {str(e)}")
def parse_to_sections(cleaned_lines: List[str]) -> Dict[str, str]:
"""
Parses a flat sequence of log lines into structural sections
demarcated by Logwatch markdown boundaries.
"""
sections_map = {}
current_section = "SYSTEM METADATA"
sections_map[current_section] = []
for line in cleaned_lines:
stripped_line = line.strip()
if (stripped_line.startswith("---") and stripped_line.endswith("---")) or \
(stripped_line.startswith("###") and stripped_line.endswith("###")):
current_section = stripped_line.strip("-# ").upper()
sections_map[current_section] = []
continue
sections_map[current_section].append(line)
return {k: "\n".join(v) for k, v in sections_map.items() if v and any(line.strip() for line in v)}
def process_log_file(filepath: str) -> Optional[Tuple[AnalysisResult, int, Dict[str, Any]]]:
"""
Ingests and slices a staging log file by structural section.
Evaluates individual sections sequentially via local SRE LLM to optimize memory bounds.
"""
filename = os.path.basename(filepath)
base_part, _ = os.path.splitext(filename)
base_name = base_part.upper()
logging.info(f"Ingesting structural staging target file: {filename} as server: {base_name}")
cleaned_lines = []
try:
with open(filepath, "r", encoding="utf-8") as f:
for line in f:
line_stripped = line.strip()
if not line_stripped:
continue
line_lower = line_stripped.lower()
if any(exclusion in line_lower for exclusion in EXCLUDED_STRINGS):
continue
cleaned_lines.append(line_stripped)
except Exception as e:
logging.error(f"Failed to read file {filepath}: {str(e)}")
return None
total_lines_read = len(cleaned_lines)
if not cleaned_lines:
logging.info(f"Ingestion queue clear for {base_name}. No lines to process.")
return AnalysisResult(hostname=base_name, is_anomaly=False, anomalies=[]), 0, {"token_count": 0, "tokens_per_sec": 0.0}
raw_sections = parse_to_sections(cleaned_lines)
aggregated_anomalies: List[LogAnomaly] = []
total_tokens = 0
tps_samples = []
for section_name, section_content in raw_sections.items():
if section_name in EXCLUDED_SECTIONS:
logging.info(f" -> Skipping excluded benign section block: [{section_name}]")
continue
if len(section_content) > SECTION_CHARACTER_LIMIT:
logging.warning(f"Section [{section_name}] on {base_name} overflows limits. Slicing buffer.")
section_content = "[SYSTEM NOTICE: Log Section Truncated]\n" + section_content[-SECTION_CHARACTER_LIMIT:]
logging.info(f" -> Evaluating section: [{section_name}] ({len(section_content.splitlines())} lines)...")
payload = {
"model": MODEL_NAME,
"messages": [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Analyze these [{section_name}] logs for server {base_name}:\n{section_content}"}
],
"options": {
"temperature": 0.0,
"num_ctx": OLLAMA_NUM_CTX
},
"format": AnalysisResult.model_json_schema(),
"stream": False
}
try:
response = requests.post(OLLAMA_API_URL, json=payload, timeout=OLLAMA_TIMEOUT_SECONDS)
response.raise_for_status()
response_data = response.json()
llm_output_text = response_data.get("message", {}).get("content", "{}").strip()
eval_count = response_data.get("eval_count", 0)
eval_duration_ns = response_data.get("eval_duration", 0)
total_tokens += eval_count
if eval_count > 0 and eval_duration_ns > 0:
tps_samples.append(eval_count / (eval_duration_ns / 1_000_000_000.0))
if llm_output_text.startswith("```"):
llm_output_text = llm_output_text.strip("`").replace("json", "", 1).strip()
section_json = json.loads(llm_output_text)
if section_json.get("is_anomaly") and "anomalies" in section_json:
for anomaly_data in section_json["anomalies"]:
if not anomaly_data.get("service"):
anomaly_data["service"] = section_name
aggregated_anomalies.append(LogAnomaly(**anomaly_data))
except requests.exceptions.RequestException as e:
error_msg = f"Ollama connection failure on section [{section_name}]: {str(e)}"
logging.error(error_msg)
dump_problematic_log(base_name, section_name, section_content, error_msg)
except (json.JSONDecodeError, TypeError) as e:
error_msg = f"Structural validation dropped on section [{section_name}]: {str(e)}"
logging.error(error_msg)
dump_problematic_log(base_name, section_name, section_content, error_msg)
pipeline_result = AnalysisResult(
hostname=base_name,
is_anomaly=len(aggregated_anomalies) > 0,
anomalies=aggregated_anomalies
)
final_perf = {
"token_count": total_tokens,
"tokens_per_sec": round(sum(tps_samples) / len(tps_samples), 2) if tps_samples else 0.0
}
return pipeline_result, total_lines_read, final_perf
File 5: The Main Execution Coordinator (parse_logs.py)
#!/usr/bin/env python3
import os
import sys
import logging
from config import (
MAIN_LOG_PATH, ENABLE_NTFY, ENABLE_CLEAN_NOTIFICATIONS, ENABLE_EMAIL
)
from alerts import send_alert_email, send_ntfy_alert
from core import process_log_file
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[
logging.FileHandler(MAIN_LOG_PATH, encoding='utf-8'),
logging.StreamHandler(sys.stdout)
],
force=True
)
if __name__ == "__main__":
logging.info("=== Starting Daily Log Analysis Execution Cycle ===")
if len(sys.argv) < 2:
logging.error("No input files provided to runtime script loop.")
if ENABLE_NTFY:
send_ntfy_alert(
title="Local Monitoring Agent: Processing Problem",
message="Script halted: Execution context received an empty array of target staging input logs.",
priority="high",
tags=["warning"]
)
sys.exit(1)
staging_files = sys.argv[1:]
pipeline_anomalies_detected = False
processing_faults_encountered = False
aggregated_report = "The local monitoring agent detected actionable anomalies:\n"
processed_hosts = []
host_line_metrics = []
total_pipeline_lines = 0
total_pipeline_tokens = 0
host_perf_metrics = []
for file_path in staging_files:
if not os.path.exists(file_path):
logging.warning(f"Target staging file path missing: {file_path}")
processing_faults_encountered = True
continue
result_bundle = process_log_file(file_path)
if result_bundle is None:
processing_faults_encountered = True
continue
analysis, lines_parsed, perf = result_bundle
processed_hosts.append(analysis.hostname)
host_line_metrics.append(f"{analysis.hostname}: {lines_parsed} lines")
total_pipeline_lines += lines_parsed
total_pipeline_tokens += perf["token_count"]
if perf["token_count"] > 0:
host_perf_metrics.append(f"{analysis.hostname}: {perf['tokens_per_sec']} t/s")
if analysis.is_anomaly and analysis.anomalies:
pipeline_anomalies_detected = True
aggregated_report += f"\nServer Host Node [{analysis.hostname}]:\n"
for anomaly in analysis.anomalies:
aggregated_report += f"* {anomaly.timestamp} - {anomaly.service}: {anomaly.message}\n"
print(f"* [{analysis.hostname}] {anomaly.service}: {anomaly.message}", flush=True)
perf_summary = f"Generated {total_pipeline_tokens} tokens total ({', '.join(host_perf_metrics) if host_perf_metrics else '0 t/s'})."
metrics_summary = f"Processed {total_pipeline_lines} lines total ({', '.join(host_line_metrics)}). {perf_summary}"
logging.info(f"Pipeline execution performance details: {metrics_summary}")
notification_tags = [host.lower() for host in processed_hosts]
if processing_faults_encountered:
logging.warning("Execution run completed with operational errors.")
if ENABLE_NTFY:
send_ntfy_alert(
title="Local Monitoring Agent: Processing Problem",
message=f"Pipeline encountered errors, Ollama timeouts, or bad JSON. {metrics_summary}",
priority="high",
tags=["warning"] + notification_tags
)
elif pipeline_anomalies_detected:
logging.info("Anomalies verified. Triggering notification arrays.")
if ENABLE_EMAIL:
send_alert_email(aggregated_report)
if ENABLE_NTFY:
send_ntfy_alert(
title="Local Monitoring Agent Alert: Actionable Anomalies Detected",
message=f"Actionable system anomalies extracted. Detailed telemetry sent. {metrics_summary}",
priority="urgent",
tags=["warning"] + notification_tags
)
else:
logging.info("Inference analysis clear: Baseline normal.")
if ENABLE_NTFY and ENABLE_CLEAN_NOTIFICATIONS:
send_ntfy_alert(
title="Local Monitoring Agent: Clean Run Success",
message=f"All processed infrastructure nodes are running clean. {metrics_summary}",
priority="default",
tags=["+1"] + notification_tags
)
logging.info("=== Execution Cycle Completed Cleanly ===")
sys.stdout.flush()
sys.stderr.flush()Phase 3: Ingestion Orchestration and Cron Automation
To connect our passive remote servers to our central Python logic, we use a defensive Bash orchestration script (fetch_and_parse.sh). This script executes on the central NUC node, logging in via secure SSH default key pairs to pull remote staging reports.
The Bash Orchestrator (/path/to/log_processor/fetch_and_parse.sh)
#!/usr/bin/env bash
# ==============================================================================
# Centralized Log Ingestion Orchestrator (Single-Pass Pull Architecture)
# Co-authored & Engineered in partnership with Google AI
# ==============================================================================
set -o pipefail
shopt -s nullglob
# ------------------------------------------------------------------------------
# GLOBAL SECURITY & ENVIRONMENT CONFIGURATION
# ------------------------------------------------------------------------------
SSH_USER="root"
SSH_PORT="3022"
# Remote path where the modified Ubuntu cron.daily/00logwatch script drops reports
REMOTE_STAGING_PATH="/var/log/logwatch_staging/system_report.log"
# Central monitoring node folder infrastructure
LOCAL_STAGING_DIR="/var/log/logwatch_staging"
VENV_PYTHON="/path/to/log_processor/.venv/bin/python3"
SCRIPT_PATH="/path/to/log_processor/parse_logs.py"
# Array of remote servers to poll (space-separated list of addresses or hostnames)
REMOTE_NODES=(
"://domain.com"
"://domain.com"
)
# ------------------------------------------------------------------------------
# INGESTION RUNTIME ENGINE
# ------------------------------------------------------------------------------
logging_info() {
echo "$(date '+%Y-%m-%d %H:%M:%S') [INFO] $1"
}
# 1. Securely pull remote logwatch data blocks
for node in "${REMOTE_NODES[@]}"; do
logging_info("Initiating single-pass pull ingestion loop for node: [${node}]")
# Execute secure download mapping remote file to local node-named storage
scp -P "${SSH_PORT}" "${SSH_USER}@${node}:${REMOTE_STAGING_PATH}" "${LOCAL_STAGING_DIR}/${node}.log" 2>/dev/null
SCP_STATUS=$?
if [ ${SCP_STATUS} -eq 0 ]; then
logging_info(" -> Ingestion successful. Executing atomic clear on remote target [${node}]...")
# Purge remote staging cache to prevent double-processing on next run
ssh -p "${SSH_PORT}" "${SSH_USER}@${node}" "rm -f ${REMOTE_STAGING_PATH}"
else
logging_info(" -> No report found or node unreachable for [${node}]. Skipping.")
fi
done
# 2. Compile staging queue arrays and execute local inference engine
STAGING_FILES=("${LOCAL_STAGING_DIR}"/*.log)
if [ ${#STAGING_FILES[@]} -gt 0 ]; then
logging_info("Staging queue active. Passing ${#STAGING_FILES[@]} files to parsing pipeline...")
# Run the modular Python orchestrator across the collected files array
$VENV_PYTHON $SCRIPT_PATH "${STAGING_FILES[@]}"
EXIT_CODE=$?
# On a clean processing verdict, obliterate the local staging files cleanly
if [ ${EXIT_CODE} -eq 0 ]; then
logging_info("Pipeline cycle completed successfully. Cleardown staging area.")
rm -f "${LOCAL_STAGING_DIR}"/*.log
else
logging_info("Pipeline exited with code ${EXIT_CODE}. Retaining log files for triage.")
fi
else
logging_info("Ingestion queue clear. No processed log logs found.")
fiAutomating with System Cron
To execute this analysis sequence silently in the background at 4:30 AM every morning, update your system crontab file using crontab -e:
30 04 * * * /path/to/log_processor/fetch_and_parse.sh > /dev/null 2>&1Elegant Notifications
The final piece of the puzzle was the notification state machine. I didn’t want messy text notifications. I configured the alerts.py module to target the ntfy server payload with explicit shortcode array tags.
Now, when the cron job completes cleanly, my phone buzzes with a clean thumbs-up emoji (+1) and lowercase server name badges detailing exactly how many lines and tokens per second were processed. If something goes wrong, or a service actually crashes, it fires an urgent warning badge (warning), dumps a raw copy of the offending log segment to a local directory for safe keeping, and drops a comprehensive telemetry breakdown into my email box.
Anyway, cool. It’s up, it’s running, and it works perfectly.
If you are running local hardware, stop sending your system traces to commercial cloud endpoints. With a little Python engineering, a solid open-source 7B model, and some automation scripting, you can run your own completely private monitoring workspace right from your server closet.
Leave a Reply
You must be logged in to post a comment.