Class: Fluent::BaritoBatchK8sOutput
- Inherits:
-
BufferedOutput
- Object
- BufferedOutput
- Fluent::BaritoBatchK8sOutput
- Defined in:
- lib/fluent/plugin/out_barito_batch_k8s.rb
Constant Summary collapse
- PLUGIN_NAME =
'barito_batch_k8s'
Instance Method Summary collapse
- #clean_attribute(record) ⇒ Object
-
#format(tag, time, record) ⇒ Object
Overide from BufferedOutput.
- #merge_log_attribute(record) ⇒ Object
-
#start ⇒ Object
Overide from BufferedOutput.
-
#write(chunk) ⇒ Object
Overide from BufferedOutput.
Instance Method Details
#clean_attribute(record) ⇒ Object
72 73 74 75 76 77 |
# File 'lib/fluent/plugin/out_barito_batch_k8s.rb', line 72 def clean_attribute(record) # Delete kubernetes & docker field record.delete('kubernetes') record.delete('docker') record end |
#format(tag, time, record) ⇒ Object
Overide from BufferedOutput
25 26 27 |
# File 'lib/fluent/plugin/out_barito_batch_k8s.rb', line 25 def format(tag, time, record) [tag, time, record].to_msgpack end |
#merge_log_attribute(record) ⇒ Object
79 80 81 82 83 84 85 86 87 88 89 90 91 |
# File 'lib/fluent/plugin/out_barito_batch_k8s.rb', line 79 def merge_log_attribute(record) = nil begin = JSON.parse(record['log']) rescue end if !.nil? return record.merge() end record end |
#start ⇒ Object
Overide from BufferedOutput
20 21 22 |
# File 'lib/fluent/plugin/out_barito_batch_k8s.rb', line 20 def start super end |
#write(chunk) ⇒ Object
Overide from BufferedOutput
30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 |
# File 'lib/fluent/plugin/out_barito_batch_k8s.rb', line 30 def write(chunk) data = { 'items' => [] } transport = Fluent::Plugin::BaritoTransport.new(@produce_url, log) chunk.msgpack_each do |tag, time, record| # Kubernetes annotations = record['kubernetes'] record = clean_attribute(record) trail = Fluent::Plugin::ClientTrail.new(true) timber = Fluent::Plugin::TimberFactory::create_timber(tag, time, record, trail) new_timber = merge_log_attribute(timber) # Add kubernetes information new_timber['k8s_metadata'] = { 'pod_name' => ['pod_name'], 'namespace_name' => ['namespace_name'], 'container_name' => ['container_name'], 'host' => ['host'], 'cluster_name' => @cluster_name } data['items'] << new_timber end if @application_secret.nil? or @application_secret.empty? return if @application_group_secret.nil? or @application_name.nil? header = { content_type: :json, 'X-App-Group-Secret' => @application_group_secret, 'X-App-Name' => @application_name } else header = {content_type: :json, 'X-App-Secret' => @application_secret} end transport.send_compressed(data, header) end |