Elasticsearch에 저장된 데이터 확인하기

ELK

Elasticsearch에 저장된 데이터 확인하기

31514 2024. 12. 19. 15:24

샘플 로그 파일이 kibana에서 시각화되기까지의 과정은 위의 그림과 같습니다.

logstash가 로그 파일을 읽고 필터링을 거친 다음, elasticsearch에 저장
kibana는 elasticsearch에 저장된 데이터를 바탕으로 시각화

실제로 elasticsearch에 저장된 데이터를 확인하기 위해 kibana의 Dev Tools Console에서 다음 쿼리를 실행했습니다.

GET logstash-*/_search
{
  "query": {
    "match_all": {}
  },
  "size": 10
}

그 결과, 크게 2가지 유형의 데이터가 나온 것을 확인할 수 있었습니다.

<첫 번째 유형>

{
  "_index": "logstash-2024.12.18",
  "_id": "_WPQ3JMBJySkpvXRWzOU",
  "_score": 1,
  "_source": {
    "log_message": "Begin connection : 172.75.1.105:2004",
    "timestamp": "2024-12-18 09:17:32.973",
    "log": {
      "file": {
        "path": "/usr/share/logstash/ingest_data/tsI/20241218.log"
      }
    },
    "method": "BeginConnect",
    "@timestamp": "2024-12-18T00:17:32.973Z",
    "event": {
      "original": "[2024-12-18 09:17:32.973]::[Comm]::[ClientSocketBase.cs]::[BeginConnect]::[153]::Begin connection : 172.75.1.105:2004"
    },
    "host": {
      "name": "1f544ba933f5"
    },
    "log_level": "Comm",
    "@version": "1",
    "source_file": "ClientSocketBase.cs",
    "line": "153"
  }
}

이 데이터는 실제 로그 파일에 있는 데이터입니다.

한 가지 주의할 점은 원래 `@timestamp`는 로그 발생 시각이 아니라, 데이터가 처리된 시간(logstash가 ES에 데이터를 적재한 시간)을 가리키는데, 아래와 같은 코드를 사용하면 `@timestamp`를 로그 발생 시각으로 덮어쓸 수 있습니다.

date {
  match => [ "timestamp", "yyyy-MM-dd HH:mm:ss.SSS" ]
  timezone => "Asia/Seoul"
  target => "@timestamp"
}

그리고 `timestamp`와 `@timestamp`의 값이 다른 이유는 각각 KST, UTC이기 때문입니다.

사실 꼭 `@timestamp`의 값을 덮어쓸 필요는 없고, 서로 다른 장단점이 존재합니다.

덮어쓴 경우: 스토리지 절약, 단순화, 일관성, 메타데이터 손실 등
덮어쓰지 않은 경우: 원본 로그와 메타데이터 분리, 유연한 분석, 스토리지 사용량 증가 등

<두 번째 유형>

{
  "_index": "logstash-2024.12.19",
  "_id": "s2HQ3JMBJySkpvXRRbgu",
  "_score": 1,
  "_source": {
    "tags": [
      "_grokparsefailure"
    ],
    "@timestamp": "2024-12-19T02:47:35.492332337Z",
    "event": {
      "original": "   위치: System.IO.File.InternalAppendAllText(String path, String contents, Encoding encoding)"
    },
    "host": {
      "name": "1f544ba933f5"
    },
    "@version": "1",
    "log": {
      "file": {
          "path": "/usr/share/logstash/ingest_data/hs/20241217.log"
      }
    }
  }
}

`tags` 키의 값을 보면 `_grokparsefailure`라는 값을 볼 수 있는데, 이는 grok 필터가 실패했을 때를 말합니다. 즉, " 위치: System.IO.File.InternalAppendAllText(String path, String contents, Encoding encoding)"데이터가 제가 지정한 `grok` 필터와 맞지 않아서 발생한 문제입니다. 왜 이런 문제가 발생했을까요? 원본 로그 파일을 살펴보니 원인을 찾을 수 있었습니다.

[2024-12-17 09:35:04.592]::[Error]::[TaskQueue.cs]::[TaskProc]::[205]::'C:\IISYS_OHLHS\Log\20241217_kp.csv' 파일은 다른 프로세스에서 사용 중이므로 프로세스에서 액세스할 수 없습니다.
[2024-12-17 09:35:04.598]::[Debug]::[TaskQueue.cs]::[TaskProc]::[205]::   위치: System.IO.__Error.WinIOError(Int32 errorCode, String maybeFullPath)
   위치: System.IO.FileStream.Init(String path, FileMode mode, FileAccess access, Int32 rights, Boolean useRights, FileShare share, Int32 bufferSize, FileOptions options, SECURITY_ATTRIBUTES secAttrs, String msgPath, Boolean bFromProxy, Boolean useLongPath, Boolean checkHost)
   위치: System.IO.FileStream..ctor(String path, FileMode mode, FileAccess access, FileShare share, Int32 bufferSize, FileOptions options, String msgPath, Boolean bFromProxy, Boolean useLongPath, Boolean checkHost)
   위치: System.IO.StreamWriter.CreateFile(String path, Boolean append, Boolean checkHost)

이를 해결하기 위해 logstash.conf 파일의 `input` 섹션 부분에 `multiline` 옵션과 정규표현식을 사용하여 여러 줄로 구성되어 있는 로그 데이터도 인식할 수 있도록 하였습니다.

