Class: Fluent::ScribeOutput
- Inherits:
-
BufferedOutput
- Object
- BufferedOutput
- Fluent::ScribeOutput
- Defined in:
- lib/fluent/plugin/out_scribe.rb
Instance Method Summary collapse
- #configure(conf) ⇒ Object
- #format(tag, time, record) ⇒ Object
-
#initialize ⇒ ScribeOutput
constructor
A new instance of ScribeOutput.
- #shutdown ⇒ Object
- #start ⇒ Object
- #write(chunk) ⇒ Object
Constructor Details
#initialize ⇒ ScribeOutput
Returns a new instance of ScribeOutput.
37 38 39 40 41 42 43 44 45 46 47 |
# File 'lib/fluent/plugin/out_scribe.rb', line 37 def initialize require 'thrift' $:.unshift File.join(File.dirname(__FILE__), 'thrift') require 'fb303_types' require 'fb303_constants' require 'facebook_service' require 'scribe_types' require 'scribe_constants' require 'scribe' super end |
Instance Method Details
#configure(conf) ⇒ Object
49 50 51 52 53 54 |
# File 'lib/fluent/plugin/out_scribe.rb', line 49 def configure(conf) # override default buffer_chunk_limit conf['buffer_chunk_limit'] ||= '1m' super end |
#format(tag, time, record) ⇒ Object
69 70 71 72 73 74 75 76 77 |
# File 'lib/fluent/plugin/out_scribe.rb', line 69 def format(tag, time, record) if @remove_prefix and ( (tag[0, @removed_length] == @removed_prefix_string and tag.length > @removed_length) or tag == @remove_prefix) [(tag[@removed_length..-1] || @default_category), record].to_msgpack else [tag, record].to_msgpack end end |
#shutdown ⇒ Object
65 66 67 |
# File 'lib/fluent/plugin/out_scribe.rb', line 65 def shutdown super end |
#start ⇒ Object
56 57 58 59 60 61 62 63 |
# File 'lib/fluent/plugin/out_scribe.rb', line 56 def start super if @remove_prefix @removed_prefix_string = @remove_prefix + '.' @removed_length = @removed_prefix_string.length end end |
#write(chunk) ⇒ Object
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 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 |
# File 'lib/fluent/plugin/out_scribe.rb', line 79 def write(chunk) socket = Thrift::Socket.new @host, @port, @timeout transport = Thrift::FramedTransport.new socket protocol = Thrift::BinaryProtocol.new transport, false, false client = Scribe::Client.new protocol transport.open begin entries = [] chunk.msgpack_each do |arr| tag, record = arr next unless @format_to_json || record.has_key?(@field_ref) = @format_to_json ? record : record[@field_ref] if .kind_of?(Array) or .kind_of?(Hash) begin = .to_json rescue => e if @ignore_invalid_record # This warning can be disabled by 'log_level error' log.warn "got invalid message", message: , error: e, error_class: e.class next else # Keep existence behaviour raise end end end if @add_newline = + "\n" end entry = LogEntry.new entry.category = tag entry. = .force_encoding('ASCII-8BIT') entries << entry end log.debug "Writing #{entries.count} entries to scribe" client.Log(entries) ensure transport.close end end |