1 Star 0 Fork 0

yebuseng / fluent-plugin-elasticsearch

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
克隆/下载
贡献代码
同步代码
取消
提示: 由于 Git 不支持空文件夾,创建文件夹后会生成空的 .keep 文件
Loading...
README
Apache-2.0

Fluent::Plugin::Elasticsearch, a plugin for Fluentd

Gem Version Build Status Coverage Status Code Climate Issue Stats Issue Stats

Send your logs to ElasticSearch (and search them with Kibana maybe?)

Note: For Amazon Elasticsearch Service please consider using fluent-plugin-aws-elasticsearch-service

Current maintainers: @cosmo0920

Requirements

fluent-plugin-elasticsearch fluentd ruby
>= 2.0.0 >= v0.14.0 >= 2.1
< 2.0.0 >= v0.12.0 >= 1.9

NOTE: fluent-plugin-elasticsearch v2.0.0 is now RC. We will release stable v2.0.0 soon.

NOTE: For v0.12 version, you should use 1.x.y version. Please send patch into v0.12 branch if you encountered 1.x version's bug.

Installation

$ gem install fluent-plugin-elasticsearch

Usage

In your Fluentd configuration, use @type elasticsearch. Additional configuration is optional, default values would look like this:

<match my.logs>
  @type elasticsearch
  host localhost
  port 9200
  index_name fluentd
  type_name fluentd
</match>

Index templates

This plugin creates ElasticSearch indices by merely writing to them. Consider using Index Templates to gain control of what get indexed and how. See this example for a good starting point.

Configuration

hosts

hosts host1:port1,host2:port2,host3:port3
# or
hosts https://customhost.com:443/path,https://username:password@host-failover.com:443

You can specify multiple ElasticSearch hosts with separator ",".

If you specify multiple hosts, this plugin will load balance updates to ElasticSearch. This is an elasticsearch-ruby feature, the default strategy is round-robin.

And this plugin will escape required URL encoded characters within %{} placeholders.

hosts https://%{j+hn}:%{passw@rd}@host1:443/elastic/,http://host2

user, password, path, scheme, ssl_verify

If you specify this option, host and port options are ignored.

user demo
password secret
path /elastic_search/
scheme https

You can specify user and password for HTTP basic auth. If used in conjunction with a hosts list, then these options will be used by default i.e. if you do not provide any of these options within the hosts listed.

And this plugin will escape required URL encoded characters within %{} placeholders.

user %{demo+}
password %{@secret}

Specify ssl_verify false to skip ssl verification (defaults to true)

logstash_format

logstash_format true # defaults to false

This is meant to make writing data into ElasticSearch indices compatible to what Logstash calls them. By doing this, one could take advantage of Kibana. See logstash_prefix and logstash_dateformat to customize this index name pattern. The index name will be #{logstash_prefix}-#{formated_date}

:warning: Setting this option to true will ignore the index_name setting. The default index name prefix is logstash-.

logstash_prefix

logstash_prefix mylogs # defaults to "logstash"

logstash_prefix_separator

logstash_prefix_separator _ # defaults to "-"

logstash_dateformat

The strftime format to generate index target index name when logstash_format is set to true. By default, the records are inserted into index logstash-YYYY.MM.DD. This option, alongwith logstash_prefix lets us insert into specified index like mylogs-YYYYMM for a monthly index.

logstash_dateformat %Y.%m. # defaults to "%Y.%m.%d"

pipeline

Only in ES >= 5.x is available to use this parameter. This param is to set a pipeline id of your elasticsearch to be added into the request, you can configure ingest node. For more information: [Ingest node]

pipeline pipeline_id

time_key_format

The format of the time stamp field (@timestamp or what you specify with time_key). This parameter only has an effect when logstash_format is true as it only affects the name of the index we write to. Please see Time#strftime for information about the value of this format.

Setting this to a known format can vastly improve your log ingestion speed if all most of your logs are in the same format. If there is an error parsing this format the timestamp will default to the ingestion time. If you are on Ruby 2.0 or later you can get a further performance improvment by installing the "strptime" gem: fluent-gem install strptime.

