Creating JSON structure for sensor.community API

Elastic Stack Logstash
1

I will send data from my logstash pipeline to the sensor.community API. The API requires the following structure which works with my curl command:

curl --location --request POST 'https://api.sensor.community/v1/push-sensor-data/' --header 'Content-Type:application/json' --header 'X-Pin:15' --header 'X-Sensor:esp32-REMOVED' --data-raw '{"software_version": "1.0", "sensordatavalues":[{"value_type":"noise_LAeq","value":"21.21"},{"value_type":"noise_LA_max","value":"22.22"}, {"value_type":"noise_LA_min","value":"23.23"}]}'

Currently I use the following pipeline to build the structure for the first parameter but it does not work:

input { pipeline { address => "sensor.community-noise" } }

filter {
  # process data for sensor.community
    mutate {
      add_field => [ "[@metadata][app_id]", "%{[end_device_ids][application_ids][application_id]}" ]
      add_field => [ "[@metadata][device_id]", "%{[end_device_ids][device_id]}" ]
    } # mutate
  # 
  if ([@metadata][device_id] == "akiot-sound01") {
    mutate {
      add_field => { "[@metadata][send_data]" => "true" }
      add_field => { "[@metadata][sensor]" => "esp32-REMOVED" }
    }
    mutate {
      add_field => { "software_version" => "1.0" } 
      add_field => { "[sensordatavalues][value_type]" => "noise_LAeq"} 
      add_field => { "[sensordatavalues][value]"  => "%{[uplink_message][decoded_payload][la][avg]}"} 
    }
  }

  mutate { remove_field => ["app_id","@version","@timestamp","host","received_at","_source","headers","correlation_ids","uplink_message","end_device_ids"] }
} # filter

output {
  if ([@metadata][app_id] == "adlerkiez-soundsensors" and [@metadata][send_data] == "true") {
#    http {
#     url => "https://api.sensor.community/v1/push-sensor-data/"
#     http_method => "post"
#     format => "json"
#     headers => [
#      'X-Pin:15',
#      'X-Sensor:%{[@metadata][sensor]}'
#     ]
#    } # http

    file {
      path => "/tmp/sensor.community-%{[@metadata][device_id]}.log"

    }

  stdout { 
    codec => rubydebug { metadata => true } 
  }
 } # if [@metadata][app_id] 
} # output

That pipeline generates the following output in the file:

{"software_version":"1.0","sensordatavalues":{"value_type":"noise_LAeq","value":"46.94"}}

How can I create the right JSON structure with all key/value pairs in sensordatavalues?

2

Unless we know what the original event looks like I don't think we can tell you how to mutate it to get the output format that you want.

3

Here is the JSON document from my Kibana GUI, which was processed by the second Elasticsearch pipeline:

