日志传输工具Logstash使用中遇到的几个问题(1)

intro：去年记录的logstash几个问题，因为涉及较多且杂乱，主题并不是那么清晰，今天整理时候发现写起来太长了，所以打算分成2-4篇文章，几个小块逐个讨论。

系统概况：
1）各服务产生事件(日志)另有一套Event机制支持，本文讨论的是如何跨数据中心(机房)传输。
2）普适性的，适合各语言实现的设计，故基于文件传输，即服务在本地产生滚动的日志，组件去 tail 文件的方式，将这些数据 即时 传输至另一个机房的Kafka。
3）logstash是5.6.4版本，kafka版本 0.10.0.1版本问题（再高版本的logstash 就不支持该版本的Kafka了）
4）logstash的input是滚动的日志文件，即tail_mode，output为kafka，无解析，插件开启了metric，queue 为 persistent，即文件模式，硬件为磁盘。
5）文件系统为ext4，系统inode是重用的。
6）本文贴出配置或数据已做脱敏处理，如有疑问请忽略。

本文先讨论遇到的两个问题
1）即便是 kill -15， logstash重启后会丢失数据
2）性能问题，可能消息大小会有影响，但观察下来logstash上传速率上限在 1.2万条/秒.

logstash的配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
input {
    file {
      path => ["/data/xxx-*.csv"]
      start_position => "beginning"
      type => "xyz"
      sincedb_write_interval => "1"
      sincedb_path => ".sincedbxxx"
      discover_interval => "1"
      tags => "xxx"
    }
    file {
      path => ["/data/yyy-*.csv"]
      start_position => "beginning"
      type => "xyz"
      sincedb_write_interval => "1"
      sincedb_path => ".sincedbyyy"
      discover_interval => "1"
      tags => "yyy"
    }

}
filter {
  mutate {
    gsub => [
      "message","[\\\\]{2,}n","",
      "message","[\\\\]{2,}r",""
      ]
  }
  metrics {
     meter => "events@%{[type]}@%{[path]}"
     meter => "events@%{[type]}@sum"
     ...
     clear_interval => 86401
  }
}
output {
  if "metric" in [tags] {
    stdout { codec => rubydebug }
    ...
  } else {
      kafka {
        ...
      }
  }
}

先看问题1–重启后会丢失数据

导致原因是 系统inode重用，通常在高请求的机器上，文件系统都会选择重用inode，那么logstash重启时加载的.sincedb文件里的inode，很可能就是上一次其他文件上报文件的位移记录，不是本文件的，从而导致丢数据。
举例来说：
因为sincedb记录了inode和inode对应文件的位移，logstash重启时候会加载这些inode和位移，当inode可重用的，就会产生一个问题：
假设我们的日志文件是滚动的，比如一小时一个文件，
假设之前inode有一条历史记录 “111222 20460”，表示inode=111222对应的文件file-2018030403 的位移是 20460(字节)，即表示上次111222对应文件读到20460这个字节处，但是，其实这时侯 111222 这个inode因为重用，对应的文件已经是 file-2018030503，且其大小是 1020460 字节了。
当我们重启时，logstash会加载111222这个inode对应的进度，即重启后 file-2018030503 这个文件其实是从 20460 字节处开始读取并上报，而不是配置的 从0即文件开头开始上报。
这就导致在上报 file-2018030503 这个文件时丢失了 20460 个字节(对于行读取还会另做处理)。

理解上面后，你可能会疑问，既然 重启 + inode重 用会丢失数据，那么 为什么平时上报滚动的日志文件没有丢失，没有出现上面的情况？

这个原因就在于，logstash 在判断新的变更时，会比较inode对应新文件的大小和记录中的文件大小，小于记录值则将记录值清零：

1）logstash未重启，内存中记录一条 “111222 20460” 这条记录
2）当 111222 这个inode对应文件 file-2018030503 新建时，filewatch插件会检测到 111222 这个inode 的文件大小为0
3）0小于20460，则logstash 会将 内存中 “111222 20460” 这条记录更新为 “111222 0”，这个0就是配置中的 from beginning。