For example to parse ISO8601 times with sub-second precision:

time_key_format %Y-%m-%dT%H:%M:%S.%N%z

time_precision

Should the record not include a time_key, define the degree of sub-second time precision to preserve from the time portion of the routed event.

For example, should your input plugin not include a time_key in the record but it able to pass a time to the router when emitting the event (AWS CloudWatch events are an example of this), then this setting will allow you to preserve the sub-second time resolution of those events. This is the case for: fluent-plugin-cloudwatch-ingest.

time_key

By default, when inserting records in Logstash format, @timestamp is dynamically created with the time at log ingestion. If you'd like to use a custom time, include an @timestamp with your record.

{"@timestamp": "2014-04-07T000:00:00-00:00"}

You can specify an option time_key (like the option described in tail Input Plugin) to replace @timestamp key.

Suppose you have settings

logstash_format true
time_key vtm

Your input is:

{
  "title": "developer",
  "vtm": "2014-12-19T08:01:03Z"
}

The output will be

{
  "title": "developer",
  "@timestamp": "2014-12-19T08:01:03Z",
  "vtm": "2014-12-19T08:01:03Z"
}

See time_key_exclude_timestamp to avoid adding @timestamp.

time_key_exclude_timestamp

time_key_exclude_timestamp false

By default, setting time_key will copy the value to an additional field @timestamp. When setting time_key_exclude_timestamp true, no additional field will be added.

utc_index

utc_index true

By default, the records inserted into index logstash-YYMMDD with UTC (Coordinated Universal Time). This option allows to use local time if you describe utc_index to false.

target_index_key

Tell this plugin to find the index name to write to in the record under this key in preference to other mechanisms. Key can be specified as path to nested record using dot ('.') as a separator.

If it is present in the record (and the value is non falsey) the value will be used as the index name to write to and then removed from the record before output; if it is not found then it will use logstash_format or index_name settings as configured.

Suppose you have the following settings

target_index_key @target_index
index_name fallback

If your input is:

{
  "title": "developer",
  "@timestamp": "2014-12-19T08:01:03Z",
  "@target_index": "logstash-2014.12.19"
}

The output would be

{
  "title": "developer",
  "@timestamp": "2014-12-19T08:01:03Z",
}

and this record will be written to the specified index (logstash-2014.12.19) rather than fallback.

target_type_key

Similar to target_index_key config, find the type name to write to in the record under this key (or nested record). If key not found in record - fallback to type_name (default "fluentd").

template_name

The name of the template to define. If a template by the name given is already present, it will be left unchanged.

This parameter along with template_file allow the plugin to behave similarly to Logstash (it installs a template at creation time) so that raw records are available. See https://github.com/uken/fluent-plugin-elasticsearch/issues/33.

template_file must also be specified.

template_file

The path to the file containing the template to install.

template_name must also be specified.

templates

Specify index templates in form of hash. Can contain multiple templates.

templates { "templane_name_1": "path_to_template_1_file", "templane_name_2": "path_to_template_2_file"}

If template_file and template_name are set, then this parameter will be ignored.

request_timeout

You can specify HTTP request timeout.

This is useful when ElasticSearch cannot return response for bulk request within the default of 5 seconds.

request_timeout 15s # defaults to 5s

reload_connections

You can tune how the elasticsearch-transport host reloading feature works. By default it will reload the host list from the server every 10,000th request to spread the load. This can be an issue if your ElasticSearch cluster is behind a Reverse Proxy, as Fluentd process may not have direct network access to the ElasticSearch nodes.

reload_connections false # defaults to true

reload_on_failure

Indicates that the elasticsearch-transport will try to reload the nodes addresses if there is a failure while making the request, this can be useful to quickly remove a dead node from the list of addresses.

reload_on_failure true # defaults to false

resurrect_after

You can set in the elasticsearch-transport how often dead connections from the elasticsearch-transport's pool will be resurrected.

