Multiple consumer subject filters #3500

Jarema · 2022-09-27T09:48:05Z

Assumptions:

Separate single filter and multiple filters logic to not break anything in current codebase (not all done)
Make it reasonable performant, but not at the cost of increasing complexity (a first pass)

What's missing:

Make sure all tests pass
Add more coverage for multiple subject filters edge cases

ripienaar · 2022-09-27T12:33:14Z

This would allow bypassing the recently added filter specific consumer create subject mechanism - I personally dont think that's fit for purpose and should be remove - but anyway it needs consideration.

derekcollison · 2022-09-29T14:09:13Z

This would allow bypassing the recently added filter specific consumer create subject mechanism - I personally dont think that's fit for purpose and should be remove - but anyway it needs consideration.

Any thoughts on how to secure without a callout architecture?

ripienaar · 2022-09-29T14:37:40Z

Any thoughts on how to secure without a callout architecture?

not right now as my thinking is around the larger roadmap item, but I know the model we have now will not work long term and as we already know have significant holes and bypasses via consumer update to name one. It's unfortunate that clients already started adopting the new subject as there's no way this will work long term

derekcollison · 2022-09-29T21:12:40Z

Any thoughts on how to secure without a callout architecture?

not right now as my thinking is around the larger roadmap item, but I know the model we have now will not work long term and as we already know have significant holes and bypasses via consumer update to name one. It's unfortunate that clients already started adopting the new subject as there's no way this will work long term

With correct permissions works quite well for what we wanted for a large customer.

ppd0705 · 2022-10-26T07:15:53Z

I'm a new user of nats. in my usage scenario I really need this feature, otherwise I need to create multiple consumers.

derekcollison · 2022-10-26T18:50:21Z

This will most likely make it in for 2.10.0 release..

In the meantime, you could have the subjects in separate streams and source them into a single one in which you connect your consumer.

Or have a common prefix for the ones you want to bundle together and use wildcard matching as it exists today.

derekcollison · 2022-12-20T14:58:38Z

Rebase please don't merge fir feature branches.

server/consumer.go

server/jetstream_test.go

server/consumer.go

piotrpio

LGTM!

server/errors.json

server/jetstream_test.go

derekcollison · 2023-01-23T20:51:43Z

server/jetstream_test.go

+	})
+	require_NoError(t, err)
+
+	js.Publish("other", []byte(fmt.Sprintf("%d", 100)))


derekcollison · 2023-01-23T20:56:20Z

