Class: LogStash::Codecs::CSV

Inherits:
Base
  • Object
show all
Extended by:
PluginMixins::ValidatorSupport::FieldReferenceValidationAdapter
Includes:
PluginMixins::ECSCompatibilitySupport::TargetCheck, PluginMixins::EventSupport::EventFactoryAdapter
Defined in:
lib/logstash/codecs/csv.rb

Constant Summary collapse

CONVERTERS = 
{
  :integer => lambda do |value|
    CSV::Converters[:integer].call(value)
  end,

  :float => lambda do |value|
    CSV::Converters[:float].call(value)
  end,

  :date => lambda do |value|
    result = CSV::Converters[:date].call(value)
    result.is_a?(Date) ? LogStash::Timestamp.new(result.to_time) : result
  end,

  :date_time => lambda do |value|
    result = CSV::Converters[:date_time].call(value)
    result.is_a?(DateTime) ? LogStash::Timestamp.new(result.to_time) : result
  end,

  :boolean => lambda do |value|
     value = value.strip.downcase
     return false if value == "false"
     return true  if value == "true"
     return value
  end
}

Instance Method Summary collapse

Constructor Details

#initialize(*params) ⇒ CSV

Returns a new instance of CSV.



111
112
113
114
115
116
117
118
 
# File 'lib/logstash/codecs/csv.rb', line 111

def initialize(*params)
  super

  @original_field = ecs_select[disabled: nil, v1: '[event][original]']

  @converter = LogStash::Util::Charset.new(@charset)
  @converter.logger = @logger
end

Instance Method Details

#decode(data) ⇒ Object 



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
161
162
163
164
 
# File 'lib/logstash/codecs/csv.rb', line 136

def decode(data)
  data = @converter.convert(data)
  begin
    values = CSV.parse_line(data, :col_sep => @separator, :quote_char => @quote_char)

    if (@autodetect_column_names && @columns.empty?)
      @columns = values
      @logger.debug? && @logger.debug("Auto detected the following columns", :columns => @columns.inspect)
      return
    end

    decoded = {}
    values.each_with_index do |value, i|
      unless (@skip_empty_columns && (value.nil? || value.empty?))
        unless ignore_field?(i)
          field_name = @columns[i] || "column#{i + 1}"
          decoded[field_name] = transform(field_name, value)
        end
      end
    end

    event = targeted_event_factory.new_event(decoded)
    event.set(@original_field, data.dup.freeze) if @original_field
    yield event
  rescue CSV::MalformedCSVError => e
    @logger.error("CSV parse failure. Falling back to plain-text", :exception => e.class, :message => e.message, :data => data)
    yield event_factory.new_event("message" => data, "tags" => ["_csvparsefailure"])
  end
end

#encode(event) ⇒ Object 



166
167
168
169
170
171
172
173
174
175
176
177
 
# File 'lib/logstash/codecs/csv.rb', line 166

def encode(event)
  if @include_headers
    csv_data = CSV.generate_line(select_keys(event), :col_sep => @separator, :quote_char => @quote_char, :headers => true)
    @on_event.call(event, csv_data)

    # output headers only once per codec lifecycle
    @include_headers = false
  end

  csv_data = CSV.generate_line(select_values(event), :col_sep => @separator, :quote_char => @quote_char)
  @on_event.call(event, csv_data)
end

#register ⇒ Object

Raises:

  • (LogStash::ConfigurationError) 


120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
 
# File 'lib/logstash/codecs/csv.rb', line 120

def register
  # validate conversion types to be the valid ones.
  bad_types = @convert.values.select do |type|
    !CONVERTERS.has_key?(type.to_sym)
  end.uniq
  raise(LogStash::ConfigurationError, "Invalid conversion types: #{bad_types.join(', ')}") unless bad_types.empty?

  # @convert_symbols contains the symbolized types to avoid symbol conversion in the transform method
  @convert_symbols = @convert.each_with_object({}) { |(k, v), result| result[k] = v.to_sym }

  # if the zero byte character is entered in the config, set the value
  @quote_char = "\x00" if @quote_char == "\\x00"

  @logger.debug? && @logger.debug("CSV parsing options", :col_sep => @separator, :quote_char => @quote_char)
end