LogStash - sql_last_value 和 last_run_metadata_path 问题

Question

我是 ELK 的新手，目前正在为一些设置而苦苦挣扎——也许我错过了一点。 我已经设置了我的 Logstash 来解析我的数据库 (MySql)，我有 2 个案例：

1. “意外维护” -> 在这种情况下，我需要在不清除我的数据库的情况下手动停止/启动 Logstash，因此当它启动时，我需要它来跟踪尚未摄取的记录。

2. “计划维护” -> 在这种情况下，我在我的数据库上进行了截断，从而使用我的 ID 返回到 0。我认为通过更改“last_run_metadata_path”中的文件集，Logstash 将开始使用我提供的这个新值。虽然，它看起来像是将旧的 Id 保存在内存中。因此，我的新条目丢失了并且从未被摄取。

这是我的设置：

input {
  jdbc {
    codec => plain { charset=>"UTF-8" }
    clean_run => false
    jdbc_driver_library => "C:\com.mysql.jdbc_5.1.5.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://localhost:3306/talend_logs?characterEncoding=utf8"
    jdbc_user => "root"
    jdbc_password => "************"
    schedule => "* * * * *"
    statement => "SELECT id, pid, ...... FROM flowlogs WHERE id > :sql_last_value order by id" 
    use_column_value => true
    tracking_column => "id"
    last_run_metadata_path => "C:\logstash-8.6.1\config\lastValueOfMySqlForLogs.yml"
  }
}

output {
  elasticsearch {
    hosts => ["https://localhost:9200"]
    index => "myLogs"
    user => "elastic"
    password => "*********"
    cacert => "C:\elasticsearch-8.6.1\config\certs\http_ca.crt"
  }

stdout{}

}

如果我误解了带有索引的 logstash 行为中的某些内容，您能告诉我吗？甚至有可能实现我想用 ID 做的事情吗？ （不幸的是无法处理日期）

非常感谢。

我尝试使用 clean_run 参数。设置为 true 时，它会在每次重新启动时从 0 重新启动，这在文档中是正常的。 当我使用 last_run_metadata_path 将其设置为 false 时，我希望它使用此文件中设置的最后一个值重新启动。情况确实如此，但它一直在增加——我无法重置它（我从表中截断数据后需要它）。

Answer 1

没关系，我想我通过修改设置解决了我的问题。 我不再基于 id，而是最终基于时间戳。

我正在使用我的数据库中的日期（时刻），这是我将记录插入数据库的日期。然后，我让 logstash 将 sql_last_value 设置为上次查询执行日期。

我只需要处理这个配置中的时区。 到目前为止，它似乎解决了我的两个问题，所以我将结束这个话题。

这是新设置，以防对任何人有帮助。

input {
  jdbc {
    codec => plain { charset=>"UTF-8" }
    clean_run => false
    jdbc_driver_library => "C:\com.mysql.jdbc_5.1.5.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://localhost:3306/talend_logs?characterEncoding=utf8"
    jdbc_user => "root"
    jdbc_password => "**********"
    jdbc_default_timezone => "Europe/Luxembourg[dst_enabled_on_overlap:true]"
    schedule => "* * * * *"
    statement => "SELECT id, pid, ... FROM flowlogs WHERE moment > :sql_last_value order by moment ASC" 
    record_last_run => true
    plugin_timezone => "local"
  }
}

output {
  elasticsearch {
    hosts => ["https://localhost:9200"]
    index => "talend-logs-etl"
    user => "elastic"
    password => "***********"
    cacert => "C:\elasticsearch-8.6.1\config\certs\http_ca.crt"
  }

stdout{}

}