vector=>kafka=>vector=>prometheus

[petrxpx]

Hello! I need some help with vector data model:
I have a metrics from prometheus(metric format)
then vector takes it and send to kafka(in metric format)

I read from kafka by another vector in log(text ) format. And i cant't convert thes messge(metric format) to send to prometheus
I have the message after kafka:
{ "metric":{ "gauge":{ "value":23.0 }, "kind":"absolute", "name":"ingress_nginx_detail_upstream_response_seconds_bucket", "tags":{ "app":"controller-main", "content_kind":"cache-headers-not-present", "controller":"main", "ingress":"akhq", "instance":"10.244.19.165:10354", "job":"nginx-ingress-controller", "le":"1.5", "location":"/", "namespace":"preprod", "node":"s-bd", "prometheus":"deckhouse", "scrape_source":"protobuf", "service":"akhq", "service_port":"http", "tier":"cluster", "vhost":"ak" }, "timestamp":"2021-05-31T09:31:49.523Z" } }

How can i expose it by sink prometheus exporter

[spencergilbert]

@jszwedko by my understanding the sink here won't differentiate between logs/metrics and that we must default to logs?

We'd want to use log_to_metric here?

EDIT: log_to_metric isn't actually "right" tool to use here :(
See #1471 for more details

[petrxpx]

Hi!
I'm using log to metric, but i have many different(counter, gauge) prometheus metrics before kafka. After kafka they become log events. Is it possible to automatical transform log events after kafka to metrics?
My conf

[sources.kafka]
  type = "kafka" 
  bootstrap_servers =  

[transforms.my_test_parse]
  type = "remap" # required
  inputs = ["kafka"] # required
  source = '''
  . =parse_json!(.message)
  .gaugenew = to_float!(.gauge.value) 
  .counternew = to_float!(.count.value) 
  .timestamp = to_timestamp!(.timestamp)
  '''
[transforms.my_filter]
  type = "filter"
  inputs =  ["my_test_parse"]
  condition = ' .name == "ingress_nginx_detail_backend_responses_total" ' 
  
[transforms.my_transform]
  type = "log_to_metric" # required
  inputs = ["my_filter"] 

  [[transforms.my_transform.metrics]]
    name = "{{name}}"
    field = "gaugenew"
    type = "gauge" 
    tags.statuscode = "{{tags.status}}"
    tags.instance = "{{tags.instance}}"
    tags.vhost = "{{tags.vhost}}"
    tags.namespace = "{{tags.namespace}}"_

[spencergilbert]

@jszwedko suggested the lua transform should be able to be used here today like:

event.metric = event.log
event.log = nil
emit(event)

[petrxpx]

Hi, thank you for help!
But lua doesn't work with message like this :

{"name":"ingress_nginx_detail_backend_lowres_upstream_response_seconds_count","tags":{"app":"controller-main","controller":"main","ingress":"smoke-mini","instance":"***","job":"nginx-ingress-controller","location":"/","namespace":"d8-upmeter","node":"p-payment-ks-loadbalancer-2d3fdd78-5c5df-n9nf5","pod_ip":"***","prometheus":"deckhouse","scrape_source":"protobuf","service":"smoke-mini","service_port":"http","tier":"cluster","vhost":"smoke"},"timestamp":"2021-06-16T08:57:56.093Z","kind":"absolute","gauge":{"value":25.0}}

transformation can't find type of metric(gauge, counter etc) or field with metric value

[spencergilbert]

🤔 it's possible you may be running into this problem I just found this morning: #7881

Can you share what transformation configs you're trying to use?

[petrxpx]

I use transformation like this, where "message" is field of json from kafka.
Is it correct?

[transforms.my_test_parse]
  type = "remap"
  inputs = ["kafka"]
  source = '''
  . =parse_json!(.message)
  .metric= .
'''
[transforms.my_transform]
  type = "lua" 
  inputs = ["my_test_parse"] 
  version = "2" 

        hooks.process = '''
        function (event, emit)
            event.metric = event.message
            event.log = nil
            emit(event)
        end
        '''

