feat(inputs.nats_consumer): Ack messages when accepted on the output #17792

Hipska · 2025-10-09T10:16:17Z

Summary

NATS Jetstream has support for message acknowledging, so make use of that.

Checklist

No AI generated code was used in this PR

Related issues

resolves #17791

# Conflicts: # plugins/inputs/nats_consumer/nats_consumer.go

srebhan

Thanks @Hipska for the PR, however, there are major issues!

The n.undelivered list of in-flight metrics is never written to!
Usually I would assume a separate onDelivery function that ACKs the message and removes the message from the sem channel instead of mixing this into the receiver function. The onDelivery function should then just run in its own go-routine.
The n.undelivered list of in-flight metrics requires locking...
Unit-tests are missing!

plugins/inputs/nats_consumer/nats_consumer.go

srebhan · 2025-10-10T08:10:13Z

plugins/inputs/nats_consumer/nats_consumer.go

 	JsStream               string          `toml:"jetstream_stream"`
 	PendingMessageLimit    int             `toml:"pending_message_limit"`
-	PendingBytesLimit      int             `toml:"pending_bytes_limit"`
+	PendingBytesLimit      int             `toml:"pending_bytes_limit" deprecated:"1.37.0;1.40.0;unused"`


Hmmm actually this setting is used in line 160, isn't it?

Whoops, result of wrong merge. Will fix it.

Hmmm still not fixed I think so unresolving my comment...

Where do you still see it used in current code?

Well you are removing this feature from the code and then deprecate the function without giving a reason on why you remove this! Furthermore, even if there is a good reason for removal it does not belong to this PR!

plugins/inputs/nats_consumer/nats_consumer.go

Co-authored-by: Sven Rebhan <[email protected]>

Hipska · 2025-10-10T11:32:36Z

I based the onDelivery function on how its implemented in inputs.amqp_consumer:

telegraf/plugins/inputs/amqp_consumer/amqp_consumer.go

Lines 412 to 415 in 851bea2

    
           case track := <-acc.Delivered(): 
        
           	if a.onDelivery(track) { 
        
           		<-sem 
        
           	}

telegraf/plugins/inputs/amqp_consumer/amqp_consumer.go

Lines 471 to 494 in 851bea2

    
           func (a *AMQPConsumer) onDelivery(track telegraf.DeliveryInfo) bool { 
        
           	delivery, ok := a.deliveries[track.ID()] 
        
           	if !ok { 
        
           		// Added by a previous connection 
        
           		return false 
        
           	} 
        
           	if track.Delivered() { 
        
           		err := delivery.Ack(false) 
        
           		if err != nil { 
        
           			a.Log.Errorf("Unable to ack written delivery: %d: %v", delivery.DeliveryTag, err) 
        
           			a.conn.Close() 
        
           		} 
        
           	} else { 
        
           		err := delivery.Reject(false) 
        
           		if err != nil { 
        
           			a.Log.Errorf("Unable to reject failed delivery: %d: %v", delivery.DeliveryTag, err) 
        
           			a.conn.Close() 
        
           		} 
        
           	} 
        
           	delete(a.deliveries, track.ID()) 
        
           	return true 
        
           }

I added locking as requested, but I see the above mentioned code also doesn't have locking..

srebhan

Thanks @Hipska! The code reads much better with your update. Please drop removing the pending bytes limit feature as it does NOT belong to this PR. Furthermore, there are some corner-cases to consider, see my comments in the code...

plugins/inputs/nats_consumer/nats_consumer.go

srebhan · 2025-10-14T10:07:29Z

plugins/inputs/nats_consumer/nats_consumer.go

 	JsStream               string          `toml:"jetstream_stream"`
 	PendingMessageLimit    int             `toml:"pending_message_limit"`
-	PendingBytesLimit      int             `toml:"pending_bytes_limit"`
+	PendingBytesLimit      int             `toml:"pending_bytes_limit" deprecated:"1.37.0;1.40.0;unused"`


Hmmm still not fixed I think so unresolving my comment...

srebhan · 2025-10-14T10:08:02Z

