14873730023

Committed 07 May 2025 02:22AM UTC coverage: 91.339% (-0.006%) from 91.345%

Build # 14873730023

Build Type

Pull #156

github

Committed by

web-flow

Commit Message

Merge 1576e49ae into 4b6ff1a28

Pull Request Pull Request #156: Fix memory usage

Run Details

1160 of 1270 relevant lines covered (91.34%)

155.26 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.41

/lib/fluent/plugin/opensearch_index_template.rb

# SPDX-License-Identifier: Apache-2.0
#
# The fluent-plugin-opensearch Contributors require contributions made to
# this file be licensed under the Apache-2.0 license or a
# compatible open source license.
#
# Modifications Copyright fluent-plugin-opensearch Contributors. See
# GitHub history for details.
#
# Licensed to Uken Inc. under one or more contributor
# license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright
# ownership. Uken Inc. licenses this file to you 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.

require 'fluent/error'
require_relative './opensearch_error'

module Fluent::OpenSearchIndexTemplate
  def get_template(template_file)
    if !File.exist?(template_file)
      raise "If you specify a template_name you must specify a valid template file (checked '#{template_file}')!"
    end
    file_contents = IO.read(template_file).gsub(/\n/,'')
    JSON.parse(file_contents)
  end

  def get_custom_template(template_file, customize_template)
    if !File.exist?(template_file)
      raise "If you specify a template_name you must specify a valid template file (checked '#{template_file}')!"
    end
    file_contents = IO.read(template_file).gsub(/\n/,'')
    customize_template.each do |key, value|
      file_contents = file_contents.gsub(key,value.downcase)
    end
    JSON.parse(file_contents)
  end

  def template_exists?(name, host = nil)
    if @use_legacy_template
      client(host).indices.get_template(:name => name)
    else
      client(host).indices.get_index_template(:name => name)
    end
    return true
  rescue OpenSearch::Transport::Transport::Errors::NotFound
    return false
  end

  def host_unreachable_exceptions
    client.transport.transport.host_unreachable_exceptions
  end

  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_transport_exceptions
    begin
      yield
    rescue *host_unreachable_exceptions, *transport_errors, Timeout::Error => e
      @_es = nil
      @_es_info = nil
      if retries < max_retries
        retries += 1
        wait_seconds = 2**retries
        sleep wait_seconds
        log.warn "Could not communicate to OpenSearch, resetting connection and trying again. #{e.message}"
        log.warn "Remaining retry: #{max_retries - retries}. Retry to communicate after #{wait_seconds} second(s)."
        retry
      end
      message = "Could not communicate to OpenSearch after #{retries} retries. #{e.message}"
      log.warn message
      raise Fluent::Plugin::OpenSearchError::RetryableOperationExhaustedFailure,
            message if fail_on_retry_exceed
    end
  end

  def template_put(name, template, host = nil)
    if @use_legacy_template
      client(host).indices.put_template(:name => name, :body => template)
    else
      client(host).indices.put_index_template(:name => name, :body => template)
    end
  end

  def indexcreation(index_name, host = nil)
    client(host).indices.create(:index => index_name)
  rescue OpenSearch::Transport::Transport::Error => e
    if e.message =~ /"already exists"/ || e.message =~ /resource_already_exists_exception/
      log.debug("Index #{index_name} already exists")
    else
      log.error("Error while index creation - #{index_name}", error: e)
    end
  end

  def template_install(name, template_file, overwrite, host = nil, target_index = nil, index_separator = '-')
    if overwrite
      template_put(name,
                   get_template(template_file), host)

      log.debug("Template '#{name}' overwritten with #{template_file}.")
      return
    end
    if !template_exists?(name, host)
      template_put(name,
                   get_template(template_file), host)
      log.info("Template configured, but no template installed. Installed '#{name}' from #{template_file}.")
    else
      log.debug("Template '#{name}' configured and already installed.")
    end
  end

  def template_custom_install(template_name, template_file, overwrite, customize_template, host, target_index, index_separator)
    custom_template = get_custom_template(template_file, customize_template)

    if overwrite
      template_put(template_name, custom_template, host)
      log.info("Template '#{template_name}' overwritten with #{template_file}.")
    else
      if !template_exists?(template_name, host)
        template_put(template_name, custom_template, host)
        log.info("Template configured, but no template installed. Installed '#{template_name}' from #{template_file}.")
      else
        log.debug("Template '#{template_name}' configured and already installed.")
      end
    end
  end

  def templates_hash_install(templates, overwrite)
    templates.each do |key, value|
      template_install(key, value, overwrite)
    end
  end

  def rollover_alias_payload(rollover_alias)
    {
      'aliases' => {
        rollover_alias => {
          'is_write_index' =>  true
        }
      }
    }
  end
end

