Skip to content

API documentation

Lognostic

Lognostic is a logging utility class designed to record, store, and analyze logging data. It captures logging information such as logger names, message sizes, and timestamps, providing functionalities to analyze this data for monitoring and debugging purposes.

Attributes:

Name Type Description
_lock Lock

A threading lock to ensure thread-safe operations on records.
_records List[Dict[str, Timestamp | str | int]]

A list to store logging records.
Source code in src/lognostic/lognostic.py 
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class Lognostic:
    """
    Lognostic is a logging utility class designed to record, store, and analyze logging data.
    It captures logging information such as logger names, message sizes, and timestamps,
    providing functionalities to analyze this data for monitoring and debugging purposes.

    Attributes:
        _lock (Lock): A threading lock to ensure thread-safe operations on records.
        _records (List[Dict[str, pd.Timestamp | str | int]]): A list to store logging records.
    """

    def __init__(self) -> None:
        """
        Initializes the Lognostic class by setting up the threading lock and initializing the list of records.
        """
        self._lock: Lock = Lock()
        self._records: List[Dict[str, Union[pd.Timestamp, str, int]]] = []

    def record(self, log_record: logging.LogRecord) -> None:
        """
        Records a new logging event by appending it to the list of records.

        Args:
            log_record (logging.LogRecord): The log record to be recorded, containing the logger's name,
            the log message, and other metadata.
        """
        logger_name: str = log_record.name
        message_size: int = len(log_record.getMessage())
        with self._lock:
            now = pd.Timestamp.now()
            self._records.append(
                {
                    "logger_name": logger_name,
                    "message_size": message_size,
                    "timestamp": now,
                }
            )

    def _dataframe(self) -> pd.DataFrame:
        """
        Converts internal records from a list of dictionaries to a pandas DataFrame.

        Returns:
            pd.DataFrame: A DataFrame containing all logging records.
        """
        with self._lock:
            return pd.DataFrame(
                self._records, columns=["logger_name", "message_size", "timestamp"]
            )

    def _get_recent_records(self, lookback_period: int) -> pd.DataFrame:
        """
        Retrieves records within the specified lookback period.

        Parameters:
            lookback_period (int): The lookback period in seconds.

        Returns:
            pd.DataFrame: A DataFrame containing recent logging records.
        """
        df = self._dataframe()
        time_window = pd.Timestamp.now() - pd.Timedelta(seconds=lookback_period)
        recent = df[df["timestamp"] > time_window]
        return recent

    def total_size(self) -> int:
        """
        Calculates the total size of all logged messages.

        Returns:
            int: The total size of all messages.
        """
        df = self._dataframe()
        total_size = df["message_size"].sum()
        return cast(int, total_size)

    def total_size_per_logger(self) -> Dict[str, int]:
        """
        Calculates the total size of logged messages per logger.

        Returns:
            Dict[str, int]: A dictionary mapping logger names to their total message size.
        """
        df = self._dataframe()
        return df.groupby("logger_name")["message_size"].sum().to_dict()

    def total_logging_rate(self, lookback_period: int = 60) -> float:
        """
        Calculates the total logging rate over a specified lookback period.

        Parameters:
            lookback_period (int, optional): The lookback period in seconds. Defaults to 60.

        Returns:
            float: The average logging rate (message size per second).
        """
        recent_records = self._get_recent_records(lookback_period)
        total_rate = recent_records["message_size"].sum() / lookback_period
        return cast(float, total_rate)

    def logging_rate_per_logger(self, lookback_period: int = 60) -> Dict[str, float]:
        """
        Calculates the logging rate per logger over a specified lookback period.

        Parameters:
            lookback_period (int, optional): The lookback period in seconds. Defaults to 60.

        Returns:
            Dict[str, float]: A dictionary mapping logger names to their average logging rate (message size per second).
        """
        recent_records = self._get_recent_records(lookback_period)
        return (
            recent_records.groupby("logger_name")["message_size"].sum()
            / lookback_period
        ).to_dict()