这样新inode是被正确定位的，故文件不会出现数据丢失。
这是bug吗？或者说，可否避免这个问题？
可以的，升级下 logstash-input-file 即可。5.6.4版本里对应的机制底层实际是由 filewatch插件实现，新版 logstash-input-file 4.1.10 已经将filewatch并入logstash-input-file，并且新版本有个特性：
.sincedb多了几列，有文件路径(filepath)和时间戳列，加载sincedb文件同时也会加载路径，这样，当检测到该inode有数据产生时，除了和低版本一样会对比inode外，还增加了一种判断，即：

inode相等时，会判断filepath是否相等，如否则认为是新的文件。

故重启时不会再因inode重用问题导致丢数据了。
这里是代码分析，比较长，如果不敢兴趣，可略过。
入口在watch.rb里
attr_reader :discoverer, :watched_files_collection

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
----
processor在tail_mode->processor.rb
    def process_all_states(watched_files)
      process_closed(watched_files)
      return if watch.quit?
      process_ignored(watched_files)
      return if watch.quit?
      process_delayed_delete(watched_files)
      return if watch.quit?
      process_restat_for_watched_and_active(watched_files)
      return if watch.quit?
      process_rotation_in_progress(watched_files)
      return if watch.quit?
      process_watched(watched_files)
      return if watch.quit?
      process_active(watched_files)
    end
    。。。
    def process_watched(watched_files)
      logger.trace("Watched processing")
      to_take = @settings.max_active - watched_files.count{|wf| wf.active?}
      if to_take > 0
        watched_files.select {|wf| wf.watched?}.take(to_take).each do |watched_file|
          watched_file.activate
          if watched_file.initial?
            create_initial(watched_file)
          else
            create(watched_file)
          end
          break if watch.quit?
        end
      else
        now = Time.now.to_i
        if (now - watch.lastwarn_max_files) > MAX_FILES_WARN_INTERVAL
          waiting = watched_files.size - @settings.max_active
          logger.warn(@settings.max_warn_msg + ", files yet to open: #{waiting}")
          watch.lastwarn_max_files = now
        end
      end
    end
    。。。
    def create_initial(watched_file)
      @create_initial.handle(watched_file)
    end

------discove 入口
discover.rb
def discover_any_files(path, ongoing)
  fileset = Dir.glob(path).select{|f| File.file?(f)}
  logger.trace("discover_files",  "count" => fileset.size)
  logger.warn("discover_files",  "count" => fileset.size)
  fileset.each do |file|
    pathname = Pathname.new(file)
    new_discovery = false
    watched_file = @watched_files_collection.watched_file_by_path(file)
    if watched_file.nil?
      begin
        path_stat = PathStatClass.new(pathname)
      rescue Errno::ENOENT
        next
      end
      watched_file = WatchedFile.new(pathname, path_stat, @settings)
      new_discovery = true
      logger.info("discover_files handling:", "new:"=> new_discovery, "watched_file:" => watched_file.details)
    end
    # if it already unwatched or its excluded then we can skip
    next if watched_file.unwatched? || can_exclude?(watched_file, new_discovery)
    logger.trace("discover_files handling:", "new discovery"=> new_discovery, "watched_file details" => watched_file.details)
    if new_discovery
      watched_file.initial_completed if ongoing
      # initially when the sincedb collection is filled with records from the persistence file
      # each value is not associated with a watched file
      # a sincedb_value can be:
      #   unassociated
      #   associated with this watched_file
      #   associated with a different watched_file
      if @sincedb_collection.associate(watched_file)
        if watched_file.file_ignorable?
          logger.trace("Discoverer discover_files: #{file}: skipping because it was last modified more than #{@settings.ignore_older} seconds ago")
          logger.info("Discoverer discover_files: #{file}: skipping because it was last modified more than #{@settings.ignore_older} seconds ago")
          # on discovery ignorable watched_files are put into the ignored state and that
          # updates the size from the internal stat
          # so the existing contents are not read.
          # because, normally, a newly discovered file will
          # have a watched_file size of zero
          # they are still added to the collection so we know they are there for the next periodic discovery
          watched_file.ignore_as_unread
        end
        # now add the discovered file to the watched_files collection and adjust the sincedb collections
        @watched_files_collection.add(watched_file)
      end
    end
    # at this point the watched file is created, is in the db but not yet opened or being processed
  end
end

------最终发现重命名的inode文件调用是在 tail_mode->handlers->base.rb
base.rb
def handle(watched_file)
  logger.trace("handling: #{watched_file.filename}")
  logger.info("handling: #{watched_file.filename}")
  unless watched_file.has_listener?
    watched_file.set_listener(@observer)
  end
  handle_specifically(watched_file)
