jjeong

참고문서)

https://www.elastic.co/guide/en/logstash/current/plugins-filters-split.html

https://www.elastic.co/guide/en/logstash/current/plugins-filters-mutate.html#plugins-filters-mutate-split

예제코드)

라인단위로 정보가 작성 되어 있을 때

데이터 예시)
1|A|가
2|B|나

filter {
	split { }

	mutate {
		split => { "message" => "|" }
		add_field => {
			"first" => "%{[message][0]}"
			"second" => "%{[message][1]}"
			"third" => "%{[message][2]}"
		}
	}
}

$ vi .bashrc

export http_proxy=http://proxy.host:port
export https_proxy=https://proxy.host:port

$ bin/logstash-plugin install plugins...

외부 인터넷 망이 막혀 있는 경우 proxy 를 이용해서 plugin 설치를 하면 됩니다.

output elasticsearch proxy 랑은 다른 내용입니다.

logstash 사용 시 event, fields 데이터 사용방법에 대한 공홈 레퍼런스 문서 입니다.
처음 사용하시는 분들에게는 유용한 자료라고 생각 되어서 공유 합니다.

https://www.elastic.co/guide/en/logstash/current/event-api.html
https://www.elastic.co/guide/en/logstash/current/event-dependent-configuration.html

Elastic 공식 문서로는 아직 못본 것 같습니다.

기본 Docker 문서는 아래 문서를 참고 하시기 바랍니다.

[공식문서]

https://www.elastic.co/guide/en/logstash/current/docker.html

그러나 아예 예제가 없는 건 아닙니다.

Elastic 사의 github 에 들어가 보시면 참고 하실 수 있는 template 파일이 있습니다.

[github logstash-docker]

https://github.com/elastic/logstash-docker/tree/master/templates

위 템플릿 기반으로 logstash 만 설정해서 구성 하시면 쉽게 할 수 있습니다.

그 이외는 on premise 방식의 logstash 사용 방법과 크게 다른게 없습니다.

[실행방법]

$ docker-compose up -d

이와 같이 실행 하시면 background 로 자동 실행 됩니다.

[docker-compose.yml 예제]

version: '3.7'
services:
  logstash:
    image: docker.elastic.co/logstash/logstash:7.6.2
    volumes:
      - ./examples/logstash.conf/:/usr/share/logstash/pipeline/logstash.conf
    network_mode: bridge

전에 그냥 문서 링크만 걸었었는데 혹시 샘플 코드가 필요 하신 분들도 있을 수 있어서 기록해 봅니다.

[config/logstash-csv.conf]

input {
  file {
      path => ["/Users/henryjeong/Works/poc/elastic/data/*.csv"]
      start_position => "beginning"
  }
}

filter {
    csv {
        separator => ","
        columns => ["title", "cat1", "cat2", "cat3", "area", "sigungu", "description"]
    }

    mutate { rename => ["title", "tt"] }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "csv-%{+YYYY.MM.dd}"
  }
}

[*.csv]

title,cat1,cat2,cat3,area,sigungu,description
엄마손충무김밥,음식,음식점,한식,경상남도,통영시,"KBS ""1박2일"" 욕지도 복불복에 추천된 충무김밥집이다."
평창동의 봄,음식,음식점,한식,서울,종로구,"평창동의 명소, 평창동 언덕마을에 유럽풍의 아름다운 음식점 & 카페. 1층에는 커피는 물론 계절빙수, 단팥죽, 호박죽, 허브티를 파는 카페, 2층에는 식당이 있다. 2층 식당에서는 평창동 봄만의 특별한 한정식 코스 요리와 웰빙 삼계탕 등의 음식을  선보이고 있다. 3층은 최대 40명수용의 연회룸이 있고, 갤러리 전시와 건강교실도 운영하고 있다. "

음봉가든,음식,음식점,한식,경기도,가평군,"음봉가든 (구, 진짜네집)은 자연산 민물고기만을 고집하는 50년 전통의 진정한 매운탕전문점이다. 주재료로 쓰이고 있는 메기 및 쏘가리, 빠가사리
등은 주인이 직접 낚아 올린 물고기로 매우 신선하며 매운탕 역시 얼큰하고 개운하다. 또한 집앞에서 직접 재배하고 키우는 야채와 채소는 매운탕의 국물맛을 더욱 맛나게 해준다."

