OpenTelemetry + Debezium

[arnaud-thorel-of]

Environment

• PostgreSQL version: all
• PostgREST version: all
• Operating system: all

Description of issue

I come to you with a question, I need to implement OpenTelemetryTracing in a specific case
The components used are :

• SpringBoot as Back4Front
• Postgrest as API
• Postgresql as Database
• KafkaConnect + Debezium as Change Data Capture
• Kafka as Message Broker

So my journey start with a call from a user to my SpringBoot app which start observability tracing

User -> SpringBoot : Call API
SpringBoot -> Postgrest : POST /contract_v1/ {} -h tracespancontext = "context"
--> Here I need to save the content of tracespancontext in "contract_v1.tracespancontext"

That allow me to retrieve it from debezium (according to the documentation https://debezium.io/documentation/reference/stable/integrations/tracing.html)

Then after it allows me to retrieve my trace_context inside kafka header message and allow me to have a full trace, from api to kafka-listeners

Actually :

I need to do it programatically by creating table and views with column 'tracespancontext' and passing inside the body.

Is there a possibility to declare a plugin that extract header tracespancontext and put data in a pre-commit function on the table if the table contains tracespancontext ?

Thx in advance
Kind Regards

[wolfgangwalther]

Is there a possibility to declare a plugin that extract header tracespancontext and put data in a pre-commit function on the table if the table contains tracespancontext ?

You can access request headers via current_setting in a BEFORE INSERT trigger. You'd need to add this trigger to all tables which have this column.

[arnaud-thorel-of]

Thx for your anwser, I'll give it a try

[arnaud-thorel-of]

Hey !

Thx a lot @wolfgangwalther
I come to you to solved this discussion :

First I use a flyway script to add those sql on every tables listed on my debezium configuration

CREATE OR REPLACE FUNCTION otel_exporter()
                    RETURNS TRIGGER AS $$
                    DECLARE
                        headers JSON;
                        tracing_context JSON;
                        baggage TEXT;
                        traceparent TEXT;
                    BEGIN
                        tracing_context := NULL;
                        -- retrieve values of traceparent and baggage from request headers
                        headers := current_setting('request.headers', true)::json;
                        baggage := headers->>'baggage';
                        traceparent := headers->>'traceparent';
                        -- Build json object (baggage is optional)
                        IF baggage IS NOT NULL AND traceparent IS NOT NULL THEN
                            tracing_context := json_build_object('traceparent', traceparent, 'baggage', baggage);
                        ELSIF traceparent IS NOT NULL THEN
                            tracing_context := json_build_object('traceparent', traceparent);
                        END IF;
                    
                        -- Update tracingspancontext (if INSERT, UPDATE on NEW, otherwise on OLD, it's usefull if we do logical deletion)
                        IF TG_OP = 'INSERT' OR TG_OP = 'UPDATE' THEN
                            NEW.tracingspancontext := tracing_context::TEXT;
                            RETURN NEW;
                        ELSIF TG_OP = 'DELETE' THEN
                            OLD.tracingspancontext := tracing_context::TEXT;
                            RETURN OLD;
                        END IF;
                        RETURN NULL; -- Prevent from errors
                    END;
                    $$ LANGUAGE plpgsql;

Then on each tables :

ALTER TABLE XXX ADD COLUMN IF NOT EXISTS tracingspancontext TEXT;
CREATE OR REPLACE TRIGGER update_tracingspancontext BEFORE UPDATE OR INSERT OR DELETE ON XXX FOR EACH ROW EXECUTE FUNCTION otel_exporter();

On the other way inside my debezium system I can now extract data from my body and put it to the header using HeaderFrom$Value transformation (https://fossies.org/linux/kafka/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/HeaderFrom.java)

In my case I wrote directly a subclass of it because I want to be fail-safe and have the same behaviour on every tables :

• Extract value tracingspancontext from the body (if it's exists)
• For each item in the json put it in the headers
• Remove tracingspancontext from the body

Then all the official instrumentation of OpenTelemetry do the job, by receiving 'traceparent', and 'baggage' from headers and keep the propagation to the next spans