end

def add_new_value_sincedb_collection(watched_file)
  sincedb_value = get_new_value_specifically(watched_file)
  logger.trace("add_new_value_sincedb_collection", "position" => sincedb_value.position, "watched_file details" => watched_file.details)
  sincedb_collection.set(watched_file.sincedb_key, sincedb_value)
  sincedb_value
end

def get_new_value_specifically(watched_file)
  position = watched_file.position_for_new_sincedb_value
  value = SincedbValue.new(position)
  value.set_watched_file(watched_file)
  watched_file.update_bytes_read(position)
  value
end
--->
creat_initial.rb
module FileWatch module TailMode module Handlers
  class CreateInitial < Base
    def handle_specifically(watched_file)
      if open_file(watched_file)
        logger.trace("handle_specifically opened file handle: #{watched_file.file.fileno}, path: #{watched_file.filename}")
        logger.info("handle_specifically opened file handle: #{watched_file.file.fileno}, path: #{watched_file.filename}")
        add_or_update_sincedb_collection(watched_file)
      end
    end

    def update_existing_specifically(watched_file, sincedb_value)
      position = watched_file.last_stat_size
      if @settings.start_new_files_at == :beginning
        position = 0
      end
      logger.trace("update_existing_specifically - #{watched_file.path}: seeking to #{position}")
      logger.info("update_existing_specifically - #{watched_file.path}: seeking to #{position}")
      watched_file.update_bytes_read(position)
      sincedb_value.update_position(position)
    end
  end
end end end
--->
base.rb
def add_or_update_sincedb_collection(watched_file)
  sincedb_value = @sincedb_collection.find(watched_file)
  if sincedb_value.nil?
    sincedb_value = add_new_value_sincedb_collection(watched_file)
    watched_file.initial_completed
  elsif sincedb_value.watched_file == watched_file
    update_existing_sincedb_collection_value(watched_file, sincedb_value)
    watched_file.initial_completed
  else
    msg = "add_or_update_sincedb_collection: found sincedb record"
    logger.trace(msg,
      "sincedb key" => watched_file.sincedb_key,
      "sincedb value" => sincedb_value
    )
    # detected a rotation, Discoverer can't handle this because this watched file is not a new discovery.
    # we must handle it here, by transferring state and have the sincedb value track this watched file
    # rotate_as_file and rotate_from will switch the sincedb key to the inode that the path is now pointing to
    # and pickup the sincedb_value from before.
    msg = "add_or_update_sincedb_collection: the found sincedb_value has a watched_file - this is a rename, switching inode to this watched file"
    logger.trace(msg)
    existing_watched_file = sincedb_value.watched_file
    if existing_watched_file.nil?
      sincedb_value.set_watched_file(watched_file)
      logger.trace("add_or_update_sincedb_collection: switching as new file")
      watched_file.rotate_as_file
      watched_file.update_bytes_read(sincedb_value.position)
    else
      sincedb_value.set_watched_file(watched_file)
      logger.trace("add_or_update_sincedb_collection: switching from...", "watched_file details" => watched_file.details)
      watched_file.rotate_from(existing_watched_file)
    end
  end
  sincedb_value
end

watched_file.rb-->
def position_for_new_sincedb_value
  if @initial
    # this file was found in first discovery
    @settings.start_new_files_at == :beginning ? 0 : last_stat_size
  else
    # always start at the beginning if found after first discovery
    0
  end
end

logstash-input-file 4.1.10 类似代码有所改进，可找到对应逻辑，总结来说是：

  1. watch.rb的discovery机制，扫描当前目录下文件，并：
    1）如果存在sincedb的key，且文件名相等，且size相等，则放入watched_files 并且设置state为ignored，且sincedb[key]=size
    2）如果存在sincedb的key，但文件名不相等，则从sincedb中删除该key，但放入watched_files 并且设置state为watched
    3）其他…
  2. tail_mode->processor.rb定期 process_all_states中当处理到 process_watched 时，
    发现该文件，设置 watched_file.activate，并调用 create_initial，create_initial正是
    Handlers::CreateInitial.new(self, sincedb_collection, observer, @settings)类，
    则handle方法就是 tail_mode->handlers->base.rb的hadle方法，即creat_initial.rb -> handle_specifically 方法
    handle_specifically调用base.rb的add_or_update_sincedb_collection
    即最终调用 sincedb_value = add_new_value_sincedb_collection(watched_file)
    add_new_value_sincedb_collection通过 get_new_value_specifically 获取到sincedb_value，即当前文件offset(get_new_value_specifically的逻辑如下，若配置beginning则为0，否则为last stat 的size)

