transform

Documentation

Index

• func JqTransform(jqQuery string, logger *zap.Logger) (transform.MessageTransformFunc, error)

Functions

func JqTransform

func JqTransform(jqQuery string, logger *zap.Logger) (transform.MessageTransformFunc, error)

JqTransform creates a MessageTransformFunc that applies a JQ query to message payloads.

The function compiles the provided JQ query string and returns a transform that: 1. Applies the JQ query to the message payload 2. Returns a new message with the transformed payload 3. Preserves the original message context and topic

The JQ query operates on the message payload, which can be any JSON-serializable value. If the payload cannot be processed or the JQ query fails, the original message is returned unchanged.

The JQ query has access to the following variables:

• $topic: The message topic as a string

Parameters:

• jqQuery: A valid JQ query string (e.g., ".field", ".[] | select(.active)", "{data: ., topic: $topic}", etc.)
• logger: Optional logger for error reporting. If nil, errors are not logged (but still handled gracefully)

Returns:

• A MessageTransformFunc that applies the JQ transformation
• An error if the JQ query cannot be compiled

Example usage:

// Extract a specific field with logging
extractName, err := JqTransform(".name", logger)
if err != nil {
    log.Fatal(err)
}

// Filter array elements without logging
filterActive, err := JqTransform(".[] | select(.active == true)", nil)
if err != nil {
    log.Fatal(err)
}

// Transform and restructure data with topic information
restructure, err := JqTransform("{user: .name, age: .age, status: .active, source_topic: $topic}", logger)
if err != nil {
    log.Fatal(err)
}

// Use topic in conditional logic
conditionalTransform, err := JqTransform("if $topic | startswith(\"user/\") then {type: \"user_event\", data: .} else . end", logger)
if err != nil {
    log.Fatal(err)
}

// Use in a transform pipeline
transforms := []MessageTransformFunc{
    extractName,
    AddTopicPrefix("processed/"),
}

The transform handles various payload types:

• Maps/objects: Used directly (already in primitive form)
• Arrays/slices of primitives: Used directly (already in primitive form)
• Arrays/slices containing structs: Converted via JSON marshaling/unmarshaling
• Primitives: Used directly (strings, numbers, booleans, etc.)
• Structs and *structs: Converted via JSON marshaling/unmarshaling to primitive maps
• JSON strings: Parsed to primitive types
• Byte slices: Parsed as JSON if valid, otherwise treated as strings
• cty.Value types: Converted using go2cty2go

If the JQ query produces multiple results, they are collected into an array. If the JQ query produces no results, the message is dropped (returns nil).

Error Handling: When a logger is provided, all runtime errors during transformation are logged but the transform continues gracefully by passing through the original message unchanged.