Module: Fluent::ElasticsearchIndexTemplate
- Included in:
- Plugin::ElasticsearchOutput
- Defined in:
- lib/fluent/plugin/elasticsearch_index_template.rb
Instance Method Summary collapse
- #create_rollover_alias(target_index, rollover_index, deflector_alias_name, app_name, index_date_pattern, index_separator, enable_ilm, ilm_policy_id, ilm_policy, ilm_policy_overwrite, host) ⇒ Object
- #get_custom_template(template_file, customize_template) ⇒ Object
- #get_template(template_file) ⇒ Object
- #get_template_name(enable_ilm, template_name, deflector_alias_name) ⇒ Object
- #indexcreation(index_name, host = nil) ⇒ Object
- #inject_ilm_settings_to_template(deflector_alias, target_index, ilm_policy_id, template, index_separator) ⇒ Object
- #retry_operate(max_retries, fail_on_retry_exceed = true, catch_trasport_exceptions = true) ⇒ Object
- #rollover_alias_payload(rollover_alias) ⇒ Object
- #template_custom_install(template_name, template_file, overwrite, customize_template, enable_ilm, deflector_alias_name, ilm_policy_id, host, target_index, index_separator) ⇒ Object
- #template_exists?(name, host = nil) ⇒ Boolean
- #template_install(name, template_file, overwrite, enable_ilm = false, deflector_alias_name = nil, ilm_policy_id = nil, host = nil, target_index = nil, index_separator = '-') ⇒ Object
- #template_put(name, template, host = nil) ⇒ Object
- #templates_hash_install(templates, overwrite) ⇒ Object
Instance Method Details
#create_rollover_alias(target_index, rollover_index, deflector_alias_name, app_name, index_date_pattern, index_separator, enable_ilm, ilm_policy_id, ilm_policy, ilm_policy_overwrite, host) ⇒ Object
162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 |
# File 'lib/fluent/plugin/elasticsearch_index_template.rb', line 162 def create_rollover_alias(target_index, rollover_index, deflector_alias_name, app_name, index_date_pattern, index_separator, enable_ilm, ilm_policy_id, ilm_policy, ilm_policy_overwrite, host) # ILM request to create alias. if rollover_index || enable_ilm if !client.indices.exists_alias(:name => deflector_alias_name) if @logstash_format index_name_temp = '<'+target_index+'-000001>' else if index_date_pattern.empty? index_name_temp = '<'+target_index.downcase+index_separator+app_name.downcase+'-000001>' else index_name_temp = '<'+target_index.downcase+index_separator+app_name.downcase+'-{'+index_date_pattern+'}-000001>' end end indexcreation(index_name_temp, host) body = rollover_alias_payload(deflector_alias_name) client.indices.put_alias(:index => index_name_temp, :name => deflector_alias_name, :body => body) log.info("The alias '#{deflector_alias_name}' is created for the index '#{index_name_temp}'") else log.debug("The alias '#{deflector_alias_name}' is already present") end # Create ILM policy if rollover indices exist. if enable_ilm if ilm_policy.empty? setup_ilm(enable_ilm, ilm_policy_id) else setup_ilm(enable_ilm, ilm_policy_id, ilm_policy, ilm_policy_overwrite) end end else log.debug("No index and alias creation action performed because rollover_index or enable_ilm is set to: '#{rollover_index}', '#{enable_ilm}'") end end |
#get_custom_template(template_file, customize_template) ⇒ Object
13 14 15 16 17 18 19 20 21 22 |
# File 'lib/fluent/plugin/elasticsearch_index_template.rb', line 13 def get_custom_template(template_file, customize_template) if !File.exists?(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 |
#get_template(template_file) ⇒ Object
5 6 7 8 9 10 11 |
# File 'lib/fluent/plugin/elasticsearch_index_template.rb', line 5 def get_template(template_file) if !File.exists?(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 |
#get_template_name(enable_ilm, template_name, deflector_alias_name) ⇒ Object
130 131 132 |
# File 'lib/fluent/plugin/elasticsearch_index_template.rb', line 130 def get_template_name(enable_ilm, template_name, deflector_alias_name) enable_ilm ? deflector_alias_name : template_name end |
#indexcreation(index_name, host = nil) ⇒ Object
67 68 69 70 71 72 73 74 75 |
# File 'lib/fluent/plugin/elasticsearch_index_template.rb', line 67 def indexcreation(index_name, host = nil) client(host).indices.create(:index => index_name) rescue Elasticsearch::Transport::Transport::Error => e if e. =~ /"already exists"/ || e. =~ /resource_already_exists_exception/ log.debug("Index #{index_name} already exists") else log.error("Error while index creation - #{index_name}", error: e) end end |
#inject_ilm_settings_to_template(deflector_alias, target_index, ilm_policy_id, template, index_separator) ⇒ Object
134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 |
# File 'lib/fluent/plugin/elasticsearch_index_template.rb', line 134 def inject_ilm_settings_to_template(deflector_alias, target_index, ilm_policy_id, template, index_separator) log.debug("Overwriting index patterns when Index Lifecycle Management is enabled.") template['index_patterns'] = "#{target_index}#{index_separator}*" if @use_legacy_template template.delete('template') if template.include?('template') # Prepare settings Hash if !template.key?('settings') template['settings'] = {} end if template['settings'] && (template['settings']['index.lifecycle.name'] || template['settings']['index.lifecycle.rollover_alias']) log.debug("Overwriting index lifecycle name and rollover alias when Index Lifecycle Management is enabled.") end template['settings'].update({ 'index.lifecycle.name' => ilm_policy_id, 'index.lifecycle.rollover_alias' => deflector_alias}) template['order'] = template['order'] ? template['order'] + target_index.count(index_separator) + 1 : 51 + target_index.count(index_separator) else # Prepare template.settings Hash if !template['template'].key?('settings') template['template']['settings'] = {} end if template['template']['settings'] && (template['template']['settings']['index.lifecycle.name'] || template['template']['settings']['index.lifecycle.rollover_alias']) log.debug("Overwriting index lifecycle name and rollover alias when Index Lifecycle Management is enabled.") end template['template']['settings'].update({ 'index.lifecycle.name' => ilm_policy_id, 'index.lifecycle.rollover_alias' => deflector_alias}) template['priority'] = template['priority'] ? template['priority'] + target_index.count(index_separator) + 1 : 101 + target_index.count(index_separator) end template end |
#retry_operate(max_retries, fail_on_retry_exceed = true, catch_trasport_exceptions = true) ⇒ Object
35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 |
# File 'lib/fluent/plugin/elasticsearch_index_template.rb', line 35 def retry_operate(max_retries, fail_on_retry_exceed = true, catch_trasport_exceptions = true) return unless block_given? retries = 0 transport_errors = Elasticsearch::Transport::Transport::Errors.constants.map{ |c| Elasticsearch::Transport::Transport::Errors.const_get c } if catch_trasport_exceptions begin yield rescue *client.transport.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 Elasticsearch, resetting connection and trying again. #{e.}" log.warn "Remaining retry: #{max_retries - retries}. Retry to communicate after #{wait_seconds} second(s)." retry end = "Could not communicate to Elasticsearch after #{retries} retries. #{e.}" log.warn raise Fluent::Plugin::ElasticsearchError::RetryableOperationExhaustedFailure, if fail_on_retry_exceed end end |
#rollover_alias_payload(rollover_alias) ⇒ Object
202 203 204 205 206 207 208 209 210 |
# File 'lib/fluent/plugin/elasticsearch_index_template.rb', line 202 def rollover_alias_payload(rollover_alias) { 'aliases' => { rollover_alias => { 'is_write_index' => true } } } end |
#template_custom_install(template_name, template_file, overwrite, customize_template, enable_ilm, deflector_alias_name, ilm_policy_id, host, target_index, index_separator) ⇒ Object
105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 |
# File 'lib/fluent/plugin/elasticsearch_index_template.rb', line 105 def template_custom_install(template_name, template_file, overwrite, customize_template, enable_ilm, deflector_alias_name, ilm_policy_id, host, target_index, index_separator) template_custom_name = get_template_name(enable_ilm, template_name, deflector_alias_name) custom_template = if enable_ilm inject_ilm_settings_to_template(deflector_alias_name, target_index, ilm_policy_id, get_custom_template(template_file, customize_template), index_separator) else get_custom_template(template_file, customize_template) end if overwrite template_put(template_custom_name, custom_template, host) log.info("Template '#{template_custom_name}' overwritten with #{template_file}.") else if !template_exists?(template_custom_name, host) template_put(template_custom_name, custom_template, host) log.info("Template configured, but no template installed. Installed '#{template_custom_name}' from #{template_file}.") else log.debug("Template '#{template_custom_name}' configured and already installed.") end end end |
#template_exists?(name, host = nil) ⇒ Boolean
24 25 26 27 28 29 30 31 32 33 |
# File 'lib/fluent/plugin/elasticsearch_index_template.rb', line 24 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 Elasticsearch::Transport::Transport::Errors::NotFound return false end |
#template_install(name, template_file, overwrite, enable_ilm = false, deflector_alias_name = nil, ilm_policy_id = nil, host = nil, target_index = nil, index_separator = '-') ⇒ Object
77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 |
# File 'lib/fluent/plugin/elasticsearch_index_template.rb', line 77 def template_install(name, template_file, overwrite, enable_ilm = false, deflector_alias_name = nil, ilm_policy_id = nil, host = nil, target_index = nil, index_separator = '-') inject_template_name = get_template_name(enable_ilm, name, deflector_alias_name) if overwrite template_put(inject_template_name, enable_ilm ? inject_ilm_settings_to_template(deflector_alias_name, target_index, ilm_policy_id, get_template(template_file), index_separator) : get_template(template_file), host) log.debug("Template '#{inject_template_name}' overwritten with #{template_file}.") return end if !template_exists?(inject_template_name, host) template_put(inject_template_name, enable_ilm ? inject_ilm_settings_to_template(deflector_alias_name, target_index, ilm_policy_id, get_template(template_file), index_separator) : get_template(template_file), host) log.info("Template configured, but no template installed. Installed '#{inject_template_name}' from #{template_file}.") else log.debug("Template '#{inject_template_name}' configured and already installed.") end end |
#template_put(name, template, host = nil) ⇒ Object
59 60 61 62 63 64 65 |
# File 'lib/fluent/plugin/elasticsearch_index_template.rb', line 59 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 |
#templates_hash_install(templates, overwrite) ⇒ Object
196 197 198 199 200 |
# File 'lib/fluent/plugin/elasticsearch_index_template.rb', line 196 def templates_hash_install(templates, overwrite) templates.each do |key, value| template_install(key, value, overwrite) end end |