Impossible to fetch data from kafka to telegraf

Viewed 23

I have a personal project to gather data from an earthquake monitoring API and put it in real time on a kafka topic and then retrieve it on grafana.

For that I use telegraf and this plugin allowing to listen and to recover data of a topic, before sending them to an influxdb bucket and use it as a source for grafana

I did some tests and I know for sure that the data is received on the topic.

{
  "Earthquake": {
    "metadata": {
      "Metadata": {
        "generated": "1662533868000",
        "url": "https://earthquake.usgs.gov/fdsnws/event/1/query?format=geojson&starttime=2022-09-07T05:57:47",
        "title": "USGS Earthquakes",
        "api": "1.13.6",
        "count": "7",
        "status": "200"
      }
    },
    "features": [
      {
        "Features": {
          "type": "Feature",
          "properties": {
            "Properties": {
              "mag": "1.7",
              "place": "34 km NE of Paxson, Alaska",
              "time": "1662532868655",
              "updated": "1662532994982",
              "tz": "0",
              "url": "https://earthquake.usgs.gov/earthquakes/eventpage/ak022bhk6641",
              "detail": "https://earthquake.usgs.gov/fdsnws/event/1/query?eventid=ak022bhk6641&format=geojson",
              "felt": "0",
              "cdi": "0.0",
              "mmi": "0.0",
              "alert": "null",
              "status": "automatic",
              "tsunami": "0",
              "sig": "44",
              "net": "ak",
              "code": "022bhk6641",
              "ids": ",ak022bhk6641,",
              "sources": ",ak,",
              "types": ",origin,phase-data,",
              "nst": "0",
              "dmin": "0.0",
              "rms": "0.6",
              "gap": "0.0",
              "magType": "ml",
              "type": "earthquake"
            }
          },
          "geometry": {
            "Geometry": {
              "type": "Point",
              "coordinates": [
                -145.1808,
                63.3242,
                0.0
              ]
            }
          },
          "id": "ak022bhk6641"
        }
      }
    ],
    "bbox": [
      -151.5653,
      -54.375,
      0.0,
      -118.39484,
      63.3242,
      71.4
    ]
  }
}

Here is my telegraf.conf that i use for my influxdb data source:

# Telegraf Configuration
#
# Telegraf is entirely plugin driven. All metrics are gathered from the
# declared inputs, and sent to the declared outputs.
#
# Plugins must be declared in here to be active.
# To deactivate a plugin, comment out the name and any variables.
#
# Use 'telegraf -config telegraf.conf -test' to see what metrics a config
# file would generate.
#
# Environment variables can be used anywhere in this config file, simply surround
# them with ${}. For strings the variable must be within quotes (ie, "${STR_VAR}"),
# for numbers and booleans they should be plain (ie, ${INT_VAR}, ${BOOL_VAR})


# Global tags can be specified here in key="value" format.
[global_tags]
  # dc = "us-east-1" # will tag all metrics with dc=us-east-1
  # rack = "1a"
  ## Environment variables can be used as tags, and throughout the config file
  # user = "$USER"

