Numa-aware memstat collector (#45)

collectors/memstatMetric.go

@@ -1,35 +1,76 @@
 package collectors
 
 import (
+	"bufio"
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io/ioutil"
+	"os"
-	"log"
+	"path/filepath"
+	"regexp"
 	"strconv"
 	"strings"
 	"time"
 
+	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
 )
 
-const MEMSTATFILE = `/proc/meminfo`
+const MEMSTATFILE = "/proc/meminfo"
+const NUMA_MEMSTAT_BASE = "/sys/devices/system/node"
 
 type MemstatCollectorConfig struct {
 	ExcludeMetrics []string `json:"exclude_metrics"`
+	NodeStats      bool     `json:"node_stats,omitempty"`
+	NumaStats      bool     `json:"numa_stats,omitempty"`
+}
+
+type MemstatCollectorNode struct {
+	file string
+	tags map[string]string
 }
 
 type MemstatCollector struct {
 	metricCollector
-	stats   map[string]int64
+	stats     map[string]int64
-	tags    map[string]string
+	tags      map[string]string
-	matches map[string]string
+	matches   map[string]string
-	config  MemstatCollectorConfig
+	config    MemstatCollectorConfig
+	nodefiles map[int]MemstatCollectorNode
+}
+
+func getStats(filename string) map[string]float64 {
+	stats := make(map[string]float64)
+	file, err := os.Open(filename)
+	if err != nil {
+		cclog.Error(err.Error())
+	}
+	defer file.Close()
+
+	scanner := bufio.NewScanner(file)
+	for scanner.Scan() {
+		line := scanner.Text()
+		linefields := strings.Fields(line)
+		if len(linefields) == 3 {
+			v, err := strconv.ParseFloat(linefields[1], 64)
+			if err == nil {
+				stats[strings.Trim(linefields[0], ":")] = v
+			}
+		} else if len(linefields) == 5 {
+			v, err := strconv.ParseFloat(linefields[3], 64)
+			if err == nil {
+				stats[strings.Trim(linefields[0], ":")] = v
+			}
+		}
+	}
+	return stats
 }
 
 func (m *MemstatCollector) Init(config json.RawMessage) error {
 	var err error
 	m.name = "MemstatCollector"
+	m.config.NodeStats = true
+	m.config.NumaStats = false
 	if len(config) > 0 {
 		err = json.Unmarshal(config, &m.config)
 		if err != nil {
@@ -40,7 +81,8 @@ func (m *MemstatCollector) Init(config json.RawMessage) error {
 	m.stats = make(map[string]int64)
 	m.matches = make(map[string]string)
 	m.tags = map[string]string{"type": "node"}
-	matches := map[string]string{`MemTotal`: "mem_total",
+	matches := map[string]string{
+		"MemTotal":     "mem_total",
 		"SwapTotal":    "swap_total",
 		"SReclaimable": "mem_sreclaimable",
 		"Slab":         "mem_slab",
@@ -48,7 +90,9 @@ func (m *MemstatCollector) Init(config json.RawMessage) error {
 		"Buffers":      "mem_buffers",
 		"Cached":       "mem_cached",
 		"MemAvailable": "mem_available",
-		"SwapFree":     "swap_free"}
+		"SwapFree":     "swap_free",
+		"MemShared":    "mem_shared",
+	}
 	for k, v := range matches {
 		_, skip := stringArrayContains(m.config.ExcludeMetrics, k)
 		if !skip {
@@ -56,13 +100,44 @@ func (m *MemstatCollector) Init(config json.RawMessage) error {
 		}
 	}
 	if len(m.matches) == 0 {
-		return errors.New("No metrics to collect")
+		return errors.New("no metrics to collect")
 	}
 	m.setup()
-	_, err = ioutil.ReadFile(string(MEMSTATFILE))
+
-	if err == nil {
+	if m.config.NodeStats {
-		m.init = true
+		if stats := getStats(MEMSTATFILE); len(stats) == 0 {
+			return fmt.Errorf("cannot read data from file %s", MEMSTATFILE)
+		}
 	}
+
+	if m.config.NumaStats {
+		globPattern := filepath.Join(NUMA_MEMSTAT_BASE, "node[0-9]*", "meminfo")
+		regex := regexp.MustCompile(filepath.Join(NUMA_MEMSTAT_BASE, "node(\\d+)", "meminfo"))
+		files, err := filepath.Glob(globPattern)
+		if err == nil {
+			m.nodefiles = make(map[int]MemstatCollectorNode)
+			for _, f := range files {
+				if stats := getStats(f); len(stats) == 0 {
+					return fmt.Errorf("cannot read data from file %s", f)
+				}
+				rematch := regex.FindStringSubmatch(f)
+				if len(rematch) == 2 {
+					id, err := strconv.Atoi(rematch[1])
+					if err == nil {
+						f := MemstatCollectorNode{
+							file: f,
+							tags: map[string]string{
+								"type":    "memoryDomain",
+								"type-id": fmt.Sprintf("%d", id),
+							},
+						}
+						m.nodefiles[id] = f
+					}
+				}
+			}
+		}
+	}
+	m.init = true
 	return err
 }
 
@@ -71,56 +146,41 @@ func (m *MemstatCollector) Read(interval time.Duration, output chan lp.CCMetric)
 		return
 	}
 
-	buffer, err := ioutil.ReadFile(string(MEMSTATFILE))
+	sendStats := func(stats map[string]float64, tags map[string]string) {
-	if err != nil {
+		for match, name := range m.matches {
-		log.Print(err)
+			var value float64 = 0
-		return
+			if v, ok := stats[match]; ok {
-	}
+				value = v
-
+			}
-	ll := strings.Split(string(buffer), "\n")
+			y, err := lp.New(name, tags, m.meta, map[string]interface{}{"value": value}, time.Now())
-	for _, line := range ll {
+			if err == nil {
-		ls := strings.Split(line, `:`)
+				output <- y
-		if len(ls) > 1 {
+			}
-			lv := strings.Fields(ls[1])
-			m.stats[ls[0]], err = strconv.ParseInt(lv[0], 0, 64)
 		}
-	}
+		if _, skip := stringArrayContains(m.config.ExcludeMetrics, "mem_used"); !skip {
-
+			if freeVal, free := stats["MemFree"]; free {
-	if _, exists := m.stats[`MemTotal`]; !exists {
+				if bufVal, buffers := stats["Buffers"]; buffers {
-		err = errors.New("Parse error")
+					if cacheVal, cached := stats["Cached"]; cached {
-		log.Print(err)
+						memUsed := stats["MemTotal"] - (freeVal + bufVal + cacheVal)
-		return
+						y, err := lp.New("mem_used", tags, m.meta, map[string]interface{}{"value": memUsed}, time.Now())
-	}
+						if err == nil {
-
+							output <- y
-	for match, name := range m.matches {
+						}
-		if _, exists := m.stats[match]; !exists {
+					}
-			err = fmt.Errorf("Parse error for %s : %s", match, name)
-			log.Print(err)
-			continue
-		}
-		y, err := lp.New(name, m.tags, m.meta, map[string]interface{}{"value": int(float64(m.stats[match]) * 1.0e-3)}, time.Now())
-		if err == nil {
-			output <- y
-		}
-	}
-
-	if _, free := m.stats[`MemFree`]; free {
-		if _, buffers := m.stats[`Buffers`]; buffers {
-			if _, cached := m.stats[`Cached`]; cached {
-				memUsed := m.stats[`MemTotal`] - (m.stats[`MemFree`] + m.stats[`Buffers`] + m.stats[`Cached`])
-				_, skip := stringArrayContains(m.config.ExcludeMetrics, "mem_used")
-				y, err := lp.New("mem_used", m.tags, m.meta, map[string]interface{}{"value": int(float64(memUsed) * 1.0e-3)}, time.Now())
-				if err == nil && !skip {
-					output <- y
 				}
 			}
 		}
 	}
-	if _, found := m.stats[`MemShared`]; found {
+
-		_, skip := stringArrayContains(m.config.ExcludeMetrics, "mem_shared")
+	if m.config.NodeStats {
-		y, err := lp.New("mem_shared", m.tags, m.meta, map[string]interface{}{"value": int(float64(m.stats[`MemShared`]) * 1.0e-3)}, time.Now())
+		nodestats := getStats(MEMSTATFILE)
-		if err == nil && !skip {
+		sendStats(nodestats, m.tags)
-			output <- y
+	}
+
+	if m.config.NumaStats {
+		for _, nodeConf := range m.nodefiles {
+			stats := getStats(nodeConf.file)
+			sendStats(stats, nodeConf.tags)
 		}
 	}
 }