Merge pull request #50 from WarpBuilds/obs

Observability improvements and github log collector

Author: suryaoruganti

.github/workflows/release-testing.yaml

@@ -3,6 +3,7 @@ name: Branch Build and Upload to R2
 on:
   push:
     branches:
+      - "obs"
       - feat-transparent-cache
 
 env:

README.md

@@ -1,2 +1,13 @@
 # warpbuild-agent
+
 Application for runner lifecycle management
+
+## Generating the API client from the OpenAPI specification in backend-core
+
+Use Redocly to generate the API client.
+
+```bash
+./scripts/generate-from-openapi-redocly.sh pkg/warpbuild --go 1.0.0
+```
+
+The filters in redocly.yaml are used to generate the API client.

pkg/telemetry/binaries/windows/amd64/README.md

@@ -8,7 +8,6 @@ import (
 	"path/filepath"
 	"runtime"
 	"sync"
-	"syscall"
 	"text/template"
 	"time"
 
@@ -25,8 +24,9 @@ type TelemetryManager struct {
 	mu     sync.RWMutex
 
 	// Components
-	receiver   *uploader.Receiver
-	s3Uploader *uploader.S3Uploader
+	receiver         *uploader.Receiver
+	s3Uploader       *uploader.S3Uploader
+	otelCollectorCmd *exec.Cmd
 
 	// Configuration
 	port          int
@@ -78,6 +78,10 @@ func (tm *TelemetryManager) Start() error {
 	tm.wg.Add(1)
 	go tm.startOtelCollector()
 
+	// Start telemetry status monitoring
+	tm.wg.Add(1)
+	go tm.monitorTelemetryStatus()
+
 	log.Logger().Infof("Telemetry manager started successfully")
 	return nil
 }
@@ -129,26 +133,16 @@ func (tm *TelemetryManager) startOtelCollector() {
 
 	log.Logger().Infof("OpenTelemetry Collector configuration written successfully")
 
-	// Channel to signal when the application should terminate
-	done := make(chan bool, 1)
-
-	// Start OpenTelemetry Collector Contrib
-	go func() {
-		defer tm.handlePanic()
-		log.Logger().Infof("Launching OpenTelemetry Collector in background...")
-		tm.runOtelCollector(collectorPath, done)
-	}()
+	log.Logger().Infof("Launching OpenTelemetry Collector in background...")
 
-	// Wait for context cancellation
-	<-tm.ctx.Done()
-	log.Logger().Infof("Context cancelled, stopping OTEL collector...")
+	// Start the OTEL collector and wait for context cancellation
+	tm.runOtelCollector(collectorPath)
 
-	// Signal the OpenTelemetry Collector process to terminate
-	done <- true
+	log.Logger().Infof("OTEL collector goroutine exited")
 }
 
 // runOtelCollector runs the OTEL collector process
-func (tm *TelemetryManager) runOtelCollector(collectorPath string, done chan bool) {
+func (tm *TelemetryManager) runOtelCollector(collectorPath string) {
 	configPath := tm.getConfigFilePath()
 	log.Logger().Infof("Starting OpenTelemetry Collector with config: %s", configPath)
 
@@ -160,29 +154,57 @@ func (tm *TelemetryManager) runOtelCollector(collectorPath string, done chan boo
 
 	log.Logger().Infof("OpenTelemetry Collector command: %s --config %s", collectorPath, configPath)
 
-	err := cmd.Start()
-	if err != nil {
+	if err := cmd.Start(); err != nil {
 		log.Logger().Errorf("Failed to start OpenTelemetry Collector: %v", err)
 		return
 	}
 
+	// Store the command reference so we can stop it later
+	tm.mu.Lock()
+	tm.otelCollectorCmd = cmd
+	tm.mu.Unlock()
+
 	log.Logger().Infof("OpenTelemetry Collector started with PID: %d", cmd.Process.Pid)
 
+	// Channel to track when cmd.Wait() completes
+	waitDone := make(chan error, 1)
+
+	// Wait for the process to exit in a separate goroutine
 	go func() {
-		<-done
-		log.Logger().Infof("Signaling OpenTelemetry Collector to terminate...")
-		if err := cmd.Process.Signal(syscall.SIGTERM); err != nil {
-			log.Logger().Errorf("Failed to terminate OpenTelemetry Collector: %v", err)
-		}
+		waitDone <- cmd.Wait()
 	}()
 
-	go func() {
-		if err := cmd.Wait(); err != nil {
+	// Wait for either context cancellation or process exit
+	select {
+	case <-tm.ctx.Done():
+		log.Logger().Infof("Context cancelled, stopping OTEL collector (PID: %d)...", cmd.Process.Pid)
+
+		// Kill the process - Go handles OS-specific details
+		if err := cmd.Process.Kill(); err != nil {
+			log.Logger().Errorf("Failed to kill OpenTelemetry Collector process: %v", err)
+		}
+
+		// Wait for the process to actually exit (with timeout)
+		select {
+		case err := <-waitDone:
+			if err != nil {
+				log.Logger().Infof("OpenTelemetry Collector terminated with error: %v", err)
+			} else {
+				log.Logger().Infof("OpenTelemetry Collector terminated successfully")
+			}
+		case <-time.After(5 * time.Second):
+			log.Logger().Warnf("Timeout waiting for OpenTelemetry Collector to exit after 5 seconds")
+		}
+
+	case err := <-waitDone:
+		if err != nil {
 			log.Logger().Errorf("OpenTelemetry Collector exited with error: %v", err)
 		} else {
 			log.Logger().Infof("OpenTelemetry Collector exited successfully")
 		}
-	}()
+	}
+
+	log.Logger().Infof("OpenTelemetry Collector process handler completed")
 }
 
 // handlePanic handles panics in goroutines
@@ -311,3 +333,69 @@ func (tm *TelemetryManager) getOtelCollectorOutputFilePath(isMetrics bool) strin
 	}
 	return filepath.Join(tm.baseDirectory, "otel-out.log")
 }
+
+// monitorTelemetryStatus monitors the telemetry enabled status via API polling
+func (tm *TelemetryManager) monitorTelemetryStatus() {
+	defer tm.wg.Done()
+
+	log.Logger().Infof("Starting telemetry status monitoring...")
+
+	ticker := time.NewTicker(1 * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			// Poll the API to check telemetry status
+			allocationDetails, resp, err := tm.warpbuildAPI.V1RunnerInstanceAPI.
+				GetRunnerInstanceAllocationDetails(tm.ctx, tm.runnerID).
+				XPOLLINGSECRET(tm.pollingSecret).
+				Execute()
+
+			if err != nil {
+				log.Logger().Debugf("Failed to get runner instance allocation details: %v", err)
+				if resp != nil {
+					log.Logger().Debugf("Response: %+v", resp)
+				}
+				continue
+			}
+
+			if allocationDetails == nil {
+				log.Logger().Debugf("No runner instance allocation details found")
+				continue
+			}
+
+			// Check if telemetry is disabled
+			if allocationDetails.HasTelemetryEnabled() {
+				telemetryEnabled := allocationDetails.GetTelemetryEnabled()
+
+				if !telemetryEnabled {
+					log.Logger().Infof("Telemetry has been disabled via API. Stopping telemetry collection...")
+					tm.stopOtelCollector()
+
+					// Cancel the context to stop the entire telemetry manager
+					tm.cancel()
+					return
+				}
+			}
+
+		case <-tm.ctx.Done():
+			log.Logger().Infof("Context cancelled, stopping telemetry status monitoring...")
+			return
+		}
+	}
+}
+
+// stopOtelCollector stops the OTEL collector process by canceling the context
+func (tm *TelemetryManager) stopOtelCollector() {
+	tm.mu.RLock()
+	defer tm.mu.RUnlock()
+
+	if tm.otelCollectorCmd == nil || tm.otelCollectorCmd.Process == nil {
+		log.Logger().Infof("OTEL collector process not running")
+		return
+	}
+
+	log.Logger().Infof("Stopping OTEL collector process (PID: %d)...", tm.otelCollectorCmd.Process.Pid)
+	// The actual termination will be handled by the context cancellation in runOtelCollector
+}

pkg/telemetry/manager.go

@@ -14,13 +14,30 @@ receivers:
       - 'C:\warpbuilds\warpbuild-agentd-restarter.stderr.log'
       - 'C:\warpbuilds\warpbuild-agentd.stdout.log'
       - 'C:\warpbuilds\warpbuild-agentd.stderr.log'
+      - 'C:\warpbuild-agentd-debug.log'
       - 'C:\warpbuilds\warpbuild-telemetryd.stdout.log'
       - 'C:\warpbuilds\warpbuild-telemetryd.stderr.log'
+      - 'C:\warpbuilds\warpbuild-proxyd.stdout.log'
+      - 'C:\warpbuilds\warpbuild-proxyd.stderr.log'
     start_at: end
     max_log_size: 200KiB
     include_file_name: true
     include_file_path: true
     poll_interval: 20s
+
+  # GitHub Actions job logs and diagnostic logs
+  filelog/gha_logs:
+    include:
+      - 'C:\warpbuilds\runner\_diag\*.log'
+      - 'C:\warpbuilds\runner\_diag\**\*.log'
+    start_at: end
+    max_log_size: 500KiB
+    include_file_name: true
+    include_file_path: true
+    include_file_path_resolved: true
+    poll_interval: 5s
+    attributes:
+      log.type: github_actions
 {{- else}}
   filelog:
     {{- if eq .OS "darwin"}}
@@ -33,6 +50,24 @@ receivers:
     include_file_name: false
     include_file_path: false
     poll_interval: 1s
+
+  # GitHub Actions job logs and diagnostic logs
+  filelog/gha_logs:
+    include:
+      {{- if eq .OS "darwin"}}
+      - '/Users/runner/.warpbuild/github-runner/runner-app-new/_diag/*.log'
+      - '/Users/runner/.warpbuild/github-runner/runner-app-new/_diag/**/*.log'
+      {{- else if eq .OS "linux"}}
+      - '/runner/_diag/*.log'
+      - '/runner/_diag/**/*.log'
+      {{- end}}
+    start_at: 'end'
+    max_log_size: 500KiB
+    include_file_name: true
+    include_file_path: true
+    poll_interval: 10s
+    attributes:
+      log.type: github_actions
 {{- end}}
 
   hostmetrics:
@@ -65,6 +100,23 @@ receivers:
             enabled: false
           system.network.dropped:
             enabled: false
+      disk:
+        metrics:
+          # collect disk throughput and IOPS metrics
+          system.disk.io:
+            enabled: true
+          system.disk.operations:
+            enabled: true
+          system.disk.io_time:
+            enabled: true
+          system.disk.operation_time:
+            enabled: true
+          system.disk.pending_operations:
+            enabled: false
+          system.disk.merged:
+            enabled: false
+          system.disk.weighted_io_time:
+            enabled: false
       filesystem:
         # Drop pseudo/virtual/container overlay mounts; keep real disks
         exclude_fs_types:
@@ -93,7 +145,7 @@ receivers:
 
 {{- if eq .OS "windows"}}
   windowsperfcounters/processor:
-    collection_interval: 30s
+    collection_interval: 1s
     metrics:
       system.cpu.time:
         description: percentage of cpu time
@@ -122,7 +174,7 @@ processors:
     timeout: 30s
     send_batch_size: 10_000_000
 
-  # Keep ONLY these 4 metrics in the pipeline
+  # Keep ONLY these metrics in the pipeline
   filter/only_needed:
     metrics:
       include:
@@ -132,24 +184,36 @@ processors:
           - system.memory.utilization
           - system.filesystem.utilization
           - system.network.io
+          - system.disk.io
+          - system.disk.operations
+          - system.disk.io_time
+          - system.disk.operation_time
 
   filter/drop_non_internet_nics:
     error_mode: ignore
     metrics:
       datapoint:
         - 'metric.name == "system.network.io" and IsMatch(attributes["device"], "^(lo|lo0|docker.*|cni.*|veth.*|br-.*|virbr.*|wg.*|tun.*|tap.*)$")'
 
-  # Convert cumulative network counters -> delta
+  # Convert cumulative counters -> delta
   cumulativetodelta:
     include:
       match_type: strict
       metrics:
         - system.network.io
+        - system.disk.io
+        - system.disk.operations
+        - system.disk.io_time
+        - system.disk.operation_time
 
   # Convert delta -> per-second rate (required list syntax)
   deltatorate:
     metrics:
       - system.network.io
+      - system.disk.io
+      - system.disk.operations
+      - system.disk.io_time
+      - system.disk.operation_time
 
   transform/normalize:
     metric_statements:
@@ -243,16 +307,25 @@ exporters:
     metrics_endpoint: "http://localhost:{{.Port}}/v1/metrics"
     encoding: "json"
     compression: "none"
+  
+  otlphttp/gha_logs:
+    logs_endpoint: "http://localhost:{{.Port}}/v1/gha-logs"
+    encoding: "json"
+    compression: "none"
 
 service:
   telemetry:
     metrics:
       level: none
   pipelines:
     logs:
-      receivers: [{{- if eq .OS "windows"}}windowseventlog/application, windowseventlog/security, filelog/services{{- else if eq .OS "darwin"}}filelog{{- else if eq .OS "linux"}}filelog{{- end}}]
+      receivers: [{{- if eq .OS "windows"}}windowseventlog/application, windowseventlog/security, filelog/services{{- else }}filelog{{- end}}]
       processors: [batch/logs]
       exporters: [otlphttp]   # tee: still send to your HTTP service, and also print locally
+    logs/gha_logs:
+      receivers: [filelog/gha_logs]
+      processors: [batch/logs]
+      exporters: [otlphttp/gha_logs]
     metrics:
       receivers: [hostmetrics{{- if eq .OS "windows"}}, windowsperfcounters/processor{{- end}}]
       processors: