lmcache--lmcache/lmcache/observability.py

# SPDX-License-Identifier: Apache-2.0
# Standard
from contextlib import contextmanager
from copy import copy
from dataclasses import dataclass, field
from typing import (
    TYPE_CHECKING,
    Any,
    Dict,
    Iterable,
    List,
    Optional,
    Sequence,
    Union,
)
import os
import threading
import time

# Third Party
from prometheus_client import REGISTRY
import prometheus_client

# First Party
from lmcache.logging import init_logger
from lmcache.usage_telemetry import ContinuousUsageContext
from lmcache.utils import thread_safe
from lmcache.v1.metadata import LMCacheMetadata

if TYPE_CHECKING:
    # First Party
    from lmcache.v1.config import LMCacheEngineConfig

logger = init_logger(__name__)


@dataclass
class LMCacheStats:
    # Counter (Note that these are incremental values,
    # which will accumulate over time in Counter)
    interval_retrieve_requests: int
    interval_store_requests: int
    interval_lookup_requests: int
    interval_requested_tokens: int
    interval_hit_tokens: int
    interval_stored_tokens: int
    interval_lookup_tokens: int
    interval_lookup_hits: int
    interval_vllm_hit_tokens: int
    interval_prompt_tokens: int

    interval_num_slow_retrieval_by_time: int
    interval_num_slow_retrieval_by_speed: int

    interval_remote_read_requests: int
    interval_remote_read_bytes: int
    interval_remote_write_requests: int
    interval_remote_write_bytes: int

    interval_remote_time_to_get: List[float]
    interval_remote_time_to_put: List[float]
    interval_remote_time_to_get_sync: List[float]

    interval_remote_ping_latency: float  # Ping latency in milliseconds
    interval_remote_ping_errors: int  # Number of ping errors
    interval_remote_ping_success: int  # Number of ping successes
    interval_remote_ping_error_code: int  # Latest ping error code

    interval_local_cpu_evict_count: int  # evict count
    interval_local_cpu_evict_keys_count: int  # evict keys count
    interval_local_cpu_evict_failed_count: int  # evict failed count

    interval_forced_unpin_count: int  # forced unpin count due to timeout

    # Real time value measurements (will be reset after each log)
    retrieve_hit_rate: float
    lookup_hit_rate: float

    local_cache_usage_bytes: int  # Size of the used local cache in bytes
    remote_cache_usage_bytes: int  # Size of the used remote cache in bytes
    local_storage_usage_bytes: int  # Size of the used local storage in bytes

    active_memory_objs_count: int  # the number of active memory objects
    pinned_memory_objs_count: int  # the number of pinned memory objects

    # Distribution measurements
    time_to_retrieve: List[float]
    time_to_store: List[float]
    time_to_lookup: List[float]
    retrieve_speed: List[float]  # Tokens per second
    store_speed: List[float]  # Tokens per second

    # Granular profiling measurements
    retrieve_process_tokens_time: List[float]
    retrieve_broadcast_time: List[float]
    retrieve_to_gpu_time: List[float]
    remote_backend_batched_get_blocking_time: List[float]
    instrumented_connector_batched_get_time: List[float]
    store_process_tokens_time: List[float]
    store_from_gpu_time: List[float]
    store_put_time: List[float]

    # P2P transfer metrics
    interval_p2p_requests: int
    interval_p2p_transferred_tokens: int
    p2p_time_to_transfer: List[float]
    p2p_transfer_speed: List[float]  # Tokens per second

    # request lookup hit rates
    # use bucket of interval_lookup_hit_rates to represents non-0 hit requests
    # use interval_lookup_0_hit_requests to represents 0 hit requests
    interval_lookup_hit_rates: List[float]
    interval_lookup_0_hit_requests: int

    interval_request_cache_lifespan: List[float]  # cache lifespan in minutes


@dataclass
class LookupRequestStats:
    request_id: int
    num_tokens: int
    hit_tokens: int
    is_finished: bool
    start_time: float = 0
    end_time: float = 0

    def time_to_lookup(self):
        if self.end_time == 0:
            return 0
        return self.end_time - self.start_time

    def hit_rate(self):
        if self.num_tokens == 0:
            return 0
        return self.hit_tokens / self.num_tokens


@dataclass
class RetrieveRequestStats:
    request_id: int
    num_tokens: int
    local_hit_tokens: int
    remote_hit_tokens: int  # Not used for now
    start_time: float
    end_time: float
    process_tokens_time: float = 0
    broadcast_time: float = 0
    to_gpu_time: float = 0
    detailed_metrics: Dict[str, Any] = field(default_factory=dict)

    def time_to_retrieve(self):
        if self.end_time == 0:
            return 0
        return self.end_time - self.start_time

    def retrieve_speed(self):
        if self.time_to_retrieve() == 0:
            return 0
        return (
            self.local_hit_tokens + self.remote_hit_tokens
        ) / self.time_to_retrieve()

    @contextmanager
    def profile_process_tokens(self):
        start = time.perf_counter()
        try:
            yield
        finally:
            self.process_tokens_time += time.perf_counter() - start

    @contextmanager
    def profile_broadcast(self):
        start = time.perf_counter()
        try:
            yield
        finally:
            self.broadcast_time += time.perf_counter() - start

    @contextmanager
    def profile_to_gpu(self):
        start = time.perf_counter()
        try:
            yield
        finally:
            self.to_gpu_time += time.perf_counter() - start


@dataclass
class StoreRequestStats:
    request_id: int
    num_tokens: int
    start_time: float
    end_time: float
    process_tokens_time: float = 0
    from_gpu_time: float = 0
    put_time: float = 0

    def time_to_store(self):
        if self.end_time == 0:
            return 0
        return self.end_time - self.start_time

    def store_speed(self):
        if self.time_to_store() == 0:
            return 0
        return self.num_tokens / self.time_to_store()

    @contextmanager
    def profile_process_tokens(self):
        start = time.perf_counter()
        try:
            yield
        finally:
            self.process_tokens_time += time.perf_counter() - start

    @contextmanager
    def profile_from_gpu(self):
        start = time.perf_counter()
        try:
            yield
        finally:
            self.from_gpu_time += time.perf_counter() - start

    @contextmanager
    def profile_put(self):
        start = time.perf_counter()
        try:
            yield
        finally:
            self.put_time += time.perf_counter() - start


@dataclass
class P2PTransferRequestStats:
    num_tokens: int
    start_time: float
    end_time: float

    def time_to_transfer(self):
        if self.end_time == 0:
            return 0
        return self.end_time - self.start_time

    def transfer_speed(self):
        if self.time_to_transfer() == 0:
            return 0
        return self.num_tokens / self.time_to_transfer()


