우리는 ELK와 Exchange와 친구입니다. 2 부

Exchange와 ELK 친구를 사귀는 방법에 대한 이야기를 계속합니다. 여기에). 이 조합을 사용하면 매우 많은 수의 로그를 주저 없이 처리할 수 있다는 점을 상기시켜 드리겠습니다. 이번에는 Exchange가 Logstash 및 Kibana 구성 요소와 작동하도록 하는 방법에 대해 이야기하겠습니다.

ELK 스택의 Logstash는 로그를 지능적으로 처리하고 문서 형식으로 Elastic에 배치할 수 있도록 준비하는 데 사용되며, 이를 기반으로 Kibana에서 다양한 시각화를 구축하는 것이 편리합니다.

설치

두 단계로 구성됩니다.

• OpenJDK 패키지 설치 및 구성
• Logstash 패키지 설치 및 구성

OpenJDK 패키지 설치 및 구성

OpenJDK 패키지를 다운로드하여 특정 디렉터리에 압축을 풀어야 합니다. 그런 다음 이 디렉터리의 경로를 Windows 운영 체제의 $env:Path 및 $env:JAVA_HOME 변수에 입력해야 합니다.

Java 버전을 확인해 보겠습니다.

PS C:> java -version
openjdk version "13.0.1" 2019-10-15
OpenJDK Runtime Environment (build 13.0.1+9)
OpenJDK 64-Bit Server VM (build 13.0.1+9, mixed mode, sharing)

Logstash 패키지 설치 및 구성

Logstash 배포판이 포함된 아카이브 파일 다운로드 따라서. 아카이브는 디스크 루트에 압축을 풀어야 합니다. 폴더에 압축 풀기 C:Program Files 그럴 가치가 없습니다. Logstash는 정상적으로 시작을 거부합니다. 그런 다음 파일에 들어가야합니다 jvm.options Java 프로세스에 RAM 할당을 담당하는 수정 사항입니다. 서버 RAM의 절반을 지정하는 것이 좋습니다. 보드에 16GB RAM이 있는 경우 기본 키는 다음과 같습니다.

-Xms1g
-Xmx1g

다음으로 교체해야 합니다:

-Xms8g
-Xmx8g

또한 해당 행을 주석 처리하는 것이 좋습니다. -XX:+UseConcMarkSweepGC. 더 알아보기 여기에. 다음 단계는 logstash.conf 파일에 기본 구성을 생성하는 것입니다.

input {
 stdin{}
}
 
filter {
}
 
output {
 stdout {
 codec => "rubydebug"
 }
}

이 구성을 사용하면 Logstash는 콘솔에서 데이터를 읽고 이를 빈 필터를 통해 전달한 후 다시 콘솔에 출력합니다. 이 구성을 사용하면 Logstash의 기능이 테스트됩니다. 이를 위해 대화형 모드에서 실행해 보겠습니다.

PS C:...bin> .logstash.bat -f .logstash.conf
...
[2019-12-19T11:15:27,769][INFO ][logstash.javapipeline    ][main] Pipeline started {"pipeline.id"=>"main"}
The stdin plugin is now waiting for input:
[2019-12-19T11:15:27,847][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2019-12-19T11:15:28,113][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}

Logstash는 포트 9600에서 성공적으로 시작되었습니다.

마지막 설치 단계: Logstash를 Windows 서비스로 시작합니다. 예를 들어 패키지를 사용하여 이 작업을 수행할 수 있습니다. NSSM:

PS C:...bin> .nssm.exe install logstash
Service "logstash" installed successfully!

결함 허용

원본 서버에서 전송될 때 로그의 안전성은 영구 대기열 메커니즘에 의해 보장됩니다.

원리

로그 처리 중 대기열의 레이아웃은 입력 → 대기열 → 필터 + 출력입니다.

입력 플러그인은 로그 소스로부터 데이터를 수신하고 이를 큐에 쓴 다음 데이터가 소스에 수신되었다는 확인을 보냅니다.

