Ruby access log filter plugin

accesslog.conf

# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.

input {
  file {
    path => "/root/linct/access_logs-sample.csv"
    sincedb_path => "/dev/null"
    start_position => "beginning"
  }
}

filter {
  csv {
    columns => [ "id", "ip", "year", "month", "day", "user_id", "project_id", "created_at", "updated_at", "stat_type", "enterprise_id"]
    skip_header => "true"
    remove_field => ["host", "message", "@version", "@timestamp", "tags", "path"]
  }
  ruby {
    init => "
           require 'digest'
           require 'json'
           require 'date'
           "
    code => "
           hash = event.to_hash
           sha1_base ={ 'ip' => nil, 'year' => nil, 'month' => nil, 'day' => nil, 'user_id' => nil, 'project_id' => nil, 'stat_type' => nil, 'enterprise_id' => nil }
           sha1_str = sha1_base.merge(hash.slice(*sha1_base.keys).delete_if do |k, v| 
            v == 'NULL' || v == '' || v == nil
           end).values.join('|')
           event.set('id', Digest::SHA1.hexdigest(sha1_str))
           event.set('ips', hash['ip'].split(/,|，/).map(&:strip))
           event.set('created_at', DateTime.parse(hash['created_at']).to_time.to_i)
           event.set('updated_at', DateTime.parse(hash['updated_at']).to_time.to_i)
           hash.each do |k, v|
             event.remove(k) if v == 'NULL' || v == '' || v == nil
           end
           event
           "
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9201"]
    index => "access-logs-history-write"
    user => "elastic"
    password => "elastic123"
    document_id => "%{id}"
  }
}