Merge pull request #23 from toolchainX/master

add support for creating log store when using aliyun sls plugin

.gitignore

@@ -23,3 +23,4 @@ _testmain.go
 *.test
 *.prof
 bin/
+docker-images/pilot

docker-images/Dockerfile

@@ -7,7 +7,7 @@ RUN apk update && apk upgrade && \
   gem install fluentd -v "~> 0.12.0" --no-ri --no-rdoc && \
   gem install fluent-plugin-elasticsearch --no-ri --no-rdoc && \
   gem install gelf -v "~> 3.0.0" --no-ri --no-rdoc && \
-  gem install aliyun_sls_sdk -v ">=0.0.5" --no-ri --no-rdoc && \
+  gem install aliyun_sls_sdk -v ">=0.0.9" --no-ri --no-rdoc && \
   apk del build-base ruby-dev && \
   rm -rf /root/.gem
 

docker-images/config.default

@@ -78,11 +78,14 @@ cat >> $FLUENTD_CONFIG << EOF
 <match docker.**>
 @type aliyun_sls
 
-project             $ALIYUNSLS_PROJECT
-region_endpoint     $ALIYUNSLS_REGION_ENDPOINT
-access_key_id       $ALIYUNSLS_ACCESS_KEY_ID
-access_key_secret   $ALIYUNSLS_ACCESS_KEY_SECRET
-ssl_verify          $SSL_VERIFY
+project              $ALIYUNSLS_PROJECT
+region_endpoint      $ALIYUNSLS_REGION_ENDPOINT
+access_key_id        $ALIYUNSLS_ACCESS_KEY_ID
+access_key_secret    $ALIYUNSLS_ACCESS_KEY_SECRET
+ssl_verify           ${SSL_VERIFY:-false}
+need_create_logstore ${ALIYUNSLS_NEED_CREATE_LOGSTORE:-false}
+create_logstore_ttl  ${ALIYUNSLS_CREATE_LOGSTORE_TTL:-1}
+create_logstore_shard_count ${ALIYUNSLS_CREATE_LOGSTORE_SHARD_COUNT:-2}
 
 flush_interval 3s
 </match>

docker-images/plugins/out_aliyun_sls.rb

@@ -8,11 +8,15 @@ class AliyunSlsOutput < BufferedOutput
     config_param :access_key_id, :string, :default => nil
     config_param :access_key_secret, :string, :default => nil
     config_param :ssl_verify, :bool, :default => false
+    config_param :need_create_logstore, :bool, :default => false
+    config_param :create_logstore_ttl, :integer, :default => 1
+    config_param :create_logstore_shard_count, :integer, :default => 2
 
     def initialize
       super
       require "aliyun_sls_sdk/protobuf"
-      require "aliyun_sls_sdk/connection"
+      require "aliyun_sls_sdk"
+      @log_store_created = false
     end
 
     def configure(conf)
@@ -38,55 +42,78 @@ def format(tag, time, record)
 
     def client
       @topic = `hostname`.strip
-      @_sls_con ||= AliyunSlsSdk::Connection.new(@project, @region_endpoint, @access_key_id, @access_key_secret)
+      @_sls_con ||= AliyunSlsSdk::LogClient.new(@region_endpoint, @access_key_id, @access_key_secret, @ssl_verify)
+    end
+
+    def createLogStore(logstore_name)
+      retries = 2
+      begin
+        createLogStoreResp = client.create_logstore(@project, logstore_name, @create_logstore_ttl, @create_logstore_shard_count)
+      rescue AliyunSlsSdk::LogException => e
+        if e.errorCode == "LogStoreAlreadyExist"
+          log.warn "logstore #{logstore_name} already exist"
+        else
+          raise
+        end
+      rescue => e
+        if retries > 0
+          log.warn "Error caught when creating logs store: #{e}"
+          retries -= 1
+          retry
+        end
+      end
+    end
+
+    def getLogItem(record)
+      contents = {}
+      record.each { |k, v|
+        contents[k] = v
+      }
+      AliyunSlsSdk::LogItem.new(nil, contents)
     end
 
     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 not log_list_hash[logStoreName]
-            log_list = AliyunSlsSdk::Protobuf::LogGroup.new(:logs => [], :topic => @topic, :source => @source)
-            log_list_hash[logStoreName] = log_list
+            log_list_hash[logStoreName] = []
           end
