🚸 Add DIVA reporter health status to validation UI

Author: agmangas

moderate_api/config.py

@@ -100,6 +100,10 @@ def url_report(self, validator: str | None = None) -> str:
             return f"{base_url}?{urlencode({'validator': validator})}"
         return base_url
 
+    def url_report_status(self) -> str:
+        """URL for fetching reporter health status."""
+        return self.build_reporter_url("report", "status")
+
 
 class Settings(BaseSettings):
     class Config:

moderate_api/diva.py

@@ -101,6 +101,35 @@ def from_entries(
         )
 
 
+class ReporterState(str, Enum):
+    """State of the DIVA Quality Reporter service."""
+
+    STARTING = "starting"
+    HEALTHY = "healthy"
+    CATCHING_UP = "catching_up"
+    STALE = "stale"
+    ERROR = "error"
+
+
+class ReporterStatus(BaseModel):
+    """Health status of the DIVA Quality Reporter."""
+
+    state: ReporterState
+    kafka_connected: bool
+    consumer_group: str | None = None
+    last_poll_at: float | None = None
+    last_message_at: float | None = None
+    last_flush_at: float | None = None
+    seconds_since_last_flush: float | None = None
+    messages_processed_total: int = 0
+    current_batch_size: int = 0
+    validation_pending_messages: int | None = None
+    validation_pending_by_partition: dict[str, int] = Field(default_factory=dict)
+    reconnect_count: int = 0
+    last_error: str | None = None
+    is_mock: bool = False
+
+
 class DivaClient:
     """Client for DIVA Kafka REST Gateway and Quality Reporter APIs.
 
@@ -377,3 +406,34 @@ async def get_validation_results(
             processed_rows=processed_rows,
             last_requested_at=start_time,
         )
+
+    async def get_reporter_status(self) -> ReporterStatus:
+        """Fetch health status from DIVA Quality Reporter.
+
+        Returns:
+            ReporterStatus with current health state. Returns ERROR state
+            on any failure instead of raising.
+        """
+        url = self.settings.url_report_status()
+
+        _logger.debug("Fetching reporter status from DIVA: url=%s", url)
+
+        try:
+            async with httpx.AsyncClient(
+                timeout=self.settings.request_timeout
+            ) as client:
+                response = await client.get(url, auth=self._auth)
+                response.raise_for_status()
+                data = response.json()
+            return ReporterStatus(**data)
+        except (
+            httpx.HTTPStatusError,
+            httpx.RequestError,
+            Exception,
+        ) as e:
+            _logger.error("Failed to fetch reporter status: %s", str(e))
+            return ReporterStatus(
+                state=ReporterState.ERROR,
+                kafka_connected=False,
+                last_error=str(e),
+            )

moderate_api/diva_mock.py

@@ -13,6 +13,8 @@
 from moderate_api.config import DivaSettings
 from moderate_api.diva import (
     DivaClient,
+    ReporterState,
+    ReporterStatus,
     ValidationEntry,
     ValidationResult,
     ValidationStatus,
@@ -239,6 +241,19 @@ def _generate_mock_entries(
 
         return entries
 
+    async def get_reporter_status(self) -> ReporterStatus:
+        """Return a healthy mock reporter status.
+
+        Returns:
+            ReporterStatus indicating healthy mock state
+        """
+        return ReporterStatus(
+            state=ReporterState.HEALTHY,
+            kafka_connected=True,
+            consumer_group="mock-consumer-group",
+            is_mock=True,
+        )
+
     @classmethod
     def reset_mock_state(cls) -> None:
         """Reset all mock validation states.

moderate_api/entities/asset/router.py

@@ -26,7 +26,11 @@
 from moderate_api.authz.user import OptionalUserDep, User, UserDep
 from moderate_api.config import SettingsDep
 from moderate_api.db import AsyncSessionDep
