我正在设置Logstash以摄取Airflow日志。以下配置提供了所需的输出：

input {
   file {
      path => "/my_path/logs/**/*.log"
      start_position => "beginning"
      sincedb_path => "/dev/null"
   }
}

filter {
   if [path] =~ /\/my_path\/logs\/containers\/.*/ or [path] =~ /\/my_path\/logs\/scheduler\/.*/ {
      drop{}
   }
   else {
      grok {
        "match" => [ "message", "\[%{TIMESTAMP_ISO8601:log_task_execution_datetime}\]%{SPACE}\{%{DATA:log_file_line}\}%{SPACE}%{WORD:log_level}%{SPACE}-%{SPACE}%{GREEDYDATA:log_message}" ]
        "remove_field" => [ "message" ]
      }
      date {
        "match" => [ "log_task_execution_datetime", "ISO8601" ]
        "target" => "log_task_execution_datetime"
        "timezone" => "UTC"
      }
      dissect {
        "mapping" => { "path" => "/my_path/logs/%{dag_id}/%{task_id}/%{dag_execution_datetime}/%{try_number}.%{}" }
        "add_field" => { "log_id_template" => "{%{dag_id}}-{%{task_id}}-{%{dag_execution_datetime}}-{%{try_number}}" }
      }
   }
}

output {
       stdout {codec => rubydebug{metadata => true}}
}

但是我不喜欢多次指定路径“/my_path/logs/”。在我的输入部分，我尝试用途：
add_field => { "[@metadata][base_path]" => "/my_path/logs/" }
然后，在滤波器部分：

if [path] =~ /[@metadata][base_path].*/ or [path] =~ /[@metadata][base_path].*/ {
      drop{}
   }
...
dissect {
        "mapping" => { "path" => "[@metadata][base_path]%{dag_id}/%{task_id}/%{dag_execution_datetime}/%{try_number}.%{}" }

但它似乎不适用于过滤器或解剖Map中的正则表达式。当我尝试使用here所描述的环境变量时，我遇到了类似的问题。
我有一个想法--也许是天真的想法--我应该能够使用一个变量来引用所有的基路径。