-          log = AliyunSlsSdk::Protobuf::Log.new(:time => Time.now.to_i, :contents => [])
-          pack_log_item(log_list_hash[logStoreName], log, record)
+          log_list_hash[logStoreName] << getLogItem(record)
         else
           log.warn "no @target key in record: #{record}, tag: #{tag}, time: #{time}"
         end
       end
-      log_list_hash.each do |storeName, log_list|
-        retries = 2
+
+      log_list_hash.each do |storeName, logitems|
+        putLogRequest = AliyunSlsSdk::PutLogsRequest.new(@project, storeName, @topic, nil, logitems, nil, true)
+        retries = 0
         begin
-          client.puts_logs(storeName, log_list, @ssl_verify)
-        rescue Exception => e
-          if retries > 0
-            log.warn "\tCaught in puts logs: #{e}"
-            client.http.shutdown
-            @_sls_con = nil
-            retries -= 1
-            retry
+          client.put_logs(putLogRequest)
+        rescue  => e
+          if e.instance_of?(AliyunSlsSdk::LogException) && e.errorCode == "LogStoreNotExist" && @need_create_logstore
+            createLogStore(storeName)
+            # wait up to 60 seconds to create the logstore
+            if retries < 3
+              retries += 1
+              sleep(10 * retries)
+              retry
+            end
+          else
+            log.warn "\tCaught in puts logs: #{e.message}"
+            if retries < 3
+              client.http.shutdown
+              @_sls_con = nil
+              retries += 1
+              sleep(1 * retries)
+              retry
+            end
+            log.error "Could not puts logs to aliyun sls: #{e.message}"
           end
-          log.error "Could not puts logs to aliyun sls: #{e}"
         end
       end
     end
-
-    private
-
-    def pack_log_item(log_list, log, record)
-      pack_hash_log_item(log, record)
-      log_list.logs << log
-    end
-
-    def pack_hash_log_item(log, hash)
-      if hash
-        hash.each { |k, v|
-          log_item = AliyunSlsSdk::Protobuf::Log::Content.new(:key => k, :value => v || "")
-          log.contents << log_item
-        }
-      end
-    end
   end
 end

docs/output/aliyun_sls.md

@@ -2,12 +2,15 @@
 
 #### Environment variables for image `fluentd`
 
-- FLUENTD_OUTPUT=aliyun_sls  # specify your output plugin name
-- ALIYUNSLS_PROJECT=test-fluentd  # specify your aliyun sls project name
-- ALIYUNSLS_REGION_ENDPOINT=cn-hangzhou.log.aliyuncs.com  # specify your region root endpoint
-- ALIYUNSLS_ACCESS_KEY_ID="your aliyun access key id"
-- ALIYUNSLS_ACCESS_KEY_SECRET="your aliyun access key secret"
-- SSL_VERIFY="true" # use `https` scheme to access aliyun sls service, default is false.
+- FLUENTD_OUTPUT=aliyun_sls  # Required, specify your output plugin name
+- ALIYUNSLS_PROJECT=test-fluentd  # Required, specify your aliyun sls project name
+- ALIYUNSLS_REGION_ENDPOINT=cn-hangzhou.log.aliyuncs.com  # Required, specify your region root endpoint
+- ALIYUNSLS_ACCESS_KEY_ID="your aliyun access key id"     # Required 
+- ALIYUNSLS_ACCESS_KEY_SECRET="your aliyun access key secret"  # Required 
+- SSL_VERIFY="true" # Optional, use `https` scheme to access aliyun sls service, default is false.
+- ALIYUNSLS_NEED_CREATE_LOGSTORE="true" # Optional, when set `true`, logstore will be created if not exist, default is false.
+- ALIYUNSLS_CREATE_LOGSTORE_TTL=2  # Optional, used when ALIYUNSLS_NEED_CREATE_LOGSTORE set `true` and as param when creating log store, set the logging data time to live in days default is 1 day to live
+- ALIYUNSLS_CREATE_LOGSTORE_SHARD_COUNT=2 # Optional, used when ALIYUNSLS_NEED_CREATE_LOGSTORE set `true` and as param when creating log store, set the shard count, default is 2 shard count 
 
 #### Labels for your `app image` whose log will streamed to aliyun sls