feat(OBSERV-02): KPI analytics dashboard — active users, conversations & messages

.vscode/fred.code-workspace

@@ -40,7 +40,7 @@
 			"../libs/fred-core/": true,
 		},
 		"i18n-ally.localesPaths": [
-			"frontend/src/locales",
+			"apps/frontend/src/locales",
 		],
 		"i18n-ally.pathMatcher": "{locale}/translation.json",
 		"i18n-ally.keystyle": "nested",

apps/control-plane-backend/config/.env.template

@@ -1,5 +1,7 @@
 # Copy to .env and adapt if needed
+
 CONFIG_FILE=./config/configuration_prod.yaml
+
 # Required only when security.m2m.enabled=true
 KEYCLOAK_CONTROL_PLANE_CLIENT_SECRET=""
 # Required only when security.rebac.enabled=true
@@ -8,3 +10,5 @@ OPENFGA_API_TOKEN=""
 MINIO_SECRET_KEY=""
 # Postgres credentials
 FRED_POSTGRES_PASSWORD=""
+# Opensearch credentials
+OPENSEARCH_PASSWORD=""

apps/control-plane-backend/config/configuration.yaml

@@ -5,13 +5,19 @@ app:
   port: 8222
   log_level: info
   gcu_version: "v1"
-  # Production defaults: scrape via Prometheus/Grafana and avoid KPI log noise.
-  kpi_process_metrics_interval_sec: 0
-  kpi_log_summary_interval_sec: 0.0
-  kpi_log_summary_top_n: 0
   default_team_max_resources_storage_size: 5368709120
   personal_max_resources_storage_size: 5368709120
 
+observability:
+  kpi:
+    log:
+      enabled: true
+    prometheus:
+      enabled: false
+    opensearch:
+      enabled: false
+    process_metrics_interval_sec: 0
+
 scheduler:
   enabled: false
   # temporal: requires Temporal server + run-worker process

apps/control-plane-backend/config/configuration_prod.yaml

@@ -4,10 +4,6 @@ app:
   address: 0.0.0.0
   port: 8222
   log_level: debug
-  # Production defaults: scrape via Prometheus/Grafana and avoid KPI log noise.
-  kpi_process_metrics_interval_sec: 0
-  kpi_log_summary_interval_sec: 0.0
-  kpi_log_summary_top_n: 0
   default_team_max_resources_storage_size: 5368709120
   personal_max_resources_storage_size: 5368709120
 
@@ -23,6 +19,11 @@ scheduler:
     workflow_id_prefix: control-plane
     connect_timeout_seconds: 5
 
+observability:
+  kpi:
+    prometheus:
+      port: 9222
+
 security:
   m2m:
     enabled: true
@@ -46,6 +47,11 @@ storage:
     port: 5432
     database: fred
     username: fred
+  opensearch:
+    host: "https://localhost:9200"
+    username: "admin"
+    secure: true
+    verify_certs: false
   content_storage:
     type: minio
     endpoint: http://localhost:8333

apps/control-plane-backend/config/configuration_test.yaml

@@ -7,6 +7,16 @@ app:
   default_team_max_resources_storage_size: 5368709120
   personal_max_resources_storage_size: 5368709120
 
+observability:
+  kpi:
+    log:
+      enabled: true
+    prometheus:
+      enabled: false
+    opensearch:
+      enabled: false
+    process_metrics_interval_sec: 0
+
 scheduler:
   enabled: false
   # temporal: requires Temporal server + run-worker process

apps/control-plane-backend/control_plane_backend/app/context.py

@@ -9,6 +9,8 @@
     RebacEngine,
     rebac_factory,
 )
+from fred_core.kpi.base_kpi_writer import BaseKPIWriter
+from fred_core.kpi.kpi_factory import build_kpi_writer
 from fred_core.scheduler import (
     SchedulerBackend,
     TemporalClientProvider,
@@ -18,6 +20,7 @@
 from fred_core.store import ContentStore, LocalContentStore, MinioContentStore
 from fred_core.tasks.service import TaskService
 from fred_core.teams.metadata_store import TeamMetadataStore
+from prometheus_client import start_http_server
 from sqlalchemy.ext.asyncio import AsyncEngine
 
 from control_plane_backend.agent_instances.store import AgentInstanceStore
@@ -47,6 +50,7 @@ def __init__(self, configuration: Configuration):
         self._policy_catalog: ConversationPolicyCatalog | None = None
         self._policy_catalog_path = self._resolve_policy_catalog_path()
         self._pg_async_engine: AsyncEngine | None = None
+        self._kpi_writer: BaseKPIWriter | None = None
         self._session_store: BaseSessionStore | None = None
         self._purge_queue_store: PurgeQueueStore | None = None
         self._team_metadata_store: TeamMetadataStore | None = None
@@ -108,6 +112,41 @@ def get_pg_async_engine(self) -> AsyncEngine:
             )
         return self._pg_async_engine
 