# Configuration for telegraf agent
[agent]
  ## Default data collection interval for all inputs
  interval = "10s"
  ## Rounds collection interval to 'interval'
  ## ie, if interval="10s" then always collect on :00, :10, :20, etc.
  round_interval = true

  ## Telegraf will send metrics to outputs in batches of at most
  ## metric_batch_size metrics.
  ## This controls the size of writes that Telegraf sends to output plugins.
  metric_batch_size = 1000

  ## Maximum number of unwritten metrics per output.  Increasing this value
  ## allows for longer periods of output downtime without dropping metrics at the
  ## cost of higher maximum memory usage.
  metric_buffer_limit = 10000

  ## Collection jitter is used to jitter the collection by a random amount.
  ## Each plugin will sleep for a random time within jitter before collecting.
  ## This can be used to avoid many plugins querying things like sysfs at the
  ## same time, which can have a measurable effect on the system.
  collection_jitter = "0s"

  ## Default flushing interval for all outputs. Maximum flush_interval will be
  ## flush_interval + flush_jitter
  flush_interval = "10s"
  ## Jitter the flush interval by a random amount. This is primarily to avoid
  ## large write spikes for users running a large number of telegraf instances.
  ## ie, a jitter of 5s and interval 10s means flushes will happen every 10-15s
  flush_jitter = "0s"

  ## Collected metrics are rounded to the precision specified. Precision is
  ## specified as an interval with an integer + unit (e.g. 0s, 10ms, 2us, 4s).
  ## Valid time units are "ns", "us" (or "µs"), "ms", "s".
  ##
  ## By default or when set to "0s", precision will be set to the same
  ## timestamp order as the collection interval, with the maximum being 1s:
  ##   ie, when interval = "10s", precision will be "1s"
  ##       when interval = "250ms", precision will be "1ms"
  ##
  ## Precision will NOT be used for service inputs. It is up to each individual
  ## service input to set the timestamp at the appropriate precision.
  precision = "0s"

  ## Override default hostname, if empty use os.Hostname()
  hostname = ""
  ## If set to true, do no set the "host" tag in the telegraf agent.
  omit_hostname = false


###############################################################################
#                            OUTPUT PLUGINS                                   #
###############################################################################


# # Configuration for sending metrics to InfluxDB 2.0
 [[outputs.influxdb_v2]]
   ## The URLs of the InfluxDB cluster nodes.
   ##
   ## Multiple URLs can be specified for a single cluster, only ONE of the
   ## urls will be written to each interval.
   ##   ex: urls = ["https://us-west-2-1.aws.cloud2.influxdata.com"]
   urls = ["http://127.0.0.1:8086"]

   ## Token for authentication.
   token = $INFLUX_TOKEN

   ## Organization is the name of the organization you wish to write to.
   organization = "earthWatch"

   ## Destination bucket to write into.
   bucket = "telegraf"

[[inputs.kafka_consumer]]
  ## Kafka brokers.
  brokers = ["localhost:9092"]

  ## Topics to consume.
  topics = ["general-events"]

  ## Maximum length of a message to consume, in bytes (default 0/unlimited);
  ## larger messages are dropped
  max_message_len = 0

  ## Data format to consume.
  ## Each data format has its own unique set of configuration options, read
  ## more about them here:
  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
  data_format = "json_v2"
    [[inputs.file.json_v2]]
        measurement_name = "Earthquake"
        [[inputs.file.json_v2.tag]]
            path = "Earthquake.features.#.Features.properties.Properties.url"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.mag"
            type = "float"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.place"
            type = "string"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.time"
            type = "int"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.updated"
            type = "int"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.tz"
            type = "int"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.url"
            type = "string"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.detail"
            type = "string"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.felt"
            type = "bool"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.cdi"
            type = "float"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.mmi"
            type = "float"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.alert"
            type = "string"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.status"
            type = "string"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.tsunami"
            type = "bool"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.sig"
            type = "int"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.net"
            type = "string"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.code"
            type = "string"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.ids"
            type = "string"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.sources"
            type = "string"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.types"
            type = "string"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.nst"
            type = "int"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.dmin"
            type = "float"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.rms"
            type = "float"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.gap"
            type = "float"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.magType"
            type = "string"
        [[inputs.file.json_v2.field]]
            path = "Earthquake.features.#.Features.properties.Properties.type"
            type = "string"

After multiple attempts and modification of the telegraf.conf file, I was able to retrieve some fields from the data, but some fields are missing and I don't know what configuration allowed me to do this. enter image description here the image above shows a request just after the topic has received the data (so <1 minutes ago), but still impossible to display, as if they were not retransmitted.

Any idea what might be missing? Thank you

0 Answers
Related