feat: enhance process exporter metrics parsing and update sender to handle new structure

Author: guiyumin

internal/exporters/process_exporter.go

@@ -16,6 +16,8 @@ type ProcessExporter struct {
 	client   *http.Client
 }
 
+var _ Exporter = (*ProcessExporter)(nil)
+
 // NewProcessExporter creates a new ProcessExporter instance
 func NewProcessExporter(endpoint string, timeout time.Duration) *ProcessExporter {
 	// Use defaults if not specified

internal/prometheus/process_exporter_parser.go

@@ -8,38 +8,37 @@ import (
 	"time"
 )
 
-// ProcessExporterMetricSnapshot represents parsed process metrics from process_exporter
-// This captures per-process group metrics (grouped by command name)
+// ProcessExporterMetricSnapshot represents a single process group metric snapshot
+// This is a flat structure matching NodeExporterMetricSnapshot pattern
+// Each process group (e.g., "nginx", "postgres") becomes one snapshot
 type ProcessExporterMetricSnapshot struct {
-	Timestamp time.Time       `json:"timestamp"`
-	Processes []ProcessMetric `json:"processes"`
+	Timestamp       time.Time `json:"timestamp"`
+	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
 }
 
-// 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
+// processData is a temporary struct used during parsing
+type processData struct {
+	numProcs        int
+	cpuSecondsTotal float64
+	memoryBytes     int64
 }
 
 // ParseProcessExporterMetrics parses Prometheus process_exporter text format
-// Extracts per-process group metrics (CPU, memory, count)
+// Returns a slice of ProcessExporterMetricSnapshot (one per process group)
 //
 // 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{},
-	}
-
+func ParseProcessExporterMetrics(data []byte) ([]ProcessExporterMetricSnapshot, error) {
+	timestamp := time.Now().UTC()
 	scanner := bufio.NewScanner(bytes.NewReader(data))
 
 	// Track metrics per process group (groupname)
-	processMetrics := make(map[string]*ProcessMetric)
+	processMetrics := make(map[string]*processData)
 
 	for scanner.Scan() {
 		line := scanner.Text()
@@ -60,18 +59,25 @@ func ParseProcessExporterMetrics(data []byte) (*ProcessExporterMetricSnapshot, e
 		return nil, fmt.Errorf("scanner error: %w", err)
 	}
 
-	// Convert map to slice
-	for _, pm := range processMetrics {
+	// Convert map to slice of flat snapshots
+	snapshots := []ProcessExporterMetricSnapshot{}
+	for name, data := range processMetrics {
 		// Only include processes that have at least 1 running instance
-		if pm.NumProcs > 0 {
-			snapshot.Processes = append(snapshot.Processes, *pm)
+		if data.numProcs > 0 {
+			snapshots = append(snapshots, ProcessExporterMetricSnapshot{
+				Timestamp:       timestamp,
+				Name:            name,
+				NumProcs:        data.numProcs,
+				CPUSecondsTotal: data.cpuSecondsTotal,
+				MemoryBytes:     data.memoryBytes,
+			})
 		}
 	}
 
-	return snapshot, nil
+	return snapshots, nil
 }
 
-func parseProcessLine(line string, processMetrics map[string]*ProcessMetric) error {
+func parseProcessLine(line string, processMetrics map[string]*processData) error {
 	// Split metric name and value
 	parts := strings.Fields(line)
 	if len(parts) < 2 {
@@ -107,26 +113,24 @@ func parseProcessLine(line string, processMetrics map[string]*ProcessMetric) err
 
 	// Ensure process metric entry exists
 	if processMetrics[groupname] == nil {
-		processMetrics[groupname] = &ProcessMetric{
-			Name: groupname,
-		}
+		processMetrics[groupname] = &processData{}
 	}
 
 	pm := processMetrics[groupname]
 
 	// Parse specific metrics
 	switch metricName {
 	case "namedprocess_namegroup_num_procs":
-		pm.NumProcs = int(value)
+		pm.numProcs = int(value)
 
 	case "namedprocess_namegroup_cpu_seconds_total":
-		pm.CPUSecondsTotal = value
+		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)
+			pm.memoryBytes = int64(value)
 		}
 	}
 

internal/report/sender.go

@@ -178,14 +178,15 @@ func (s *Sender) drainLoop() {
 
 // processBatch loads and sends buffered files grouped by exporter
 // Returns error if send fails (files are kept for retry)
-// Payload format: { "node_exporter": [...], "postgres_exporter": [...] }
+// Payload format: { "node_exporter": [...], "process_exporter": [...] }
 func (s *Sender) processBatch(filePaths []string) error {
 	if len(filePaths) == 0 {
 		return nil
 	}
 
-	// Group entries by exporter name
-	exporterMetrics := make(map[string][]prometheus.NodeExporterMetricSnapshot)
+	// Group entries by exporter name - use separate maps for type safety
+	nodeExporterMetrics := []prometheus.NodeExporterMetricSnapshot{}
+	processExporterMetrics := []prometheus.ProcessExporterMetricSnapshot{}
 	processedFiles := []string{}
 	var serverID string
 
@@ -216,38 +217,60 @@ func (s *Sender) processBatch(filePaths []string) error {
 			serverID = entry.ServerID
 		}
 
-		// Parse Prometheus text to structured metrics
-		// Note: Currently only node_exporter is parsed, other exporters need their own parsers
-		snapshot, err := prometheus.ParseNodeExporterMetrics(entry.Data)
-		if err != nil {
-			logger.Warn("Failed to parse node_exporter metrics, using zero values",
-				logger.String("exporter", entry.ExporterName),
-				logger.String("file", filePath),
-				logger.Err(err))
-			// Use zero-value snapshot
-			snapshot = &prometheus.NodeExporterMetricSnapshot{
-				Timestamp: time.Now().UTC(),
+		// Parse Prometheus text to structured metrics based on exporter type
+		switch entry.ExporterName {
+		case "node_exporter":
+			snapshot, err := prometheus.ParseNodeExporterMetrics(entry.Data)
+			if err != nil {
+				logger.Warn("Failed to parse node_exporter metrics, using zero values",
+					logger.String("exporter", entry.ExporterName),
+					logger.String("file", filePath),
+					logger.Err(err))
+				// Use zero-value snapshot
+				snapshot = &prometheus.NodeExporterMetricSnapshot{
+					Timestamp: time.Now().UTC(),
+				}
 			}
-		}
+			nodeExporterMetrics = append(nodeExporterMetrics, *snapshot)
+
+		case "process_exporter":
+			snapshots, err := prometheus.ParseProcessExporterMetrics(entry.Data)
+			if err != nil {
+				logger.Warn("Failed to parse process_exporter metrics, skipping",
+					logger.String("exporter", entry.ExporterName),
+					logger.String("file", filePath),
+					logger.Err(err))
+				continue
+			}
+			// Append all process snapshots (one per process group)
+			processExporterMetrics = append(processExporterMetrics, snapshots...)
 
-		// Add to exporter's array
-		exporterMetrics[entry.ExporterName] = append(
-			exporterMetrics[entry.ExporterName],
-			*snapshot,
-		)
+		default:
+			logger.Warn("Unknown exporter type, skipping",
+				logger.String("exporter", entry.ExporterName),
+				logger.String("file", filePath))
+			continue
+		}
 
 		processedFiles = append(processedFiles, filePath)
 	}
 
 	// Nothing to send
-	if len(exporterMetrics) == 0 {
+	if len(nodeExporterMetrics) == 0 && len(processExporterMetrics) == 0 {
 		return nil
 	}
 
-	// Build payload: { "node_exporter": [...], "mysql_exporter": [...] }
+	// Build payload: { "node_exporter": [...], "process_exporter": [...] }
+	// Only include exporters that have data
 	payload := make(map[string]interface{})
-	for exporterName, snapshots := range exporterMetrics {
-		payload[exporterName] = snapshots
+	exporterCount := 0
+	if len(nodeExporterMetrics) > 0 {
+		payload["node_exporter"] = nodeExporterMetrics
+		exporterCount++
+	}
+	if len(processExporterMetrics) > 0 {
+		payload["process_exporter"] = processExporterMetrics
+		exporterCount++
 	}
 
 	// Convert to JSON
@@ -280,7 +303,7 @@ func (s *Sender) processBatch(filePaths []string) error {
 	if successCount > 0 {
 		logger.Info("Successfully sent buffered data",
 			logger.Int("files", successCount),
-			logger.Int("exporters", len(exporterMetrics)))
+			logger.Int("exporters", exporterCount))
 
 		// Periodically clean up old buffer files
 		if err := s.buffer.Cleanup(); err != nil {