[spencergilbert]

Ok, major credit to @jszwedko

---
sources:
  kafka_in:
    type: kafka
    bootstrap_servers: 127.0.0.1:9091
    group_id: vector
    topics:
    - example
transforms:
  parse_kafka:
    type: remap
    inputs:
    - kafka_in
    source: |
      .metric = parse_json!(.message)
      .metric.timestamp = to_timestamp!(.metric.timestamp)
  to_metric:
    type: lua
    inputs:
    - parse_kafka
    version: '2'
    hooks:
      process: |
        function (event, emit)
          event.metric = event.log.metric
          event.log = nil
          emit(event)
        end

I can convert this over to TOML if it helps, but those transforms should get you where you want

[petrxpx]

Hi, @spencergilbert thank you for help! I applied configuration, but sink prometheus exporter have nothing to expose.
configuration:

[transforms.my_test_parse]
  type = "remap"
  inputs = ["kafka"]
  source = '''
  .metric = parse_json!(.message)
  .metric.timestamp = to_timestamp!(.metric.timestamp)
  '''
[transforms.my_transform]
  type = "lua" 
  inputs = ["my_test_parse"] 
  version = "2" 
        hooks.process = '''
        function (event, emit)
            event.metric = event.log.metric
            event.log = nil
            emit(event)
        end
        '''

Messages in logs
Jun 18 09:45:31.291 DEBUG hyper::proto::h1::io: parsed 16 headers
Jun 18 09:45:31.291 DEBUG hyper::proto::h1::conn: incoming body is empty
Jun 18 09:45:31.291 DEBUG vector::internal_events::prometheus: Request to prometheus server complete. status_code=200 OK
Jun 18 09:45:31.291 DEBUG hyper::proto::h1::io: flushed 116 bytes

Do you have any ideas?

[spencergilbert]

Can you share the config that uses the above as input, and the output from curling the prometheus_exporter address on the /metrics path?

[petrxpx]

@spencergilbert, i'm awfully sorry. It works good
I have an error in configuration in subsystem(not in vector).
Thank you, and I wish you all the best.

[spencergilbert]

Excellent, glad I can help - we have some planned work to improve the documentation around using VRL with metrics, as well as handling metrics more easily when passing them through components that handle both logs and metrics.

Thanks for working through this with me!

[minuchi]

Hello!
Is this still only way to convert metrics from kafka source?

I'm trying this but the vector says like this below.

2023-06-26T05:58:44.201136Z ERROR transform{component_kind="transform" component_id=host_metrics component_type=lua component_name=host_metrics}: vector::internal_events::lua: Error in building lua script. error=RuntimeErrorHooksProcess { source: CallbackError { traceback: "stack traceback:\n\t[C]: in local 'emit'\n\t[string \"src/transforms/lua/v2/mod.rs:228:9\"]:4: in function <[string \"src/transforms/lua/v2/mod.rs:228: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_limit=true

my vector version is 0.30.0.
kafka_host_metrics data have come from vector host data via kafka and here's my transforms code.

transforms:
  parse_host_metrics:
    type: remap
    inputs:
    - kafka_host_metrics
    source: |
      .metric = parse_json!(.message)
      .metric.timestamp = to_timestamp!(.metric.timestamp)
  host_metrics:
    type: lua
    inputs:
    - parse_host_metrics
    version: "2"
    hooks:
      process: |
        function (event, emit)
          event.metric = event.log.metric
          event.log = nil
          emit(event)
        end

[jszwedko]

It sounds like the structure doesn't exactly match the expected structure for metrics. Could you log out . in the remap transform and share it?

Separately, the kafka source is capable of decoding metrics directly using the decoding.codec option of native (protobuf) or native_json (JSON) but this depends on whatever is writing to Kafka writing the data in one of those formats. If Vector is the one writing the data, the kafka sink supports the same options as encoding.codec.