class LMCStatsMonitor:
    def __init__(self):
        # Interval metrics that will be reset after each log
        # Accumulate incremental values in the Prometheus Counter
        self.interval_retrieve_requests = 0
        self.interval_store_requests = 0
        self.interval_lookup_requests = 0
        self.interval_requested_tokens = 0  # total requested tokens retrieve
        self.interval_hit_tokens = 0  # total hit tokens retrieve
        self.interval_stored_tokens = 0  # total tokens tored in LMCache
        self.interval_lookup_tokens = 0  # total requested tokens lookup
        self.interval_lookup_hits = 0  # total hit tokens lookup
        self.interval_vllm_hit_tokens = 0  # total hit tokens in vllm
        self.interval_prompt_tokens = 0  # total prompt tokens
        self.interval_lookup_0_hit_requests = 0

        self.interval_num_slow_retrieval_by_time = 0
        self.interval_num_slow_retrieval_by_speed = 0

        # P2P transfer metrics
        self.interval_p2p_requests = 0
        self.interval_p2p_transferred_tokens = 0
        self.p2p_requests: Dict[int, P2PTransferRequestStats] = {}
        self.p2p_request_id = 0

        # remote backends read/write metrics
        self.interval_remote_read_requests = 0
        self.interval_remote_read_bytes = 0
        self.interval_remote_write_requests = 0
        self.interval_remote_write_bytes = 0

        # remote backends get/put cost time metrics
        self.interval_remote_time_to_get: List[float] = []
        self.interval_remote_time_to_put: List[float] = []
        # the time of get value from remote backends synchronously,
        # which includes rpc and schedule time
        self.interval_remote_time_to_get_sync: List[float] = []

        self.interval_remote_ping_latency = 0
        self.interval_remote_ping_errors = 0
        self.interval_remote_ping_success = 0
        self.interval_remote_ping_error_code = 0  # 0 means success

        self.interval_local_cpu_evict_count = 0
        self.interval_local_cpu_evict_keys_count = 0
        self.interval_local_cpu_evict_failed_count = 0

        self.interval_forced_unpin_count = 0

        self.local_cache_usage_bytes = 0
        self.remote_cache_usage_bytes = 0
        self.local_storage_usage_bytes = 0

        self.active_memory_objs_count = 0
        self.pinned_memory_objs_count = 0

        self.retrieve_requests: Dict[int, RetrieveRequestStats] = {}
        self.store_requests: Dict[int, StoreRequestStats] = {}
        self.lookup_requests: Dict[int, LookupRequestStats] = {}

        self.retrieve_request_id = 0
        self.store_request_id = 0
        self.lookup_request_id = 0

        self.interval_request_cache_lifespan: Dict[int, float] = {}
        self.reuse_chunk_id = 0

        self._current_retrieve_stats: Optional[RetrieveRequestStats] = None
        self.retrieve_time_threshold: float = 1e9
        self.retrieve_token_speed_threshold: float = -1.0
        self.last_retrieve_warning_time: float = 0.0
        self.skipped_retrieve_warning_count: int = 0

    def set_current_retrieve_stats(self, stats: RetrieveRequestStats):
        self._current_retrieve_stats = stats

    def get_current_retrieve_stats(self) -> Optional[RetrieveRequestStats]:
        return self._current_retrieve_stats

    def clear_current_retrieve_stats(self):
        self._current_retrieve_stats = None

    @thread_safe
    def on_lookup_request(self, num_tokens: int) -> LookupRequestStats:
        """
        This function is called when a lookup request is sent to the cache.
        It will record the number of tokens requested.
        """
        curr_time = time.perf_counter()
        lookup_stats = LookupRequestStats(
            request_id=self.lookup_request_id,
            num_tokens=num_tokens,
            hit_tokens=0,
            is_finished=False,
            start_time=curr_time,
        )
        self.interval_lookup_requests += 1
        self.interval_lookup_tokens += num_tokens
        self.lookup_requests[self.lookup_request_id] = lookup_stats
        self.lookup_request_id += 1
        return lookup_stats

    @thread_safe
    def on_lookup_finished(
        self,
        stats: LookupRequestStats,
        num_hit_tokens: int,
    ):
        """
        This function is called when a lookup request is finished.
        It will record the number of tokens hit and track by node type.

        Args:
            stats: LookupRequestStats object
            num_hit_tokens: Total number of tokens found in lookup
        """
        curr_time = time.perf_counter()
        assert stats.request_id in self.lookup_requests
        stats.hit_tokens = num_hit_tokens
        stats.is_finished = True
        if stats.end_time == 0:
            stats.end_time = curr_time
        self.interval_lookup_hits += num_hit_tokens
        if num_hit_tokens == 0:
            self.interval_lookup_0_hit_requests += 1

    @thread_safe
    def on_retrieve_request(self, num_tokens: int) -> RetrieveRequestStats:
        """
        Returns the internal "request id" that will be used in
        on_retrieve_finished
        """
        curr_time = time.perf_counter()
        retrieve_stats = RetrieveRequestStats(
            request_id=self.retrieve_request_id,
            num_tokens=num_tokens,
            local_hit_tokens=0,
            remote_hit_tokens=0,
            start_time=curr_time,
            end_time=0,
        )
        self.interval_requested_tokens += num_tokens
        self.interval_retrieve_requests += 1
        self.retrieve_requests[self.retrieve_request_id] = retrieve_stats
        self.retrieve_request_id += 1
        self.set_current_retrieve_stats(retrieve_stats)
        return retrieve_stats

    @thread_safe
    def on_retrieve_finished(
        self,
        retrieve_stats: RetrieveRequestStats,
        num_retrieved_tokens: int,
    ):
        curr_time = time.perf_counter()
        assert retrieve_stats.request_id in self.retrieve_requests
        retrieve_stats.local_hit_tokens = num_retrieved_tokens
        if retrieve_stats.end_time == 0:
            retrieve_stats.end_time = curr_time
        self.interval_hit_tokens += num_retrieved_tokens
        self.clear_current_retrieve_stats()

        time_to_retrieve = retrieve_stats.time_to_retrieve()
        retrieve_speed = retrieve_stats.retrieve_speed()
        if time_to_retrieve > self.retrieve_time_threshold:
            self.interval_num_slow_retrieval_by_time += 1
        if 0 < retrieve_speed < self.retrieve_token_speed_threshold:
            self.interval_num_slow_retrieval_by_speed += 1

        # Log a warning if the retrieval performance is below defined thresholds:
        # 1. Total time taken (time_to_retrieve) exceeds the maximum allowed time.
        # 2. Retrieval speed (retrieve_speed) falls below the minimum required tokens/s.
        # The warnings are rate-limited to once every 10 seconds to avoid log flooding.
        if (
            time_to_retrieve > self.retrieve_time_threshold
            or 0 < retrieve_speed < self.retrieve_token_speed_threshold
        ):
            if curr_time - self.last_retrieve_warning_time > 10.0:
                logger.warning(
                    "Retrieve request %d surpassed threshold: "
                    "time_to_retrieve=%.4f s (threshold=%.4f s), "
                    "retrieve_speed=%.2f tokens/s (threshold=%.2f tokens/s). "
                    "Skipped %d slow retrieval logs in the last %.1f seconds. "
                    "Detailed metrics: "
                    "num_tokens=%d, local_hit_tokens=%d, remote_hit_tokens=%d, "
                    "process_tokens_time=%.5f s, "
                    "broadcast_time=%.5f s, "
                    "to_gpu_time=%.5f s, "
                    "detailed_metrics=%s",
                    retrieve_stats.request_id,
                    time_to_retrieve,
                    self.retrieve_time_threshold,
                    retrieve_speed,
                    self.retrieve_token_speed_threshold,
                    self.skipped_retrieve_warning_count,
                    curr_time - self.last_retrieve_warning_time,
                    retrieve_stats.num_tokens,
                    retrieve_stats.local_hit_tokens,
                    retrieve_stats.remote_hit_tokens,
                    retrieve_stats.process_tokens_time,
                    retrieve_stats.broadcast_time,
                    retrieve_stats.to_gpu_time,
                    retrieve_stats.detailed_metrics,
                )
                self.last_retrieve_warning_time = curr_time
                self.skipped_retrieve_warning_count = 0
            else:
                self.skipped_retrieve_warning_count += 1

    @thread_safe
    def on_store_request(self, num_tokens: int) -> StoreRequestStats:
        """
        Returns the internal "request id" that will be used in on_store_finished
        """
        curr_time = time.perf_counter()
        store_stats = StoreRequestStats(
            request_id=self.store_request_id,
            num_tokens=num_tokens,
            start_time=curr_time,
            end_time=0,
        )
        self.interval_store_requests += 1
        self.interval_stored_tokens += num_tokens
        self.store_requests[self.store_request_id] = store_stats
        self.store_request_id += 1
        return store_stats

    @thread_safe
    def on_store_finished(
        self,
        store_stats: StoreRequestStats,
        num_stored_tokens: int = -1,
    ):
        curr_time = time.perf_counter()
        assert store_stats.request_id in self.store_requests
        if store_stats.end_time == 0:
            store_stats.end_time = curr_time
        if num_stored_tokens >= 0:
            store_stats.num_tokens = num_stored_tokens

    @thread_safe
    def on_p2p_transfer_request(self, num_tokens: int) -> int:
        curr_time = time.time()
        self.interval_p2p_requests += 1
        self.p2p_requests[self.p2p_request_id] = P2PTransferRequestStats(
            num_tokens=num_tokens,
            start_time=curr_time,
            end_time=0,
        )
        self.p2p_request_id += 1
        return self.p2p_request_id - 1

    @thread_safe
    def on_p2p_transfer_finished(self, request_id: int):
        curr_time = time.time()
        assert request_id in self.p2p_requests
        p2p_stats = self.p2p_requests[request_id]
        self.interval_p2p_transferred_tokens += p2p_stats.num_tokens
        p2p_stats.end_time = curr_time

    @thread_safe
    def on_chunk_reuse(self, time_interval: float):
        """
        time_interval: float or int, in seconds
        """
        self.interval_request_cache_lifespan[self.reuse_chunk_id] = time_interval / 60.0
        self.reuse_chunk_id += 1

    @thread_safe
    def update_local_cache_usage(self, usage: int):
        self.local_cache_usage_bytes = usage

    @thread_safe
    def update_remote_cache_usage(self, usage: int):
        self.remote_cache_usage_bytes = usage

    @thread_safe
    def update_local_storage_usage(self, usage: int):
        self.local_storage_usage_bytes = usage

    @thread_safe
    def update_interval_remote_read_metrics(self, read_bytes: int):
        self.interval_remote_read_requests += 1
        self.interval_remote_read_bytes += read_bytes

    @thread_safe
    def update_interval_remote_write_metrics(self, write_bytes: int):
        self.interval_remote_write_requests += 1
        self.interval_remote_write_bytes += write_bytes

    @thread_safe
    def update_interval_remote_time_to_get(self, get_time: float):
        self.interval_remote_time_to_get.append(get_time)

    @thread_safe
    def update_interval_remote_time_to_put(self, put_time: float):
        self.interval_remote_time_to_put.append(put_time)

    @thread_safe
    def update_interval_remote_time_to_get_sync(self, get_time_sync: float):
        self.interval_remote_time_to_get_sync.append(get_time_sync)

    @thread_safe
    def update_remote_ping_latency(self, latency: float):
        self.interval_remote_ping_latency = latency

    @thread_safe
    def update_remote_ping_error_code(self, error_code: int):
        """Update ping error code"""
        self.interval_remote_ping_error_code = error_code
        if error_code != 0:
            self.interval_remote_ping_errors += 1
        else:
            self.interval_remote_ping_success += 1

    @thread_safe
    def update_local_cpu_evict_metrics(self, evict_keys_count: int):
        self.interval_local_cpu_evict_count += 1
        self.interval_local_cpu_evict_keys_count += evict_keys_count

    @thread_safe
    def update_local_cpu_evict_failed_count(self, evict_failed_count: int):
        self.interval_local_cpu_evict_failed_count += evict_failed_count

    @thread_safe
    def update_forced_unpin_count(self, delta: int):
        self.interval_forced_unpin_count += delta

    @thread_safe
    def update_active_memory_objs_count(self, active_memory_objs_count: int):
        self.active_memory_objs_count = active_memory_objs_count

    @thread_safe
    def update_pinned_memory_objs_count(self, delta: int):
        self.pinned_memory_objs_count += delta

    @thread_safe
    def update_interval_vllm_hit_tokens(self, delta: int):
        self.interval_vllm_hit_tokens += delta

    @thread_safe
    def update_interval_prompt_tokens(self, delta: int):
        self.interval_prompt_tokens += delta

    def _clear(self):
        """
        Clear all the distribution stats
        """
        self.interval_retrieve_requests = 0
        self.interval_store_requests = 0
        self.interval_lookup_requests = 0

        self.interval_requested_tokens = 0
        self.interval_hit_tokens = 0
        self.interval_stored_tokens = 0
        self.interval_lookup_tokens = 0
        self.interval_lookup_hits = 0
        self.interval_vllm_hit_tokens = 0
        self.interval_prompt_tokens = 0

        self.interval_num_slow_retrieval_by_time = 0
        self.interval_num_slow_retrieval_by_speed = 0

        self.interval_remote_read_requests = 0
        self.interval_remote_read_bytes = 0
        self.interval_remote_write_requests = 0
        self.interval_remote_write_bytes = 0

        self.interval_remote_time_to_get.clear()
        self.interval_remote_time_to_put.clear()
        self.interval_remote_time_to_get_sync.clear()

        self.interval_remote_ping_latency = 0
        self.interval_remote_ping_errors = 0
        self.interval_remote_ping_success = 0
        self.interval_remote_ping_error_code = 0

        self.interval_local_cpu_evict_count = 0
        self.interval_local_cpu_evict_keys_count = 0
        self.interval_local_cpu_evict_failed_count = 0

        self.interval_forced_unpin_count = 0

        self.interval_p2p_requests = 0
        self.interval_p2p_transferred_tokens = 0

        self.interval_lookup_0_hit_requests = 0

        new_retrieve_requests = {}
        for request_id, retrieve_stats in self.retrieve_requests.items():
            if retrieve_stats.end_time == 0:
                new_retrieve_requests[request_id] = retrieve_stats
        self.retrieve_requests = new_retrieve_requests

        new_store_requests = {}
        for request_id, store_stats in self.store_requests.items():
            if store_stats.end_time == 0:
                new_store_requests[request_id] = store_stats
        self.store_requests = new_store_requests

        new_p2p_requests = {}
        for request_id, p2p_stats in self.p2p_requests.items():
            if p2p_stats.end_time == 0:
                new_p2p_requests[request_id] = p2p_stats
        self.p2p_requests = new_p2p_requests

        new_lookup_requests = {}
        for request_id, lookup_stats in self.lookup_requests.items():
            if not lookup_stats.is_finished:
                new_lookup_requests[request_id] = lookup_stats
        self.lookup_requests = new_lookup_requests

        self.interval_request_cache_lifespan.clear()
        self.reuse_chunk_id = 0

    @thread_safe
    def get_stats_and_clear(self) -> LMCacheStats:
        """
        This function should be called with by prometheus adapter with
        a specific interval.
        The function will return the latest states between the current
        call and the previous call.
        """
        # Calculate retrieve hit rate based on requests finished in this interval
        finished_retrieve_stats = [
            s for s in self.retrieve_requests.values() if s.end_time != 0
        ]
        sum_finished_retrieve_requested_tokens = sum(
            s.num_tokens for s in finished_retrieve_stats
        )
        sum_finished_retrieve_hit_tokens = sum(
            s.local_hit_tokens + s.remote_hit_tokens for s in finished_retrieve_stats
        )
        retrieve_hit_rate = (
            1
            if len(finished_retrieve_stats) == 0
            or sum_finished_retrieve_requested_tokens == 0
            else sum_finished_retrieve_hit_tokens
            / sum_finished_retrieve_requested_tokens
        )

        # Calculate lookup hit rate based on requests finished in this interval
        finished_lookup_stats = [
            s for s in self.lookup_requests.values() if s.is_finished
        ]
        sum_finished_lookup_requested = sum(s.num_tokens for s in finished_lookup_stats)
        sum_finished_lookup_hit = sum(s.hit_tokens for s in finished_lookup_stats)
        lookup_hit_rate = (
            0
            if sum_finished_lookup_requested == 0
            else sum_finished_lookup_hit / sum_finished_lookup_requested
        )

        def filter_out_zeros(stats: Iterable[float]) -> List[float]:
            return [x for x in stats if x != 0]

        time_to_retrieve = filter_out_zeros(
            stats.time_to_retrieve() for stats in self.retrieve_requests.values()
        )

        time_to_store = filter_out_zeros(
            stats.time_to_store() for stats in self.store_requests.values()
        )

        time_to_lookup = filter_out_zeros(
            stats.time_to_lookup() for stats in self.lookup_requests.values()
        )

        retrieve_speed = filter_out_zeros(
            stats.retrieve_speed() for stats in self.retrieve_requests.values()
        )

        store_speed = filter_out_zeros(
            stats.store_speed() for stats in self.store_requests.values()
        )

        # Granular profiling measurements
        retrieve_process_tokens_time = filter_out_zeros(
            stats.process_tokens_time for stats in self.retrieve_requests.values()
        )
        retrieve_broadcast_time = filter_out_zeros(
            stats.broadcast_time for stats in self.retrieve_requests.values()
        )
        retrieve_to_gpu_time = filter_out_zeros(
            stats.to_gpu_time for stats in self.retrieve_requests.values()
        )
        remote_backend_batched_get_blocking_time = filter_out_zeros(
            stats.detailed_metrics.get("remote_backend_batched_get_blocking_time", 0.0)
            for stats in self.retrieve_requests.values()
        )
        instrumented_connector_batched_get_time = filter_out_zeros(
            stats.detailed_metrics.get("instrumented_connector_batched_get_time", 0.0)
            for stats in self.retrieve_requests.values()
        )
        store_process_tokens_time = filter_out_zeros(
            stats.process_tokens_time for stats in self.store_requests.values()
        )
        store_from_gpu_time = filter_out_zeros(
            stats.from_gpu_time for stats in self.store_requests.values()
        )
        store_put_time = filter_out_zeros(
            stats.put_time for stats in self.store_requests.values()
        )

        p2p_time_to_transfer = filter_out_zeros(
            stats.time_to_transfer() for stats in self.p2p_requests.values()
        )

        p2p_transfer_speed = filter_out_zeros(
            stats.transfer_speed() for stats in self.p2p_requests.values()
        )

        request_lookup_hit_rates = filter_out_zeros(
            stats.hit_rate()
            for stats in self.lookup_requests.values()
            if stats.is_finished
        )

        request_lifespan = list(self.interval_request_cache_lifespan.values())

        ret = LMCacheStats(
            interval_retrieve_requests=self.interval_retrieve_requests,
            interval_store_requests=self.interval_store_requests,
            interval_lookup_requests=self.interval_lookup_requests,
            interval_requested_tokens=self.interval_requested_tokens,
            interval_hit_tokens=self.interval_hit_tokens,
            interval_stored_tokens=self.interval_stored_tokens,
            interval_lookup_tokens=self.interval_lookup_tokens,
            interval_lookup_hits=self.interval_lookup_hits,
            interval_remote_read_requests=self.interval_remote_read_requests,
            interval_remote_read_bytes=self.interval_remote_read_bytes,
            interval_remote_write_requests=self.interval_remote_write_requests,
            interval_remote_write_bytes=self.interval_remote_write_bytes,
            interval_remote_time_to_get=self.interval_remote_time_to_get.copy(),
            interval_remote_time_to_put=self.interval_remote_time_to_put.copy(),
            interval_remote_time_to_get_sync=self.interval_remote_time_to_get_sync.copy(),
            interval_remote_ping_latency=self.interval_remote_ping_latency,
            interval_remote_ping_errors=self.interval_remote_ping_errors,
            interval_remote_ping_success=self.interval_remote_ping_success,
            interval_remote_ping_error_code=self.interval_remote_ping_error_code,
            retrieve_hit_rate=retrieve_hit_rate,
            lookup_hit_rate=lookup_hit_rate,
            interval_local_cpu_evict_count=self.interval_local_cpu_evict_count,
            interval_local_cpu_evict_keys_count=self.interval_local_cpu_evict_keys_count,
            interval_local_cpu_evict_failed_count=self.interval_local_cpu_evict_failed_count,
            interval_forced_unpin_count=self.interval_forced_unpin_count,
            local_cache_usage_bytes=self.local_cache_usage_bytes,
            remote_cache_usage_bytes=self.remote_cache_usage_bytes,
            local_storage_usage_bytes=self.local_storage_usage_bytes,
            active_memory_objs_count=self.active_memory_objs_count,
            pinned_memory_objs_count=self.pinned_memory_objs_count,
            time_to_retrieve=time_to_retrieve,
            time_to_store=time_to_store,
            time_to_lookup=time_to_lookup,
            retrieve_speed=retrieve_speed,
            store_speed=store_speed,
            retrieve_process_tokens_time=retrieve_process_tokens_time,
            retrieve_broadcast_time=retrieve_broadcast_time,
            retrieve_to_gpu_time=retrieve_to_gpu_time,
            remote_backend_batched_get_blocking_time=remote_backend_batched_get_blocking_time,  # noqa: E501
            instrumented_connector_batched_get_time=instrumented_connector_batched_get_time,  # noqa: E501
            store_process_tokens_time=store_process_tokens_time,
            store_from_gpu_time=store_from_gpu_time,
            store_put_time=store_put_time,
            interval_vllm_hit_tokens=self.interval_vllm_hit_tokens,
            interval_p2p_requests=self.interval_p2p_requests,
            interval_num_slow_retrieval_by_time=self.interval_num_slow_retrieval_by_time,
            interval_num_slow_retrieval_by_speed=self.interval_num_slow_retrieval_by_speed,
            interval_p2p_transferred_tokens=self.interval_p2p_transferred_tokens,
            p2p_time_to_transfer=p2p_time_to_transfer,
            p2p_transfer_speed=p2p_transfer_speed,
            interval_lookup_hit_rates=request_lookup_hit_rates,
            interval_request_cache_lifespan=request_lifespan,
            interval_prompt_tokens=self.interval_prompt_tokens,
            interval_lookup_0_hit_requests=self.interval_lookup_0_hit_requests,
        )
        self._clear()
        return ret

    _instance = None

    @staticmethod
    def GetOrCreate() -> "LMCStatsMonitor":
        if LMCStatsMonitor._instance is None:
            LMCStatsMonitor._instance = LMCStatsMonitor()
        return LMCStatsMonitor._instance

    @staticmethod
    def DestroyInstance():
        LMCStatsMonitor._instance = None

    @staticmethod
    def unregister_all_metrics():
        collectors = list(REGISTRY._collector_to_names.keys())
        for collector in collectors:
            try:
                REGISTRY.unregister(collector)
            except KeyError:
                pass


