Benjamin_Boenisch/breakpilot-pwa
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(infrastructure): Add night-scheduler for automated Docker service shutdown

Browse Source 
Implement dashboard-controlled night mode for automatic Docker service
management. Services are stopped at a configurable time (default 22:00)
and restarted in the morning (default 06:00).

Features:
- Python/FastAPI scheduler service (port 8096)
- Admin dashboard API routes at /api/admin/night-mode
- Toggle for enable/disable night mode
- Time picker for shutdown and startup times
- Manual start/stop buttons for immediate actions
- Excluded services (night-scheduler, nginx always run)

Files added:
- night-scheduler/scheduler.py - Main scheduler with REST API
- night-scheduler/Dockerfile - Container with Docker CLI
- night-scheduler/requirements.txt - FastAPI, Uvicorn, Pydantic
- night-scheduler/tests/test_scheduler.py - Unit tests
- admin-v2/app/api/admin/night-mode/* - API proxy routes
- .claude/rules/night-scheduler.md - Developer documentation

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
BreakPilot Dev
2026-02-08 22:45:03 -08:00
parent 83e32dc289
commit 3f7032260b
11 changed files with 3049 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
night-scheduler/Dockerfile Normal file
View File

@@ -0,0 +1,30 @@
FROM python:3.11-slim
# Docker CLI installieren (für docker compose Befehle)
RUN apt-get update && apt-get install -y \
curl \
gnupg \
lsb-release \
&& curl -fsSL https://download.docker.com/linux/debian/gpg | gpg --dearmor -o /usr/share/keyrings/docker-archive-keyring.gpg \
&& echo "deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/docker-archive-keyring.gpg] https://download.docker.com/linux/debian $(lsb_release -cs) stable" > /etc/apt/sources.list.d/docker.list \
&& apt-get update \
&& apt-get install -y docker-ce-cli docker-compose-plugin \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /app
# Python Dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Anwendung kopieren
COPY scheduler.py .
# Config-Verzeichnis
RUN mkdir -p /config
# Port für REST-API
EXPOSE 8096
# Start
CMD ["python", "scheduler.py"]

8
night-scheduler/config/night-mode.json Normal file
View File

@@ -0,0 +1,8 @@
{
"enabled": false,
"shutdown_time": "22:00",
"startup_time": "06:00",
"last_action": null,
"last_action_time": null,
"excluded_services": ["night-scheduler", "nginx"]
}

8
night-scheduler/requirements.txt Normal file
View File

@@ -0,0 +1,8 @@
fastapi==0.109.0
uvicorn[standard]==0.27.0
pydantic==2.5.3
# Testing
pytest==8.0.0
pytest-asyncio==0.23.0
httpx==0.26.0

389
night-scheduler/scheduler.py Normal file
View File

@@ -0,0 +1,389 @@
#!/usr/bin/env python3
"""
Night Scheduler - Leichtgewichtiger Scheduler für Nachtabschaltung
Prüft jede Minute die Konfiguration und führt docker compose up/down aus.
REST-API auf Port 8096 für Dashboard-Zugriff.
"""
import json
import os
import subprocess
import asyncio
from datetime import datetime, time, timedelta
from pathlib import Path
from typing import Optional
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
# Konfiguration
CONFIG_PATH = Path("/config/night-mode.json")
COMPOSE_FILE = Path("/app/docker-compose.yml")
COMPOSE_PROJECT = os.getenv("COMPOSE_PROJECT_NAME", "breakpilot-pwa")
# Services die NICHT gestoppt werden sollen
EXCLUDED_SERVICES = {"night-scheduler", "nginx"}
class NightModeConfig(BaseModel):
"""Konfiguration für den Nachtmodus"""
enabled: bool = False
shutdown_time: str = "22:00"
startup_time: str = "06:00"
last_action: Optional[str] = None # "shutdown" oder "startup"
last_action_time: Optional[str] = None
excluded_services: list[str] = Field(default_factory=lambda: list(EXCLUDED_SERVICES))
class NightModeStatus(BaseModel):
"""Status-Response für die API"""
config: NightModeConfig
current_time: str
next_action: Optional[str] = None # "shutdown" oder "startup"
next_action_time: Optional[str] = None
time_until_next_action: Optional[str] = None
services_status: dict[str, str] = Field(default_factory=dict)
class ExecuteRequest(BaseModel):
"""Request für sofortige Ausführung"""
action: str # "start" oder "stop"
def load_config() -> NightModeConfig:
"""Lädt die Konfiguration aus der JSON-Datei"""
if CONFIG_PATH.exists():
try:
with open(CONFIG_PATH) as f:
data = json.load(f)
return NightModeConfig(**data)
except (json.JSONDecodeError, Exception) as e:
print(f"Fehler beim Laden der Konfiguration: {e}")
return NightModeConfig()
def save_config(config: NightModeConfig) -> None:
"""Speichert die Konfiguration in die JSON-Datei"""
CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True)
with open(CONFIG_PATH, "w") as f:
json.dump(config.model_dump(), f, indent=2)
def parse_time(time_str: str) -> time:
"""Parst einen Zeit-String im Format HH:MM"""
parts = time_str.split(":")
return time(int(parts[0]), int(parts[1]))
def get_services_to_manage() -> list[str]:
"""Ermittelt alle Services, die verwaltet werden sollen"""
try:
result = subprocess.run(
["docker", "compose", "-f", str(COMPOSE_FILE), "config", "--services"],
capture_output=True,
text=True,
timeout=30
)
if result.returncode == 0:
services = result.stdout.strip().split("\n")
config = load_config()
excluded = set(config.excluded_services)
return [s for s in services if s and s not in excluded]
except Exception as e:
print(f"Fehler beim Ermitteln der Services: {e}")
return []
def get_services_status() -> dict[str, str]:
"""Ermittelt den Status aller Services"""
status = {}
try:
result = subprocess.run(
["docker", "compose", "-f", str(COMPOSE_FILE), "ps", "--format", "json"],
capture_output=True,
text=True,
timeout=30
)
if result.returncode == 0 and result.stdout.strip():
# Docker compose ps gibt JSON-Lines aus
for line in result.stdout.strip().split("\n"):
if line:
try:
container = json.loads(line)
service = container.get("Service", container.get("Name", "unknown"))
state = container.get("State", container.get("Status", "unknown"))
status[service] = state
except json.JSONDecodeError:
continue
except Exception as e:
print(f"Fehler beim Abrufen des Service-Status: {e}")
return status
def execute_docker_command(action: str) -> tuple[bool, str]:
"""
Führt docker compose Befehl aus.
Args:
action: "start" oder "stop"
Returns:
(success, message)
"""
services = get_services_to_manage()
if not services:
return False, "Keine Services zum Verwalten gefunden"
if action == "stop":
cmd = ["docker", "compose", "-f", str(COMPOSE_FILE), "stop"] + services
elif action == "start":
cmd = ["docker", "compose", "-f", str(COMPOSE_FILE), "start"] + services
else:
return False, f"Unbekannte Aktion: {action}"
try:
print(f"Führe aus: {' '.join(cmd)}")
result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
if result.returncode == 0:
return True, f"Aktion '{action}' erfolgreich ausgeführt für {len(services)} Services"
else:
return False, f"Fehler: {result.stderr}"
except subprocess.TimeoutExpired:
return False, "Timeout beim Ausführen des Befehls"
except Exception as e:
return False, f"Ausnahme: {str(e)}"
def calculate_next_action(config: NightModeConfig) -> tuple[Optional[str], Optional[datetime], Optional[timedelta]]:
"""
Berechnet die nächste Aktion basierend auf der aktuellen Zeit.
Returns:
(action, next_time, time_until)
"""
if not config.enabled:
return None, None, None
now = datetime.now()
today = now.date()
shutdown_time = parse_time(config.shutdown_time)
startup_time = parse_time(config.startup_time)
shutdown_dt = datetime.combine(today, shutdown_time)
startup_dt = datetime.combine(today, startup_time)
# Wenn startup vor shutdown ist, bedeutet das über Mitternacht
# z.B. shutdown 22:00, startup 06:00
if startup_time < shutdown_time:
# Wir sind in der Nacht
if now.time() < startup_time:
# Vor Startup-Zeit -> nächste Aktion ist Startup heute
return "startup", startup_dt, startup_dt - now
elif now.time() < shutdown_time:
# Zwischen Startup und Shutdown -> nächste Aktion ist Shutdown heute
return "shutdown", shutdown_dt, shutdown_dt - now
else:
# Nach Shutdown -> nächste Aktion ist Startup morgen
next_startup = startup_dt + timedelta(days=1)
return "startup", next_startup, next_startup - now
else:
# Startup nach Shutdown am selben Tag (ungewöhnlich, aber unterstützt)
if now.time() < shutdown_time:
return "shutdown", shutdown_dt, shutdown_dt - now
elif now.time() < startup_time:
return "startup", startup_dt, startup_dt - now
else:
next_shutdown = shutdown_dt + timedelta(days=1)
return "shutdown", next_shutdown, next_shutdown - now
def format_timedelta(td: timedelta) -> str:
"""Formatiert ein timedelta als lesbaren String"""
total_seconds = int(td.total_seconds())
hours, remainder = divmod(total_seconds, 3600)
minutes, _ = divmod(remainder, 60)
if hours > 0:
return f"{hours}h {minutes}min"
return f"{minutes}min"
async def scheduler_loop():
"""Haupt-Scheduler-Schleife, prüft jede Minute"""
print("Scheduler-Loop gestartet")
while True:
try:
config = load_config()
if config.enabled:
now = datetime.now()
current_time = now.time()
shutdown_time = parse_time(config.shutdown_time)
startup_time = parse_time(config.startup_time)
# Prüfe ob Shutdown-Zeit erreicht
if (current_time.hour == shutdown_time.hour and
current_time.minute == shutdown_time.minute):
if config.last_action != "shutdown" or (
config.last_action_time and
datetime.fromisoformat(config.last_action_time).date() < now.date()
):
print(f"Shutdown-Zeit erreicht: {config.shutdown_time}")
success, msg = execute_docker_command("stop")
print(f"Shutdown: {msg}")
config.last_action = "shutdown"
config.last_action_time = now.isoformat()
save_config(config)
# Prüfe ob Startup-Zeit erreicht
elif (current_time.hour == startup_time.hour and
current_time.minute == startup_time.minute):
if config.last_action != "startup" or (
config.last_action_time and
datetime.fromisoformat(config.last_action_time).date() < now.date()
):
print(f"Startup-Zeit erreicht: {config.startup_time}")
success, msg = execute_docker_command("start")
print(f"Startup: {msg}")
config.last_action = "startup"
config.last_action_time = now.isoformat()
save_config(config)
except Exception as e:
print(f"Fehler in Scheduler-Loop: {e}")
# Warte 60 Sekunden
await asyncio.sleep(60)
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Lifecycle-Manager für FastAPI"""
# Startup: Starte den Scheduler
task = asyncio.create_task(scheduler_loop())
yield
# Shutdown: Stoppe den Scheduler
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
# FastAPI App
app = FastAPI(
title="Night Scheduler API",
description="API für die Dashboard-gesteuerte Nachtabschaltung",
version="1.0.0",
lifespan=lifespan
)
# CORS für Admin-Dashboard
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/health")
async def health():
"""Health-Check-Endpoint"""
return {"status": "healthy", "service": "night-scheduler"}
@app.get("/api/night-mode", response_model=NightModeStatus)
async def get_status():
"""Gibt den aktuellen Status und die Konfiguration zurück"""
config = load_config()
now = datetime.now()
next_action, next_time, time_until = calculate_next_action(config)
return NightModeStatus(
config=config,
current_time=now.strftime("%H:%M:%S"),
next_action=next_action,
next_action_time=next_time.strftime("%H:%M") if next_time else None,
time_until_next_action=format_timedelta(time_until) if time_until else None,
services_status=get_services_status()
)
@app.post("/api/night-mode", response_model=NightModeConfig)
async def update_config(new_config: NightModeConfig):
"""Aktualisiert die Nachtmodus-Konfiguration"""
# Validiere Zeitformate
try:
parse_time(new_config.shutdown_time)
parse_time(new_config.startup_time)
except (ValueError, IndexError):
raise HTTPException(status_code=400, detail="Ungültiges Zeitformat. Erwartet: HH:MM")
# Behalte last_action bei, wenn nicht explizit gesetzt
if new_config.last_action is None:
old_config = load_config()
new_config.last_action = old_config.last_action
new_config.last_action_time = old_config.last_action_time
save_config(new_config)
return new_config
@app.post("/api/night-mode/execute")
async def execute_action(request: ExecuteRequest):
"""Führt eine Aktion sofort aus (start/stop)"""
if request.action not in ["start", "stop"]:
raise HTTPException(status_code=400, detail="Aktion muss 'start' oder 'stop' sein")
success, message = execute_docker_command(request.action)
if success:
# Aktualisiere last_action
config = load_config()
config.last_action = "startup" if request.action == "start" else "shutdown"
config.last_action_time = datetime.now().isoformat()
save_config(config)
return {"success": True, "message": message}
else:
raise HTTPException(status_code=500, detail=message)
@app.get("/api/night-mode/services")
async def get_services():
"""Gibt die Liste aller verwaltbaren Services zurück"""
return {
"all_services": get_services_to_manage(),
"excluded_services": list(load_config().excluded_services),
"status": get_services_status()
}
@app.get("/api/night-mode/logs")
async def get_logs():
"""Gibt die letzten Aktionen zurück"""
config = load_config()
return {
"last_action": config.last_action,
"last_action_time": config.last_action_time,
"enabled": config.enabled
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8096)

1
night-scheduler/tests/__init__.py Normal file
View File

@@ -0,0 +1 @@
# Night Scheduler Tests

342
night-scheduler/tests/test_scheduler.py Normal file
View File

@@ -0,0 +1,342 @@
"""
Tests für den Night Scheduler
Unit Tests für:
- Konfiguration laden/speichern
- Zeit-Parsing
- Nächste Aktion berechnen
- API Endpoints
"""
import json
import pytest
from datetime import datetime, time, timedelta
from pathlib import Path
from unittest.mock import patch, MagicMock
from fastapi.testclient import TestClient
# Importiere die zu testenden Funktionen
import sys
sys.path.insert(0, str(Path(__file__).parent.parent))
from scheduler import (
app,
NightModeConfig,
NightModeStatus,
parse_time,
calculate_next_action,
format_timedelta,
load_config,
save_config,
CONFIG_PATH,
)
# Test Client für FastAPI
client = TestClient(app)
class TestParseTime:
"""Tests für die parse_time Funktion"""
def test_parse_time_valid_morning(self):
"""Gültige Morgenzeit parsen"""
result = parse_time("06:00")
assert result.hour == 6
assert result.minute == 0
def test_parse_time_valid_evening(self):
"""Gültige Abendzeit parsen"""
result = parse_time("22:30")
assert result.hour == 22
assert result.minute == 30
def test_parse_time_midnight(self):
"""Mitternacht parsen"""
result = parse_time("00:00")
assert result.hour == 0
assert result.minute == 0
def test_parse_time_end_of_day(self):
"""23:59 parsen"""
result = parse_time("23:59")
assert result.hour == 23
assert result.minute == 59
class TestFormatTimedelta:
"""Tests für die format_timedelta Funktion"""
def test_format_hours_and_minutes(self):
"""Stunden und Minuten formatieren"""
td = timedelta(hours=4, minutes=23)
result = format_timedelta(td)
assert result == "4h 23min"
def test_format_only_minutes(self):
"""Nur Minuten formatieren"""
td = timedelta(minutes=45)
result = format_timedelta(td)
assert result == "45min"
def test_format_zero(self):
"""Null formatieren"""
td = timedelta(minutes=0)
result = format_timedelta(td)
assert result == "0min"
def test_format_many_hours(self):
"""Viele Stunden formatieren"""
td = timedelta(hours=15, minutes=30)
result = format_timedelta(td)
assert result == "15h 30min"
class TestCalculateNextAction:
"""Tests für die calculate_next_action Funktion"""
def test_disabled_returns_none(self):
"""Deaktivierter Modus gibt None zurück"""
config = NightModeConfig(enabled=False)
action, next_time, time_until = calculate_next_action(config)
assert action is None
assert next_time is None
assert time_until is None
def test_before_shutdown_time(self):
"""Vor Shutdown-Zeit: Nächste Aktion ist Shutdown"""
config = NightModeConfig(
enabled=True,
shutdown_time="22:00",
startup_time="06:00"
)
# Mock datetime.now() auf 18:00
with patch('scheduler.datetime') as mock_dt:
mock_now = datetime(2026, 2, 9, 18, 0, 0)
mock_dt.now.return_value = mock_now
mock_dt.combine = datetime.combine
action, next_time, time_until = calculate_next_action(config)
assert action == "shutdown"
assert next_time is not None
assert next_time.hour == 22
assert next_time.minute == 0
def test_after_shutdown_before_midnight(self):
"""Nach Shutdown, vor Mitternacht: Nächste Aktion ist Startup morgen"""
config = NightModeConfig(
enabled=True,
shutdown_time="22:00",
startup_time="06:00"
)
with patch('scheduler.datetime') as mock_dt:
mock_now = datetime(2026, 2, 9, 23, 0, 0)
mock_dt.now.return_value = mock_now
mock_dt.combine = datetime.combine
action, next_time, time_until = calculate_next_action(config)
assert action == "startup"
assert next_time is not None
# Startup sollte am nächsten Tag sein
assert next_time.day == 10
def test_early_morning_before_startup(self):
"""Früher Morgen vor Startup: Nächste Aktion ist Startup heute"""
config = NightModeConfig(
enabled=True,
shutdown_time="22:00",
startup_time="06:00"
)
with patch('scheduler.datetime') as mock_dt:
mock_now = datetime(2026, 2, 9, 4, 0, 0)
mock_dt.now.return_value = mock_now
mock_dt.combine = datetime.combine
action, next_time, time_until = calculate_next_action(config)
assert action == "startup"
assert next_time is not None
assert next_time.hour == 6
class TestNightModeConfig:
"""Tests für das NightModeConfig Model"""
def test_default_config(self):
"""Standard-Konfiguration erstellen"""
config = NightModeConfig()
assert config.enabled is False
assert config.shutdown_time == "22:00"
assert config.startup_time == "06:00"
assert config.last_action is None
assert "night-scheduler" in config.excluded_services
def test_config_with_values(self):
"""Konfiguration mit Werten erstellen"""
config = NightModeConfig(
enabled=True,
shutdown_time="23:00",
startup_time="07:30",
last_action="startup",
last_action_time="2026-02-09T07:30:00"
)
assert config.enabled is True
assert config.shutdown_time == "23:00"
assert config.startup_time == "07:30"
assert config.last_action == "startup"
class TestAPIEndpoints:
"""Tests für die API Endpoints"""
def test_health_endpoint(self):
"""Health Endpoint gibt Status zurück"""
response = client.get("/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "healthy"
assert data["service"] == "night-scheduler"
def test_get_status_endpoint(self):
"""GET /api/night-mode gibt Status zurück"""
with patch('scheduler.load_config') as mock_load:
mock_load.return_value = NightModeConfig()
response = client.get("/api/night-mode")
assert response.status_code == 200
data = response.json()
assert "config" in data
assert "current_time" in data
def test_update_config_endpoint(self):
"""POST /api/night-mode aktualisiert Konfiguration"""
with patch('scheduler.save_config') as mock_save:
with patch('scheduler.load_config') as mock_load:
mock_load.return_value = NightModeConfig()
new_config = {
"enabled": True,
"shutdown_time": "23:00",
"startup_time": "07:00",
"excluded_services": ["night-scheduler", "nginx"]
}
response = client.post("/api/night-mode", json=new_config)
assert response.status_code == 200
mock_save.assert_called_once()
def test_update_config_invalid_time(self):
"""POST /api/night-mode mit ungültiger Zeit gibt Fehler"""
with patch('scheduler.load_config') as mock_load:
mock_load.return_value = NightModeConfig()
new_config = {
"enabled": True,
"shutdown_time": "invalid",
"startup_time": "06:00",
"excluded_services": []
}
response = client.post("/api/night-mode", json=new_config)
assert response.status_code == 400
def test_execute_stop_endpoint(self):
"""POST /api/night-mode/execute mit stop"""
with patch('scheduler.execute_docker_command') as mock_exec:
with patch('scheduler.load_config') as mock_load:
with patch('scheduler.save_config'):
mock_exec.return_value = (True, "Services gestoppt")
mock_load.return_value = NightModeConfig()
response = client.post(
"/api/night-mode/execute",
json={"action": "stop"}
)
assert response.status_code == 200
mock_exec.assert_called_once_with("stop")
def test_execute_start_endpoint(self):
"""POST /api/night-mode/execute mit start"""
with patch('scheduler.execute_docker_command') as mock_exec:
with patch('scheduler.load_config') as mock_load:
with patch('scheduler.save_config'):
mock_exec.return_value = (True, "Services gestartet")
mock_load.return_value = NightModeConfig()
response = client.post(
"/api/night-mode/execute",
json={"action": "start"}
)
assert response.status_code == 200
mock_exec.assert_called_once_with("start")
def test_execute_invalid_action(self):
"""POST /api/night-mode/execute mit ungültiger Aktion"""
response = client.post(
"/api/night-mode/execute",
json={"action": "invalid"}
)
assert response.status_code == 400
def test_get_services_endpoint(self):
"""GET /api/night-mode/services gibt Services zurück"""
with patch('scheduler.get_services_to_manage') as mock_services:
with patch('scheduler.get_services_status') as mock_status:
with patch('scheduler.load_config') as mock_load:
mock_services.return_value = ["backend", "frontend"]
mock_status.return_value = {"backend": "running", "frontend": "running"}
mock_load.return_value = NightModeConfig()
response = client.get("/api/night-mode/services")
assert response.status_code == 200
data = response.json()
assert "all_services" in data
assert "excluded_services" in data
assert "status" in data
def test_get_logs_endpoint(self):
"""GET /api/night-mode/logs gibt Logs zurück"""
with patch('scheduler.load_config') as mock_load:
mock_load.return_value = NightModeConfig(
last_action="shutdown",
last_action_time="2026-02-09T22:00:00"
)
response = client.get("/api/night-mode/logs")
assert response.status_code == 200
data = response.json()
assert data["last_action"] == "shutdown"
class TestConfigPersistence:
"""Tests für Konfigurations-Persistenz"""
def test_load_missing_config_returns_default(self):
"""Fehlende Konfiguration gibt Standard zurück"""
with patch.object(CONFIG_PATH, 'exists', return_value=False):
config = load_config()
assert config.enabled is False
assert config.shutdown_time == "22:00"
def test_save_and_load_config(self, tmp_path):
"""Konfiguration speichern und laden"""
config_file = tmp_path / "night-mode.json"
with patch('scheduler.CONFIG_PATH', config_file):
original = NightModeConfig(
enabled=True,
shutdown_time="21:00",
startup_time="05:30"
)
save_config(original)
loaded = load_config()
assert loaded.enabled == original.enabled
assert loaded.shutdown_time == original.shutdown_time
assert loaded.startup_time == original.startup_time
if __name__ == "__main__":
pytest.main([__file__, "-v"])