adding max interval

Author: samber

README.md

@@ -58,7 +58,7 @@ type Option struct {
 	Level slog.Leveler
 
 	// parquet rows buffer
-	Buffer parquet.ParquetBuffer
+	Buffer slogparquet.ParquetBuffer
 
 	// optional: customize json payload builder
 	Converter Converter
@@ -68,7 +68,7 @@ type Option struct {
 ### Parquet buffer
 
 ```go
-func NewParquetBuffer(bucket objstore.Bucket, prefix string, rows int) parquet.ParquetBuffer
+func NewParquetBuffer(bucket objstore.Bucket, prefix string, maxRecords int, maxInterval time.Duration) slogparquet.ParquetBuffer
 ```
 
 Attributes will be injected in log payload.
@@ -88,20 +88,20 @@ import (
 )
 
 func main() {
-	bucket, _ := s3.NewBucketWithConfig(
-		slogparquet.NewLogger(),
-		s3.Config{
-			Endpoint:  os.Getenv("AWS_S3_ENDPOINT"),
-			Region:    os.Getenv("AWS_S3_REGION"),
-			Bucket:    os.Getenv("AWS_S3_BUCKET"),
-			AccessKey: os.Getenv("AWS_ACCESS_KEY"),
-			SecretKey: os.Getenv("AWS_SECRET_KEY"),
-			PartSize:  16 * 1024 * 1024, // 16MB
-		},
-		"logger",
-	)
-
-	buffer := slogparquet.NewParquetBuffer(bucket, "api/logs", 10*1024*1024)
+    bucket, _ := s3.NewBucketWithConfig(
+        slogparquet.NewLogger(),
+        s3.Config{
+            Endpoint:  os.Getenv("AWS_S3_ENDPOINT"),
+            Region:    os.Getenv("AWS_S3_REGION"),
+            Bucket:    os.Getenv("AWS_S3_BUCKET"),
+            AccessKey: os.Getenv("AWS_ACCESS_KEY"),
+            SecretKey: os.Getenv("AWS_SECRET_KEY"),
+            PartSize:  16 * 1024 * 1024, // 16MB
+        },
+        "logger",
+    )
+
+    buffer := slogparquet.NewParquetBuffer(bucket, "api/logs", 10*1024*1024)
 
     logger := slog.New(slogparquet.Option{Level: slog.LevelDebug, Buffer: buffer}.NewParquetHandler())
     logger = logger.
@@ -126,8 +126,8 @@ func main() {
         ).
         Info("user registration")
 
-	buffer.Flush(true)
-	bucket.Close()
+    buffer.Flush(true)
+    bucket.Close()
 }
 ```
 

docker-compose.yml

@@ -0,0 +1,12 @@
+version: '3'
+
+services:
+  minio:
+    image: minio/minio
+    ports:
+      - "9000:9000" # s3-compatible api
+      - "9001:9001" # web ui
+    environment:
+      MINIO_ROOT_USER: helloworld
+      MINIO_ROOT_PASSWORD: helloworld
+    command: server --console-address ":9001" /data

example/example.go

@@ -13,6 +13,13 @@ import (
 )
 
 func main() {
+	// export AWS_ACCESS_KEY=helloworld
+	// export AWS_SECRET_KEY=helloworld
+	// export AWS_S3_REGION=fr-par
+	// export AWS_S3_ENDPOINT=localhost:9000
+	// export AWS_S3_BUCKET=slog-test
+	// go run *.go
+
 	bucket, err := s3.NewBucketWithConfig(
 		slogparquet.NewLogger(),
 		s3.Config{
@@ -29,7 +36,7 @@ func main() {
 		log.Fatal(err)
 	}
 
-	buffer := slogparquet.NewParquetBuffer(bucket, "logs", 10*1024*1024)
+	buffer := slogparquet.NewParquetBuffer(bucket, "logs", 10*1024*1024, 1*time.Second)
 
 	logger := slog.New(slogparquet.Option{Level: slog.LevelDebug, Buffer: buffer}.NewParquetHandler())
 	logger = logger.
@@ -54,26 +61,20 @@ func main() {
 		).
 		Info("user registration")
 
+	for i := 0; i < 10_000_000; i++ {
+		logger.
+			With(
+				slog.Group("user",
+					slog.String("id", "user-123"),
+					slog.Time("created_at", time.Now().AddDate(0, 0, -1)),
+				),
+			).
+			With("a", i).
+			With("environment", "dev").
+			With("error", fmt.Errorf("an error")).
+			Error("A message")
+	}
+
 	buffer.Flush(true)
 	bucket.Close()
-
-	// logger := slog.New(slogparquet.Option{Level: slog.LevelDebug, Buffer: buffer}.NewParquetHandler())
-	// logger = logger.With("release", "v1.0.0")
-
-	// for i := 0; i < 10_000_000; i++ {
-	// 	logger.
-	// 		With(
-	// 			slog.Group("user",
-	// 				slog.String("id", "user-123"),
-	// 				slog.Time("created_at", time.Now().AddDate(0, 0, -1)),
-	// 			),
-	// 		).
-	// 		With("a", i).
-	// 		With("environment", "dev").
-	// 		With("error", fmt.Errorf("an error")).
-	// 		Error("A message")
-	// }
-
-	// buffer.Flush(true)
-	// bucket.Close()
 }

example/go.mod

@@ -7,6 +7,8 @@ require (
 	github.com/thanos-io/objstore v0.0.0-20230816175749-20395bffdf26
 )
 
+replace github.com/samber/slog-parquet => ../
+
 require (
 	github.com/andybalholm/brotli v1.0.5 // indirect
 	github.com/aws/aws-sdk-go-v2 v1.16.0 // indirect

example/go.sum

@@ -311,8 +311,6 @@ github.com/rs/xid v1.5.0 h1:mKX4bl4iPYJtEIxp6CYiUuLQ/8DYMoz0PUdtGgMFRVc=
 github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
 github.com/samber/lo v1.38.1 h1:j2XEAqXKb09Am4ebOg31SpvzUTTs6EN3VfgeLUhPdXM=
 github.com/samber/lo v1.38.1/go.mod h1:+m/ZKRl6ClXCE2Lgf3MsQlWfh4bn1bz6CXEOxnEXnEA=
-github.com/samber/slog-parquet v0.0.0-20230819003546-f6e0e0bb6a8a h1:Bi9kTzF4F75Lry3MDi7hRk9FaNIYwTgXbWz9Ipth97s=
-github.com/samber/slog-parquet v0.0.0-20230819003546-f6e0e0bb6a8a/go.mod h1:mz88Z+AwBfpgRltNWNbD3Mu4yxNNgHFsraPVzmCCNgk=
 github.com/segmentio/asm v1.1.3/go.mod h1:Ld3L4ZXGNcSLRg4JBsZ3//1+f/TjYl0Mzen/DQy1EJg=
 github.com/segmentio/encoding v0.3.6 h1:E6lVLyDPseWEulBmCmAKPanDd3jiyGDo5gMcugCRwZQ=
 github.com/segmentio/encoding v0.3.6/go.mod h1:n0JeuIqEQrQoPDGsjo8UNd1iA0U8d8+oHAA4E3G3OxM=

parquet.go

@@ -26,7 +26,8 @@ type parquetBuffer struct {
 	prefix string
 	id     string
 
-	size         int
+	maxRecords   int
+	maxInterval  time.Duration
 	rowGroupSize int
 
 	mutex  sync.Mutex // ☢️
@@ -36,13 +37,14 @@ type parquetBuffer struct {
 	buffer *parquet.GenericBuffer[log]
 }
 
-func NewParquetBuffer(bucket objstore.Bucket, prefix string, size int) ParquetBuffer {
+func NewParquetBuffer(bucket objstore.Bucket, prefix string, maxRecords int, maxInterval time.Duration) ParquetBuffer {
 	return &parquetBuffer{
 		prefix: prefix,
 		id:     lo.Must(uuid.NewV4()).String()[0:5],
 
-		size:         size,
-		rowGroupSize: int(math.Ceil(float64(size) / 10)),
+		maxRecords:   maxRecords,
+		maxInterval:  maxInterval,
+		rowGroupSize: int(math.Ceil(float64(maxRecords) / 10)),
 
 		mutex:  sync.Mutex{},
 		bucket: bucket,
@@ -52,7 +54,7 @@ func NewParquetBuffer(bucket objstore.Bucket, prefix string, size int) ParquetBu
 	}
 }
 
-func (b *parquetBuffer) Append(time time.Time, logLevel slog.Level, message string, attributes map[string]any) error {
+func (b *parquetBuffer) Append(tIme time.Time, logLevel slog.Level, message string, attributes map[string]any) error {
 	serializedAttrs, err := json.Marshal(attributes)
 	if err != nil {
 		return err
@@ -62,7 +64,7 @@ func (b *parquetBuffer) Append(time time.Time, logLevel slog.Level, message stri
 
 	_, err = b.buffer.Write([]log{
 		{
-			Time:       time,
+			Time:       tIme,
 			LogLevel:   logLevel.String(),
 			Message:    message,
 			Attributes: serializedAttrs,
@@ -73,7 +75,7 @@ func (b *parquetBuffer) Append(time time.Time, logLevel slog.Level, message stri
 		return err
 	}
 
-	if b.buffer.Len() >= b.size {
+	if b.buffer.Len() >= b.maxRecords || b.start.Add(b.maxInterval).Before(time.Now()) {
 		b.mutex.Unlock()
 		return b.Flush(false)
 	}