Replace priority queue with limited queue to prevent OOM polling issue #392

Sarah-Singh16 · 2025-05-13T23:10:11Z

Why I did it

fixes #26

How I did it

Replaced all priority queues with limited queues so that when limit is reached, a final end-process message is sent to the user and the process is stopped. The limited queue will check if the current queue size has reached the specified maximum queue size (Default 100MB). If the queue has reached its maximum size, it will return a Subscribe output queue exhausted error.

How to verify it

Connect telegraf to sonic device. Run telegraf on lower CPU limit ~20% and poll in SAMPLE mode.

UT: Shows Subscribe output queue exhausted error when item value queue size reaches the maximum output queue size (Default 100MB)

2025-07-25T19:04:12Z D! [inputs.gnmi] Connection to gNMI device 100.94.113.225:8080 closed
2025-07-25T19:04:12Z E! [inputs.gnmi] Error in plugin: aborted gNMI subscription: rpc error: code = InvalidArgument desc = Queue error:  rpc error: code = ResourceExhausted desc = Subscribe output queue exhausted

UT2: Shows Subscribe output queue exhausted error when item notification queue size reaches the maximum output queue size (Default 100MB)

2025-07-25T19:15:45Z D! [inputs.gnmi] Connection to gNMI device 100.94.113.225:8080 closed
2025-07-25T19:15:45Z E! [inputs.gnmi] Error in plugin: aborted gNMI subscription: rpc error: code = InvalidArgument desc = Subscribe output queue exhausted

UT3: Shows Maximum number of subscriptions reached error when the amount of subscription clients reaches the maximum client amount (Default 10)

2025-07-25T19:18:32Z D! [inputs.gnmi] Connection to gNMI device 100.94.113.225:8080 established
2025-07-25T19:18:33Z D! [inputs.gnmi] Connection to gNMI device 100.94.113.225:8080 closed
2025-07-25T19:18:33Z E! [inputs.gnmi] Error in plugin: aborted gNMI subscription: rpc error: code = ResourceExhausted desc = Maximum number of subscriptions reached

Which release branch to backport (provide reason below if selected)

Description for the changelog

Link to config_db schema for YANG module changes

A picture of a cute animal (not mandatory but encouraged)

linux-foundation-easycla · 2025-05-13T23:10:17Z

The committers listed above are authorized under a signed CLA.

✅ login: Sarah-Singh16 / name: Sarah Singh (baeac70, b475b8e, 38fc2a1, 48ce988, d0e0b34, 6914692)

mssonicbld · 2025-05-13T23:10:19Z

/azp run

azure-pipelines · 2025-05-13T23:10:30Z

Azure Pipelines successfully started running 1 pipeline(s).

… issue

Sarah-Singh16 · 2025-05-27T23:23:49Z

/azp run

azure-pipelines · 2025-05-27T23:23:55Z

Commenter does not have sufficient privileges for PR 392 in repo sonic-net/sonic-gnmi

… issue

mssonicbld · 2025-05-28T00:00:52Z

/azp run

azure-pipelines · 2025-05-28T00:01:31Z

Azure Pipelines successfully started running 1 pipeline(s).

sachinholla · 2025-05-29T10:27:34Z

gnmi_server/client_subscribe.go

@@ -15,6 +14,10 @@ import (
 	"sync"
 )

