Logstash - Filter & Codec

이번에 알아볼 녀석

//날짜 포맷 변경
filter {
  date {
    match => [ "logdate", "MMM dd yyyy HH:mm:ss" ]
  }
}

//drop 필터. 버린다. 
filter {
  if [loglevel] == "debug" {
    drop { }
  }
}

// source 에 속한 녀석들과 hash를 합쳐서 genrated_id 라는 녀석을 만든다. 
filter {
  fingerprint {
    source => ["IP", "@timestamp", "message"]
    method => "SHA1"
    key => "0123"
    target => "[@metadata][generated_id]"
  }
}

// 필드 값 CRUD.
filter {
  mutate {
    rename => { "HOSTORIP" => "client_ip" }
  }
}
// Strip whitespace from field. NOTE: this only works on leading and trailing whitespace.
filter {
  mutate {
    strip => ["field1", "field2"]
  }
}

// 90 % 이벤트를 버린다. 
filter {
  ruby {
    code => "event.cancel if rand <= 0.90"
  }
}

---

input {
  kafka {
    codec => {
      avro => {
        schema_uri => "/tmp/schema.avsc"
      }
    }
  }
}


filter {
  csv {
    separator => ","
    columns => [ "Transaction Number", "Date", "Description", "Amount Debit", "Amount Credit", "Balance" ]
  }
}


input {
  tcp {
    codec => fluent
    port => 4000
  }
}


input {
  file {
    path => "/path/to/myfile.json"
    codec =>"json"
}


input
  kafka {
    zk_connect => "127.0.0.1"
    topic_id => "your_topic_goes_here"
    codec => protobuf {
      class_name => "Animal::Unicorn"
      include_path => ['/path/to/protobuf/definitions/UnicornProtobuf.pb.rb']
    }
  }
}


filter {
  xml {
    source => "message"
  }
}