-
Notifications
You must be signed in to change notification settings - Fork 9
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add jq generic mapper transformation
- Loading branch information
Showing
8 changed files
with
704 additions
and
3 deletions.
There are no files selected for viewing
15 changes: 15 additions & 0 deletions
15
assets/docs/configuration/transformations/builtin/jq-full-example.hcl
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,15 @@ | ||
| transform { | ||
| use "jq" { | ||
| jq_command = <<JQEOT | ||
| { | ||
| my_app_id: .app_id, | ||
| my_nested_prop: { | ||
| playback_rate: .contexts_com_snowplowanalytics_snowplow_media_player_1[0].playbackRate | ||
| } | ||
| } | ||
| JQEOT | ||
|
|
||
| timeout_sec = 5 | ||
| snowplow_mode = true | ||
| } | ||
| } |
5 changes: 5 additions & 0 deletions
5
assets/docs/configuration/transformations/builtin/jq-minimal-example.hcl
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,5 @@ | ||
| transform { | ||
| use "jq" { | ||
| jq_command = "[.]" | ||
| } | ||
| } |
11 changes: 11 additions & 0 deletions
11
assets/test/transformconfig/TestGetTransformations/configs/jq.hcl
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| transform { | ||
| use "jq" { | ||
| jq_command = <<JQEOT | ||
| { | ||
| my_app_id: .app_id, | ||
| } | ||
| JQEOT | ||
|
|
||
| snowplow_mode = true | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,156 @@ | ||
| /** | ||
| * Copyright (c) 2020-present Snowplow Analytics Ltd. | ||
| * All rights reserved. | ||
| * | ||
| * This software is made available by Snowplow Analytics, Ltd., | ||
| * under the terms of the Snowplow Limited Use License Agreement, Version 1.0 | ||
| * located at https://docs.snowplow.io/limited-use-license-1.0 | ||
| * BY INSTALLING, DOWNLOADING, ACCESSING, USING OR DISTRIBUTING ANY PORTION | ||
| * OF THE SOFTWARE, YOU AGREE TO THE TERMS OF SUCH LICENSE AGREEMENT. | ||
| */ | ||
|
|
||
| package transform | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/json" | ||
| "fmt" | ||
| "time" | ||
|
|
||
| "github.com/itchyny/gojq" | ||
|
|
||
| "github.com/snowplow/snowbridge/config" | ||
| "github.com/snowplow/snowbridge/pkg/models" | ||
| ) | ||
|
|
||
| // JQMapperConfig represents the configuration for the JQ transformation | ||
| type JQMapperConfig struct { | ||
| JQCommand string `hcl:"jq_command"` | ||
| RunTimeoutMs int `hcl:"timeout_ms,optional"` | ||
| SpMode bool `hcl:"snowplow_mode,optional"` | ||
| } | ||
|
|
||
| // JQMapper handles jq generic mapping as a transformation | ||
| type jqMapper struct { | ||
| JQCode *gojq.Code | ||
| RunTimeoutMs time.Duration | ||
| SpMode bool | ||
| } | ||
|
|
||
| // RunFunction runs a jq mapper transformation | ||
| func (jqm *jqMapper) RunFunction() TransformationFunction { | ||
| return func(message *models.Message, interState interface{}) (*models.Message, *models.Message, *models.Message, interface{}) { | ||
| input, err := mkJQInput(jqm, message, interState) | ||
| if err != nil { | ||
| message.SetError(err) | ||
| return nil, nil, message, nil | ||
| } | ||
|
|
||
| ctx, cancel := context.WithTimeout(context.Background(), jqm.RunTimeoutMs) | ||
| defer cancel() | ||
|
|
||
| iter := jqm.JQCode.RunWithContext(ctx, input) | ||
| // no looping since we only keep first value | ||
| v, ok := iter.Next() | ||
| if !ok { | ||
| message.SetError(fmt.Errorf("jq query got no output")) | ||
| return nil, nil, message, nil | ||
| } | ||
|
|
||
| if err, ok := v.(error); ok { | ||
| message.SetError(err) | ||
| return nil, nil, message, nil | ||
| } | ||
|
|
||
| // here v is any, so we Marshal. alternative: gojq.Marshal | ||
| data, err := json.Marshal(v) | ||
| if err != nil { | ||
| message.SetError(fmt.Errorf("error encoding jq query output data")) | ||
| return nil, nil, message, nil | ||
| } | ||
|
|
||
| message.Data = data | ||
| return message, nil, nil, nil | ||
| } | ||
| } | ||
|
|
||
| // jqMapperAdapter implements the Pluggable interface | ||
| type jqMapperAdapter func(i interface{}) (interface{}, error) | ||
|
|
||
| // ProvideDefault implements the ComponentConfigurable interface | ||
| func (f jqMapperAdapter) ProvideDefault() (interface{}, error) { | ||
| return &JQMapperConfig{ | ||
| RunTimeoutMs: 100, | ||
| }, nil | ||
| } | ||
|
|
||
| // Create implements the ComponentCreator interface | ||
| func (f jqMapperAdapter) Create(i interface{}) (interface{}, error) { | ||
| return f(i) | ||
| } | ||
|
|
||
| // jqMapperAdapterGenerator returns a jqAdapter | ||
| func jqMapperAdapterGenerator(f func(c *JQMapperConfig) (TransformationFunction, error)) jqMapperAdapter { | ||
| return func(i interface{}) (interface{}, error) { | ||
| cfg, ok := i.(*JQMapperConfig) | ||
| if !ok { | ||
| return nil, fmt.Errorf("invalid input, expected JQMapperConfig") | ||
| } | ||
|
|
||
| return f(cfg) | ||
| } | ||
| } | ||
|
|
||
| // jqMapperConfigFunction returns a jq mapper transformation function from a JQMapperConfig | ||
| func jqMapperConfigFunction(c *JQMapperConfig) (TransformationFunction, error) { | ||
| query, err := gojq.Parse(c.JQCommand) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("error parsing jq command: %q", err.Error()) | ||
| } | ||
|
|
||
| code, err := gojq.Compile(query) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("error compiling jq query: %q", err.Error()) | ||
| } | ||
|
|
||
| jq := &jqMapper{ | ||
| JQCode: code, | ||
| RunTimeoutMs: time.Duration(c.RunTimeoutMs) * time.Millisecond, | ||
| SpMode: c.SpMode, | ||
| } | ||
|
|
||
| return jq.RunFunction(), nil | ||
| } | ||
|
|
||
| // JQMapperConfigPair is a configuration pair for the jq mapper transformation | ||
| var JQMapperConfigPair = config.ConfigurationPair{ | ||
| Name: "jq", | ||
| Handle: jqMapperAdapterGenerator(jqMapperConfigFunction), | ||
| } | ||
|
|
||
| // mkJQInput ensures the input to JQ query is of expected type | ||
| func mkJQInput(jqm *jqMapper, message *models.Message, interState interface{}) (map[string]interface{}, error) { | ||
| if !jqm.SpMode { | ||
| // gojq input can only be map[string]any or []any | ||
| // here we only consider the first, but we could also expand | ||
| var input map[string]interface{} | ||
| err := json.Unmarshal(message.Data, &input) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| return input, nil | ||
| } | ||
|
|
||
| parsedEvent, err := IntermediateAsSpEnrichedParsed(interState, message) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| spInput, err := parsedEvent.ToMap() | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| return spInput, nil | ||
| } |
Oops, something went wrong.