resurrect_after 5 # defaults to 60s

include_tag_key, tag_key

include_tag_key true # defaults to false
tag_key tag # defaults to tag

This will add the Fluentd tag in the JSON record. For instance, if you have a config like this:

<match my.logs>
  @type elasticsearch
  include_tag_key true
  tag_key _key
</match>

The record inserted into ElasticSearch would be

{"_key": "my.logs", "name": "Johnny Doeie"}

id_key

id_key request_id # use "request_id" field as a record id in ES

By default, all records inserted into ElasticSearch get a random _id. This option allows to use a field in the record as an identifier.

This following record {"name": "Johnny", "request_id": "87d89af7daffad6"} will trigger the following ElasticSearch command

{ "index" : { "_index": "logstash-2013.01.01", "_type": "fluentd", "_id": "87d89af7daffad6" } }
{ "name": "Johnny", "request_id": "87d89af7daffad6" }

parent_key

parent_key a_parent # use "a_parent" field value to set _parent in elasticsearch command

If your input is

{ "name": "Johnny", "a_parent": "my_parent" }

ElasticSearch command would be

{ "index" : { "_index": "****", "_type": "****", "_id": "****", "_parent": "my_parent" } }
{ "name": "Johnny", "a_parent": "my_parent" }

if parent_key is not configed or the parent_key is absent in input record, nothing will happen.

routing_key

Similar to parent_key config, will add _routing into elasticsearch command if routing_key is set and the field does exist in input event.

remove_keys

parent_key a_parent
routing_key a_routing
remove_keys a_parent, a_routing # a_parent and a_routing fields won't be sent to elasticsearch

remove_keys_on_update

Remove keys on update will not update the configured keys in elasticsearch when a record is being updated. This setting only has any effect if the write operation is update or upsert.

If the write setting is upsert then these keys are only removed if the record is being updated, if the record does not exist (by id) then all of the keys are indexed.

remove_keys_on_update foo,bar

remove_keys_on_update_key

This setting allows remove_keys_on_update to be configured with a key in each record, in much the same way as target_index_key works. The configured key is removed before indexing in elasticsearch. If both remove_keys_on_update and remove_keys_on_update_key is present in the record then the keys in record are used, if the remove_keys_on_update_key is not present then the value of remove_keys_on_update is used as a fallback.

remove_keys_on_update_key keys_to_skip

write_operation

The write_operation can be any of:

Operation Description
index (default) new data is added while existing data (based on its id) is replaced (reindexed).
create adds new data - if the data already exists (based on its id), the op is skipped.
update updates existing data (based on its id). If no data is found, the op is skipped.
upsert known as merge or insert if the data does not exist, updates if the data exists (based on its id).

Please note, id is required in create, update, and upsert scenario. Without id, the message will be dropped.

time_parse_error_tag

With logstash_format true, elasticsearch plugin parses timestamp field for generating index name. If the record has invalid timestamp value, this plugin emits an error event to @ERROR label with time_parse_error_tag configured tag.

Default value is Fluent::ElasticsearchOutput::TimeParser.error for backward compatibility. :: separated tag is not good for tag routing because some plugins assume tag is separated by .. We recommend to set this parameter like time_parse_error_tag es_plugin.output.time.error. We will change default value to . separated tag.

reconnect_on_error

Indicates that the plugin should reset connection on any error (reconnect on next send). By default it will reconnect only on "host unreachable exceptions". We recommended to set this true in the presence of elasticsearch shield.

reconnect_on_error true # defaults to false

Client/host certificate options

Need to verify ElasticSearch's certificate? You can use the following parameter to specify a CA instead of using an environment variable.

ca_file /path/to/your/ca/cert

Does your ElasticSearch cluster want to verify client connections? You can specify the following parameters to use your client certificate, key, and key password for your connection.

client_cert /path/to/your/client/cert
client_key /path/to/your/private/key
client_key_pass password

If you want to configure SSL/TLS version, you can specify ssl_version parameter.

