Merge pull request #30 from BSWANG/master

process `aliyun_logs_` env to support k8s pod and containers

docker-images/fluentd.tpl

@@ -32,6 +32,7 @@ host "#{Socket.gethostname}"
 
 {{if $.source.Application}}docker_app {{ $.source.Application }} {{end}}
 {{if $.source.Service}}docker_service {{ $.source.Service }} {{end}}
+{{if $.source.POD}}k8s_pod {{ $.source.POD }} {{end}}
 {{if $.source.Container}}docker_container {{ $.source.Container }} {{end}}
 </record>
 </filter>

pilot/pilot.go

@@ -25,10 +25,12 @@ aliyun.log: /var/log/hello.log[:json][;/var/log/abc/def.log[:txt]]
 */
 
 const LABEL_SERVICE_LOGS = "aliyun.logs."
+const ENV_SERVICE_LOGS = "aliyun_logs_"
 const FLUENTD_CONF_HOME = "/etc/fluentd"
 
 const LABEL_PROJECT = "com.docker.compose.project"
 const LABEL_SERVICE = "com.docker.compose.service"
+const LABEL_POD = "io.kubernetes.pod.name"
 
 type Pilot struct {
 	mutex        sync.Mutex
@@ -97,6 +99,7 @@ func (p *Pilot) watch() error {
 type Source struct {
 	Application string
 	Service     string
+	POD         string
 	Container   string
 }
 
@@ -172,6 +175,7 @@ func (p *Pilot) newContainer(containerJSON types.ContainerJSON) error {
 	jsonLogPath := containerJSON.LogPath
 	mounts := containerJSON.Mounts
 	labels := containerJSON.Config.Labels
+	env := containerJSON.Config.Env
 
 	//logConfig.containerDir match types.mountPoint
 	/**
@@ -185,9 +189,22 @@ func (p *Pilot) newContainer(containerJSON types.ContainerJSON) error {
 	source := Source{
 		Application: labels[LABEL_PROJECT],
 		Service:     labels[LABEL_SERVICE],
+		POD:         labels[LABEL_POD],
 		Container:   strings.TrimPrefix(containerJSON.Name, "/"),
 	}
 
+	for _, e := range env {
+		if !strings.HasPrefix(e, ENV_SERVICE_LOGS) {
+			continue
+		}
+		envLabel := strings.Split(e, "=")
+		if len(envLabel) == 2 {
+			labelKey := strings.Replace(envLabel[0], "_", ".", -1)
+			labels[labelKey] = envLabel[1]
+		}
+
+	}
+
 	logConfigs, err := p.getLogConfigs(jsonLogPath, mounts, labels)
 	if err != nil {
 		return err