소개

다른 시스템을 배포하는 동안 우리는 많은 수의 다양한 로그를 처리해야 하는 필요성에 직면했습니다. ELK가 도구로 선택되었습니다. 이 기사에서는 이 스택을 설정한 경험에 대해 설명합니다.

모든 기능을 설명하는 목표를 설정하지는 않지만 실질적인 문제를 해결하는 데 집중하고 싶습니다. 이것은 충분히 많은 양의 문서와 기성 이미지로 인해 많은 함정이 있다는 사실 때문입니다. 적어도 우리는 그것들을 발견했습니다.

docker-compose를 통해 스택을 배포했습니다. 또한 잘 작성된 docker-compose.yml을 사용하여 거의 문제 없이 스택을 올릴 수 있었습니다. 승리가 이미 가까워진 것 같았습니다. 이제 우리의 필요에 맞게 약간 비틀면 그게 다입니다.

안타깝게도 애플리케이션에서 로그를 수신하고 처리하도록 시스템을 조정하려는 시도는 바로 성공하지 못했습니다. 따라서 각 구성 요소를 개별적으로 연구한 다음 해당 연결로 돌아갈 가치가 있다고 결정했습니다.

이제 logstash부터 시작하겠습니다.

환경, 배포, 컨테이너에서 Logstash 실행

배포를 위해 docker-compose를 사용하며 여기에 설명된 실험은 MacOS 및 Ubuntu 18.0.4에서 수행되었습니다.

원래 docker-compose.yml에 있던 logstash 이미지는 docker.elastic.co/logstash/logstash:6.3.2입니다.

우리는 그것을 실험에 사용할 것입니다.

logstash를 실행하기 위해 별도의 docker-compose.yml을 작성했습니다. 물론 명령 줄에서 이미지를 시작할 수 있었지만 결국 docker-compose의 모든 것이 시작되는 특정 작업을 해결했습니다.

구성 파일에 대한 간략한 설명

설명에서 다음과 같이 logstash는 하나의 채널에 대해 실행될 수 있으며 이 경우 *.conf 파일을 전송해야 하거나 여러 채널에 대해 실행될 경우 pipelines.yml 파일을 전송해야 합니다. , 각 채널에 대한 .conf 파일을 참조합니다.
우리는 두 번째 길을 택했습니다. 그것은 우리에게 더 다재다능하고 확장 가능한 것처럼 보였습니다. 따라서 우리는 pipelines.yml을 생성하고 각 채널에 대한 .conf 파일을 넣을 파이프라인 디렉터리를 만들었습니다.

컨테이너 내부에는 또 다른 구성 파일인 logstash.yml이 있습니다. 우리는 그것을 만지지 않고 그대로 사용합니다.

따라서 우리의 디렉토리 구조는 다음과 같습니다.

당분간 입력 데이터를 수신하기 위해 포트 5046의 tcp라고 가정하고 출력에 stdout을 사용합니다.

다음은 첫 번째 실행을 위한 간단한 구성입니다. 초기 작업은 시작하는 것이기 때문입니다.

그래서 우리는 이 docker-compose.yml을 가지고 있습니다.

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      	- elk
    ports:
      	- 5046:5046
    volumes:
      	- ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
	- ./config/pipelines:/usr/share/logstash/config/pipelines:ro

여기서 무엇을 볼 수 있습니까?

1. 네트워크와 볼륨은 원래 docker-compose.yml(전체 스택이 실행되는 곳)에서 가져왔으며 여기서 전체 그림에 큰 영향을 미치지 않는다고 생각합니다.
2. docker.elastic.co/logstash/logstash:6.3.2 이미지에서 하나의 서비스(services) logstash를 생성하고 이름을 logstash_one_channel로 지정합니다.
3. 컨테이너 내부의 포트 5046을 동일한 내부 포트로 전달하고 있습니다.
4. 우리는 ./config/pipelines.yml 파이프 구성 파일을 컨테이너 내부의 /usr/share/logstash/config/pipelines.yml 파일에 매핑합니다. 여기서 logstash는 만일을 대비하여 이를 선택하고 읽기 전용으로 만듭니다.
5. 파이프 구성 파일이 있는 ./config/pipelines 디렉터리를 /usr/share/logstash/config/pipelines 디렉터리에 매핑하고 읽기 전용으로 만듭니다.

파이핑.yml 파일

- pipeline.id: HABR
  pipeline.workers: 1
  pipeline.batch.size: 1
  path.config: "./config/pipelines/habr_pipeline.conf"

HABR 식별자와 해당 구성 파일의 경로가 있는 하나의 채널을 설명합니다.

그리고 마지막으로 "./config/pipelines/habr_pipeline.conf" 파일

input {
  tcp {
    port => "5046"
   }
  }
