Merge pull request #1635 from andyxning/add_cluster_name_to_influxdb_sink

add cluster_name to influxdb sink

Author: DirectXMan12

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
 }

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.

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)

metrics/sinks/influxdb/influxdb.go

@@ -93,6 +93,9 @@ func (sink *influxdbSink) ExportData(dataBatch *core.DataBatch) {
 					point.Tags[key] = value
 				}
 			}
+
+			point.Tags["cluster_name"] = sink.c.ClusterName
+
 			dataPoints = append(dataPoints, point)
 			if len(dataPoints) >= maxSendBatchSize {
 				sink.sendData(dataPoints)
@@ -131,6 +134,7 @@ func (sink *influxdbSink) ExportData(dataBatch *core.DataBatch) {
 				},
 				Time: dataBatch.Timestamp.UTC(),
 			}
+
 			for key, value := range metricSet.Labels {
 				if value != "" {
 					point.Tags[key] = value
@@ -141,6 +145,7 @@ func (sink *influxdbSink) ExportData(dataBatch *core.DataBatch) {
 					point.Tags[key] = value
 				}
 			}
+			point.Tags["cluster_name"] = sink.c.ClusterName
 
 			dataPoints = append(dataPoints, point)
 			if len(dataPoints) >= maxSendBatchSize {