electron-rare/meet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

wip réécriture

Browse Source
This commit is contained in:
lebaudantoine
2024-10-31 00:35:08 +01:00
parent 95792c7e8b
commit f85dc6eceb
17 changed files with 743 additions and 759 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/backend/core/api/permissions.py
+37
View File
@@ -1,5 +1,7 @@
"""Permission handlers for the Meet core app."""
from django.conf import settings
from rest_framework import permissions
from ..models import RoleChoices
@@ -65,6 +67,21 @@ class RoomPermissions(permissions.BasePermission):
return obj.is_administrator(user)
class IsRoomOwnerOrAdministrator(permissions.BasePermission):
"""Temporary"""
message = "You must be an admin or owner to start a recording."
def has_permission(self, request, view):
# Get the room object
room = view.get_object()
if not room.is_owner(request.user) and not room.is_administrator(request.user):
return False
return True
class ResourceAccessPermission(permissions.BasePermission):
"""
Permissions for a room that can only be updated by room administrators.
@@ -83,3 +100,23 @@ class ResourceAccessPermission(permissions.BasePermission):
return obj.user == user
return obj.resource.is_administrator(user)
class IsRecordingEnabled(permissions.BasePermission):
"""Check if the recording feature is enabled."""
message = "Access denied, recording is disabled."
def has_permission(self, request, view):
"""Determine if access is allowed based on settings."""
return settings.RECORDING_ENABLE
class IsStorageEventEnabled(permissions.BasePermission):
"""Check if the storage event feature is enabled."""
message = "Access denied, storage event is disabled."
def has_permission(self, request, view):
"""Determine if access is allowed based on settings."""
return settings.RECORDING_STORAGE_EVENT_ENABLE
src/backend/core/api/viewsets.py
+97 -74
View File
@@ -4,8 +4,6 @@ import uuid
from logging import getLogger
from django.conf import settings
from django.core.exceptions import PermissionDenied, ValidationError
from django.db import IntegrityError
from django.db.models import Q
from django.http import Http404
from django.shortcuts import get_object_or_404
@@ -17,6 +15,9 @@ from rest_framework import (
pagination,
viewsets,
)
from rest_framework import (
exceptions as drf_exceptions,
)
from rest_framework import (
response as drf_response,
)
@@ -25,16 +26,18 @@ from rest_framework import (
)
from core import models, utils
from core.recording import (
IgnoreNotificationError,
LivekitEgressWorker,
MinioParser,
RecordingNotFound,
RecordingSessionManager,
from core.recording.storage import (
InvalidBucketError,
InvalidFileTypeError,
InvalidRequestDataError,
StorageEventAuthentication,
get_parser,
)
from core.recording.worker import (
RecordingStartError,
RecordingStopError,
RecordingUpdateError,
StorageHandler,
WorkerServiceMediator,
worker_service_provider,
)
from ..analytics import analytics
@@ -184,12 +187,6 @@ class RoomViewSet(
permission_classes = [permissions.RoomPermissions]
queryset = models.Room.objects.all()
serializer_class = serializers.RoomSerializer
session_manager = RecordingSessionManager(
worker_class=LivekitEgressWorker,
output_folder="recordings",
server_configurations=settings.LIVEKIT_CONFIGURATION,
enable_output_logging=settings.LOG_RECORDING_OUTPUT,
)
def get_object(self):
"""Allow getting a room by its slug."""
@@ -236,68 +233,81 @@ class RoomViewSet(
return drf_response.Response(data)
@decorators.action(detail=True, methods=["post"], url_path="start-recording")
@decorators.action(
detail=True,
methods=["post"],
url_path="start-recording",
permission_classes=[
permissions.IsRoomOwnerOrAdministrator,
permissions.IsRecordingEnabled,
],
)
def start_room_recording(self, request, pk=None): # pylint: disable=unused-argument
"""Start room recording."""
if not settings.ENABLE_RECORDING:
raise PermissionDenied({"error": "Recording is disabled."})
"""Start recording a room."""
room = self.get_object()
if not room.is_owner_or_administrator(request.user):
raise PermissionDenied(
"You must be an admin or owner to start a recording."
mode = request.data.get("mode")
if mode is None:
return drf_response.Response(
{"error": "Recording mode is required."},
status=drf_status.HTTP_400_BAD_REQUEST,
)
mode = request.data.get(
"mode", models.RecordingModeChoices.SCREEN_RECORDING
# May raise exception if an active recording already exist for the room
recording = models.Recording.objects.create(
creator=request.user, room=room, mode=mode
)
if mode not in models.RecordingModeChoices.values:
raise ValidationError({"error": "Invalid recording mode specified."})
worker_service = worker_service_provider.create(mode=recording.mode)
worker_manager = WorkerServiceMediator(worker_service=worker_service)
try:
recording = models.Recording.objects.create(
creator=request.user, room=room, mode=mode
)
except IntegrityError as e:
# todo - integrity error not specific enough
logger.error(
"An active recording already exists for room %s: %s", room.slug, e
)
worker_manager.start(recording)
except RecordingStartError:
return drf_response.Response(
{"error": f"An active recording already exists for room {room.slug}."},
status=drf_status.HTTP_200_OK,
{"error": f"Recording failed to start for room {room.slug}"},
status=drf_status.HTTP_500_INTERNAL_SERVER_ERROR,
)
# May raise exception
self.session_manager.start_recording(recording)
return drf_response.Response(
{"message": f"Recording started for room {room.slug}"},
{"message": f"Recording successfully started for room {room.slug}"},
status=drf_status.HTTP_201_CREATED,
)
@decorators.action(detail=True, methods=["post"], url_path="stop-recording")
@decorators.action(
detail=True,
methods=["post"],
url_path="stop-recording",
permission_classes=[
permissions.IsRoomOwnerOrAdministrator,
permissions.IsRecordingEnabled,
],
)
def stop_room_recording(self, request, pk=None): # pylint: disable=unused-argument
"""Stop room recording."""
if not settings.ENABLE_RECORDING:
raise PermissionDenied({"error": "Recording is disabled."})
room = self.get_object()
if not room.is_owner_or_administrator(request.user):
raise PermissionDenied("You must be an admin or owner to stop a recording.")
try:
recording = models.Recording.objects.get(
room=room, status=models.RecordingStatusChoices.ACTIVE
)
except models.Recording.DoesNotExist:
return drf_response.Response(
{"error": "No active recording found for this room."},
status=drf_status.HTTP_404_NOT_FOUND,
)
except models.Recording.DoesNotExist as e:
raise drf_exceptions.NotFound(
"No active recording found for this room."
) from e
# May raise exception
self.session_manager.stop_recording(recording)
worker_service = worker_service_provider.create(mode=recording.mode)
worker_manager = WorkerServiceMediator(worker_service=worker_service)
try:
worker_manager.stop(recording)
except RecordingStopError:
return drf_response.Response(
{"error": f"Recording failed to stop for room {room.slug}"},
status=drf_status.HTTP_500_INTERNAL_SERVER_ERROR,
)
return drf_response.Response(
{"message": f"Recording stopped for room {room.slug}."}
@@ -390,33 +400,46 @@ class RecordingViewSet(viewsets.GenericViewSet, mixins.ListModelMixin):
API endpoints to access and perform actions on recording.
"""
storage = StorageHandler(
bucket_name=settings.AWS_STORAGE_BUCKET_NAME, parser_class=MinioParser
@decorators.action(
detail=False,
methods=["post"],
url_path="storage-hook",
authentication_classes=[StorageEventAuthentication],
permission_classes=[permissions.IsStorageEventEnabled],
)
@decorators.action(detail=False, methods=["post"], url_path="storage-hook")
def on_save(self, request, pk=None): # pylint: disable=unused-argument
def on_storage_event_received(self, request, pk=None): # pylint: disable=unused-argument
"""Handle incoming storage hook events for recordings."""
if not settings.AWS_ENABLE_STORAGE_HOOK:
raise PermissionDenied({"error": "Storage hook is disabled."})
parser = get_parser()
try:
self.storage.on_save(request.data)
except IgnoreNotificationError:
recording_id = parser.get_recording_id(request.data)
except InvalidRequestDataError as e:
raise drf_exceptions.PermissionDenied(f"Invalid request data: {e}") from e
except InvalidBucketError as e:
raise drf_exceptions.PermissionDenied("Invalid bucket specified") from e
except InvalidFileTypeError as e:
return drf_response.Response(
{"message": "Ignore, doesn't match criteria"},
)
except RecordingUpdateError as e:
return drf_response.Response(
{"error": f"Error updating recording: {e}"},
status=drf_status.HTTP_403_FORBIDDEN,
)
except RecordingNotFound:
return drf_response.Response(
{"error": "No recording found for this event."},
status=drf_status.HTTP_404_NOT_FOUND,
{"message": f"Ignore this file type, {e}"},
)
# TODO - trigger postprocessing based on recording's mode
try:
recording = models.Recording.objects.get(id=recording_id)
except models.Recording.DoesNotExist as e:
raise drf_exceptions.NotFound("No recording found for this event.") from e
if not recording.is_savable():
raise drf_exceptions.PermissionDenied(
f"Recording with ID {recording_id} cannot be saved because it is either in an error state or has already been saved."
)
recording.status = models.RecordingStatusChoices.SAVED
recording.save()
# todo - trigger post-processing
return drf_response.Response(
{"message": "Event processed."},
src/backend/core/models.py
+10 -8
View File
@@ -256,10 +256,6 @@ class Resource(BaseModel):
"""Check if a user is owner of the resource."""
return RoleChoices.check_owner_role(self.get_role(user))
def is_owner_or_administrator(self, user):
"""Check if a user is owner or administrator of the resource."""
return self.is_owner(user) or self.is_administrator(user)
class ResourceAccess(BaseModel):
"""Link table between resources and users"""
@@ -366,7 +362,6 @@ class Room(Resource):
# todo - discuss how the path could changed, and we could loose track of file
# todo - discuss the uniqueness of worker_id field
class Recording(BaseModel):
"""Model for recordings that take place in a room"""
@@ -390,9 +385,8 @@ class Recording(BaseModel):
mode = models.CharField(
max_length=20,
choices=RecordingModeChoices.choices,
default=RecordingModeChoices.SCREEN_RECORDING,
verbose_name=_("Recording mode"),
help_text=_("Defines the type of recording being performed."),
verbose_name=_("Worker kind"),
help_text=_("Defines the kind of worker being called."),
)
worker_id = models.CharField(
max_length=255,
@@ -433,3 +427,11 @@ class Recording(BaseModel):
"stop": is_creator and not is_final_status,
"update": False,
}
def is_savable(self) -> bool:
"""Wip."""
is_in_error = RecordingStatusChoices.is_error_status(self.status)
is_already_saved = self.status == RecordingStatusChoices.SAVED
return not is_in_error and not is_already_saved
src/backend/core/recording/__init__.py
-5
View File
@@ -1,5 +0,0 @@
"""Module for recording classes."""
from .exceptions import *
from .storage import *
from .worker import *
src/backend/core/recording/event/__init__.py
+3
View File
@@ -0,0 +1,3 @@
from .authentification import *
from .exceptions import *
from .wip import *
src/backend/core/recording/event/authentification.py
+65
View File
@@ -0,0 +1,65 @@
"""Authentication class for storage event token validation."""
import logging
import secrets
from django.conf import settings
from django.utils.translation import gettext_lazy as _
from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed
logger = logging.getLogger(__name__)
class StorageEventAuthentication(BaseAuthentication):
"""Authenticate requests using a Bearer token for storage event integration.
This class validates Bearer tokens for storage events that don't map to database users.
It's designed for S3-compatible storage integrations and similar use cases.
Events are submitted when a webhook is configured on some bucket's events.
"""
AUTH_HEADER = "Authorization"
TOKEN_TYPE = "Bearer" # noqa S105
def authenticate(self, request):
"""Validate the Bearer token from the Authorization header."""
required_token = settings.RECORDING_STORAGE_EVENT_TOKEN
if not required_token:
return None
auth_header = request.headers.get(self.AUTH_HEADER)
if not auth_header:
logger.warning(
"Authentication failed: Missing Authorization header (ip: %s)",
request.META.get("REMOTE_ADDR"),
)
raise AuthenticationFailed(_("Authorization header is required"))
auth_parts = auth_header.split(" ")
if len(auth_parts) != 2 or auth_parts[0] != self.TOKEN_TYPE:
logger.warning(
"Authentication failed: Invalid authorization header (ip: %s)",
request.META.get("REMOTE_ADDR"),
)
raise AuthenticationFailed(_("Invalid authorization header."))
token = auth_parts[1]
# Use constant-time comparison to prevent timing attacks
if not secrets.compare_digest(token.encode(), required_token.encode()):
logger.warning(
"Authentication failed: Invalid token (ip: %s)",
request.META.get("REMOTE_ADDR"),
)
raise AuthenticationFailed(_("Invalid token"))
return None
def authenticate_header(self, request):
"""Return the WWW-Authenticate header value."""
return f"{self.TOKEN_TYPE} realm='Storage event API'"
src/backend/core/recording/event/exceptions.py
+18
View File
@@ -0,0 +1,18 @@
"""Recording and storage services specific exceptions."""
class ParsingEventDataError(Exception):
"""Raised when the request data is malformed, incomplete, or missing."""
class InvalidBucketError(Exception):
"""Raised when the bucket name in the request does not match the expected one."""
class InvalidFileTypeError(Exception):
"""Raised when the file type in the request is not supported."""
class InvalidFilepathError(Exception):
"""Raised when the filepath in the request is invalid."""
src/backend/core/recording/event/parsers.py
+139
View File
@@ -0,0 +1,139 @@
import re
import logging
from dataclasses import dataclass
from functools import lru_cache
from typing import Any, Dict, Protocol, Optional
from django.conf import settings
from django.utils.module_loading import import_string
from .exceptions import ParsingEventDataError, InvalidBucketError, InvalidFileTypeError, InvalidFilepathError
logger = logging.getLogger(__name__)
@dataclass
class StorageEvent:
"""Represents a storage event with relevant metadata.
Attributes:
filepath: Identifier for the affected recording
filetype: Type of storage event
bucket_name: When the event occurred
metadata: Additional event data
"""
filepath: str
filetype: str
bucket_name: str
metadata: Optional[Dict[str, Any]]
class EventParser(Protocol):
"""Wip."""
def __init__(self, bucket_name, allowed_filetypes = None):
"""Wip."""
def parse(self, data: Dict) -> StorageEvent:
"""Wip."""
def validate(self, data: StorageEvent) -> None:
"""Wip."""
def get_recording_id(self, data: Dict) -> str:
"""Wip."""
# todo - explain we don't need a factory class, a function with cache is enough
@lru_cache(maxsize=1)
def get_parser() -> EventParser:
"""Wip."""
event_parser_cls = import_string(settings.RECORDING_EVENT_PARSER_CLASS)
return event_parser_cls(bucket_name=settings.AWS_STORAGE_BUCKET_NAME)
class MinioParser:
"""Wip."""
def __init__(self, bucket_name, allowed_filetypes = None):
"""Wip."""
self._bucket_name = bucket_name
self._allowed_filetypes = allowed_filetypes or {"audio/ogg", "video/mp4"}
self._filepath_regex = re.compile(
r"(?P<folder>(?:[^%]+%2F)*)?(?P<recording_id>[0-9a-fA-F\-]{36})\.(?P<extension>[a-zA-Z0-9]+)"
)
@staticmethod
def parse(data):
"""Wip."""
if not data:
raise ParsingEventDataError("Received empty data")
try:
record = data["Records"][0]
s3 = record["s3"]
bucket_name = s3["bucket"]["name"]
file_object = s3["object"]
filepath = file_object["key"]
filetype = file_object["contentType"]
except (KeyError, IndexError) as e:
# todo - be more actionable
raise ParsingEventDataError(f"Missing or malformed field in event data: {e}") from e
try:
return StorageEvent(
filepath=filepath,
filetype=filetype,
bucket_name=bucket_name,
)
except TypeError as e:
# todo - be more actionable
raise ParsingEventDataError(
"Missing essential data fields: filepath, filetype, or bucket name"
) from e
def validate(self, event_data: StorageEvent) -> str:
"""Wip."""
if event_data.bucket_name != self._bucket_name:
raise InvalidBucketError(
f"Invalid bucket: expected {self._bucket_name}, got {event_data.bucket_name}"
)
if not event_data.filetype in self._allowed_filetypes:
raise InvalidFileTypeError(
f"Invalid file type, expected {self._allowed_filetypes}, got '{event_data.filetype}'"
)
match = self._filepath_regex.match(event_data.filepath)
if not match:
raise InvalidFilepathError(f"Invalid filepath structure: {event_data.filepath}")
recording_id = match.group("recording_id")
return recording_id
def get_recording_id(self, data):
"""Wip."""
event_data = self.parse(data)
recording_id = self.validate(event_data)
return recording_id
src/backend/core/recording/exceptions.py
-66
View File
@@ -1,66 +0,0 @@
"""Recording specific exceptions."""
# todo - discuss if it's a good idea to prefix Exception with Worker
class WorkerRequestError(Exception):
"""
Exception raised when the request is invalid
"""
class WorkerConnectionError(Exception):
"""
Exception raised when the client is not accessible
"""
class WorkerResponseError(Exception):
"""
Exception raised when the response is not as expected
"""
class RecordingStartError(Exception):
"""
Exception raised when the response is not as expected
"""
class RecordingStopError(Exception):
"""
Exception raised when the response is not as expected
"""
class InvalidBucketError(Exception):
"""Exception raised when the bucket name in the request does not match the expected one."""
class InvalidRequestDataError(Exception):
"""
Exception raised when the request data is malformed, incomplete, or missing.
"""
class InvalidFileTypeError(Exception):
"""
Exception raised when the file type in the request is not supported.
"""
class IgnoreNotificationError(Exception):
"""
Exception raised when a notification should be ignored due to non-critical issues.
"""
class RecordingNotFound(Exception):
"""
Exception raised when the requested recording cannot be found in the database.
"""
class RecordingUpdateError(Exception):
"""
Exception raised when the requested recording can not be updated.
"""
src/backend/core/recording/storage.py
-252
View File
@@ -1,252 +0,0 @@
"""Recording storage classes"""
import logging
import re
from abc import ABC, abstractmethod
from core.models import Recording, RecordingStatusChoices
from .exceptions import (
IgnoreNotificationError,
InvalidBucketError,
InvalidFileTypeError,
InvalidRequestDataError,
RecordingNotFound,
RecordingUpdateError,
)
logger = logging.getLogger(__name__)
class AbstractHookParser(ABC):
"""Abstract base class for hook parsers.
This class defines the interface for hook parsers, which are responsible
for parsing and validating incoming hook data from external sources.
"""
@abstractmethod
def __init__(self, bucket_name):
"""Initialize the hook parser with a bucket name.
Args:
bucket_name (str): The name of the bucket where data is stored.
"""
@abstractmethod
def extract_recording_id(self, data) -> str:
"""Parse and validate the incoming hook data.
Args:
data (dict): The data received from the hook.
Returns:
recording_id: The ID of the recording object.
Raises:
InvalidRequestDataError: If the data format is invalid.
InvalidBucketError: If the bucket in the data does not match the expected one.
InvalidFileTypeError: If the file type is not supported.
"""
class StorageHandler:
"""
Handle storage-related operations for recording data.
This class processes incoming data, validates it, and updates the
status of the recording in the database accordingly.
"""
def __init__(self, bucket_name, parser_class):
"""Initialize the storage handler with a bucket and parser class.
Args:
bucket_name (str): The name of the bucket where data is stored.
parser_class (class): A class that parses incoming hook data.
"""
self._parser = parser_class(bucket_name=bucket_name)
@staticmethod
def _update_recording(recording):
"""Update the status of the recording to 'SAVED'.
Args:
recording (Recording): The recording instance to update.
Raises:
RecordingUpdateError: If the recording is already saved or
is in an error state.
"""
# todo - check others status
if RecordingStatusChoices.is_error_status(recording.status):
logger.exception(
"Recording with ID %s is in an error state and cannot be saved.",
recording.id,
)
raise RecordingUpdateError(
f"Recording with ID {recording.id} is in an error state and cannot be saved."
)
if recording.status == RecordingStatusChoices.SAVED:
logger.exception("Recording with ID %s is already saved.", recording.id)
raise RecordingUpdateError(
f"Recording with ID {recording.id} is already saved."
)
recording.status = RecordingStatusChoices.SAVED
recording.save()
logger.info("Recording with ID %s was successfully saved.", recording)
def on_save(self, data) -> Recording:
"""Handle the process of updating a recording's status based on incoming data.
This method parses the incoming data, validates the recording,
and updates its status to "SAVED" in the database.
Args:
data (dict): The data received from the hook.
Returns:
Recording: The updated recording object.
Raises:
IgnoreNotificationError: If the data is invalid or irrelevant.
RecordingNotFound: If the recording with the given ID cannot be found.
"""
try:
recording_id = self._parser.extract_recording_id(data)
except InvalidRequestDataError as e:
logger.exception("Could not handle hook event %s", e)
logger.debug("Invalid request data: %s", data)
raise IgnoreNotificationError("Invalid request data received.") from e
except InvalidBucketError as e:
logger.exception("Invalid bucket queried: %s", e)
raise IgnoreNotificationError("Invalid bucket specified.") from e
except InvalidFileTypeError as e:
logger.info("Ignored event as it does not pertain to a recording. %s", e)
logger.debug("Non-recording file detected in request: %s", data)
raise IgnoreNotificationError("Invalid file type received.") from e
try:
recording = Recording.objects.get(id=recording_id)
except Recording.DoesNotExist as e:
logger.exception("Recording with ID %s not found.", recording_id)
raise RecordingNotFound(
f"Recording with ID {recording_id} not found."
) from e
self._update_recording(recording)
return recording
class MinioParser(AbstractHookParser):
"""Parser for handling incoming Minio hook data.
This class extracts essential information such as the bucket name,
filename, and file type from the incoming hook data. It also provides
methods for extracting and validating recording IDs.
"""
# Todo - discuss if it should be an instance attribute
FILENAME_PATTERN = re.compile(
r"(?P<parent_folder>(?:[^%]+%2F)*)?(?P<recording_id>[0-9a-fA-F\-]{36})\.(?P<file_type>[a-zA-Z0-9]+)"
)
def __init__(self, bucket_name):
"""Initialize the MinioParser with the expected bucket name.
Args:
bucket_name (str): The name of the bucket where recording files are stored.
"""
self._bucket_name = bucket_name
@staticmethod
def _extract(data):
"""Extract the bucket name, filename, and file type from the Minio event data.
This method parses the incoming event data to extract key fields such as the
bucket name, the filename (which contains the recording ID), and the file type.
Args:
data (dict): The event payload data received from Minio, containing the file
and bucket details.
Returns:
tuple: A tuple containing the extracted bucket name (str), filename (str),
and file type (str).
Raises:
InvalidRequestDataError: If the data is missing or has unexpected structure.
KeyError: If required fields are missing in the data.
IndexError: If the event data doesn't follow the expected structure.
"""
if not data:
raise InvalidRequestDataError("Received empty data")
try:
record = data["Records"][0]
s3 = record["s3"]
bucket_name = s3["bucket"]["name"]
filepath = s3["object"]["key"]
filetype = s3["object"]["contentType"]
except KeyError as e:
raise InvalidRequestDataError(f"Missing required field: {e}") from e
except IndexError as e:
raise InvalidRequestDataError(f"Unexpected data structure: {e}") from e
if not filepath or not filetype or not bucket_name:
raise InvalidRequestDataError(
"Missing essential data fields: filepath, filetype, or bucket name"
)
return bucket_name, filepath, filetype
def extract_recording_id(self, data):
"""Extract and validate the recording ID from the event's filename.
This method checks if the extracted bucket name matches the expected bucket,
validates that the file type is correct, and ensures that the filename conforms
to the expected format (`{recording_id}.{file_type}`). The recording ID must
be a valid UUID (v4).
Args:
data (dict): The event payload data from Minio.
Returns:
str: The validated recording ID extracted from the filename.
Raises:
InvalidBucketError: If the bucket name in the data doesn't match the expected bucket.
InvalidFileTypeError: If the file type is not 'audio/ogg'.
InvalidRequestDataError: If the filename doesn't follow the expected format.
"""
bucket_name, filename, filetype = self._extract(data)
if bucket_name != self._bucket_name:
raise InvalidBucketError(
f"Invalid bucket: expected {self._bucket_name}, got {bucket_name}"
)
# FIXME - bulky, not extensible
if filetype not in {"audio/ogg", "video/mp4"}:
raise InvalidFileTypeError(
f"Invalid file type, expected 'ogg' or 'mp4', got '{filetype}'"
)
match = self.FILENAME_PATTERN.match(filename)
if not match:
raise InvalidRequestDataError(f"Invalid filename structure: {filename}")
recording_id = match.group("recording_id")
return recording_id
src/backend/core/recording/worker.py
-345
View File
@@ -1,345 +0,0 @@
"""Recording worker and session classes"""
# pylint: disable=no-member
import logging
from abc import ABC, abstractmethod
import aiohttp
from asgiref.sync import async_to_sync
from livekit.api import TwirpError
from livekit.api.egress_service import EgressService
from livekit.protocol import egress as proto_egress
from core.models import Recording, RecordingModeChoices, RecordingStatusChoices
from .exceptions import (
RecordingStartError,
RecordingStopError,
WorkerConnectionError,
WorkerRequestError,
WorkerResponseError,
)
logger = logging.getLogger(__name__)
class AbstractRecordingWorker(ABC):
"""
Abstract base class for recording workers.
This class defines the interface that recording workers must implement
for starting and stopping recordings. It ensures that any recording
worker used will have the necessary methods for managing the recording
lifecycle.
Methods:
__init__: Initializes the worker with the required configurations.
start_recording: Asynchronously starts a recording session.
stop_recording: Asynchronously stops a recording session.
"""
@abstractmethod
def __init__(
self,
output_folder: str,
server_configurations: dict,
enable_output_logging: bool,
):
"""Initialize the recording worker with the output folder and server configurations.
Args:
output_folder (str): Path where the recording files will be stored.
server_configurations (dict): Configuration settings for the recording service.
"""
@abstractmethod
async def start_recording(self, recording: Recording) -> str:
"""Start a recording session.
Args:
recording (Recording): A recording object containing the necessary details.
Returns:
str: The unique identifier for the recording session (e.g., egress ID).
"""
@abstractmethod
async def stop_recording(self, recording: Recording) -> RecordingStatusChoices:
"""Stop a recording session.
Args:
recording (Recording): A recording object containing the necessary details.
Returns:
RecordingStatusChoices: The final status of the recording (e.g., STOPPED or ABORTED).
"""
class RecordingSessionManager:
"""
Manage the lifecycle of a recording session using a recording worker.
This class acts as a wrapper around an `AbstractRecordingWorker` instance.
It is responsible for starting and stopping recordings, updating the recording's
status in the database, and handling errors.
Methods:
start_recording: Start a recording session using the provided worker.
stop_recording: Stop a recording session using the provided worker.
"""
def __init__(self, worker_class, **kwargs):
"""Initialize the RecordingWorker with the specified recording worker.
Args:
worker_class (AbstractRecordingWorker): An implementation of the recording worker.
"""
self._worker: AbstractRecordingWorker = worker_class(**kwargs)
def start_recording(self, recording: Recording):
"""Start a recording session.
This method attempts to start a recording using the configured worker.
It updates the recording's status and handles any potential errors during
the process.
Args:
recording (Recording): The recording object with necessary details.
Raises:
RecordingStartError: If the recording fails to start.
"""
try:
worker_id = self._worker.start_recording(recording)
recording.worker_id = worker_id
recording.status = RecordingStatusChoices.ACTIVE
except (WorkerRequestError, WorkerConnectionError, WorkerResponseError) as e:
logger.error(
"Failed to start recording for room %s: %s", recording.room.slug, e
)
recording.status = RecordingStatusChoices.FAILED_TO_START
raise RecordingStartError() from e
finally:
recording.save()
logger.info(
"Worker started for room %s (worker ID: %s, mode: %s)",
recording.room,
recording.worker_id,
recording.mode,
)
def stop_recording(self, recording: Recording):
"""Stop a recording session.
This method attempts to stop a recording using the configured worker.
It updates the recording's status and handles any potential errors during
the process.
Args:
recording (Recording): The recording object with necessary details.
Raises:
RecordingStopError: If the recording fails to stop.
"""
try:
recording.status = self._worker.stop_recording(recording)
except (WorkerConnectionError, WorkerResponseError) as e:
logger.error(
"Failed to stop recording for room %s: %s", recording.room.slug, e
)
recording.status = RecordingStatusChoices.FAILED_TO_STOP
raise RecordingStopError() from e
finally:
recording.save()
logger.info("Worker stopped for room %s", recording.room)
class LivekitEgressWorker(AbstractRecordingWorker):
"""
Worker class to handle LiveKit recording egress.
This class implements the necessary methods to start and stop recordings
through LiveKit's egress service. It builds requests based on recording
details and communicates with LiveKit to manage recording sessions.
Methods:
start_recording: Start a recording in LiveKit.
stop_recording: Stop a recording in LiveKit.
"""
# FIXME - I feel it bulky
MODE_MAPPINGS = {
RecordingModeChoices.SCREEN_RECORDING: {
"type": proto_egress.EncodedFileType.MP4,
"extension": "mp4",
"extra_params": {},
},
RecordingModeChoices.TRANSCRIPT: {
"type": proto_egress.EncodedFileType.OGG,
"extension": "ogg",
"extra_params": {"audio_only": True},
},
}
def __init__(
self,
output_folder: str,
server_configurations: dict,
enable_output_logging: bool,
):
"""Initialize the LiveKit worker with the output folder and server configurations.
Args:
output_folder (str): Path where the recording files will be stored.
server_configurations (dict): Configuration settings for connecting to LiveKit.
enable_output_logging (bool): Flag indicating whether to log the recording output data.
"""
self._output_folder = output_folder
self._server_configurations = server_configurations
self._enable_output_logging = enable_output_logging
def _create_start_request(self, recording):
"""Create a request to start a recording in LiveKit.
Args:
recording (Recording): A recording object containing the necessary details.
Returns:
proto_egress.RoomCompositeEgressRequest: The request to start recording in LiveKit.
Raises:
WorkerRequestError: If the file type is unknown or unsupported by LiveKit.
"""
try:
file_details = self.MODE_MAPPINGS[recording.mode]
except KeyError as e:
logger.error("Unknown recording mode for %s: %s", recording.id, e)
raise WorkerRequestError(
f"Unknown recording mode: {recording.mode}"
) from e
filepath = f"{self._output_folder}/{recording.id}.{file_details['extension']}"
# FIXME - align room's name everywhere else in the code
slug = f"{recording.room.id!s}".replace("-", "")
return proto_egress.RoomCompositeEgressRequest(
room_name=slug,
file_outputs=[
proto_egress.EncodedFileOutput(
file_type=file_details["type"],
filepath=filepath,
)
],
**file_details["extra_params"],
)
@async_to_sync
async def start_recording(self, recording):
"""Start a recording session in LiveKit.
Args:
recording (Recording): A recording object containing the necessary details.
Returns:
str: The egress ID for the started recording session.
Raises:
WorkerConnectionError: If there is a connection error with LiveKit.
WorkerResponseError: If the response from LiveKit is invalid or missing data.
"""
async with aiohttp.ClientSession() as session:
client = EgressService(session, **self._server_configurations)
request = self._create_start_request(recording)
try:
response = await client.start_room_composite_egress(start=request)
except TwirpError as e:
raise WorkerConnectionError(
f"LiveKit client connection error, {e.message}."
) from e
if not response.egress_id:
raise WorkerResponseError("Egress ID not found in the response.")
return response.egress_id
@staticmethod
def _create_stop_request(recording):
"""Create a request to stop a recording in LiveKit.
Args:
recording (Recording): A recording object containing the necessary details.
Returns:
proto_egress.StopEgressRequest: The request to stop the recording in LiveKit.
"""
return proto_egress.StopEgressRequest(
egress_id=recording.worker_id,
)
@staticmethod
def _log_recording_data(egress_info):
"""Extract and logs file information from the recording data."""
try:
filename = egress_info.file.filename
started_at = egress_info.file.started_at
ended_at = egress_info.file.ended_at
duration = egress_info.file.duration
# todo - enhance, see with Jacques how and where to log, or log to posthog
logger.info(
"Extracted file info - Filename: %s, Started At: %d, Ended At: %d, Duration: %d",
filename,
started_at or "N/A",
ended_at or "N/A",
duration or "N/A",
)
except:
pass
@async_to_sync
async def stop_recording(self, recording):
"""Stop a recording session in LiveKit.
Args:
recording (Recording): A recording object containing the necessary details.
Returns:
RecordingStatusChoices: The final status of the recording session.
Raises:
WorkerConnectionError: If there is a connection error with LiveKit.
WorkerResponseError: If the response from LiveKit is invalid or missing data.
"""
async with aiohttp.ClientSession() as session:
client = EgressService(session, **self._server_configurations)
request = self._create_stop_request(recording)
try:
response = await client.stop_egress(stop=request)
except TwirpError as e:
raise WorkerConnectionError(
f"LiveKit client connection error, {e.message}."
) from e
if not response.status:
raise WorkerResponseError(
"LiveKit response is missing the recording status."
)
if response.status == proto_egress.EgressStatus.EGRESS_ABORTED:
return RecordingStatusChoices.ABORTED
# FIXME - bulky should be improved
if self._enable_output_logging:
self._log_recording_data(response)
return RecordingStatusChoices.STOPPED
src/backend/core/recording/worker/__init__.py
+13
View File
@@ -0,0 +1,13 @@
"""Meet worker services classes and exceptions."""
from .exceptions import *
from .factories import WorkerServiceFactory
from .mediator import WorkerServiceMediator
from .services import AudioCompositeEgressService, VideoCompositeEgressService
# Expose the worker service factory instance at the module level.
# This provides a centralized access point and abstracts away the service creation
# details from the code using the services.
worker_service_provider = WorkerServiceFactory()
worker_service_provider.register("transcript", AudioCompositeEgressService)
worker_service_provider.register("screen_recording", VideoCompositeEgressService)
src/backend/core/recording/worker/exceptions.py
+21
View File
@@ -0,0 +1,21 @@
"""Recording and worker services specific exceptions."""
class WorkerRequestError(Exception):
"""Raised when there is an issue with the worker request"""
class WorkerConnectionError(Exception):
"""Raised when there is an issue connecting to the worker."""
class WorkerResponseError(Exception):
"""Raised when the worker's response is not as expected."""
class RecordingStartError(Exception):
"""Raised when there is an error starting the recording."""
class RecordingStopError(Exception):
"""Raised when there is an error stopping the recording."""
src/backend/core/recording/worker/factories.py
+90
View File
@@ -0,0 +1,90 @@
"""Factory, configurations and Protocol to create worker services"""
import logging
from dataclasses import dataclass
from functools import lru_cache
from typing import Any, ClassVar, Dict, Optional, Protocol
from django.conf import settings
logger = logging.getLogger(__name__)
@dataclass
class WorkerServiceConfig:
"""Declare Worker Service common configurations"""
output_folder: str
server_configurations: Dict[str, Any]
verify_ssl: Optional[bool]
bucket_args: Optional[dict]
@classmethod
@lru_cache
def from_settings(cls) -> "WorkerServiceConfig":
"""Load configuration from Django settings with caching for efficiency."""
logger.debug("Loading WorkerServiceConfig from settings.")
return cls(
output_folder=settings.RECORDING_OUTPUT_FOLDER,
server_configurations=settings.LIVEKIT_CONFIGURATION,
verify_ssl=settings.RECORDING_VERIFY_SSL,
bucket_args={
"endpoint": settings.AWS_S3_ENDPOINT_URL,
"access_key": settings.AWS_S3_ACCESS_KEY_ID,
"secret": settings.AWS_S3_SECRET_ACCESS_KEY,
"region": settings.AWS_S3_REGION_NAME,
"bucket": settings.AWS_STORAGE_BUCKET_NAME,
"force_path_style": True,
},
)
class WorkerService(Protocol):
"""Define the interface for interacting with a worker service."""
hrid: ClassVar[str]
def __init__(self, config: WorkerServiceConfig):
"""Initialize the service with the given configuration."""
def start(self, room_id: str, recording_id: str) -> str:
"""Start a recording for a specified room."""
def stop(self, worker_id: str) -> str:
"""Stop recording for a specified worker."""
class WorkerServiceFactory:
"""Factory to instantiate worker services based on a specified mode
This factory currently uses a common configuration (`_default_config`) to initialize all
workers. In the future, if workers require different configurations, consider refactoring
to a builder pattern. With a builder pattern, specific builders could be registered per
WorkerService type, allowing each builder to handle instantiation of the worker with its
unique configuration requirements.
"""
def __init__(self):
"""Initialize the WorkerServiceFactory with a default configuration and worker registry."""
self._worker_service_registry = {}
self._default_config = WorkerServiceConfig.from_settings()
def register(self, mode, worker_service: WorkerService):
"""Register a worker service for a specific mode."""
if mode in self._worker_service_registry:
raise KeyError(f"Worker service for mode '{mode}' is already registered.")
self._worker_service_registry[mode] = worker_service
def create(self, mode: str) -> WorkerService:
"""Instantiate a worker service for the specified mode."""
worker_service_cls = self._worker_service_registry.get(mode)
if not worker_service_cls:
raise ValueError(f"Unknown worker service for mode: {mode}.")
return worker_service_cls(config=self._default_config)
src/backend/core/recording/worker/mediator.py
+94
View File
@@ -0,0 +1,94 @@
"""Mediator between the worker service and recording instances in the Django ORM."""
import logging
from core.models import Recording, RecordingStatusChoices
from .exceptions import (
RecordingStartError,
RecordingStopError,
WorkerConnectionError,
WorkerRequestError,
WorkerResponseError,
)
from .factories import WorkerService
logger = logging.getLogger(__name__)
class WorkerServiceMediator:
"""Mediate interactions between a worker service and a recording instance.
This class avoids direct coupling between the worker service and the Django ORM.
It is responsible for updating the recording instance based on the worker service's
status and responses. It also encapsulates worker-related errors into more
user-friendly higher-level exceptions.
This class follows the Mediator design pattern to centralize and coordinate
the communication between the worker service and the recording instances. It's not only
a facade for the worker service, because it adds functionalities to the worker service.
"""
def __init__(self, worker_service: WorkerService):
"""Initialize the WorkerServiceMediator with the provided worker service."""
self._worker_service = worker_service
def start(self, recording: Recording):
"""Start the recording process using the worker service.
Args:
recording (Recording): The recording instance to start.
Raises:
RecordingStartError: If there is an error starting the recording.
"""
# FIXME - no manipulations of room_name should be required
room_name = f"{recording.room.id!s}".replace("-", "")
try:
worker_id = self._worker_service.start(room_name, recording.id)
except (WorkerRequestError, WorkerConnectionError, WorkerResponseError) as e:
logger.error(
"Failed to start recording for room %s: %s", recording.room.slug, e
)
recording.status = RecordingStatusChoices.FAILED_TO_START
raise RecordingStartError() from e
else:
recording.worker_id = worker_id
recording.status = RecordingStatusChoices.ACTIVE
finally:
recording.save()
logger.info(
"Worker started for room %s (worker ID: %s, mode: %s)",
recording.room,
recording.worker_id,
recording.mode,
)
def stop(self, recording: Recording):
"""Stop the recording process using the worker service.
Args:
recording (Recording): The recording instance to stop.
Raises:
RecordingStopError: If there is an error stopping the recording.
"""
try:
response = self._worker_service.stop(worker_id=recording.worker_id)
except (WorkerConnectionError, WorkerResponseError) as e:
logger.error(
"Failed to stop recording for room %s: %s", recording.room.slug, e
)
recording.status = RecordingStatusChoices.FAILED_TO_STOP
raise RecordingStopError() from e
else:
recording.status = RecordingStatusChoices[response]
finally:
recording.save()
logger.info("Worker stopped for room %s", recording.room)
src/backend/core/recording/worker/services.py
+139
View File
@@ -0,0 +1,139 @@
"""Worker services in charge of recording a room."""
# pylint: disable=no-member
import aiohttp
from asgiref.sync import async_to_sync
from livekit import api as livekit_api
from livekit.api.egress_service import EgressService
from .exceptions import WorkerConnectionError, WorkerResponseError
from .factories import WorkerServiceConfig
class BaseEgressService:
"""Base egress defining common method to manage and interact with LiveKit egress processes."""
def __init__(self, config: WorkerServiceConfig):
self._config = config
self._s3 = livekit_api.S3Upload(**config.bucket_args)
def _get_filepath(self, filename, extension):
"""Construct the file path for a given filename and extension."""
return f"{self._config.output_folder}/{filename}.{extension}"
@async_to_sync
async def _handle_request(self, request, method_name: str):
"""Handle making a request to the LiveKit API and returns the response."""
# Use HTTP connector for local development with Tilt,
# where cluster communications are unsecure
connector = aiohttp.TCPConnector(ssl=self._config.verify_ssl)
async with aiohttp.ClientSession(connector=connector) as session:
client = EgressService(session, **self._config.server_configurations)
method = getattr(client, method_name)
# todo - test method, to make sure it exists
try:
response = await method(request)
except livekit_api.TwirpError as e:
raise WorkerConnectionError(
f"LiveKit client connection error, {e.message}."
) from e
return response
def stop(self, worker_id):
"""Stop an ongoing egress worker.
The StopEgressRequest is shared among all types of egress,
so a single implementation in the base class should be sufficient.
"""
request = livekit_api.StopEgressRequest(
egress_id=worker_id,
)
response = self._handle_request(request, "stop_egress")
if not response.status:
raise WorkerResponseError(
"LiveKit response is missing the recording status."
)
# To avoid exposing EgressStatus values and coupling with LiveKit outside of this class,
# the response status is mapped to simpler "ABORTED" or "STOPPED" strings.
if response.status == livekit_api.EgressStatus.EGRESS_ABORTED:
return "ABORTED"
return "STOPPED"
def start(self, room_name, recording_id):
"""Start the egress process for a recording (not implemented in the base class).
Each derived class must implement this method, providing the necessary parameters for
its specific egress type (e.g. audio_only, streaming output).
"""
raise NotImplementedError("Subclass must implement this method.")
class VideoCompositeEgressService(BaseEgressService):
"""Record multiple participant video and audio tracks into a single output '.mp4' file."""
hrid = "video-recording-composite-livekit-egress"
def start(self, room_name, recording_id):
"""Start the video composite egress process for a recording."""
file_type = livekit_api.EncodedFileType.MP4
filepath = self._get_filepath(filename=recording_id, extension="mp4")
file_output = livekit_api.EncodedFileOutput(
file_type=file_type,
filepath=filepath,
s3=self._s3,
)
request = livekit_api.RoomCompositeEgressRequest(
room_name=room_name,
file_outputs=[file_output],
)
response = self._handle_request(request, "start_room_composite_egress")
if not response.egress_id:
raise WorkerResponseError("Egress ID not found in the response.")
return response.egress_id
class AudioCompositeEgressService(BaseEgressService):
"""Record multiple participant audio tracks into a single output '.ogg' file."""
hrid = "audio-recording-composite-livekit-egress"
def start(self, room_name, recording_id):
"""Start the audio composite egress process for a recording."""
file_type = livekit_api.EncodedFileType.OGG
filepath = self._get_filepath(filename=recording_id, extension="ogg")
file_output = livekit_api.EncodedFileOutput(
file_type=file_type,
filepath=filepath,
s3=self._s3,
)
request = livekit_api.RoomCompositeEgressRequest(
room_name=room_name, file_outputs=[file_output], audio_only=True
)
response = self._handle_request(request, "start_room_composite_egress")
if not response.egress_id:
raise WorkerResponseError("Egress ID not found in the response.")
return response.egress_id
src/backend/meet/settings.py
+17 -9
View File
@@ -137,9 +137,6 @@ class Base(Configuration):
environ_name="AWS_STORAGE_BUCKET_NAME",
environ_prefix=None,
)
AWS_ENABLE_STORAGE_HOOK = values.BooleanValue(
False, environ_name="AWS_ENABLE_STORAGE_HOOK", environ_prefix=None
)
# Internationalization
# https://docs.djangoproject.com/en/3.1/topics/i18n/
@@ -404,16 +401,27 @@ class Base(Configuration):
ALLOW_UNREGISTERED_ROOMS = values.BooleanValue(
True, environ_name="ALLOW_UNREGISTERED_ROOMS", environ_prefix=None
)
ENABLE_RECORDING = values.BooleanValue(
False, environ_name="ENABLE_RECORDING", environ_prefix=None
)
LOG_RECORDING_OUTPUT = values.BooleanValue(
False, environ_name="LOG_RECORDING_OUTPUT", environ_prefix=None
)
ANALYTICS_KEY = values.Value(
None, environ_name="ANALYTICS_KEY", environ_prefix=None
)
# Recording settings
RECORDING_ENABLE = values.BooleanValue(
False, environ_name="RECORDING_ENABLE", environ_prefix=None
)
RECORDING_OUTPUT_FOLDER = values.Value(
"recordings", environ_name="RECORDING_OUTPUT_FOLDER", environ_prefix=None
)
RECORDING_VERIFY_SSL = values.BooleanValue(
True, environ_name="RECORDING_VERIFY_SSL", environ_prefix=None
)
RECORDING_STORAGE_EVENT_ENABLE = values.BooleanValue(
False, environ_name="RECORDING_STORAGE_EVENT_ENABLE", environ_prefix=None
)
RECORDING_STORAGE_EVENT_TOKEN = values.Value(
None, environ_name="RECORDING_STORAGE_HOOK_TOKEN", environ_prefix=None
)
# pylint: disable=invalid-name
@property
def ENVIRONMENT(self):