대기열의 메시지는 Logstash에 의해 처리되고 필터와 출력 플러그인을 통해 전달됩니다. 출력에서 로그가 전송되었다는 확인을 받으면 Logstash는 큐에서 처리된 로그를 제거합니다. Logstash가 중지되면 처리되지 않은 모든 메시지와 확인을 받지 못한 메시지는 대기열에 남아 있으며 Logstash는 다음에 시작할 때 해당 메시지를 계속 처리합니다.

조정

파일의 키로 조정 가능 C:Logstashconfiglogstash.yml:

• queue.type: (가능한 값 ​​- persisted и memory (default)).
• path.queue: (기본적으로 C:Logstashqueue에 저장되는 대기열 파일이 있는 폴더 경로).
• queue.page_capacity: (최대 대기열 페이지 크기, 기본값은 64MB).
• queue.drain: (true/false - Logstash를 종료하기 전에 대기열 처리 중지를 활성화/비활성화합니다. 활성화하지 않는 것이 좋습니다. 이는 서버 종료 속도에 직접적인 영향을 미치기 때문입니다.)
• queue.max_events: (큐의 최대 이벤트 수, 기본값은 0(무제한)).
• queue.max_bytes: (최대 대기열 크기(바이트), 기본값 - 1024mb(1gb)).

구성된 경우 queue.max_events и queue.max_bytes을 선택하면 이러한 설정 값에 도달하면 메시지가 대기열에 더 이상 허용되지 않습니다. 지속 대기열에 대해 자세히 알아보기 여기에.

대기열 설정을 담당하는 logstash.yml 부분의 예:

queue.type: persisted
queue.max_bytes: 10gb

조정

Logstash 구성은 일반적으로 수신 로그 처리의 다양한 단계, 즉 수신(입력 섹션), 구문 분석(필터 섹션), Elastic으로 전송(출력 섹션)을 담당하는 세 부분으로 구성됩니다. 아래에서는 각각에 대해 자세히 살펴보겠습니다.

입력

Filebeat 에이전트로부터 원시 로그가 포함된 수신 스트림을 받습니다. 입력 섹션에 표시되는 것은 이 플러그인입니다.

input {
  beats {
    port => 5044
  }
}

이 구성 후에 Logstash는 포트 5044를 수신하기 시작하고 로그를 수신할 때 필터 섹션의 설정에 따라 처리합니다. 필요한 경우 SSL의 filebit에서 로그를 수신하기 위해 채널을 래핑할 수 있습니다. Beats 플러그인 설정에 대해 자세히 알아보세요. 여기에.

필터

처리에 관심이 있는 Exchange에서 생성하는 모든 텍스트 로그는 로그 파일 자체에 설명된 필드가 있는 csv 형식입니다. csv 레코드를 구문 분석하기 위해 Logstash는 세 가지 플러그인을 제공합니다. 해부하다, csv 및 grok. 첫 번째가 가장 быстрый, 그러나 가장 간단한 로그만 구문 분석하는 데 대처합니다.
예를 들어, 다음 레코드를 두 개로 분할합니다(필드 내부에 쉼표가 있기 때문에). 이로 인해 로그가 잘못 구문 분석됩니다.

…,"MDB:GUID1, Mailbox:GUID2, Event:526545791, MessageClass:IPM.Note, CreationTime:2020-05-15T12:01:56.457Z, ClientType:MOMT, SubmissionAssistant:MailboxTransportSubmissionEmailAssistant",…

예를 들어 IIS와 같은 로그를 구문 분석할 때 사용할 수 있습니다. 이 경우 필터 섹션은 다음과 같습니다.

filter {
  if "IIS" in [tags] {
    dissect {
      mapping => {
        "message" => "%{date} %{time} %{s-ip} %{cs-method} %{cs-uri-stem} %{cs-uri-query} %{s-port} %{cs-username} %{c-ip} %{cs(User-Agent)} %{cs(Referer)} %{sc-status} %{sc-substatus} %{sc-win32-status} %{time-taken}"
      }
      remove_field => ["message"]
      add_field => { "application" => "exchange" }
    }
  }
} 

