String to UUID implementation in telegraf

input plugin kafka
output plugin postgresql

when i consuming event id and timestamps are convert to string so its giving error because its creating conflicts between postgres columns datatype

CREATE TABLE IF NOT EXISTS public.integration_rate_limits(
id UUID,
reset_date_time TIMESTAMP,
“organization_id” UUID,
created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP)

[agent]
interval = “1s”
round_interval = true
metric_batch_size = 1000
metric_buffer_limit = 10000
collection_jitter = “0s”
flush_interval = “1s”
flush_jitter = “0s”
precision = “”
hostname = “”
omit_hostname = true

[[inputs.kafka_consumer]]
name_override = “integration_rate_limits”
brokers = [“localhost:29092”]
topics = [“otg.ratelimit-telegraf.t”]
metadata_full = true
consumer_group = “consumer_group_rate_limit”
max_undelivered_messages = 1
max_message_len = 1000000
data_format=“json_v2”

[[inputs.kafka_consumer.json_v2]]
[[inputs.kafka_consumer.json_v2.field]]
path = “event.body.id”
rename = “id”
type = “string”
optional = true
[[inputs.kafka_consumer.json_v2.field]]
path = “event.body.type”
rename = “type”
type = “string”
optional = true
[[inputs.kafka_consumer.json_v2.field]]
path = “event.body.limit”
type = “int”
optional = true
[[inputs.kafka_consumer.json_v2.field]]
path = “event.body.remaining”
type = “int”
optional = true
[[inputs.kafka_consumer.json_v2.field]]
path = “event.body.used”
type = “int”
optional = true
[[inputs.kafka_consumer.json_v2.field]]
path = “event.body.resetDateTime”
rename = “reset_date_time”
optional = true
# [[inputs.kafka_consumer.json_v2.field]]
# path = “event.body.lastPolledDateTime”
# rename = “last_polled_date_time”
# type = “string”
# optional = true
# [[inputs.kafka_consumer.json_v2.tag]]
# path = “event.body.organization.id”
# rename = “organization_id”
# type = “string”
# optional = true
# [[inputs.kafka_consumer.json_v2.tag]]
# path = “event.body.integration.id”
# rename = “integration_id”
# type = “string”
# optional = true

[[inputs.kafka_consumer.json_v2]]

timestamp_path = “event.body.resetDateTime”

timestamp_format = “rfc3339”

[[processors.timestamp]]
field = “reset_date_time”

date_format = “unix”

timezone = “UTC”

source_timestamp_format = “RFC3339”
source_timestamp_timezone=“UTC”
destination_timestamp_format = “RFC3339”
destination_timestamp_timezone = “UTC”

[[outputs.postgresql]]
schema = “public”
connection = "host=localhost user=postgres password=password sslmode=disable dbname=postgres port=5432 pool_max_conns = 10 pool_min_conns = 1 pool_max_conn_lifetime = 2000s pool_max_conn_idle_time = 10s pool_health_check_period = 10s "
timestamp_column_name = “created_date”
timestamp_column_type = “timestamp with time zone”
create_templates = [
‘’’
INSERT INTO public.integration_rate_limits (id, type) VALUES({{ .id}}::UUID, {{ .type}}, {{ .limit}}, {{ .remaining}}, {{ .resetDateTime}}, {{ .lastPolledDateTime}}, {{ .organization_id}}, {{ .integration_id}} );
‘’’
]
log_level = “debug”

can anyone help with this

Are you sure your id has a valid UUID format?

Thank you for quick reply

When its come to kafka plugin im getting as jsonv2 so id is converted to string but i need that id as UUID type for postgres column. Any example or anything can you share with that will really helpful.

I don’t think this is currently possible as we do automatically determine which Postgres type to use based on the field/tag type. Could you please open a feature request and provide your vision of how to configure individual column types?!?

Thank you so much for your reply srebhan

I will definitely raise a feature request.

Two more last questions i have .
1.is there any processor plugin i can use to convert string to uuid before its going to postgres.like starlak

  1. In db i have column type timestamp but from telegraf im not able to convert the string timestamp to timestamp datatyle for postgres

Thnks in adv. kindly give me your Guidance here

@Mk_Anbu for 1. there is no way as Telegraf itself does not provide a UUID type but it is typically transported as string. So if Postgres cannot automatically convert a string to its UUID type we need an explicit conversion in the output plugin. Seems like there is a way to add support

For 2: Please check the documentation of the plugin, it offers timestamp_column_name option to tell Telegraf which column the metric timestamp should go to. It furthermore provides the timestamp_column_type option to tune how the timestamp is stored.
Could you provide a Telegraf metric and which information to use as timestamp? I’m pretty sure we find a way to make this work…

1 Like