filter {
  mutate {
    add_field => [ "habra_field", "Hello Habr" ]
    }
  }
output {
  stdout {
      
    }
  }

지금은 설명에 들어가지 않고 다음을 실행해 봅니다.

docker-compose up

우리는 무엇을 보나요?

컨테이너가 시작되었습니다. 작업을 확인할 수 있습니다.

echo '13123123123123123123123213123213' | nc localhost 5046

그리고 컨테이너 콘솔에 응답이 표시됩니다.

그러나 동시에 다음과 같은 것도 볼 수 있습니다.

logstash_one_channel | [2019-04-29T11:28:59,790][오류][logstash.licensechecker.licensereader] Unable to retrieve license information from license server {:message=>"Elasticsearch Unreachable: [http://elasticsearch:9200/][Manticore ::해결실패]elasticsearch", ...

logstash_one_channel | [2019-04-29T11:28:59,894][INFO ][logstash.pipeline ] 파이프라인이 {:pipeline_id=>".monitoring-logstash", :thread=>"# »}

logstash_one_channel | [2019-04-29T11:28:59,988][INFO ][logstash.agent ] {:count=>2, :running_pipelines=>[:HABR, :".monitoring-logstash"], :non_running_pipelines=>[를 실행하는 파이프라인 ]}
logstash_one_channel | [2019-04-29T11:29:00,015][ERROR][logstash.inputs.metrics ] X-Pack이 Logstash에는 설치되지만 Elasticsearch에는 설치되지 않습니다. 모니터링 기능을 사용하려면 Elasticsearch에 X-Pack을 설치하세요. 다른 기능을 사용할 수 있습니다.
logstash_one_channel | [2019-04-29T11:29:00,526][INFO ][logstash.agent ] Logstash API 엔드포인트 {:port=>9600} 시작 성공
logstash_one_channel | [2019-04-29T11:29:04,478][INFO ][logstash.outputs.elasticsearch] Elasticsearch 연결이 작동하는지 확인하기 위해 상태 확인 실행 {:healthcheck_url=>http://elasticsearch:9200/, :path=> "/"}
logstash_one_channel | [2019-04-29T11:29:04,487][WARN ][logstash.outputs.elasticsearch] 죽은 ES 인스턴스에 대한 연결을 다시 시도했지만 오류가 발생했습니다. {:URL=>"탄성 검색:9200/", :error_type=>LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=>"Elasticsearch에 연결할 수 없음: [http://elasticsearch:9200/][Manticore::ResolutionFailure] 엘라스틱서치"}
logstash_one_channel | [2019-04-29T11:29:04,704][INFO ][logstash.licensechecker.licensereader] Elasticsearch 연결이 작동하는지 확인하기 위해 상태 확인 실행 {:healthcheck_url=>http://elasticsearch:9200/, :path=> "/"}
logstash_one_channel | [2019-04-29T11:29:04,710][WARN ][logstash.licensechecker.licensereader] 죽은 ES 인스턴스에 대한 연결을 다시 시도했지만 오류가 발생했습니다. {:URL=>"탄성 검색:9200/", :error_type=>LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=>"Elasticsearch에 연결할 수 없음: [http://elasticsearch:9200/][Manticore::ResolutionFailure] 엘라스틱서치"}

그리고 우리 로그는 항상 크롤링됩니다.

여기에서 파이프라인이 성공적으로 시작되었다는 메시지를 녹색으로, 오류 메시지를 빨간색으로, 연결 시도에 대한 메시지를 노란색으로 강조 표시했습니다. 탄성 검색: 9200.
이는 이미지에 포함된 logstash.conf에 elasticsearch의 가용성에 대한 확인이 있기 때문에 발생합니다. 결국 logstash는 Elk 스택의 일부로 작동한다고 가정하고 분리했습니다.

일할 수는 있지만 편리하지는 않습니다.

해결책은 XPACK_MONITORING_ENABLED 환경 변수를 통해 이 검사를 비활성화하는 것입니다.

docker-compose.yml을 변경하고 다시 실행해 보겠습니다.

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro

이제 모든 것이 괜찮습니다. 컨테이너는 실험 준비가 되었습니다.

인접한 콘솔에 다시 입력할 수 있습니다.

echo '13123123123123123123123213123213' | nc localhost 5046

그리고 봐라:

logstash_one_channel | {
logstash_one_channel |         "message" => "13123123123123123123123213123213",
logstash_one_channel |      "@timestamp" => 2019-04-29T11:43:44.582Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |            "host" => "gateway",
logstash_one_channel |            "port" => 49418
logstash_one_channel | }

하나의 채널 내에서 작업

그래서 우리는 시작했습니다. 이제 실제로 시간을 들여 logstash를 직접 구성할 수 있습니다. 지금은 pipelines.yml 파일을 건드리지 말고 하나의 채널로 작업하여 무엇을 얻을 수 있는지 봅시다.

채널 구성 파일 작업의 일반적인 원칙은 여기 공식 매뉴얼에 잘 설명되어 있습니다. 여기에
러시아어로 읽고 싶다면 이것을 사용했습니다. 기사(하지만 쿼리 구문이 오래되었으므로 이를 고려해야 합니다.)

입력 부분부터 순차적으로 가보겠습니다. 우리는 이미 tcp에 대한 작업을 보았습니다. 여기서 또 무엇이 흥미로울 수 있습니까?

하트비트를 사용하여 메시지 테스트

자동 테스트 메시지를 생성할 수 있는 흥미로운 가능성이 있습니다.
이렇게 하려면 입력 섹션에 heartbean 플러그인을 포함해야 합니다.

input {
  heartbeat {
    message => "HeartBeat!"
   }
  } 

전원을 켜면 XNUMX분에 한 번씩 받기 시작합니다.

logstash_one_channel | {
logstash_one_channel |      "@timestamp" => 2019-04-29T13:52:04.567Z,
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "HeartBeat!",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "host" => "a0667e5c57ec"
logstash_one_channel | }

더 자주 수신하려면 간격 매개변수를 추가해야 합니다.
이것이 우리가 10초마다 메시지를 받는 방법입니다.

input {
  heartbeat {
    message => "HeartBeat!"
    interval => 10
   }
  }

파일에서 데이터 가져오기

파일 모드도 살펴보기로 했습니다. 파일에서 제대로 작동하는 경우 적어도 로컬 사용에는 에이전트가 필요하지 않을 수 있습니다.

설명에 따르면 작동 모드는 tail -f와 유사해야 합니다. 개행을 읽거나 선택적으로 전체 파일을 읽습니다.

그래서 우리가 얻고자 하는 것:

1. 하나의 로그 파일에 추가된 행을 수신하려고 합니다.
2. 우리는 여러 로그 파일에 기록된 데이터를 수신하면서 수신된 내용을 분리할 수 있기를 원합니다.
3. 우리는 logstash가 다시 시작될 때 이 데이터를 다시 수신하지 않도록 하고 싶습니다.
4. logstash가 비활성화되어 있고 데이터가 파일에 계속 기록되는지 확인하고 실행하면 이 데이터를 수신하게 됩니다.

실험을 수행하기 위해 docker-compose.yml에 한 줄을 더 추가하여 파일을 저장할 디렉터리를 엽니다.

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro
      - ./logs:/usr/share/logstash/input

그리고 habr_pipeline.conf의 입력 섹션을 변경합니다.

input {
  file {
    path => "/usr/share/logstash/input/*.log"
   }
  }

우리는 시작한다:

docker-compose up

로그 파일을 만들고 쓰려면 다음 명령을 사용합니다.


echo '1' >> logs/number1.log

{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:28:53.876Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log"
logstash_one_channel | }

네, 작동합니다!

동시에 경로 필드가 자동으로 추가된 것을 볼 수 있습니다. 따라서 앞으로는 이를 기준으로 레코드를 필터링할 수 있습니다.

다시 해보자:

echo '2' >> logs/number1.log

{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:28:59.906Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "2",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log"
logstash_one_channel | }

이제 다른 파일로 이동:

 echo '1' >> logs/number2.log

{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:29:26.061Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number2.log"
logstash_one_channel | }

엄청난! 파일이 선택되었고 경로가 올바르게 지정되었으며 모든 것이 정상입니다.

logstash를 중지하고 다시 시작하십시오. 기다리 자. 고요. 저것들. 우리는 이러한 기록을 다시 받지 않습니다.

그리고 이제 가장 대담한 실험입니다.

logstash를 넣고 다음을 실행합니다.

echo '3' >> logs/number2.log
echo '4' >> logs/number1.log

logstash를 다시 실행하고 다음을 확인하십시오.

logstash_one_channel | {
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "3",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number2.log",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:48:50.589Z
logstash_one_channel | }
logstash_one_channel | {
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "4",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:48:50.856Z
logstash_one_channel | }

만세! 모든 것이 픽업되었습니다.

그러나 다음 사항에 대해 경고할 필요가 있습니다. logstash 컨테이너가 제거되면(docker stop logstash_one_channel && docker rm logstash_one_channel) 아무것도 선택되지 않습니다. 읽은 파일의 위치는 컨테이너 내부에 저장되었습니다. 처음부터 시작하면 새 줄만 허용됩니다.

기존 파일 읽기

처음으로 logstash를 실행하고 있지만 이미 로그가 있고 이를 처리하고 싶다고 가정해 보겠습니다.
위에서 사용한 입력 섹션으로 logstash를 실행하면 아무 것도 얻을 수 없습니다. 개행만 logstash에서 처리됩니다.

기존 파일에서 줄을 가져오려면 입력 섹션에 추가 줄을 추가합니다.

input {
  file {
    start_position => "beginning"
    path => "/usr/share/logstash/input/*.log"
   }
  }

또한 뉘앙스가 있습니다. 이것은 logstash가 아직 보지 못한 새 파일에만 영향을 미칩니다. logstash의 시야에 이미 있는 동일한 파일의 경우 이미 크기를 기억하고 있으며 이제 새 레코드만 가져옵니다.

입력 섹션을 공부하면서 이것에 대해 멈추자. 더 많은 옵션이 있지만 지금은 추가 실험에 충분합니다.

라우팅 및 데이터 변환

다음 문제를 해결해 봅시다. 한 채널에서 온 메시지가 있고 일부는 정보용이고 일부는 오류 메시지라고 가정해 봅시다. 태그가 다릅니다. 일부는 INFO이고 다른 일부는 ERROR입니다.

출구에서 분리해야 합니다. 저것들. 한 채널에는 정보 메시지를 작성하고 다른 채널에는 오류 메시지를 작성합니다.

이렇게 하려면 입력 섹션에서 필터 및 출력으로 이동합니다.

필터 섹션을 사용하여 들어오는 메시지를 구문 분석하여 이미 작업할 수 있는 해시(키-값 쌍)를 얻습니다. 조건에 따라 파싱합니다. 출력 섹션에서 메시지를 선택하고 각 메시지를 자체 채널로 보냅니다.

grok으로 메시지 구문 분석

텍스트 문자열을 구문 분석하고 필드 집합을 가져오기 위해 필터 섹션에 특수 플러그인인 grok가 있습니다.

여기에 자세한 설명을 제공하려는 목표를 설정하지 않고(이에 대해서는 공식 문서) 간단한 예를 들겠습니다.

이렇게 하려면 입력 라인의 형식을 결정해야 합니다. 나는 그것들을 다음과 같이 가지고 있습니다 :

1 정보 메시지1
2 오류 메시지2

저것들. 먼저 식별자, INFO/ERROR, 공백 없는 단어.
어렵지는 않지만 작동 원리를 이해하기에 충분합니다.

따라서 필터 섹션의 grok 플러그인에서 문자열을 구문 분석하기 위한 패턴을 정의해야 합니다.

다음과 같이 표시됩니다.

filter {
  grok {
    match => { "message" => ["%{INT:message_id} %{LOGLEVEL:message_type} %{WORD:message_text}"] }
   }
  } 

기본적으로 정규 표현식입니다. INT, LOGLEVEL, WORD와 같은 기성품 패턴이 사용됩니다. 그들의 설명과 다른 패턴은 여기에서 볼 수 있습니다. 여기에

이제 이 필터를 통과하면 문자열이 message_id, message_type, message_text라는 세 필드의 해시로 바뀝니다.

출력 섹션에 표시됩니다.

if 명령을 사용하여 출력 섹션의 메시지 라우팅

우리가 기억하는 것처럼 출력 섹션에서 메시지를 두 개의 스트림으로 분할하려고 했습니다. 일부 - iNFO인 경우 콘솔에 출력하고 오류가 있는 경우 파일로 출력합니다.

이 메시지를 어떻게 공유할 수 있습니까? 문제의 조건은 이미 해결책을 제안합니다. 결국 우리는 이미 INFO와 ERROR의 두 값만 사용할 수 있는 전용 message_type 필드를 가지고 있습니다. if 문을 사용하여 선택할 것입니다.

if [message_type] == "ERROR" {
        # Здесь выводим в файл
       } else
     {
      # Здесь выводим в stdout
    }

필드 및 연산자 작업에 대한 설명은 이 섹션에서 찾을 수 있습니다. 공식 매뉴얼.

이제 결론 자체에 대해.

콘솔 출력, 여기에서 모든 것이 명확합니다 - stdout {}

그러나 파일에 대한 출력 - 컨테이너에서 이 모든 것을 실행하고 있으며 결과를 작성하는 파일을 외부에서 액세스할 수 있도록 하려면 docker-compose.yml에서 이 디렉터리를 열어야 합니다.

총계 :

파일의 출력 섹션은 다음과 같습니다.


output {
  if [message_type] == "ERROR" {
    file {
          path => "/usr/share/logstash/output/test.log"
          codec => line { format => "custom format: %{message}"}
         }
    } else
     {stdout {
             }
     }
  }

출력을 위해 docker-compose.yml에 볼륨을 하나 더 추가합니다.

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro
      - ./logs:/usr/share/logstash/input
      - ./output:/usr/share/logstash/output

우리는 시작하고, 시도하고, 두 흐름으로 나뉘는 것을 봅니다.

출처 : habr.com