fix issue #150 and update fluentd version to 1.2.6

Dockerfile.fluentd

@@ -17,8 +17,8 @@ RUN apk update && \
     apk add python && \
     apk add lsof && \
     apk add ca-certificates wget && \
-    gem install fluentd -v 1.1.0 --no-ri --no-rdoc && \
-    gem install fluent-plugin-elasticsearch --no-ri --no-rdoc && \
+    gem install fluentd -v 1.2.6 --no-ri --no-rdoc && \
+    gem install fluent-plugin-elasticsearch -v ">=2.0.0" --no-ri --no-rdoc && \
     gem install gelf -v "~> 3.0.0" --no-ri --no-rdoc && \
     gem install aliyun_sls_sdk -v ">=0.0.9" --no-ri --no-rdoc && \
     gem install remote_syslog_logger -v ">=1.0.1" --no-ri --no-rdoc && \
@@ -42,4 +42,5 @@ HEALTHCHECK CMD /pilot/healthz
 VOLUME /etc/fluentd/conf.d
 VOLUME /pilot/pos
 WORKDIR /pilot/
+ENV PILOT_TYPE=fluentd
 ENTRYPOINT ["/pilot/entrypoint"]

assets/fluentd/plugins/out_aliyun_sls.rb

@@ -33,11 +33,7 @@ def shutdown
     end
 
     def format(tag, time, record)
-      if record["@target"]
-        [tag, time, record].to_msgpack
-      else
-        super
-      end
+      [tag, time, record].to_msgpack
     end
 
     def client
@@ -75,15 +71,15 @@ def getLogItem(record)
     def write(chunk)
       log_list_hash = {}
       chunk.msgpack_each do |tag, time, record|
-        if record and record["@target"]
-          logStoreName = record["@target"]
-          record.delete("@target")
+        if record and record["_target"]
+          logStoreName = record["_target"]
+          record.delete("_target")
           if not log_list_hash[logStoreName]
             log_list_hash[logStoreName] = []
           end
           log_list_hash[logStoreName] << getLogItem(record)
         else
-          log.warn "no @target key in record: #{record}, tag: #{tag}, time: #{time}"
+          log.warn "no _target key in record: #{record}, tag: #{tag}, time: #{time}"
         end
       end
 

docs/filebeat/docs.md

@@ -17,6 +17,7 @@ docker run --rm -it \
    -v /var/run/docker.sock:/var/run/docker.sock \
    -v /etc/localtime:/etc/localtime \
    -v /:/host:ro \
+   --cap-add SYS_ADMIN \
    registry.cn-hangzhou.aliyuncs.com/acs/log-pilot:0.9.5-filebeat
 ```
 
@@ -31,6 +32,7 @@ docker run --rm -it \
     -v /var/run/docker.sock:/var/run/docker.sock \
     -v /etc/localtime:/etc/localtime \
     -v /:/host:ro \
+    --cap-add SYS_ADMIN \
     -e LOGGING_OUTPUT=elasticsearch \
     -e ELASTICSEARCH_HOST=${ELASTICSEARCH_HOST} \
     -e ELASTICSEARCH_PORT=${ELASTICSEARCH_PORT} \

docs/fluentd/docs.md

@@ -31,6 +31,7 @@ The command below run pilot with elastichsearch output, this makes log-pilot sen
 docker run --rm -it \
     -v /var/run/docker.sock:/var/run/docker.sock \
     -v /:/host:ro \
+    --cap-add SYS_ADMIN \
     -e LOGGING_OUTPUT=elasticsearch \
     -e ELASTICSEARCH_HOST=${ELASTICSEARCH_HOST} \
     -e ELASTICSEARCH_PORT=${ELASTICSEARCH_PORT} \

docs/fluentd/output/aliyun_sls.md

@@ -36,11 +36,13 @@ docker run --rm -it \
     -v /var/run/docker.sock:/var/run/docker.sock \
     -v /etc/localtime:/etc/localtime \
     -v /:/host:ro \
+    --cap-add SYS_ADMIN \
     -e LOGGING_OUTPUT=aliyun_sls \
     -e ALIYUNSLS_PROJECT="your-aliyun-sls-project-name"  \
     -e ALIYUNSLS_REGION_ENDPOINT=cn-hangzhou.log.aliyuncs.com \
     -e ALIYUNSLS_ACCESS_KEY_ID="your-access-key-id" \
     -e ALIYUNSLS_ACCESS_KEY_SECRET="your-access-key-secret"  \
+    -e ALIYUNSLS_NEED_CREATE_LOGSTORE="true" \
     registry.cn-hangzhou.aliyuncs.com/acs/log-pilot:0.9.5-fluentd
 ````
 

pilot/fluentd_piloter.go

@@ -7,6 +7,8 @@ import (
 	"os/exec"
 	"syscall"
 	"time"
+	"strconv"
+	"strings"
 )
 
 const (
@@ -17,6 +19,7 @@ const (
 	FLUENTD_PLUGINS   = FLUENTD_BASE_CONF + "/plugins"
 
 	ENV_FLUENTD_OUTPUT = "FLUENTD_OUTPUT"
+	ENV_FLUENTD_WORKER = "FLUENTD_WORKER"
 )
 
 var fluentd *exec.Cmd
@@ -39,9 +42,15 @@ func (p *FluentdPiloter) Start() error {
 	}
 
 	log.Info("starting fluentd")
+	worker := os.Getenv(ENV_FLUENTD_WORKER)
+	if _, err := strconv.Atoi(worker); worker == "" || err != nil {
+		worker = "1"
+	}
+
 	fluentd = exec.Command(FLUENTD_EXEC_CMD,
 		"-c", FLUENTD_CONF_FILE,
-		"-p", FLUENTD_PLUGINS)
+		"-p", FLUENTD_PLUGINS,
+		"--workers", worker)
 	fluentd.Stderr = os.Stderr
 	fluentd.Stdout = os.Stdout
 	err := fluentd.Start()
@@ -108,7 +117,7 @@ func shell(command string) string {
 	if err != nil {
 		fmt.Printf("error %v", err)
 	}
-	return string(out)
+	return strings.TrimSpace(string(out))
 }
 
 func (p *FluentdPiloter) GetConfHome() string {

pilot/pilot.go

@@ -646,7 +646,7 @@ func (p *Pilot) getLogConfigs(jsonLogPath string, mounts []types.MountPoint, lab
 
 	var labelNames []string
 	//sort keys
-	for k, _ := range labels {
+	for k := range labels {
 		labelNames = append(labelNames, k)
 	}