샘플 로그 파일 업로드하기

로그 파일 구조

docker-compose.yml 파일에서 logstash 부분을 보면 다음과 같습니다.

logstash01:
    depends_on:
      es01:
        condition: service_healthy
      kibana:
        condition: service_healthy
    image: docker.elastic.co/logstash/logstash:${STACK_VERSION}
    labels:
      co.elastic.logs/module: logstash
    user: root
    volumes:
      - certs:/usr/share/logstash/certs
      - logstashdata01:/usr/share/logstash/data
      - "./logstash_ingest_data/:/usr/share/logstash/ingest_data/"
      - "./logstash.conf:/usr/share/logstash/pipeline/logstash.conf:ro"
    environment:
      - xpack.monitoring.enabled=false
      - ELASTIC_USER=elastic
      - ELASTIC_PASSWORD=${ELASTIC_PASSWORD}
      - ELASTIC_HOSTS=https://es01:9200

위 코드에서 volumes 섹션 밑에 - "./logstash_ingest_data/:/usr/share/logstash/ingest_data/" 코드는 호스트 환경의 ./logstash_ingest_data/ 경로에 있는 모든 파일이 logstash 컨테이너 안의 /usr/share/logstash/ingest_data/ 경로에 모두 저장된다는 뜻입니다. 이에 따라 샘플 로그 파일을 모두 호스트 환경의 ./logstash_ingest_data/ 경로 밑에 넣었습니다.

다음으로는 logstash가 해당 로그 파일을 인식할 수 있도록 logstash.conf 파일을 수정했습니다.

input {
  file {
    #https://www.elastic.co/guide/en/logstash/current/plugins-inputs-file.html
    #default is TAIL which assumes more data will come into the file.
    #change to mode => "read" if the file is a complete file.  
    #by default, the file will be removed once reading is complete -- backup your files if you need them.
    # we will be using READ with the completed file action to log to a file.
    mode => "read"
    path => "/usr/share/logstash/ingest_data/**/*.log"
    exit_after_read => true # this tells logstash to exit after reading the file.  This is useful for running logstash as a "job". if you want logstash to continue to run and monitor for files, remove this line.
    file_completed_action => "log" # this tells logstash to log to the file specified in file_completed_log_path once its done reading the input file.
    file_completed_log_path => "/usr/share/logstash/ingest_data/logstash_completed.log"
  }
}

filter {
  if [path] == "/usr/share/logstash/ingest_data/logstash_completed.log" {
    drop {}
  }
  
  grok {
    match => { "message" => "^\[%{TIMESTAMP_ISO8601:timestamp}\]::\[%{WORD:log_level}\]::\[%{DATA:source_file}\]::\[%{DATA:method}\]::\[%{NUMBER:line}\]::%{GREEDYDATA:log_message}$" }
  }
}

output {
  elasticsearch {
    index => "logstash-%{+YYYY.MM.dd}" # ES에 저장되는 인덱스 이름
    hosts=> "${ELASTIC_HOSTS}"
    user=> "${ELASTIC_USER}"
    password=> "${ELASTIC_PASSWORD}"
    cacert=> "certs/ca/ca.crt"
  }
}

위 코드에서 input 섹션의 path 부분을 수정하여 확장자가 log인 모든 파일을 인식하도록 변경했습니다. 하지만 file_completed_log_path 또한 같은 경로에 저장되므로, 이 부분은 filter 섹션에서 필터링하였습니다. 그 이유는 filter 섹션에서 grok을 통해 비정형 상태의 로그 데이터를 정형으로 만드는 과정에서 데이터의 패턴이 달라 오류가 발생하기 때문입니다.

참고로 로그 데이터의 형식은 다음과 같습니다.

[2024-12-17 09:13:51.296]::[Debug]::[CameraQueueTaskBase.cs]::[MultiObjectProc]::[244]::====== Camera Queue 누적 이미지 갯수 ======
[2024-12-17 09:13:51.319]::[Debug]::[CameraQueueTaskBase.cs]::[MultiObjectProc]::[287]::Camera No.0 : 1 EA
[2024-12-17 09:13:51.319]::[Debug]::[CameraQueueTaskBase.cs]::[MultiObjectProc]::[291]::Division Image Time : 23 ms
[2024-12-17 09:13:51.429]::[Error]::[OHLHSCameraQueueTask.cs]::[InspectImage]::[205]::개체 참조가 개체의 인스턴스로 설정되지 않았습니다.

데이터 패턴 매칭이 끝나면 총 136만 개의 데이터가 Elasticsearch에 logstash-2024.12.18의 이름을 가진 인덱스로 저장되고, 이를 변경하기 위해서는 index => "logstash-%{+YYYY.MM.dd}" 부분을 수정하면 됩니다.

 

모든 과정이 끝나고 kibana → Analytics → Discover에 접속하면 다음과 같은 화면을 볼 수 있습니다.