ssl_version TLSv1_2 # or [SSLv23, TLSv1, TLSv1_1]

Proxy Support

Starting with version 0.8.0, this gem uses excon, which supports proxy with environment variables - https://github.com/excon/excon#proxy-support

Buffer options

fluentd-plugin-elasticsearch extends Fluentd's builtin Output plugin and use compat_parameters plugin helper. It adds the following options:

buffer_type memory
flush_interval 60
retry_limit 17
retry_wait 1.0
num_threads 1

The value for option buffer_chunk_limit should not exceed value http.max_content_length in your Elasticsearch setup (by default it is 100mb).

Note: If you use or evaluate Fluentd v0.14, you can use <buffer> directive to specify buffer configuration, too. In more detail, please refer to the buffer configuration options for v0.14

Hash flattening

Elasticsearch will complain if you send object and concrete values to the same field. For example, you might have logs that look this, from different places:

{"people" => 100} {"people" => {"some" => "thing"}}

The second log line will be rejected by the Elasticsearch parser because objects and concrete values can't live in the same field. To combat this, you can enable hash flattening.

flatten_hashes true
flatten_hashes_separator _

This will produce elasticsearch output that looks like this: {"people_some" => "thing"}

Note that the flattener does not deal with arrays at this time.

Not seeing a config you need?

We try to keep the scope of this plugin small and not add too many configuration options. If you think an option would be useful to others, feel free to open an issue or contribute a Pull Request.

Alternatively, consider using fluent-plugin-forest. For example, to configure multiple tags to be sent to different ElasticSearch indices:

<match my.logs.*>
  @type forest
  subtype elasticsearch
  remove_prefix my.logs
  <template>
    logstash_prefix ${tag}
    # ...
  </template>
</match>

And yet another option is described in Dynamic Configuration section.

Note: If you use or evaluate Fluentd v0.14, you can use builtin placeholders. In more detail, please refer to Placeholders section.

Dynamic configuration

If you want configurations to depend on information in messages, you can use elasticsearch_dynamic. This is an experimental variation of the ElasticSearch plugin allows configuration values to be specified in ways such as the below:

<match my.logs.*>
  @type elasticsearch_dynamic
  hosts ${record['host1']}:9200,${record['host2']}:9200
  index_name my_index.${Time.at(time).getutc.strftime(@logstash_dateformat)}
  logstash_prefix ${tag_parts[3]}
  port ${9200+rand(4)}
  index_name ${tag_parts[2]}-${Time.at(time).getutc.strftime(@logstash_dateformat)}
</match>

Please note, this uses Ruby's eval for every message, so there are performance and security implications.

Placeholders

v0.14 placeholders can handle ${tag} for tag, %Y%m%d like strftime format, and custom record keys like as record["mykey"].

Note that custom chunk key is diffrent notations for record_reformer and record_modifier. They uses record["some_key"] to specify placeholders, but this feature uses ${key1}, ${key2} notation. And tag, time, and some arbitrary keys must be included in buffer directive attributes.

They are used as below:

tag

<match my.logs>
  @type elasticsearch
  index_name elastic.${tag} #=> replaced with each event's tag. e.g.) elastic.test.tag
  <buffer tag>
    @type memory
  </buffer>
  # <snip>
</match>

time

<match my.logs>
  @type elasticsearch
  index_name elastic.%Y%m%d #=> e.g.) elastic.20170811
  <buffer tag, time>
    @type memory
    timekey 3600
  </buffer>
  # <snip>
</match>

custom key

records = {key1: "value1", key2: "value2"}
<match my.logs>
  @type elasticsearch
  index_name elastic.${key1}.${key2} # => e.g.) elastic.value1.value2
  <buffer tag, key1, key2>
    @type memory
  </buffer>
  # <snip>
</match>

Multi workers

Since Fluentd v0.14, multi workers feature had been implemented to increase throughput with multiple processes. This feature allows Fluentd processes to use one or more CPUs. This feature will be enabled by the following system configuration:

<system>
  workers N # where N is a natural number (N >= 1).