{
  "_index": "akiot-2023.11.05",
  "_type": "_doc",
  "_id": "bsuIoIsBpHce2ysqZGXl",
  "_version": 1,
  "_score": 1,
  "_source": {
    "lz.avg": "57.07",
    "end_device_ids": {
      "device_id": "akiot-sound01",
      "dev_addr": "260BBDCE",
      "application_ids": {
        "application_id": "adlerkiez-soundsensors"
      },
      "dev_eui": "0000480F5BBD9E7C",
      "join_eui": "0000000000000000"
    },
    "lc.max": "77",
    "devid": "akiot-sound01",
    "received_at": "2023-11-05T17:29:34.968108274Z",
    "la.min": "30.8",
    "host": "127.0.0.1",
    "app_id": "adlerkiez-soundsensors",
    "@version": "1",
    "la.max": "75.49",
    "@timestamp": "2023-11-05T17:29:35.091Z",
    "correlation_ids": [
      "gs:uplink:01HEG8GRSA8CE647H6PFK720H0"
    ],
    "headers": {
      "http_version": "HTTP/1.1",
      "x_forwarded_server": "elk-logstash.cargobikometer.de",
      "http_host": "localhost:8080",
      "accept_encoding": "gzip",
      "content_type": "application/json",
      "http_accept": null,
      "request_path": "/",
      "x_forwarded_for": "34.255.49.188",
      "x_forwarded_host": "elk-logstash.cargobikometer.de",
      "http_user_agent": "TheThingsStack/3.28.0-rc2-SNAPSHOT-ee7fad721 (linux/arm64)",
      "x_tts_domain": "eu1.cloud.thethings.network",
      "content_length": "2230",
      "request_method": "POST",
      "connection": "Keep-Alive"
    },
    "la.avg": "54.35",
    "uplink_message": {
      "rx_metadata": [
        {
          "rssi": -109,
          "snr": -12,
          "time": "2023-11-05T17:29:34Z",
          "timestamp": 328770964,
          "channel_rssi": -109,
          "gateway_ids": {
            "eui": "647FDAFFFE010D7D",
            "gateway_id": "berlin-schlossstrasse"
          },
          "received_at": "2023-11-05T17:29:34.737379914Z",
          "uplink_token": "CiMKIQoVYmVybGluLXNjaGxvc3NzdHJhc3NlEghkf9r//gENfRCUy+KcARoMCP6hn6oGEPqOz+sCIKCUm+LI4N0B",
          "location": {
            "latitude": 52.43657133897713,
            "longitude": 13.549353182315828,
            "source": "SOURCE_REGISTRY"
          }
        },
        {
          "rssi": -115,
          "snr": -2.25,
          "time": "2023-11-05T17:29:34.739227056Z",
          "timestamp": 3399229380,
          "channel_rssi": -115,
          "gateway_ids": {
            "eui": "FCC23DFFFE0B987C",
            "gateway_id": "adlerkiez-iot-lorix-one"
          },
          "received_at": "2023-11-05T17:29:34.520085821Z",
          "uplink_token": "CiUKIwoXYWRsZXJraWV6LWlvdC1sb3JpeC1vbmUSCPzCPf/+C5h8EMS/8NQMGgwI/qGfqgYQv5K+8AIgoKvwjvdi",
          "location": {
            "latitude": 52.437863163422385,
            "longitude": 13.551034927368166,
            "altitude": 50,
            "source": "SOURCE_REGISTRY"
          }
        }
      ],
      "frm_payload": "TWb6tHf/vXj/vXR6hqS1pI9zZg==",
      "consumed_airtime": "0.133632s",
      "version_ids": {
        "band_id": "EU_863_870",
        "model_id": "wifi-lora-32-class-a-otaa",
        "firmware_version": "1.0",
        "brand_id": "heltec",
        "hardware_version": "_unknown_hw_version_"
      },
      "received_at": "2023-11-05T17:29:34.763231951Z",
      "f_cnt": 294,
      "session_key_id": "AYuekdbBmIKq1liti9q31Q==",
      "network_ids": {
        "ns_id": "EC656E0000000181",
        "tenant_id": "ttn",
        "net_id": "000013",
        "cluster_address": "eu1.cloud.thethings.network",
        "cluster_id": "eu1"
      },
      "settings": {
        "data_rate": {
          "lora": {
            "spreading_factor": 8,
            "bandwidth": 125000,
            "coding_rate": "4/5"
          }
        },
        "time": "2023-11-05T17:29:34Z",
        "frequency": "867900000",
        "timestamp": 328770964
      },
      "f_port": 22,
      "decoded_payload": {
        "lc": {
          "avg": 57.07,
          "spectrum": [
            32.03,
            36.04,
            40.26,
            49.52,
            54.65,
            49.52,
            43.38,
            35.03,
            27.8
          ],
          "max": 77,
          "min": 35.93
        },
        "la": {
          "avg": 54.35,
          "spectrum": [
            -4.37,
            10.64,
            24.36,
            40.92,
            51.45,
            49.52,
            44.38,
            35.73,
            29.7
          ],
          "max": 75.49,
          "min": 30.8
        },
        "lz": {
          "avg": 57.07,
          "spectrum": [
            35.03,
            36.84,
            40.46,
            49.52,
            54.65,
            49.52,
            43.18,
            34.73,
            30.8
          ],
          "max": 77,
          "min": 36.24
        }
      }
    },
    "lc.avg": "57.07",
    "lz.max": "77",
    "lz.min": "36.24",
    "dev_id": "akiot-sound01",
    "lc.min": "35.93"
  },
  "fields": {
    "headers.connection": [
      "Keep-Alive"
    ],
    "uplink_message.decoded_payload.la.max": [
      75.49
    ],
    "uplink_message.rx_metadata.location.altitude": [
      50
    ],
    "uplink_message.rx_metadata.gateway_ids.eui": [
      "647FDAFFFE010D7D",
      "FCC23DFFFE0B987C"
    ],
    "uplink_message.rx_metadata.location.latitude": [
      52.43657133897713,
      52.437863163422385
    ],
    "headers.x_forwarded_host": [
      "elk-logstash.cargobikometer.de"
    ],
    "uplink_message.session_key_id": [
      "AYuekdbBmIKq1liti9q31Q=="
    ],
    "uplink_message.rx_metadata.channel_rssi": [
      -109,
      -115
    ],
    "headers.request_path": [
      "/"
    ],
    "end_device_ids.device_id": [
      "akiot-sound01"
    ],
    "lc.min": [
      35
    ],
    "uplink_message.decoded_payload.la.spectrum": [
      -4.37,
      10.64,
      24.36,
      40.92,
      51.45,
      49.52,
      44.38,
      35.73,
      29.7
    ],
    "lc.avg": [
      57
    ],
    "uplink_message.network_ids.ns_id": [
      "EC656E0000000181"
    ],
    "uplink_message.f_cnt": [
      294
    ],
    "host": [
      "127.0.0.1"
    ],
    "headers.http_version": [
      "HTTP/1.1"
    ],
    "uplink_message.rx_metadata.uplink_token": [
      "CiMKIQoVYmVybGluLXNjaGxvc3NzdHJhc3NlEghkf9r//gENfRCUy+KcARoMCP6hn6oGEPqOz+sCIKCUm+LI4N0B",
      "CiUKIwoXYWRsZXJraWV6LWlvdC1sb3JpeC1vbmUSCPzCPf/+C5h8EMS/8NQMGgwI/qGfqgYQv5K+8AIgoKvwjvdi"
    ],
    "headers.x_forwarded_server": [
      "elk-logstash.cargobikometer.de"
    ],
    "uplink_message.version_ids.model_id": [
      "wifi-lora-32-class-a-otaa"
    ],
    "lz.max": [
      77
    ],
    "app_id": [
      "adlerkiez-soundsensors"
    ],
    "devid.keyword": [
      "akiot-sound01"
    ],
    "headers.x_tts_domain": [
      "eu1.cloud.thethings.network"
    ],
    "uplink_message.rx_metadata.timestamp": [
      328770964,
      3399229380
    ],
    "uplink_message.settings.data_rate.lora.bandwidth": [
      125000
    ],
    "uplink_message.rx_metadata.location.longitude": [
      13.549353182315828,
      13.551034927368166
    ],
    "uplink_message.frm_payload": [
      "TWb6tHf/vXj/vXR6hqS1pI9zZg=="
    ],
    "uplink_message.decoded_payload.lz.avg": [
      57.07
    ],
    "uplink_message.decoded_payload.lz.min": [
      36.24
    ],
    "uplink_message.rx_metadata.snr": [
      -12,
      -2.25
    ],
    "uplink_message.version_ids.hardware_version": [
      "_unknown_hw_version_"
    ],
    "end_device_ids.dev_addr": [
      "260BBDCE"
    ],
    "uplink_message.rx_metadata.received_at": [
      "2023-11-05T17:29:34.737379914Z",
      "2023-11-05T17:29:34.520085821Z"
    ],
    "uplink_message.settings.time": [
      "2023-11-05T17:29:34Z"
    ],
    "headers.content_type": [
      "application/json"
    ],
    "uplink_message.decoded_payload.la.avg": [
      54.35
    ],
    "uplink_message.decoded_payload.la.min": [
      30.8
    ],
    "lc.max": [
      77
    ],
    "uplink_message.received_at": [
      "2023-11-05T17:29:34.763231951Z"
    ],
    "uplink_message.decoded_payload.lz.max": [
      77
    ],
    "uplink_message.decoded_payload.lc.spectrum": [
      32.03,
      36.04,
      40.26,
      49.52,
      54.65,
      49.52,
      43.38,
      35.03,
      27.8
    ],
    "devid": [
      "akiot-sound01"
    ],
    "correlation_ids": [
      "gs:uplink:01HEG8GRSA8CE647H6PFK720H0"
    ],
    "uplink_message.network_ids.cluster_address": [
      "eu1.cloud.thethings.network"
    ],
    "uplink_message.settings.data_rate.lora.spreading_factor": [
      8
    ],
    "la.max": [
      75
    ],
    "headers.http_host": [
      "localhost:8080"
    ],
    "uplink_message.decoded_payload.lc.avg": [
      57.07
    ],
    "uplink_message.decoded_payload.lc.min": [
      35.93
    ],
    "uplink_message.settings.frequency": [
      "867900000"
    ],
    "uplink_message.version_ids.firmware_version": [
      "1.0"
    ],
    "@version": [
      "1"
    ],
    "uplink_message.decoded_payload.lz.spectrum": [
      35.03,
      36.84,
      40.46,
      49.52,
      54.65,
      49.52,
      43.18,
      34.73,
      30.8
    ],
    "uplink_message.network_ids.net_id": [
      "000013"
    ],
    "uplink_message.rx_metadata.gateway_ids.gateway_id": [
      "berlin-schlossstrasse",
      "adlerkiez-iot-lorix-one"
    ],
    "uplink_message.rx_metadata.time": [
      "2023-11-05T17:29:34Z",
      "2023-11-05T17:29:34.739227056Z"
    ],
    "la.min": [
      30
    ],
    "la.avg": [
      54
    ],
    "uplink_message.f_port": [
      22
    ],
    "uplink_message.network_ids.tenant_id": [
      "ttn"
    ],
    "end_device_ids.application_ids.application_id": [
      "adlerkiez-soundsensors"
    ],
    "uplink_message.network_ids.cluster_id": [
      "eu1"
    ],
    "uplink_message.decoded_payload.lc.max": [
      77
    ],
    "end_device_ids.dev_eui": [
      "0000480F5BBD9E7C"
    ],
    "uplink_message.version_ids.brand_id": [
      "heltec"
    ],
    "uplink_message.rx_metadata.location.source": [
      "SOURCE_REGISTRY",
      "SOURCE_REGISTRY"
    ],
    "lz.avg": [
      57
    ],
    "uplink_message.rx_metadata.rssi": [
      -109,
      -115
    ],
    "lz.min": [
      36
    ],
    "headers.x_forwarded_for": [
      "34.255.49.188"
    ],
    "headers.accept_encoding": [
      "gzip"
    ],
    "uplink_message.consumed_airtime": [
      "0.133632s"
    ],
    "dev_id": [
      "akiot-sound01"
    ],
    "headers.http_user_agent": [
      "TheThingsStack/3.28.0-rc2-SNAPSHOT-ee7fad721 (linux/arm64)"
    ],
    "headers.content_length": [
      "2230"
    ],
    "@timestamp": [
      "2023-11-05T17:29:35.091Z"
    ],
    "uplink_message.settings.timestamp": [
      328770964
    ],
    "received_at": [
      "2023-11-05T17:29:34.968108274Z"
    ],
    "headers.request_method": [
      "POST"
    ],
    "uplink_message.settings.data_rate.lora.coding_rate": [
      "4/5"
    ],
    "uplink_message.version_ids.band_id": [
      "EU_863_870"
    ],
    "end_device_ids.join_eui": [
      "0000000000000000"
    ]
  }
}

