将 Logstash 中的时间戳时区转换为输出索引名称答案

【问题标题】：Convert timestamp timezone in Logstash for output index name将 Logstash 中的时间戳时区转换为输出索引名称
【发布时间】：2023-03-30 15:10:01
【问题描述】：

在我的场景中，Logstash 接收的系统日志行的“时间戳”是 UTC，我们在 Elasticsearch 输出中使用事件“时间戳”：

output {
    elasticsearch {
        embedded => false
        host => localhost
        port => 9200
        protocol => http
        cluster => 'elasticsearch'
        index => "syslog-%{+YYYY.MM.dd}"
    }
}

我的问题是，在 UTC 午夜，Logstash 在一天结束之前在时区（GMT-4 => America/Montreal）将日志发送到不同的索引，并且索引在 20 小时（晚上 8 点）之后没有日志，因为“时间戳”是 UTC。

我们已经完成了转换时区的工作，但我们遇到了显着的性能下降：

filter {
    mutate {
        add_field => {
            # Create a new field with string value of the UTC event date
            "timestamp_zoned" => "%{@timestamp}"
        }
    }

    date {
        # Parse UTC string value and convert it to my timezone into a new field
        match => [ "timestamp_zoned", "yyyy-MM-dd HH:mm:ss Z" ]
        timezone => "America/Montreal"
        locale => "en"
        remove_field => [ "timestamp_zoned" ]
        target => "timestamp_zoned_obj"
    }

    ruby {
        # Output the zoned date to a new field
        code => "event['index_day'] = event['timestamp_zoned_obj'].strftime('%Y.%m.%d')"
        remove_field => [ "timestamp_zoned_obj" ]
    }
}

output {
    elasticsearch {
        embedded => false
        host => localhost
        port => 9200
        protocol => http
        cluster => 'elasticsearch'
        # Use of the string value
        index => "syslog-%{index_day}"
    }
}

有没有办法优化这个配置？

【问题讨论】：

@timestamp 字段应该是 UTC。其他工具（如 Kibana 和 Elasticsearch Curator）依赖于此，除非您有充分的理由，否则不应更改它。

标签： ruby indexing elasticsearch timezone logstash

【解决方案1】：

这是优化配置，请试一试性能。

您无需使用mutate 和date 插件。直接使用ruby 插件。

input {
    stdin {
    }
}

filter {
    ruby {
            code => "
                    event['index_day'] = event['@timestamp'].localtime.strftime('%Y.%m.%d')
            "
    }
}

output {
    stdout { codec => rubydebug }
}

示例输出：

{
       "message" => "test",
      "@version" => "1",
    "@timestamp" => "2015-03-30T05:27:06.310Z",
          "host" => "BEN_LIM",
     "index_day" => "2015.03.29"
}

【讨论】：

哦，谢谢！我不知道localtime可以不带任何参数调用。
为什么会出现错误：发生 Ruby 异常：未定义的方法 `localtime'
更新更现代的 Logstash 版本：event.set('localdate', event.get('@timestamp').time.localtime.strftime('%Y-%m-%dT%H:%M:%S.%3N%z'))

【解决方案2】：

在logstash version 5.0 and later，你可以使用这个：

filter{
ruby {
        code => "event.set('index_day', event.get('[@timestamp]').time.localtime.strftime('%Y%m%d'))"
    }
}

【讨论】：

【解决方案3】：

在 1.5.0 版本中，我们可以将时间戳按本地时区转换为索引名称。这是我的配置：

filter {
    ruby {
        code => "event['index_day'] = event.timestamp.time.localtime.strftime('%Y.%m.%d')"
    }
}
output {
    elasticsearch {
        host => localhost
        index => "thrall-%{index_day}"
    }
}

【讨论】：

【解决方案4】：

在 Logstash 版本 5.0.2 中，API 被修改。我们可以将时间戳按本地时区转换为索引名称。这是我的配置：

filter { 
   ruby { 
       code => "event['index_day'] = event.timestamp.time.localtime.strftime('%Y.%m.%d')" 
   } 
}

【讨论】：

filter { ruby { code => "event['index_day'] = event.timestamp.time.localtime.strftime('%Y.%m.%d')" } }

【解决方案5】：

类似的用例 - 但使用logstash file output plugin 并写入以事件到达的当地时间为日期的文件。在logstash version 7.12 上验证。

改编自discuss.elastic.co，主要是补零偏移小时。注意！如果您的偏移量有半小时，您将需要相应调整。

filter {
    ruby {
        code => "
        require 'tzinfo'
        tz = 'Europe/Oslo'
        offset = TZInfo::Timezone.get(tz).current_period.utc_total_offset / (60*60)
        event.set('[@metadata][local_date]',
                  event.get('@timestamp').time.localtime(
                      sprintf('+%02i:00', offset.to_s)
                  ).strftime('%Y%m%d'))
        "
    }   
    if ([agent][type] == "filebeat") {
        mutate {
            add_field => ["file_path", "%{[host][name]}_%{[log][file][path]}.%{[@metadata][local_date]}"]
        }  
    } else {   
        mutate {
            add_field => ["file_path", "%{[agent][hostname]}_%{[agent][type]}.%{[@metadata][local_date]}"]
        }
    }
}

【讨论】：