jpmschweitzer/portainer-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(scheduler): add Pydantic models and improve task API

Browse Source 
Scheduler API:
- Add Pydantic models for request/response validation
- Improve API documentation with examples
- Add detailed schedule pattern documentation
- Document priority levels and executor types

Models:
- TaskCreate, TaskUpdate, TaskResponse models
- Field validation and constraints
- Type safety for task operations

Documentation:
- Add TASK_REGISTRATION.md guide
- Document schedule patterns and executor configs
This commit is contained in:
jpmschweitzer
2025-12-10 01:26:21 +01:00
parent 0bd4f8056a
commit 8c0ced68eb
3 changed files with 570 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/scheduler/TASK_REGISTRATION.md
+289
View File
@@ -0,0 +1,289 @@
# Scheduler Task Registration Guide
## Overview
The Scheduler is **DB-driven** (not YAML-based). Tasks are registered via REST API and stored in PostgreSQL.
## Quick Reference
### Register a New Task
```bash
curl -X POST http://192.168.86.149:8090/tasks \
-H "Authorization: Bearer $SCHEDULER_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"task_name": "my_task",
"service": "my-service",
"executor": "rest_api",
"priority": 25,
"description": "Task description",
"minute": 0,
"hour": -1,
"config": {
"method": "POST",
"url": "http://service:port/endpoint",
"headers": {
"Authorization": "Bearer ${ENV_VAR}"
},
"body": {}
}
}'
```
### List All Tasks
```bash
curl http://192.168.86.149:8090/tasks \
-H "Authorization: Bearer $SCHEDULER_API_KEY"
```
### Trigger Task Manually
```bash
curl -X POST http://192.168.86.149:8090/tasks/{task_name}/trigger \
-H "Authorization: Bearer $SCHEDULER_API_KEY"
```
### View Task Execution History
```bash
curl http://192.168.86.149:8090/executions?task_name={task_name} \
-H "Authorization: Bearer $SCHEDULER_API_KEY"
```
## Schedule Patterns
Use cron-style fields where **`-1` means "every"**:
| Pattern | minute | hour | day_of_month | month | day_of_week | Description |
|---------|--------|------|--------------|-------|-------------|-------------|
| Every minute | -1 | -1 | -1 | -1 | -1 | Runs every minute |
| Hourly | 0 | -1 | -1 | -1 | -1 | Runs at minute 0 of every hour |
| Daily at 2 AM | 0 | 2 | -1 | -1 | -1 | Runs at 2:00 AM every day |
| Weekly (Monday 2 AM) | 0 | 2 | -1 | -1 | 0 | Runs at 2:00 AM every Monday |
| Monthly (1st at 2 AM) | 0 | 2 | 1 | -1 | -1 | Runs at 2:00 AM on 1st of month |
**Note:** `day_of_week` is 0-6 (Monday-Sunday)
## Priority Levels
| Priority Range | Level | Use Cases |
|----------------|-------|-----------|
| 1-5 | Emergency/System | Critical system tasks |
| 10-30 | User | User-initiated tasks |
| 40-70 | Maintenance | Background maintenance |
| 70+ | Low | Low priority cleanup |
## Executor Types
### `rest_api` Executor
Calls HTTP endpoints. Supports environment variable substitution in headers/body.
**Config Format:**
```json
{
"method": "POST|GET|PUT|DELETE|PATCH",
"url": "http://service:port/endpoint",
"headers": {
"Authorization": "Bearer ${ENV_VAR}",
"Content-Type": "application/json"
},
"body": {
"key": "value"
},
"timeout": 30,
"auth": {
"type": "bearer|basic|api_key",
"value": "${TOKEN_ENV_VAR}"
}
}
```
**Environment Variables:** Use `${VAR_NAME}` syntax to reference env vars from the scheduler container.
### Other Executors
- `shell`: Execute shell commands
- `python`: Execute Python scripts
- `docker`: Docker operations
- `backup`: Backup operations
- `doc_sync`: Documentation sync
## Complete Task Schema
### Required Fields
- `task_name` (string): Unique task identifier
- `service` (string): Service that owns this task (e.g., "library-desk", "core-api")
- `executor` (string): Executor type ("rest_api", "shell", etc.)
- `priority` (int): Priority level (1-100)
### Optional Fields
- `description` (string): Human-readable description
- `minute` (int): -1 to 59, default: -1 (every minute)
- `hour` (int): -1 to 23, default: -1 (every hour)
- `day_of_month` (int): -1 to 31, default: -1 (every day)
- `month` (int): -1 to 12, default: -1 (every month)
- `day_of_week` (int): -1 to 6, default: -1 (every day)
- `enabled` (bool): Whether task is enabled, default: true
- `max_retries` (int): Max retry attempts, default: 3
- `timeout_seconds` (int): Execution timeout, default: 3600
- `config` (object): Executor-specific configuration
- `created_by` (string): Creator identifier, default: "api"
## Example: Librarian Knowledge Consolidation
This task processes unprocessed search queries and consolidates knowledge into wiki pages.
**Registration:**
```bash
curl -X POST http://192.168.86.149:8090/tasks \
-H "Authorization: Bearer $SCHEDULER_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"task_name": "librarian_consolidation",
"service": "library-desk",
"executor": "rest_api",
"priority": 25,
"description": "Processes unprocessed search queries and consolidates knowledge",
"minute": 0,
"hour": -1,
"config": {
"method": "POST",
"url": "http://library-desk:8089/consolidation/run",
"headers": {
"Authorization": "Bearer ${LIBRARY_API_KEY}",
"Content-Type": "application/json"
},
"body": {
"dry_run": false,
"process_limit": 10,
"lookback_days": 7,
"min_web_results": 2
}
}
}'
```
**Schedule:** Runs hourly (at minute 0 of every hour)
**What it does:**
1. Queries Neo4j for unprocessed `SearchQuery` nodes
2. Analyzes web results with Ollama
3. Creates/updates wiki pages via WikiPageWriter
4. Extracts and adds new entities to knowledge graph
5. Marks searches as processed
## API Endpoints
| Method | Endpoint | Description |
|--------|----------|-------------|
| GET | `/tasks` | List all tasks |
| GET | `/tasks/{task_name}` | Get task details |
| POST | `/tasks` | Create new task |
| PUT | `/tasks/{task_name}` | Update task |
| DELETE | `/tasks/{task_name}` | Delete task |
| POST | `/tasks/{task_name}/trigger` | Manually trigger task |
| GET | `/executions` | View execution history |
| GET | `/stats` | Get scheduler statistics |
| GET | `/health` | Health check |
## Interactive Documentation
Visit **http://192.168.86.149:8090/docs** for interactive Swagger UI with:
- Full API documentation
- Request/response schemas
- Try-it-out functionality
- Example payloads
## Environment Variables
Get API keys from containers:
```bash
# Scheduler API key
docker exec scheduler printenv SCHEDULER_API_KEY
# Library Desk API key
docker exec library-desk printenv LIBRARY_API_KEY
```
## Task Management Tips
1. **Test with dry_run**: Set `enabled: false` initially or use dry_run in task config
2. **Monitor executions**: Check `/executions` endpoint regularly
3. **Start conservative**: Use longer intervals (hourly, daily) before going to frequent runs
4. **Use priorities wisely**: Reserve 1-10 for critical tasks, use 20-30 for user tasks
5. **Set timeouts appropriately**: Long-running tasks need higher timeout_seconds
## Troubleshooting
### Task not running
1. Check if task is enabled: `GET /tasks/{task_name}`
2. Check execution history: `GET /executions?task_name={task_name}`
3. Verify schedule matches current time
4. Check scheduler logs: `docker logs scheduler`
### Task failing
1. Check execution history for error messages
2. Verify endpoint is accessible from scheduler container
3. Test endpoint manually with curl
4. Check API keys and authentication
5. Verify executor configuration (especially for rest_api)
### Debug a task
```bash
# Get task details
curl http://192.168.86.149:8090/tasks/librarian_consolidation \
-H "Authorization: Bearer $SCHEDULER_API_KEY"
# View recent executions
curl http://192.168.86.149:8090/executions?task_name=librarian_consolidation&limit=5 \
-H "Authorization: Bearer $SCHEDULER_API_KEY"
# Trigger manually
curl -X POST http://192.168.86.149:8090/tasks/librarian_consolidation/trigger \
-H "Authorization: Bearer $SCHEDULER_API_KEY"
```
## Architecture
```
┌─────────────────────────────────────────────────────────┐
│ The Scheduler (AsyncIOScheduler) │
│ │
│ ┌────────────────────────────────────────────┐ │
│ │ Cron Job (every minute) │ │
│ │ ↓ │ │
│ │ Query PostgreSQL for tasks scheduled │ │
│ │ for current minute/hour/day │ │
│ │ ↓ │ │
│ │ Execute up to 5 tasks concurrently │ │
│ │ (based on priority) │ │
│ │ ↓ │ │
│ │ Call appropriate executor: │ │
│ │ - rest_api → HTTP request │ │
│ │ - shell → subprocess │ │
│ │ - python → exec │ │
│ └────────────────────────────────────────────┘ │
│ │
│ PostgreSQL: │
│ - scheduled_tasks (task definitions) │
│ - task_executions (execution history) │
└─────────────────────────────────────────────────────────┘
```
## Current Registered Tasks
As of 2025-12-09:
1. **backup_docker_configs_daily** - Daily at 3:05 AM
2. **librarian_consolidation** - Hourly (minute 0)
3. **test_example_task** - Every minute (testing)
4. **sync_fastapi_docs_monthly** - 11th of month at 4:00 AM
5. **sync_ollama_docs_monthly** - 12th of month at 4:00 AM
services/scheduler/src/main.py
+71 -17
View File
@@ -16,6 +16,7 @@ import logging
from src.config import get_settings, Settings
from src.tasks.executor import TaskExecutor
from src.models import TaskCreate, TaskUpdate, TaskResponse
# Configure logging
logging.basicConfig(
@@ -194,29 +195,82 @@ async def get_task_details(
return dict(task)
@app.post("/tasks")
@app.post("/tasks", response_model=TaskResponse, tags=["Task Management"])
async def create_task(
task_data: dict,
task: TaskCreate,
api_key: str = Depends(verify_api_key),
executor: TaskExecutor = Depends(get_task_executor)
):
"""Create a new scheduled task."""
"""
Create a new scheduled task.
## Schedule Pattern
Use cron-style fields where `-1` means "every":
- `minute: -1, hour: -1` → Runs every minute
- `minute: 0, hour: -1` → Runs at minute 0 of every hour (hourly)
- `minute: 0, hour: 2` → Runs at 2:00 AM every day
- `minute: 0, hour: 2, day_of_week: 0` → Runs at 2:00 AM every Monday
## Priority Levels
- **1-5**: Emergency/critical system tasks
- **10-30**: User-initiated tasks
- **40-70**: Background maintenance
- **70+**: Low priority cleanup
## Executor Types
### `rest_api` Executor
Calls HTTP endpoints. Config format:
```json
{
"method": "POST",
"url": "http://service:port/endpoint",
"headers": {
"Authorization": "Bearer ${ENV_VAR}",
"Content-Type": "application/json"
},
"body": {"key": "value"}
}
```
Environment variables can be referenced with `${VAR_NAME}` syntax.
## Example: Librarian Consolidation Task
```json
{
"task_name": "librarian_consolidation",
"service": "library-desk",
"executor": "rest_api",
"priority": 25,
"description": "Processes search queries and consolidates knowledge",
"minute": 0,
"hour": -1,
"config": {
"method": "POST",
"url": "http://library-desk:8089/consolidation/run",
"headers": {
"Authorization": "Bearer ${LIBRARY_DESK_API_KEY}",
"Content-Type": "application/json"
},
"body": {
"dry_run": false,
"process_limit": 10
}
}
}
```
This creates an hourly task that calls the consolidation endpoint.
"""
import psycopg2.extras
import json
required_fields = ['task_name', 'service', 'executor', 'priority']
if not all(field in task_data for field in required_fields):
raise HTTPException(400, f"Missing required fields: {required_fields}")
# Set defaults
task_data.setdefault('minute', -1)
task_data.setdefault('hour', -1)
task_data.setdefault('day_of_month', -1)
task_data.setdefault('month', -1)
task_data.setdefault('day_of_week', -1)
task_data.setdefault('enabled', True)
task_data.setdefault('max_retries', 3)
task_data.setdefault('timeout_seconds', 3600)
# Convert task model to dict
task_data = task.model_dump()
# Convert config dict to JSON string if present
if 'config' in task_data and isinstance(task_data['config'], dict):
@@ -236,7 +290,7 @@ async def create_task(
%(config)s::jsonb, %(max_retries)s, %(timeout_seconds)s,
%(created_by)s)
RETURNING *
""", {**task_data, 'created_by': task_data.get('created_by', 'api')})
""", task_data)
new_task = dict(cur.fetchone())
conn.commit()
services/scheduler/src/models.py
+210
View File
@@ -0,0 +1,210 @@
"""
Pydantic models for The Scheduler API.
"""
from pydantic import BaseModel, Field
from typing import Optional, Dict, Any
from enum import Enum
class SchedulePattern(str, Enum):
"""Common schedule patterns."""
EVERY_MINUTE = "every_minute"
HOURLY = "hourly"
DAILY = "daily"
WEEKLY = "weekly"
MONTHLY = "monthly"
class TaskExecutor(str, Enum):
"""Available task executors."""
REST_API = "rest_api"
SHELL = "shell"
PYTHON = "python"
class TaskPriorityLevel(str, Enum):
"""Task priority levels."""
EMERGENCY = "emergency" # 1-5: Critical system tasks
SYSTEM = "system" # 5-10: System maintenance
USER = "user" # 10-30: User-initiated tasks
MAINTENANCE = "maintenance" # 40-70: Background maintenance
LOW = "low" # 70+: Low priority cleanup
class TaskBase(BaseModel):
"""Base task fields."""
task_name: str = Field(
...,
description="Unique task identifier (e.g., 'librarian_consolidation')",
example="librarian_consolidation"
)
service: str = Field(
...,
description="Service that owns this task (e.g., 'library-desk', 'core-api')",
example="library-desk"
)
executor: str = Field(
...,
description="Executor type: 'rest_api', 'shell', or 'python'",
example="rest_api"
)
priority: int = Field(
...,
ge=1,
le=100,
description="Priority level (1-5: emergency, 10-30: user, 40-70: maintenance, 70+: low)",
example=25
)
description: Optional[str] = Field(
None,
description="Human-readable task description",
example="Processes unprocessed search queries and consolidates knowledge into wiki pages"
)
class TaskSchedule(BaseModel):
"""Cron-style schedule fields."""
minute: int = Field(
-1,
ge=-1,
le=59,
description="Minute to run (-1 = every minute, 0-59 = specific minute)",
example=-1
)
hour: int = Field(
-1,
ge=-1,
le=23,
description="Hour to run (-1 = every hour, 0-23 = specific hour)",
example=-1
)
day_of_month: int = Field(
-1,
ge=-1,
le=31,
description="Day of month to run (-1 = every day, 1-31 = specific day)",
example=-1
)
month: int = Field(
-1,
ge=-1,
le=12,
description="Month to run (-1 = every month, 1-12 = specific month)",
example=-1
)
day_of_week: int = Field(
-1,
ge=-1,
le=6,
description="Day of week to run (-1 = every day, 0-6 = Monday-Sunday)",
example=-1
)
class TaskConfig(BaseModel):
"""Task execution configuration."""
enabled: bool = Field(
True,
description="Whether task is enabled",
example=True
)
max_retries: int = Field(
3,
ge=0,
le=10,
description="Maximum retry attempts on failure",
example=3
)
timeout_seconds: int = Field(
3600,
ge=1,
description="Execution timeout in seconds",
example=3600
)
config: Optional[Dict[str, Any]] = Field(
None,
description="Executor-specific configuration (varies by executor type)",
example={
"method": "POST",
"url": "http://library-desk:8089/consolidation/run",
"headers": {
"Authorization": "Bearer ${LIBRARY_DESK_API_KEY}",
"Content-Type": "application/json"
},
"body": {
"dry_run": False,
"process_limit": 10,
"lookback_days": 7,
"min_web_results": 2
}
}
)
class TaskCreate(TaskBase, TaskSchedule, TaskConfig):
"""Request model for creating a new scheduled task."""
created_by: Optional[str] = Field(
"api",
description="User or system that created this task",
example="api"
)
class Config:
json_schema_extra = {
"example": {
"task_name": "librarian_consolidation",
"service": "library-desk",
"executor": "rest_api",
"priority": 25,
"description": "Processes unprocessed search queries and consolidates knowledge",
"minute": 0,
"hour": -1,
"day_of_month": -1,
"month": -1,
"day_of_week": -1,
"enabled": True,
"max_retries": 3,
"timeout_seconds": 3600,
"config": {
"method": "POST",
"url": "http://library-desk:8089/consolidation/run",
"headers": {
"Authorization": "Bearer ${LIBRARY_DESK_API_KEY}",
"Content-Type": "application/json"
},
"body": {
"dry_run": False,
"process_limit": 10,
"lookback_days": 7,
"min_web_results": 2
}
},
"created_by": "api"
}
}
class TaskUpdate(BaseModel):
"""Request model for updating a scheduled task."""
service: Optional[str] = None
executor: Optional[str] = None
priority: Optional[int] = Field(None, ge=1, le=100)
minute: Optional[int] = Field(None, ge=-1, le=59)
hour: Optional[int] = Field(None, ge=-1, le=23)
day_of_month: Optional[int] = Field(None, ge=-1, le=31)
month: Optional[int] = Field(None, ge=-1, le=12)
day_of_week: Optional[int] = Field(None, ge=-1, le=6)
enabled: Optional[bool] = None
description: Optional[str] = None
config: Optional[Dict[str, Any]] = None
max_retries: Optional[int] = Field(None, ge=0, le=10)
timeout_seconds: Optional[int] = Field(None, ge=1)
class TaskResponse(TaskCreate):
"""Response model for task operations."""
created_at: str
updated_at: Optional[str] = None
class Config:
from_attributes = True