Integrate Notification Center with Background Task System (bgtask)
- Dominant language
- Python
- Stars
- 670
- Forks
- 183
- Avg merge
- 17h 7m
- Merged PRs (30d)
- 358
Description
## User Story
As a Backend.AI developer, I want background tasks to automatically emit notifications on completion/failure, so that users and administrators are informed of task status without manual intervention.
## Context
This story integrates the Notification Center with the existing background task (bgtask) system, enabling automatic notification emission for task lifecycle events.
## Functional Requirements
### 1. Integration Points in BackgroundTaskReporter
Modify BackgroundTaskReporter to emit notifications:
```python
# src/ai/backend/common/bgtask/reporter.py
class BackgroundTaskReporter:
def __init__(
self,
...,
notification_service: NotificationService | None = None,
):
self.notification_service = notification_service
async def report_completion(
self,
task_id: str,
success_count: int,
failure_count: int,
duration: float,
):
# Existing: Update Redis state
await self._update_task_state(...)
# New: Emit notification
if self.notification_service:
await self.notification_service.emit_task_completed(
task_id=task_id,
success_count=success_count,
failure_count=failure_count,
duration=duration,
task_type=self.task_type,
domain=self.domain,
)
async def report_failure(
self,
task_id: str,
error: Exception,
):
# Existing: Update Redis state
await self._update_task_state(...)
# New: Emit notification
if self.notification_service:
await self.notification_service.emit(
event_type="bgtask.failed",
subject=f"Task {task_id} failed",
severity="error",
context={
"task_id": task_id,
"task_type": self.task_type,
"error_type": type(error).__name__,
"error_message": str(error),
},
)
```
### 2. Event Types
Define standard bgtask event types:
```python
# Event types
BGTASK_STARTED = "bgtask.started"
BGTASK_PROGRESS = "bgtask.progress" # Optional: for long-running tasks
BGTASK_COMPLETED = "bgtask.completed"
BGTASK_FAILED = "bgtask.failed"
BGTASK_SUBTASK_UPDATED = "bgtask.subtask_updated" # Optional
```
### 3. Notification Service Injection
Add NotificationService to shared_config or context:
```python
# In manager initialization
notification_service = NotificationService(
redis_client=shared_config.redis_client,
db_session=shared_config.db_session_factory,
)
# Inject into bgtask context
bgtask_context = BackgroundTaskContext(
...,
notification_service=notification_service,
)
```
### 4. Default Notification Rules
Create default notification rules for common scenarios:
```python
# Migration or initialization script
default_rules = [
{
"name": "Admin notification for failed tasks",
"event_type": "bgtask.failed",
"channel": admin_email_channel,
"message_template": '''
{
"subject": "Background task failed: {{task_id}}",
"body": "Task {{task_type}} failed with error: {{error_message}}"
}
''',
"enabled": True,
},
{
"name": "Webhook for all task completions",
"event_type": "bgtask.completed",
"channel": webhook_channel,
"message_template": '''
{
"task_id": "{{task_id}}",
"status": "completed",
"success_count": {{success_count}},
"failure_count": {{failure_count}},
"duration": {{duration}}
}
''',
"enabled": False, # Opt-in
}
]
```
### 5. Backward Compatibility
Ensure existing bgtask functionality works without notification:
```python
# NotificationService is optional
if self.notification_service:
await self.notification_service.emit(...)
# Continue normal operation even without notifications
```
### 6. Configuration
Add bgtask notification settings:
```
[bgtask]
# Existing settings...
[bgtask.notification]
enabled = true
emit_on_completion = true
emit_on_failure = true
emit_on_progress = false # Optional: for long tasks
```
## Testing Scenarios
1. **Task completes successfully**: Notification emitted with success metrics
1. **Task fails**: Error notification emitted with stack trace
1. **Notification service disabled**: Task continues normally
1. **Multiple subtasks**: Each subtask update can emit notification (optional)
1. **Long-running task**: Progress notifications at intervals (optional)
## Acceptance Criteria
- [ ] BackgroundTaskReporter integration complete
- [ ] Notification emission on task completion
- [ ] Notification emission on task failure
- [ ] NotificationService injection via context
- [ ] Backward compatibility maintained
- [ ] Configuration options added
- [ ] Default notification rules created
- [ ] Unit tests for integration
- [ ] Integration test with full flow
- [ ] Documentation updated
## Related Issues
Epic: BA-302
Related: BA-2849 (Background Task Plugin System)
Depends on: BA-2863 (Service Layer API)
JIRA Issue: BA-2866
Contributor guide
Assessment
This issue has not been assessed yet.