Merge branch 'main' into denopink/feat/optimize-qos-trim-v2

xmidt-org · Aug 5, 2024 · 84bbd43 · 84bbd43
2 parents 1d2bbb6 + a5309a6
commit 84bbd43
Show file tree

Hide file tree

Showing 6 changed files with 24 additions and 11 deletions.
diff --git a/cmd/xmidt-agent/default-config.yaml b/cmd/xmidt-agent/default-config.yaml
@@ -127,8 +127,7 @@ mock_tr_181:
 xmidt_agent_crud:
   service_name: xmidt_agent
 qos:
-  max_queue_bytes:  1048576  # 1 * 1024 * 1024 // 1MB max/queue,
-  max_message_bytes: 262144 # 256 * 1024      // 256 KB
+  max_queue_bytes:  104857600  # 100 * 1024 * 1024 // 100MB max/queue,
   priority: newest
 metadata:
   fields:

diff --git a/internal/wrphandlers/mocktr181/handler.go b/internal/wrphandlers/mocktr181/handler.go
@@ -143,7 +143,7 @@ func (h Handler) HandleWrp(msg wrp.Message) error {
 	default:
 		// currently only get and set are implemented for existing mocktr181
 		statusCode = 520
-		payloadResponse = []byte(fmt.Sprintf(`{"message": "command %s is not support", "statusCode": %d}`, command, statusCode))
+		payloadResponse = []byte(fmt.Sprintf(`{"message": "command '%s' is not supported", "statusCode": %d}`, command, statusCode))
 	}
 
 	response := msg

diff --git a/internal/wrphandlers/qos/options.go b/internal/wrphandlers/qos/options.go
@@ -45,8 +45,6 @@ func MaxMessageBytes(s int) Option {
 		func(h *Handler) error {
 			if s < 0 {
 				return fmt.Errorf("%w: negative MaxMessageBytes", ErrMisconfiguredQOS)
-			} else if s == 0 {
-				s = DefaultMaxMessageBytes
 			}
 
 			h.maxMessageBytes = s

diff --git a/internal/wrphandlers/qos/priority_queue.go b/internal/wrphandlers/qos/priority_queue.go
@@ -25,7 +25,8 @@ type priorityQueue struct {
 	priority PriorityType
 	// tieBreaker breaks any QualityOfService ties.
 	tieBreaker tieBreaker
-	// maxQueueBytes is the allowable max size of the queue based on the sum of all queued wrp message's payloads
+	// maxQueueBytes is the allowable max size of the queue based on the sum of all queued wrp message's payloads.
+	// Zero value will disable individual message size validation.
 	maxQueueBytes int64
 	// MaxMessageBytes is the largest allowable wrp message payload.
 	maxMessageBytes int
@@ -81,7 +82,8 @@ func (pq *priorityQueue) Dequeue() (wrp.Message, bool) {
 // Enqueue queues the given message.
 func (pq *priorityQueue) Enqueue(msg wrp.Message) error {
 	// Check whether msg violates maxMessageBytes.
-	if len(msg.Payload) > pq.maxMessageBytes {
+	// The zero value of `pq.maxMessageBytes` will disable individual message size validation.
+	if pq.maxMessageBytes != 0 && len(msg.Payload) > pq.maxMessageBytes {
 		return fmt.Errorf("%w: %v", ErrMaxMessageBytes, pq.maxMessageBytes)
 	}
 

diff --git a/internal/wrphandlers/qos/priority_queue_test.go b/internal/wrphandlers/qos/priority_queue_test.go
@@ -185,6 +185,12 @@ func testEnqueueDequeue(t *testing.T) {
 			maxMessageBytes:   len(largeCriticalQOSMsg.Payload) - 1,
 			expectedQueueSize: 0,
 		},
+		{
+			description:       "allow any message size",
+			messages:          []wrp.Message{largeCriticalQOSMsg},
+			maxQueueBytes:     len(largeCriticalQOSMsg.Payload),
+			expectedQueueSize: 1,
+		},
 		{
 			description:       "message too large with a nonempty queue",
 			messages:          []wrp.Message{largeCriticalQOSMsg, largeCriticalQOSMsg},

diff --git a/internal/wrphandlers/qos/qos_test.go b/internal/wrphandlers/qos/qos_test.go
@@ -33,7 +33,6 @@ func TestHandler_HandleWrp(t *testing.T) {
 	tests := []struct {
 		description string
 		options     []qos.Option
-		priority    qos.PriorityType
 		// int64 required for nextCallCount atomic.Int64 comparison
 		nextCallCount        int64
 		next                 wrpkit.Handler
@@ -54,6 +53,16 @@ func TestHandler_HandleWrp(t *testing.T) {
 				return nil
 			}),
 		},
+		{
+			description:   "enqueued and delivered message prioritizing newer messages with no message size restriction",
+			options:       []qos.Option{qos.MaxQueueBytes(int64(100)), qos.Priority(qos.NewestType)},
+			nextCallCount: 1,
+			next: wrpkit.HandlerFunc(func(wrp.Message) error {
+				nextCallCount.Add(1)
+
+				return nil
+			}),
+		},
 		{
 			description:   "enqueued and delivered message prioritizing older messages",
 			options:       []qos.Option{qos.MaxQueueBytes(int64(100)), qos.MaxMessageBytes(50), qos.Priority(qos.OldestType)},
@@ -93,9 +102,8 @@ func TestHandler_HandleWrp(t *testing.T) {
 			shouldHalt: true,
 		},
 		{
-			description: "zero MaxQueueBytes option value",
-			options:     []qos.Option{qos.MaxQueueBytes(int64(100)), qos.MaxMessageBytes(50), qos.Priority(qos.NewestType)},
-
+			description:   "zero MaxQueueBytes option value",
+			options:       []qos.Option{qos.MaxQueueBytes(int64(100)), qos.MaxMessageBytes(50), qos.Priority(qos.NewestType)},
 			nextCallCount: 1,
 			next: wrpkit.HandlerFunc(func(wrp.Message) error {
 				nextCallCount.Add(1)