# Filebeat -\> logstash 순서 보장

**URL:** <https://discuss.elastic.co/t/filebeat-logstash/52590>\
**Category:** 한국어 질문 및 토론\
**Created:** [June 13, 2016, 10:12am UTC](https://discuss.elastic.co/t/filebeat-logstash/52590 "2016-06-13T10:12:36Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![Donovan](https://avatars.discourse-cdn.com/v4/letter/d/ea666f/32.png) [@Donovan](https://discuss.elastic.co/u/Donovan)\
**Post date:** [June 13, 2016, 10:12am UTC](https://discuss.elastic.co/t/filebeat-logstash/52590/1 "2016-06-13T10:12:36Z")

</div>

현재 filebeat으로 log file을 읽어서 logstash로 보내고 logstash에서 s3와 kafka로 전송하는 간단한 스트림을 구성했습니다.  
file beat 설정은 아래와 같고

```auto
  # List of prospectors to fetch data.
  prospectors:
    -
      paths:
        - D:\*.log
      input_type: log
      document_type: rhyme
      ignore_older: 2h
output:
  logstash:
    # The Logstash hosts
    hosts: ["internal.logstash:5100"]

logging:
  level: debug

  # enable file rotation with default configuration
  to_files: true

  # do not log to syslog
  to_syslog: false

  files:
    path: C:\Program Files\Filebeat\log
    rotateeverybytes: 10485760 # = 10MB
    name: beat.log
    keepfiles: 7

```

logstash의 경우에도 beat에서 받고 grok처리 후 output으로 가는 단순한 구조입니다.

```auto
input {
    beats {
        port => 5100
        type => "rhyme"
        congestion_threshold => 60
    }
}
filter {
    if [type] == "rhyme" {
        grok {
            match => ["message", "%{DATA:topic_id}\|%{DATA:message_id}\|%{GREEDYDATA:body}"]
        }
    }
}
output {
    if [type] == "rhyme" {
        s3 {
            access_key_id => "???"
            secret_access_key => "???"
            region => "???"
            bucket => "???"
            time_file => 60
            prefix => "rhyme-"
        }
        kafka {
            topic_id => "%{topic_id}"
            message_key => "%{message_id}"
            codec => plain {
                format => "%{body}"
            }
            bootstrap_servers => "internal.kafka:6667"
        }
    }
}

```

그런데 순서보장이 되지 않습니다. log파일은 분명히 순서대로 되어있습니다만...  
전부 다 출력을 찍어보기에는 간헐적이라 잡아낼 끈기도 없구요.  
혹시 체크해야할 포인트들만 알려주시면 감사하겠습니다.

---

<div class="post-metadata">

**Author:** ![MINSOO\_JUN](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/minsoo_jun/32/10927_2.png) [@MINSOO\_JUN](https://discuss.elastic.co/u/MINSOO_JUN)\
**Post date:** [July 18, 2016, 12:18am UTC](https://discuss.elastic.co/t/filebeat-logstash/52590/2 "2016-07-18T00:18:43Z")

</div>

Log파일 자체에 timestamp가 있으면 그것을 이용하는게(필터에서 타임스템프를 덮어쓰기) 좋지 않을까요? 한 쓰레드에서 움직이는 모델이 아니라 동시에 네트워크상에 흘러 가는 것이라 반드시 보낸 순서대로 도착을 한다는 것을 보증하기는 힘들거라 봅니다.

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [July 6, 2017, 1:47pm UTC](https://discuss.elastic.co/t/filebeat-logstash/52590/3 "2017-07-06T13:47:02Z")

</div>