Skip to content

Automatically create data streams #44

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Apr 4, 2022
Merged

Automatically create data streams #44

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions lib/fluent/plugin/opensearch_index_template.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,10 +62,10 @@ def host_unreachable_exceptions
client.transport.transport.host_unreachable_exceptions
end

def retry_operate(max_retries, fail_on_retry_exceed = true, catch_trasport_exceptions = true)
def retry_operate(max_retries, fail_on_retry_exceed = true, catch_transport_exceptions = true)
return unless block_given?
retries = 0
transport_errors = OpenSearch::Transport::Transport::Errors.constants.map{ |c| OpenSearch::Transport::Transport::Errors.const_get c } if catch_trasport_exceptions
transport_errors = OpenSearch::Transport::Transport::Errors.constants.map{ |c| OpenSearch::Transport::Transport::Errors.const_get c } if catch_transport_exceptions
begin
yield
rescue *host_unreachable_exceptions, *transport_errors, Timeout::Error => e
Expand Down
15 changes: 0 additions & 15 deletions lib/fluent/plugin/out_opensearch_data_stream.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,6 @@ def configure(conf)
@fail_on_putting_template_retry_exceed,
@catch_transport_exception_on_retry) do
create_index_template(@data_stream_name, @data_stream_template_name)
create_data_stream(@data_stream_name)
end
rescue => e
raise Fluent::ConfigError, "Failed to create data stream: <#{@data_stream_name}> #{e.message}"
Expand Down Expand Up @@ -105,19 +104,6 @@ def data_stream_exist?(datastream_name, host = nil)
end
end

def create_data_stream(datastream_name, host = nil)
return if data_stream_exist?(datastream_name, host)
params = {
name: datastream_name
}
retry_operate(@max_retry_putting_template,
@fail_on_putting_template_retry_exceed,
@catch_transport_exception_on_retry) do
# TODO: Use X-Pack equivalent performing DataStream operation method on the following line
client(host).perform_request('PUT', "/_data_stream/#{datastream_name}", {}, params)
end
end

def template_exists?(name, host = nil)
if @use_legacy_template
client(host).indices.get_template(:name => name)
Expand Down Expand Up @@ -176,7 +162,6 @@ def write(chunk)
unless @data_stream_names.include?(data_stream_name)
begin
create_index_template(data_stream_name, data_stream_template_name, host)
create_data_stream(data_stream_name, host)
@data_stream_names << data_stream_name
rescue => e
raise Fluent::ConfigError, "Failed to create data stream: <#{data_stream_name}> #{e.message}"
Expand Down