Injecting json string into different Elasticsearch indices using Logstash

Question

I have these sample data stored in postgresql events table:

{"playhead":0,"metadata":"{\"class\":\"Broadway\\\\Domain\\\\Metadata\",\"payload\":{\"console\":{\"command\":\"Doctrine\\\\Bundle\\\\FixturesBundle\\\\Command\\\\LoadDataFixturesDoctrineCommand\",\"arguments\":\"'doctrine:fixtures:load' --env=prod -n\"}}}","@timestamp":"2020-05-13T18:03:24.295Z","payload":"{\"class\":\"OpenLoyalty\\\\Component\\\\Transaction\\\\Domain\\\\Event\\\\TransactionWasRegistered\",\"payload\":{\"transactionId\":\"91d83147-0280-481e-a984-c899e9720ec2\",\"transactionData\":{\"documentNumber\":\"not-matched\",\"purchasePlace\":\"wroclaw\",\"purchaseDate\":1582622706,\"documentType\":\"sell\"},\"customerData\":{\"name\":\"Jan Nowak\",\"email\":\"[email protected]\",\"nip\":\"aaa\",\"phone\":\"+0954730076810\",\"loyaltyCardNumber\":\"not_matched_card_number\",\"address\":{\"street\":\"Ko\\u015bciuszki\",\"address1\":\"12\",\"city\":\"Warsaw\",\"country\":\"PL\",\"province\":\"Mazowieckie\",\"postal\":\"00-800\"}},\"items\":[{\"sku\":{\"code\":\"SKU1\"},\"name\":\"item 1\",\"quantity\":1,\"grossValue\":1,\"category\":\"aaa\",\"labels\":[{\"key\":\"test\",\"value\":\"label\"},{\"key\":\"test\",\"value\":\"label2\"}],\"maker\":\"sss\"},{\"sku\":{\"code\":\"SKU2\"},\"name\":\"item 2\",\"quantity\":2,\"grossValue\":2,\"category\":\"bbb\",\"labels\":[],\"maker\":\"ccc\"}],\"posId\":null,\"excludedDeliverySKUs\":null,\"excludedLevelSKUs\":null,\"excludedLevelCategories\":null,\"revisedDocument\":null,\"labels\":[]}}","@version":"1","id":1,"type":"OpenLoyalty.Component.Transaction.Domain.Event.TransactionWasRegistered","uuid":"91d83147-0280-481e-a984-c899e9720ec2","recorded_on":"2020-02-24T09:25:06.222986+00:00"}

 {"playhead":1,"metadata":"{\"class\":\"Broadway\\\\Domain\\\\Metadata\",\"payload\":{\"console\":{\"command\":\"Doctrine\\\\Bundle\\\\FixturesBundle\\\\Command\\\\LoadDataFixturesDoctrineCommand\",\"arguments\":\"'doctrine:fixtures:load' --env=prod -n\"}}}","@timestamp":"2020-05-13T18:03:24.430Z","payload":"{\"class\":\"OpenLoyalty\\\\Component\\\\Customer\\\\Domain\\\\Event\\\\AssignedAccountToCustomer\",\"payload\":{\"customerId\":\"00000000-0000-474c-b092-b0dd880c07aa\",\"accountId\":\"3c067099-486a-41a8-87ca-343305126c5e\"}}","@version":"1","id":80,"type":"OpenLoyalty.Component.Customer.Domain.Event.AssignedAccountToCustomer","uuid":"00000000-0000-474c-b092-b0dd880c07aa","recorded_on":"2020-02-24T09:25:20.012550+00:00"}

