Add self monitoring (#8)

Author: onrylmz

README.md

@@ -12,7 +12,7 @@ A high-performance load generator for testing log ingestion pipelines, observabi
 - **Continuous mode**: Maximum speed testing with `--period 0`
 - **Real-time stats**: Current and average throughput metrics
 - **Backpressure detection**: Tracks 429/503 responses
-- **Process monitoring**: Monitor CPU, memory, and threads of any process
+- **Process monitoring**: Monitor CPU, memory, and threads of target processes and loadgen itself
 - **Cross-platform**: Supports macOS and Linux
 - **Zero dependencies**: Only requires `gofakeit` for data generation
 
@@ -121,6 +121,9 @@ Process Monitoring Options:
   -monitor-pid int
         PID of process to monitor (0 means disabled)
 
+  -monitor-self
+        Monitor loadgen's own resource usage (default false)
+
   -monitor-interval duration
         Interval for process monitoring stats (default 5s)
 ```
@@ -248,7 +251,7 @@ Output (JSON array):
 
 ### Process Monitoring
 
-Monitor a process without generating load:
+Monitor a target process without generating load:
 
 ```bash
 # Monitor by process name
@@ -263,25 +266,24 @@ Monitor a process without generating load:
 
 **Output:**
 ```
-[MONITOR] pid: 12345 | cpu: 45.3% | memory: 234.5MB | threads: 28
+[MONITOR - TARGET] edgedelta | pid: 12345 | cpu: 45.3% | memory: 234.5MB | threads: 28
 ```
 
-Monitor a process while generating load:
+Monitor loadgen's own resource usage:
 
 ```bash
-# Observe how load impacts the monitored process
-./loadgen \
-    --endpoint http://localhost:8085 \
-    --format nginx_log \
-    --workers 48 \
-    --period 10ms \
-    --monitor-process edgedelta
+# Monitor self only
+./loadgen --monitor-self --endpoint http://localhost:8085 --workers 100
+
+# Monitor both loadgen and target process
+./loadgen --monitor-self --monitor-process edgedelta --endpoint http://localhost:8085 --workers 100
 ```
 
-**Output:**
+**Output (dual monitoring):**
 ```
 [STATS] current: 1000 logs/sec, 0.64 MB/s | avg: 1000 logs/sec | total: 5000 | errors: 0 | backpressure: 0 (0.0%)
-[MONITOR] pid: 12345 | cpu: 45.3% | memory: 234.5MB | threads: 28
+[MONITOR - SELF] loadgen | pid: 8045 | cpu: 15.2% | memory: 45.3MB | threads: 11
+[MONITOR - TARGET] edgedelta | pid: 12345 | cpu: 45.3% | memory: 234.5MB | threads: 28
 ```
 
 **Monitoring Metrics:**
@@ -316,6 +318,9 @@ The tool prints real-time statistics every 5 seconds:
    - Smaller pools (10-50): Less memory, faster startup, sufficient for most load tests
    - Larger pools (500-1000): More variety, useful for testing unique value handling
 6. **Format style**: NDJSON (`--format-style ndjson`) is default and works with most log systems
+7. **Self-monitoring**: Use `--monitor-self` to ensure loadgen isn't the bottleneck
+   - If loadgen CPU is maxed out, reduce workers or use multiple instances
+   - Helps distinguish between generator and target system performance issues
 
 ## How It Works
 

main.go

@@ -33,6 +33,7 @@ type Config struct {
 	Workers         int
 	MonitorPID      int
 	MonitorProcess  string
+	MonitorSelf     bool
 	MonitorInterval time.Duration
 	PayloadPoolSize int
 }
@@ -188,40 +189,68 @@ func main() {
 		}()
 	}
 
