[12.0 stable] Prevent deferred queue from growing

pkg/pillar/cmd/zedagent/zedagent.go

@@ -295,7 +295,7 @@ func queueInfoToDest(ctx *zedagentContext, dest destinationBitset,
 			devUUID, "info")
 		// Ignore errors for all the LOC info messages
 		const ignoreErr = true
-		zedcloudCtx.DeferredPeriodicCtx.SetDeferred(key, buf, size, url,
+		zedcloudCtx.DeferredLOCPeriodicCtx.SetDeferred(key, buf, size, url,
 			bailOnHTTPErr, withNetTracing, ignoreErr, itemType)
 	}
 }
@@ -442,6 +442,9 @@ func Run(ps *pubsub.PubSub, loggerArg *logrus.Logger, logArg *base.LogObject, ar
 	zedcloudCtx.DeferredPeriodicCtx = zedcloud.CreateDeferredCtx(zedcloudCtx,
 		zedagentCtx.ps, agentName, "DeferredPeriodic",
 		warningTime, errorTime, nil)
+	zedcloudCtx.DeferredLOCPeriodicCtx = zedcloud.CreateDeferredCtx(zedcloudCtx,
+		zedagentCtx.ps, agentName, "DeferredLOCPeriodic",
+		warningTime, errorTime, nil)
 	// XXX defer this until we have some config from cloud or saved copy
 	getconfigCtx.pubAppInstanceConfig.SignalRestarted()
 

pkg/pillar/zedcloud/deferred.go

@@ -132,6 +132,33 @@ func (ctx *DeferredContext) processQueueTask(ps *pubsub.PubSub,
 	}
 }
 
+// mergeQueuesNoLock merges requests which were not sent (argument)
+// with incoming requests, accumulated in the `ctx.deferredItems`.
+// Context: `ctx.lock` held.
+func (ctx *DeferredContext) mergeQueuesNoLock(notSentReqs []*deferredItem) {
+	if len(ctx.deferredItems) > 0 {
+		// During the send new items land into the `ctx.deferredItems`
+		// queue, which keys can exist in the `notSentReqs` queue.
+		// Traverse requests which were not sent, find items with same
+		// keys in the `ctx.deferredItems` and replace item in the
+		// `notSentReqs`.
+		for i, oldItem := range notSentReqs {
+			for j, newItem := range ctx.deferredItems {
+				if oldItem.key == newItem.key {
+					// Replace item in head
+					notSentReqs[i] = newItem
+					// Remove from tail
+					ctx.deferredItems =
+						append(ctx.deferredItems[:j], ctx.deferredItems[j+1:]...)
+					break
+				}
+			}
+		}
+	}
+	// Merge the rest adding new items to the tail
+	ctx.deferredItems = append(notSentReqs, ctx.deferredItems...)
+}
+
 // handleDeferred try to send all deferred items
 func (ctx *DeferredContext) handleDeferred() bool {
 	ctx.lock.Lock()
@@ -236,8 +263,7 @@ func (ctx *DeferredContext) handleDeferred() bool {
 	}
 
 	ctx.lock.Lock()
-	// Merge with the incoming requests, recently added are in the tail
-	ctx.deferredItems = append(notSentReqs, ctx.deferredItems...)
+	ctx.mergeQueuesNoLock(notSentReqs)
 	if len(ctx.deferredItems) == 0 {
 		stopTimer(log, ctx)
 	}

pkg/pillar/zedcloud/send.go

@@ -71,15 +71,20 @@ type ZedCloudContext struct {
 	serverSigningCertHash []byte
 	onBoardCertBytes      []byte
 	log                   *base.LogObject
-	// All HTTP requests which can't be dropped and send should be
-	// repeated in case of a transmission error are added to this
-	// queue.
+	// All controller HTTP requests which can't be dropped and send
+	// should be repeated in case of a transmission error are added to
+	// this queue.
 	DeferredEventCtx *DeferredContext
-	// All periodic HTTP requests are added to this queue, sending
-	// errors of which can be ignored. This means even the request has
-	// failed, it will be removed from the queue, so there is no need
-	// to `kick` this queue once connectivity has restored.
+	// All periodic controller HTTP requests are added to this queue,
+	// sending errors of which can be ignored. This means even the
+	// request has failed, it will be removed from the queue, so there
+	// is no need to `kick` this queue once connectivity has restored.
 	DeferredPeriodicCtx *DeferredContext
+	// All periodic LOC HTTP requests are added to this queue,
+	// sending errors of which can be ignored. This means even the
+	// request has failed, it will be removed from the queue, so there
+	// is no need to `kick` this queue once connectivity has restored.
+	DeferredLOCPeriodicCtx *DeferredContext
 }
 
 // ContextOptions - options to be passed at NewContext