대자골토속음식,음식,음식점,한식,경기도,고양시,경기 북부지방에선 국수나 수제비를 넣어 국물을 넉넉하게 만든 음식을 '털레기'라 하는데 이곳은 '미꾸라지털레기'를 주메뉴로 하는 30여년
내력의 경기 북부지방 대표향토음식점이다. 주인이 직접 야채와 채소를 재배하는데 고춧가루까지도 직접 재배한 것을 사용한다고하니 그 사명감이 대단하다 할 수 있겠다. 통미꾸라지매운탕의 옛맛을 찾는 사람이라면 꼭 한번은 들려봐야 전통 맛집이다.

장수촌,음식,음식점,한식,경기도,광명시,"부드러운 닭고기 육질에 구수한 누룽지가 함께 하는 '누룽지삼계탕'. 거기다 잘 익은 김치 한 점 더 한다면 그 어떤 맛도 부러울 것이 없다. 식사메뉴인 '누룽지삼계탕'과 '쟁반막국수', 안주메뉴인 골뱅이무침 메뉴가 전부인 '장수촌'은 경기도 광명의 대표맛집으로 토종닭 선별부터 양념 재료 하나 하나까지 일일이 주인이 직접 선별하는 그 정성과 끈기가 맛의 비결이라 할 수 있겠다."
청기와뼈다귀해장국,음식,음식점,한식,경기도,부천시,"부천 사람이라면 모르는 이 없을 정도로 유명한 집이다. 돼지고기와 국물에서 냄새가 안나 여성이나 어린이들 특히 어르신들 보양식으로
도 입소문이 난 곳인데, 부재료 보다는 뼈다귀로 양을 채우는 그 푸짐함 또한 그 소문이 자자하다. 양질의 돼지 뼈에 사골 국물과 우거지를 넣어 맛을 낸 뼈다귀 해장국. 그 맛을 제대로 만날 수 있는 대표적인 음식점이다. "

일번지,음식,음식점,한식,경기도,성남시,닭요리의 으뜸이라 해도 과언이 아닌 '남한산성 닭죽촌민속마을' 에서 남한산성 등산후 가장 많이 찾는 집 중에 한 집이다. 특히 이집은 닭백숙 이외에 '닭도가니'로도 유명한데 이집의 '도가니'는 소의 도가니를 뜻 하는 것이 아니라 장독대의 '독'에서 유래된 말로 '독'에 밥과 닭과 여러보약재를 넣어 만드는 것으로 그 맛과 영양면에서 닭요리 중에 최고라 할 수 있겠다.

장금이,음식,음식점,한식,경기도,시흥시,"경기도 시흥의 지역특산물 중의 하나가 바로 '연'이다. 연은 수련과에 속하는 다년생 수생식물로 뿌리채소로는 드물게 다량의 비타민과 무기질을 함유하고 있어 최근 건강식 식품원료로 각광받고 있으나 그 효능에 비해 다양한 조리방법이 개발되어 있지 않아 흔히 '연'하면 '연근' 반찬 이외엔 생각나는 것이 없는데, 물왕동 연요리전문점 '장금이'를 찾으면 그렇지 않음을 직접 확인 할 수 있다. 흔치 않은 색다른 한정식을 원한다면 한 번쯤은 꼭 한 번 들러 연밥정식과 연잎수육을 맛 봐야 할 곳이다. "

안성마춤갤러리,음식,음식점,한식,경기도,안성시,경기도 안성의 농산물 브랜드 '안성마춤'을 내세워 만든 고품격 갤러리풍 식당으로 각종 공연과 작품전시회 감상과 동시에 농협에서 직접 운영하는 특등급 안성한우를 맞볼 수 있는 곳으로 유명한 집이다. 특히 안성마춤한우 중에 10%만 생산된다는 슈프림급 한우는 늦어도 하루 전에는 꼭 예약을 해야 그 맛을 볼 수 있다 하여 그 희소성에 더더욱 인기가 높다.

