How to map array inside message in Logstash HTTP Output

Question

I am using Logstash to update by query existing Elasticsearch documents with an additional field that contains aggregate values extracted from Potgresql table. I use elastichsearch output to load one index using document_id and http output to update another index that have different document_id but receving errors:

[2023-02-08T17:58:12,086][ERROR][logstash.outputs.http ][main][b64f19821b11ee0df1bd165920785876cd6c5fab079e27d39bb7ee19a3d642a4] [HTTP Output Failure] Encountered non-2xx HTTP code 400 {:response_code=>400, :url=>"http://localhost:9200/medico/_update_by_query", :event=>#LogStash::Event:0x19a14c08}

This is my pipeline configuration:

input {
    jdbc {
        # Postgres jdbc connection string to our database, mydb
        jdbc_connection_string => "jdbc:postgresql://handel:5432/mydb"
        statement_filepath => "D:\ProgrammiUnsupported\logstash-7.15.2\config\nota_sede.sql"
    }
}

filter {
    aggregate {
        task_id => "%{idCso}"
        code => "
            map['idCso'] = event.get('idCso')
            map['noteSede'] ||= []
            map['noteSede'] << {
                'id' => event.get('idNota'),
                'tipo' => event.get('tipoNota'),
                'descrizione' => event.get('descrizione'),
                'data' => event.get('data'),
                'dataInizio' => event.get('dataInizio'),
                'dataFine' => event.get('dataFine')
            }
            event.cancel()"
        push_previous_map_as_event => true
        timeout => 60
        timeout_tags => ['_aggregatetimeout']       
    }
   }
}

output {

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

#       this works
    elasticsearch {
        hosts => "https://localhost:9200"
        document_id => "STRUTTURA_%{idCso}" 
        index => "struttura"
        action => "update"
        user => "user"
        password => "password"
        ssl => true
        cacert => "/usr/share/logstash/config/ca.crt"   
    }
    
    http {
        url => "http://localhost:9200/medico/_update_by_query"
        user => "elastic"
        password => "changeme"
        http_method => "post"
        format => "message"
        content_type => "application/json"
        message => '{
                        "query":{
                            "term":{
                                "idCso":"%{idCso}"
                            }
                        },
                        "script":{
                            "source":"ctx._source.noteSede=params.noteSede",
                            "lang":"painless",
                            "params":{
                                "noteSede":"%{noteSede}"
                                }
                            }
                        }
                    }'
    }
}

The stdout output show me the sended docs to output like this:

{
     "query" => {
        "term" => {
            "idCso" => "859119"
        }
    },
    "script" => {
        "source" => "ctx._source.noteSede=params.noteSede",
        "lang" => "painless",
        "params" => {
            "noteSede" => "{dataFine=null, dataInizio=2020-02-13, descrizione=?, tipo=DB, id=6390644, data=2020-02-13 12:26:58.409},{dataFine=null, dataInizio=2020-02-13, descrizione=?, tipo=DE, id=6390645, data=2020-02-13 12:26:58.41}"
        }
        }
    }
}

How could I set noteSede array field into message to _update_by_query ?

Answer 1

I found the trick using ruby code for setting params array end set the format of http output to json. Possibile code optimization but it works!

ruby {
    code => '
        temp = event.get("noteSede")
        note = {"noteSede" => temp}
        event.set("script", "params" => note)
    '
}
mutate {
    add_field => {
        "[script][lang]" => "painless"
        "[query][term][idSede]" => '%{idSede}'
        "[script][source]" => "ctx._source.noteSede = params.noteSede"
    }
    remove_field => ["tags", "idSede", "noteSede", "@version", "@timestamp"]            
}   

Bye