235 lines
9.0 KiB
Python
235 lines
9.0 KiB
Python
"""Low-cardinality operational metrics for the Dify side of KnowledgeFS.
|
|
|
|
The metric contracts intentionally exclude tenant, resource, principal, token, URL, and free-form
|
|
error values. Callers own best-effort failure isolation so telemetry can never change product state.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable, Iterable, Mapping
|
|
from functools import lru_cache
|
|
from typing import Literal, NamedTuple, Protocol
|
|
|
|
from opentelemetry.metrics import Observation, get_meter
|
|
|
|
|
|
class KnowledgeFSCapabilityIssuanceMetric(NamedTuple):
|
|
caller_kind: str
|
|
operation_id: str
|
|
outcome: Literal["denied", "failed", "issued"]
|
|
reason: str
|
|
|
|
|
|
class KnowledgeFSBatchStatusMetric(NamedTuple):
|
|
duration_seconds: float
|
|
missing_spaces: int
|
|
outcome: Literal["degraded", "failed", "success"]
|
|
requested_spaces: int
|
|
returned_spaces: int
|
|
|
|
|
|
class KnowledgeFSControlSpaceStateMetric(NamedTuple):
|
|
duration_seconds: float
|
|
from_state: str
|
|
to_state: str
|
|
|
|
|
|
class KnowledgeFSLifecycleTaskMetric(NamedTuple):
|
|
duration_seconds: float | None
|
|
operation: str
|
|
status: Literal["dispatch_error", "queued", "retry", "running", "succeeded"]
|
|
|
|
|
|
class KnowledgeFSOperationAdmissionMetric(NamedTuple):
|
|
operation_id: str
|
|
bucket: str
|
|
phase: Literal["commit", "refund", "reserve"]
|
|
outcome: Literal["failure", "success"]
|
|
|
|
|
|
class KnowledgeFSOperationalMetricsPort(Protocol):
|
|
def record_batch_status(self, event: KnowledgeFSBatchStatusMetric) -> None: ...
|
|
|
|
def record_capability_issuance(self, event: KnowledgeFSCapabilityIssuanceMetric) -> None: ...
|
|
|
|
def record_control_space_state(self, event: KnowledgeFSControlSpaceStateMetric) -> None: ...
|
|
|
|
def record_lifecycle_task(self, event: KnowledgeFSLifecycleTaskMetric) -> None: ...
|
|
|
|
def record_operation_admission(self, event: KnowledgeFSOperationAdmissionMetric) -> None: ...
|
|
|
|
def register_control_space_state_gauge(self, read_counts: Callable[[], Mapping[str, int]]) -> None: ...
|
|
|
|
|
|
class _CounterPort(Protocol):
|
|
def add(self, amount: int, *, attributes: dict[str, str]) -> None: ...
|
|
|
|
|
|
class _HistogramPort(Protocol):
|
|
def record(self, amount: float, *, attributes: dict[str, str]) -> None: ...
|
|
|
|
|
|
class _MeterPort(Protocol):
|
|
def create_counter(self, name: str, *, description: str, unit: str) -> _CounterPort: ...
|
|
|
|
def create_histogram(self, name: str, *, description: str, unit: str) -> _HistogramPort: ...
|
|
|
|
def create_observable_gauge(
|
|
self,
|
|
name: str,
|
|
*,
|
|
callbacks: tuple[Callable[[object], Iterable[Observation]], ...],
|
|
description: str,
|
|
unit: str,
|
|
) -> object: ...
|
|
|
|
|
|
class OpenTelemetryKnowledgeFSOperationalMetrics:
|
|
"""Map sanitized events to OpenTelemetry counters and duration histograms."""
|
|
|
|
def __init__(self, *, meter: _MeterPort | None = None) -> None:
|
|
resolved_meter = meter or get_meter("dify.knowledge_fs")
|
|
self._meter = resolved_meter
|
|
self._control_space_state_gauge: object | None = None
|
|
self._capability_issuance = resolved_meter.create_counter(
|
|
"dify.knowledge_fs.capability_issuance",
|
|
description="KnowledgeFS capability issuance outcomes",
|
|
unit="{issuance}",
|
|
)
|
|
self._batch_requests = resolved_meter.create_counter(
|
|
"dify.knowledge_fs.batch_status_requests",
|
|
description="KnowledgeFS batch status request outcomes",
|
|
unit="{request}",
|
|
)
|
|
self._batch_spaces = resolved_meter.create_counter(
|
|
"dify.knowledge_fs.batch_status_spaces",
|
|
description="KnowledgeFS batch status returned and missing spaces",
|
|
unit="{space}",
|
|
)
|
|
self._batch_latency = resolved_meter.create_histogram(
|
|
"dify.knowledge_fs.batch_status_latency",
|
|
description="KnowledgeFS batch status latency",
|
|
unit="s",
|
|
)
|
|
self._control_space_transitions = resolved_meter.create_counter(
|
|
"dify.knowledge_fs.control_space_transitions",
|
|
description="KnowledgeFS control-space lifecycle transitions",
|
|
unit="{transition}",
|
|
)
|
|
self._control_space_state_duration = resolved_meter.create_histogram(
|
|
"dify.knowledge_fs.control_space_state_duration",
|
|
description="Time spent in a KnowledgeFS control-space state before transition",
|
|
unit="s",
|
|
)
|
|
self._lifecycle_tasks = resolved_meter.create_counter(
|
|
"dify.knowledge_fs.lifecycle_tasks",
|
|
description="KnowledgeFS durable lifecycle task state observations",
|
|
unit="{task}",
|
|
)
|
|
self._lifecycle_task_latency = resolved_meter.create_histogram(
|
|
"dify.knowledge_fs.lifecycle_task_latency",
|
|
description="KnowledgeFS durable lifecycle task terminal latency",
|
|
unit="s",
|
|
)
|
|
self._revoke_latency = resolved_meter.create_histogram(
|
|
"dify.knowledge_fs.revoke_latency",
|
|
description="KnowledgeFS capability revoke enqueue-to-ack latency",
|
|
unit="s",
|
|
)
|
|
self._operation_admission = resolved_meter.create_counter(
|
|
"dify.knowledge_fs.operation_admission",
|
|
description="KnowledgeFS direct-operation reserve and finalization outcomes",
|
|
unit="{operation}",
|
|
)
|
|
|
|
def record_capability_issuance(self, event: KnowledgeFSCapabilityIssuanceMetric) -> None:
|
|
self._capability_issuance.add(
|
|
1,
|
|
attributes={
|
|
"caller_kind": event.caller_kind,
|
|
"operation_id": event.operation_id,
|
|
"outcome": event.outcome,
|
|
"reason": event.reason,
|
|
},
|
|
)
|
|
|
|
def record_batch_status(self, event: KnowledgeFSBatchStatusMetric) -> None:
|
|
outcome: dict[str, str] = {"outcome": event.outcome}
|
|
self._batch_requests.add(1, attributes=outcome)
|
|
self._batch_latency.record(event.duration_seconds, attributes=outcome)
|
|
self._batch_spaces.add(event.returned_spaces, attributes={"result": "returned"})
|
|
self._batch_spaces.add(event.missing_spaces, attributes={"result": "missing"})
|
|
|
|
def record_control_space_state(self, event: KnowledgeFSControlSpaceStateMetric) -> None:
|
|
attributes = {"from_state": event.from_state, "to_state": event.to_state}
|
|
self._control_space_transitions.add(1, attributes=attributes)
|
|
self._control_space_state_duration.record(event.duration_seconds, attributes=attributes)
|
|
|
|
def record_lifecycle_task(self, event: KnowledgeFSLifecycleTaskMetric) -> None:
|
|
attributes = {"operation": event.operation, "status": event.status}
|
|
self._lifecycle_tasks.add(1, attributes=attributes)
|
|
if event.duration_seconds is None:
|
|
return
|
|
self._lifecycle_task_latency.record(event.duration_seconds, attributes=attributes)
|
|
if event.operation == "revoke" and event.status == "succeeded":
|
|
self._revoke_latency.record(event.duration_seconds, attributes={"operation": "revoke"})
|
|
|
|
def record_operation_admission(self, event: KnowledgeFSOperationAdmissionMetric) -> None:
|
|
self._operation_admission.add(
|
|
1,
|
|
attributes={
|
|
"bucket": event.bucket,
|
|
"operation_id": event.operation_id,
|
|
"outcome": event.outcome,
|
|
"phase": event.phase,
|
|
},
|
|
)
|
|
|
|
def register_control_space_state_gauge(self, read_counts: Callable[[], Mapping[str, int]]) -> None:
|
|
"""Register one DB-backed current-state instrument per process."""
|
|
|
|
if self._control_space_state_gauge is not None:
|
|
return
|
|
|
|
def observe(_: object) -> Iterable[Observation]:
|
|
try:
|
|
counts = read_counts()
|
|
return tuple(
|
|
Observation(
|
|
max(0, counts.get(state, 0)),
|
|
attributes={"aggregation_scope": "global_database_snapshot", "state": state},
|
|
)
|
|
for state in ("provisioning", "deleting", "error")
|
|
)
|
|
except Exception:
|
|
return ()
|
|
|
|
self._control_space_state_gauge = self._meter.create_observable_gauge(
|
|
"dify.knowledge_fs.control_spaces",
|
|
callbacks=(observe,),
|
|
description=(
|
|
"Global KnowledgeFS control-space database snapshot by operational state; "
|
|
"aggregate replica series with max, never sum"
|
|
),
|
|
unit="{space}",
|
|
)
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def get_knowledge_fs_operational_metrics() -> KnowledgeFSOperationalMetricsPort:
|
|
"""Return one process-wide instrument set to avoid duplicate OpenTelemetry registration."""
|
|
|
|
return OpenTelemetryKnowledgeFSOperationalMetrics()
|
|
|
|
|
|
__all__ = [
|
|
"KnowledgeFSBatchStatusMetric",
|
|
"KnowledgeFSCapabilityIssuanceMetric",
|
|
"KnowledgeFSControlSpaceStateMetric",
|
|
"KnowledgeFSLifecycleTaskMetric",
|
|
"KnowledgeFSOperationAdmissionMetric",
|
|
"KnowledgeFSOperationalMetricsPort",
|
|
"OpenTelemetryKnowledgeFSOperationalMetrics",
|
|
"get_knowledge_fs_operational_metrics",
|
|
]
|