Add locking to loki.write sink + nits (#5694)

* Address PR comments

* added bench

* add locking to loki.write sink

* adding changelog

* imports order

* fix changelog

* remove unused client write to

* added missing entry

* typos

* add missing changelog entry

* remove duplicate entry

CHANGELOG.md

@@ -95,6 +95,8 @@ Main (unreleased)
 - Grafana Agent Operator: `config-reloader` container no longer runs as root.
   (@rootmout)
 
+- Added support for replaying not sent data for `loki.write` when WAL is enabled. (@thepalbi)
+
 ### Bugfixes
 
 - Fixed an issue where `loki.process` validation for stage `metric.counter` was
@@ -129,6 +131,8 @@ Main (unreleased)
 
 - Fixed a bug where UDP syslog messages were never processed (@joshuapare)
 
+- Updating configuration for `loki.write` no longer drops data. (@thepalbi)
+
 v0.37.4 (2023-11-06)
 -----------------
 

component/common/loki/client/internal/marker_encoding.go

@@ -15,7 +15,7 @@ var (
 func EncodeMarkerV1(segment uint64) ([]byte, error) {
 	// marker format v1
 	// marker [ 0 , 1 ] - HEADER, which is used to track version
-	// marker [ 2 , 9 ] - encoded unit 64 which is the content of the marker, the last "consumed" segment
+	// marker [ 2 , 9 ] - encoded uint64 which is the content of the marker, the last "consumed" segment
 	// marker [ 10, 13 ] - CRC32 of the first 10 bytes of the marker, using IEEE polynomial
 	bs := make([]byte, 14)
 	// write header with marker format version

component/common/loki/wal/watcher_test.go

@@ -64,12 +64,9 @@ func (t *testWriteTo) AssertContainsLines(tst *testing.T, lines ...string) {
 	}
 	t.ReadEntries.DoneIterate()
 
-	allSeen := true
-	for _, wasSeen := range seen {
-		allSeen = allSeen && wasSeen
+	for line, wasSeen := range seen {
+		require.True(tst, wasSeen, "expected to have received line: %s", line)
 	}
-
-	require.True(tst, allSeen, "expected all entries to have been received")
 }
 
 // watcherTestResources contains all resources necessary to test an individual Watcher functionality

component/loki/write/write.go

@@ -119,11 +119,14 @@ func (c *Component) Run(ctx context.Context) error {
 		case <-ctx.Done():
 			return nil
 		case entry := <-c.receiver.Chan():
+			c.mut.RLock()
 			select {
 			case <-ctx.Done():
+				c.mut.RUnlock()
 				return nil
 			case c.sink.Chan() <- entry:
 			}
+			c.mut.RUnlock()
 		}
 	}
 }