Logstash 구성을 사용하면 다음을 사용할 수 있습니다. 조건문, 따라서 filebeat 태그가 지정된 로그만 dissect 플러그인으로 보낼 수 있습니다. IIS. 플러그인 내에서 필드 값을 이름과 일치시키고 원래 필드를 삭제합니다. message, 로그 항목이 포함되어 있으며, 예를 들어 로그를 수집하는 애플리케이션의 이름을 포함하는 사용자 정의 필드를 추가할 수 있습니다.

로그를 추적하는 경우 csv 플러그인을 사용하는 것이 더 좋으며 복잡한 필드를 올바르게 처리할 수 있습니다.

filter {
  if "Tracking" in [tags] {
    csv {
      columns => ["date-time","client-ip","client-hostname","server-ip","server-hostname","source-context","connector-id","source","event-id","internal-message-id","message-id","network-message-id","recipient-address","recipient-status","total-bytes","recipient-count","related-recipient-address","reference","message-subject","sender-address","return-path","message-info","directionality","tenant-id","original-client-ip","original-server-ip","custom-data","transport-traffic-type","log-id","schema-version"]
      remove_field => ["message", "tenant-id", "schema-version"]
      add_field => { "application" => "exchange" }
    }
}

플러그인 내에서 필드 값을 이름과 일치시키고 원래 필드를 삭제합니다. message (또한 필드 tenant-id и schema-version)에는 로그 항목이 포함되어 있으며, 예를 들어 로그를 수집하는 애플리케이션의 이름이 포함되는 사용자 정의 필드를 추가할 수 있습니다.

필터링 단계를 종료하면 Kibana에서 시각화할 준비가 된 첫 번째 근사치의 문서를 받게 됩니다. 다음 사항이 누락됩니다.

• 숫자 필드는 텍스트로 인식되므로 해당 필드에 대한 작업이 불가능합니다. 즉, 필드 time-taken IIS 로그 및 필드 recipient-count и total-bites 로그 추적.
• 표준 문서 타임스탬프에는 로그가 서버 측에 기록된 시간이 아니라 로그가 처리된 시간이 포함됩니다.
• 분야 recipient-address 편지 수신자를 계산하는 분석을 허용하지 않는 하나의 건설 현장처럼 보일 것입니다.

이제 로그 처리 프로세스에 약간의 마법을 추가할 시간입니다.

숫자 필드 변환

해부 플러그인에는 옵션이 있습니다 convert_datatype, 텍스트 필드를 디지털 형식으로 변환하는 데 사용할 수 있습니다. 예를 들어 다음과 같습니다.

dissect {
  …
  convert_datatype => { "time-taken" => "int" }
  …
}

이 방법은 필드에 확실히 문자열이 포함될 경우에만 적합하다는 점을 기억할 가치가 있습니다. 이 옵션은 필드의 Null 값을 처리하지 않으며 예외를 발생시킵니다.

로그를 추적하려면 유사한 변환 방법을 사용하지 않는 것이 좋습니다. recipient-count и total-bites 비어 있을 수 있습니다. 이 필드를 변환하려면 플러그인을 사용하는 것이 좋습니다 돌연변이:

mutate {
  convert => [ "total-bytes", "integer" ]
  convert => [ "recipient-count", "integer" ]
}

receive_address를 개별 수신자로 분할

이 문제는 mutate 플러그인을 사용하여 해결할 수도 있습니다.

mutate {
  split => ["recipient_address", ";"]
}

타임스탬프 변경

로그 추적의 경우 플러그인을 사용하면 문제가 매우 쉽게 해결됩니다. 데이터, 이는 현장에서 글을 쓰는 데 도움이 될 것입니다. timestamp 필드에서 필요한 형식으로 날짜 및 시간 date-time:

date {
  match => [ "date-time", "ISO8601" ]
  timezone => "Europe/Moscow"
  remove_field => [ "date-time" ]
}

IIS 로그의 경우 필드 데이터를 결합해야 합니다. date и time mutate 플러그인을 사용하여 필요한 시간대를 등록하고 이 타임스탬프를 timestamp 날짜 플러그인을 사용하여:

mutate { 
  add_field => { "data-time" => "%{date} %{time}" }
  remove_field => [ "date", "time" ]
}
date { 
  match => [ "data-time", "YYYY-MM-dd HH:mm:ss" ]
  timezone => "UTC"
  remove_field => [ "data-time" ]
}

산출

출력 섹션은 처리된 로그를 로그 수신자에게 보내는 데 사용됩니다. Elastic으로 직접 전송하는 경우 플러그인을 사용합니다. 탄성 검색, 생성된 문서를 보내기 위한 서버 주소 및 인덱스 이름 템플릿을 지정합니다.

output {
  elasticsearch {
    hosts => ["127.0.0.1:9200", "127.0.0.2:9200"]
    manage_template => false
    index => "Exchange-%{+YYYY.MM.dd}"
  }
}

최종 구성

최종 구성은 다음과 같습니다.

input {
  beats {
    port => 5044
  }
}
 
filter {
  if "IIS" in [tags] {
    dissect {
      mapping => {
        "message" => "%{date} %{time} %{s-ip} %{cs-method} %{cs-uri-stem} %{cs-uri-query} %{s-port} %{cs-username} %{c-ip} %{cs(User-Agent)} %{cs(Referer)} %{sc-status} %{sc-substatus} %{sc-win32-status} %{time-taken}"
      }
      remove_field => ["message"]
      add_field => { "application" => "exchange" }
      convert_datatype => { "time-taken" => "int" }
    }
    mutate { 
      add_field => { "data-time" => "%{date} %{time}" }
      remove_field => [ "date", "time" ]
    }
    date { 
      match => [ "data-time", "YYYY-MM-dd HH:mm:ss" ]
      timezone => "UTC"
      remove_field => [ "data-time" ]
    }
  }
  if "Tracking" in [tags] {
    csv {
      columns => ["date-time","client-ip","client-hostname","server-ip","server-hostname","source-context","connector-id","source","event-id","internal-message-id","message-id","network-message-id","recipient-address","recipient-status","total-bytes","recipient-count","related-recipient-address","reference","message-subject","sender-address","return-path","message-info","directionality","tenant-id","original-client-ip","original-server-ip","custom-data","transport-traffic-type","log-id","schema-version"]
      remove_field => ["message", "tenant-id", "schema-version"]
      add_field => { "application" => "exchange" }
    }
    mutate {
      convert => [ "total-bytes", "integer" ]
      convert => [ "recipient-count", "integer" ]
      split => ["recipient_address", ";"]
    }
    date {
      match => [ "date-time", "ISO8601" ]
      timezone => "Europe/Moscow"
      remove_field => [ "date-time" ]
    }
  }
}
 
output {
  elasticsearch {
    hosts => ["127.0.0.1:9200", "127.0.0.2:9200"]
    manage_template => false
    index => "Exchange-%{+YYYY.MM.dd}"
  }
}

유용한 링크 :

• Windows에 OpenJDK 11을 설치하는 방법은 무엇입니까?
• 로그스태시 다운로드
• Elastic에서는 더 이상 사용되지 않는 옵션 UseConcMarkSweepGC #36828을 사용합니다.
• NSSM
• 지속 대기열
• Beats 입력 플러그인
• Logstash 야, 내 전기톱은 어디 있지? 내 로그를 분석해야 해요
• 해부 필터 플러그인
• 조건부
• 필터 플러그인 변형
• 날짜 필터 플러그인
• Elasticsearch 출력 플러그인

출처 : habr.com