+    def get_kpi_writer(self) -> BaseKPIWriter:
+        if self._kpi_writer is None:
+            self._kpi_writer = build_kpi_writer(
+                kpi_config=self.configuration.observability.kpi,
+                opensearch_config=self.configuration.storage.opensearch,
+                service_name="control-plane",
+                log_level=self.configuration.app.log_level,
+            )
+        return self._kpi_writer
+
+    def get_kpi_store(self):  # -> OpenSearchKPIStore | None
+        from fred_core.kpi.kpi_writer import KPIWriter
+        from fred_core.kpi.opensearch_kpi_store import OpenSearchKPIStore
+        from fred_core.kpi.prometheus_kpi_store import PrometheusKPIStore
+
+        writer = self.get_kpi_writer()
+        if not isinstance(writer, KPIWriter):
+            return None
+        store = writer.store
+        if isinstance(store, PrometheusKPIStore):
+            store = store._delegate
+        return store if isinstance(store, OpenSearchKPIStore) else None
+
+    def start_metrics_exporter(self) -> None:
+        """Start the Prometheus scrape endpoint when configured."""
+        prom_cfg = self.configuration.observability.kpi.prometheus
+        if not prom_cfg.enabled:
+            return
+        start_http_server(prom_cfg.port, addr=prom_cfg.address)
+        logger.info(
+            "[control-plane] Prometheus metrics exporter ready at %s:%s",
+            prom_cfg.address,
+            prom_cfg.port,
+        )
+
     def get_session_store(self) -> BaseSessionStore:
         if self._session_store is None:
             self._session_store = PostgresSessionStore(

apps/control-plane-backend/control_plane_backend/config/models.py

@@ -8,6 +8,8 @@
     SecurityConfiguration,
 )
 from fred_core.common import (
+    KpiObservabilityConfig,
+    OpenSearchStoreConfig,
     PostgresStoreConfig,
     TemporalSchedulerConfig,
 )
@@ -33,6 +35,10 @@ class AppConfig(BaseModel):
     )
 
 
+class ObservabilityConfig(BaseModel):
+    kpi: KpiObservabilityConfig = Field(default_factory=KpiObservabilityConfig)
+
+
 class FrontendFeatureFlags(BaseModel):
     """Typed feature flags exposed to the frontend bootstrap."""
 
@@ -259,6 +265,7 @@ class StorageConfig(BaseModel):
     content_storage: ContentStorageConfig = Field(
         default_factory=_default_content_storage
     )
+    opensearch: Optional[OpenSearchStoreConfig] = None
 
 
 class Configuration(BaseModel):
@@ -267,6 +274,7 @@ class Configuration(BaseModel):
     scheduler: SchedulerConfig
     security: SecurityConfiguration = Field(default_factory=_default_security)
     storage: StorageConfig = Field(default_factory=StorageConfig)
+    observability: ObservabilityConfig = Field(default_factory=ObservabilityConfig)
     policies: PolicyConfig = Field(default_factory=PolicyConfig)
 
 

apps/control-plane-backend/control_plane_backend/kpi/README.md

