Welcome to Stress Injector’s documentation!

Read Me:

StressInjector - CPU Stress

class stressinjector.cpu.CPUStress(seconds: int = 40, logger: Logger = None)

Controller for CPU stress using multiprocessing. Gets duration as user input.

>>> CPUStress

CPU is stressed using multiprocessing.Process to run the infinite loop on each process.

Warning

  • CPU stress is induced in real time.

  • A relatively low performing machine may stall when stress is induced for a long duration.

References

>>> CPUStress._infinite()
    Triggers an infinite loop for the number of logical cores.

>>> CPUStress._measure_cpu()
    Measures the impact on each logical core in a dedicated thread.

Instantiates the members of the class.

Parameters:

  • seconds – The number of seconds CPU has to be stressed. Defaults to five times the number of cores.

  • logger – Custom logger.

classmethod _format_number(n: float) → int

Converts numbers with float value .0 to integers.

Parameters:

n – Raw numbers which is either a str or float

Returns:

Processed integers without any float value extensions in it.

Return type:

int

_infinite() → None

Infinite loop to stress each core on the CPU for the number of logical cores available.

See also

The loop runs on each core as this function is triggered by processing.Process that runs as a loop.

_measure_cpu() → None

Uses cpu_percent() to get the current CPU utilization and print the utilization percentage on each core.

Runs in a forever loop. Stops when the flag stop_thread is set to True.

_run() → None

Initiator for stress injector.

infinite()

To kick off stress injector.

measure()

To measure the usage in the background running in a dedicated thread.

StressInjector - Memory Stress

class stressinjector.memory.MemoryStress(gigabytes: int = 32, logger: Logger = None)

Controller to trigger the memory stress. Gets number of GBs as user input.

>>> MemoryStress
Parameters:

gigabytes –

  • The number of gigabytes for which the memory has to be stressed. Defaults to twice the number of GBs.

See also

Suggests twice the amount of physical memory.

Warning

  • Memory stress is induced in real time.

  • A low RAM equipped machine may stall or be un-responsive when stress is induced for a higher byte value.

References

>>> MemoryStress._stress()

Generates random bytes, 1024 times the GigaBytes value entered during prompt or class intialization.

>>> _size_converter()

Converts bytes to human-readable size format.

Instantiates the members of the class.

Parameters:

  • gigabytes – The number of gigabytes, memory has to be stressed. Defaults to twice the physical memory.

  • logger – Custom logger.

classmethod _memory_util_check() → Union[int, float]

Returns memory used only the current script.

Returns:

The memory used by the current process.

Return type:

int or float

References

macOS or Linux:
>>> resource.getrusage(resource.RUSAGE_SELF).ru_maxrss

getrusage

Windows:
>>> psutil.Process(settings.pid).memory_info().peak_wset

memory_info

_run() → None

Initiator for stress injector. Converts GigaBytes to Bytes.

stress()

To kick off stress injector with the desired bytes converted from user input.

memory_util_check()

To measure the usage post completion.

classmethod _stress(mb: int) → str

Generates random bytes.

Bytes are generated with the multiple of 1024 ~ 1GB. Uses tqdm module to show a progress bar.

Parameters:

mb – The number of MegaBytes of stress has be induced A.k.a. random bytes have to be generated.

Returns:

Calls the size_converter method to get the human-readable size of stress that was induced.

Return type:

str

stressinjector.memory._size_converter(byte_size: Union[int, float]) → str

Gets the current memory consumed and converts it to human friendly format.

Parameters:

byte_size – Receives byte size as argument.

Returns:

Converted human understandable size.

Return type:

str

StressInjector - URL Stress

class stressinjector.url.URLStress(url: str, rate: int = 100000.0, timeout: Union[int, float] = None, logger: Logger = None, request_type: str = RequestType.get, **kwargs)

Controller for URL stress using threadpool. Gets url as input.

>>> URLStress

Instantiate the object, parse and validate the URL.

Parameters:

  • url – URL to inject stress.

  • rate – Number of calls to make.

  • timeout – Timeout for each request.

  • retry_limit – Retry limit if the system is unable to spinup more threads.

  • circuit_break – Wait time in seconds between retries.

  • logger – Custom logger.

  • request_type – Function from requests module.

  • kwargs – Keyword arguments to use in the request.

_run() → None

Runs initiate request injection and prints success and error count.

initiate_injection() → bool

Initiates injection in a thread pool.

Returns:

Returns a boolean flag based on successful completion.

Return type:

bool

make_request(sample: bool = False) → None

Makes a GET request to the endpoint.

Parameters:

sample – Boolean flag to indicate if the request is sample.

StressInjector - Models

class stressinjector.models.OperatingSystem(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Wrapper for supported operating systems.

_generate_next_value_(start, count, last_values)

Generate the next value when not given.

name: the name of the member start: the initial start value or None count: the number of existing members last_values: the list of values assigned

_member_type_

alias of str

_new_member_(**kwargs)

Create and return a new object. See help(type) for accurate signature.

_value_repr_()

Return repr(self).

class stressinjector.models.RequestType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Wrapper for request types.

_generate_next_value_(start, count, last_values)

Generate the next value when not given.

name: the name of the member start: the initial start value or None count: the number of existing members last_values: the list of values assigned

_member_type_

alias of str

_new_member_(**kwargs)

Create and return a new object. See help(type) for accurate signature.

_value_repr_()

Return repr(self).

class stressinjector.models.Settings

Wrapper for settings.

exception stressinjector.models.UnsupportedOS

Custom OSError for unsupported operating system.

stressinjector.models.operating_system

alias of OperatingSystem

stressinjector.models.settings

alias of Settings

StressInjector - Helper

stressinjector.helper.flush_screen() → None

Flushes the screen output.

See also

Writes new set of empty strings for the size of the terminal if ran using one.

stressinjector.helper.write_screen(text: Any) → None

Write text to a screen that can be cleared later.

Parameters:

text – Text to be written.

Indices and tables