input {
 # beat에서 데이터를 받을 logstash의 port지정
  beats {
    port => 5044 
  }
}

filter {
  mutate {
    # 실제 데이터는 "message" 필드로 오기 때문에 csv형태의 내용을 분할하여 새로운 이름으로 필드를 추가 
    split => [ "message",  "," ] 
    add_field => {
      "date" => "%{[message][0]}"
      "regieon" => "%{[message][1]}"
      "confirmed" => "%{[message][2]}"
      "death" => "%{[message][3]}"
      "released" => "%{[message][4]}"
    }

    # 기본으로 전송되는 데이터 분석에 불필요한 필드 제거
    remove_field => ["ecs", "host", "@version", "agent", "log", "tags",  "input", "message"]
  }

  # "date" 필드를 이용하여 Elasticsearch에서 인식할 수 있는 date 타입의 형태로 필드를 추가
  # index pattern 형식으로 키바나에서 다양한 시각화 등에 사용을 위한 기본 설정: 시계열 데이터(시간을 기준으로 데이터 분석)
  # timestamp 포멧으로 형성하는 새로운 field 구성
  # convert date: 가변적인 사용자 정의 이름. 단, 날짜시분초를 표현 가능한 field값으로 새롭게 생성
  # data field 값으로 가공해서 파생
  date {
    match => [ "date", "yyyyMMdd"]
    timezone => "Asia/Seoul"
    locale => "ko"
    target => "convert_date"
  }

  # Kibana에서 데이터 분석시 필요하기 때문에 숫자 타입으로 변경
  mutate {
    convert => {
      "confirmed" => "integer"
      "death" => "integer"
      "released" => "integer"
    }
    remove_field => [ "@timestamp" ]
  }
}

output {

  # 콘솔창에 어떤 데이터들로 필터링 되었는지 확인
  stdout {
          codec => rubydebug
  }

  # 위에서 설치한 Elasticsearch 로 "covid-19" 라는 이름으로 인덱싱 
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "covid-19"
	
  }
}