1	# SPDX-License-Identifier: Apache-2.0
2	#
3	# The fluent-plugin-opensearch Contributors require contributions made to
4	# this file be licensed under the Apache-2.0 license or a
5	# compatible open source license.
6	#
7	# Modifications Copyright fluent-plugin-opensearch Contributors. See
8	# GitHub history for details.
9	#
10	# Licensed to Uken Inc. under one or more contributor
11	# license agreements. See the NOTICE file distributed with
12	# this work for additional information regarding copyright
13	# ownership. Uken Inc. licenses this file to you under
14	# the Apache License, Version 2.0 (the "License"); you may
15	# not use this file except in compliance with the License.
16	# You may obtain a copy of the License at
17	#
18	# http://www.apache.org/licenses/LICENSE-2.0
19	#
20	# Unless required by applicable law or agreed to in writing,
21	# software distributed under the License is distributed on an
22	# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
23	# KIND, either express or implied. See the License for the
24	# specific language governing permissions and limitations
25	# under the License.
26
27	require 'fluent/error'	1✔
28	require_relative './opensearch_error'	1✔
29
30	module Fluent::OpenSearchIndexTemplate	1✔
31	def get_template(template_file)	1✔
32	if !File.exist?(template_file)	28✔
33	raise "If you specify a template_name you must specify a valid template file (checked '#{template_file}')!"	3✔
34	end
35	file_contents = IO.read(template_file).gsub(/\n/,'')	25✔
36	JSON.parse(file_contents)	25✔
37	end
38
39	def get_custom_template(template_file, customize_template)	1✔
40	if !File.exist?(template_file)	8✔
41	raise "If you specify a template_name you must specify a valid template file (checked '#{template_file}')!"	×
42	end
43	file_contents = IO.read(template_file).gsub(/\n/,'')	8✔
44	customize_template.each do \|key, value\|	8✔
45	file_contents = file_contents.gsub(key,value.downcase)	16✔
46	end
47	JSON.parse(file_contents)	8✔
48	end
49
50	def template_exists?(name, host = nil)	1✔
51	if @use_legacy_template	86✔
52	client(host).indices.get_template(:name => name)	23✔
53	else
54	client(host).indices.get_index_template(:name => name)	63✔
55	end
56	return true	4✔
57	rescue OpenSearch::Transport::Transport::Errors::NotFound
58	return false	25✔
59	end
60
61	def host_unreachable_exceptions	1✔
62	client.transport.transport.host_unreachable_exceptions	74✔
63	end
64
65	def retry_operate(max_retries, fail_on_retry_exceed = true, catch_transport_exceptions = true)	1✔
66	return unless block_given?	470✔
67	retries = 0	470✔
68	transport_errors = OpenSearch::Transport::Transport::Errors.constants.map{ \|c\| OpenSearch::Transport::Transport::Errors.const_get c } if catch_transport_exceptions	20,168✔
69	begin
70	yield	491✔
71	rescue host_unreachable_exceptions, transport_errors, Timeout::Error => e	74✔
72	@_es = nil	69✔
73	@_es_info = nil	69✔
74	if retries < max_retries	69✔
75	retries += 1	21✔
76	wait_seconds = 2**retries	21✔
77	sleep wait_seconds	21✔
78	log.warn "Could not communicate to OpenSearch, resetting connection and trying again. #{e.message}"	21✔
79	log.warn "Remaining retry: #{max_retries - retries}. Retry to communicate after #{wait_seconds} second(s)."	21✔
80	retry	21✔
81	end
82	message = "Could not communicate to OpenSearch after #{retries} retries. #{e.message}"	48✔
83	log.warn message	48✔
84	raise Fluent::Plugin::OpenSearchError::RetryableOperationExhaustedFailure,
85	message if fail_on_retry_exceed	48✔
86	end
87	end
88
89	def template_put(name, template, host = nil)	1✔
90	if @use_legacy_template	33✔
91	client(host).indices.put_template(:name => name, :body => template)	16✔
92	else
93	client(host).indices.put_index_template(:name => name, :body => template)	17✔
94	end
95	end
96
97	def indexcreation(index_name, host = nil)	1✔
98	client(host).indices.create(:index => index_name)	×
99	rescue OpenSearch::Transport::Transport::Error => e
100	if e.message =~ /"already exists"/ \|\| e.message =~ /resource_already_exists_exception/	×
101	log.debug("Index #{index_name} already exists")	×
102	else
103	log.error("Error while index creation - #{index_name}", error: e)	×
104	end
105	end
106
107	def template_install(name, template_file, overwrite, host = nil, target_index = nil, index_separator = '-')	1✔
108	if overwrite	93✔
109	template_put(name,	8✔
110	get_template(template_file), host)
111
112	log.debug("Template '#{name}' overwritten with #{template_file}.")	8✔
113	return	8✔
114	end
115	if !template_exists?(name, host)	85✔
116	template_put(name,	20✔
117	get_template(template_file), host)
118	log.info("Template configured, but no template installed. Installed '#{name}' from #{template_file}.")	17✔
119	else
120	log.debug("Template '#{name}' configured and already installed.")	4✔
121	end
122	end
123
124	def template_custom_install(template_name, template_file, overwrite, customize_template, host, target_index, index_separator)	1✔
125	custom_template = get_custom_template(template_file, customize_template)	8✔
126
127	if overwrite	8✔
128	template_put(template_name, custom_template, host)	2✔
129	log.info("Template '#{template_name}' overwritten with #{template_file}.")	2✔
130	else
131	if !template_exists?(template_name, host)	6✔
132	template_put(template_name, custom_template, host)	6✔
133	log.info("Template configured, but no template installed. Installed '#{template_name}' from #{template_file}.")	6✔
134	else
135	log.debug("Template '#{template_name}' configured and already installed.")	×
136	end
137	end
138	end
139
140	def templates_hash_install(templates, overwrite)	1✔
141	templates.each do \|key, value\|	8✔
142	template_install(key, value, overwrite)	20✔
143	end
144	end
145
146	def rollover_alias_payload(rollover_alias)	1✔
147	{
148	'aliases' => {	×
149	rollover_alias => {
150	'is_write_index' => true
151	}
152	}
153	}
154	end
155	end

fluent / fluent-plugin-opensearch / 14873730023

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous