feat: add process exporter support and configuration to metrics agent

Author: guiyumin

cmd/start.go

@@ -103,6 +103,8 @@ func runAgent(cmd *cobra.Command, args []string) error {
 		switch exporterCfg.Name {
 		case "node_exporter":
 			exp = exporters.NewNodeExporter(exporterCfg.Endpoint, exporterCfg.Timeout)
+		case "process_exporter":
+			exp = exporters.NewProcessExporter(exporterCfg.Endpoint, exporterCfg.Timeout)
 		default:
 			logger.Warn("Unknown exporter type, skipping", logger.String("name", exporterCfg.Name))
 			continue

internal/exporters/exporter.go

@@ -17,12 +17,4 @@ type Exporter interface {
 
 	// Verify checks if the exporter is accessible (used at startup)
 	Verify() error
-
-	// DefaultEndpoint returns the standard endpoint for this exporter
-	// Example: "http://localhost:9100/metrics" for node_exporter
-	DefaultEndpoint() string
-
-	// DefaultInterval returns the recommended scrape interval
-	// Example: 15s for node_exporter, 30s for postgres_exporter
-	DefaultInterval() time.Duration
 }

internal/exporters/node_exporter.go

@@ -82,11 +82,3 @@ func (n *NodeExporter) Verify() error {
 	logger.Info("node_exporter verified", logger.String("endpoint", n.endpoint))
 	return nil
 }
-
-func (n *NodeExporter) DefaultEndpoint() string {
-	return "http://localhost:9100/metrics"
-}
-
-func (n *NodeExporter) DefaultInterval() time.Duration {
-	return 15 * time.Second
-}

internal/exporters/process_exporter.go

@@ -0,0 +1,81 @@
+package exporters
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+)
+
+// ProcessExporter represents a Prometheus process_exporter instance
+type ProcessExporter struct {
+	name     string
+	endpoint string
+	timeout  time.Duration
+	client   *http.Client
+}
+
+// NewProcessExporter creates a new ProcessExporter instance
+func NewProcessExporter(endpoint string, timeout time.Duration) *ProcessExporter {
+	// Use defaults if not specified
+	if endpoint == "" {
+		endpoint = "http://127.0.0.1:9256/metrics"
+	}
+	if timeout == 0 {
+		timeout = 3 * time.Second
+	}
+
+	return &ProcessExporter{
+		name:     "process_exporter",
+		endpoint: endpoint,
+		timeout:  timeout,
+		client: &http.Client{
+			Timeout: timeout,
+		},
+	}
+}
+
+// Name returns the exporter name
+func (e *ProcessExporter) Name() string {
+	return e.name
+}
+
+// Endpoint returns the metrics endpoint URL
+func (e *ProcessExporter) Endpoint() string {
+	return e.endpoint
+}
+
+// Scrape fetches metrics from process_exporter
+func (e *ProcessExporter) Scrape(ctx context.Context) ([]byte, error) {
+	req, err := http.NewRequestWithContext(ctx, "GET", e.endpoint, nil)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create request: %w", err)
+	}
+
+	resp, err := e.client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("HTTP request failed: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+	}
+
+	data, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read response: %w", err)
+	}
+
+	return data, nil
+}
+
+// Verify checks if the exporter is accessible
+func (e *ProcessExporter) Verify() error {
+	ctx, cancel := context.WithTimeout(context.Background(), e.timeout)
+	defer cancel()
+
+	_, err := e.Scrape(ctx)
+	return err
+}

internal/prometheus/process_exporter_parser.go

