support sls sink with external labels. (#267)

* support sls sink for external labels.

* fix markdown bug

docs/en/sls-sink.md

@@ -3,20 +3,21 @@
 *This sink supports sls (Log Service of Alibaba Cloud)*.
 To use the sls sink add the following flag:
 
-	--sink=sls:<SLS_ENDPOINTL>&logStore=[your_logstore]&project=[your_project]&topic=[topic_for_log]
+	--sink=sls:<SLS_ENDPOINTL>?logStore=[your_logstore]&project=[your_project]&topic=[topic_for_log]&label=<key,value>
 
 
 The following options are available:
 * `project` - Project of SLS instance. 
 * `logStore` - logStore of SLS instance project. 
 * `topic` - topic for every log sent to SLS. 
+* `label` - Custom labels on alerting message.(such as clusterId), format is label=ClusterId,test_clusterId&label=RegionId,test_regionId&label=UserId,test_uid
 * `accessKeyId` - optional param. aliyun access key to sink to sls. 
 * `accessKeySecret` - optional param. aliyun access key secret to sink to sls.
 * `internal` - optional param. if true, it will sink to sls through aliyun internal network connection. 
 
 For example:
 
-    --sink=sls:https://sls.aliyuncs.com?project=my_sls_project&logStore=my_sls_project_logStore&topic=k8s-cluster-dev
+    --sink=sls:https://sls.aliyuncs.com?project=my_sls_project&logStore=my_sls_project_logStore&topic=k8s-cluster-dev&label=Key1,Value1&label=Key2,Value2
 
 #### How to config aliyun access key.
 

sinks/sls/sls.go

@@ -4,7 +4,7 @@
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-//     http://www.apache.org/licenses/LICENSE-2.0
+//	http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -28,6 +28,7 @@ import (
 	"net/url"
 	"os"
 	"strconv"
+	"strings"
 )
 
 const (
@@ -38,9 +39,9 @@ const (
 )
 
 /*
-	Usage:
-	--sink=sls:https://sls.aliyuncs.com?logStore=[your_log_store]&project=[your_project_name]
-*/
+ * Usage:
+ * --sink=sls:https://sls.aliyuncs.com?logStore=[your_log_store]&project=[your_project_name]&label=<key,value>
+ */
 type SLSSink struct {
 	Config   *Config
 	Project  string
@@ -56,6 +57,7 @@ type Config struct {
 	internal        bool
 	accessKeyId     string
 	accessKeySecret string
+	label           map[string]string
 }
 
 func (s *SLSSink) Name() string {
@@ -152,6 +154,15 @@ func (s *SLSSink) eventToContents(event *v1.Event) []*sls.Log_Content {
 		})
 	}
 
+	if len(s.Config.label) > 0 {
+		for key, value := range s.Config.label {
+			contents = append(contents, &sls.Log_Content{
+				Key:   &key,
+				Value: &value,
+			})
+		}
+	}
+
 	return contents
 }
 
@@ -226,9 +237,28 @@ func parseConfig(uri *url.URL) (*Config, error) {
 			c.internal = internal
 		}
 	}
+
+	if len(opts["label"]) >= 1 {
+		labelsStrs := opts["label"]
+		c.label = parseLabels(labelsStrs)
+	}
+
 	return c, nil
 }
 
+func parseLabels(labelsStrs []string) map[string]string {
+	labels := make(map[string]string)
+	for _, kv := range labelsStrs {
+		kvItems := strings.Split(kv, ",")
+		if len(kvItems) == 2 {
+			labels[kvItems[0]] = kvItems[1]
+		} else {
+			klog.Errorf("parse sls labels error. labelsStr: %v, kv format error: %v", labelsStrs, kv)
+		}
+	}
+	return labels
+}
+
 // newClient creates client using AK or metadata
 func newClient(c *Config) (*sls.Client, error) {
 	// get region from env

sinks/sls/sls_test.go

@@ -0,0 +1,12 @@
+package sls
+
+import (
+	"net/url"
+	"testing"
+)
+
+func TestSLSSinkParse(t *testing.T) {
+	u, _ := url.Parse("sls:https://sls.aliyuncs.com?internal=true&logStore=k8s-event&project=test_projectId&topic=&label=ClusterId,test_clusterId&label=RegionId,test_regionId&label=UserId,test_uid")
+	d, _ := NewSLSSink(u)
+	t.Logf("sls sink config: %v", d.Config)
+}