-	// Determine monitoring PID if monitoring is enabled.
-	var processStats *ProcessStats
+	// Determine monitoring targets.
+	var targetStats *ProcessStats
+	var selfStats *ProcessStats
+
 	if config.MonitorProcess != "" {
 		pid, err := findProcessByName(config.MonitorProcess)
 		if err != nil {
 			log.Fatalf("Failed to find process '%s': %v", config.MonitorProcess, err)
 		}
-		processStats = &ProcessStats{
+		targetStats = &ProcessStats{
 			pid:         pid,
 			processName: config.MonitorProcess,
+			monitorType: MonitorTypeTarget,
 		}
 		log.Printf("Found process '%s' with PID: %d", config.MonitorProcess, pid)
 	} else if config.MonitorPID > 0 {
-		processStats = &ProcessStats{
-			pid: config.MonitorPID,
+		targetStats = &ProcessStats{
+			pid:         config.MonitorPID,
+			monitorType: MonitorTypeTarget,
 		}
 		log.Printf("Monitoring process PID: %d", config.MonitorPID)
 	}
 
+	if config.MonitorSelf {
+		selfStats = &ProcessStats{
+			pid:         os.Getpid(),
+			processName: "loadgen",
+			monitorType: MonitorTypeSelf,
+		}
+		log.Printf("Monitoring self PID: %d", selfStats.pid)
+	}
+
 	// Determine mode: monitoring-only vs concurrent vs normal.
-	monitoringEnabled := processStats != nil
+	monitoringEnabled := targetStats != nil || selfStats != nil
 	endpointProvided := flag.Lookup("endpoint").Value.String() != flag.Lookup("endpoint").DefValue
 
 	if monitoringEnabled && !endpointProvided {
 		// Mode 2: Standalone monitoring only.
 		log.Printf("Starting monitoring-only mode (interval: %s)", config.MonitorInterval)
-		runMonitoringOnly(ctx, processStats, config.MonitorInterval)
+
+		if targetStats != nil {
+			go monitorProcess(ctx, targetStats, config.MonitorInterval)
+		}
+		if selfStats != nil {
+			go monitorProcess(ctx, selfStats, config.MonitorInterval)
+		}
+
+		// Block until context is cancelled.
+		<-ctx.Done()
+		log.Println("Monitoring stopped")
 		return
 	}
 
 	if monitoringEnabled {
 		// Mode 1: Concurrent monitoring + load generation.
 		log.Printf("Starting concurrent monitoring (interval: %s)", config.MonitorInterval)
-		go monitorProcess(ctx, processStats, config.MonitorInterval)
+		if targetStats != nil {
+			go monitorProcess(ctx, targetStats, config.MonitorInterval)
+		}
+		if selfStats != nil {
+			go monitorProcess(ctx, selfStats, config.MonitorInterval)
+		}
 	}
 
 	// Normal load generation (with or without monitoring).
@@ -245,6 +274,7 @@ func parseFlags() *Config {
 	workers := flag.Int("workers", 1, "Number of concurrent workers")
 	monitorPID := flag.Int("monitor-pid", 0, "PID of process to monitor (0 means disabled)")
 	monitorProcess := flag.String("monitor-process", "", "Process name to monitor (e.g., 'edgedelta')")
+	monitorSelf := flag.Bool("monitor-self", false, "Monitor loadgen's own resource usage")
 	monitorInterval := flag.Duration("monitor-interval", 5*time.Second, "Interval for process monitoring stats")
 	payloadPoolSize := flag.Int("payload-pool-size", 100, "Number of unique payloads to generate in the pool")
 	flag.Parse()
@@ -261,6 +291,7 @@ func parseFlags() *Config {
 		Workers:         *workers,
 		MonitorPID:      *monitorPID,
 		MonitorProcess:  *monitorProcess,
+		MonitorSelf:     *monitorSelf,
 		MonitorInterval: *monitorInterval,
 		PayloadPoolSize: *payloadPoolSize,
 	}

monitor.go

@@ -13,6 +13,16 @@ import (
 	"time"
 )
 
+// MonitorType represents the type of process monitoring.
+type MonitorType string
+
+const (
+	// MonitorTypeSelf indicates monitoring of loadgen's own process.
+	MonitorTypeSelf MonitorType = "SELF"
+	// MonitorTypeTarget indicates monitoring of a target process.
+	MonitorTypeTarget MonitorType = "TARGET"
+)
+
 // ProcessStats tracks process monitoring metrics.
 type ProcessStats struct {
 	mu          sync.Mutex
@@ -21,6 +31,7 @@ type ProcessStats struct {
 	memoryMB    float64
 	threadCount int
 	processName string
+	monitorType MonitorType
 }
 
 func (ps *ProcessStats) update(cpu float64, mem float64, threads int) {
@@ -34,8 +45,15 @@ func (ps *ProcessStats) update(cpu float64, mem float64, threads int) {
 func (ps *ProcessStats) print() {
 	ps.mu.Lock()
 	defer ps.mu.Unlock()
-	fmt.Printf("[MONITOR] pid: %d | cpu: %.1f%% | memory: %.1fMB | threads: %d\n",
-		ps.pid, ps.cpuPercent, ps.memoryMB, ps.threadCount)
+
+	prefix := fmt.Sprintf("[MONITOR - %s]", ps.monitorType)
+	if ps.processName != "" {
+		fmt.Printf("%s %s | pid: %d | cpu: %.1f%% | memory: %.1fMB | threads: %d\n",
+			prefix, ps.processName, ps.pid, ps.cpuPercent, ps.memoryMB, ps.threadCount)
+	} else {
+		fmt.Printf("%s pid: %d | cpu: %.1f%% | memory: %.1fMB | threads: %d\n",
+			prefix, ps.pid, ps.cpuPercent, ps.memoryMB, ps.threadCount)
+	}
 }
 
 // findProcessByName finds a process PID by name using pgrep.
@@ -264,36 +282,3 @@ func monitorProcess(ctx context.Context, stats *ProcessStats, interval time.Dura
 		}
 	}
 }
-
-// runMonitoringOnly runs monitoring as the main execution path (blocking).
-func runMonitoringOnly(ctx context.Context, stats *ProcessStats, interval time.Duration) {
-	log.Printf("Monitoring process PID %d (press Ctrl+C to stop)", stats.pid)
-
-	ticker := time.NewTicker(interval)
-	defer ticker.Stop()
-
-	sampleInterval := calculateSampleInterval(interval)
-
-	cpu, mem, threads, err := collectProcessMetrics(stats.pid, sampleInterval)
-	if err != nil {
-		log.Fatalf("Failed to collect initial metrics: %v", err)
-	}
-	stats.update(cpu, mem, threads)
-	stats.print()
-
-	for {
-		select {
-		case <-ctx.Done():
-			log.Println("Monitoring stopped")
-			return
-		case <-ticker.C:
-			cpu, mem, threads, err := collectProcessMetrics(stats.pid, sampleInterval)
-			if err != nil {
-				log.Printf("Process terminated or monitoring failed: %v", err)
-				return
-			}
-			stats.update(cpu, mem, threads)
-			stats.print()
-		}
-	}
-}