+var (
+	OutputQueSize uint64


No value is assigned to this global variable. So, NewLimitedQueue is always created with maxsize 0 (line 48). Does is mean unlimted size?

sachinholla · 2025-05-29T10:55:25Z

sonic_data_client/db_client.go

@@ -172,6 +176,56 @@ type DbClient struct {
 	errors  int64
 }

+var queueLengthSum uint64


Can you move all LimitedQueue related code to separate file? It is not a db_client specific utility

sachinholla · 2025-05-29T11:19:35Z

sonic_data_client/db_client.go

+
+type LimitedQueue struct {
+	Q       *queue.PriorityQueue
+	maxSize uint64


Queue size is global but the maxsize is per-subscribe value. What is the usecase?

sachinholla · 2025-05-29T12:06:52Z

sonic_data_client/db_client.go

+func (q *LimitedQueue) EnqueueItem(item Value) error {
+	queueLengthLock.Lock()
+	defer queueLengthLock.Unlock()
+	ilen := uint64(len(item.Val.GetJsonIetfVal()))


Cannot assume the value will always be ietf json encoded. Translib client uses the scalar values as prescribed by the gNMI spec -- fills the Notification field. Other clients are filling string & proto_bytes too.

There is a proto.Size() API to calculate message size. But calling it here may be costly. Needs some benchmarking. Alternatively we can think of limiting by number of messages.

sachinholla · 2025-05-29T12:10:52Z

sonic_data_client/db_client.go

+		return q.Q.Put(item)
+	} else {
+		log.Error("Telemetry output queue full, discarding item!")
+		return nil


Issue #26 talks about closing the subscription with RESOURCE_EXHAUSTED status.

… issue

mssonicbld · 2025-07-22T00:11:40Z

/azp run

azure-pipelines · 2025-07-22T00:11:49Z

Azure Pipelines successfully started running 1 pipeline(s).

sachinholla · 2025-07-23T07:00:57Z

sonic_data_client/db_client.go

@@ -173,6 +176,55 @@ type DbClient struct {
 	errors  int64
 }

+type LimitedQueue struct {


Consider moving to a separate file, say queue.go

sachinholla · 2025-07-23T07:04:36Z

sonic_data_client/db_client.go

+func (q *LimitedQueue) EnqueueItem(item Value) error {
+	q.queueLengthLock.Lock()
+	defer q.queueLengthLock.Unlock()
+	ilen := (uint64)(proto.Size(item.Notification))


Not all data clients populate Value.Notification field -- AFAIK only the TranslClient uses it. Queue limit checks will not be applied to subscriptions served by other data clients

sachinholla · 2025-07-23T07:14:15Z

sonic_data_client/db_client.go

-	putFatalMsg(c.q, msg)
-}
-
-func putFatalMsg(q *queue.PriorityQueue, msg string) {


Generic putFatalMsg() function is better than having separate enqueueFatalMsgXYZ functions per data client. You can even consider defining the generic function as a member of LimitedQueue struct.

sachinholla · 2025-07-23T07:20:46Z

sonic_data_client/db_client.go

-
-func putFatalMsg(q *queue.PriorityQueue, msg string) {
-	q.Put(Value{
+	log.Error(msg)


Use log.ErrorDepth(1, msg) to preserve caller's file/line info

sachinholla · 2025-07-23T07:23:32Z

sonic_data_client/db_client.go

@@ -330,7 +394,7 @@ func (c *DbClient) AppDBPollRun(q *queue.PriorityQueue, poll chan struct{}, w *s
 							},
 						},
 					}
-					c.q.Put(Value{spbv})
+					c.q.EnqueueItem(Value{spbv})


Possible "queue full" error is ignored

sachinholla · 2025-07-23T07:26:11Z

sonic_data_client/transl_subscriber.go

+					if st.Code() == codes.ResourceExhausted {
+						enqueFatalMsgTranslib(c, st.Message())
+					}
+				}
 				log.Warning(err)


Duplicate log.. enqueFatalMsgTranslib() logs it again

sachinholla · 2025-07-23T07:26:54Z

sonic_data_client/transl_subscriber.go

@@ -186,8 +193,16 @@ func (ts *translSubscriber) notify(v *translib.SubscribeResponse) error {
 	}

 	spbv := &spb.Value{Notification: msg}
-	ts.client.q.Put(Value{spbv})
+	ts.client.q.EnqueueItem(Value{spbv})


Duplicate EnqueueItem() call.. There is one more in line 198, which seems to be the correct one

sachinholla · 2025-07-23T07:41:31Z

telemetry/telemetry.go

@@ -175,6 +177,8 @@ func setupFlags(fs *flag.FlagSet) (*TelemetryConfig, *gnmi.Config, error) {
 		Vrf:                   fs.String("vrf", "", "VRF name, when zmq_address belong on a VRF, need VRF name to bind ZMQ."),
 		EnableCrl:             fs.Bool("enable_crl", false, "Enable certificate revocation list"),
 		CrlExpireDuration:     fs.Int("crl_expire_duration", 86400, "Certificate revocation list cache expire duration"),
+		OutputQueSz:           fs.Uint64("output_queue_size", 100, "Output Queue Maximum Size (MB)"),


As of now this limit is only applied to notification messages and not for Get response messages. Consider renaming this option as "notification_queue_limit"

sachinholla · 2025-07-23T07:52:11Z

Hi @qiluo-msft, @ganglyu -- this change touches all db clients. Please take a look

sachinholla · 2025-07-23T08:45:16Z

sonic_data_client/transl_subscriber.go

+			return err
+		}
+	} else {
+		log.V(6).Infof("Added spbv #%v", spbv)


This log is never reached -- status.FromError() returns ok=true for nil errors.
See https://pkg.go.dev/google.golang.org/grpc/status#FromError

sachinholla · 2025-07-23T08:47:02Z

sonic_data_client/transl_subscriber.go

@@ -164,6 +166,11 @@ func (ts *translSubscriber) processResponses(q *queue.PriorityQueue) {
 			}

 			if err := ts.notify(v); err != nil {
+				if st, ok := status.FromError(err); ok {
+					if st.Code() == codes.ResourceExhausted {
+						enqueFatalMsgTranslib(c, st.Message())


Stop the loop after adding a fatal message

… issue

mssonicbld · 2025-07-25T19:30:44Z

/azp run

azure-pipelines · 2025-07-25T19:30:55Z

Azure Pipelines successfully started running 1 pipeline(s).

Replace priority queue with limited queue to prevent OOM slow polling…

baeac70

… issue

Replace priority queue with limited queue to prevent OOM slow polling…

38fc2a1

… issue

sachinholla suggested changes May 29, 2025

View reviewed changes

Sarah-Singh16 added 2 commits July 21, 2025 23:26

Merge branch 'master' into add_limited_queue_for_oom_fix

d0e0b34

Replace priority queue with limited queue to prevent OOM slow polling…

b475b8e

… issue

Sarah-Singh16 force-pushed the add_limited_queue_for_oom_fix branch from 7f5a572 to b475b8e Compare July 22, 2025 00:11

sachinholla suggested changes Jul 23, 2025

View reviewed changes

ganglyu requested a review from zbud-msft July 23, 2025 07:54

sachinholla suggested changes Jul 23, 2025

View reviewed changes

Sarah-Singh16 added 2 commits July 25, 2025 18:38

Merge branch 'master' into add_limited_queue_for_oom_fix

6914692

Replace priority queue with limited queue to prevent OOM slow polling…

48ce988

… issue

Sarah-Singh16 requested a review from sachinholla July 25, 2025 19:33

Replace priority queue with limited queue to prevent OOM polling issue #392

Are you sure you want to change the base?

Replace priority queue with limited queue to prevent OOM polling issue #392

Uh oh!

Conversation

Sarah-Singh16 commented May 13, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Why I did it

How I did it

How to verify it

Which release branch to backport (provide reason below if selected)

Description for the changelog

Link to config_db schema for YANG module changes

A picture of a cute animal (not mandatory but encouraged)

Uh oh!

linux-foundation-easycla bot commented May 13, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mssonicbld commented May 13, 2025

Uh oh!

azure-pipelines bot commented May 13, 2025

Uh oh!

Sarah-Singh16 commented May 27, 2025

Uh oh!

azure-pipelines bot commented May 27, 2025

Uh oh!

mssonicbld commented May 28, 2025

Uh oh!

azure-pipelines bot commented May 28, 2025

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!

mssonicbld commented Jul 22, 2025

Uh oh!

azure-pipelines bot commented Jul 22, 2025

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!

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!

sachinholla commented Jul 23, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mssonicbld commented Jul 25, 2025

Uh oh!

azure-pipelines bot commented Jul 25, 2025

Uh oh!

Uh oh!

Sarah-Singh16 commented May 13, 2025 •

edited

Loading

linux-foundation-easycla bot commented May 13, 2025 •

edited

Loading