언덕너머매운탕,음식,음식점,한식,경기도,연천군,민물고기 중에 살이 탱탱하고 쫄깃한 맛으로 매운탕 재료 중에 으뜸이라 불리우는 '쏘가리'를 메인메뉴로 자랑하는 이집의 '쏘가리매운탕'은
임진강에서 직접 잡아올린 자연 그대로의 그 담백하고 칼칼한 맛이 일품이라 할 수 있다.

보기 좋으라고 개행을 추가 했습니다.

실제 개행 없이 들어 있습니다.

위 데이터는 공공데이터에서 제가 추려 온 데이터 입니다.

위 데이터에서는 Datatype 에 대한 변환을 고민 하지 않아도 되지만 필요한 경우가 있을 수도 있습니다.

공식문서)

https://www.elastic.co/guide/en/logstash/current/plugins-filters-csv.html#plugins-filters-csv-convert

convert

• Value type is hash
• Default value is {}

Define a set of datatype conversions to be applied to columns. Possible conversions are integer, float, date, date_time, boolean

Example:

    filter {
      csv {
        convert => {
          "column1" => "integer"
          "column2" => "boolean"
        }
      }
    }

keyword 나 text 는 지원 하지 않으며 지원하는 datatype 은 integer, float, date, date_time, boolean 입니다.

csv 파일을 밀어 넣을 때 주의 하셔야 하는 점은)

- input 에서 codec 으로 csv 를 지정 하시게 되면 column 명 지정이 원하는 데로 되지 않습니다.

- filter 에서 처리를 하셔야 정상적으로 column 명이 field 명으로 들어 가게 됩니다.

- input codec csv 와 filter 모두 설정 안하게 되면 그냥 message field 에 row 단위로 들어 가게 됩니다.

mutate { rename => ["title", "tt"] }

- 이건 뭔지 딱 보셔도 아시겠죠?

- column 명을 title 에서 tt 로 변경 해서 field 로 생성 되게 됩니다.

공식 문서에 잘 나와 있습니다.

https://www.elastic.co/guide/en/logstash/current/tuning-logstash.html

https://www.elastic.co/guide/en/logstash/current/performance-tuning.html

기본적으로 아래 두 개 설정만 잘 세팅 하셔도 성능 뽑아 낼 수 있습니다.

pipeline.workers)

이 설정 값은 그냥 기본으로 Core 수 만큼 잡아 주고 시작 하시면 됩니다.

pipeline.batch.size)

Worker thread 가 한 번에 처리 하기 위한 이벤트의 크기 입니다.
최적 크기는 직접 구하셔야 합니다.

결국 Elasticsearch 로 Bulk Request 를 보내기 위한 최적의 크기로 설정 한다고 보시면 됩니다.

이외 더 봐주시면 좋은 건

- CPU

- MEM

- I/O (Disk, Network)

- JVM Heap

문의가 들어 왔습니다.

여러 필드에 대해서 date format 이 다른데 어떻게 적용을 해야 하나요?

그래서 소스코드를 열어 보고 아래와 같이 해보라고 했습니다.

date {
...
}

date {
...
}

결국 date {...} 를 필드 별로 선언을 해주면 되는 내용입니다.

공식 문서)

https://www.elastic.co/guide/en/logstash/current/plugins-filters-date.html

Common Options)

https://www.elastic.co/guide/en/logstash/current/plugins-filters-date.html#plugins-filters-date-common-options

Date Filter Configuration Options)

전체 옵션이 필수가 아니긴 합니다.

그래도 꼭 아셔야 하는 설정은 match, target 입니다.

- match 의 첫 번째 값은 field 명이고, 그 이후는 format 들이 되겠습니다.

(공식 문서에 잘 나와 있습니다.)

An array with field name first, and format patterns following, [ field, formats... ]

If your time field has multiple possible formats, you can do this:

- target 은 지정을 하지 않게 되면 기본 @timestamp 필드로 설정이 됩니다. 변경 하고자 하면 target 에 원하시는 field name 을 넣으시면 됩니다.

