lz4 decompress in file worker

Author: DmitryRomanov

go.mod

@@ -31,6 +31,7 @@ require (
 	github.com/klauspost/compress v1.17.8
 	github.com/minio/minio-go v6.0.14+incompatible
 	github.com/ozontech/insane-json v0.1.9
+	github.com/pierrec/lz4/v4 v4.1.21
 	github.com/prometheus/client_golang v1.16.0
 	github.com/prometheus/procfs v0.10.1
 	github.com/rjeczalik/notify v0.9.3
@@ -118,7 +119,6 @@ require (
 	github.com/onsi/ginkgo v1.16.5 // indirect
 	github.com/opencontainers/runtime-spec v1.0.2 // indirect
 	github.com/pascaldekloe/name v1.0.1 // indirect
-	github.com/pierrec/lz4/v4 v4.1.21 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.3.0 // indirect
 	github.com/prometheus/common v0.42.0 // indirect

plugin/input/file/worker.go

@@ -3,12 +3,15 @@ package file
 import (
 	"bytes"
 	"io"
+	"mime"
 	"os"
+	"path/filepath"
 	"strings"
 
 	"github.com/ozontech/file.d/pipeline"
 	"github.com/ozontech/file.d/pipeline/metadata"
 	k8s_meta "github.com/ozontech/file.d/plugin/input/k8s/meta"
+	"github.com/pierrec/lz4/v4"
 
 	"go.uber.org/zap"
 )
@@ -86,12 +89,22 @@ func (w *worker) work(controller inputer, jobProvider *jobProvider, readBufferSi
 			}
 		}
 
+		mimeType := getMimeType(file.Name())
+		var reader io.Reader
+
+		if mimeType == "application/x-lz4" {
+			lz4Reader := lz4.NewReader(file)
+			reader = lz4Reader
+		} else {
+			reader = file
+		}
+
 		// append the data of the old work, this happens when the event was not completely written to the file
 		// for example: {"level": "info", "message": "some...
 		// the end of the message can be added later and will be read in this iteration
 		accumBuf = append(accumBuf[:0], job.tail...)
 		for {
-			n, err := file.Read(readBuf)
+			n, err := reader.Read(readBuf)
 			controller.IncReadOps()
 			// if we read to end of file it's time to check truncation etc and process next job
 			if err == io.EOF || n == 0 {
@@ -173,6 +186,16 @@ func (w *worker) work(controller inputer, jobProvider *jobProvider, readBufferSi
 	}
 }
 
+func getMimeType(filename string) string {
+	ext := filepath.Ext(filename)
+	mimeType := mime.TypeByExtension(ext)
+	if mimeType == "" {
+		mimeType = "application/octet-stream"
+	}
+
+	return mimeType
+}
+
 func (w *worker) processEOF(file *os.File, job *Job, jobProvider *jobProvider, totalOffset int64) error {
 	stat, err := file.Stat()
 	if err != nil {