add cluster_name to influxdb sink

kubernetes-retired · May 9, 2017 · f55b10c · f55b10c
1 parent b109f11
commit f55b10c
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 0 deletions.
diff --git a/common/influxdb/influxdb.go b/common/influxdb/influxdb.go
@@ -40,6 +40,7 @@ type InfluxdbConfig struct {
 	WithFields      bool
 	InsecureSsl     bool
 	RetentionPolicy string
+	ClusterName     string
 }
 
 func NewClient(c InfluxdbConfig) (InfluxdbClient, error) {
@@ -79,6 +80,7 @@ func BuildConfig(uri *url.URL) (*InfluxdbConfig, error) {
 		WithFields:      false,
 		InsecureSsl:     false,
 		RetentionPolicy: "0",
+		ClusterName:     "default",
 	}
 
 	if len(uri.Host) > 0 {
@@ -121,5 +123,9 @@ func BuildConfig(uri *url.URL) (*InfluxdbConfig, error) {
 		config.InsecureSsl = val
 	}
 
+	if len(opts["cluster_name"]) >= 1 {
+		config.ClusterName = opts["cluster_name"][0]
+	}
+
 	return &config, nil
 }
diff --git a/docs/sink-configuration.md b/docs/sink-configuration.md
@@ -35,6 +35,7 @@ The following options are available:
 * `secure` - Connect securely to InfluxDB (default: `false`)
 * `insecuressl` - Ignore SSL certificate validity (default: `false`)
 * `withfields` - Use [InfluxDB fields](storage-schema.md#using-fields) (default: `false`)
+* `cluster_name` - cluster name for different Kubernetes clusters. (default: `default`)
 
 ### Google Cloud Monitoring
 This sink supports monitoring metrics only.

diff --git a/events/sinks/influxdb/influxdb.go b/events/sinks/influxdb/influxdb.go
@@ -131,6 +131,9 @@ func (sink *influxdbSink) ExportEvents(eventBatch *core.EventBatch) {
 		if err != nil {
 			glog.Warningf("Failed to convert event to point: %v", err)
 		}
+
+		point.Tags["cluster_name"] = sink.c.ClusterName
+
 		dataPoints = append(dataPoints, *point)
 		if len(dataPoints) >= maxSendBatchSize {
 			sink.sendData(dataPoints)

diff --git a/metrics/sinks/influxdb/influxdb.go b/metrics/sinks/influxdb/influxdb.go
@@ -88,6 +88,9 @@ func (sink *influxdbSink) ExportData(dataBatch *core.DataBatch) {
 				},
 				Time: dataBatch.Timestamp.UTC(),
 			}
+
+			point.Tags["cluster_name"] = sink.c.ClusterName
+
 			dataPoints = append(dataPoints, point)
 			if len(dataPoints) >= maxSendBatchSize {
 				sink.sendData(dataPoints)
@@ -126,12 +129,14 @@ func (sink *influxdbSink) ExportData(dataBatch *core.DataBatch) {
 				},
 				Time: dataBatch.Timestamp.UTC(),
 			}
+
 			for key, value := range metricSet.Labels {
 				point.Tags[key] = value
 			}
 			for key, value := range labeledMetric.Labels {
 				point.Tags[key] = value
 			}
+			point.Tags["cluster_name"] = sink.c.ClusterName
 
 			dataPoints = append(dataPoints, point)
 			if len(dataPoints) >= maxSendBatchSize {