# Logstash wrapping the data with document

**URL:** <https://discuss.elastic.co/t/logstash-wrapping-the-data-with-document/344135>\
**Category:** Logstash\
**Created:** [September 29, 2023, 12:13pm UTC](https://discuss.elastic.co/t/logstash-wrapping-the-data-with-document/344135 "2023-09-29T12:13:46Z")\
**Posts on this page:** 6\
**Page:** 1

<div class="post-metadata">

**Author:** ![Keremcan\_Seker](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/keremcan_seker/32/126097_2.png) [@Keremcan\_Seker](https://discuss.elastic.co/u/Keremcan_Seker)\
**Post date:** [September 29, 2023, 12:13pm UTC](https://discuss.elastic.co/t/logstash-wrapping-the-data-with-document/344135/1 "2023-09-29T12:13:46Z")

</div>

I'm sending data with python on both kafka and http request however kafka pipeline wraps the data with "document". the data on the top is coming from kafka pipeline and the below from http pipeline

 ![image](https://us1.discourse-cdn.com/elastic/original/3X/4/2/42a7c3a9788be6e409978183c1069fa8f735847f.png)

And these are my client codes

```auto
from confluent_kafka import Producer

import json

try:
    # producer = Producer({'bootstrap.servers': '192.168.:29092'})
    producer = Producer({'bootstrap.servers': 'localhost:9092'})
except Exception as e:
    print(e)

def send_message(topic, message):
    try:
        message = json.dumps(message)
        # print(message)
        producer.produce(topic, message)

        producer.flush()
        print("Message sent successfully")
    except Exception as e:
        print(e)

data = {
    "service_name": "kafka-service",
    "activity_type": "info",
    "error": {
        "component_id": "1234",
        "message": "dogru index girildi",
                
    },
    "additional_data": {
        "key": "value"
    }
}

# print(json.dumps(data.get('error')))
send_message('kafka-activity-log', data)

```

```auto
import requests
import json

# Logstash HTTP endpoint
LOGSTASH_URL = 'http://localhost:8080' # Replace with your Logstash endpoint URL

# Data to be sent to Logstash
data = {
    "user_id": "user123",
    "service_name": "example_service",
    "activity_type": "info",
    "error": {
        "component_id": "1234",
        "message": "dogru index girildi",
                
    },
    "additional_data": {
        "key": "now1"
    }
}

# Convert data to JSON format
json_data = json.dumps(data)

# Set up HTTP headers
headers = {
    'Content-Type': 'application/json'
}

# Send POST request to Logstash
response = requests.post(LOGSTASH_URL, headers=headers, data=json_data)

# Check the response
if response.status_code == 200:
    print("Data sent to Logstash successfully!")
else:
    print(f"Failed to send data to Logstash. Status code: {response.status_code}")
    print(response.text)

```

this is my pipeline they both are the same just with different inputs

```auto
 input {
    kafka {
      bootstrap_servers => "kafka:29092"
      topics => ["kafka-activity-log"]
      codec => json {
        target => "[document]"
      }
    } 
    }

    filter{
      json {
        source => "message"
        target => "parsed_json"
      }

    if "_jsonparsefailure" not in [tags] {
      mutate {
        remove_field => ["message", "@version", "host", "url", "event", "user_agent"]
      }

      ruby {
        code => '
          empty_fields = 0
          ["user_id", "service_name", "activity_type"].each do |field|
            if event.get(field).to_s.empty?
              empty_fields += 1
            end
          end

          if empty_fields >= 2
            event.set("[@metadata][index]", "user_activity_log_missing_data")
          elsif event.get("[user_id]").to_s.empty?
            event.set("[@metadata][index]", "user_activity_log_empty_user_id")
          elsif event.get("[service_name]").to_s.empty?
            event.set("[@metadata][index]", "user_activity_log_empty_service_name")
          elsif event.get("[activity_type]").to_s.empty?
            event.set("[@metadata][index]", "user_activity_log_empty_activity_type")
          elsif event.get("[error][message]").to_s.empty?
            event.set("[@metadata][index]", "user_activity_log_empty_error_message")
          elsif event.get("[error][component_id]").to_s.empty?
            event.set("[@metadata][index]", "user_activity_log_empty_component_id")
          else
            event.set("[@metadata][index]", "user_activity_log")
          end
        '
      }
    } else {
      ruby {
        code => '
          event.set("[@metadata][index]", "user_activity_log_invalid_data")
        '
      }
    }
    }

```

how can i prevent this so it does not wrap the element with document

---

<div class="post-metadata">

**Author:** ![leandrojmp](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/leandrojmp/32/107231_2.png) [@leandrojmp](https://discuss.elastic.co/u/leandrojmp)\
**Post date:** [September 29, 2023, 12:32pm UTC](https://discuss.elastic.co/t/logstash-wrapping-the-data-with-document/344135/2 "2023-09-29T12:32:11Z")

</div>

> [@Keremcan\_Seker](#):
>
> how can i prevent this so it does not wrap the element with document

It is your kafka config that is putting the data from kafka inside the `document` field.

```auto
      codec => json {
        target => "[document]"
      }

```

The target option does exactly what you do not want it to do, it will parse your json and put it as nested fields inside `document`, if you do not want it, remove the target option, use just `codec => json`.

---

<div class="post-metadata">

**Author:** ![Keremcan\_Seker](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/keremcan_seker/32/126097_2.png) [@Keremcan\_Seker](https://discuss.elastic.co/u/Keremcan_Seker)\
**Post date:** [September 29, 2023, 12:35pm UTC](https://discuss.elastic.co/t/logstash-wrapping-the-data-with-document/344135/3 "2023-09-29T12:35:52Z")

</div>

thanks a lot !  
but my http pipeline is configured the same

```auto
codec => json {
      target => "[document]" # Parse the JSON into the [document] field
    }

```

why it does not do that ?

---

<div class="post-metadata">

**Author:** ![leandrojmp](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/leandrojmp/32/107231_2.png) [@leandrojmp](https://discuss.elastic.co/u/leandrojmp)\
**Post date:** [September 29, 2023, 12:53pm UTC](https://discuss.elastic.co/t/logstash-wrapping-the-data-with-document/344135/4 "2023-09-29T12:53:59Z")

</div>

> [@Keremcan\_Seker](#):
>
> why it does not do that ?

It is a bug, the http input per default will expect the massages as `json`, so if your message is a `json` it will basically ignore anything you put in the `codec` option.

Check this [answer](https://discuss.elastic.co/t/json-codec-plugin-target-option-http-input/304217/4) about it.

---

<div class="post-metadata">

**Author:** ![Keremcan\_Seker](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/keremcan_seker/32/126097_2.png) [@Keremcan\_Seker](https://discuss.elastic.co/u/Keremcan_Seker)\
**Post date:** [September 29, 2023, 1:25pm UTC](https://discuss.elastic.co/t/logstash-wrapping-the-data-with-document/344135/5 "2023-09-29T13:25:13Z")

</div>

i did not know that again thanks a lot for your help

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [October 27, 2023, 1:26pm UTC](https://discuss.elastic.co/t/logstash-wrapping-the-data-with-document/344135/6 "2023-10-27T13:26:06Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.