{"playhead":2,"metadata":"{\"class\":\"Broadway\\\\Domain\\\\Metadata\",\"payload\":{\"console\":{\"command\":\"Doctrine\\\\Bundle\\\\FixturesBundle\\\\Command\\\\LoadDataFixturesDoctrineCommand\",\"arguments\":\"'doctrine:fixtures:load' --env=prod -n\"}}}","@timestamp":"2020-05-13T18:03:24.430Z","payload":"{\"class\":\"OpenLoyalty\\\\Component\\\\Customer\\\\Domain\\\\Event\\\\CustomerWasMovedToLevel\",\"payload\":{\"customerId\":\"00000000-0000-474c-b092-b0dd880c07aa\",\"levelId\":\"e82c96cf-32a3-43bd-9034-4df343e50000\",\"oldLevelId\":null,\"updatedAt\":1582536320,\"manually\":false,\"removeLevelManually\":false}}","@version":"1","id":81,"type":"OpenLoyalty.Component.Customer.Domain.Event.CustomerWasMovedToLevel","uuid":"00000000-0000-474c-b092-b0dd880c07aa","recorded_on":"2020-02-24T09:25:20.014810+00:00"}

and many others with different type values. what I'm looking for is to insert the payloadof each these data into their appropriate index of elasticsearch. for instance data with type = OpenLoyalty.Component.Customer.Domain.Event.AssignedAccountToCustomer has to go into oloy.account_details index. and the index should be like:

    {
    "_index": "oloy.account_details",
    "_type": "OpenLoyalty\Component\Account\Domain\ReadModel\AccountDetails",
    "_id": "AXGjvhDQUCC0J4ATWiCP",
    "_version": 1,
    "_score": 1,
    "_source": {
    "accountId": "5a9bdb83-f7e8-442c-b06f-387e1b1e95a7",
    "customerId": "11111111-0000-474c-b092-b0dd880c07e1"
    }
  }

I wonder what configuration should I set for Logstash. so far I have this:

input {
    jdbc {
        jdbc_connection_string => "jdbc:postgresql://localhost:5432/openloyalty"
        jdbc_user => "openloyalty"
        jdbc_password => "openloyalty"
        jdbc_driver_library => "C:\logstash\postgresql-42.2.12.jar"
        jdbc_driver_class => "org.postgresql.Driver"
        statement => "SELECT * from events"
    }
}

output {
    elasticsearch {
            index => "???"
            document_type => "???"
            document_id => "???"
            hosts => ["localhost:9200"]
    }
}

You can add meta data fields to your table such as target_index and fill this fields in the first place when adding your DB row. then in your logstash config use data from these filed to set dynamic index name and etc. something like index => "%{taget_index}" could work. — Masoud Aghaei
– Masoud Aghaei, Commented May 14, 2020 at 13:30

Nejatians · Accepted Answer · 2020-06-09 13:18:54Z

Your answer is here:

input {
    file{
        path => "C:/Users/Khatere Tajfar/Desktop/Log/application.log"
 start_position => "beginning"
        ignore_older => 0  
    }
}
filter {
    grok {
        match=>{"message"=>"%{GREEDYDATA:part1} INFO %{GREEDYDATA:part2}\{\"indexType\"\:\"%{GREEDYDATA:type}\"\,\"indexName\"\:\"%{GREEDYDATA:index}\"\,\"model\"\:\"%{GREEDYDATA:jsonpart}\"\,\"id\"\:\"%{GREEDYDATA:id}\"\}"}

    }

    mutate {
             remove_field => [ "part1","part2","message","path","host","@timestamp","@version"]
         gsub  => ["jsonpart","[\\\\\\]+","X"]
         gsub  => ["jsonpart","[\\]","X"]
         gsub  => ["jsonpart","X",""]
        }
    json {
        source => "jsonpart"
    }
    mutate {
        remove_field => [ "jsonpart"]   
    }
     mutate { add_field => { "[@metadata][type]" => "%{type}" } }
         mutate { remove_field => ["type"] }

     mutate { add_field => { "[@metadata][index]" => "%{index}" } }
         mutate { remove_field => ["index"] }

     mutate { add_field => { "[@metadata][id]" => "%{id}" } }
         mutate { remove_field => ["id"] }

}
output {
    stdout { codec => json_lines} 
    elasticsearch {
            index =>  "%{[@metadata][index]}" 
            document_type =>  "%{[@metadata][type]}" 
            document_id => "%{[@metadata][id]}" 
            hosts => ["localhost:9200"]
    }
}

Collectives™ on Stack Overflow

Injecting json string into different Elasticsearch indices using Logstash

1 Answer 1

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Related