Update collectors to be able to directly access containers by their i…

…p address.

collector/config.go

@@ -17,12 +17,13 @@ package collector
 import (
 	"time"
 
+	"encoding/json"
 	"github.com/google/cadvisor/info/v1"
 )
 
 type Config struct {
 	//the endpoint to hit to scrape metrics
-	Endpoint string `json:"endpoint"`
+	Endpoint EndpointConfig `json:"endpoint"`
 
 	//holds information about different metrics that can be collected
 	MetricsConfig []MetricConfig `json:"metrics_config"`
@@ -52,11 +53,48 @@ type MetricConfig struct {
 
 type Prometheus struct {
 	//the endpoint to hit to scrape metrics
-	Endpoint string `json:"endpoint"`
+	Endpoint EndpointConfig `json:"endpoint"`
 
 	//the frequency at which metrics should be collected
 	PollingFrequency time.Duration `json:"polling_frequency"`
 
 	//holds names of different metrics that can be collected
 	MetricsConfig []string `json:"metrics_config"`
 }
+
+type EndpointConfig struct {
+	// The full URL of the endpoint to reach
+	URL string
+	// A configuration in which an actual URL is constructed from, using the container's ip address
+	URLConfig URLConfig
+}
+
+type URLConfig struct {
+	// the protocol to use for connecting to the endpoint. Eg 'http' or 'https'
+	Protocol string `json:"protocol"`
+
+	// the port to use for connecting to the endpoint. Eg '8778'
+	Port json.Number `json:"port"`
+
+	// the path to use for the endpoint. Eg '/metrics'
+	Path string `json:"path"`
+}
+
+func (ec *EndpointConfig) UnmarshalJSON(b []byte) error {
+	url := ""
+	config := URLConfig{
+		Protocol: "http",
+		Port:     "8000",
+	}
+
+	if err := json.Unmarshal(b, &url); err == nil {
+		ec.URL = url
+		return nil
+	}
+	err := json.Unmarshal(b, &config)
+	if err == nil {
+		ec.URLConfig = config
+		return nil
+	}
+	return err
+}

collector/config/sample_config_endpoint_config.json

@@ -0,0 +1,38 @@
+{
+  "endpoint" : {
+    "protocol": "https",
+    "port": 8000,
+    "path": "/nginx_status"
+  },
+  "metrics_config"  : [
+    { "name" : "activeConnections",
+      "metric_type" : "gauge",
+      "units" : "number of active connections",
+      "data_type" : "int",
+      "polling_frequency" : 10,
+      "regex" : "Active connections: ([0-9]+)"
+    },
+    { "name" : "reading",
+      "metric_type" : "gauge",
+      "units" : "number of reading connections",
+      "data_type" : "int",
+      "polling_frequency" : 10,
+      "regex" : "Reading: ([0-9]+) .*"
+    },
+    { "name" : "writing",
+      "metric_type" : "gauge",
+      "data_type" : "int",
+      "units" : "number of writing connections",
+      "polling_frequency" : 10,
+      "regex" : ".*Writing: ([0-9]+).*"
+    },
+    { "name" : "waiting",
+      "metric_type" : "gauge",
+      "units" : "number of waiting connections",
+      "data_type" : "int",
+      "polling_frequency" : 10,
+      "regex" : ".*Waiting: ([0-9]+)"
+    }
+  ]
+
+}

collector/config/sample_config_prometheus_endpoint_config.json

@@ -0,0 +1,10 @@
+{
+  "endpoint" : {
+    "protocol": "http",
+    "port": 8081,
+    "path": "/METRICS"
+  },
+  "polling_frequency" : 10,
+  "metrics_config" : [
+  ]
+}

collector/generic_collector.go

@@ -24,6 +24,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/google/cadvisor/container"
 	"github.com/google/cadvisor/info/v1"
 )
 
@@ -51,13 +52,15 @@ type collectorInfo struct {
 }
 
 //Returns a new collector using the information extracted from the configfile
-func NewCollector(collectorName string, configFile []byte, metricCountLimit int) (*GenericCollector, error) {
+func NewCollector(collectorName string, configFile []byte, metricCountLimit int, containerHandler container.ContainerHandler) (*GenericCollector, error) {
 	var configInJSON Config
 	err := json.Unmarshal(configFile, &configInJSON)
 	if err != nil {
 		return nil, err
 	}
 
+	configInJSON.Endpoint.configure(containerHandler)
+
 	//TODO : Add checks for validity of config file (eg : Accurate JSON fields)
 
 	if len(configInJSON.MetricsConfig) == 0 {
@@ -130,7 +133,7 @@ func (collector *GenericCollector) Collect(metrics map[string][]v1.MetricVal) (t
 	currentTime := time.Now()
 	nextCollectionTime := currentTime.Add(time.Duration(collector.info.minPollingFrequency))
 
-	uri := collector.configFile.Endpoint
+	uri := collector.configFile.Endpoint.URL
 	response, err := http.Get(uri)
 	if err != nil {
 		return nextCollectionTime, nil, err

collector/generic_collector_test.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/google/cadvisor/info/v1"
 
+	containertest "github.com/google/cadvisor/container/testing"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -44,7 +45,8 @@ func TestEmptyConfig(t *testing.T) {
 	configFile, err := ioutil.ReadFile("temp.json")
 	assert.NoError(err)
 
-	_, err = NewCollector("tempCollector", configFile, 100)
+	containerHandler := containertest.NewMockContainerHandler("mockContainer")
+	_, err = NewCollector("tempCollector", configFile, 100, containerHandler)
 	assert.Error(err)
 
 	assert.NoError(os.Remove("temp.json"))
@@ -74,7 +76,8 @@ func TestConfigWithErrors(t *testing.T) {
 	configFile, err := ioutil.ReadFile("temp.json")
 	assert.NoError(err)
 
-	_, err = NewCollector("tempCollector", configFile, 100)
+	containerHandler := containertest.NewMockContainerHandler("mockContainer")
+	_, err = NewCollector("tempCollector", configFile, 100, containerHandler)
 	assert.Error(err)
 
 	assert.NoError(os.Remove("temp.json"))
@@ -112,7 +115,8 @@ func TestConfigWithRegexErrors(t *testing.T) {
 	configFile, err := ioutil.ReadFile("temp.json")
 	assert.NoError(err)
 
-	_, err = NewCollector("tempCollector", configFile, 100)
+	containerHandler := containertest.NewMockContainerHandler("mockContainer")
+	_, err = NewCollector("tempCollector", configFile, 100, containerHandler)
 	assert.Error(err)
 
 	assert.NoError(os.Remove("temp.json"))
@@ -125,10 +129,28 @@ func TestConfig(t *testing.T) {
 	configFile, err := ioutil.ReadFile("config/sample_config.json")
 	assert.NoError(err)
 
-	collector, err := NewCollector("nginx", configFile, 100)
+	containerHandler := containertest.NewMockContainerHandler("mockContainer")
+	collector, err := NewCollector("nginx", configFile, 100, containerHandler)
 	assert.NoError(err)
 	assert.Equal(collector.name, "nginx")
-	assert.Equal(collector.configFile.Endpoint, "http://localhost:8000/nginx_status")
+	assert.Equal(collector.configFile.Endpoint.URL, "http://localhost:8000/nginx_status")
+	assert.Equal(collector.configFile.MetricsConfig[0].Name, "activeConnections")
+}
+
+func TestEndpointConfig(t *testing.T) {
+	assert := assert.New(t)
+	configFile, err := ioutil.ReadFile("config/sample_config_endpoint_config.json")
+	assert.NoError(err)
+
+	containerHandler := containertest.NewMockContainerHandler("mockContainer")
+	containerHandler.On("GetContainerIPAddress").Return(
+		"111.111.111.111",
+	)
+
+	collector, err := NewCollector("nginx", configFile, 100, containerHandler)
+	assert.NoError(err)
+	assert.Equal(collector.name, "nginx")
+	assert.Equal(collector.configFile.Endpoint.URL, "https://111.111.111.111:8000/nginx_status")
 	assert.Equal(collector.configFile.MetricsConfig[0].Name, "activeConnections")
 }
 
@@ -139,15 +161,16 @@ func TestMetricCollection(t *testing.T) {
 	configFile, err := ioutil.ReadFile("config/sample_config.json")
 	assert.NoError(err)
 
-	fakeCollector, err := NewCollector("nginx", configFile, 100)
+	containerHandler := containertest.NewMockContainerHandler("mockContainer")
+	fakeCollector, err := NewCollector("nginx", configFile, 100, containerHandler)
 	assert.NoError(err)
 
 	tempServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		fmt.Fprintln(w, "Active connections: 3\nserver accepts handled requests")
 		fmt.Fprintln(w, "5 5 32\nReading: 0 Writing: 1 Waiting: 2")
 	}))
 	defer tempServer.Close()
-	fakeCollector.configFile.Endpoint = tempServer.URL
+	fakeCollector.configFile.Endpoint.URL = tempServer.URL
 
 	metrics := map[string][]v1.MetricVal{}
 	_, metrics, errMetric := fakeCollector.Collect(metrics)
@@ -174,6 +197,7 @@ func TestMetricCollectionLimit(t *testing.T) {
 	configFile, err := ioutil.ReadFile("config/sample_config.json")
 	assert.NoError(err)
 
-	_, err = NewCollector("nginx", configFile, 1)
+	containerHandler := containertest.NewMockContainerHandler("mockContainer")
+	_, err = NewCollector("nginx", configFile, 1, containerHandler)
 	assert.Error(err)
 }

collector/prometheus_collector.go

@@ -24,6 +24,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/google/cadvisor/container"
 	"github.com/google/cadvisor/info/v1"
 )
 
@@ -46,13 +47,15 @@ type PrometheusCollector struct {
 }
 
 //Returns a new collector using the information extracted from the configfile
-func NewPrometheusCollector(collectorName string, configFile []byte, metricCountLimit int) (*PrometheusCollector, error) {
+func NewPrometheusCollector(collectorName string, configFile []byte, metricCountLimit int, containerHandler container.ContainerHandler) (*PrometheusCollector, error) {
 	var configInJSON Prometheus
 	err := json.Unmarshal(configFile, &configInJSON)
 	if err != nil {
 		return nil, err
 	}
 
+	configInJSON.Endpoint.configure(containerHandler)
+
 	minPollingFrequency := configInJSON.PollingFrequency
 
 	// Minimum supported frequency is 1s
@@ -108,7 +111,7 @@ func getMetricData(line string) string {
 
 func (collector *PrometheusCollector) GetSpec() []v1.MetricSpec {
 	specs := []v1.MetricSpec{}
-	response, err := http.Get(collector.configFile.Endpoint)
+	response, err := http.Get(collector.configFile.Endpoint.URL)
 	if err != nil {
 		return specs
 	}
@@ -153,7 +156,7 @@ func (collector *PrometheusCollector) Collect(metrics map[string][]v1.MetricVal)
 	currentTime := time.Now()
 	nextCollectionTime := currentTime.Add(time.Duration(collector.pollingFrequency))
 
-	uri := collector.configFile.Endpoint
+	uri := collector.configFile.Endpoint.URL
 	response, err := http.Get(uri)
 	if err != nil {
 		return nextCollectionTime, nil, err