Promtail: Add compressed files support

CHANGELOG.md

@@ -21,8 +21,8 @@
 * [6375](https://github.com/grafana/loki/pull/6375) **dannykopping**: Fix bug that prevented users from using the `json` parser after a `line_format` pipeline stage.
 * [6505](https://github.com/grafana/loki/pull/6375) **dmitri-lerko** Fixes `failed to receive pubsub messages` error with promtail GCPLog client.
 ##### Changes
+* [6726](https://github.com/grafana/loki/pull/6726) **kavirajk**: upgrades go from 1.17.9 -> 1.18.4
 * [6415](https://github.com/grafana/loki/pull/6415) **salvacorts**: Evenly spread queriers across kubernetes nodes.
-* [6726](https://github.com/grafana/loki/pull/6726) **kavirajk** upgrades go from 1.17.9 -> 1.18.4
 * [6349](https://github.com/grafana/loki/pull/6349) **simonswine**: Update the default HTTP listen port from 80 to 3100. Make sure to configure the port explicitly if you are using port 80.
 * [6835](https://github.com/grafana/loki/pull/6835) **DylanGuedes**: Add new per-tenant query timeout configuration and remove engine query timeout.
 
@@ -31,6 +31,7 @@
 ##### Enhancements
 
 * [6708](https://github.com/grafana/loki/pull/6708) **DylanGuedes**: Add compressed files support to Promtail.
+* [5977](https://github.com/grafana/loki/pull/5977) **juissi-t** lambda-promtail: Add support for Kinesis data stream events
 * [6828](https://github.com/grafana/loki/pull/6828) **alexandre1984rj** Add the BotScore and BotScoreSrc fields once the Cloudflare API returns those two fields on the list of all available log fields.
 * [6656](https://github.com/grafana/loki/pull/6656) **carlospeon**: Allow promtail to add matches to the journal reader
 

clients/pkg/promtail/targets/file/decompresser.go

@@ -3,7 +3,6 @@ package file
 import (
 	"bufio"
 	"compress/bzip2"
-	"compress/flate"
 	"compress/gzip"
 	"compress/zlib"
 	"fmt"
@@ -32,7 +31,6 @@ import (
 
 func supportedCompressedFormats() map[string]struct{} {
 	return map[string]struct{}{
-		".zip":    {},
 		".gz":     {},
 		".tar.gz": {},
 		".z":      {},
@@ -114,9 +112,6 @@ func mountReader(f *os.File, logger log.Logger) (reader io.Reader, err error) {
 	} else if ext == ".z" {
 		decompressLib = "compress/zlib"
 		reader, err = zlib.NewReader(f)
-	} else if ext == ".zip" {
-		decompressLib = "compress/flate"
-		reader = flate.NewReader(f)
 	} else if ext == ".bz2" {
 		decompressLib = "bzip2"
 		reader = bzip2.NewReader(f)
@@ -192,8 +187,12 @@ func (t *decompressor) readLines() {
 
 	level.Info(t.logger).Log("msg", "successfully mounted reader", "path", t.path, "ext", filepath.Ext(t.path))
 
+	megabyte := 1000000
+	maxLoglineSize := 2 * megabyte
+	buffer := make([]byte, 3*megabyte)
 	scanner := bufio.NewScanner(r)
-	for i := 0; ; i++ {
+	scanner.Buffer(buffer, maxLoglineSize)
+	for line := 1; ; line++ {
 		if !scanner.Scan() {
 			break
 		}
@@ -206,7 +205,7 @@ func (t *decompressor) readLines() {
 			break
 		}
 
-		if i <= int(t.position) {
+		if line <= int(t.position) {
 			// skip already seen lines.
 			continue
 		}

clients/pkg/promtail/targets/file/decompresser_test.go

@@ -0,0 +1,175 @@
+package file
+
+import (
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/go-kit/log"
+	"github.com/grafana/loki/clients/pkg/promtail/api"
+	"github.com/grafana/loki/clients/pkg/promtail/client/fake"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/atomic"
+)
+
+type noopClient struct {
+	noopChan chan api.Entry
+	wg       sync.WaitGroup
+	once     sync.Once
+}
+
+func (n *noopClient) Chan() chan<- api.Entry {
+	return n.noopChan
+}
+
+func (n *noopClient) Stop() {
+	n.once.Do(func() { close(n.noopChan) })
+}
+
+func newNoopClient() *noopClient {
+	c := &noopClient{noopChan: make(chan api.Entry)}
+	c.wg.Add(1)
+	go func() {
+		defer c.wg.Done()
+		for range c.noopChan {
+			// noop
+		}
+	}()
+	return c
+}
+
+func BenchmarkReadlines(b *testing.B) {
+	entryHandler := newNoopClient()
+
+	scenarios := []struct {
+		name string
+		file string
+	}{
+		{
+			name: "2000 lines of log .tar.gz compressed",
+			file: "test_fixtures/short-access.tar.gz",
+		},
+		{
+			name: "100000 lines of log .gz compressed",
+			file: "test_fixtures/long-access.gz",
+		},
+	}
+
+	for _, tc := range scenarios {
+		b.Run(tc.name, func(b *testing.B) {
+			decBase := &decompressor{
+				logger:  log.NewNopLogger(),
+				running: atomic.NewBool(false),
+				handler: entryHandler,
+				path:    tc.file,
+			}
+
+			for i := 0; i < b.N; i++ {
+				newDec := decBase
+				newDec.metrics = NewMetrics(prometheus.NewRegistry())
+				newDec.done = make(chan struct{})
+				newDec.readLines()
+				<-newDec.done
+			}
+		})
+	}
+}
+
+func TestGigantiqueGunzipFile(t *testing.T) {
+	file := "test_fixtures/long-access.gz"
+	handler := fake.New(func() {})
+
+	d := &decompressor{
+		logger:  log.NewNopLogger(),
+		running: atomic.NewBool(false),
+		handler: handler,
+		path:    file,
+		done:    make(chan struct{}),
+		metrics: NewMetrics(prometheus.NewRegistry()),
+	}
+
+	d.readLines()
+
+	<-d.done
+	time.Sleep(time.Millisecond * 200)
+
+	entries := handler.Received()
+	require.Equal(t, 100000, len(entries))
+}
+
+func TestOnelineFiles(t *testing.T) {
+	fileContent, err := os.ReadFile("test_fixtures/onelinelog.log")
+	require.NoError(t, err)
+	t.Run("gunzip file", func(t *testing.T) {
+		file := "test_fixtures/onelinelog.log.gz"
+		handler := fake.New(func() {})
+
+		d := &decompressor{
+			logger:  log.NewNopLogger(),
+			running: atomic.NewBool(false),
+			handler: handler,
+			path:    file,
+			done:    make(chan struct{}),
+			metrics: NewMetrics(prometheus.NewRegistry()),
+		}
+
+		d.readLines()
+
+		<-d.done
+		time.Sleep(time.Millisecond * 200)
+
+		entries := handler.Received()
+		require.Equal(t, 1, len(entries))
+		require.Equal(t, string(fileContent), entries[0].Line)
+	})
+
+	t.Run("bzip2 file", func(t *testing.T) {
+		file := "test_fixtures/onelinelog.log.bz2"
+		handler := fake.New(func() {})
+
+		d := &decompressor{
+			logger:  log.NewNopLogger(),
+			running: atomic.NewBool(false),
+			handler: handler,
+			path:    file,
+			done:    make(chan struct{}),
+			metrics: NewMetrics(prometheus.NewRegistry()),
+		}
+
+		d.readLines()
+
+		<-d.done
+		time.Sleep(time.Millisecond * 200)
+
+		entries := handler.Received()
+		require.Equal(t, 1, len(entries))
+		require.Equal(t, string(fileContent), entries[0].Line)
+	})
+
+	t.Run("tar.gz file", func(t *testing.T) {
+		file := "test_fixtures/onelinelog.tar.gz"
+		handler := fake.New(func() {})
+
+		d := &decompressor{
+			logger:  log.NewNopLogger(),
+			running: atomic.NewBool(false),
+			handler: handler,
+			path:    file,
+			done:    make(chan struct{}),
+			metrics: NewMetrics(prometheus.NewRegistry()),
+		}
+
+		d.readLines()
+
+		<-d.done
+		time.Sleep(time.Millisecond * 200)
+
+		entries := handler.Received()
+		require.Equal(t, 1, len(entries))
+		firstEntry := entries[0]
+		require.Contains(t, firstEntry.Line, "onelinelog.log") // contains .tar.gz headers
+		require.Contains(t, firstEntry.Line, `5.202.214.160 - - [26/Jan/2019:19:45:25 +0330] "GET / HTTP/1.1" 200 30975 "https://www.zanbil.ir/" "Mozilla/5.0 (Windows NT 6.2; WOW64; rv:21.0) Gecko/20100101 Firefox/21.0" "-"`)
+	})
+}

clients/pkg/promtail/targets/file/test_fixtures/onelinelog.log

@@ -0,0 +1 @@
+5.202.214.160 - - [26/Jan/2019:19:45:25 +0330] "GET / HTTP/1.1" 200 30975 "https://www.zanbil.ir/" "Mozilla/5.0 (Windows NT 6.2; WOW64; rv:21.0) Gecko/20100101 Firefox/21.0" "-"