将参数从 Airflow 传递到 Logstash

Question

我已将 logstash 配置为在默认气流日志路径中侦听日志。我想在 elasticsearch 中创建索引为 {dag_id}-{task_id}-{execution_date}-{try_number}。所有这些都是来自 Airflow 的参数。这些是airflow.cfg 中修改后的值。

[core]
remote_logging = True

[elasticsearch]
host = 127.0.0.1:9200                                                                            
log_id_template = {{dag_id}}-{{task_id}}-{{execution_date}}-{{try_number}}                       
end_of_log_mark = end_of_log
write_stdout = True                                                                              
json_format = True                                                                               
json_fields = asctime, filename, lineno, levelname, message

这些任务实例详细信息需要从 Airflow 传递到 logstash。 dag_id, 任务ID， 执行日期， try_number

这是我的 logstash 配置文件。

input {
     file{                                                                                                  
       path => "/home/kmeenaravich/airflow/logs/Helloworld/*/*/*.log"                                   
       start_position => beginning                                                                    
     } 
}                                                                                                
output {                                                                                                  
     elasticsearch { 
        hosts => ["127.0.0.1:9200"]                                                                      
        index => "logginapp-%{+YYYY.MM.dd}"                                                        
     }                                                                                                
     stdout { codec => rubydebug }                                                          
} 

我有 2 个问题。如何将 Airflow 中的参数传递给 Logstash？

我已将 logstash 配置为监听日志路径。由于 airfow.cfg 中的 remote_logging 为 True，因此日志不会写入基本日志文件夹。如果这是错误的，或者如果我连接到 Amazon S3，日志也会写入 base_log_folder 路径。但是，我要配置 logstash，需要将日志写入本地文件夹。我使用气流版本 1.10.9 。如何将日志流式传输到 Elasticsearch 索引。

Answer 1

要回答您的第一个问题（我假设您的意思是直接将日志传递给 Elasticsearch），您不能。 Airflow“Elasticsearch Logging”并不是真正的 Elasticsearch 日志记录，而是更多的配置，以使日志记录能够发送到 Elasticsearch。属性的命名（在我看来）有点令人困惑，因为它表明您可以直接写入 Elasticsearch。 您可以将 Airflow 配置为从 Elasticsearch 中读取日志。见Airflow Elasticsearch documentation for more information：

Airflow 可以配置为从 Elasticsearch 中读取任务日志和 可选择以标准或 json 格式将日志写入标准输出。这些日志 稍后可以收集并转发到 Elasticsearch 集群 使用 fluentd、logstash 或其他工具。

当您启用write_stdout = True 时，输出将写入标准输出。如果您希望将输出写入文件中，您必须设置 write_stdout = False 或将其留空。然后，您的 logstash 配置应该会找到这些文件，这会回答您的第二个问题。

干杯 迈克尔