[ADDED] Stream subject mapping transforms #3814

jnmoyne · 2023-01-25T20:17:35Z

Documentation added (if applicable)
Tests added
Branch rebased on top of current dev
Build is green in Travis CI
You have certified that the contribution is your original work and that you license the work to the project under the Apache 2 license

Adds support for subject mapping transforms in streams

Introduces the following changes:

Extracts subject transform code and tests from accounts.go to it's own separate set of files.
Ability to add a subject mapping transform destination to sourced streams, allowing to map message subjects as they are received from the stream being sourced
Ability to source more than once from the same stream, each time using a specific subject filter and destination transform
Ability to add a stream input subject mapping transform to a stream which will not act as a filter but will transform the subject of messages matching the transform's source according to the transform's destination. This mapping happens regardless of how the messages are received into the stream (mirror, or subject(s) and/or source(s))

This enables many interesting use cases, such as KV bucket sourcing, or inserting a partition number token in the subject in order to then create a consumer per partition, directly at the stream definition level rather than having to resort to using the Core NATS subject mapping functionality (which requires administrative access to the server config file or account key to be configured).

/cc @nats-io/core

Extract subject transformation code out of accounts.go Stream sources can now have a subject mapping transform You can source the same stream more than once Remove limitation that the subject filter for a source, mirror or consumer must have an overlap with the sourced/mirrored's stream or the stream's subjects

derekcollison

Just a few minor nits.

derekcollison · 2023-01-25T21:27:22Z

server/jetstream_test.go

@@ -11647,6 +11631,14 @@ func TestJetStreamSourceBasics(t *testing.T) {
 		return nil
 	})

