from __future__ import annotations import json import logging from typing import Any, Dict, List from django import forms from django.db.models import Q from django.http import HttpRequest, HttpResponse from django.shortcuts import get_object_or_404, redirect from django.utils import formats, timezone from django.urls import reverse from django.views import View from django.views.generic.detail import SingleObjectMixin from django.views.generic.edit import FormMixin from django.views.generic.list import MultipleObjectMixin from inertia import render from labs.lab_manager_v2.lab_instance import ( LabInstance as LabInstanceManager, get_lab_ready_duration_seconds, ) from labs.lab_manager_v2.lab_spec import LABS_SPEC_DIR from labs.models import ( LabInstance, LabTask as DBLabTask, LabTaskDependency as DBLabTaskDependency, ) from .base import PermissionRequiredAdminMixin logger = logging.getLogger(__name__) class LabInstanceForm(forms.ModelForm): """Form for editing lab instance basics.""" class Meta: model = LabInstance fields = [ "lab_code", "cloud_provider", "region_name", "tidb_version", "status", ] class LabInstanceListView(PermissionRequiredAdminMixin, MultipleObjectMixin, View): """List view for lab instances with filters and search.""" model = LabInstance permission_required = "labs.view_labinstance" paginate_by = 20 page_kwarg = "page" ordering = "-created_at" search_fields = ["lab_code", "unique_tag", "user__email"] component = "lab-instances/Index" def get_queryset(self): queryset = self.model._default_manager.select_related("user").all() if self.ordering: queryset = queryset.order_by(self.ordering) query = self.request.GET.get("q", "").strip() if query: conditions = Q() for field in self.search_fields: conditions |= Q(**{f"{field}__icontains": query}) queryset = queryset.filter(conditions) status = self.request.GET.get("status") valid_status = {choice[0] for choice in LabInstance.LabInstanceStatus.choices} if status in valid_status: queryset = queryset.filter(status=status) access_type = self.request.GET.get("access_type") valid_access = {choice[0] for choice in LabInstance.LabAccessType.choices} if access_type in valid_access: queryset = queryset.filter(access_type=access_type) cloud_provider = self.request.GET.get("cloud_provider", "").strip().lower() if cloud_provider: queryset = queryset.filter(cloud_provider=cloud_provider) region = self.request.GET.get("region_name", "").strip() if region: queryset = queryset.filter(region_name__icontains=region) return queryset def serialize_instance(self, instance: LabInstance) -> Dict[str, Any]: return { "id": instance.id, "user_email": instance.user.email if instance.user_id else None, "user_id": instance.user_id, "lab_code": instance.lab_code, "cloud_provider": instance.cloud_provider, "region_name": instance.region_name, "status": instance.status, "access_type": instance.access_type, "access_detail": instance.access_detail, "created_at": instance.created_at.isoformat(), "debug_url": reverse("labs_admin:lab_instances_detail", args=[instance.id]) + "?tab=debug", } def get(self, request: HttpRequest) -> HttpResponse: self.object_list = self.get_queryset() paginate_by = self.get_paginate_by(self.object_list) if paginate_by: paginator, page, object_list, is_paginated = self.paginate_queryset( self.object_list, paginate_by ) else: paginator = None page = None object_list = self.object_list is_paginated = False instances: List[Dict[str, Any]] = [ self.serialize_instance(inst) for inst in object_list ] pagination: Dict[str, Any] | None = None if paginator and page: pagination = { "page": page.number, "per_page": paginator.per_page, "total": paginator.count, "num_pages": paginator.num_pages, "is_paginated": is_paginated, } return render( request, self.component, { "instances": instances, "filters": { "q": request.GET.get("q", "").strip(), "status": request.GET.get("status"), "access_type": request.GET.get("access_type"), "cloud_provider": request.GET.get("cloud_provider", "") .strip() .lower() or None, "region_name": request.GET.get("region_name", ""), }, "pagination": pagination, "status_choices": list(LabInstance.LabInstanceStatus.values), "access_type_choices": list(LabInstance.LabAccessType.values), "cloud_provider_choices": ["aws", "alicloud"], }, ) class LabInstanceDetailView( PermissionRequiredAdminMixin, FormMixin, SingleObjectMixin, View ): """Detail/Edit view for a single lab instance with tabs.""" model = LabInstance permission_required = "labs.view_labinstance" form_class = LabInstanceForm component = "lab-instances/Detail" def _format_dt(self, dt) -> str | None: """Format datetime for debug tab display.""" if not dt: return None try: dt_utc = timezone.localtime(dt, timezone=timezone.utc) except Exception: dt_utc = dt try: return formats.date_format(dt_utc, format="DATETIME_FORMAT", use_l10n=True) except Exception: return str(dt_utc) def _to_jsonable(self, value: Any) -> Any: """Best-effort conversion to JSON-serializable data for Inertia props.""" try: return json.loads(json.dumps(value, default=str)) except Exception: return str(value) def serialize_instance(self, instance: LabInstance) -> Dict[str, Any]: """Serialize instance for the Lab Info tab.""" return { "id": instance.id, "user_email": instance.user.email if instance.user_id else None, "user_id": instance.user_id, "lab_code": instance.lab_code, "cloud_provider": instance.cloud_provider, "region_name": instance.region_name, "tidb_version": instance.tidb_version, "status": instance.status, "access_type": instance.access_type, "access_detail": instance.access_detail, "created_at": instance.created_at.isoformat(), "ready_at": instance.ready_at.isoformat() if instance.ready_at else None, "start_terminating_at": instance.start_terminating_at.isoformat() if instance.start_terminating_at else None, "terminated_at": instance.terminated_at.isoformat() if instance.terminated_at else None, "exited_at": instance.exited_at.isoformat() if instance.exited_at else None, } def serialize_debug_instance(self, instance: LabInstance) -> Dict[str, Any]: """Serialize instance for the Debug tab.""" return { "id": instance.id, "user": { "id": instance.user_id, "email": instance.user.email if instance.user_id else None, } if instance.user_id else None, "lab_code": instance.lab_code, "cloud_provider": instance.cloud_provider, "region_name": instance.region_name, "tidb_environment": instance.tidb_environment, "tidb_version": instance.tidb_version, "unique_tag": instance.unique_tag, "created_at": self._format_dt(instance.created_at), "ready_at": self._format_dt(instance.ready_at), "scheduled_terminate_at": self._format_dt(instance.scheduled_terminate_at), "exited_at": self._format_dt(instance.exited_at), } def serialize_task(self, task: DBLabTask) -> Dict[str, Any]: """Serialize a task for display.""" return { "id": task.id, "name": task.name, "status": task.status, "retries": task.retries, "comment": task.comment, "payload": task.payload, "started_at": self._format_dt(task.started_at), "finished_at": self._format_dt(task.finished_at), "created_at": self._format_dt(task.created_at), "updated_at": self._format_dt(task.updated_at), } def get_form_kwargs(self) -> Dict[str, Any]: kwargs = super().get_form_kwargs() kwargs["instance"] = self.object return kwargs def build_basic_instance_detail(self, db_lab_instance: LabInstance) -> Dict[str, Any]: creation_progress = None estimated_creation_duration_seconds = None now = timezone.now() if db_lab_instance.scheduled_terminate_at: remaining_seconds = max( int((db_lab_instance.scheduled_terminate_at - now).total_seconds()), 0, ) else: remaining_seconds = None if ( db_lab_instance.status == LabInstance.LabInstanceStatus.CREATING and db_lab_instance.create_job_id ): creation_progress = { "total": DBLabTask.objects.filter( lab_job=db_lab_instance.create_job ).count(), "completed": DBLabTask.objects.filter( lab_job=db_lab_instance.create_job, status=DBLabTask.StatusChoices.SUCCEEDED, ).count(), } estimated_creation_duration_seconds = get_lab_ready_duration_seconds( db_lab_instance.lab_code ) return { "id": db_lab_instance.id, "lab_code": db_lab_instance.lab_code, "cloud_provider": db_lab_instance.cloud_provider, "region_name": db_lab_instance.region_name, "tidb_version": db_lab_instance.tidb_version, "tidb_environment": db_lab_instance.tidb_environment, "status": db_lab_instance.get_user_status(), "created_at": db_lab_instance.created_at, "ready_at": db_lab_instance.ready_at, "exited_at": db_lab_instance.exited_at, "creation_progress": creation_progress, "estimated_creation_duration_seconds": estimated_creation_duration_seconds, "remaining_seconds": remaining_seconds, } def get_debug_context(self, db_lab_instance: LabInstance) -> Dict[str, Any]: """Build the context for the debug tab.""" instance_detail = self.build_basic_instance_detail(db_lab_instance) environment_variables: dict[str, Any] = {} runtime_resources: list[Any] = [] lab_spec_path = LABS_SPEC_DIR / db_lab_instance.lab_code / "manifest.yaml" lab_spec_available = lab_spec_path.exists() lab_spec_error = None if lab_spec_available: try: lab_instance = LabInstanceManager(db_lab_instance) instance_detail = lab_instance.get_detail() runtime_resources = lab_instance.get_resources() environment_variables = lab_instance.get_environment_variables() except Exception as e: logger.exception( "Failed to build debug runtime info for lab instance %s", db_lab_instance.id, ) lab_spec_error = str(e) else: lab_spec_error = f"Lab spec not found: {db_lab_instance.lab_code}" # Creation job tasks and dependencies (if any) create_job = db_lab_instance.create_job if create_job: create_tasks = DBLabTask.objects.filter(lab_job=create_job).order_by( "created_at" ) create_task_dependencies = DBLabTaskDependency.objects.filter( lab_job=create_job ) else: create_tasks = DBLabTask.objects.none() create_task_dependencies = DBLabTaskDependency.objects.none() # Build dependency edges for create job task_id_to_name = {t.id: t.name for t in create_tasks} create_task_edges = [ { "task_id": dep.lab_task_id, "task_name": task_id_to_name.get(dep.lab_task_id, str(dep.lab_task_id)), "depends_on_id": dep.depends_on_id, "depends_on_name": task_id_to_name.get( dep.depends_on_id, str(dep.depends_on_id) ), } for dep in create_task_dependencies ] failed_create_tasks = [ t for t in create_tasks if t.status == DBLabTask.StatusChoices.FAILED ] # Termination job tasks and dependencies (if any) terminate_job = db_lab_instance.terminate_job if terminate_job: terminate_tasks = DBLabTask.objects.filter(lab_job=terminate_job).order_by( "created_at" ) terminate_task_dependencies = DBLabTaskDependency.objects.filter( lab_job=terminate_job ) else: terminate_tasks = DBLabTask.objects.none() terminate_task_dependencies = DBLabTaskDependency.objects.none() term_task_id_to_name = {t.id: t.name for t in terminate_tasks} terminate_task_edges = [ { "task_id": dep.lab_task_id, "task_name": term_task_id_to_name.get( dep.lab_task_id, str(dep.lab_task_id) ), "depends_on_id": dep.depends_on_id, "depends_on_name": term_task_id_to_name.get( dep.depends_on_id, str(dep.depends_on_id) ), } for dep in terminate_task_dependencies ] failed_terminate_tasks = [ t for t in terminate_tasks if t.status == DBLabTask.StatusChoices.FAILED ] return { "debug_instance": self.serialize_debug_instance(db_lab_instance), "instance_detail": self._to_jsonable(instance_detail), "environment_variables": self._to_jsonable(environment_variables), "runtime_resources": self._to_jsonable(runtime_resources), "lab_spec_available": lab_spec_available, "lab_spec_error": lab_spec_error, "create_tasks": [self.serialize_task(t) for t in create_tasks], "create_task_edges": create_task_edges, "failed_create_tasks": [ self.serialize_task(t) for t in failed_create_tasks ], "terminate_tasks": [self.serialize_task(t) for t in terminate_tasks], "terminate_task_edges": terminate_task_edges, "failed_terminate_tasks": [ self.serialize_task(t) for t in failed_terminate_tasks ], } def get(self, request: HttpRequest, pk: int) -> HttpResponse: self.object = self.get_object() tab = request.GET.get("tab", "info") context = { "instance": self.serialize_instance(self.object), "tab": tab, "status_choices": list(LabInstance.LabInstanceStatus.values), } # Only load debug data when the debug tab is active if tab == "debug": context.update(self.get_debug_context(self.object)) return render(request, self.component, context) def post(self, request: HttpRequest, pk: int) -> HttpResponse: if not request.user.has_perm("labs.change_labinstance"): return redirect(f"/admin/lab-instances/{pk}/") self.object = self.get_object() form = self.get_form() if form.is_valid(): return self.form_valid(form) return self.form_invalid(form) def form_valid(self, form: forms.ModelForm) -> HttpResponse: form.save() self.request.session["flash_success"] = "Lab instance updated" return redirect(f"/admin/lab-instances/{self.object.id}/") def form_invalid(self, form: forms.ModelForm) -> HttpResponse: return render( self.request, self.component, { "instance": self.serialize_instance(self.object), "tab": "info", "errors": form.errors, "form": form.data, "status_choices": list(LabInstance.LabInstanceStatus.values), }, )