receiver: Add a prototype VM metrics receiver. (#515)

Author: songy23

cmd/ocagent/main.go

@@ -43,6 +43,7 @@ import (
 	"github.com/census-instrumentation/opencensus-service/receiver/jaegerreceiver"
 	"github.com/census-instrumentation/opencensus-service/receiver/opencensusreceiver"
 	"github.com/census-instrumentation/opencensus-service/receiver/prometheusreceiver"
+	"github.com/census-instrumentation/opencensus-service/receiver/vmmetricsreceiver"
 	"github.com/census-instrumentation/opencensus-service/receiver/zipkinreceiver"
 	"github.com/census-instrumentation/opencensus-service/receiver/zipkinreceiver/zipkinscribereceiver"
 )
@@ -172,6 +173,15 @@ func runOCAgent() {
 		closeFns = append(closeFns, promDoneFn)
 	}
 
+	// If the VMMetrics receiver is enabled, then run it.
+	if agentConfig.VMMetricsReceiverEnabled() {
+		vmmDoneFn, err := runVMMetricsReceiver(viperCfg, commonMetricsSink, asyncErrorChan)
+		if err != nil {
+			log.Fatal(err)
+		}
+		closeFns = append(closeFns, vmmDoneFn)
+	}
+
 	// Always cleanup finally
 	defer func() {
 		for _, closeFn := range closeFns {
@@ -342,3 +352,18 @@ func runPrometheusReceiver(v *viper.Viper, next consumer.MetricsConsumer, asyncE
 	log.Print("Running Prometheus receiver")
 	return doneFn, nil
 }
+
+func runVMMetricsReceiver(v *viper.Viper, next consumer.MetricsConsumer, asyncErrorChan chan<- error) (doneFn func() error, err error) {
+	vmr, err := vmmetricsreceiver.New(v.Sub("receivers.vmmetrics"), next)
+	if err != nil {
+		return nil, err
+	}
+	if err := vmr.StartMetricsReception(context.Background(), asyncErrorChan); err != nil {
+		return nil, err
+	}
+	doneFn = func() error {
+		return vmr.StopMetricsReception(context.Background())
+	}
+	log.Print("Running VMMetrics receiver")
+	return doneFn, nil
+}

internal/config/config.go

@@ -101,6 +101,7 @@ type Receivers struct {
 	Zipkin     *ReceiverConfig       `mapstructure:"zipkin"`
 	Jaeger     *ReceiverConfig       `mapstructure:"jaeger"`
 	Scribe     *ScribeReceiverConfig `mapstructure:"zipkin-scribe"`
+	VMMetrics  *ReceiverConfig       `mapstructure:"vmmetrics"`
 
 	// Prometheus contains the Prometheus configurations.
 	// Such as:
@@ -374,6 +375,14 @@ func (c *Config) OpenCensusReceiverTLSCredentialsServerOption() (opt opencensusr
 	return tlsCreds.ToOpenCensusReceiverServerOption()
 }
 
+// VMMetricsReceiverEnabled returns true if Config is non-nil.
+func (c *Config) VMMetricsReceiverEnabled() bool {
+	if c == nil {
+		return false
+	}
+	return c.Receivers != nil && c.Receivers.VMMetrics != nil
+}
+
 // CheckLogicalConflicts serves to catch logical errors such as
 // if the Zipkin receiver port conflicts with that of the exporter,
 // lest we'll have a self DOS because spans will be exported "out" from

receiver/vmmetricsreceiver/.nocover

@@ -0,0 +1 @@
+FIXME: This will get re-written and moved.

receiver/vmmetricsreceiver/doc.go

@@ -0,0 +1,17 @@
+// Copyright 2019, OpenCensus Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package vmmetricsreceiver has the logic for scraping VM metrics
+// and then passing them onto a metric consumer instance.
+package vmmetricsreceiver

receiver/vmmetricsreceiver/example_config.yaml

@@ -0,0 +1,11 @@
+receivers:
+  vmmetrics:
+    scrape_interval: 10s
+
+zpages:
+  port: 55679
+
+exporters:
+  prometheus:
+    namespace: "vmmetrics_test"
+    address: "localhost:8888"

receiver/vmmetricsreceiver/metrics_receiver.go

@@ -0,0 +1,109 @@
+// Copyright 2019, OpenCensus Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package vmmetricsreceiver
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"runtime"
+	"sync"
+	"time"
+
+	"github.com/spf13/viper"
+
+	"github.com/census-instrumentation/opencensus-service/consumer"
+)
+
+var (
+	errAlreadyStarted     = errors.New("already started")
+	errAlreadyStopped     = errors.New("already stopped")
+	errNilMetricsConsumer = errors.New("expecting a non-nil MetricsConsumer")
+)
+
+// 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"`
+}
+
+// Receiver is the type used to handle metrics from VM metrics.
+type Receiver struct {
+	mu sync.Mutex
+
+	vmc *VMMetricsCollector
+
+	stopOnce  sync.Once
+	startOnce sync.Once
+}
+
+// New creates a new vmmetricsreceiver.Receiver reference.
+func New(v *viper.Viper, consumer consumer.MetricsConsumer) (*Receiver, error) {
+	if consumer == nil {
+		return nil, errNilMetricsConsumer
+	}
+
+	var cfg Configuration
+
+	// Unmarshal our config values (using viper's mapstructure)
+	err := v.Unmarshal(&cfg)
+	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)
+	if err != nil {
+		return nil, err
+	}
+
+	vmr := &Receiver{
+		vmc: vmc,
+	}
+	return vmr, nil
+}
+
+// StartMetricsReception scrapes VM metrics based on the OS platform.
+func (vmr *Receiver) StartMetricsReception(ctx context.Context, asyncErrorChan chan<- error) error {
+	vmr.mu.Lock()
+	defer vmr.mu.Unlock()
+
+	var err = errAlreadyStarted
+	vmr.startOnce.Do(func() {
+		switch runtime.GOOS {
+		case "linux":
+			vmr.vmc.StartCollection()
+		case "darwin", "freebsd", "windows":
+			// TODO: add support for other platforms.
+			return
+		}
+
+		err = nil
+	})
+	return err
+}
+
+// StopMetricsReception stops and cancels the underlying VM metrics scrapers.
+func (vmr *Receiver) StopMetricsReception(ctx context.Context) error {
+	vmr.mu.Lock()
+	defer vmr.mu.Unlock()
+
+	var err = errAlreadyStopped
+	vmr.stopOnce.Do(func() {
+		vmr.vmc.StopCollection()
+		err = nil
+	})
+	return err
+}