denet 0.7.0

a simple process monitor
Documentation 
1
2
3
4
5
6
7
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

#!/usr/bin/env python3
"""
Example demonstrating child process monitoring with denet.

This example shows how to use denet to monitor a process that spawns child processes,
properly capturing and aggregating metrics across the entire process tree.
"""

import json
import os
import sys
import time
from denet import execute_with_monitoring


def run_stress_test():
    """Run a CPU-intensive stress test that spawns multiple child processes."""
    print(f"Parent process: {os.getpid()}")

    # Create child processes for CPU-intensive work
    # This is a simple way to spawn child processes in Python
    for i in range(4):
        pid = os.fork()
        if pid == 0:  # Child process
            # Each child process will do CPU-intensive work
            print(f"Child process {i + 1}: {os.getpid()}")
            start_time = time.time()
            # Run for about 2 seconds
            while time.time() - start_time < 2:
                # Do some CPU-intensive calculations
                _ = [i * i * i for i in range(100000)]

            # Exit the child process
            sys.exit(0)

    # Parent process waits for children to finish
    print("Parent waiting for children to complete...")
    for _ in range(4):
        os.wait()

    print("All child processes completed!")
    return 0


def main():
    """Run the example and display the monitoring results."""
    print("Starting child process monitoring example...")

    # Run the stress test with monitoring
    # Note the include_children=True parameter which is key for monitoring process trees
    exit_code, monitor = execute_with_monitoring(
        cmd=[
            sys.executable,
            "-c",
            "import os, sys, time; "
            "print(f'Parent: {os.getpid()}'); "
            # Create 4 child processes
            "for i in range(4): "
            "    pid = os.fork(); "
            "    if pid == 0: "  # Child process
            "        print(f'Child {i+1}: {os.getpid()}'); "
            "        start = time.time(); "
            "        while time.time() - start < 3: "  # Each child runs for 3 seconds
            "            _ = [i*i*i for i in range(50000)]; "  # CPU-intensive work
            "        sys.exit(0); "
            # Parent waits for all children
            "for _ in range(4): os.wait(); "
            "print('All children completed');",
        ],
        base_interval_ms=100,  # Sample every 100ms
        max_interval_ms=500,  # Maximum interval 500ms
        store_in_memory=True,  # Keep samples in memory
        output_file="stress_monitor.jsonl",  # Also write to file
        include_children=True,  # This is critical for monitoring child processes
    )

    # Process and display the results
    samples = monitor.get_samples()

    # Get summary statistics
    summary_json = monitor.get_summary()
    summary = json.loads(summary_json)

    print("\n===== Monitoring Results =====")
    print(f"Total samples collected: {len(samples)}")
    print(f"Average CPU usage: {summary['avg_cpu_usage']:.2f}%")
    print(f"Peak memory usage: {summary['peak_mem_rss_kb'] / 1024:.2f} MB")
    print(f"Maximum processes monitored: {summary['max_processes']}")

    # Show the process count over time
    process_counts = []
    for sample in samples:
        if isinstance(sample, str):
            sample = json.loads(sample)

        # Look for process_count in aggregated data
        if "aggregated" in sample:
            count = sample["aggregated"].get("process_count", 1)
            process_counts.append(count)
        elif "process_count" in sample:
            process_counts.append(sample["process_count"])

    if process_counts:
        print("\nProcess count over time:")
        for i, count in enumerate(process_counts):
            print(f"Sample {i + 1}: {count} processes")

    print("\nExample complete. Results written to stress_monitor.jsonl")

    return exit_code


if __name__ == "__main__":
    sys.exit(main())