Add deduplicated log line printing to quesma_logger, use it in places that logged most often (#1266)

This PR introduces `logger.DeduplicatedWarn()` and
`logger.DeduplicatedInfo()` methods to `quesma_logger`, which
automatically deduplicate log messages.

For example,

```
  Ingesting via _bulk API, batch size=2 lines
```

is a log line that the code tries to print very often, but with
`DeduplicatedInfo()` that exact log line is only printed once. An
expirable LRU cache is used with 1 minute expiry time, so the log line
is still printed occasionally, but at most 1 every minute.

Additionally, a couple places that produced a lot of logs were migrated
to use `DeduplicatedWarn()`/`DeduplicatedInfo()`.

---------

Signed-off-by: Piotr Grabowski <[email protected]>

Author: avelanarius

quesma/clickhouse/schema.go

@@ -280,13 +280,13 @@ func NewType(value any, valueOrigin string) (Type, error) {
 			cols = append(cols, &Column{Name: k, Type: innerType, Codec: Codec{Name: ""}})
 		}
 		if len(cols) == 0 {
-			logger.Warn().Msgf("Empty map type (origin: %s).", valueOrigin)
+			logger.DeduplicatedWarn().Msgf("Empty map type (origin: %s).", valueOrigin)
 			return nil, fmt.Errorf("empty map type (origin: %s)", valueOrigin)
 		}
 		return MultiValueType{Name: "Tuple", Cols: cols}, nil
 	case []interface{}:
 		if len(valueCasted) == 0 {
-			logger.Warn().Msgf("Empty array type (origin: %s).", valueOrigin)
+			logger.DeduplicatedWarn().Msgf("Empty array type (origin: %s).", valueOrigin)
 			return nil, fmt.Errorf("empty array type (origin: %s)", valueOrigin)
 		}
 		innerName := fmt.Sprintf("%s[0]", valueOrigin)
@@ -296,11 +296,11 @@ func NewType(value any, valueOrigin string) (Type, error) {
 		}
 		return CompoundType{Name: "Array", BaseType: innerType}, nil
 	case nil:
-		logger.Warn().Msgf("Nil type (origin: %s).", valueOrigin)
+		logger.DeduplicatedWarn().Msgf("Nil type (origin: %s).", valueOrigin)
 		return nil, fmt.Errorf("nil type (origin: %s)", valueOrigin)
 	}
 
-	logger.Warn().Msgf("Unsupported type '%T' of value: %v (origin: %s).", value, value, valueOrigin)
+	logger.DeduplicatedWarn().Msgf("Unsupported type '%T' of value: %v (origin: %s).", value, value, valueOrigin)
 	return nil, fmt.Errorf("unsupported type '%T' of value: %v (origin: %s)", value, value, valueOrigin)
 }
 

quesma/go.mod