input {
  file {
    mode => "read"
    path => "/usr/share/logstash/ingest_data/**/*.log"
    exit_after_read => true
    file_completed_action => "log"
    file_completed_log_path => "/usr/share/logstash/ingest_data/logstash_completed.log"
    start_position => "beginning"
    sincedb_path => "/dev/null"
    codec => multiline {
      pattern => "^\[\d{4}-\d{2}-\d{2}.*$"
      negate => true
      what => "previous"
    }
  }
}

그 결과 Elasticsearch에 저장된 데이터의 예시는 다음과 같습니다.

{
  "_index": "logstash-2024.12.17",
  "_id": "4suB3ZMBk44LdPIjwEKW",
  "_score": 1,
  "_ignored": [
    "log_message.keyword",
    "event.original.keyword"
  ],
  "_source": {
    "tags": [
      "multiline"
    ],
    "event": {
      "original": """[2024-12-17 12:21:35.714]::[Debug]::[TaskQueue.cs]::[TaskProc]::[205]::   위치: System.IO.__Error.WinIOError(Int32 errorCode, String maybeFullPath)
위치: System.IO.FileStream.Init(String path, FileMode mode, FileAccess access, Int32 rights, Boolean useRights, FileShare share, Int32 bufferSize, FileOptions options, SECURITY_ATTRIBUTES secAttrs, String msgPath, Boolean bFromProxy, Boolean useLongPath, Boolean checkHost)
위치: System.IO.FileStream..ctor(String path, FileMode mode, FileAccess access, FileShare share, Int32 bufferSize, FileOptions options, String msgPath, Boolean bFromProxy, Boolean useLongPath, Boolean checkHost)
위치: System.IO.StreamWriter.CreateFile(String path, Boolean append, Boolean checkHost)
위치: System.IO.StreamWriter..ctor(String path, Boolean append, Encoding encoding, Int32 bufferSize, Boolean checkHost)
위치: System.IO.StreamWriter..ctor(String path, Boolean append, Encoding encoding)
위치: System.IO.File.InternalAppendAllText(String path, String contents, Encoding encoding)
위치: IISYS.Inspection.ServiceProvider.OHLHS.OHLHSSystemService.OnInspectionResultsCallback(IEnumerable`1 inspectResults) 파일 C:\WorkSpace\IISYS.inspector_OHLHS\Application\IISYS.Inspector\IISYS.Inspection\ServiceProvider\OHLHS\OHLHSSystemService.cs:줄 116
위치: IISYS.Inspection.Tasks.ProfilerQueueTaskBase.MultiObjectProc(List`1 dataList) 파일 C:\WorkSpace\IISYS.inspector_OHLHS\Application\IISYS.Inspector\IISYS.Inspection\Tasks\ProfilerQueueTaskBase.cs:줄 122
위치: Utility.Tasks.TaskQueue`1.TaskProc() 파일 C:\WorkSpace\IISYS.inspector_OHLHS\Common\Utility\Tasks\TaskQueue.cs:줄 188
"""
    },
    "@version": "1",
    "log": {
      "file": {
        "path": "/usr/share/logstash/ingest_data/hs/20241217.log"
      }
    },
    "line": "205",
    "method": "TaskProc",
    "log_message": """   위치: System.IO.__Error.WinIOError(Int32 errorCode, String maybeFullPath)
위치: System.IO.FileStream.Init(String path, FileMode mode, FileAccess access, Int32 rights, Boolean useRights, FileShare share, Int32 bufferSize, FileOptions options, SECURITY_ATTRIBUTES secAttrs, String msgPath, Boolean bFromProxy, Boolean useLongPath, Boolean checkHost)
위치: System.IO.FileStream..ctor(String path, FileMode mode, FileAccess access, FileShare share, Int32 bufferSize, FileOptions options, String msgPath, Boolean bFromProxy, Boolean useLongPath, Boolean checkHost)
위치: System.IO.StreamWriter.CreateFile(String path, Boolean append, Boolean checkHost)
위치: System.IO.StreamWriter..ctor(String path, Boolean append, Encoding encoding, Int32 bufferSize, Boolean checkHost)
위치: System.IO.StreamWriter..ctor(String path, Boolean append, Encoding encoding)
위치: System.IO.File.InternalAppendAllText(String path, String contents, Encoding encoding)
위치: IISYS.Inspection.ServiceProvider.OHLHS.OHLHSSystemService.OnInspectionResultsCallback(IEnumerable`1 inspectResults) 파일 C:\WorkSpace\IISYS.inspector_OHLHS\Application\IISYS.Inspector\IISYS.Inspection\ServiceProvider\OHLHS\OHLHSSystemService.cs:줄 116
위치: IISYS.Inspection.Tasks.ProfilerQueueTaskBase.MultiObjectProc(List`1 dataList) 파일 C:\WorkSpace\IISYS.inspector_OHLHS\Application\IISYS.Inspector\IISYS.Inspection\Tasks\ProfilerQueueTaskBase.cs:줄 122
위치: Utility.Tasks.TaskQueue`1.TaskProc() 파일 C:\WorkSpace\IISYS.inspector_OHLHS\Common\Utility\Tasks\TaskQueue.cs:줄 188
""",
    "timestamp": "2024-12-17 12:21:35.714",
    "@timestamp": "2024-12-17T03:21:35.714Z",
    "host": {
      "name": "0998624e8f3b"
    },
    "log_level": "Debug",
    "source_file": "TaskQueue.cs"
  }
}