@@ -0,0 +1,81 @@
+# KPI presets
+
+Each KPI preset is a self-contained query that runs against OpenSearch and returns
+structured data. Presets are auto-registered as GET endpoints under `/kpi/presets/<name>`.
+
+## How it works
+
+```
+api.py          — iterates PRESETS, mounts one route per preset
+presets/
+  __init__.py   — PRESETS list (add your preset here)
+  base.py       — PresetDef dataclass
+  common.py     — shared response types (TimeSeriesResponse, …)
+  <name>.py     — one file per preset
+utils.py        — resolve_interval(): picks OpenSearch bucket size from time range
+```
+
+Every preset is a `PresetDef`:
+
+```python
+PresetDef(
+    name="my_preset",          # becomes GET /kpi/presets/my_preset
+    response_model=MyResponse, # Pydantic model — drives OpenAPI schema
+    handler=query_my_preset,   # called with (store, user=…, since=…, until=…)
+    summary="One-line description for OpenAPI docs",
+)
+```
+
+The handler receives:
+- `store: OpenSearchKPIStore` — call `store.client.search(index=store.index, body=…)`
+- `user: KeycloakUser` — call `require_admin(user)` if admin-only
+- `since / until: datetime` — the requested time range (UTC, always set)
+
+## Adding a preset
+
+1. Create `presets/my_preset.py`. Define a Pydantic response model and a handler
+   function. Use `TimeSeriesResponse` from `common.py` for time-bucketed data, or
+   define a custom model if the shape doesn't fit.
+
+2. Use `resolve_interval(since, until)` from `utils.py` to get the right OpenSearch
+   bucket interval and `strftime` format for the time range.
+
+3. Register in `presets/__init__.py`:
+
+```python
+from control_plane_backend.kpi.presets.my_preset import MY_PRESET
+
+PRESETS: list[PresetDef] = [
+    ACTIVE_USERS_OVER_TIME_PRESET,
+    MY_PRESET,               # add here
+]
+```
+
+4. Regenerate the frontend types: `cd apps/frontend && make update-control-plane-api`
+
+## Common response types
+
+**`TimeSeriesResponse`** (`common.py`) — use for any time-bucketed metric:
+
+```python
+TimeSeriesResponse(
+    rows=[TimeSeriesPoint(date="2026-06-12", value=42.0), …],
+    since=since,   # AwareDatetime, passed through from the handler
+    until=until,
+    interval="1d", # the OpenSearch fixed_interval used
+)
+```
+
+The frontend `TimeSeriesLineChart` molecule consumes this shape directly.
+
+**`ScalarResponse`** (`common.py`) — use for any single integer metric over a time range:
+
+```python
+ScalarResponse(
+    value=42,
+    since=since,   # AwareDatetime, passed through from the handler
+    until=until,
+)
+```
+
+The frontend `KpiStatCard` molecule consumes this shape directly.

apps/control-plane-backend/control_plane_backend/kpi/api.py

@@ -0,0 +1,71 @@
+# Copyright Thales 2026
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from fred_core import KeycloakUser, get_current_user
+from fred_core.kpi.opensearch_kpi_store import OpenSearchKPIStore
+
+from control_plane_backend.app.dependencies import get_application_container
+from control_plane_backend.kpi.presets import PRESETS
+
+
+def get_kpi_store(request: Request) -> OpenSearchKPIStore:
+    container = get_application_container(request)
+    store = container.get_kpi_store()
+    if store is None:
+        raise HTTPException(status_code=503, detail="KPI store not available")
+    return store
+
+
+def build_kpi_router() -> APIRouter:
+    router = APIRouter(prefix="/kpi", tags=["KPI"])
+
+    for preset in PRESETS:
+
+        def make_handler(p=preset):
+            async def handler(
+                since: datetime | None = Query(
+                    default=None,
+                    description="Start of the time range (ISO 8601 datetime). Defaults to 30 days ago.",
+                ),
+                until: datetime | None = Query(
+                    default=None,
+                    description="End of the time range (ISO 8601 datetime). Defaults to now.",
+                ),
+                user: KeycloakUser = Depends(get_current_user),
+                store: OpenSearchKPIStore = Depends(get_kpi_store),
+            ):
+                now = datetime.now(tz=timezone.utc)
+                resolved_since = since or (now - timedelta(days=30))
+                resolved_until = until or now
+                return p.handler(
+                    store, user=user, since=resolved_since, until=resolved_until
+                )
+
+            return handler
+
+        router.add_api_route(
+            f"/presets/{preset.name}",
+            make_handler(),
+            methods=["GET"],
+            response_model=preset.response_model,
+            summary=preset.summary,
+            response_model_exclude_none=True,
+        )
+
+    return router

apps/control-plane-backend/control_plane_backend/kpi/presets/__init__.py

@@ -0,0 +1,38 @@
+# Copyright Thales 2026
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+from control_plane_backend.kpi.presets.active_users_over_time import (
+    ACTIVE_USERS_OVER_TIME_PRESET,
+)
+from control_plane_backend.kpi.presets.base import PresetDef
+from control_plane_backend.kpi.presets.messages_over_time import (
+    MESSAGES_OVER_TIME_PRESET,
+)
+from control_plane_backend.kpi.presets.sessions_over_time import (
+    SESSIONS_OVER_TIME_PRESET,
+)
+from control_plane_backend.kpi.presets.unique_users_total import (
+    UNIQUE_USERS_TOTAL_PRESET,
+)
+
+PRESETS: list[PresetDef] = [
+    ACTIVE_USERS_OVER_TIME_PRESET,
+    UNIQUE_USERS_TOTAL_PRESET,
+    SESSIONS_OVER_TIME_PRESET,
+    MESSAGES_OVER_TIME_PRESET,
+]
+
+__all__ = ["PRESETS", "PresetDef"]