lablup / lablup/backend.ai

Integrate Notification Center with Background Task System (bgtask)

Open
#6,440 0 comments 0 reactions 0 assignees View on GitHub
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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.