Конфигурация Logstash — имена файлов S3?

Я пытаюсь использовать Logstash для сбора данных о продажах для нескольких клиентов от разных поставщиков.

Пока у меня есть корзина S3 (Входящие), в которую я могу поместить свои файлы (в настоящее время CSV), и в соответствии с префиксом клиентского кода в файле данные помещаются в эластичный индекс для каждого клиента. Все это прекрасно работает.

У меня проблема в том, что у меня есть данные от нескольких поставщиков, и мне нужен способ определить, какой файл принадлежит тому или иному поставщику. Добавление дополнительного столбца в CSV не вариант, поэтому я планировал добавить его к имени файла, чтобы в итоге я получил соглашение об именах файлов, например clientcode_vendorcode_reportdate.csv.

Я могу что-то упустить, но кажется, что на S3 я не могу получить доступ к имени файла внутри моей конфигурации, что кажется сумасшедшим, учитывая, что в какой-то момент префикс явно читается. Я собирался использовать фильтр Grok или Ruby, чтобы просто разделить имя файла на символ подчеркивания, что дало бы мне три ключевые переменные, которые я могу использовать в своей конфигурации, но пока все попытки не увенчались успехом. Я даже не могу получить полный source_path или имя файла в качестве переменной.

Моя конфигурация пока выглядит примерно так: я удалил неудачные попытки использования фильтров Grok/Ruby.

input {
    s3 {
    access_key_id =>"MYACCESSKEYID"
    bucket => "sales-inbox"
    backup_to_bucket => "sales-archive"
    delete => "true"
    secret_access_key => "MYSECRETACCESSKEY"
        region => "eu-west-1"
        codec => plain
        prefix => "ABC"
        type => "ABC"
  }
    s3 {
    access_key_id =>"MYACCESSKEYID"
    bucket => "sales-inbox"
    backup_to_bucket => "sales-archive"
    delete => "true"
    secret_access_key => "MYSECRETACCESSKEY"
        region => "eu-west-1"
        codec => plain
        prefix => "DEF"
        type => "DEF"
  }
}

filter {
    if [type] == "ABC" {
        csv {
            columns => ["Date","Units","ProductID","Country"]
            separator => ","
        }
        mutate {
            add_field => { "client_code" => "abc" }
        }      
    } 
    else if [type] == "DEF" {
        csv {
            columns => ["Date","Units","ProductID","Country"]
            separator => ","
        }
        mutate {
            add_field => { "client_code" => "def" }
        }
    } 
    mutate
    {
        remove_field => [ "message" ]
    }   
}
output {
        elasticsearch {
            codec => json
            hosts => "myelasticcluster.com:9200"
            index => "sales_%{client_code}"
            document_type => "sale"
          }
        stdout { codec => rubydebug }
}

Мы будем очень признательны за рекомендации тех, кто хорошо разбирается в конфигурациях Logstash!