Support tags, both server-wide and per-job

config.go

@@ -9,11 +9,12 @@ import (
 
 // Config is the base of our configuration data structure
 type Config struct {
-	Jobs                    []Job          `yaml:"jobs"`
-	Selenium                SeleniumConfig `yaml:"selenium"`
-	Storage                 StorageConfig  `yaml:"storage,omitempty"`
-	ReportInternalMetrics   bool           `yaml:"report-internal-metrics,omitempty"`
-	InternalMetricsInterval uint           `yaml:"internal-metrics-gathering-interval,omitempty"`
+	Jobs                    []Job             `yaml:"jobs"`
+	Selenium                SeleniumConfig    `yaml:"selenium"`
+	Storage                 StorageConfig     `yaml:"storage,omitempty"`
+	ReportInternalMetrics   bool              `yaml:"report-internal-metrics,omitempty"`
+	InternalMetricsInterval uint              `yaml:"internal-metrics-gathering-interval,omitempty"`
+	ServerTags              map[string]string `yaml:"server-tags,omitempty"`
 }
 
 // SeleniumConfig holds the configuration for our Selenium service

example/config.yaml

@@ -32,7 +32,7 @@ storage:
         port: 8125
         metric-namespace: crabby.crabby-sfo-1
         tags:
-            - crabby-sfo-1
+            - instance: crabby-sfo-1
     prometheus:
         host: prometheus.mysite.org
         port: 9091
@@ -42,6 +42,8 @@ storage:
         port: 5555
         metric-namespace: crabby.crabby-sfo-1
         tags:
-            - crabby-sfo-1
+            - instance: crabby-sfo-1
 report-internal-metrics: true
 internal-metrics-gathering-interval: 15
+server-tags:
+    - region: us-east-1

jobs.go

@@ -6,17 +6,19 @@ import (
 	"log"
 	"math/rand"
 	"net/http"
+	"net/url"
 	"sync"
 	"time"
 )
 
 // Job holds a single Selenium
 type Job struct {
-	Name     string   `yaml:"name"`
-	URL      string   `yaml:"url"`
-	Type     string   `yaml:"type"`
-	Interval uint16   `yaml:"interval"`
-	Cookies  []Cookie `yaml:"cookies,omitempty"`
+	Name     string            `yaml:"name"`
+	URL      string            `yaml:"url"`
+	Type     string            `yaml:"type"`
+	Tags     map[string]string `yaml:"tags,omitempty"`
+	Interval uint16            `yaml:"interval"`
+	Cookies  []Cookie          `yaml:"cookies,omitempty"`
 }
 
 // JobRunner holds channels and state related to running Jobs
@@ -72,6 +74,8 @@ func StartJobs(ctx context.Context, wg *sync.WaitGroup, c *Config, storage *Stor
 
 	for _, j := range jobs {
 
+		j.populateAutomaticTags(c)
+
 		// If we've been provided with an offset for staggering jobs, sleep for a random
 		// time interval (where: 0 < sleepDur < offset) before starting that job's timer
 		if c.Selenium.JobStaggerOffset > 0 {
@@ -85,3 +89,21 @@ func StartJobs(ctx context.Context, wg *sync.WaitGroup, c *Config, storage *Stor
 	}
 
 }
+
+func (j *Job) populateAutomaticTags(c *Config) {
+	u, err := url.Parse(j.URL)
+	if err != nil {
+		return
+	}
+
+	// Add all of our server tags
+	for k, v := range c.ServerTags {
+		j.Tags[k] = v
+	}
+
+	// Add the hostname from our job
+	j.Tags["crabby.job_hostname"] = u.Hostname()
+
+	// Add the type of check (simple, selenium)
+	j.Tags["crabby.job_type"] = j.Type
+}

selenium.go

@@ -120,20 +120,11 @@ func RunSeleniumTest(j Job, seleniumServer string, storage *Storage) {
 		return
 	}
 
-	fmt.Println(j.Name, "DNS time:", wr.ri.dnsDuration)
-	storage.MetricDistributor <- makeMetric(j.Name, "dns_duration", wr.ri.dnsDuration)
-
-	fmt.Println(j.Name, "Connection establishment time", wr.ri.serverConnectionDuration)
-	storage.MetricDistributor <- makeMetric(j.Name, "server_connection_duration", wr.ri.serverConnectionDuration)
-
-	fmt.Println(j.Name, "Response time:", wr.ri.serverResponseDuration)
-	storage.MetricDistributor <- makeMetric(j.Name, "server_response_duration", wr.ri.serverResponseDuration)
-
-	fmt.Println(j.Name, "Server processing time:", wr.ri.serverProcessingDuration)
-	storage.MetricDistributor <- makeMetric(j.Name, "server_processing_duration", wr.ri.serverProcessingDuration)
-
-	fmt.Println(j.Name, "DOM rendering time:", wr.ri.domRenderingDuration)
-	storage.MetricDistributor <- makeMetric(j.Name, "dom_rendering_duration", wr.ri.domRenderingDuration)
+	storage.MetricDistributor <- makeMetric(j, "dns_duration", wr.ri.dnsDuration)
+	storage.MetricDistributor <- makeMetric(j, "server_connection_duration", wr.ri.serverConnectionDuration)
+	storage.MetricDistributor <- makeMetric(j, "server_response_duration", wr.ri.serverResponseDuration)
+	storage.MetricDistributor <- makeMetric(j, "server_processing_duration", wr.ri.serverProcessingDuration)
+	storage.MetricDistributor <- makeMetric(j, "dom_rendering_duration", wr.ri.domRenderingDuration)
 
 	err = wr.wd.Close()
 	if err != nil {

simple.go

@@ -88,7 +88,7 @@ func RunSimpleTest(ctx context.Context, j Job, storage *Storage, client *http.Cl
 	}
 
 	// Send our server response code as an event
-	storage.EventDistributor <- makeEvent(j.Name, resp.StatusCode)
+	storage.EventDistributor <- makeEvent(j, resp.StatusCode)
 
 	// Even though we never read the response body, if we don't close it,
 	// the http.Transport goroutines will terminate and the app will eventually
@@ -109,16 +109,16 @@ func RunSimpleTest(ctx context.Context, j Job, storage *Storage, client *http.Cl
 
 	switch url.Scheme {
 	case "https":
-		storage.MetricDistributor <- makeMetric(j.Name, "dns_duration", t1.Sub(t0).Seconds()*1000)
-		storage.MetricDistributor <- makeMetric(j.Name, "server_connection_duration", t2.Sub(t1).Seconds()*1000)
-		storage.MetricDistributor <- makeMetric(j.Name, "tls_handshake_duration", t3.Sub(t2).Seconds()*1000)
-		storage.MetricDistributor <- makeMetric(j.Name, "server_processing_duration", t4.Sub(t3).Seconds()*1000)
-		storage.MetricDistributor <- makeMetric(j.Name, "server_response_duration", t5.Sub(t4).Seconds()*1000)
+		storage.MetricDistributor <- makeMetric(j, "dns_duration", t1.Sub(t0).Seconds()*1000)
+		storage.MetricDistributor <- makeMetric(j, "server_connection_duration", t2.Sub(t1).Seconds()*1000)
+		storage.MetricDistributor <- makeMetric(j, "tls_handshake_duration", t3.Sub(t2).Seconds()*1000)
+		storage.MetricDistributor <- makeMetric(j, "server_processing_duration", t4.Sub(t3).Seconds()*1000)
+		storage.MetricDistributor <- makeMetric(j, "server_response_duration", t5.Sub(t4).Seconds()*1000)
 
 	case "http":
-		storage.MetricDistributor <- makeMetric(j.Name, "dns_duration", t1.Sub(t0).Seconds()*1000)
-		storage.MetricDistributor <- makeMetric(j.Name, "server_connection_duration", t3.Sub(t1).Seconds()*1000)
-		storage.MetricDistributor <- makeMetric(j.Name, "server_processing_duration", t4.Sub(t3).Seconds()*1000)
-		storage.MetricDistributor <- makeMetric(j.Name, "server_response_duration", t5.Sub(t4).Seconds()*1000)
+		storage.MetricDistributor <- makeMetric(j, "dns_duration", t1.Sub(t0).Seconds()*1000)
+		storage.MetricDistributor <- makeMetric(j, "server_connection_duration", t3.Sub(t1).Seconds()*1000)
+		storage.MetricDistributor <- makeMetric(j, "server_processing_duration", t4.Sub(t3).Seconds()*1000)
+		storage.MetricDistributor <- makeMetric(j, "server_response_duration", t5.Sub(t4).Seconds()*1000)
 	}
 }

storage.go

@@ -11,14 +11,17 @@ import (
 // Metric holds one metric data point
 type Metric struct {
 	Name      string
+	Timing    string
 	Value     float64
+	Tags      map[string]string
 	Timestamp time.Time
 }
 
 // Event holds one monitoring event
 type Event struct {
 	Name         string
 	ServerStatus int
+	Tags         map[string]string
 	Timestamp    time.Time
 }
 
@@ -168,25 +171,34 @@ func (s *Storage) storageDistributor(ctx context.Context, wg *sync.WaitGroup) er
 }
 
 // makeMetric creates a Metric from raw values and metric names
-func makeMetric(name string, timing string, value float64) Metric {
+func makeMetric(j Job, timing string, value float64) Metric {
 
 	m := Metric{
-		Name:      fmt.Sprintf("%v.%v", name, timing),
+		Name:      j.Name,
+		Timing:    timing,
 		Value:     value,
 		Timestamp: time.Now(),
 	}
 
+	for k, v := range j.Tags {
+		m.Tags[k] = v
+	}
+
 	return m
 }
 
 // makeEvent creates an Event from raw values and event names
-func makeEvent(name string, status int) Event {
+func makeEvent(j Job, status int) Event {
 
 	e := Event{
-		Name:         name,
+		Name:         j.Name,
 		ServerStatus: status,
 		Timestamp:    time.Now(),
 	}
 
+	for k, v := range j.Tags {
+		e.Tags[k] = v
+	}
+
 	return e
 }

storage_dogstatsd.go

@@ -12,10 +12,9 @@ import (
 // DogstatsdConfig describes the YAML-provided configuration for a Datadog
 // DogstatsD storage backend
 type DogstatsdConfig struct {
-	Host      string   `yaml:"host"`
-	Port      int      `yaml:"port"`
-	Namespace string   `yaml:"metric-namespace"`
-	Tags      []string `yaml:"tags,omitempty"`
+	Host      string `yaml:"host"`
+	Port      int    `yaml:"port"`
+	Namespace string `yaml:"metric-namespace"`
 }
 
 // DogstatsdStorage holds the configuration for a DogstatsD storage backend
@@ -64,14 +63,19 @@ func (d DogstatsdStorage) sendMetric(m Metric) error {
 	var metricName string
 
 	if d.Namespace == "" {
-		metricName = fmt.Sprintf("crabby.%v", m.Name)
+		metricName = fmt.Sprintf("crabby.%v.%v", m.Name, m.Timing)
 	} else {
-		metricName = fmt.Sprintf("%v.%v", d.Namespace, m.Name)
+		metricName = fmt.Sprintf("%v.%v.%v", d.Namespace, m.Name, m.Timing)
+	}
+
+	// Add all of our metric tags to the metric payload
+	for k, v := range m.Tags {
+		d.DogstatsdConn.Tags = append(d.DogstatsdConn.Tags, k+":"+v)
 	}
 
 	err := d.DogstatsdConn.TimeInMilliseconds(metricName, m.Value, nil, 1)
 	if err != nil {
-		log.Printf("Could not send metric %v: %v\n", m.Name, err)
+		log.Printf("Could not send metric %v.%v: %v\n", m.Name, m.Timing, err)
 		return err
 	}
 
@@ -100,16 +104,17 @@ func (d DogstatsdStorage) sendEvent(e Event) error {
 		Message: fmt.Sprintf("%v is returning a HTTP status code of %v", e.Name, e.ServerStatus),
 	}
 
+	// Add all of our metric tags to the metric payload
+	for k, v := range e.Tags {
+		d.DogstatsdConn.Tags = append(d.DogstatsdConn.Tags, k+":"+v)
+	}
+
 	if (e.ServerStatus < 400) && (e.ServerStatus > 0) {
 		sc.Status = statsd.Ok
 	} else {
 		sc.Status = statsd.Critical
 	}
 
-	for _, t := range d.DogstatsdConn.Tags {
-		sc.Tags = append(sc.Tags, t)
-	}
-
 	err := d.DogstatsdConn.ServiceCheck(sc)
 
 	return err
@@ -127,8 +132,5 @@ func NewDogstatsdStorage(c *Config) DogstatsdStorage {
 		log.Println("Warning: could not create dogstatsd connection", err)
 	}
 
-	for _, t := range c.Storage.Dogstatsd.Tags {
-		d.DogstatsdConn.Tags = append(d.DogstatsdConn.Tags, t)
-	}
 	return d
 }

storage_graphite.go

@@ -67,9 +67,9 @@ func (g GraphiteStorage) sendMetric(m Metric) error {
 	valStr := strconv.FormatFloat(m.Value, 'f', 3, 64)
 
 	if g.Namespace == "" {
-		metricName = fmt.Sprintf("crabby.%v", m.Name)
+		metricName = fmt.Sprintf("crabby.%v.%v", m.Name, m.Timing)
 	} else {
-		metricName = fmt.Sprintf("%v.%v", g.Namespace, m.Name)
+		metricName = fmt.Sprintf("%v.%v.%v", g.Namespace, m.Name, m.Timing)
 	}
 
 	if m.Timestamp.IsZero() {
@@ -89,7 +89,7 @@ func (g GraphiteStorage) sendMetric(m Metric) error {
 
 	err := g.GraphiteConn.SendMetric(gm)
 	if err != nil {
-		log.Printf("Could not send metric %v: %v\n", m.Name, err)
+		log.Printf("Could not send metric %v.%v: %v\n", m.Name, m.Timing, err)
 		return err
 	}
 

storage_prometheus.go

@@ -18,11 +18,28 @@ type PrometheusConfig struct {
 	Namespace string `yaml:"metric-namespace,omitempty"`
 }
 
+// CurrentMetrics is a map of Metrics + a mutex that maintains the most recent metrics
+// that we are collecting.  These are maintained so that a Prometheus client can connect
+// to the built-in Prometheus endpoint and fetch the very latest result for this
+// metric.
+type CurrentMetrics struct {
+	metrics map[string]MetricStore
+	sync.RWMutex
+}
+
+// MetricStore holds a Prometheus GaugeVec for a given metric
+type MetricStore struct {
+	name     string
+	timing   string
+	gaugeVec prometheus.GaugeVec
+}
+
 // PrometheusStorage holds the configuration for a Graphite storage backend
 type PrometheusStorage struct {
-	Namespace string
-	Registry  *prometheus.Registry
-	url       string
+	Namespace      string
+	Registry       *prometheus.Registry
+	url            string
+	currentMetrics *CurrentMetrics
 }
 
 // NewPrometheusStorage sets up a new Prometheus storage backend
@@ -61,10 +78,7 @@ func (p PrometheusStorage) processMetrics(ctx context.Context, wg *sync.WaitGrou
 	for {
 		select {
 		case m := <-mchan:
-			err := p.sendMetric(m)
-			if err != nil {
-				log.Println(err)
-			}
+			p.storeCurrentMetric(m)
 		case <-ctx.Done():
 			log.Println("Cancellation request recieved.  Cancelling metrics processor.")
 			return
@@ -105,3 +119,47 @@ func (p PrometheusStorage) sendEvent(e Event) error {
 	var err error
 	return err
 }
+
+func (p PrometheusStorage) storeCurrentMetric(m Metric) {
+	p.currentMetrics.Lock()
+	defer p.currentMetrics.Unlock()
+
+	// See if this metric name exists in the current metrics map
+	cm, exists := p.currentMetrics.metrics[m.Timing]
+
+	// If the metric doesn't exist, create a new MetricsStore and Gauge for it
+	if !exists {
+		var glabel []string
+
+		for k := range m.Tags {
+			glabel = append(glabel, k)
+		}
+
+		gauge := prometheus.NewGaugeVec(
+			prometheus.GaugeOpts{
+				Name: m.Timing,
+			}, glabel,
+		)
+
+		// Create a MetricsStore/Gauge
+		ms := MetricStore{
+			name:     m.Name,
+			gaugeVec: *gauge,
+		}
+
+		// ...and add that to our map
+		p.currentMetrics.metrics[m.Name] = ms
+
+		// Register the Gauge with Prometheus
+		prometheus.MustRegister(p.currentMetrics.metrics[m.Timing].gaugeVec)
+
+	} else {
+		labels := prometheus.Labels{}
+		for k, v := range m.Tags {
+			labels[k] = v
+		}
+		labels["name"] = m.Name
+		cm.gaugeVec.With(labels).Set(m.Value)
+	}
+
+}