+	m, err := js.GetMsg("MS", 1)
+	if err != nil {


requie_NoError(t, err)

Done, since this pattern was present in many places in the file I made it consistent and replaced with require_NoError in all these other places in the file as well

derekcollison · 2023-01-25T21:28:43Z

server/jetstream_test.go

+	}
+
+	m, err := js.GetMsg("T1", 1)
+	if err != nil {


require_NoError()

Done (see above)

derekcollison · 2023-01-25T21:29:01Z

server/stream.go

@@ -58,6 +58,9 @@ type StreamConfig struct {
 	Mirror       *StreamSource   `json:"mirror,omitempty"`
 	Sources      []*StreamSource `json:"sources,omitempty"`

+	// Allow applying a subject transform to incoming messages before doing anything else
+	InputSubjectTransform *InputSubjectTransform `json:"input_subject_transform,omitempty"`


Shorter json name possible?

I thought of just input_transform but then thought maybe we want to have transform on other things than subjects later. And also thought since it's only in stream config / stream info it's not critical to optimize the size and as humans will be editing those JSONs a lot of the time I erred on the side of maybe a bit too verbose figuring it couldn't hurt 🤷‍♂️ but yeah it's like 'input' it's the best I could come up with but not especially attached to it... Open to anything you'd rather use

Given it’s “subjects” when setting the interest subjects for the stream naming this input sometning is confusing so maybe subject_transform or even subjects_transform to be very clear

Yeah I put input in there to try and denote that it's not just messages from the 'subjects' that get transformed but also those from sources or the mirror, but really for lack of a better term. But just "subjects_transform" may lead people to forget it also applies to mirror/sources.

Trying to convey it's a 'stream pre-processing' transform that happens after the subjects/mirror/sources step (which itself can already have another transform, at least for sources)

We cant possibly convey all this by choice of a single word :) So seems we should at least just pick a term and stick with it, even better if its one we already use. It's subject(s).

Maybe trying to cram too much meaning in a field name (picking good names can be hard 😅) and "subject_transform" is enough... @derekcollison let me know what name you like best and I'll adjust everywhere (companion branches) for it.

@ripienaar fair enough, I will go with "subject_transform" then (unless someone else comes with a better one by tomorrow morning)

derekcollison · 2023-01-25T21:31:30Z

server/stream.go

 // Lock should be held
 func (mset *stream) sourceInfo(si *sourceInfo) *StreamSourceInfo {
 	if si == nil {
 		return nil
 	}

-	ssi := &StreamSourceInfo{Name: si.name, Lag: si.lag, Error: si.err}
+	var ssi *StreamSourceInfo


Maybe keep same original assignment and just add the field if si.tr != nil?

derekcollison · 2023-01-25T21:32:46Z

server/subject_transform.go

+
+// Helper to ingest and index the subjectTransform destination token (e.g. $x or {{}}) in the token
+// returns a transformation type, and three function arguments: an array of source subject token indexes, and a single number (e.g. number of partitions, or a slice size), and a string (e.g.a split delimiter)
+


Remove empty line, make line above 2.

… and calling Fatalf throughout all the tests in the file Improvements to readability

derekcollison · 2023-01-26T17:44:38Z

LMK when you want me to take another look for review.

jnmoyne · 2023-01-26T18:01:11Z

Still need to do the changes for input_subject_transform naming but I’m in meetings most of the time until sometime in the afternoon today, I’ll let you know. I believe that name change should be the only thing left.

…

On Jan 26, 2023, at 9:44 AM, Derek Collison ***@***.***> wrote: LMK when you want me to take another look for review. — Reply to this email directly, view it on GitHub <#3814 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AB3U6WEROTVTPEYP6HLFF4TWUKZZDANCNFSM6AAAAAAUGX2NKQ>. You are receiving this because you authored the thread.

…ce's SubjectTransform)

jnmoyne · 2023-01-26T21:01:37Z

Change of the StreamConfig's input subject transform to just subject transform. Because I was already using "subject_transform" inside source config/source info which is not a complete transform (source+dest) but just the dest (since the source is in the subject_filter field) in turn I had to rename that to something else.

Commits have also been pushed for this changes to the companion branches in nats.go, jsm.go, and natscli

derekcollison

LGTM

bruth · 2023-01-29T12:13:32Z

@derekcollison @jnmoyne Playing around, I was able to construct a cycle with the two stream setup with this PR and #3825 applied by setting the subject_transform_dest to the local subject subject, e.g.

{
  "name": "n1",
  "subjects": ["n1.>"],
  "sources": [
    {
      "name": "n2",
      "filter_subject": "n2.>",
      "subject_transform_dest": "n1.>"
    }
  ]
}

And the equivalent for n2 (swapping n1 and n2). This is to be expected since the transform occurs before sourcing occurs, but just wanted to call it out.

jnmoyne · 2023-01-29T15:26:08Z

Will take a look

derekcollison · 2023-01-29T18:28:56Z

I added in ability to have two sources reference each other as long as their subjects did not overlap.

Just need to add in the transform logic in dev branch.

bruth · 2023-01-29T19:07:03Z

I had created a branch that rebased this one on top of main, incorporating both sets of changes called bjr-suse.

jnmoyne · 2023-01-30T21:48:46Z

Can't reproduce using the current top of dev, which also has #3825 merged in

jnmoyne requested a review from derekcollison January 25, 2023 20:17

jnmoyne force-pushed the jnm/streamsourcetransform branch from fbd63d9 to a953e84 Compare January 25, 2023 20:26

rebase fix

e2561a6

derekcollison reviewed Jan 25, 2023

View reviewed changes

Change test to use require_NoError instead of explicitly checking err…

4fbe13e

… and calling Fatalf throughout all the tests in the file Improvements to readability

jnmoyne mentioned this pull request Jan 26, 2023

Sourced stream with subject + source subject mapping nats-io/nats.go#1197

Closed

Rename StreamConfig's InputSubjectTransform (and therefore StreamSour…

ddce3ce

…ce's SubjectTransform)

derekcollison self-requested a review January 26, 2023 22:06

derekcollison approved these changes Jan 26, 2023

View reviewed changes

derekcollison merged commit c2d3d9c into dev Jan 26, 2023

derekcollison deleted the jnm/streamsourcetransform branch January 26, 2023 22:09

ripienaar mentioned this pull request Jan 27, 2023

Mappings breaking changes #3823

Merged

aricart mentioned this pull request Feb 14, 2023

Subject mapping transforms for Streams nats-io/nats-architecture-and-design#201

Open

15 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[ADDED] Stream subject mapping transforms #3814

[ADDED] Stream subject mapping transforms #3814

jnmoyne commented Jan 25, 2023 •

edited

derekcollison left a comment

derekcollison Jan 25, 2023

jnmoyne Jan 26, 2023

derekcollison Jan 25, 2023

jnmoyne Jan 26, 2023

derekcollison Jan 25, 2023

jnmoyne Jan 26, 2023 •

edited

ripienaar Jan 26, 2023

jnmoyne Jan 26, 2023 •

edited

jnmoyne Jan 26, 2023

ripienaar Jan 26, 2023

jnmoyne Jan 26, 2023

jnmoyne Jan 26, 2023

derekcollison Jan 25, 2023

jnmoyne Jan 26, 2023

derekcollison Jan 25, 2023

jnmoyne Jan 26, 2023

derekcollison commented Jan 26, 2023

jnmoyne commented Jan 26, 2023 via email

jnmoyne commented Jan 26, 2023

derekcollison left a comment

bruth commented Jan 29, 2023

jnmoyne commented Jan 29, 2023

derekcollison commented Jan 29, 2023

bruth commented Jan 29, 2023

jnmoyne commented Jan 30, 2023


		// Helper to ingest and index the subjectTransform destination token (e.g. $x or {{}}) in the token
		// returns a transformation type, and three function arguments: an array of source subject token indexes, and a single number (e.g. number of partitions, or a slice size), and a string (e.g.a split delimiter)

[ADDED] Stream subject mapping transforms #3814

[ADDED] Stream subject mapping transforms #3814

Conversation

jnmoyne commented Jan 25, 2023 • edited

derekcollison left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jnmoyne Jan 26, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jnmoyne Jan 26, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

derekcollison commented Jan 26, 2023

jnmoyne commented Jan 26, 2023 via email

jnmoyne commented Jan 26, 2023

derekcollison left a comment

Choose a reason for hiding this comment

bruth commented Jan 29, 2023

jnmoyne commented Jan 29, 2023

derekcollison commented Jan 29, 2023

bruth commented Jan 29, 2023

jnmoyne commented Jan 30, 2023

jnmoyne commented Jan 25, 2023 •

edited

jnmoyne Jan 26, 2023 •

edited

jnmoyne Jan 26, 2023 •

edited