class PrometheusLogger:
    lmcache_is_healthy: prometheus_client.Gauge
    periodic_threads_total_count: prometheus_client.Gauge
    periodic_threads_running_count: prometheus_client.Gauge
    periodic_threads_active_count: prometheus_client.Gauge

    _gauge_cls = prometheus_client.Gauge
    _counter_cls = prometheus_client.Counter
    _histogram_cls = prometheus_client.Histogram

    def _create_counter(
        self, name: str, documentation: str, labelnames: List[str]
    ) -> prometheus_client.Counter:
        """Create a Counter and register it for reset_counters()."""
        counter = self._counter_cls(
            name=name, documentation=documentation, labelnames=labelnames
        )
        self._counters.append(counter)
        return counter

    def _create_histogram(
        self,
        name: str,
        documentation: str,
        labelnames: List[str],
        buckets: Sequence[float],
    ) -> prometheus_client.Histogram:
        """Create a Histogram and register it for reset_histograms().

        If the ``config`` object's extra_config contains a key
        ``histogram_bucket_<short_name>`` (where ``<short_name>``
        is the metric name without the ``lmcache:`` prefix),
        the value will be used as the bucket list, overriding
        the default *buckets* argument.
        """
        short_name = name.split(":", 1)[-1]
        config_key = "histogram_bucket_%s" % short_name
        custom = (
            self.config.get_extra_config_value(config_key)
            if self.config is not None
            else None
        )
        if custom is not None:
            buckets = custom
            logger.info(
                "Using custom buckets for histogram %s from extra_config key '%s'",
                name,
                config_key,
            )
        histogram = self._histogram_cls(
            name=name,
            documentation=documentation,
            labelnames=labelnames,
            buckets=buckets,
        )
        self._histograms.append(histogram)
        return histogram

    @staticmethod
    def _ensure_multiprocess_dir() -> None:
        multiprocess_dir = os.environ.get("PROMETHEUS_MULTIPROC_DIR")
        if not multiprocess_dir:
            multiprocess_dir = "/tmp/lmcache_prometheus"
            os.environ["PROMETHEUS_MULTIPROC_DIR"] = multiprocess_dir
        os.makedirs(multiprocess_dir, exist_ok=True)

    def __init__(
        self,
        metadata: LMCacheMetadata,
        config: Optional["LMCacheEngineConfig"] = None,
    ):
        # Ensure PROMETHEUS_MULTIPROC_DIR is set before any metric registration
        PrometheusLogger._ensure_multiprocess_dir()

        self.metadata = metadata
        self.config = config

        self.labels = self._metadata_to_labels(metadata)
        labelnames = list(self.labels.keys())

        # List to track all counters/histograms for reset methods
        self._counters: List[prometheus_client.Counter] = []
        self._histograms: List[prometheus_client.Histogram] = []

        self.counter_num_retrieve_requests = self._create_counter(
            name="lmcache:num_retrieve_requests",
            documentation="Total number of retrieve requests sent to lmcache",
            labelnames=labelnames,
        )

        self.counter_num_store_requests = self._create_counter(
            name="lmcache:num_store_requests",
            documentation="Total number of store requests sent to lmcache",
            labelnames=labelnames,
        )

        self.counter_num_lookup_requests = self._create_counter(
            name="lmcache:num_lookup_requests",
            documentation="Total number of lookup requests sent to lmcache",
            labelnames=labelnames,
        )

        self.counter_num_requested_tokens = self._create_counter(
            name="lmcache:num_requested_tokens",
            documentation="Total number of tokens requested from lmcache",
            labelnames=labelnames,
        )

        self.counter_num_hit_tokens = self._create_counter(
            name="lmcache:num_hit_tokens",
            documentation="Total number of tokens hit in lmcache",
            labelnames=labelnames,
        )

        self.counter_num_stored_tokens = self._create_counter(
            name="lmcache:num_stored_tokens",
            documentation=(
                "Total number of tokens stored in lmcache including evicted ones"
            ),
            labelnames=labelnames,
        )

        self.counter_num_lookup_tokens = self._create_counter(
            name="lmcache:num_lookup_tokens",
            documentation="Total number of tokens requested in lookup from lmcache",
            labelnames=labelnames,
        )

        self.counter_num_lookup_hits = self._create_counter(
            name="lmcache:num_lookup_hits",
            documentation="Total number of tokens hit in lookup from lmcache",
            labelnames=labelnames,
        )

        self.counter_num_vllm_hit_tokens = self._create_counter(
            name="lmcache:num_vllm_hit_tokens",
            documentation="Number of hit tokens in vllm",
            labelnames=labelnames,
        )

        self.counter_num_prompt_tokens = self._create_counter(
            name="lmcache:num_prompt_tokens",
            documentation="Number of prompt tokens in lmcache",
            labelnames=labelnames,
        )

        self.counter_num_remote_read_requests = self._create_counter(
            name="lmcache:num_remote_read_requests",
            documentation="Total number of requests read from "
            "remote backends in lmcache",
            labelnames=labelnames,
        )

        self.counter_num_remote_read_bytes = self._create_counter(
            name="lmcache:num_remote_read_bytes",
            documentation="Total number of bytes read from remote backends in lmcache",
            labelnames=labelnames,
        )

        self.counter_num_remote_write_requests = self._create_counter(
            name="lmcache:num_remote_write_requests",
            documentation="Total number of requests write to "
            "remote backends in lmcache",
            labelnames=labelnames,
        )

        self.counter_num_remote_write_bytes = self._create_counter(
            name="lmcache:num_remote_write_bytes",
            documentation="Total number of bytes write to remote backends in lmcache",
            labelnames=labelnames,
        )

        self.counter_local_cpu_evict_count = self._create_counter(
            name="lmcache:local_cpu_evict_count",
            documentation="Total number of evict in local cpu backend",
            labelnames=labelnames,
        )

        self.counter_local_cpu_evict_keys_count = self._create_counter(
            name="lmcache:local_cpu_evict_keys_count",
            documentation="Total number of evict keys in local cpu backend",
            labelnames=labelnames,
        )

        self.counter_local_cpu_evict_failed_count = self._create_counter(
            name="lmcache:local_cpu_evict_failed_count",
            documentation="Total number of failed eviction in local cpu backend",
            labelnames=labelnames,
        )

        self.counter_forced_unpin_count = self._create_counter(
            name="lmcache:forced_unpin_count",
            documentation="Total number of forced unpin due to timeout",
            labelnames=labelnames,
        )

        self.counter_lookup_0_hit_requests = self._create_counter(
            name="lmcache:lookup_0_hit_requests",
            documentation="Total number of 0 hit lookup requests",
            labelnames=labelnames,
        )

        self.counter_num_slow_retrieval_by_time = self._create_counter(
            name="lmcache:num_slow_retrieval_by_time",
            documentation="Total number of slow retrievals by time threshold",
            labelnames=labelnames,
        )

        self.counter_num_slow_retrieval_by_speed = self._create_counter(
            name="lmcache:num_slow_retrieval_by_speed",
            documentation="Total number of slow retrievals by speed threshold",
            labelnames=labelnames,
        )

        self.gauge_retrieve_hit_rate = self._gauge_cls(
            name="lmcache:retrieve_hit_rate",
            documentation="Hit rate of lmcache retrieve requests since last log",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )

        self.gauge_lookup_hit_rate = self._gauge_cls(
            name="lmcache:lookup_hit_rate",
            documentation="Hit rate of lmcache lookup requests since last log",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )

        self.gauge_local_cache_usage = self._gauge_cls(
            name="lmcache:local_cache_usage",
            documentation="Local cache usage (bytes) of lmcache",
            labelnames=labelnames,
            multiprocess_mode="sum",
        )

        self.gauge_remote_cache_usage = self._gauge_cls(
            name="lmcache:remote_cache_usage",
            documentation="Remote cache usage (bytes) of lmcache",
            labelnames=labelnames,
            multiprocess_mode="sum",
        )

        self.gauge_local_storage_usage = self._gauge_cls(
            name="lmcache:local_storage_usage",
            documentation="Local storage usage (bytes) of lmcache",
            labelnames=labelnames,
            multiprocess_mode="sum",
        )

        self.gauge_active_memory_objs_count = self._gauge_cls(
            name="lmcache:active_memory_objs_count",
            documentation="The number of active memory objects",
            labelnames=labelnames,
            multiprocess_mode="sum",
        )

        self.gauge_pinned_memory_objs_count = self._gauge_cls(
            name="lmcache:pinned_memory_objs_count",
            documentation="The number of pinned memory objects",
            labelnames=labelnames,
            multiprocess_mode="sum",
        )

        time_to_retrieve_buckets = [
            0.001,
            0.005,
            0.01,
            0.02,
            0.04,
            0.06,
            0.08,
            0.1,
            0.25,
            0.5,
            0.75,
            1.0,
            2.5,
            5.0,
            7.5,
            10.0,
        ]
        self.histogram_time_to_retrieve = self._create_histogram(
            name="lmcache:time_to_retrieve",
            documentation="Time to retrieve from lmcache (seconds)",
            labelnames=labelnames,
            buckets=time_to_retrieve_buckets,
        )

        time_to_store_buckets = [
            0.001,
            0.005,
            0.01,
            0.02,
            0.04,
            0.06,
            0.08,
            0.1,
            0.25,
            0.5,
            0.75,
            1.0,
            2.5,
            5.0,
            7.5,
            10.0,
        ]
        self.histogram_time_to_store = self._create_histogram(
            name="lmcache:time_to_store",
            documentation="Time to store to lmcache (seconds)",
            labelnames=labelnames,
            buckets=time_to_store_buckets,
        )

        time_to_lookup_buckets = [
            0.00001 * 2**i for i in range(20)
        ]  # 0.01 ms to 5000 ms
        self.histogram_time_to_lookup = self._create_histogram(
            name="lmcache:time_to_lookup",
            documentation="Time to lookup in lmcache (seconds)",
            labelnames=labelnames,
            buckets=time_to_lookup_buckets,
        )

        profiling_buckets = [0.00001 * 2**i for i in range(20)]  # 0.01 ms to 5000 ms
        self.histogram_retrieve_process_tokens_time = self._create_histogram(
            name="lmcache:retrieve_process_tokens_time",
            documentation="Time to process tokens in retrieve (seconds)",
            labelnames=labelnames,
            buckets=profiling_buckets,
        )
        self.histogram_retrieve_broadcast_time = self._create_histogram(
            name="lmcache:retrieve_broadcast_time",
            documentation="Time to broadcast memory objects in retrieve (seconds)",
            labelnames=labelnames,
            buckets=profiling_buckets,
        )
        self.histogram_retrieve_to_gpu_time = self._create_histogram(
            name="lmcache:retrieve_to_gpu_time",
            documentation="Time to move data to GPU in retrieve (seconds)",
            labelnames=labelnames,
            buckets=profiling_buckets,
        )
        self.histogram_remote_backend_batched_get_blocking_time = (
            self._create_histogram(
                name="lmcache:remote_backend_batched_get_blocking_time",
                documentation="Time to get data from remote backend (seconds)",
                labelnames=labelnames,
                buckets=profiling_buckets,
            )
        )
        self.histogram_instrumented_connector_batched_get_time = self._create_histogram(
            name="lmcache:instrumented_connector_batched_get_time",
            documentation="Time used by the connector (seconds)",
            labelnames=labelnames,
            buckets=profiling_buckets,
        )
        self.histogram_store_process_tokens_time = self._create_histogram(
            name="lmcache:store_process_tokens_time",
            documentation="Time to process tokens in store (seconds)",
            labelnames=labelnames,
            buckets=profiling_buckets,
        )
        self.histogram_store_from_gpu_time = self._create_histogram(
            name="lmcache:store_from_gpu_time",
            documentation="Time to move data from GPU in store (seconds)",
            labelnames=labelnames,
            buckets=profiling_buckets,
        )
        self.histogram_store_put_time = self._create_histogram(
            name="lmcache:store_put_time",
            documentation="Time to put data to storage in store (seconds)",
            labelnames=labelnames,
            buckets=profiling_buckets,
        )

        retrieve_speed_buckets = [
            1,
            8,
            16,
            32,
            64,
            128,
            256,
            512,
            1024,
            2048,
            4096,
            8192,
            16384,
            32768,
            65536,
        ]
        self.histogram_retrieve_speed = self._create_histogram(
            name="lmcache:retrieve_speed",
            documentation="Retrieve speed of lmcache (tokens per second)",
            labelnames=labelnames,
            buckets=retrieve_speed_buckets,
        )

        store_speed_buckets = [
            1,
            8,
            16,
            32,
            64,
            128,
            256,
            512,
            1024,
            2048,
            4096,
            8192,
            16384,
            32768,
            65536,
        ]
        self.histogram_store_speed = self._create_histogram(
            name="lmcache:store_speed",
            documentation="Store speed of lmcache (tokens per second)",
            labelnames=labelnames,
            buckets=store_speed_buckets,
        )

        # P2P transfer metrics
        p2p_time_buckets = [
            0.001,  # 1ms
            0.005,  # 5ms
            0.01,  # 10ms
            0.02,  # 20ms
            0.04,  # 40ms
            0.06,  # 60ms
            0.08,  # 80ms
            0.1,  # 100ms
            0.25,  # 250ms
            0.5,  # 500ms
            0.75,  # 750ms
            1.0,  # 1s
            2.5,  # 2.5s
            5.0,  # 5s
            7.5,  # 7.5s
            10.0,  # 10s
        ]
        self.histogram_p2p_time_to_transfer = self._create_histogram(
            name="lmcache:p2p_time_to_transfer",
            documentation="Time to transfer via P2P (seconds)",
            labelnames=labelnames,
            buckets=p2p_time_buckets,
        )

        p2p_speed_buckets = [
            1,
            8,
            16,
            32,
            64,
            128,
            256,
            512,
            1024,
            2048,
            4096,
            8192,
            16384,
            32768,
            65536,
        ]
        self.histogram_p2p_transfer_speed = self._create_histogram(
            name="lmcache:p2p_transfer_speed",
            documentation="P2P transfer speed (tokens per second)",
            labelnames=labelnames,
            buckets=p2p_speed_buckets,
        )

        remote_time_to_get = [
            1,
            5,
            10,
            20,
            40,
            60,
            80,
            100,
            250,
            500,
            750,
            1000,
            2500,
            5000,
            7500,
            10000,
        ]
        self.histogram_remote_time_to_get = self._create_histogram(
            name="lmcache:remote_time_to_get",
            documentation="Time to get from remote backends (ms)",
            labelnames=labelnames,
            buckets=remote_time_to_get,
        )

        remote_time_to_put = [
            1,
            5,
            10,
            20,
            40,
            60,
            80,
            100,
            250,
            500,
            750,
            1000,
            2500,
            5000,
            7500,
            10000,
        ]
        self.histogram_remote_time_to_put = self._create_histogram(
            name="lmcache:remote_time_to_put",
            documentation="Time to put to remote backends (ms)",
            labelnames=labelnames,
            buckets=remote_time_to_put,
        )

        remote_time_to_get_sync = [
            1,
            5,
            10,
            20,
            40,
            60,
            80,
            100,
            250,
            500,
            750,
            1000,
            2500,
            5000,
            7500,
            10000,
        ]
        self.histogram_remote_time_to_get_sync = self._create_histogram(
            name="lmcache:remote_time_to_get_sync",
            documentation="Time to get from remote backends synchronously(ms)",
            labelnames=labelnames,
            buckets=remote_time_to_get_sync,
        )

        request_cache_hit_rate = [
            0.1,
            0.2,
            0.3,
            0.4,
            0.5,
            0.6,
            0.7,
            0.8,
            0.9,
            1.0,
        ]
        self.histogram_request_cache_hit_rate = self._create_histogram(
            name="lmcache:request_cache_hit_rate",
            documentation="Request cache hit rate",
            labelnames=labelnames,
            buckets=request_cache_hit_rate,
        )

        request_cache_lifespan_buckets = [
            0,
            1,
            5,
            10,
            20,
            40,
            60,
            80,
            100,
            250,
            500,
            750,
            1000,
            2500,
            5000,
        ]
        self.histogram_request_cache_lifespan = self._create_histogram(
            name="lmcache:request_cache_lifespan",
            documentation="Request cache lifespan in minutes",
            labelnames=labelnames,
            buckets=request_cache_lifespan_buckets,
        )

        # Ping latency metrics: use a gauge to record the latest ping latency
        self.gauge_remote_ping_latency = self._gauge_cls(
            name="lmcache:remote_ping_latency",
            documentation="Latest ping latency to remote backends (ms)",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )
        self.counter_remote_ping_errors = self._create_counter(
            name="lmcache:remote_ping_errors",
            documentation="Number of ping errors to remote backends",
            labelnames=labelnames,
        )
        self.counter_remote_ping_successes = self._create_counter(
            name="lmcache:remote_ping_successes",
            documentation="Number of ping successes to remote backends",
            labelnames=labelnames,
        )
        self.gauge_remote_ping_error_code = self._gauge_cls(
            name="lmcache:remote_ping_error_code",
            documentation="Latest ping error code to remote backends",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )
        self._dynamic_metrics(labelnames)

    def _create_dynamic_gauge(
        self,
        name: str,
        documentation: str,
        labelnames: List[str],
        multiprocess_mode: str,
    ) -> None:
        metric_attr = name.removeprefix("lmcache:")
        if metric_attr == name or not metric_attr.isidentifier():
            raise ValueError(f"Invalid dynamic metric name: {name}")

        gauge = self._gauge_cls(
            name=name,
            documentation=documentation,
            labelnames=labelnames,
            multiprocess_mode=multiprocess_mode,
        )
        # Store the shared collector separately from the labeled child.
        # Shallow-copied label views reuse the collector, then call labels()
        # with their own labels so set_function callbacks publish to that view.
        self._dynamic_gauge_collectors[metric_attr] = gauge
        setattr(self, metric_attr, gauge.labels(**self.labels))

    def _bind_dynamic_metric_children(self) -> None:
        PrometheusLogger._ensure_multiprocess_dir()
        for metric_attr, gauge in self._dynamic_gauge_collectors.items():
            setattr(self, metric_attr, gauge.labels(**self.labels))

    def _dynamic_metrics(self, labelnames: List[str]) -> None:
        """
        Dynamically get value by lambda function while capture
        """
        self._dynamic_gauge_collectors: Dict[str, prometheus_client.Gauge] = {}
        self._create_dynamic_gauge(
            name="lmcache:local_cpu_hot_cache_count",
            documentation="The size of the hot_cache",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )
        self._create_dynamic_gauge(
            name="lmcache:local_cpu_keys_in_request_count",
            documentation="The size of the keys_in_request",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )
        self._create_dynamic_gauge(
            name="lmcache:kv_msg_queue_size",
            documentation="The size of the KV message queue in BatchedMessageSender",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )
        self._create_dynamic_gauge(
            name="lmcache:remote_put_task_num",
            documentation="The number of remote put tasks",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )
        self._create_dynamic_gauge(
            name="lmcache:pin_monitor_pinned_objects_count",
            documentation="The number of pinned objects in PinMonitor",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )
        self._create_dynamic_gauge(
            name="lmcache:lmcache_is_healthy",
            documentation="The health status of LMCache (1=healthy, 0=unhealthy)",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )
        self._create_dynamic_gauge(
            name="lmcache:get_blocking_failed_count",
            documentation="The number of get blocking failed",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )
        self._create_dynamic_gauge(
            name="lmcache:put_failed_count",
            documentation="The number of put failed",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )

        event_statuses = ["ongoing", "done", "not_found"]
        for status in event_statuses:
            metric_name = f"storage_events_{status}_count"
            self._create_dynamic_gauge(
                name=f"lmcache:{metric_name}",
                documentation=f"The number of {status.replace('_', ' ')} events",
                labelnames=labelnames,
                multiprocess_mode="sum",
            )

        # Chunk statistics metrics (dynamic)
        self._create_dynamic_gauge(
            name="lmcache:chunk_statistics_enabled",
            documentation="Whether chunk statistics collection is enabled",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )
        self._create_dynamic_gauge(
            name="lmcache:chunk_statistics_total_requests",
            documentation="Total number of requests processed by chunk statistics",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )
        self._create_dynamic_gauge(
            name="lmcache:chunk_statistics_total_chunks",
            documentation="Total number of chunks processed",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )
        self._create_dynamic_gauge(
            name="lmcache:chunk_statistics_unique_chunks",
            documentation="Number of unique chunks (estimated)",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )
        self._create_dynamic_gauge(
            name="lmcache:chunk_statistics_reuse_rate",
            documentation="Chunk reuse rate (0.0 to 1.0)",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )
        self._create_dynamic_gauge(
            name="lmcache:chunk_statistics_bloom_filter_size_mb",
            documentation="Bloom Filter memory usage in MB",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )
        self._create_dynamic_gauge(
            name="lmcache:chunk_statistics_bloom_filter_fill_rate",
            documentation="Bloom Filter fill rate (0.0 to 1.0)",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )
        self._create_dynamic_gauge(
            name="lmcache:chunk_statistics_file_count",
            documentation="Number of files created for file_hash strategy",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )
        self._create_dynamic_gauge(
            name="lmcache:chunk_statistics_current_file_size",
            documentation="Current file size in bytes for file_hash strategy",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )

        # Connector metrics
        connector_metrics = [
            "scheduler_unfinished_requests_count",
            "connector_load_specs_count",
            "connector_request_trackers_count",
            "connector_kv_caches_count",
            "connector_layerwise_retrievers_count",
            "connector_invalid_block_ids_count",
            "connector_requests_priority_count",
        ]

        for metric_name in connector_metrics:
            self._create_dynamic_gauge(
                name=f"lmcache:{metric_name}",
                documentation=f"The count of {metric_name.replace('_', ' ')}",
                labelnames=labelnames,
                multiprocess_mode="livemostrecent",
            )

        # PeriodicThread metrics
        self._create_dynamic_gauge(
            name="lmcache:periodic_threads_total_count",
            documentation="Total number of registered periodic threads",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )
        self._create_dynamic_gauge(
            name="lmcache:periodic_threads_running_count",
            documentation="Number of running periodic threads",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )
        self._create_dynamic_gauge(
            name="lmcache:periodic_threads_active_count",
            documentation="Number of active periodic threads (recently executed)",
            labelnames=labelnames,
            multiprocess_mode="livemostrecent",
        )

        # Per-level metrics for periodic threads
        for level_name in ["critical", "high", "medium", "low"]:
            self._create_dynamic_gauge(
                name=f"lmcache:periodic_threads_{level_name}_total",
                documentation=f"Total number of {level_name} level periodic threads",
                labelnames=labelnames,
                multiprocess_mode="livemostrecent",
            )

            self._create_dynamic_gauge(
                name=f"lmcache:periodic_threads_{level_name}_running",
                documentation=f"Number of running {level_name} level periodic threads",
                labelnames=labelnames,
                multiprocess_mode="livemostrecent",
            )

            self._create_dynamic_gauge(
                name=f"lmcache:periodic_threads_{level_name}_active",
                documentation=f"Number of active {level_name} level periodic threads",
                labelnames=labelnames,
                multiprocess_mode="livemostrecent",
            )

    def _log_gauge(self, gauge, data: Union[int, float]) -> None:
        # Convenience function for logging to gauge.
        gauge.labels(**self.labels).set(data)

    def _log_counter(self, counter, data: Union[int, float]) -> None:
        # Convenience function for logging to counter.
        # Prevent ValueError from negative increment
        if data < 0:
            return
        counter.labels(**self.labels).inc(data)

    def _log_histogram(self, histogram, data: Union[List[int], List[float]]) -> None:
        # Convenience function for logging to histogram.
        for value in data:
            histogram.labels(**self.labels).observe(value)

    def log_prometheus(self, stats: LMCacheStats):
        self._log_counter(
            self.counter_num_retrieve_requests, stats.interval_retrieve_requests
        )
        self._log_counter(
            self.counter_num_store_requests, stats.interval_store_requests
        )
        self._log_counter(
            self.counter_num_lookup_requests, stats.interval_lookup_requests
        )

        self._log_counter(
            self.counter_num_requested_tokens, stats.interval_requested_tokens
        )
        self._log_counter(self.counter_num_hit_tokens, stats.interval_hit_tokens)
        self._log_counter(self.counter_num_stored_tokens, stats.interval_stored_tokens)
        self._log_counter(self.counter_num_lookup_tokens, stats.interval_lookup_tokens)
        self._log_counter(self.counter_num_lookup_hits, stats.interval_lookup_hits)
        self._log_counter(self.counter_num_prompt_tokens, stats.interval_prompt_tokens)
        self._log_counter(
            self.counter_num_vllm_hit_tokens, stats.interval_vllm_hit_tokens
        )

        self._log_counter(
            self.counter_num_remote_read_requests,
            stats.interval_remote_read_requests,
        )
        self._log_counter(
            self.counter_num_remote_read_bytes, stats.interval_remote_read_bytes
        )
        self._log_counter(
            self.counter_num_remote_write_requests,
            stats.interval_remote_write_requests,
        )
        self._log_counter(
            self.counter_num_remote_write_bytes,
            stats.interval_remote_write_bytes,
        )
        self._log_counter(
            self.counter_local_cpu_evict_count,
            stats.interval_local_cpu_evict_count,
        )
        self._log_counter(
            self.counter_local_cpu_evict_keys_count,
            stats.interval_local_cpu_evict_keys_count,
        )
        self._log_counter(
            self.counter_local_cpu_evict_failed_count,
            stats.interval_local_cpu_evict_failed_count,
        )
        self._log_counter(
            self.counter_forced_unpin_count,
            stats.interval_forced_unpin_count,
        )
        self._log_counter(
            self.counter_lookup_0_hit_requests,
            stats.interval_lookup_0_hit_requests,
        )
        self._log_counter(
            self.counter_num_slow_retrieval_by_time,
            stats.interval_num_slow_retrieval_by_time,
        )
        self._log_counter(
            self.counter_num_slow_retrieval_by_speed,
            stats.interval_num_slow_retrieval_by_speed,
        )

        self._log_gauge(self.gauge_retrieve_hit_rate, stats.retrieve_hit_rate)

        self._log_gauge(self.gauge_lookup_hit_rate, stats.lookup_hit_rate)

        self._log_gauge(self.gauge_local_cache_usage, stats.local_cache_usage_bytes)

        self._log_gauge(self.gauge_remote_cache_usage, stats.remote_cache_usage_bytes)

        self._log_gauge(self.gauge_local_storage_usage, stats.local_storage_usage_bytes)

        self._log_histogram(self.histogram_time_to_retrieve, stats.time_to_retrieve)

        self._log_histogram(self.histogram_time_to_store, stats.time_to_store)

        self._log_histogram(self.histogram_time_to_lookup, stats.time_to_lookup)

        self._log_histogram(self.histogram_retrieve_speed, stats.retrieve_speed)

        self._log_histogram(self.histogram_store_speed, stats.store_speed)

        self._log_histogram(
            self.histogram_retrieve_process_tokens_time,
            stats.retrieve_process_tokens_time,
        )
        self._log_histogram(
            self.histogram_retrieve_broadcast_time, stats.retrieve_broadcast_time
        )
        self._log_histogram(
            self.histogram_retrieve_to_gpu_time, stats.retrieve_to_gpu_time
        )
        self._log_histogram(
            self.histogram_remote_backend_batched_get_blocking_time,
            stats.remote_backend_batched_get_blocking_time,
        )
        self._log_histogram(
            self.histogram_instrumented_connector_batched_get_time,
            stats.instrumented_connector_batched_get_time,
        )
        self._log_histogram(
            self.histogram_store_process_tokens_time, stats.store_process_tokens_time
        )
        self._log_histogram(
            self.histogram_store_from_gpu_time, stats.store_from_gpu_time
        )
        self._log_histogram(self.histogram_store_put_time, stats.store_put_time)

        self._log_histogram(
            self.histogram_p2p_time_to_transfer, stats.p2p_time_to_transfer
        )

        self._log_histogram(self.histogram_p2p_transfer_speed, stats.p2p_transfer_speed)

        self._log_histogram(
            self.histogram_remote_time_to_get, stats.interval_remote_time_to_get
        )
        self._log_histogram(
            self.histogram_remote_time_to_put, stats.interval_remote_time_to_put
        )
        self._log_histogram(
            self.histogram_remote_time_to_get_sync,
            stats.interval_remote_time_to_get_sync,
        )
        self._log_histogram(
            self.histogram_request_cache_hit_rate, stats.interval_lookup_hit_rates
        )
        self._log_histogram(
            self.histogram_request_cache_lifespan, stats.interval_request_cache_lifespan
        )
        self._log_gauge(
            self.gauge_remote_ping_latency, stats.interval_remote_ping_latency
        )
        self._log_counter(
            self.counter_remote_ping_errors, stats.interval_remote_ping_errors
        )
        self._log_counter(
            self.counter_remote_ping_successes, stats.interval_remote_ping_success
        )
        self._log_gauge(
            self.gauge_remote_ping_error_code, stats.interval_remote_ping_error_code
        )
        self._log_gauge(
            self.gauge_active_memory_objs_count, stats.active_memory_objs_count
        )
        self._log_gauge(
            self.gauge_pinned_memory_objs_count, stats.pinned_memory_objs_count
        )

    _instances: Dict[tuple[tuple[str, str], ...], "PrometheusLogger"] = {}

    @staticmethod
    @thread_safe
    def GetOrCreate(
        metadata: LMCacheMetadata,
        config: Optional["LMCacheEngineConfig"] = None,
    ) -> "PrometheusLogger":
        metadata_key = PrometheusLogger._metadata_to_key(metadata)
        if metadata_key in PrometheusLogger._instances:
            return PrometheusLogger._instances[metadata_key]

        base_logger = PrometheusLogger._get_base_logger()
        if base_logger is None:
            logger_instance = PrometheusLogger(metadata, config=config)
        else:
            logger_instance = PrometheusLogger._create_label_view(
                base_logger,
                metadata,
            )

        PrometheusLogger._instances[metadata_key] = logger_instance
        return logger_instance

    @staticmethod
    def _get_base_logger() -> Optional["PrometheusLogger"]:
        """
        Return an existing logger to reuse registered Prometheus collectors.
        """
        return next(iter(PrometheusLogger._instances.values()), None)

    @thread_safe
    def reset_counters(self) -> None:
        """
        Reset all Prometheus Counter metrics by calling clear().
        After clearing, re-initialize with labels so metrics remain visible.
        """
        label_views = self._reset_label_views()
        for counter in self._counters:
            counter.clear()
            # Re-initialize all known label views to keep each series visible.
            for label_view in label_views:
                counter.labels(**label_view.labels)

    @thread_safe
    def reset_histograms(self) -> None:
        """
        Reset all Prometheus Histogram metrics by calling clear().
        After clearing, re-initialize with labels so metrics remain visible.
        """
        label_views = self._reset_label_views()
        for histogram in self._histograms:
            histogram.clear()
            # Re-initialize all known label views to keep each series visible.
            for label_view in label_views:
                histogram.labels(**label_view.labels)

    @staticmethod
    def _metadata_to_labels(metadata: LMCacheMetadata) -> Dict[str, Any]:
        labels = {
            "model_name": metadata.model_name,
            "worker_id": metadata.worker_id,
            "role": metadata.role,
            "served_model_name": metadata.served_model_name or "",
        }
        return labels

    @staticmethod
    def _metadata_to_key(metadata: LMCacheMetadata) -> tuple[tuple[str, str], ...]:
        labels = PrometheusLogger._metadata_to_labels(metadata)
        return tuple(sorted((name, str(value)) for name, value in labels.items()))

    @staticmethod
    def _create_label_view(
        base_logger: "PrometheusLogger",
        metadata: LMCacheMetadata,
    ) -> "PrometheusLogger":
        """Reuse registered collectors with a different metadata/label view."""
        label_view = copy(base_logger)
        label_view.metadata = metadata
        label_view.labels = PrometheusLogger._metadata_to_labels(metadata)
        label_view._bind_dynamic_metric_children()
        return label_view

    def _reset_label_views(self) -> List["PrometheusLogger"]:
        """Return label views whose children must be restored after clear().

        Counter.clear() and Histogram.clear() remove every child from the
        shared collector, so reset must recreate children for all known labels.
        """
        views = list(PrometheusLogger._instances.values())
        if self not in views:
            views.append(self)
        return views