</system>

Contact

If you have a question, open an Issue.

Contributing

There are usually a few feature requests, tagged Easy, Normal and Hard. Feel free to work on any one of them.

Pull Requests are welcomed.

Pull Request Graph

Running tests

Install dev dependencies:

$ gem install bundler
$ bundle install
$ bundle exec rake test
Apache License Version 2.0, January 2004 http://www.apache.org/licenses/ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION 1. Definitions. "License" shall mean the terms and conditions for use, reproduction, and distribution as defined by Sections 1 through 9 of this document. "Licensor" shall mean the copyright owner or entity authorized by the copyright owner that is granting the License. "Legal Entity" shall mean the union of the acting entity and all other entities that control, are controlled by, or are under common control with that entity. For the purposes of this definition, "control" means (i) the power, direct or indirect, to cause the direction or management of such entity, whether by contract or otherwise, or (ii) ownership of fifty percent (50%) or more of the outstanding shares, or (iii) beneficial ownership of such entity. "You" (or "Your") shall mean an individual or Legal Entity exercising permissions granted by this License. "Source" form shall mean the preferred form for making modifications, including but not limited to software source code, documentation source, and configuration files. "Object" form shall mean any form resulting from mechanical transformation or translation of a Source form, including but not limited to compiled object code, generated documentation, and conversions to other media types. "Work" shall mean the work of authorship, whether in Source or Object form, made available under the License, as indicated by a copyright notice that is included in or attached to the work (an example is provided in the Appendix below). "Derivative Works" shall mean any work, whether in Source or Object form, that is based on (or derived from) the Work and for which the editorial revisions, annotations, elaborations, or other modifications represent, as a whole, an original work of authorship. For the purposes of this License, Derivative Works shall not include works that remain separable from, or merely link (or bind by name) to the interfaces of, the Work and Derivative Works thereof. "Contribution" shall mean any work of authorship, including the original version of the Work and any modifications or additions to that Work or Derivative Works thereof, that is intentionally submitted to Licensor for inclusion in the Work by the copyright owner or by an individual or Legal Entity authorized to submit on behalf of the copyright owner. For the purposes of this definition, "submitted" means any form of electronic, verbal, or written communication sent to the Licensor or its representatives, including but not limited to communication on electronic mailing lists, source code control systems, and issue tracking systems that are managed by, or on behalf of, the Licensor for the purpose of discussing and improving the Work, but excluding communication that is conspicuously marked or otherwise designated in writing by the copyright owner as "Not a Contribution." "Contributor" shall mean Licensor and any individual or Legal Entity on behalf of whom a Contribution has been received by Licensor and subsequently incorporated within the Work. 2. Grant of Copyright License. Subject to the terms and conditions of this License, each Contributor hereby grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free, irrevocable copyright license to reproduce, prepare Derivative Works of, publicly display, publicly perform, sublicense, and distribute the Work and such Derivative Works in Source or Object form. 3. Grant of Patent License. Subject to the terms and conditions of this License, each Contributor hereby grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free, irrevocable (except as stated in this section) patent license to make, have made, use, offer to sell, sell, import, and otherwise transfer the Work, where such license applies only to those patent claims licensable by such Contributor that are necessarily infringed by their Contribution(s) alone or by combination of their Contribution(s) with the Work to which such Contribution(s) was submitted. If You institute patent litigation against any entity (including a cross-claim or counterclaim in a lawsuit) alleging that the Work or a Contribution incorporated within the Work constitutes direct or contributory patent infringement, then any patent licenses granted to You under this License for that Work shall terminate as of the date such litigation is filed. 4. Redistribution. You may reproduce and distribute copies of the Work or Derivative Works thereof in any medium, with or without modifications, and in Source or Object form, provided that You meet the following conditions: (a) You must give any other recipients of the Work or Derivative Works a copy of this License; and (b) You must cause any modified files to carry prominent notices stating that You changed the files; and (c) You must retain, in the Source form of any Derivative Works that You distribute, all copyright, patent, trademark, and attribution notices from the Source form of the Work, excluding those notices that do not pertain to any part of the Derivative Works; and (d) If the Work includes a "NOTICE" text file as part of its distribution, then any Derivative Works that You distribute must include a readable copy of the attribution notices contained within such NOTICE file, excluding those notices that do not pertain to any part of the Derivative Works, in at least one of the following places: within a NOTICE text file distributed as part of the Derivative Works; within the Source form or documentation, if provided along with the Derivative Works; or, within a display generated by the Derivative Works, if and wherever such third-party notices normally appear. The contents of the NOTICE file are for informational purposes only and do not modify the License. You may add Your own attribution notices within Derivative Works that You distribute, alongside or as an addendum to the NOTICE text from the Work, provided that such additional attribution notices cannot be construed as modifying the License. You may add Your own copyright statement to Your modifications and may provide additional or different license terms and conditions for use, reproduction, or distribution of Your modifications, or for any such Derivative Works as a whole, provided Your use, reproduction, and distribution of the Work otherwise complies with the conditions stated in this License. 5. Submission of Contributions. Unless You explicitly state otherwise, any Contribution intentionally submitted for inclusion in the Work by You to the Licensor shall be under the terms and conditions of this License, without any additional terms or conditions. Notwithstanding the above, nothing herein shall supersede or modify the terms of any separate license agreement you may have executed with Licensor regarding such Contributions. 6. Trademarks. This License does not grant permission to use the trade names, trademarks, service marks, or product names of the Licensor, except as required for reasonable and customary use in describing the origin of the Work and reproducing the content of the NOTICE file. 7. Disclaimer of Warranty. Unless required by applicable law or agreed to in writing, Licensor provides the Work (and each Contributor provides its Contributions) on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied, including, without limitation, any warranties or conditions of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A PARTICULAR PURPOSE. You are solely responsible for determining the appropriateness of using or redistributing the Work and assume any risks associated with Your exercise of permissions under this License. 8. Limitation of Liability. In no event and under no legal theory, whether in tort (including negligence), contract, or otherwise, unless required by applicable law (such as deliberate and grossly negligent acts) or agreed to in writing, shall any Contributor be liable to You for damages, including any direct, indirect, special, incidental, or consequential damages of any character arising as a result of this License or out of the use or inability to use the Work (including but not limited to damages for loss of goodwill, work stoppage, computer failure or malfunction, or any and all other commercial damages or losses), even if such Contributor has been advised of the possibility of such damages. 9. Accepting Warranty or Additional Liability. While redistributing the Work or Derivative Works thereof, You may choose to offer, and charge a fee for, acceptance of support, warranty, indemnity, or other liability obligations and/or rights consistent with this License. However, in accepting such obligations, You may act only on Your own behalf and on Your sole responsibility, not on behalf of any other Contributor, and only if You agree to indemnify, defend, and hold each Contributor harmless for any liability incurred by, or claims asserted against, such Contributor by reason of your accepting any such warranty or additional liability. END OF TERMS AND CONDITIONS APPENDIX: How to apply the Apache License to your work. To apply the Apache License to your work, attach the following boilerplate notice, with the fields enclosed by brackets "{}" replaced with your own identifying information. (Don't include the brackets!) The text should be enclosed in the appropriate comment syntax for the file format. We also recommend that a file or class name and description of purpose be included on the same "printed page" as the copyright notice for easier identification within third-party archives. Copyright 2017 Uken Studios, Inc. Licensed under the Apache License, Version 2.0 (the "License"); 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 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

简介

暂无描述 展开 收起
Ruby
Apache-2.0
取消

发行版

暂无发行版

贡献者

全部

近期动态

加载更多
不能加载更多了
Ruby
1
https://gitee.com/yebuseng/fluent-plugin-elasticsearch.git
git@gitee.com:yebuseng/fluent-plugin-elasticsearch.git
yebuseng
fluent-plugin-elasticsearch
fluent-plugin-elasticsearch
master

搜索帮助