server/stream.go

 		if preq != nil && !o.isFilteredMatch(preq.Subject) {
+			o.mu.RUnlock()


Maybe capture in variable to simplify locking.

o.mu.RLock() doPurge := preq == nil || o.isFilteredMatch(preq.Subject) o.mu.RUnlock() if doPurge { o.purge(fseq, lseq) }

server/stream.go

server/consumer.go

derekcollison · 2023-01-25T21:07:45Z

server/consumer.go

+			}
+		}
+	}
+	if subjectFilters[0] != _EMPTY_ {


Can subjectFilters be empty slice or nil?

We should check len(subjectFilters) > 0

It can't.

func subjectFilters(filter string, filters []string) []string { if len(filters) == 0 { return []string{filter} } return filters }

Will always return slice of len >= 1, but I don't like that part of the code too, as that's implicit and simple change in that function can break it.

What about changing subjectFilters to:

// Utility for simpler if conditions in Consumer config checks. // In future iteration, we can immediately create `o.subjf` and // use it to validate things. func subjectFilters(filter string, filters []string) ([]string, bool) { if len(filters) == 0 { return []string{filter}, filter == _EMPTY_ } return filters, filters[0] == _EMPTY_ }

That simplifies most uses of it and removes the assumption that len() > 0.

I am not a fan of that, if not filters subjf should be nil, and check for len(subjf) > 0 should always return true iff there are filters.

Suggest we rework that.

What about..

func subjectFilters(filter string, filters []string) ([]string, bool) { if filter != _EMPTY_ { filters = append(filters, filter) } return filters, len(filters) > 0 }

Doing the rework now.

derekcollison · 2023-01-25T21:09:57Z

server/consumer.go

+		sub := &subjectFilter{
+			subject: filter,
+		}
+		if filter != _EMPTY_ && subjectHasWildcard(filter) {


Why could filter be empty?

If we change above this can be simplified.

server/consumer.go

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

server/consumer.go

@@ -988,8 +1019,9 @@ func (o *consumer) setLeader(isLeader bool) {
 		}

 		// Update the group on the our starting sequence if we are starting but we skipped some in the stream.
+		// TODO(tp): check how we should approach this for multiple filters. Ask Derek about this one.


sseq should be proper starting point, so code below should be fine regardless of multiple subjects.

Can remove comment.

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

server/consumer.go

@@ -3942,20 +4126,37 @@ func (o *consumer) selectStartingSeqNo() {

 		if state.FirstSeq == 0 {
 			o.sseq = 1
+			for i := range o.subjf {


Is subjf ptr now? Just use normal range if pointer vs using index.

ok let's switch to updating the value?

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

server/consumer.go

+				o.subjf[i].nextSeq = 1
+			}
+			if filter.nextSeq < state.FirstSeq {
+				o.subjf[i].nextSeq = state.FirstSeq


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

server/consumer.go

+				o.subjf[i].nextSeq = state.FirstSeq
+			}
+			if filter.nextSeq > state.LastSeq {
+				o.subjf[i].nextSeq = state.LastSeq + 1


server/consumer.go

derekcollison · 2023-01-25T21:24:48Z

server/stream.go

@@ -4610,6 +4612,11 @@ func (mset *stream) setConsumer(o *consumer) {
 	if o.cfg.FilterSubject != _EMPTY_ {
 		mset.numFilter++
 	}
+	for _, subject := range o.cfg.FilterSubjects {


Can it be valid after checking config that an entry be == EMPTY?

if you pass `filterSubjects: []string{"", "two"} it will fail the validation and return overlapping subjects error.

So here subject can not ever equal EMPTY correct?

correct, I now treat this as an error (client passing []string{""}

wallyqs · 2023-01-26T03:12:20Z

server/jetstream_errors_generated.go

@@ -536,6 +545,9 @@ var (
 		JSStreamWrongLastSequenceErrF:              {Code: 400, ErrCode: 10071, Description: "wrong last sequence: {seq}"},
 		JSTempStorageFailedErr:                     {Code: 500, ErrCode: 10072, Description: "JetStream unable to open temp storage for restore"},
 		JSTemplateNameNotMatchSubjectErr:           {Code: 400, ErrCode: 10073, Description: "template name in subject does not match request"},
+		JsConsumerDuplicateFilterSubjects:          {Code: 400, ErrCode: 10134, Description: "Consumer cannot have both FilterSubject and FilterSubjects specified"},


Suggested change

JsConsumerDuplicateFilterSubjects: {Code: 400, ErrCode: 10134, Description: "Consumer cannot have both FilterSubject and FilterSubjects specified"},

JsConsumerDuplicateFilterSubjects: {Code: 400, ErrCode: 10134, Description: "consumer cannot have both FilterSubject and FilterSubjects specified"},

wallyqs · 2023-01-26T03:12:29Z

server/jetstream_errors_generated.go

@@ -536,6 +545,9 @@ var (
 		JSStreamWrongLastSequenceErrF:              {Code: 400, ErrCode: 10071, Description: "wrong last sequence: {seq}"},
 		JSTempStorageFailedErr:                     {Code: 500, ErrCode: 10072, Description: "JetStream unable to open temp storage for restore"},
 		JSTemplateNameNotMatchSubjectErr:           {Code: 400, ErrCode: 10073, Description: "template name in subject does not match request"},
+		JsConsumerDuplicateFilterSubjects:          {Code: 400, ErrCode: 10134, Description: "Consumer cannot have both FilterSubject and FilterSubjects specified"},
+		JsConsumerMultipleFiltersNotAllowed:        {Code: 400, ErrCode: 10135, Description: "Consumer with multiple subject filters cannot use subject based API"},


Suggested change

JsConsumerMultipleFiltersNotAllowed: {Code: 400, ErrCode: 10135, Description: "Consumer with multiple subject filters cannot use subject based API"},

JsConsumerMultipleFiltersNotAllowed: {Code: 400, ErrCode: 10135, Description: "consumer with multiple subject filters cannot use subject based API"},

wallyqs · 2023-01-26T03:12:38Z

server/jetstream_errors_generated.go

@@ -536,6 +545,9 @@ var (
 		JSStreamWrongLastSequenceErrF:              {Code: 400, ErrCode: 10071, Description: "wrong last sequence: {seq}"},
 		JSTempStorageFailedErr:                     {Code: 500, ErrCode: 10072, Description: "JetStream unable to open temp storage for restore"},
 		JSTemplateNameNotMatchSubjectErr:           {Code: 400, ErrCode: 10073, Description: "template name in subject does not match request"},
+		JsConsumerDuplicateFilterSubjects:          {Code: 400, ErrCode: 10134, Description: "Consumer cannot have both FilterSubject and FilterSubjects specified"},
+		JsConsumerMultipleFiltersNotAllowed:        {Code: 400, ErrCode: 10135, Description: "Consumer with multiple subject filters cannot use subject based API"},
+		JsConsumerOverlappingSubjectFilters:        {Code: 400, ErrCode: 10136, Description: "Consumer subject filters cannot overlap"},


Suggested change

JsConsumerOverlappingSubjectFilters: {Code: 400, ErrCode: 10136, Description: "Consumer subject filters cannot overlap"},

JsConsumerOverlappingSubjectFilters: {Code: 400, ErrCode: 10136, Description: "consumer subject filters cannot overlap"},

derekcollison · 2023-02-12T15:45:11Z

server/stream.go

-	if o.cfg.FilterSubject != _EMPTY_ {
-		mset.numFilter++
+	if o.subjf != nil {
+		mset.numFilter += len(o.subjf)


This just is number of filtered consumers, so still could be just ++ here.

so would suggest.

if len(o.subjf) > 0 { mset.numFilter++ }

ah! Thought it's number of filters.
Fixed and added comment to struct field.

derekcollison · 2023-02-12T15:48:56Z

server/stream.go

-	}
-	if newFilter != _EMPTY_ {
-		mset.numFilter++
+	mset.numFilter += len(newFilters)


I would just check once all updated, if len(o.subjf) == 0 { mset.numFilter-- }

derekcollison · 2023-02-12T15:53:01Z

server/stream.go

+func (mset *stream) partitionUnique(partitions []string) bool {
+	for _, partition := range partitions {
+		for _, o := range mset.consumers {
+


Prefer tighter, so no new line here I think. Also can remove check for nil since not in fast path range setup ok.

derekcollison

One small change and we should be good to go.

derekcollison · 2023-02-13T08:32:25Z

server/stream.go

@@ -4625,7 +4625,7 @@ func (mset *stream) numConsumers() int {
 func (mset *stream) setConsumer(o *consumer) {
 	mset.consumers[o.name] = o
 	if o.subjf != nil {


Would do len(o.subjf) > 0.. I think that is more succinct.

Also would match decrement version.

derekcollison

LGTM

wallyqs · 2023-02-13T08:49:51Z

server/jetstream_errors_generated.go

+	JsConsumerMultipleFiltersNotAllowed ErrorIdentifier = 10137
+
+	// JsConsumerOverlappingSubjectFilters consumer subject filters cannot overlap
+	JsConsumerOverlappingSubjectFilters ErrorIdentifier = 10138


Should this be like the others above to be consistent, like:

Suggested change

JsConsumerOverlappingSubjectFilters ErrorIdentifier = 10138

JSConsumerOverlappingSubjectFilters ErrorIdentifier = 10138

wallyqs · 2023-02-13T08:50:02Z

server/jetstream_errors_generated.go

+	JsConsumerDuplicateFilterSubjects ErrorIdentifier = 10136
+
+	// JsConsumerMultipleFiltersNotAllowed consumer with multiple subject filters cannot use subject based API
+	JsConsumerMultipleFiltersNotAllowed ErrorIdentifier = 10137


Suggested change

JsConsumerMultipleFiltersNotAllowed ErrorIdentifier = 10137

JSConsumerMultipleFiltersNotAllowed ErrorIdentifier = 10137

wallyqs · 2023-02-13T08:50:10Z

server/jetstream_errors_generated.go

@@ -406,6 +409,15 @@ const (

 	// JSTemplateNameNotMatchSubjectErr template name in subject does not match request
 	JSTemplateNameNotMatchSubjectErr ErrorIdentifier = 10073
+
+	// JsConsumerDuplicateFilterSubjects consumer cannot have both FilterSubject and FilterSubjects specified
+	JsConsumerDuplicateFilterSubjects ErrorIdentifier = 10136


Suggested change

JsConsumerDuplicateFilterSubjects ErrorIdentifier = 10136

JSConsumerDuplicateFilterSubjects ErrorIdentifier = 10136

wallyqs · 2023-02-13T08:50:50Z

server/jetstream_errors_generated.go

@@ -2124,3 +2150,33 @@ func NewJSTemplateNameNotMatchSubjectError(opts ...ErrorOption) *ApiError {

 	return ApiErrors[JSTemplateNameNotMatchSubjectErr]
 }
+
+// NewJsConsumerDuplicateFilterSubjectsError creates a new JsConsumerDuplicateFilterSubjects error: "consumer cannot have both FilterSubject and FilterSubjects specified"
+func NewJsConsumerDuplicateFilterSubjectsError(opts ...ErrorOption) *ApiError {


Suggested change

func NewJsConsumerDuplicateFilterSubjectsError(opts ...ErrorOption) *ApiError {

func NewJSConsumerDuplicateFilterSubjectsError(opts ...ErrorOption) *ApiError {

wallyqs · 2023-02-13T08:50:59Z

server/jetstream_errors_generated.go

+}
+
+// NewJsConsumerOverlappingSubjectFiltersError creates a new JsConsumerOverlappingSubjectFilters error: "consumer subject filters cannot overlap"
+func NewJsConsumerOverlappingSubjectFiltersError(opts ...ErrorOption) *ApiError {


Suggested change

func NewJsConsumerOverlappingSubjectFiltersError(opts ...ErrorOption) *ApiError {

func NewJSConsumerOverlappingSubjectFiltersError(opts ...ErrorOption) *ApiError {

I think @Jarema will do in separate PR, but great feedback!

wallyqs · 2023-02-13T08:51:54Z

server/jetstream_test.go

@@ -9387,7 +9387,7 @@ func TestJetStreamPubWithSyncPerf(t *testing.T) {

 func TestJetStreamConsumerPerf(t *testing.T) {
 	// Comment out to run, holding place for now.
-	t.SkipNow()
+	// t.SkipNow()


maybe meant to leave it uncommented to not run the consumer perf?

Good catch!

Jarema changed the base branch from main to dev December 14, 2022 12:17

Jarema force-pushed the multiple-consumer-filters branch 3 times, most recently from 822e1a5 to 0090ee1 Compare January 6, 2023 09:59

Jarema force-pushed the multiple-consumer-filters branch 2 times, most recently from 7982181 to 5f2ea90 Compare January 16, 2023 14:50

piotrpio reviewed Jan 18, 2023

View reviewed changes

Jarema force-pushed the multiple-consumer-filters branch 2 times, most recently from 756885f to 3a36f36 Compare January 18, 2023 12:31

This was referenced Jan 18, 2023

Add Consumers with Multiple Filters ADR-34 nats-io/nats-architecture-and-design#192

Merged

Add support for consumers with multiple filters nats-io/nats.rs#814

Merged

piotrpio reviewed Jan 23, 2023

View reviewed changes

piotrpio approved these changes Jan 23, 2023

View reviewed changes

Jarema force-pushed the multiple-consumer-filters branch 2 times, most recently from 82ffb54 to 4d208d5 Compare January 23, 2023 13:56

Jarema marked this pull request as ready for review January 23, 2023 13:56

Jarema requested a review from derekcollison January 23, 2023 13:56

Jarema changed the title ~~[DRAFT WIP] Multiple consumer subject filters~~ Multiple consumer subject filters Jan 23, 2023

derekcollison requested changes Jan 23, 2023

View reviewed changes

Jarema force-pushed the multiple-consumer-filters branch from c7d7201 to 03d702c Compare January 25, 2023 15:31

derekcollison self-requested a review January 25, 2023 20:54

derekcollison requested changes Jan 25, 2023

View reviewed changes

wallyqs reviewed Jan 26, 2023

View reviewed changes

Jarema force-pushed the multiple-consumer-filters branch 5 times, most recently from 6754604 to 618027b Compare February 1, 2023 21:57

Jarema requested a review from derekcollison February 1, 2023 22:59

Jarema force-pushed the multiple-consumer-filters branch 3 times, most recently from e94752e to 123ec50 Compare February 6, 2023 22:10

derekcollison requested changes Feb 12, 2023

View reviewed changes

Jarema force-pushed the multiple-consumer-filters branch from 75a2476 to 33eb6a9 Compare February 13, 2023 07:12

derekcollison self-requested a review February 13, 2023 08:31

derekcollison requested changes Feb 13, 2023

View reviewed changes

Add multiple subject filters

af338d0

Jarema force-pushed the multiple-consumer-filters branch from 33eb6a9 to af338d0 Compare February 13, 2023 08:38

derekcollison self-requested a review February 13, 2023 08:42

derekcollison approved these changes Feb 13, 2023

View reviewed changes

derekcollison merged commit fcd6f87 into nats-io:dev Feb 13, 2023

wallyqs reviewed Feb 13, 2023

View reviewed changes

Jarema mentioned this pull request Mar 5, 2023

Jetstream: Support for multiple "filter subject" patterns in consumer #2515

Closed

cjbottaro mentioned this pull request Apr 21, 2023

Only one consumer allowed when using filter_subjects #4081

Closed

kindermoumoute mentioned this pull request May 3, 2023

[Jetstream] Add MaxBytesPerSubject stream configuration #4128

Open

sravan-s mentioned this pull request Aug 13, 2023

[draft] NATS AyWa/quickwit#2

Draft

3 tasks

		if preq != nil && !o.isFilteredMatch(preq.Subject) {
		o.mu.RUnlock()

	JsConsumerDuplicateFilterSubjects: {Code: 400, ErrCode: 10134, Description: "Consumer cannot have both FilterSubject and FilterSubjects specified"},
	JsConsumerDuplicateFilterSubjects: {Code: 400, ErrCode: 10134, Description: "consumer cannot have both FilterSubject and FilterSubjects specified"},

	JsConsumerMultipleFiltersNotAllowed: {Code: 400, ErrCode: 10135, Description: "Consumer with multiple subject filters cannot use subject based API"},
	JsConsumerMultipleFiltersNotAllowed: {Code: 400, ErrCode: 10135, Description: "consumer with multiple subject filters cannot use subject based API"},

	JsConsumerOverlappingSubjectFilters: {Code: 400, ErrCode: 10136, Description: "Consumer subject filters cannot overlap"},
	JsConsumerOverlappingSubjectFilters: {Code: 400, ErrCode: 10136, Description: "consumer subject filters cannot overlap"},

	JsConsumerOverlappingSubjectFilters ErrorIdentifier = 10138
	JSConsumerOverlappingSubjectFilters ErrorIdentifier = 10138

	JsConsumerMultipleFiltersNotAllowed ErrorIdentifier = 10137
	JSConsumerMultipleFiltersNotAllowed ErrorIdentifier = 10137

	JsConsumerDuplicateFilterSubjects ErrorIdentifier = 10136
	JSConsumerDuplicateFilterSubjects ErrorIdentifier = 10136

	func NewJsConsumerDuplicateFilterSubjectsError(opts ...ErrorOption) *ApiError {
	func NewJSConsumerDuplicateFilterSubjectsError(opts ...ErrorOption) *ApiError {

	func NewJsConsumerOverlappingSubjectFiltersError(opts ...ErrorOption) *ApiError {
	func NewJSConsumerOverlappingSubjectFiltersError(opts ...ErrorOption) *ApiError {

Multiple consumer subject filters #3500

Multiple consumer subject filters #3500

Conversation

Jarema commented Sep 27, 2022 • edited

ripienaar commented Sep 27, 2022

derekcollison commented Sep 29, 2022

ripienaar commented Sep 29, 2022

derekcollison commented Sep 29, 2022

ppd0705 commented Oct 26, 2022

derekcollison commented Oct 26, 2022

derekcollison commented Dec 20, 2022

piotrpio 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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Jarema Jan 25, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Jarema Jan 27, 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

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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Jarema commented Sep 27, 2022 •

edited

Jarema Jan 25, 2023 •

edited

Jarema Jan 27, 2023 •

edited