receiver/vmmetrics: Allow to have a different process mount point. (#521)

* receiver/vmmetrics: Allow to have a different process mount point.

* Update unmarshal.

Author: songy23

receiver/vmmetricsreceiver/example_config.yaml

@@ -1,6 +1,8 @@
 receivers:
   vmmetrics:
-    scrape_interval: 10s
+    scrape_interval: 10
+    mount_point: "/proc"
+    # process_mount_point: "/data/proc" # Should only be used for Daemon Set within a container.
 
 zpages:
   port: 55679

receiver/vmmetricsreceiver/metrics_receiver.go

@@ -35,9 +35,10 @@ var (
 
 // Configuration defines the behavior and targets of the VM metrics scrapers.
 type Configuration struct {
-	scrapeInterval time.Duration `mapstructure:"scrape_interval"`
-	mountPoint     string        `mapstructure:"mount_point"`
-	metricPrefix   string        `mapstructure:"metric_prefix"`
+	scrapeInterval    time.Duration `mapstructure:"scrape_interval"`
+	mountPoint        string        `mapstructure:"mount_point"`
+	processMountPoint string        `mapstructure:"process_mount_point"`
+	metricPrefix      string        `mapstructure:"metric_prefix"`
 }
 
 // Receiver is the type used to handle metrics from VM metrics.
@@ -59,12 +60,12 @@ func New(v *viper.Viper, consumer consumer.MetricsConsumer) (*Receiver, error) {
 	var cfg Configuration
 
 	// Unmarshal our config values (using viper's mapstructure)
-	err := v.Unmarshal(&cfg)
+	err := unmarshal(&cfg, v.AllSettings())
 	if err != nil {
 		return nil, fmt.Errorf("vmmetrics receiver failed to parse config: %s", err)
 	}
 
-	vmc, err := NewVMMetricsCollector(cfg.scrapeInterval, cfg.mountPoint, cfg.metricPrefix, consumer)
+	vmc, err := NewVMMetricsCollector(cfg.scrapeInterval, cfg.mountPoint, cfg.processMountPoint, cfg.metricPrefix, consumer)
 	if err != nil {
 		return nil, err
 	}
@@ -107,3 +108,21 @@ func (vmr *Receiver) StopMetricsReception(ctx context.Context) error {
 	})
 	return err
 }
+
+// TODO(songya): investigate why viper.Unmarshal didn't work, remove this method and use viper.Unmarshal instead.
+func unmarshal(cfg *Configuration, settings map[string]interface{}) error {
+	if interval, ok := settings["scrape_interval"]; ok {
+		intervalInSecs := interval.(int)
+		cfg.scrapeInterval = time.Duration(intervalInSecs * int(time.Second))
+	}
+	if mountPoint, ok := settings["mount_point"]; ok {
+		cfg.mountPoint = mountPoint.(string)
+	}
+	if processMountPoint, ok := settings["process_mount_point"]; ok {
+		cfg.processMountPoint = processMountPoint.(string)
+	}
+	if prefix, ok := settings["metric_prefix"]; ok {
+		cfg.metricPrefix = prefix.(string)
+	}
+	return nil
+}

receiver/vmmetricsreceiver/vm_metrics_collector.go

@@ -37,9 +37,12 @@ import (
 type VMMetricsCollector struct {
 	consumer consumer.MetricsConsumer
 
-	startTime      time.Time
-	views          []*view.View
-	fs             procfs.FS
+	startTime time.Time
+	views     []*view.View
+
+	fs        procfs.FS
+	processFs procfs.FS
+
 	scrapeInterval time.Duration
 	metricPrefix   string
 	done           chan struct{}
@@ -51,14 +54,17 @@ const (
 )
 
 // NewVMMetricsCollector creates a new set of VM and Process Metrics (mem, cpu).
-func NewVMMetricsCollector(si time.Duration, mpoint, mprefix string, consumer consumer.MetricsConsumer) (*VMMetricsCollector, error) {
-	if mpoint == "" {
-		mpoint = defaultMountPoint
+func NewVMMetricsCollector(si time.Duration, mountPoint, processMountPoint, prefix string, consumer consumer.MetricsConsumer) (*VMMetricsCollector, error) {
+	if mountPoint == "" {
+		mountPoint = defaultMountPoint
+	}
+	if processMountPoint == "" {
+		processMountPoint = defaultMountPoint
 	}
 	if si <= 0 {
 		si = defaultScrapeInterval
 	}
-	fs, err := procfs.NewFS(mpoint)
+	fs, err := procfs.NewFS(mountPoint)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create new VMMetricsCollector: %s", err)
 	}
@@ -68,10 +74,18 @@ func NewVMMetricsCollector(si time.Duration, mpoint, mprefix string, consumer co
 		views:          vmViews,
 		fs:             fs,
 		scrapeInterval: si,
-		metricPrefix:   mprefix,
+		metricPrefix:   prefix,
 		done:           make(chan struct{}),
 	}
 	view.Register(vmc.views...)
+
+	if processMountPoint != mountPoint {
+		vmc.processFs, err = procfs.NewFS(processMountPoint)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create new VMMetricsCollector: %s", err)
+		}
+	}
+
 	return vmc, nil
 }
 
@@ -108,7 +122,13 @@ func (vmc *VMMetricsCollector) scrape() {
 		mRuntimeSysMem.M(int64(ms.Sys)))
 
 	pid := os.Getpid()
-	proc, err := procfs.NewProc(pid)
+	var proc procfs.Proc
+	var err error
+	if vmc.processFs == "" {
+		proc, err = vmc.fs.NewProc(pid)
+	} else {
+		proc, err = vmc.processFs.NewProc(pid)
+	}
 	if err == nil {
 		if procStat, err := proc.NewStat(); err == nil {
 			stats.Record(ctx, mCPUSeconds.M(int64(procStat.CPUTime())))