上述完成后，则会进行下一步的process_active，active就是通过监控文件的 @size 是否大于 @bytes_read，来判断文件当前是否在grown

需要指出的是，升级 logstash-input-file 重启前，最好手动修改下 .sincedb文件，否则直到 .sincedb文件里的相关行inode都拥有 filepath 这一列，还是会出现上述丢失情况的。

更多思考：
inode相等，logstash是根据filepath解决重用问题，如果同名新文件或位置变更，应该依旧会导致数据遗失或重复问题。
其实这里也可以根据文件的创建时间来判断，但不能解决系统时钟回退问题，即只能处理正向时间。

性能问题

一开始以为是跨数据中心 9ms的网络延迟或kafka参数配置导致写kafka性能不高，这个不难优化，但已排除，并非该原因。
实际测试同网段kafka性能瓶颈依旧维持1.2万/s。
排查过程比较冗长，这里不详述，简单记下结论：

瓶颈在 config/logstash.yml里配置的queue.type: persisted，即使用了持久化的Ack队列。

测试机器为本人mac机器，硬盘为SSD/8G内存，如果使用 memory模式，性能可至少在3.6万/秒。
但这里为避丢数据，所以优化时不能作改动，其次是把 input 一分为二，性能略提升，在生产机器上至少可达 2.0万条/秒，提升约50%+。
更多：如果是SSD，即本人开发机上测试，但input是可以到2.8万/秒，双input可到4万/秒

起初认为增加配置中的 worker threads 数量可以改进性能，但其实不是，因为 worker threads 仅是针对pipeline的多线程，处理的是事件，这里的瓶颈在input和Ackqueue的交互，不过我没找到对应多线程处理input的配置参数，【如果你有发现或了解，感谢邮件给 aXRob21hc2xhdUBxcS5jb20= (base64)】，所以拆分为两个input，确实起到优化效果。

其他

后续几篇文章，还会分析 logstash 使用过程遇到的一个莫名罢机的问题，以及 传输数据完整性保障的问题，并会分析下 logstash的 WrappedSynchronousQueue/WrappedAckedQueue机制，也会分析一下怎么解决传输Kafka失败不丢数据的问题。

大概说下，虽然logstash那个queue叫做持久化且ACK的queue，听起来像是支持传输失败会重试从而保证数据完整不丢失。
但其实不是，但这个ACK甚至都不是针对一次完整性传输的ACK，即 这个ACK机制和 Flume的 Source-Channel-Sink之间的Transaction机制一样，即便是flume支持的持久化的File channel，这个Transaction也只是针对 Source和Channel(Channel和sink)之间传输(或者说 传递)这个动作是transactional的，而不是保证数据一定被传输成功，即事务是针对 传递 这个动作，而不是 传输 整个过程。这点和大多数人理解的新版kafka支持exactly once概念基本是不对的一样。

总结：

1，或许你或者很多文章认为开发一款普适的日志即时上报工具应该不难，本人一开始也这么认为，即便是只要求支持本地文件读取并推送至kafka。
但希望了解logstash使用运行原理后，你可以取消这种想法，除非你有比logstash团队更多的时间。
2，或许你可以搜索到网上许多比较几大流行日志上报开源工具的优劣分析，考虑到每个人的开发语言/习惯/经历不同，甚至还有基于容器的上报机制，去推荐用哪个，并不是合适的行为。
但几乎可以肯定的是，logstash 相比其他实现，最大(或者唯一)的劣势就是logstash 时JRuby语言写的。
3，如果非要建议，最好就是选择一个，把其中的坑和不足都踩一遍(除非该组件不再维护了)，了解了存在的问题后，才能判断使用哪个好。
4，除非必须，尽量不要使用logstash解析数据，重复三遍！首先是性能不可控，更重要是解析异常不可控，毕竟对于非Ruby系的开发者来说有点困难。
所以我建议尽量留给上下游去处理，仅把logstash做为纯无结构的数据传输工具，尽管其解析插件很丰富。
但若对数据异常并不是很敏感，使用Logstash插件解析不影响。