I hope you can find the required information. The input comes from the LoRaSoundkit sensor, which will be forwarded with the TTN http integration to my ELK cluster.

4

With the following add_field commands I get following output, but it is still wrong:

    mutate {
      add_field => { "software_version" => "1.0" } 
      #add_field => { "sensordatavalues" => { "value_type" => "noise_LAeq" "value"  => "%{[uplink_message][decoded_payload][la][avg]}" } 
      add_field => { "sensordatavalues" => [
            { "value_type" =>  "noise_LAeq"  "value" =>  "%{[uplink_message][decoded_payload][la][avg]}" }, 
            { "value_type" =>  "noise_LA_min"  "value" =>  "%{[uplink_message][decoded_payload][la][min]}" }, 
            { "value_type" =>  "noise_LA_max"  "value" =>  "%{[uplink_message][decoded_payload][la][max]}" } 
                     ]
      }

output:

{"software_version":"1.0","sensordatavalues":["{\"value_type\"=>\"noise_LAeq\", \"value\"=>\"44.69\"}","{\"value_type\"=>\"noise_LA_min\", \"value\"=>\"30.76\"}","{\"value_type\"=>\"noise_LA_max\", \"value\"=>\"68.78\"}"]}

But the line contains no key/value parameters. It should look like:

{"software_version":"1.0","sensordatavalues":[{"value_type":"noise_LAeq","value":"44.69"},{"value_type":"noise_LA_min","value":"30.76"},{"value_type":"noise_LA_max","value":"68.78"}]}
5

You need to share the event Logstash is receiving.

You shared a logstash pipeline with a pipeline input, but where is the configuration that directs data to this pipeline?

To be able to replicate your configuration you would need to share the source input you are receiving.

6

This are my configuration files:
cat ../pipelines.yml

# This file is where you define your pipelines. You can define multiple.
# For more information on multiple pipelines, see the documentation:
#   https://www.elastic.co/guide/en/logstash/current/multiple-pipelines.html

- pipeline.id: input
  queue.type: persisted
  path.config: "/etc/logstash/conf.d/input.conf"

- pipeline.id: buffered-es
  queue.type: persisted
  path.config: "/etc/logstash/conf.d/es.conf"

- pipeline.id: buffered-opensensemap-sound
  queue.type: persisted
  path.config: "/etc/logstash/conf.d/opensensemap-sound.conf"

- pipeline.id: buffered-sensor-community-noise
  queue.type: persisted
  path.config: "/etc/logstash/conf.d/sensor.community-noise.conf"

/etc/logstash/conf.d# cat input.conf

input {
  http {
    # get data from TTN integrations or from WZePaperDisplays
    host => "127.0.0.1" # default: 0.0.0.0
    port => 8080 # default: 8080
  }
} # input

#output { pipeline { send_to => ["internal-es", "opensensemap-sound"] } }
output { pipeline { send_to => ["internal-es", "opensensemap-sound", "sensor.community-noise"] } }
#output { pipeline { send_to => ["internal-es"] } }

Are that the right information? Is it possible to append the input event to a file during processing in the http input plugin?

7

You need to share the data received by this input, the source input before it being sent to the other pipelines.

Add a file output in this configuration to save some sample of what is being received.

8

I added a file output and get the following:

output { 
  pipeline { send_to => ["internal-es", "opensensemap-sound", "sensor.community-noise"] } 
  file { path => "/tmp/logstash-http-input.log" }

}


cat /tmp/logstash-http-input.log 
{"@timestamp":"2023-11-06T14:49:06.251Z","correlation_ids":["gs:uplink:01HEJHQMKG7MSKRPQQ9HTEJ669"],"host":"127.0.0.1","received_at":"2023-11-06T14:49:06.110358653Z","headers":{"http_version":"HTTP/1.1","http_host":"localhost:8080","http_accept":null,"x_forwarded_server":"elk-logstash.cargobikometer.de","accept_encoding":"gzip","x_forwarded_host":"elk-logstash.cargobikometer.de","x_tts_domain":"eu1.cloud.thethings.network","connection":"Keep-Alive","content_length":"1804","x_forwarded_for":"63.34.43.96","content_type":"application/json","request_method":"POST","request_path":"/","http_user_agent":"TheThingsStack/3.28.0-rc2-SNAPSHOT-ee7fad721 (linux/arm64)"},"end_device_ids":{"join_eui":"0000000000000000","application_ids":{"application_id":"adlerkiez-soundsensors"},"device_id":"akiot-sound01","dev_eui":"0000480F5BBD9E7C","dev_addr":"260BB027"},"uplink_message":{"frm_payload":"R3Pwnnr+snz/tIKnmJmHf4aGdw==","received_at":"2023-11-06T14:49:05.905163847Z","session_key_id":"AYukbTm4x2X8ANYDawZTxw==","rx_metadata":[{"uplink_token":"CiUKIwoXYWRsZXJraWV6LWlvdC1sb3JpeC1vbmUSCPzCPf/+C5h8EMT20MsDGgwI4fmjqgYQ5sairwMgoIOe8IYc","received_at":"2023-11-06T14:49:05.872423590Z","snr":8.5,"channel_rssi":-94,"gateway_ids":{"eui":"FCC23DFFFE0B987C","gateway_id":"adlerkiez-iot-lorix-one"},"rssi":-94,"time":"2023-11-06T14:49:05.844245910Z","location":{"altitude":50,"longitude":13.551034927368166,"latitude":52.437863163422385,"source":"SOURCE_REGISTRY"},"timestamp":963918660}],"consumed_airtime":"0.071936s","version_ids":{"model_id":"wifi-lora-32-class-a-otaa","hardware_version":"_unknown_hw_version_","band_id":"EU_863_870","brand_id":"heltec","firmware_version":"1.0"},"network_ids":{"ns_id":"EC656E0000000181","tenant_id":"ttn","net_id":"000013","cluster_id":"eu1","cluster_address":"eu1.cloud.thethings.network"},"f_port":22,"f_cnt":104,"decoded_payload":{"lz":{"spectrum":[36.2,46.5,42.32,42.6,37.59,35.36,37.31,37.31,33.13],"avg":50.12,"max":71,"min":34.53},"lc":{"spectrum":[33.2,45.7,42.12,42.6,37.59,35.36,37.51,37.61,30.13],"avg":49.56,"max":70.72,"min":33.97},"la":{"spectrum":[-3.2,20.3,26.22,34,34.39,35.36,38.51,38.31,32.03],"avg":43.99,"max":66.82,"min":32.02}},"settings":{"frequency":"867900000","data_rate":{"lora":{"spreading_factor":7,"bandwidth":125000,"coding_rate":"4/5"}},"time":"2023-11-06T14:49:05.844245910Z","timestamp":963918660}},"@version":"1"}

Hope that this is what you need.

9

Today I found a solution.

    mutate {
      add_field => { "software_version" => "1.0" } 
    }

    ruby {
        code => '
          event.set("sensordatavalues", [ 
            {"value" => event.get("[uplink_message][decoded_payload][la][avg]"),"value_type" => "noise_LAeq"},
            {"value" => event.get("[uplink_message][decoded_payload][la][min]"),"value_type" => "noise_LA_min"},
            {"value" => event.get("[uplink_message][decoded_payload][la][max]"),"value_type" => "noise_LA_max"} 
          ] )
        '
    }

This generates the following output:

{"software_version":"1.0","sensordatavalues":[{"value_type":"noise_LAeq","value":50.19},{"value_type":"noise_LA_min","value":32.22},{"value_type":"noise_LA_max","value":75.59}]}
10

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