指标

MetricModelType module-attribute

MetricModelType = TypeVar('MetricModelType', bound=MetricModel)

T module-attribute

T = TypeVar('T')

__doc__ module-attribute

__doc__ = '\n指标模块\n'

IMetricCollector

指标收集器

源代码位于： logis/metric/__init__.py

class IMetricCollector(ABC):
    """
    指标收集器
    """

    def __init__(self, write_buffer: Optional[WriteBuffer] = None, **kwargs):
        super().__init__(**kwargs)
        self._buffer = write_buffer or AsyncWriteBuffer(
            WriteBufferConfig(batch_size=100, flush_interval=10, handler=self.submit)
        )
        invoke(self._buffer.start)

    @abstractmethod
    def submit(self, metrics: List[T]):
        """
        提交指标
        """
        pass

    def collect(self, metric: T):
        """
        缓冲指标
        """
        invoke(self._buffer.put, metric)

    def flush(self, flush_all: bool = False, **kwargs):
        """
        刷新指标
        """
        invoke(self._buffer.flush, flush_all=flush_all, **kwargs)

__init__

__init__(write_buffer: Optional[WriteBuffer] = None, **kwargs)

源代码位于： logis/metric/__init__.py

def __init__(self, write_buffer: Optional[WriteBuffer] = None, **kwargs):
    super().__init__(**kwargs)
    self._buffer = write_buffer or AsyncWriteBuffer(
        WriteBufferConfig(batch_size=100, flush_interval=10, handler=self.submit)
    )
    invoke(self._buffer.start)

collect

collect(metric: T)

缓冲指标

源代码位于： logis/metric/__init__.py

def collect(self, metric: T):
    """
    缓冲指标
    """
    invoke(self._buffer.put, metric)

flush

flush(flush_all: bool = False, **kwargs)

刷新指标

源代码位于： logis/metric/__init__.py

def flush(self, flush_all: bool = False, **kwargs):
    """
    刷新指标
    """
    invoke(self._buffer.flush, flush_all=flush_all, **kwargs)

submit abstractmethod

submit(metrics: List[T])

提交指标

源代码位于： logis/metric/__init__.py

@abstractmethod
def submit(self, metrics: List[T]):
    """
    提交指标
    """
    pass

MetricModel

指标模型

源代码位于： logis/metric/__init__.py

class MetricModel(ABC):
    """
    指标模型
    """

    @classmethod
    @abstractmethod
    def get_measurement(
        cls, prefix: Optional[str] = None, suffix: Optional[str] = None, sep: str = ""
    ) -> str:
        """
        获取指标名称
        """
        pass

    def to_influxdb_point(self, measurement: Optional[str] = None):
        """
        转换为InfluxDB点
        """
        raise NotImplementedError("to_influxdb_point")

get_measurement abstractmethod classmethod

get_measurement(prefix: Optional[str] = None, suffix: Optional[str] = None, sep: str = '') -> str

获取指标名称

源代码位于： logis/metric/__init__.py

@classmethod
@abstractmethod
def get_measurement(
    cls, prefix: Optional[str] = None, suffix: Optional[str] = None, sep: str = ""
) -> str:
    """
    获取指标名称
    """
    pass

to_influxdb_point

to_influxdb_point(measurement: Optional[str] = None)

转换为InfluxDB点

源代码位于： logis/metric/__init__.py

def to_influxdb_point(self, measurement: Optional[str] = None):
    """
    转换为InfluxDB点
    """
    raise NotImplementedError("to_influxdb_point")