Stats

This page contains the API documentation for the gradual.reporting.stats module.

gradual.reporting.stats

The stats module provides the Stats class which handles collection and processing of test statistics in the stress testing framework. It implements a singleton pattern and uses multiprocessing for efficient stats collection and processing.

Classes

Stats(phase_config: PhaseConfig, run_name: str)

Singleton class for managing test statistics collection and processing.

This class provides functionality for: 1. Collecting test statistics in a thread-safe manner 2. Processing statistics in a separate process 3. Managing test timing and runtime tracking 4. Supporting database persistence of statistics

Attributes:

Name	Type	Description
_instance	Stats	Singleton instance of the Stats class
stop_writing	Event	Event to signal when to stop processing stats
stats_queue	Queue	Queue for passing stats between processes
phase_config	PhaseConfig	Configuration for the current test phase
test_start_time	int	Timestamp when the test started
test_end_time	int	Timestamp when the test ended
write_db_process	Process	Process for handling stats persistence
run_name	str	Name of the current test run

Initialize a new Stats instance.

Parameters:

Name	Type	Description	Default
phase_config	PhaseConfig	Configuration for the current test phase	required
run_name	str	Name of the current test run	required

Source code in src/gradual/reporting/stats.py

def __init__(self, phase_config: PhaseConfig, run_name: str):
    """
    Initialize a new Stats instance.

    Args:
        phase_config (PhaseConfig): Configuration for the current test phase
        run_name (str): Name of the current test run
    """
    self.stats_queue: Queue = Queue()
    self.phase_config = phase_config
    self.test_start_time: int
    self.test_end_time: int
    self.write_db_process = Process(target=self.process_stats, args=())
    self.run_name = run_name

Attributes

stop_writing = Event() class-attribute instance-attribute

stats_queue: Queue = Queue() instance-attribute

phase_config = phase_config instance-attribute

test_start_time: int instance-attribute

test_end_time: int instance-attribute

write_db_process = Process(target=(self.process_stats), args=()) instance-attribute

run_name = run_name instance-attribute

Functions

start_process_stats()

Start the process that processes the stats.

Source code in src/gradual/reporting/stats.py

def start_process_stats(self):
    """
    Start the process that processes the stats.
    """
    self.write_db_process.start()

close_process_stats()

Terminate the process that processes the stats.

Source code in src/gradual/reporting/stats.py

def close_process_stats(self):
    """
    Terminate the process that processes the stats.
    """
    self.write_db_process.terminate()

process_stats()

Process statistics in a separate process.

This method runs in a separate process and: 1. Listens to the stats queue for new statistics 2. Processes received statistics using the provided adapters 3. Continues until stop_writing event is set

Note

The method uses a timeout of 1 second when waiting for new stats to allow for graceful shutdown. The timeout is used to avoid blocking the main process.

Source code in src/gradual/reporting/stats.py

def process_stats(self):
    """
    Process statistics in a separate process.

    This method runs in a separate process and:
    1. Listens to the stats queue for new statistics
    2. Processes received statistics using the provided adapters
    3. Continues until stop_writing event is set

    Note:
        The method uses a timeout of 1 second when waiting for new stats
        to allow for graceful shutdown. The timeout is used to avoid
        blocking the main process.
    """
    while not self.stop_writing.is_set():
        try:
            stats, adapters = self.stats_queue.get(
                timeout=1
            )  # wait up to 1 sec for a stat to be available
            for adapter in adapters:
                try:
                    adapter.process_stats(stats)
                except Exception as e:
                    error(
                        f"Stat processing failed with {adapter} with exception: {e}"
                    )
                    error(traceback.format_exc())
        except Empty:
            # queue is empty. so do nothing
            pass

persist_stats(stats, adapters: Optional[list[Adapter]] = None)

Add statistics to the processing queue.

Parameters:

Name	Type	Description	Default
stats		Statistics to be processed and persisted	required
adapters	Optional[list[Adapter]]	Adapters to be used to process the stats.	None

Source code in src/gradual/reporting/stats.py

def persist_stats(self, stats, adapters: Optional[list[Adapter]] = None):
    """
    Add statistics to the processing queue.

    Args:
        stats: Statistics to be processed and persisted
        adapters: Adapters to be used to process the stats.
    """
    if adapters is None:
        adapters = [LoggingAdapter()]
    self.stats_queue.put((stats, adapters))

get_stats_instance() classmethod

Get the singleton instance of the Stats class.

Returns:

Name	Type	Description
Stats		The singleton instance

Source code in src/gradual/reporting/stats.py

@classmethod
def get_stats_instance(cls):
    """
    Get the singleton instance of the Stats class.

    Returns:
        Stats: The singleton instance
    """
    return cls._instance

current_runtime()

Calculate the current runtime of the test.

Returns:

Name	Type	Description
float		Time elapsed since test start in seconds

Source code in src/gradual/reporting/stats.py

def current_runtime(self):
    """
    Calculate the current runtime of the test.

    Returns:
        float: Time elapsed since test start in seconds
    """
    return time.time() - self.test_start_time