@@ -0,0 +1,137 @@
+package prometheus
+
+import (
+	"bufio"
+	"bytes"
+	"fmt"
+	"strings"
+	"time"
+)
+
+// ProcessExporterMetricSnapshot represents parsed process metrics from process_exporter
+// This captures per-process group metrics (grouped by command name)
+type ProcessExporterMetricSnapshot struct {
+	Timestamp time.Time       `json:"timestamp"`
+	Processes []ProcessMetric `json:"processes"`
+}
+
+// ProcessMetric represents metrics for a single process group (e.g., all "nginx" processes)
+type ProcessMetric struct {
+	Name            string  `json:"name"`              // Process name (groupname from process_exporter)
+	NumProcs        int     `json:"num_procs"`         // Number of processes in this group
+	CPUSecondsTotal float64 `json:"cpu_seconds_total"` // Total CPU time consumed (counter)
+	MemoryBytes     int64   `json:"memory_bytes"`      // Resident memory (RSS) in bytes
+}
+
+// ParseProcessExporterMetrics parses Prometheus process_exporter text format
+// Extracts per-process group metrics (CPU, memory, count)
+//
+// Expected metrics from process_exporter:
+// - namedprocess_namegroup_num_procs{groupname="nginx"} 4
+// - namedprocess_namegroup_cpu_seconds_total{groupname="nginx"} 1234.56
+// - namedprocess_namegroup_memory_bytes{groupname="nginx",memtype="resident"} 104857600
+func ParseProcessExporterMetrics(data []byte) (*ProcessExporterMetricSnapshot, error) {
+	snapshot := &ProcessExporterMetricSnapshot{
+		Timestamp: time.Now().UTC(),
+		Processes: []ProcessMetric{},
+	}
+
+	scanner := bufio.NewScanner(bytes.NewReader(data))
+
+	// Track metrics per process group (groupname)
+	processMetrics := make(map[string]*ProcessMetric)
+
+	for scanner.Scan() {
+		line := scanner.Text()
+
+		// Skip comments and empty lines
+		if len(line) == 0 || line[0] == '#' {
+			continue
+		}
+
+		// Parse metric line
+		if err := parseProcessLine(line, processMetrics); err != nil {
+			// Log but don't fail on individual parse errors
+			continue
+		}
+	}
+
+	if err := scanner.Err(); err != nil {
+		return nil, fmt.Errorf("scanner error: %w", err)
+	}
+
+	// Convert map to slice
+	for _, pm := range processMetrics {
+		// Only include processes that have at least 1 running instance
+		if pm.NumProcs > 0 {
+			snapshot.Processes = append(snapshot.Processes, *pm)
+		}
+	}
+
+	return snapshot, nil
+}
+
+func parseProcessLine(line string, processMetrics map[string]*ProcessMetric) error {
+	// Split metric name and value
+	parts := strings.Fields(line)
+	if len(parts) < 2 {
+		return fmt.Errorf("invalid line format")
+	}
+
+	metricPart := parts[0]
+	valuePart := parts[1]
+
+	// Extract metric name and labels
+	var metricName string
+	var labels map[string]string
+
+	if idx := strings.Index(metricPart, "{"); idx != -1 {
+		metricName = metricPart[:idx]
+		labelsStr := metricPart[idx+1 : len(metricPart)-1] // Remove {}
+		labels = parseLabels(labelsStr)
+	} else {
+		metricName = metricPart
+		labels = make(map[string]string)
+	}
+
+	value, err := parseValue(valuePart)
+	if err != nil {
+		return err
+	}
+
+	// Extract groupname (process name)
+	groupname, ok := labels["groupname"]
+	if !ok {
+		return fmt.Errorf("missing groupname label")
+	}
+
+	// Ensure process metric entry exists
+	if processMetrics[groupname] == nil {
+		processMetrics[groupname] = &ProcessMetric{
+			Name: groupname,
+		}
+	}
+
+	pm := processMetrics[groupname]
+
+	// Parse specific metrics
+	switch metricName {
+	case "namedprocess_namegroup_num_procs":
+		pm.NumProcs = int(value)
+
+	case "namedprocess_namegroup_cpu_seconds_total":
+		pm.CPUSecondsTotal = value
+
+	case "namedprocess_namegroup_memory_bytes":
+		// Only use resident memory (RSS)
+		memtype, ok := labels["memtype"]
+		if ok && memtype == "resident" {
+			pm.MemoryBytes = int64(value)
+		}
+	}
+
+	return nil
+}
+
+// Note: parseLabels() and parseValue() are already defined in node_exporter_parser.go
+// They are package-level functions shared across all parsers in the prometheus package

nodepulse.yml

@@ -31,6 +31,15 @@ exporters:
     interval: 15s  # Optional: Fast scraping for system metrics (falls back to agent.interval if not specified)
     timeout: 3s
 
+  # Process Exporter - Per-process metrics (CPU, memory by process name)
+  # NOTE: Requires process_exporter to be installed and running
+  # See: ansible/roles/process-exporter/ for deployment
+  - name: process_exporter
+    enabled: false  # Disabled by default - enable after deploying process_exporter
+    endpoint: "http://localhost:9256/metrics"
+    interval: 15s  # Same as node_exporter for consistency
+    timeout: 3s
+
   # Example: PostgreSQL Exporter (uncomment to enable)
   # - name: postgres_exporter
   #   enabled: true