Initial implementation of metrics (Prometheus support)

Created a metrics package that leans on the official Prometheus client to serve
metrics. Adds a new Pollable for Sensors that record metrics received from user-
defined functions.

containerbuddy/config.go

@@ -8,6 +8,7 @@ import (
 	"flag"
 	"fmt"
 	"io/ioutil"
+	"metrics"
 	"os"
 	"os/exec"
 	"strings"
@@ -50,6 +51,7 @@ type Config struct {
 	StopTimeout  int              `json:"stopTimeout"`
 	Services     []*ServiceConfig `json:"services"`
 	Backends     []*BackendConfig `json:"backends"`
+	Metrics      *metrics.Metrics `json:"metrics,omitempty"`
 	preStartCmd  *exec.Cmd
 	preStopCmd   *exec.Cmd
 	postStopCmd  *exec.Cmd
@@ -209,13 +211,31 @@ func initializeConfig(config *Config) (*Config, error) {
 			service.healthCheckCmd = cmd
 		}
 
-		interfaces, ifaceErr := utils.ParseInterfaces(service.Interfaces)
-		if ifaceErr != nil {
-			return nil, ifaceErr
+		if ipAddress, err := utils.IpFromInterfaces(service.Interfaces); err != nil {
+			return nil, err
+		} else {
+			service.ipAddress = ipAddress
 		}
+	}
 
-		if service.ipAddress, err = utils.GetIP(interfaces); err != nil {
+	if config.Metrics != nil {
+		m := config.Metrics
+		if err := m.Parse(); err != nil {
 			return nil, err
+		} else {
+			// create a new service for Metrics
+			metricsService := &ServiceConfig{
+				ID:               fmt.Sprintf("%s-%s", m.ServiceName, hostname),
+				Name:             m.ServiceName,
+				Poll:             m.Poll,
+				TTL:              m.TTL,
+				Interfaces:       m.Interfaces,
+				Tags:             m.Tags,
+				discoveryService: discovery,
+				ipAddress:        m.IpAddress,
+				healthCheckCmd:   nil, // no health check code
+			}
+			config.Services = append(config.Services, metricsService)
 		}
 	}
 

containerbuddy/main.go

@@ -66,6 +66,12 @@ func handlePolling(config *Config) {
 	for _, service := range config.Services {
 		quit = append(quit, poll(service))
 	}
+	if config.Metrics != nil {
+		for _, sensor := range config.Metrics.Sensors {
+			quit = append(quit, poll(sensor))
+		}
+		config.Metrics.Serve()
+	}
 	config.QuitChannels = quit
 }
 

containerbuddy/services.go

@@ -53,6 +53,11 @@ func (s *ServiceConfig) Deregister() {
 
 // CheckHealth runs the service's health command, returning the results
 func (s *ServiceConfig) CheckHealth() (int, error) {
+	// if we have a valid ServiceConfig but there's no health check
+	// set, assume it always passes (ex. metrics service).
+	if s.healthCheckCmd == nil {
+		return 0, nil
+	}
 	exitCode, err := run(s.healthCheckCmd)
 	// Reset command object - since it can't be reused
 	s.healthCheckCmd = utils.ArgsToCmd(s.healthCheckCmd.Args)

main.go

@@ -1,6 +1,9 @@
 package main
 
-import "containerbuddy"
+import (
+	"containerbuddy"
+	_ "metrics"
+)
 
 func main() {
 	containerbuddy.Main()

makefile

@@ -58,7 +58,7 @@ docker: build/containerbuddy_build consul etcd
 
 # top-level target for vendoring our packages: godep restore requires
 # being in the package directory so we have to run this for each package
-vendor: containerbuddy/vendor utils/vendor
+vendor: containerbuddy/vendor utils/vendor metrics/vendor
 %/vendor: build/containerbuddy_build
 	docker run --rm \
 	    -v ${ROOT}:/go/cb \
@@ -86,6 +86,7 @@ add-dep: build/containerbuddy_build
 	 	containerbuddy_build godep save
 	mv $(PKG)/src $(PKG)/vendor
 
+
 # ----------------------------------------------
 # develop and test
 
@@ -95,7 +96,7 @@ lint: vendor
 # run unit tests
 test: docker vendor
 	@mkdir -p cover
-	${DOCKERRUN} go test -v containerbuddy utils
+	${DOCKERRUN} go test -v containerbuddy utils metrics
 
 cover:
 	@mkdir -p cover

metrics/metrics.go

@@ -0,0 +1,43 @@
+package metrics
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/prometheus/client_golang/prometheus"
+	"net/http"
+	"utils"
+)
+
+// Metrics represents the service to advertise for finding the metrics
+// endpoint, and the collection of Sensors.
+type Metrics struct {
+	ServiceName string          `json:"name"`
+	Url         string          `json:"url"`
+	Port        int             `json:"port"`
+	TTL         int             `json:"ttl"`
+	Poll        int             `json:"poll"`
+	Interfaces  json.RawMessage `json:"interfaces"` // optional override
+	Tags        []string        `json:"tags,omitempty"`
+	Sensors     []*Sensor       `json:"sensors"`
+	IpAddress   string
+}
+
+func (m *Metrics) Parse() error {
+	if ipAddress, err := utils.IpFromInterfaces(m.Interfaces); err != nil {
+		return err
+	} else {
+		m.IpAddress = ipAddress
+	}
+	for _, sensor := range m.Sensors {
+		if err := sensor.Parse(); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (m *Metrics) Serve() {
+	http.Handle(m.Url, prometheus.Handler())
+	listen := fmt.Sprintf("%s:%v", m.IpAddress, m.Port)
+	http.ListenAndServe(listen, nil)
+}

metrics/sensors.go

@@ -0,0 +1,116 @@
+package metrics
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/prometheus/client_golang/prometheus"
+	"os"
+	"os/exec"
+	"strconv"
+	"utils"
+
+	log "github.com/Sirupsen/logrus"
+)
+
+// A Sensor is a single measurement of the application.
+type Sensor struct {
+	Namespace string          `json:"namespace"`
+	Subsystem string          `json:"subsystem"`
+	Name      string          `json:"name"`
+	Help      string          `json:"help"` // help string returned by API
+	Type      string          `json:"type"`
+	Poll      int             `json:"poll"` // time in seconds
+	CheckExec json.RawMessage `json:"check"`
+	checkCmd  *exec.Cmd
+	collector prometheus.Collector
+}
+
+// PollTime implements Pollable for Sensor
+// It returns the sensor's poll interval.
+func (s Sensor) PollTime() int {
+	return s.Poll
+}
+
+// PollAction implements Pollable for Sensor.
+func (s Sensor) PollAction() {
+	if metricValue, err := s.getMetrics(); err != nil {
+		s.record(metricValue)
+	} else {
+		log.Errorln(err)
+	}
+}
+
+func (s Sensor) getMetrics() (string, error) {
+	// we'll pass stderr to the container's stderr, but stdout must
+	// be "clean" and not have anything other than what we intend
+	// to write to our collector.
+	s.checkCmd.Stderr = os.Stderr
+	if out, err := s.checkCmd.Output(); err != nil {
+		return "", err
+	} else {
+		return string(out[:]), nil
+	}
+}
+
+func (s Sensor) record(metricValue string) {
+	if val, err := strconv.ParseFloat(metricValue, 64); err != nil {
+		log.Errorln(err)
+	} else {
+		switch collector := s.collector.(type) {
+		case prometheus.Counter:
+			collector.Add(val)
+		case prometheus.Gauge:
+			collector.Set(val)
+		case prometheus.Histogram:
+			collector.Observe(val)
+		case prometheus.Summary:
+			collector.Observe(val)
+		}
+	}
+}
+
+func (s *Sensor) Parse() error {
+	if check, err := utils.ParseCommandArgs(s.CheckExec); err != nil {
+		return err
+	} else {
+		s.checkCmd = check
+	}
+	// the prometheus client lib's API here is baffling... they don't expose
+	// an interface or embed their Opts type in each of the Opts "subtypes",
+	// so we can't share the initialization.
+	switch {
+	case s.Type == "counter":
+		s.collector = prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: s.Namespace,
+			Subsystem: s.Subsystem,
+			Name:      s.Name,
+			Help:      s.Help,
+		})
+	case s.Type == "gauge":
+		s.collector = prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: s.Namespace,
+			Subsystem: s.Subsystem,
+			Name:      s.Name,
+			Help:      s.Help,
+		})
+	case s.Type == "histogram":
+		s.collector = prometheus.NewHistogram(prometheus.HistogramOpts{
+			Namespace: s.Namespace,
+			Subsystem: s.Subsystem,
+			Name:      s.Name,
+			Help:      s.Help,
+		})
+	case s.Type == "summary":
+		s.collector = prometheus.NewSummary(prometheus.SummaryOpts{
+			Namespace: s.Namespace,
+			Subsystem: s.Subsystem,
+			Name:      s.Name,
+			Help:      s.Help,
+		})
+	default:
+		return fmt.Errorf("Invalid sensor type: %s\n", s.Type)
+	}
+
+	prometheus.MustRegister(s.collector)
+	return nil
+}

utils/ips.go

@@ -2,6 +2,8 @@ package utils
 
 import (
 	"bytes"
+	"encoding/json"
+	"errors"
 	"fmt"
 	"net"
 	"regexp"
@@ -12,6 +14,37 @@ import (
 	log "github.com/Sirupsen/logrus"
 )
 
+func IpFromInterfaces(raw json.RawMessage) (string, error) {
+	interfaces, ifaceErr := ParseInterfaces(raw)
+	if ifaceErr != nil {
+		return "", ifaceErr
+	}
+
+	if ipAddress, err := GetIP(interfaces); err != nil {
+		return "", err
+	} else {
+		return ipAddress, nil
+	}
+}
+
+func ParseInterfaces(raw json.RawMessage) ([]string, error) {
+	if raw == nil {
+		return []string{}, nil
+	}
+	// Parse as a string
+	var jsonString string
+	if err := json.Unmarshal(raw, &jsonString); err == nil {
+		return []string{jsonString}, nil
+	}
+
+	var jsonArray []string
+	if err := json.Unmarshal(raw, &jsonArray); err == nil {
+		return jsonArray, nil
+	}
+
+	return []string{}, errors.New("interfaces must be a string or an array")
+}
+
 // GetIP determines the IP address of the container
 func GetIP(specList []string) (string, error) {
 

utils/utils.go

@@ -40,21 +40,3 @@ func StrToCmd(command string) *exec.Cmd {
 	}
 	return nil
 }
-
-func ParseInterfaces(raw json.RawMessage) ([]string, error) {
-	if raw == nil {
-		return []string{}, nil
-	}
-	// Parse as a string
-	var jsonString string
-	if err := json.Unmarshal(raw, &jsonString); err == nil {
-		return []string{jsonString}, nil
-	}
-
-	var jsonArray []string
-	if err := json.Unmarshal(raw, &jsonArray); err == nil {
-		return jsonArray, nil
-	}
-
-	return []string{}, errors.New("interfaces must be a string or an array")
-}