logstash-input-oss外掛程式 - Elasticsearch

logstash-input-oss外掛程式基於阿里雲輕量訊息佇列（簡稱SMQ，原MNS），實現了當關聯的Object Storage Service服務OSS（Object Storage Service）檔案變化時，觸發SMQ通知阿里雲Logstash從OSS檔案系統中擷取最新的資料。您可以在OSS的事件通知區域，配置當檔案發生變化時，自動發送訊息給SMQ。

說明

logstash-input-oss是一款開源外掛程式，詳情請參見logstash-input-oss。

注意事項

當logstash-input-oss外掛程式接收到SMQ通知訊息後，會自動提取通知訊息中的oss object（PutObject，AppendObject等）進行全量同步。
如果OSS儲存的是.gz或.gzip結尾的文字檔，阿里雲Logstash會以.gzip的檔案格式對其進行處理，其他格式的檔案以文字檔進行處理。
檔案是以文字檔的方式讀取的，如果您的檔案是不可解析的格式（例如.jar、.bin等格式），有可能讀取出來是亂碼。

前提條件

您已完成以下操作：

安裝logstash-input-oss外掛程式。
具體操作，請參見安裝或卸載外掛程式。
開通阿里雲OSS服務和阿里雲SMQ服務，且兩者在相同地區。
具體操作，請參見開通阿里雲OSS服務和開通輕量訊息佇列並授權。
在OSS中配置事件通知。
具體操作，請參見設定事件通知規則。

使用logstash-input-oss外掛程式

參見通過設定檔管理管道建立管道任務，在建立管道任務時，需要按照以下說明配置管道參數。配置完成後進行儲存與部署，即可觸發阿里雲Logstash從OSS中擷取資料。

以從OSS中擷取資料，然後寫入到Elasticsearch為例，配置樣本如下。

input {
 oss {
  endpoint => "oss-cn-hangzhou-internal.aliyuncs.com"
  bucket => "zl-ossou****"
  access_key_id => "******"
  access_key_secret => "*********"
  prefix => "file-sample-prefix"
  mns_settings => {
   endpoint => "******.mns.cn-hangzhou-internal.aliyuncs.com"
   queue => "aliyun-es-sample-mns"
  }
  codec => json {
   charset => "UTF-8"
  }
 }
}

output {
  elasticsearch {
    hosts => ["http://es-cn-***.elasticsearch.aliyuncs.com:9200"]
    index => "aliyun-es-sample"
    user => "elastic"
    password => "changeme"
  }
}

重要

Endpoint不能以HTTP為首碼，並且需要internal網域名稱，否則會報錯。

參數說明

logstash-input-oss外掛程式支援的參數如下。

參數	類型	是否必選	說明
endpoint	string	是	OSS對外服務的訪問網域名稱，擷取方式請參見訪問網域名稱和資料中心。
bucket	string	是	OSS的Bucket名稱。
access_key_id	string	是	阿里雲帳號的AccessKey ID。
access_key_secret	string	是	阿里雲帳號的Access Key Secret。
prefix	string	否	如果指定了該參數，則Bucket中目錄或檔案名稱的首碼必須與之匹配（不是Regex）。通過配置該參數，您可以讀取指定Bucket下的某一個或者幾個目錄。
additional_oss_settings	hash	否	附加的OSS用戶端配置。可選值 secure_connection_enabled：是否啟用安全連線。 max_connections_to_oss：OSS的最大串連數。
delete	boolean	否	是否從原始Bucket中刪除已處理的檔案： true：是 false（預設）：否
backup_to_bucket	string	否	用來備份已處理過的檔案的OSS Bucket名稱。
backup_to_dir	string	否	用來備份已經處理過的檔案的本地目錄路徑。
backup_add_prefix	string	否	檔案處理後，為key（OSS中包含檔案名稱的完整路徑）附加一個首碼。當您將資料備份到另一個（或同一個）Bucket時，這個參數將有效地讓您選擇一個新的檔案夾來放置檔案。
include_object_properties	boolean	否	是否在[@metadata][oss]中包含OSS對象的屬性（last_modified，content_type，metadata）： true：是 false：否如果不設定此參數，[@metadata][oss][key]將始終存在。
exclude_pattern	string	否	要從Bucket中排除的key的rubyRegex。
mns_settings	hash	是	輕量訊息佇列配置，可選值： endpoint：SMQ連接埠連結。不能以HTTP為首碼，並且需要internal網域名稱，否則會報錯。 queue：隊列名。 poll_interval_seconds：當隊列中沒有訊息時，針對該隊列的ReceiveMessage請求最長的等待時間，預設為10秒。 wait_seconds：本次ReceiveMessage請求最長的Polling等待時間，單位為秒。 ReceiveMessage的詳細資料請參見ReceiveMessage。

常見問題

Q：為什麼基於SMQ設計logstash-input-oss外掛程式？
A：因為OSS檔案的變更需要有一種機制通知用戶端，而目前OSS檔案事件變更可以無縫的寫入到SMQ中。
Q：為什麼不使用OSS的ListObjects API擷取變更的檔案？
A：OSS在記錄未處理的檔案及已經處理的檔案時會增加本機存放區，當本機存放區較大時，ListObjects API效能會降低。目前其他檔案儲存體系統，如S3開源社區，也將ListObjects API改為了訊息通知機制。
Q：用戶端已經觸發了阿里雲Logstash從OSS中擷取資料，但是OSS還在寫入檔案， logstash-input-oss會怎麼處理，會不會導致部分資料丟失？
A：OSS還在寫入檔案時，觸發了阿里雲Logstash從OSS中擷取資料，此時已經寫入OSS中的資料會記錄到SMQ的隊列中，並通過Logstash管道傳輸到Elasticsearch中。還未寫入到OSS的這部分資料會繼續寫入到OSS中，等待下次事件觸發後再進行寫入。