@@ -18,6 +18,7 @@ require (
 	github.com/gorilla/securecookie v1.1.2
 	github.com/gorilla/sessions v1.4.0
 	github.com/hashicorp/go-multierror v1.1.1
+	github.com/hashicorp/golang-lru/v2 v2.0.7
 	github.com/jackc/pgx/v5 v5.7.2
 	github.com/k0kubun/pp v3.0.1+incompatible
 	github.com/knadh/koanf/parsers/json v0.1.0

quesma/go.sum

@@ -65,6 +65,8 @@ github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY
 github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
 github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo=
 github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM=
+github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
+github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
 github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
 github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=

quesma/logger/logger.go

@@ -268,3 +268,11 @@ func Panic() *zerolog.Event {
 func ReasonUnsupportedQuery(queryType string) string {
 	return ReasonPrefixUnsupportedQueryType + queryType
 }
+
+func DeduplicatedInfo() quesma_v2.DeduplicatedEvent {
+	return logger.DeduplicatedInfo()
+}
+
+func DeduplicatedWarn() quesma_v2.DeduplicatedEvent {
+	return logger.DeduplicatedWarn()
+}

quesma/quesma/functionality/bulk/bulk.go

@@ -22,7 +22,6 @@ import (
 	"net/http"
 	"sort"
 	"strings"
-	"sync"
 )
 
 type (
@@ -72,7 +71,7 @@ func Write(ctx context.Context, defaultIndex *string, bulk types.NDJSON, ip *ing
 	defer recovery.LogPanic()
 
 	maxBulkSize := len(bulk)
-	maybeLogBatchSize(maxBulkSize)
+	logger.DeduplicatedInfo().Msgf("Ingesting via _bulk API, batch size=%d lines", maxBulkSize)
 
 	// The returned results should be in the same order as the input request, however splitting the bulk might change the order.
 	// Therefore, each BulkRequestEntry has a corresponding pointer to the result entry, allowing us to freely split and reshuffle the bulk.
@@ -333,17 +332,3 @@ func sendToClickhouse(ctx context.Context, clickhouseBulkEntries map[string][]Bu
 		}
 	}
 }
-
-// Global set to keep track of logged batch sizes
-var loggedBatchSizes = make(map[int]struct{})
-var mutex sync.Mutex
-
-// maybeLogBatchSize logs only unique batch sizes
-func maybeLogBatchSize(batchSize int) {
-	mutex.Lock()
-	defer mutex.Unlock()
-	if _, alreadyLogged := loggedBatchSizes[batchSize]; !alreadyLogged {
-		logger.Info().Msgf("Ingesting via _bulk API, batch size=%d lines", batchSize)
-		loggedBatchSizes[batchSize] = struct{}{}
-	}
-}

quesma/v2/core/quesma_logger.go

@@ -4,8 +4,11 @@ package quesma_api
 
 import (
 	"context"
+	"fmt"
 	"github.com/QuesmaOrg/quesma/quesma/v2/core/tracing"
+	"github.com/hashicorp/golang-lru/v2/expirable"
 	"github.com/rs/zerolog"
+	"hash/fnv"
 	"os"
 	"time"
 )
@@ -20,6 +23,9 @@ const (
 	DefaultBurstSamplerPeriodSeconds    = 20  // burst up to 600 lines of logs per 20 seconds period
 	DefaultBurstSamplerMaxLogsPerSecond = 30  // ~100k lines of logs per hour
 	DefaultSheddingFrequency            = 100 // when the limit is exhausted, log every ~ 100 log lines
+
+	DeduplicatedLogsCacheSize  = 1000
+	DeduplicatedLogsExpiryTime = 1 * time.Minute
 )
 
 type QuesmaLogger interface {
@@ -30,6 +36,10 @@ type QuesmaLogger interface {
 	Fatal() *zerolog.Event
 	Panic() *zerolog.Event
 
+	// TODO: Add similar for other log levels
+	DeduplicatedInfo() DeduplicatedEvent
+	DeduplicatedWarn() DeduplicatedEvent
+
 	DebugWithCtx(ctx context.Context) *zerolog.Event
 	InfoWithCtx(ctx context.Context) *zerolog.Event
 	WarnWithCtx(ctx context.Context) *zerolog.Event
@@ -45,18 +55,19 @@ type QuesmaLogger interface {
 
 type QuesmaLoggerImpl struct {
 	zerolog.Logger
+
+	deduplicatedLogs *expirable.LRU[any, struct{}]
 }
 
 func NewQuesmaLogger(log zerolog.Logger) QuesmaLogger {
 	return &QuesmaLoggerImpl{
-		Logger: log,
+		Logger:           log,
+		deduplicatedLogs: expirable.NewLRU[any, struct{}](DeduplicatedLogsCacheSize, nil, DeduplicatedLogsExpiryTime),
 	}
 }
 
 func (l *QuesmaLoggerImpl) WithComponent(name string) QuesmaLogger {
-	return &QuesmaLoggerImpl{
-		Logger: l.Logger.With().Str("component", name).Logger(),
-	}
+	return NewQuesmaLogger(l.Logger.With().Str("component", name).Logger())
 }
 
 func (l *QuesmaLoggerImpl) MarkTraceEndWithCtx(ctx context.Context) *zerolog.Event {
@@ -124,6 +135,45 @@ func (l *QuesmaLoggerImpl) ErrorWithCtx(ctx context.Context) *zerolog.Event {
 	return event
 }
 
+func (l *QuesmaLoggerImpl) DeduplicatedInfo() DeduplicatedEvent {
+	return DeduplicatedEvent{
+		event: l.Info(),
+		l:     l,
+	}
+}
+
+func (l *QuesmaLoggerImpl) DeduplicatedWarn() DeduplicatedEvent {
+	return DeduplicatedEvent{
+		event: l.Warn(),
+		l:     l,
+	}
+}
+
+type DeduplicatedEvent struct {
+	event *zerolog.Event
+	l     *QuesmaLoggerImpl
+}
+
+func hashMsgf(format string, v ...interface{}) uint32 {
+	// []interface{} is not hashable, so we need to hash it manually
+	// For the convenience sake we just hash a Print representation
+	h := fnv.New32a()
+	fmt.Fprint(h, format, v)
+	return h.Sum32()
+}
+
+// TODO: Add wrappers for other *zerolog.Event methods
+func (m DeduplicatedEvent) Msgf(format string, v ...interface{}) {
+	hash := hashMsgf(format, v)
+
+	if m.l.deduplicatedLogs.Contains(hash) {
+		return
+	}
+
+	m.l.deduplicatedLogs.Add(hash, struct{}{})
+	m.event.Msgf(format, v...)
+}
+
 func EmptyQuesmaLogger() QuesmaLogger {
 	// not so empty :D
 	return NewQuesmaLogger(zerolog.New(

quesma/v2/core/quesma_logger_test.go

@@ -0,0 +1,44 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package quesma_api
+
+import (
+	"bytes"
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/assert"
+	"strings"
+	"testing"
+)
+
+func TestDeduplicatedEvents(t *testing.T) {
+	var buf bytes.Buffer
+
+	log := zerolog.New(&buf).Level(zerolog.DebugLevel)
+	logger := NewQuesmaLogger(log)
+
+	logger.DeduplicatedInfo().Msgf("info test %d", 42)
+	logger.DeduplicatedInfo().Msgf("info test %d", 42) // duplicate should be skipped
+
+	logger.DeduplicatedWarn().Msgf("warn test %d", 42)
+	logger.DeduplicatedWarn().Msgf("warn test %d", 42)   // duplicate should be skipped
+	logger.DeduplicatedWarn().Msgf("warn test %d", 1000) // distinct argument
+	logger.DeduplicatedWarn().Msgf("WARN test %d", 42)   // distinct format string
+
+	logger.DeduplicatedWarn().Msgf("two args %d %s %v", 42, "asda", []byte{1, 2, 3})
+	logger.DeduplicatedWarn().Msgf("two args %d %s %v", 42, "asda", []byte{1, 2, 3})  // duplicate should be skipped
+	logger.DeduplicatedWarn().Msgf("two args %d %s %v", 45, "asda", []byte{1, 2, 3})  // distinct argument
+	logger.DeduplicatedWarn().Msgf("two args %d %s %v", 42, "asda2", []byte{1, 2, 3}) // distinct argument
+	logger.DeduplicatedWarn().Msgf("two args %d %s %v", 42, "asda", []byte{5, 2, 3})  // distinct argument
+	logger.DeduplicatedWarn().Msgf("two args %d %s %v", 42, "asda2", []byte{1, 2, 3}) // duplicate should be skipped
+
+	output := buf.String()
+
+	assert.Equal(t, 1, strings.Count(output, "info test 42"))
+	assert.Equal(t, 1, strings.Count(output, "warn test 42"))
+	assert.Equal(t, 1, strings.Count(output, "warn test 1000"))
+	assert.Equal(t, 1, strings.Count(output, "WARN test 42"))
+	assert.Equal(t, 1, strings.Count(output, "two args 42 asda [1 2 3]"))
+	assert.Equal(t, 1, strings.Count(output, "two args 45 asda [1 2 3]"))
+	assert.Equal(t, 1, strings.Count(output, "two args 42 asda2 [1 2 3]"))
+	assert.Equal(t, 1, strings.Count(output, "two args 42 asda [5 2 3]"))
+}