Function _emit_online_audit

sdk/python/feast/feature_server.py:191–219

Best-effort audit log emission for online feature requests.

(
    request: GetOnlineFeaturesRequest,
    features: Union[List[str], "feast.FeatureService"],
    entity_count: int,
    status: str,
    latency_ms: float,
)

def _emit_online_audit(
    request: GetOnlineFeaturesRequest,
    features: Union[List[str], "feast.FeatureService"],
    entity_count: int,
    status: str,
    latency_ms: float,
):
    """Best-effort audit log emission for online feature requests."""
    try:
        from feast.permissions.security_manager import get_security_manager

        requestor_id = "anonymous"
        sm = get_security_manager()
        if sm and sm.current_user:
            requestor_id = sm.current_user.username or "anonymous"

        fv_names, feat_count = _parse_feature_info(features)

        feast_metrics.emit_online_audit_log(
            requestor_id=requestor_id,
            entity_keys=list(request.entities.keys()),
            entity_count=entity_count,
            feature_views=fv_names,
            feature_count=feat_count,
            status=status,
            latency_ms=latency_ms,
        )
    except Exception:
        logger.warning("Failed to emit online audit log", exc_info=True)


async def _get_features(