Telegraf -> Kafka -> Vector -> Prometheus

[sushovan23]

I'm trying to achieve the flow from telegraf -> kafka -> vector -> prometheus
In telegraf we have configured kafka output plugin and sending data in json format.

{
    "fields": {
        "field_1": 30,
        "field_2": 4,
        "field_N": 59,
        "n_images": 660
    },
    "name": "docker",
    "tags": {
        "host": "raynor"
    },
    "timestamp": 1458229140
}

Now in vector side. We have configured kafka source and able to read it in vector as verified by using console sink.
However my goal is to remote_write the kafka logs (which are essentially metrics) into prometheus using remote_write sink.

I read through documentation and there isn't a parser available for it. Is there anyone who has tried this scenario?

[jszwedko]

Hi @sushovan23 !

Can you provide examples of the metrics you want to write out in OpenMetrics format? For example:

field_1{host=raynor} 30 1458229140

? Or something else?

[sushovan23]

@jszwedko kafka input source is making the message as a string instead of table.. so I am not able to unpack it using for loop in lua.
any suggestions on how to extract the metrics?

[sushovan23]

Made a bunch of progress:

checked other post little similar to this and I added a remap before lua transform:

[transforms.my_test_parse]
  type = "remap"
  inputs = ["kafka_input"]
  source = '''
  .metric = parse_json!(.message)
  .metric.timestamp = to_timestamp!(.metric.timestamp)
  '''

now my next transform looks like this:

[transforms.my_transform_id]
type = "lua"
inputs = [ "my_test_parse" ]
version = "2"
hooks.process = """
  function(event, emit)
    for field, value in pairs(event.log.metric.fields) do
      emit {
        metric = {
          name = event.log.metric.name .. "_" .. field,
          kind = "incremental",
          timestamp = os.date("*t", XXXX),     -> problem here as now my timestamp is also a tableformat
          counter = {
            value = value
          },
          tags = event.log.metric.tags
        }
      }
    end
  end
"""

timestamp is a table which contains following:

year	2022
wday	5
yday	139
sec	15
isdst	false
min	59
month	5
day	19
hour	3
nanosec	0

I got no idea how to convert this to timestamp.

[sushovan23]

@jszwedko Thanks for all the help. timestamp = os.time(event.log.metric.timestamp), was the function which takes table as input.

Getting following error which doesn't really say what is wrong:

2022-05-19T04:23:05.794640Z ERROR transform{component_kind="transform" component_id=my_transform_id component_type=lua component_name=my_transform_id}: vector::internal_events::lua: Error in lua script; discarding event. error=RuntimeErrorHooksProcess { source: CallbackError { traceback: "stack traceback:\n\t[C]: in local 'emit'\n\t[string \"src/transforms/lua/v2/mod.rs:130:9\"]:3: in function <[string \"src/transforms/lua/v2/mod.rs:130:9\"]:1>", cause: FromLuaConversionError { from: "integer", to: "table", message: None } } } error_type="script_failed" error_code="runtime_error_hook_process" stage="processing" internal_log_rate_secs=30
2022-05-19T04:23:05.794876Z ERROR transform{component_kind="transform" component_id=my_transform_id component_type=lua component_name=my_transform_id}: vector::internal_events::lua: Internal log [Error in lua script; discarding event.] is being rate limited.

when I try with following:
timestamp = event.log.metric.timestamp
then get slightly different error

2022-05-19T04:39:45.961848Z ERROR transform{component_kind="transform" component_id=my_transform_id component_type=lua component_name=my_transform_id}: vector::internal_events::lua: Error in lua script; discarding event. error=RuntimeErrorHooksProcess { source: CallbackError { traceback: "stack traceback:\n\t[C]: in local 'emit'\n\t[string \"src/transforms/lua/v2/mod.rs:130:9\"]:3: in function <[string \"src/transforms/lua/v2/mod.rs:130:9\"]:1>", cause: FromLuaConversionError { from: "string", to: "f64", message: Some("expected number or string coercible to number") } } } error_type="script_failed" error_code="runtime_error_hook_process" stage="processing" internal_log_rate_secs=30
2022-05-19T04:39:45.976858Z ERROR transform{component_kind="transform" component_id=my_transform_id component_type=lua component_name=my_transform_id}: vector::internal_events::lua: Internal log [Error in lua script; discarding event.] is being rate limited.

[sushovan23]

@jszwedko Could you let me know where I can find about the formatting for timestamp in the emit:

      emit {
        metric = {
          name = event.log.metric.name .. "_" .. field,
          kind = "incremental",
          timestamp = os.date("*t", XXXX),     -> problem here as now my timestamp is also a tableformat
          counter = {
            value = value
          },
          tags = event.log.metric.tags
        }
      }

I have tried providing the timestamp in string format, in table format but nothing works

[sushovan23]

@jszwedko finally got it working!!!

lua transform I did this - timestamp = event.log.metric.timestamp,

seems like timestamp is taking data in table format

and I can see data in prometheus.

Is there a way to make data expire after 60s or 120s. right now prometheus I can get data in graph but current time doesnt written since its not written yet.

Guess I'm looking something similar to expiration_interval in telegraf output plugin. something similar for remote_write too are you aware