# Forwarding the http input data to http output data
**URL:** https://discuss.elastic.co/t/forwarding-the-http-input-data-to-http-output-data/375567
**Category:** Logstash
**Created:** [March 7, 2025, 4:41pm UTC](https://discuss.elastic.co/t/forwarding-the-http-input-data-to-http-output-data/375567 "2025-03-07T16:41:31Z")
**Posts on this page:** 13
**Page:** 2
### Author:  [@Francesco\_Esposito](https://discuss.elastic.co/u/Francesco_Esposito)
#### Post date: [March 11, 2025, 6:01pm UTC](https://discuss.elastic.co/t/forwarding-the-http-input-data-to-http-output-data/375567/21 "2025-03-11T18:01:06Z")
```auto
input {
http {
port => 5043
}
}
filter {
mutate {
add_field => {
"full_url1" => "http://localhost:9200%{[headers][request_path]}"
}
add_field => {
"full_url2" => "https://anotherendpoint.com%{[headers][request_path]}"
}
add_field => {
"original_content_type" => "%{[headers][content_type]}"
}
add_field => {
"original_path" => "%{[headers][request_path]}"
}
add_field => {
"request_body" => "%{[message]}"
}
}
}
output {
file {
path => "/log_streaming/my_app/log-%{+yyyy-MM-dd_HH.mm.ss.SSS}.log"
}
http {
http_method=>"post"
format => "message"
url => "%{[full_url1]}"
content_type => "application/json"
message => "%{[request_body]}"
}
http {
http_method=>"post"
format => "message"
url => "%{[full_url2]}"
content_type => "application/json"
message => "%{[request_body]}"
}
}
```
---
### Author:  [@RainTown](https://discuss.elastic.co/u/RainTown)
#### Post date: [March 11, 2025, 8:22pm UTC](https://discuss.elastic.co/t/forwarding-the-http-input-data-to-http-output-data/375567/22 "2025-03-11T20:22:10Z")
Excellent, yes, that seems to work well.
You should likely accept your own answer, and good luck with your project going forward.
---
### Author:  [@Francesco\_Esposito](https://discuss.elastic.co/u/Francesco_Esposito)
#### Post date: [March 12, 2025, 5:04pm UTC](https://discuss.elastic.co/t/forwarding-the-http-input-data-to-http-output-data/375567/23 "2025-03-12T17:04:47Z")
@Badger, I am following the suggestion of moving away from http output plugin to use elasticsearch output plugin.
I have a request coming from http input plugin as explained:
```auto
{
"@timestamp": "2025-03-10T20:16:17.408Z",
"message": "{ \"index\" : { \"_index\" : \"journaling_insert\"}}\r\n{\"REMOTEIP\":\"1.111.1.11\",\"CHAINCODE\":\"8971\",\"EVENTID\":\"16\",\"STOREATTRIBUTE3\":\"Passed Value\",\"DATETIME\":\"2025-03-10T20:16:17.404\",\"STOREATTRIBUTE2\":\"StoreDB Value\",\"STOREATTRIBUTE4\":\"StoreDB Value\",\"STOREATTRIBUTE5\":\"StoreDB Value\",\"FLSECURITY\":{\"SID\":\"1111\"},\"FLCUSTOMER\":{\"FIRSTNAME\":\"Francesco\",\"LASTNAME\":\"Esposito\"}}\r\n{ \"index\" : { \"_index\" : \"journaling_insert\"}}\r\n{\"REMOTEIP\":\"1.111.1.11\",\"CHAINCODE\":\"8971\",\"EVENTID\":\"17\",\"DRAWERIDENT\":\"test\",\"DATETIME\":\"2025-03-10T20:16:17.404\",\"STOREATTRIBUTE2\":\"StoreDB Value\",\"STOREATTRIBUTE3\":\"StoreDB Value\",\"STOREATTRIBUTE4\":\"StoreDB Value\",\"STOREATTRIBUTE5\":\"StoreDB Value\",\"FLTRANSACTIONATTRIBUTES\":{\"INVOICENUMBER\":\"1111\"},\"FLCUSTOMER\":{\"FIRSTNAME\":\"Sam\",\"LASTNAME\":\"Stein\"}}\r\n",
"host": "127.0.0.1",
"headers": {
"cache_control": "no-cache",
"http_version": "HTTP/1.1",
"connection": "Keep-Alive",
"request_method": "POST",
"http_user_agent": "Mozilla/4.0 (compatible; Clever Internet Suite)",
"content_type": "application/x-ndjson; charset=utf-8",
"request_path": "/_bulk",
"http_accept": "*/*",
"accept_encoding": "gzip",
"content_length": "776",
"http_host": "localhost:5043"
},
"@version": "1"
}
```
And I am not able to format the request in right way to feed the elasticsearch output plugin something like:
```auto
{ "index" : { "_index" : "journaling_insert"}}
{"REMOTEIP":"1.111.1.11","CHAINCODE":"8971","EVENTID":"16","STOREATTRIBUTE3":"Passed Value","DATETIME":"2025-03-10T20:16:17.404","STOREATTRIBUTE2":"StoreDB Value","STOREATTRIBUTE4":"StoreDB Value","STOREATTRIBUTE5":"StoreDB Value","FLSECURITY":{"SID":"1111"},"FLCUSTOMER":{"FIRSTNAME":"Francesco","LASTNAME":"Esposito"}}
{ "index" : { "_index" : "journaling_insert"}}
{"REMOTEIP":"1.111.1.11","CHAINCODE":"8971","EVENTID":"17","DRAWERIDENT":"test","DATETIME":"2025-03-10T20:16:17.404","STOREATTRIBUTE2":"StoreDB Value","STOREATTRIBUTE3":"StoreDB Value","STOREATTRIBUTE4":"StoreDB Value","STOREATTRIBUTE5":"StoreDB Value","FLTRANSACTIONATTRIBUTES":{"INVOICENUMBER":"1111"},"FLCUSTOMER":{"FIRSTNAME":"Sam","LASTNAME":"Stein"}}
```
@RainTown I tried your approach, but it doesn't work generating this json, that I output to file:
```auto
{
"DRAWERIDENT": "test",
"STOREATTRIBUTE3": "StoreDB Value",
"@version": "1",
"FLTRANSACTIONATTRIBUTES": {
"INVOICENUMBER": "1111"
},
"DATETIME": "2025-03-11T20:33:51.690",
"REMOTEIP": "1.111.1.11",
"STOREATTRIBUTE5": "StoreDB Value",
"CHAINCODE": "8971",
"STOREATTRIBUTE4": "StoreDB Value",
"EVENTID": "17",
"@timestamp": "2025-03-11T20:33:51.877Z",
"STOREATTRIBUTE2": "StoreDB Value",
"FLCUSTOMER": {
"LASTNAME": "Stein",
"FIRSTNAME": "Sam"
}
}
```
Any suggestion?
---
### Author:  [@Francesco\_Esposito](https://discuss.elastic.co/u/Francesco_Esposito)
#### Post date: [March 12, 2025, 7:34pm UTC](https://discuss.elastic.co/t/forwarding-the-http-input-data-to-http-output-data/375567/24 "2025-03-12T19:34:07Z")
I have made some changes to my application and now I am sending something like this:
```auto
{ "journaling_index": [
{"REMOTEIP":"1.111.1.11","CHAINCODE":"8971","EVENTID":"16","STOREATTRIBUTE3":"Passed Value","DATETIME":"2025-03-07T19:14:58.400","STOREATTRIBUTE2":"StoreDB Value","STOREATTRIBUTE4":"StoreDB Value","STOREATTRIBUTE5":"StoreDB Value","FLSECURITY":{"SID":"1111"},"FLCUSTOMER":{"FIRSTNAME":"Gandalf","LASTNAME":"the Grey"}},
{"REMOTEIP":"1.111.1.11","CHAINCODE":"8971","EVENTID":"17","DRAWERIDENT":"test","DATETIME":"2025-03-07T19:14:58.400","STOREATTRIBUTE2":"StoreDB Value","STOREATTRIBUTE3":"StoreDB Value","STOREATTRIBUTE4":"StoreDB Value","STOREATTRIBUTE5":"StoreDB Value","FLTRANSACTIONATTRIBUTES":{"INVOICENUMBER":"1111"},"FLCUSTOMER":{"FIRSTNAME":"Gandalf","LASTNAME":"the Grey"}}
]}
```
that generates:
```auto
{
"@timestamp": "2025-03-12T19:24:41.307Z",
"@version": "1",
"journaling_index": [
{
"FLSECURITY": {
"SID": "1111"
},
"STOREATTRIBUTE4": "StoreDB Value",
"REMOTEIP": "1.111.1.11",
"CHAINCODE": "8971",
"EVENTID": "16",
"STOREATTRIBUTE5": "StoreDB Value",
"STOREATTRIBUTE3": "Passed Value",
"DATETIME": "2025-03-07T19:14:58.400",
"FLCUSTOMER": {
"LASTNAME": "the Grey",
"FIRSTNAME": "Gandalf"
},
"STOREATTRIBUTE2": "StoreDB Value"
},
{
"STOREATTRIBUTE4": "StoreDB Value",
"CHAINCODE": "8971",
"FLTRANSACTIONATTRIBUTES": {
"INVOICENUMBER": "1111"
},
"STOREATTRIBUTE3": "StoreDB Value",
"DATETIME": "2025-03-07T19:14:58.400",
"DRAWERIDENT": "test",
"STOREATTRIBUTE2": "StoreDB Value",
"REMOTEIP": "1.111.1.11",
"EVENTID": "17",
"STOREATTRIBUTE5": "StoreDB Value",
"FLCUSTOMER": {
"LASTNAME": "the Grey",
"FIRSTNAME": "Gandalf"
}
}
]
}
```
Using the config:
```auto
input {
http {
port => 5043
}
}
filter {
mutate {
remove_field => "headers"
remove_field => "host"
}
json {
source => "message"
remove_field => "message"
}
}
output {
file {
path => "/log_streaming/my_app/log-%{+yyyy-MM-dd_HH.mm.ss.SSS}.log"
}
elasticsearch {
hosts => "localhost:9200"
index => "journaling_insert"
}
}
```
I am still missing how I can use my journaling\_insert as index and the different records as insert record. Could you guys support me in that?
---
### Author:  [@RainTown](https://discuss.elastic.co/u/RainTown)
#### Post date: [March 12, 2025, 7:41pm UTC](https://discuss.elastic.co/t/forwarding-the-http-input-data-to-http-output-data/375567/25 "2025-03-12T19:41:53Z")
> [@Francesco\_Esposito](#):
>
> And I am not able to format the request in right way to feed the elasticsearch output plugin something like:
Different elasticsearch endpoints expect different input. You probably dont need the
```auto
{ "index" : { "_index" : "journaling_insert"}}
```
parts, just the documents. But still, you need do something similar with the http input to push just the documents to the elasticsearch output.
If you change your Delphi application to POST one document at a time to the HTTP endpoint then something like this will work:
```auto
input {
http { host => "0.0.0.0" port => "8080" }
}
filter {
json {
source => "message" remove_field => "message" remove_field => "host" remove_field => "headers" remove_field => "@timestamp" remove_field => "@version"
}
}
output {
file { path => "/tmp/debug.txt" }
elasticsearch { hosts=>["localhost:9200"] index=>"new_index" }
}
```
_NOTE: I am not claiming this is the optimal approach._
Lets go through the 3 phases
input: your document (singular) arrives the HTTP port, here port 8080, path does not matter. That received document is effectively stored into `message` inside a larger document, alongside various other fields like HTTP headers, methods, etc.
filter: this part essentially extracts the original document from the field `message`, and then throws away the message and rest of the fields.
output: this document is added to the (logstash internal) Q of docs to be indexed to elasticsearch. At some later time point (but within a very short time interval) elasticsearch will index all the docs in its current Q into the `new_index` index. Note logstash itself uses the \_bulk endpoint by default.
So I tested the above config this way - I have your original docs stored in a file, the actual docs are lines 2 and 4. So I send the, line 2 and then line 4, to the HTTP input (port 8080 in my case) and they get stored into the index `new_index`.
```auto
$ cat sample.json
{ "index" : { "_index" : "new_index"}}
{"REMOTEIP":"1.111.1.11","CHAINCODE":"8971","EVENTID":"16","STOREATTRIBUTE3":"Passed Value","DATETIME":"2025-03-07T19:14:58.400","STOREATTRIBUTE2":"StoreDB Value","STOREATTRIBUTE4":"StoreDB Value","STOREATTRIBUTE5":"StoreDB Value","FLSECURITY":{"SID":"1111"},"FLCUSTOMER":{"FIRSTNAME":"Gandalf","LASTNAME":"the Grey"}}
{ "index" : { "_index" : "new_index"}}
{"REMOTEIP":"1.111.1.11","CHAINCODE":"8971","EVENTID":"17","DRAWERIDENT":"test","DATETIME":"2025-03-07T19:14:58.400","STOREATTRIBUTE2":"StoreDB Value","STOREATTRIBUTE3":"StoreDB Value","STOREATTRIBUTE4":"StoreDB Value","STOREATTRIBUTE5":"StoreDB Value","FLTRANSACTIONATTRIBUTES":{"INVOICENUMBER":"1111"},"FLCUSTOMER":{"FIRSTNAME":"Gandalf","LASTNAME":"the Grey"}}
$ sed -n 2p sample.json | curl -s http://localhost:8080 --data-binary @- && echo $?
ok0
$ sed -n 4p sample.json | curl -s http://localhost:8080 --data-binary @- && echo $?
ok0
$ curl -s http://localhost:9200/new_index/_search | jq -Sc ".hits.hits[]._source"
{"CHAINCODE":"8971","DATETIME":"2025-03-07T19:14:58.400","EVENTID":"16","FLCUSTOMER":{"FIRSTNAME":"Gandalf","LASTNAME":"the Grey"},"FLSECURITY":{"SID":"1111"},"REMOTEIP":"1.111.1.11","STOREATTRIBUTE2":"StoreDB Value","STOREATTRIBUTE3":"Passed Value","STOREATTRIBUTE4":"StoreDB Value","STOREATTRIBUTE5":"StoreDB Value"}
{"CHAINCODE":"8971","DATETIME":"2025-03-07T19:14:58.400","DRAWERIDENT":"test","EVENTID":"17","FLCUSTOMER":{"FIRSTNAME":"Gandalf","LASTNAME":"the Grey"},"FLTRANSACTIONATTRIBUTES":{"INVOICENUMBER":"1111"},"REMOTEIP":"1.111.1.11","STOREATTRIBUTE2":"StoreDB Value","STOREATTRIBUTE3":"StoreDB Value","STOREATTRIBUTE4":"StoreDB Value","STOREATTRIBUTE5":"StoreDB Value"}
```
I am sure there is a not very complex way to handle multiple documents at once, i.e. sending multiple docs to HTTP endpoint at same time, but ... I didn't have time to figure that out.
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [March 12, 2025, 8:24pm UTC](https://discuss.elastic.co/t/forwarding-the-http-input-data-to-http-output-data/375567/26 "2025-03-12T20:24:26Z")
For the original data format the following might work
```
mutate { gsub => ["message", "\r", ""] }
mutate { split => { "message" => "
" } }
ruby {
code => '
msg = event.get("message")
if msg.is_a? Array
while msg.length > 1 do
clone = event.clone
clone.set("message", msg.shift(2))
new_event_block.call(clone)
end
if msg.length > 0
logger.warn("Content is uneven")
event.set("message", msg)
else
event.cancel
end
end
'
}
if [message][1] {
json { source => "[message][1]" }
json { source => "[message][0]" target => "[@metadata][operation]" }
}
```
and in the output section
```
if [@metadata][index] {
elasticsearch {
index => "%{[@metadata][operation][index][_index]}"
....
```
As noted before, this doesn't handle create, update, etc.
---
### Author:  [@Francesco\_Esposito](https://discuss.elastic.co/u/Francesco_Esposito)
#### Post date: [March 12, 2025, 8:35pm UTC](https://discuss.elastic.co/t/forwarding-the-http-input-data-to-http-output-data/375567/27 "2025-03-12T20:35:10Z")
@Badger , what do you think about the new way I am testing, with a node journaling\_insert that is named as the index I want to send the data to and containing the the data itself?
Could this make a easier to build configuration? Could you help me again?
Sorry to bother you again in creating this configuration, trying to learn.
@RainTown unfortunately, I cannot send one document at the time. My operation needs to be atomic. I was able to work with one document simply sending the json formatted new document and add the index = journaling\_insert in the elasticsearch output. But, unfortunately, that is not usable because the only specification is that the insert needs to be in bulk of all the needed record.
---
### Author:  [@RainTown](https://discuss.elastic.co/u/RainTown)
#### Post date: [March 12, 2025, 8:39pm UTC](https://discuss.elastic.co/t/forwarding-the-http-input-data-to-http-output-data/375567/28 "2025-03-12T20:39:05Z")
> [@Francesco\_Esposito](#):
>
> unfortunately, I cannot send one document at the time. My operation needs to be atomic
someone more knowledgeable than I will need to confirm if logstash guarantees this (atomicity) and, if so, in which scenarios (and specifically if the HTTP input is used). Also in context of persistent queues.
> [@Francesco\_Esposito](#):
>
> that is not usable because the only specification is that the insert needs to be in bulk of all the needed record
Plus I didn't _think_ elasticsearch itself guarantees atomicity anyways, so ...
---
### Author:  [@Francesco\_Esposito](https://discuss.elastic.co/u/Francesco_Esposito)
#### Post date: [March 12, 2025, 8:43pm UTC](https://discuss.elastic.co/t/forwarding-the-http-input-data-to-http-output-data/375567/29 "2025-03-12T20:43:15Z")
I know and, really really, thank you for all the support, you have been great helping me getting me started.
About this problem, I am just thinking that if I was able to do it thru http output, I could be able to do it thru elasticsearch output, more if bulk is the standard api used. Let's hope!
---
### Author:  [@Francesco\_Esposito](https://discuss.elastic.co/u/Francesco_Esposito)
#### Post date: [March 12, 2025, 9:56pm UTC](https://discuss.elastic.co/t/forwarding-the-http-input-data-to-http-output-data/375567/30 "2025-03-12T21:56:47Z")
@Badger , what do you think about the new way I am testing, with a node journaling\_insert that is named as the index I want to send the data to and containing the the data itself?
Could this make a easier to build configuration? Could you help me again?
Sorry to bother you again in creating this configuration, trying to learn.
---
### Author:  [@Francesco\_Esposito](https://discuss.elastic.co/u/Francesco_Esposito)
#### Post date: [March 13, 2025, 8:32pm UTC](https://discuss.elastic.co/t/forwarding-the-http-input-data-to-http-output-data/375567/31 "2025-03-13T20:32:43Z")
I am closing this thread with a specific one to solve my problem using http input and elasticsearch output.
---
### Author:  [@RainTown](https://discuss.elastic.co/u/RainTown)
#### Post date: [March 13, 2025, 10:25pm UTC](https://discuss.elastic.co/t/forwarding-the-http-input-data-to-http-output-data/375567/32 "2025-03-13T22:25:40Z")
Thats fine. But can I just remind you that you wrote:
> [@Francesco\_Esposito](#):
>
> My operation needs to be atomic. I was able to work with one document simply sending the json formatted new document and add the index = journaling\_insert in the elasticsearch output. But, unfortunately, that is not usable because the only specification is that the insert needs to be in bulk of all the needed record.
and I replied
> [@RainTown](#):
>
> I didn't _think_ elasticsearch itself guarantees atomicity anyways
To quote from an older thread here:
> All items of a bulk request are indexed individually, so failures will just affect parts of the bulk request. The success or failure of individual documents is reported in the response. You could therefore have e.g. 3 updates to the same document in a bulk request and have only the second one fail. There is no transactions or atomicity.
This is nothing to do with your http/elasticsearch/logstash issue reported here. But unless I'm wrong or misunderstanding what you wrote, a lack of guaranteed atomicity is likely something you should consider how to mitigate. Not for this thread, but ... you maybe have a model based on a incorrect assumption. You probably don't want to find out the downsides of that the hard way.
Good luck with your project.
---
### Author:  [@Francesco\_Esposito](https://discuss.elastic.co/u/Francesco_Esposito)
#### Post date: [March 14, 2025, 3:24pm UTC](https://discuss.elastic.co/t/forwarding-the-http-input-data-to-http-output-data/375567/33 "2025-03-14T15:24:47Z")
Just to clarify, what I meant as "atomic" is that I need to send records to Logstash in one shot. When I have received the 200 OK response, I am done with my job, after that Logstash will need to take care of my packet.
[Previous page](https://discuss.elastic.co/t/forwarding-the-http-input-data-to-http-output-data/375567.md?page=1)