예제)

date {
    match => ["time" , "yyyy-MM-dd'T'HH:mm:ssZ", "yyyy-MM-dd'T'HH:mm:ss.SSSZ"]
    target => "@timestamp"
}

date {
    match => ["localtime" , "yyyy-MM-dd HH:mm:ssZ"]
    target => "time"
}

DateFilter.java)

/*
 * Licensed to Elasticsearch under one or more contributor
 * license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright
 * ownership. Elasticsearch licenses this file to you under
 * the Apache License, Version 2.0 (the "License"); you may
 * not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.logstash.filters;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.joda.time.Instant;
import org.logstash.Event;
import org.logstash.ext.JrubyEventExtLibrary.RubyEvent;
import org.logstash.filters.parser.CasualISO8601Parser;
import org.logstash.filters.parser.JodaParser;
import org.logstash.filters.parser.TimestampParser;
import org.logstash.filters.parser.TimestampParserFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class DateFilter {
  private static Logger logger = LogManager.getLogger();
  private final String sourceField;
  private final String[] tagOnFailure;
  private RubyResultHandler successHandler;
  private RubyResultHandler failureHandler;
  private final List<ParserExecutor> executors = new ArrayList<>();
  private final ResultSetter setter;

  public interface RubyResultHandler {
    void handle(RubyEvent event);
  }

  public DateFilter(String sourceField, String targetField, List<String> tagOnFailure, RubyResultHandler successHandler, RubyResultHandler failureHandler) {
    this(sourceField, targetField, tagOnFailure);
    this.successHandler = successHandler;
    this.failureHandler = failureHandler;
  }

  public DateFilter(String sourceField, String targetField, List<String> tagOnFailure) {
    this.sourceField = sourceField;
    this.tagOnFailure = tagOnFailure.toArray(new String[0]);
    if (targetField.equals("@timestamp")) {
      this.setter = new TimestampSetter();
    } else {
      this.setter = new FieldSetter(targetField);
    }
  }

  public void acceptFilterConfig(String format, String locale, String timezone) {
    TimestampParser parser = TimestampParserFactory.makeParser(format, locale, timezone);
    logger.debug("Date filter with format={}, locale={}, timezone={} built as {}", format, locale, timezone, parser.getClass().getName());
    if (parser instanceof JodaParser || parser instanceof CasualISO8601Parser) {
      executors.add(new TextParserExecutor(parser, timezone));
    } else {
      executors.add(new NumericParserExecutor(parser));
    }
  }

 public List<RubyEvent> receive(List<RubyEvent> rubyEvents) {
    for (RubyEvent rubyEvent : rubyEvents) {
      Event event = rubyEvent.getEvent();

      switch (executeParsers(event)) {
        case FIELD_VALUE_IS_NULL_OR_FIELD_NOT_PRESENT:
        case IGNORED:
          continue;
        case SUCCESS:
          if (successHandler != null) {
            successHandler.handle(rubyEvent);
          }
          break;
        case FAIL: // fall through
        default:
          for (String t : tagOnFailure) {
            event.tag(t);
          }
          if (failureHandler != null) {
            failureHandler.handle(rubyEvent);
          }
      }
    }
    return rubyEvents;
  }

  public ParseExecutionResult executeParsers(Event event) {
    Object input = event.getField(sourceField);
    if (event.isCancelled()) { return ParseExecutionResult.IGNORED; }
    if (input == null) { return ParseExecutionResult.FIELD_VALUE_IS_NULL_OR_FIELD_NOT_PRESENT; }

    for (ParserExecutor executor : executors) {
      try {
        Instant instant = executor.execute(input, event);
        setter.set(event, instant);
        return ParseExecutionResult.SUCCESS;
      } catch (IllegalArgumentException | IOException e) {
        // do nothing, try next ParserExecutor
      }
    }
    return ParseExecutionResult.FAIL;
  }
}

공홈에 올라와 있는 문서의 번역 본 정도로 정리를 해보려고 합니다.

별거 아니지만 JSON filter 를 많이 사용하면서 Validation 에 대한 인식이 부족해서 오류를 발생 시키는 경우가 꽤 많이 있습니다.

기억력을 돕기 위해 작성해 봅니다.

공식문서)

https://www.elastic.co/guide/en/logstash/current/plugins-filters-json.html

이 내용은 logstash reference 문서 내 filter 항목에 해당 합니다.

용도는 말 그대로 입니다. JSON  parsing 을 하는 filter 입니다.

여기서 문제가 들어 오는 JSON 데이터가 항상 validate 할 거라고 생각 하고 구현 하시는 분들이 계시는데 이 부분이 문제가 됩니다.

기본 이라고 생각 하지만 validation 에 대한 개발자의 생각과 경험이 요즘은 다른 것 같더라구요.

암튼, 그래서 JSON filter 사용 시 제공하는 Option 에 대해서 숙지 하고 사용하시면 좋겠습니다.

기본적으로는 Common Options 를 먼저 보시는게 좋습니다.

JSON Filter Configuration Options)

• skip_on_invalid_json
  json 이 아닌 데이터가 들어 올 경우 에러를 발생 시키지 않고 skip 시키기 위한 옵션 입니다.
  기본 설정 값이 false 이기 때문에 잘 못된 데이터에 대해서 오류가 발생 하게 됩니다.

• source
  json parsing 을 하기 위한 field 를 지정 하게 됩니다.

• tag_on_failure
  정상적으로 처리 되지 않았을 경우 tags 라는 filed 에 "_jsonparsefailure" 값이 추가 됩니다.

•  target
  source field 내 json value 가 target filed 로 등록 되며, 이미 target field 가 있다면 overwrite 됩니다.

위 설정 중에서 skip_on_invalid_json  과 tag_on_failure 만 잘 설정 하셔도 invalid data 에 대한 오류는 잘 넘길 수 있습니다.

간혹 이 오류로 인해서 logstash 가 먹통이 되는 걸 예방 할 수 있기 때문 입니다.

Logstash 사용 시 --config.reload.automatic 설정을 통해서 conf 파일에 대한 변경 사항을 데몬 재시작 없이 할 수 있습니다.

하지만 모든 변경 사항에 대해서 반영이 가능 하지 않기 때문에 사용 시 주의 하셔야 합니다.

크게 사용 방법은 두 가지로 나뉩니다.

2번의 과정을 하고 싶지 않기 때문에 1번을 설정해서 사용을 하는데 문제는 이게 모든 plugins 에서 동작 하지는 않는 다는 것입니다.

만약 아래와 같은 에러를 접하셨다면, 해당 plugin 또는 pipeline 은 auto reload 설정이 동작 하지 않는 것들 이니 참고 하시기 바랍니다.

[에러내용]

쉽게는 기본적인 syntax 오류 같은건 바로 바로 반영 됩니다. :)

가끔 사용하다가도 처음 사용할  때 이해했던 내용과 다르게 기억 될 때가 있습니다.

그래서 또 복습 합니다.

구성은 아래와 같습니다.

각 서버의 file log 를 input file 로 읽고 output sqs 로 보냅니다.

그런 후 input sqs 를 읽고 multi output 으로 file 과 elasticsearch 로 저장을 하게 되는 구조 입니다.

여기서 AWS 의 SQS 에 대한 메시지 수명 주기에 대한 이해가 필요 합니다.

원문)

https://aws.amazon.com/ko/sqs/details/

결국 정리 하면, consumer 가 읽어 가면 다른 consumer 는 이미 읽어간 데이터를 읽어 가지 못합니다. 이유는 위 설명에서와 같이 삭제 되기 때문 입니다.

단일 큐를 사용하면서 큐에 쌓인 메시지의 소비는 다중 consumer 로 처리 하고 단일 저장소로 저장 할 경우 쉽게 구성 할 수 있다는 이야기 였습니다.

여기서 제가 착각한 포인트는 읽어간 메시지가 삭제 되지 않는다는 것이였구요.

이런 오해는 "메시지는 최대 14일 동안 대기열에 보관됩니다." 라는 걸 보고 착각한 것이였습니다.