__init__()

Initializes the Lognostic class by setting up the threading lock and initializing the list of records.

Source code in src/lognostic/lognostic.py 
19
20
21
22
23
24
def __init__(self) -> None:
    """
    Initializes the Lognostic class by setting up the threading lock and initializing the list of records.
    """
    self._lock: Lock = Lock()
    self._records: List[Dict[str, Union[pd.Timestamp, str, int]]] = []

logging_rate_per_logger(lookback_period=60)

Calculates the logging rate per logger over a specified lookback period.

Parameters:

Name Type Description Default
lookback_period int

The lookback period in seconds. Defaults to 60.

 60

Returns:

Type Description
Dict[str, float]

Dict[str, float]: A dictionary mapping logger names to their average logging rate (message size per second).
Source code in src/lognostic/lognostic.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def logging_rate_per_logger(self, lookback_period: int = 60) -> Dict[str, float]:
    """
    Calculates the logging rate per logger over a specified lookback period.

    Parameters:
        lookback_period (int, optional): The lookback period in seconds. Defaults to 60.

    Returns:
        Dict[str, float]: A dictionary mapping logger names to their average logging rate (message size per second).
    """
    recent_records = self._get_recent_records(lookback_period)
    return (
        recent_records.groupby("logger_name")["message_size"].sum()
        / lookback_period
    ).to_dict()

record(log_record)

Records a new logging event by appending it to the list of records.

Parameters:

Name Type Description Default
log_record LogRecord

The log record to be recorded, containing the logger's name,

 required
Source code in src/lognostic/lognostic.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def record(self, log_record: logging.LogRecord) -> None:
    """
    Records a new logging event by appending it to the list of records.

    Args:
        log_record (logging.LogRecord): The log record to be recorded, containing the logger's name,
        the log message, and other metadata.
    """
    logger_name: str = log_record.name
    message_size: int = len(log_record.getMessage())
    with self._lock:
        now = pd.Timestamp.now()
        self._records.append(
            {
                "logger_name": logger_name,
                "message_size": message_size,
                "timestamp": now,
            }
        )

total_logging_rate(lookback_period=60)

Calculates the total logging rate over a specified lookback period.

Parameters:

Name Type Description Default
lookback_period int

The lookback period in seconds. Defaults to 60.

 60

Returns:

Name Type Description
float float

The average logging rate (message size per second).
Source code in src/lognostic/lognostic.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def total_logging_rate(self, lookback_period: int = 60) -> float:
    """
    Calculates the total logging rate over a specified lookback period.

    Parameters:
        lookback_period (int, optional): The lookback period in seconds. Defaults to 60.

    Returns:
        float: The average logging rate (message size per second).
    """
    recent_records = self._get_recent_records(lookback_period)
    total_rate = recent_records["message_size"].sum() / lookback_period
    return cast(float, total_rate)

total_size()

Calculates the total size of all logged messages.

Returns:

Name Type Description
int int

The total size of all messages.
Source code in src/lognostic/lognostic.py 
73
74
75
76
77
78
79
80
81
82
def total_size(self) -> int:
    """
    Calculates the total size of all logged messages.

    Returns:
        int: The total size of all messages.
    """
    df = self._dataframe()
    total_size = df["message_size"].sum()
    return cast(int, total_size)

total_size_per_logger()

Calculates the total size of logged messages per logger.

Returns:

Type Description
Dict[str, int]

Dict[str, int]: A dictionary mapping logger names to their total message size.
Source code in src/lognostic/lognostic.py 
84
85
86
87
88
89
90
91
92
def total_size_per_logger(self) -> Dict[str, int]:
    """
    Calculates the total size of logged messages per logger.

    Returns:
        Dict[str, int]: A dictionary mapping logger names to their total message size.
    """
    df = self._dataframe()
    return df.groupby("logger_name")["message_size"].sum().to_dict()