fix: Resolve pool crash on start due to new event queue structure, improve admin API output, and fix streams not counting completedLines

Author: driskell

lc-lib/admin/api/data.go

@@ -70,7 +70,7 @@ func (n Bytes) HumanReadable(string) ([]byte, error) {
 		size = float64(n) / 1024 / 1024 / 1024 / 1024
 	}
 
-	return []byte(strconv.FormatFloat(size, 'g', 2, 64) + suffix), nil
+	return []byte(strconv.FormatFloat(size, 'f', 2, 64) + suffix), nil
 }
 
 // Float represents a floating point number in the API

lc-lib/publisher/endpoint/endpoint.go

@@ -79,7 +79,7 @@ func (e *Endpoint) Init() {
 	e.ctx = context.WithValue(context.Background(), ContextSelf, e)
 
 	e.warming = true
-	backoffName := fmt.Sprintf("[E %s] Failure", e.poolEntry.Desc)
+	backoffName := fmt.Sprintf("[E %s] Recovery", e.poolEntry.Desc)
 	e.backoff = core.NewExpBackoff(backoffName, e.sink.config.Backoff, e.sink.config.BackoffMax)
 
 	e.readyElement.Value = e

lc-lib/receiver/pool.go

@@ -118,8 +118,11 @@ func (r *Pool) Run() {
 ReceiverLoop:
 	for {
 		var nextSpool *spoolEntry = nil
+		var nextSpoolEvents []*event.Event = nil
+
 		if len(r.spool) != 0 {
 			nextSpool = r.spool[0]
+			nextSpoolEvents = nextSpool.events
 		}
 
 		select {
@@ -214,6 +217,8 @@ ReceiverLoop:
 				} else {
 					// Reset idle timeout
 					r.startIdleTimeout(eventImpl.Context(), receiver, connection)
+					// Count lines here as there is no ACK to do it
+					connectionStatus.lines += int64(eventImpl.Count())
 				}
 				connectionStatus.bytes += eventImpl.Size()
 				r.connectionLock.Unlock()
@@ -291,7 +296,7 @@ ReceiverLoop:
 					r.startIdleTimeout(eventImpl.Context(), receiver, connection)
 				}
 			}
-		case spoolChan <- nextSpool.events:
+		case spoolChan <- nextSpoolEvents:
 			copy(r.spool, r.spool[1:])
 			r.spool = r.spool[:len(r.spool)-1]
 			r.spoolSize -= int64(nextSpool.size)

lc-lib/receiver/status.go

@@ -70,7 +70,7 @@ func (p *poolConnectionStatus) Update() error {
 	p.SetEntry("listener", api.String(p.listener))
 	p.SetEntry("description", api.String(p.desc))
 	p.SetEntry("completedLines", api.Number(p.lines))
-	p.SetEntry("completedBytes", api.Bytes(p.bytes))
+	p.SetEntry("receivedBytes", api.Bytes(p.bytes))
 	p.SetEntry("pendingPayloads", api.Number(len(p.progress)))
 	return nil
 }