-from moderate_api.diva import ValidationResult, ValidationStatus
+from moderate_api.diva import (
+    ReporterStatus,
+    ValidationResult,
+    ValidationStatus,
+)
 from moderate_api.diva_deps import DivaClientDep
 from moderate_api.entities.asset import trust_routes
 from moderate_api.entities.asset.models import (
@@ -996,6 +1000,24 @@ async def get_supported_extensions(
     return settings.diva.supported_extensions
 
 
+@router.get(
+    "/validation/reporter-status",
+    response_model=ReporterStatus,
+    tags=[_TAG],
+    summary="Get Quality Reporter health status",
+)
+async def get_reporter_status(
+    user: UserDep,
+    diva: DivaClientDep,
+) -> ReporterStatus:
+    """Fetch the current health status of the DIVA Quality Reporter.
+
+    Returns telemetry including state machine status, Kafka connectivity,
+    processing lag, and pending messages.
+    """
+    return await diva.get_reporter_status()
+
+
 @router.post(
     "/{id}/object/{object_id}/validate",
     response_model=StartValidationResponse,

moderate_ui/public/locales/en/common.json

@@ -129,5 +129,28 @@
       "includeMine": "Toggle this to view private datasets you have uploaded, in addition to public ones.",
       "fileFormat": "Filter results to show only specific file types (e.g., CSV, JSON)."
     }
+  },
+  "reporter": {
+    "healthy": "Validation ready",
+    "healthyTooltip": "Data Quality service is available and validation can run now.",
+    "starting": "Validation starting",
+    "startingTooltip": "Data Quality service is starting up. Validation requests may take a moment.",
+    "catchingUp": "Validation delayed",
+    "catchingUpTooltip": "Data Quality service is catching up on queued work. New validation requests may take longer.",
+    "stale": "Validation delayed",
+    "staleTitle": "Validation delayed",
+    "staleTooltip": "Data Quality service has not processed recent work. Validation results may be delayed.",
+    "staleMessage": "The Data Quality service has not processed recent work. Validation results may be delayed.",
+    "error": "Validation unavailable",
+    "errorTitle": "Validation unavailable",
+    "errorTooltip": "Data Quality service is currently unavailable. Starting a new validation may fail.",
+    "errorMessage": "The Data Quality service is currently unavailable. Starting a new validation may fail.",
+    "kafkaConnected": "Kafka Connected",
+    "kafkaDisconnected": "Kafka Disconnected"
+  },
+  "validation": {
+    "serviceStatusLabel": "Service status",
+    "serviceUnavailableTitle": "Validation temporarily unavailable",
+    "serviceUnavailableMessage": "Validation is temporarily unavailable because the Data Quality service is offline."
   }
 }

moderate_ui/src/api/validation.ts

@@ -61,6 +61,36 @@ export interface AssetObjectRowCountResponse {
   estimated: boolean;
 }
 
+/**
+ * State of the DIVA Quality Reporter service.
+ */
+export type ReporterState =
+  | "starting"
+  | "healthy"
+  | "catching_up"
+  | "stale"
+  | "error";
+
+/**
+ * Health status of the DIVA Quality Reporter.
+ */
+export interface ReporterStatus {
+  state: ReporterState;
+  kafka_connected: boolean;
+  consumer_group?: string;
+  last_poll_at?: number;
+  last_message_at?: number;
+  last_flush_at?: number;
+  seconds_since_last_flush?: number;
+  messages_processed_total: number;
+  current_batch_size: number;
+  validation_pending_messages?: number;
+  validation_pending_by_partition: Record<string, number>;
+  reconnect_count: number;
+  last_error?: string;
+  is_mock?: boolean;
+}
+
 /**
  * Add computed stats to validation entries.
  */
@@ -157,3 +187,12 @@ export async function getAssetObjectRowCount({
   const response = await axios.get(url);
   return response.data;
 }
+
+/**
+ * Get the current health status of the DIVA Quality Reporter.
+ */
+export async function getReporterStatus(): Promise<ReporterStatus> {
+  const url = buildApiUrl("asset", "validation", "reporter-status");
+  const response = await axios.get(url);
+  return response.data;
+}