如何配置Logstash收集嵌套Airflow日志并发送至Elasticsearch

修正后的完整配置 #

input {
    file {
        path => "/home/ubuntu/run/logs/scheduler/**/*.log"
        start_position => "beginning"
        codec => "line"
        sincedb_path => "/dev/null" # 可选：重启后重新读取所有日志时添加
    }
}

filter {
    # 从路径提取日期目录和日志文件名
    grok {
        match => { "path" => "%{DATA:base_path}/%{DATE:log_date}/%{DATA:dag_file}\.log" }
    }
    # 从Airflow日志内容提取核心字段（需根据实际日志格式调整grok模式）
    grok {
        match => { "message" => "\[%{DATA:dag_id}\] \[%{DATA:task_id}\] \[%{DATA:execution_date}\] \[%{NUMBER:try_number:int}\]" }
    }
    # 生成唯一log_id
    mutate {
        add_field => {
            "log_id" => "%{dag_id}-%{task_id}-%{execution_date}-%{try_number}"
        }
        # 可选：清理冗余字段
        remove_field => ["base_path", "@version"]
    }
}

output {
    elasticsearch {
        hosts => ["localhost:9200"]
        index => "airflow-scheduler-logs-%{+YYYY.MM.dd}" # 按日期分片索引
    }
    stdout { codec => rubydebug } # 可选：控制台调试输出
}

关键修改点 #

• 路径修正：

  • 原路径~/home/ubuntu/run/log/scheduler/错误：/home/ubuntu是绝对路径，无需叠加~；仅指定目录无法收集文件，需用**/*.log递归匹配所有子目录下的.log文件。
  • **匹配任意层级子目录，*.log匹配所有日志文件。

• 语法修复：

  • 原配置中codec -> "line"是语法错误，Logstash配置必须用=>而非->。

• 过滤逻辑完善：

  • 新增路径grok匹配：提取日期log_date和DAG文件名dag_file，方便后续日志筛选。
  • 新增日志内容grok匹配：需根据你的Airflow实际日志格式调整规则，确保能提取dag_id、task_id等字段，否则log_id会生成无效占位符。
  • 可选sincedb_path => "/dev/null"：重启Logstash后若要重新读取所有历史日志，添加此行（默认Logstash会记录已读文件位置，避免重复收集）。

注意事项 #

• 权限配置：Logstash默认以logstash用户运行，需确保该用户对/home/ubuntu/run/logs/scheduler/目录及所有子文件有读取权限，执行以下命令设置：

  sudo setfacl -R -m u:logstash:r /home/ubuntu/run/logs/scheduler/
  

• 日志格式适配：如果Airflow日志格式与示例grok模式不符，可使用Logstash自带的grok命令调试，调整匹配规则。

内容的提问来源于stack exchange，提问作者Eby