def reset_observability_metrics() -> None:
    """
    Reset observability metrics to their initial state.
    """

    prometheus_logger = PrometheusLogger._get_base_logger()
    if prometheus_logger is not None:
        prometheus_logger.reset_counters()
        prometheus_logger.reset_histograms()


class LMCacheStatsLogger:
    def __init__(
        self,
        metadata: LMCacheMetadata,
        log_interval: int,
        config: Optional["LMCacheEngineConfig"] = None,
    ):
        self.metadata = metadata
        self.log_interval = log_interval
        self.monitor = LMCStatsMonitor.GetOrCreate()
        self.prometheus_logger = PrometheusLogger.GetOrCreate(metadata, config=config)
        self.lmc_usage_logger = ContinuousUsageContext.GetOrCreate(metadata)
        self.is_running = True
        # Event for interruptible sleep during shutdown
        self.shutdown_event = threading.Event()

        self.thread = threading.Thread(
            target=self.log_worker, daemon=True, name="stats-logger-thread"
        )
        self.thread.start()

    def log_worker(self):
        while self.is_running:
            stats = self.monitor.get_stats_and_clear()
            self.prometheus_logger.log_prometheus(stats)
            self.lmc_usage_logger.incr_or_send_stats(stats)
            # Use Event.wait() instead of time.sleep() for interruptible sleep
            # Returns True if event was set, False if timeout occurred
            self.shutdown_event.wait(self.log_interval)

    def shutdown(self):
        """Shutdown the stats logger gracefully with immediate wake-up"""
        logger.info("Shutting down LMCacheStatsLogger...")

        # Signal the worker thread to stop
        self.is_running = False

        # Signal the event to wake up the thread immediately from sleep
        self.shutdown_event.set()

        # Wait for thread with a reasonable timeout
        if self.thread.is_alive():
            # Since we wake up the thread immediately, use a shorter timeout
            # Just enough time for the thread to finish its current iteration
            timeout = 5.0
            logger.info(
                f"Waiting for stats logger thread to finish (timeout: {timeout}s)..."
            )

            try:
                self.thread.join(timeout=timeout)

                if self.thread.is_alive():
                    logger.warning(
                        f"Stats logger thread did not terminate "
                        f"within {timeout}s timeout. "
                        "Thread may be blocked in logging operations. "
                        "Proceeding with shutdown anyway."
                    )
                else:
                    logger.info("Stats logger thread terminated successfully")
            except Exception as e:
                logger.error("Error waiting for stats logger thread: %s", e)
        else:
            logger.info("Stats logger thread already stopped")

        logger.info("LMCacheStatsLogger shutdown complete")