plugins/inputs/nats_consumer/nats_consumer.go

 func (n *NatsConsumer) Start(acc telegraf.Accumulator) error {
+	n.sem = make(semaphore, n.MaxUndeliveredMessages)
 	n.acc = acc.WithTracking(n.MaxUndeliveredMessages)
+	n.undelivered = make(map[telegraf.TrackingID]*nats.Msg)


How about

Suggested change

n.undelivered = make(map[telegraf.TrackingID]*nats.Msg)

n.undelivered = make(map[telegraf.TrackingID]*nats.Msg, PendingMessageLimit)

Not very sure, I would use MaxUndeliveredMessages instead if needed.

And I'm starting to think they are actually meaning the same..

Well I wonder if those two parameters don't mean the same thing. But yeah if the mean different things I agree to use MaxUndeliveredMessages as this is what is meant here.

plugins/inputs/nats_consumer/nats_consumer.go

srebhan · 2025-10-14T10:22:15Z

plugins/inputs/nats_consumer/nats_consumer.go

+func (n *NatsConsumer) waitForDelivery(parentCtx context.Context) {
+	for {
+		select {
+		case <-parentCtx.Done():
+			return
+		case track := <-n.acc.Delivered():
+			<-n.sem
+			msg := n.removeDelivered(track.ID())
+
+			if msg != nil {
+				if track.Delivered() {
+					err := msg.Ack()
+					if err != nil {
+						n.Log.Errorf("Failed to Ack message on subject %s: %v", msg.Subject, err)
+					}
+				} else {
+					err := msg.Nak()
+					if err != nil {
+						n.Log.Errorf("Failed to Nak message on subject %s: %v", msg.Subject, err)
+					}
+				}
+			}
+		}
+	}
+}
+
+func (n *NatsConsumer) removeDelivered(id telegraf.TrackingID) *nats.Msg {
+	n.Lock()
+	defer n.Unlock()
+
+	msg, ok := n.undelivered[id]
+	if !ok {
+		return nil
+	}
+	delete(n.undelivered, id)
+	return msg
+}


How about

Suggested change

func (n *NatsConsumer) waitForDelivery(parentCtx context.Context) {

for {

select {

case <-parentCtx.Done():

return

case track := <-n.acc.Delivered():

<-n.sem

msg := n.removeDelivered(track.ID())

if msg != nil {

if track.Delivered() {

err := msg.Ack()

if err != nil {

n.Log.Errorf("Failed to Ack message on subject %s: %v", msg.Subject, err)

}

} else {

err := msg.Nak()

if err != nil {

n.Log.Errorf("Failed to Nak message on subject %s: %v", msg.Subject, err)

}

}

}

}

}

}

func (n *NatsConsumer) removeDelivered(id telegraf.TrackingID) *nats.Msg {

n.Lock()

defer n.Unlock()

msg, ok := n.undelivered[id]

if !ok {

return nil

}

delete(n.undelivered, id)

return msg

}

func (n *NatsConsumer) handleDelivery(ctx context.Context) {

for {

select {

case <-ctx.Done():

// Plugin is stopping

return

case track := <-n.acc.Delivered():

// Get the delivered message and remove it from the internal tracking

// mechanism

n.Lock()

msg, found := n.undelivered[id]

delete(n.undelivered, id)

defer n.Unlock()

// Make space for a new message to be received despite any error case

<-n.sem

if !found {

n.Log.Errorf("received delivery event for unknown message %v", id)

continue

}

// Acknowledge the message

if track.Delivered() {

if err := msg.Ack(); err != nil {

n.Log.Errorf("Failed to Ack message on subject %s: %v", msg.Subject, err)

}

} else {

if err := msg.Nak(); err != nil {

n.Log.Errorf("Failed to Nak message on subject %s: %v", msg.Subject, err)

}

}

}

}

}

I'm not sure we should NAK the message here if this means the message is re-queued to be honest. The background is that if only a single metric is rejected (for whatever reason) the whole NATS message is NAK'ed. If the underlying reason for rejecting the metric is permanent, this may trigger an infinite loop for the message as it will always be NAK'ed and thus may render the input dysfunctional if enough of those messages are consumed (i.e. more than the allowed max pending).

Isn't that the whole purpose of tracking metrics?

I don't think it is. The purpose of tracking metrics is to allow an input to know if a message was delivered or is "in flight". If you receive a delivery event it means that the metric was processed by the endpoint (it might be a processor as well, just for clarity). The Delivered function tells you if all messages were delivered successfully.

Now there are different ways to handle "unsuccessful" messages. In PR #15796 we implemented NAK'ing messages for AMQP if delivery fails ensuring that the messages are not re-enqueued in the topic.

Coming back to this PR: If delivery fails, i.e. the metric could not be handled by the output plugin (e.g. serialization fails, format is invalid etc or the endpoint got the metric but something went wrong, and you NAK the message, you must make sure that this message is not sent again, otherwise you end up in an infinite loop of getting the message -> delivery fails -> NAK message - back to square one. Now if you not only have one of those metrics but they do appear sporadically, you will end up with a dysfunctional input only being busy with NAK'ed messages...

So please either make sure the NAK'ed messages are not re-enqueued in the topic or do not NAK those messages!

plugins/inputs/nats_consumer/nats_consumer.go

plugins/inputs/nats_consumer/nats_consumer_test.go

telegraf-tiger · 2025-10-15T15:05:52Z

Download PR build artifacts for linux_amd64.tar.gz, darwin_arm64.tar.gz, and windows_amd64.zip.
Downloads for additional architectures and packages are available below.

☺️ This pull request doesn't significantly change the Telegraf binary size (less than 1%)

📦 Click here to get additional PR build artifacts

Artifact URLs

. DEB	. RPM	. TAR . GZ	. ZIP
amd64.deb	aarch64.rpm	darwin_amd64.tar.gz	windows_amd64.zip
arm64.deb	armel.rpm	darwin_arm64.tar.gz	windows_arm64.zip
armel.deb	armv6hl.rpm	freebsd_amd64.tar.gz	windows_i386.zip
armhf.deb	i386.rpm	freebsd_armv7.tar.gz
i386.deb	ppc64le.rpm	freebsd_i386.tar.gz
mips.deb	riscv64.rpm	linux_amd64.tar.gz
mipsel.deb	s390x.rpm	linux_arm64.tar.gz
ppc64el.deb	x86_64.rpm	linux_armel.tar.gz
riscv64.deb		linux_armhf.tar.gz
s390x.deb		linux_i386.tar.gz
		linux_mips.tar.gz
		linux_mipsel.tar.gz
		linux_ppc64le.tar.gz
		linux_riscv64.tar.gz
		linux_s390x.tar.gz

srebhan

Thanks @Hipska! There are still some issues here:

Why do you remove the PendingBytesLimit option? You did not provide any reason for doing so and this does not belong to this PR. Please keep the option!
You are leaking messages which are not acknowledged if they are non-Jetstream.
NAK'ing messages might re-queue the failed message ending up in an infinite loop blocking the plugin.

Please address those issues!

srebhan · 2025-10-21T14:59:59Z

plugins/inputs/nats_consumer/nats_consumer.go

 func (n *NatsConsumer) Start(acc telegraf.Accumulator) error {
+	n.sem = make(semaphore, n.MaxUndeliveredMessages)
 	n.acc = acc.WithTracking(n.MaxUndeliveredMessages)
+	n.undelivered = make(map[telegraf.TrackingID]*nats.Msg)


Suggested change

n.undelivered = make(map[telegraf.TrackingID]*nats.Msg)

n.undelivered = make(map[telegraf.TrackingID]*nats.Msg, n.MaxUndeliveredMessages)

srebhan · 2025-10-21T15:00:32Z

plugins/inputs/nats_consumer/nats_consumer.go

-			// set the subscription pending limits
-			err = sub.SetPendingLimits(n.PendingMessageLimit, n.PendingBytesLimit)


Why do you remove this feature?

srebhan · 2025-10-21T15:01:55Z

plugins/inputs/nats_consumer/nats_consumer.go

+				for _, s := range n.jsSubs {
+					if msg.Sub == s {
+						n.handleJetstreamMessage(msg)
+						break L
+					}
 				}


Why do you need this? It wasn't required before... This reads like yet another thing folded into this PR!

srebhan · 2025-10-21T15:05:13Z

plugins/inputs/nats_consumer/nats_consumer.go

+					err := msg.Nak()
+					if err != nil {
+						n.Log.Errorf("Failed to Nak JetStream message on subject %s: %v", msg.Subject, err)
+					}


It won't make my concerns go away if you do not address them. :-) How can we make sure that messages with permanent errors are not re-enqueued/re-send by NATS?

srebhan · 2025-10-21T18:58:55Z

plugins/inputs/nats_consumer/nats_consumer.go

+	for _, m := range metrics {
+		m.AddTag("subject", msg.Subject)
+	}
+	return n.acc.AddTrackingMetricGroup(metrics), nil


You are not adding the tracking ID to the n.undelivered lookup so you will never be able to ACK this message!

Hipska added 2 commits October 9, 2025 12:13

feat(inputs.nats_consumer): Ack messages when accepted on the output

0022d04

Change to channel based queue subscription

fe4722b

telegraf-tiger bot added feat Improvement on an existing feature such as adding a new setting/mode to an existing plugin plugin/input 1. Request for new input plugins 2. Issues/PRs that are related to input plugins labels Oct 9, 2025

Hipska added 2 commits October 9, 2025 13:17

Linter fix

2b54502

Merge branch 'master' into feat/inputs/nats_consumer/ack_on_delivery

1628b33

# Conflicts: # plugins/inputs/nats_consumer/nats_consumer.go

srebhan requested changes Oct 10, 2025

View reviewed changes

srebhan self-assigned this Oct 10, 2025

Hipska marked this pull request as draft October 10, 2025 09:21

Hipska and others added 7 commits October 10, 2025 11:24

Apply suggestion from @srebhan

95f15a7

Co-authored-by: Sven Rebhan <[email protected]>

Pending limit not applicable for channel subscriptions

f73359c

Don't use deprecated option.

2d198a9

Actually store undelivered messages

065b6b5

Add tests

824ebbe

Reset value

16fd2ec

Add locking

c05920e

Hipska added 2 commits October 10, 2025 14:04

Rewrite delivery handler.

47e47ae

Add extra locking

b70746d

Hipska marked this pull request as ready for review October 10, 2025 12:29

Try to figure out if message is from a JetStream..

935a5eb

Hipska force-pushed the feat/inputs/nats_consumer/ack_on_delivery branch from 1d31e7e to 935a5eb Compare October 10, 2025 15:05

Hipska added 2 commits October 10, 2025 17:38

Better jetstream detection

f860006

Merge branch 'master' into feat/inputs/nats_consumer/ack_on_delivery

c3beafd

Hipska mentioned this pull request Oct 10, 2025

feat(inputs.nats_consumer): Allow configuring Jetstream stream #17710

Merged

1 task

Hipska requested a review from srebhan October 13, 2025 08:23

Merge branch 'master' into feat/inputs/nats_consumer/ack_on_delivery

aa52017

srebhan requested changes Oct 14, 2025

View reviewed changes

Hipska added 2 commits October 14, 2025 14:39

Terminate message when unable to parse

adef87c

Rename context argument

2716049

Hipska added 2 commits October 14, 2025 15:00

Move sync.Mutex

0f60378

Change error handling

8d0eb11

Hipska requested a review from srebhan October 20, 2025 08:48

srebhan reviewed Oct 21, 2025

View reviewed changes

srebhan added the waiting for response waiting for response from contributor label Oct 23, 2025

	n.undelivered = make(map[telegraf.TrackingID]*nats.Msg)
	n.undelivered = make(map[telegraf.TrackingID]*nats.Msg, PendingMessageLimit)

		// set the subscription pending limits
		err = sub.SetPendingLimits(n.PendingMessageLimit, n.PendingBytesLimit)

feat(inputs.nats_consumer): Ack messages when accepted on the output #17792

Are you sure you want to change the base?

feat(inputs.nats_consumer): Ack messages when accepted on the output #17792

Conversation

Hipska commented Oct 9, 2025

Summary

Checklist

Related issues

Uh oh!

srebhan left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Hipska commented Oct 10, 2025

Uh oh!

srebhan left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Hipska Oct 14, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

srebhan Oct 20, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

telegraf-tiger bot commented Oct 15, 2025

Artifact URLs

Uh oh!

srebhan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Hipska Oct 14, 2025 •

edited

Loading

srebhan Oct 20, 2025 •

edited

Loading