bulker: improved logging

all projects: upgraded to go 1.22
jitsucom · Mar 21, 2024 · 876c7e2 · 876c7e2
1 parent f65ddf6
commit 876c7e2
Show file tree

Hide file tree

Showing 21 changed files with 347 additions and 17 deletions.
diff --git a/.github/workflows/test-build-image.yml b/.github/workflows/test-build-image.yml
@@ -29,7 +29,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v3
       with:
-        go-version: 1.21.5
+        go-version: 1.22.1
         cache: false
 
     - name: Bulker Test

diff --git a/admin/go.mod b/admin/go.mod
@@ -1,6 +1,6 @@
 module github.com/jitsucom/bulker/admin
 
-go 1.21
+go 1.22
 
 require (
 	github.com/confluentinc/confluent-kafka-go/v2 v2.3.0

diff --git a/bulkerapp/app/abstract_batch_consumer.go b/bulkerapp/app/abstract_batch_consumer.go
@@ -264,6 +264,7 @@ func (bc *AbstractBatchConsumer) ConsumeAll() (counters BatchCounters, err error
 		bc.Debugf("Consumer should not consume. offsets: %d-%d", lowOffset, highOffset)
 		return BatchCounters{}, nil
 	}
+	bc.Infof("Starting consuming messages from topic. Messages in topic: ~%d. ", highOffset-lowOffset)
 	batchNumber := 1
 	for {
 		if bc.retired.Load() {

diff --git a/bulkerapp/app/batch_consumer.go b/bulkerapp/app/batch_consumer.go
@@ -38,6 +38,7 @@ func NewBatchConsumer(repository *Repository, destinationId string, batchPeriodS
 }
 
 func (bc *BatchConsumerImpl) processBatchImpl(destination *Destination, batchNum, batchSize, retryBatchSize int, highOffset int64) (counters BatchCounters, nextBatch bool, err error) {
+	bc.Infof("Processing batch #%d", batchNum)
 	counters.firstOffset = int64(kafka.OffsetBeginning)
 	startTime := time.Now()
 	var bulkerStream bulker.BulkerStream

diff --git a/bulkerapp/go.mod b/bulkerapp/go.mod
@@ -1,6 +1,6 @@
 module github.com/jitsucom/bulker/bulkerapp
 
-go 1.21
+go 1.22
 
 require (
 	github.com/confluentinc/confluent-kafka-go/v2 v2.3.0

diff --git a/bulkerlib/go.mod b/bulkerlib/go.mod
@@ -1,6 +1,6 @@
 module github.com/jitsucom/bulker/bulkerlib
 
-go 1.21
+go 1.22
 
 require (
 	cloud.google.com/go v0.112.0

diff --git a/bulkerlib/implementations/file_storage/abstract.go b/bulkerlib/implementations/file_storage/abstract.go
@@ -45,6 +45,8 @@ type AbstractFileStorageStream struct {
 
 	state  bulker.State
 	inited bool
+
+	startTime time.Time
 }
 
 func newAbstractFileStorageStream(id string, p implementations2.FileAdapter, filenameFunc func(ctx context.Context) string, mode bulker.BulkMode, streamOptions ...bulker.StreamOption) (AbstractFileStorageStream, error) {
@@ -66,6 +68,7 @@ func newAbstractFileStorageStream(id string, p implementations2.FileAdapter, fil
 	}
 	ps.csvHeader = utils.NewSet[string]()
 	ps.state = bulker.State{Status: bulker.Active}
+	ps.startTime = time.Now()
 	return ps, nil
 }
 
@@ -130,6 +133,8 @@ func (ps *AbstractFileStorageStream) postComplete(err error) (bulker.State, erro
 		ps.state.SetError(err)
 		ps.state.Status = bulker.Failed
 	} else {
+		sec := time.Since(ps.startTime).Seconds()
+		logging.Infof("[%s] Stream completed successfully in %.2f s. Avg Speed: %.2f events/sec.", ps.id, sec, float64(ps.state.SuccessfulRows)/sec)
 		ps.state.Status = bulker.Completed
 	}
 	return ps.state, err
@@ -154,6 +159,13 @@ func (ps *AbstractFileStorageStream) flushBatchFile(ctx context.Context) (err er
 		if err != nil {
 			return errorj.Decorate(err, "failed to sync batch file")
 		}
+		stat, _ := ps.batchFile.Stat()
+		var batchSizeMb float64
+		if stat != nil {
+			batchSizeMb = float64(stat.Size()) / 1024 / 1024
+			sec := time.Since(ps.startTime).Seconds()
+			logging.Infof("[%s] Flushed %d events to batch file. Size: %.2f mb in %.2f s. Speed: %.2f mb/s", ps.id, ps.eventsInBatch, batchSizeMb, sec, batchSizeMb/sec)
+		}
 		workingFile := ps.batchFile
 		needToConvert := false
 		convertStart := time.Now()
@@ -215,7 +227,12 @@ func (ps *AbstractFileStorageStream) flushBatchFile(ctx context.Context) (err er
 			workingFile.Sync()
 		}
 		if needToConvert {
-			logging.Infof("[%s] Converted batch file from %s to %s in %s", ps.id, ps.marshaller.Format(), ps.targetMarshaller.Format(), time.Now().Sub(convertStart))
+			stat, _ = workingFile.Stat()
+			var convertedSizeMb float64
+			if stat != nil {
+				convertedSizeMb = float64(stat.Size()) / 1024 / 1024
+			}
+			logging.Infof("[%s] Converted batch file from %s (%.2f mb) to %s (%.2f mb) in %.2f s.", ps.id, ps.marshaller.FileExtension(), batchSizeMb, ps.targetMarshaller.FileExtension(), convertedSizeMb, time.Since(convertStart).Seconds())
 		}
 		//create file reader for workingFile
 		_, err = workingFile.Seek(0, 0)

diff --git a/bulkerlib/implementations/sql/abstract.go b/bulkerlib/implementations/sql/abstract.go
@@ -8,6 +8,7 @@ import (
 	"github.com/jitsucom/bulker/jitsubase/logging"
 	"github.com/jitsucom/bulker/jitsubase/utils"
 	jsoniter "github.com/json-iterator/go"
+	"time"
 )
 
 // TODO: check whether COPY is transactional ?
@@ -31,6 +32,8 @@ type AbstractSQLStream struct {
 	customTypes     types.SQLTypes
 	pkColumns       []string
 	timestampColumn string
+
+	startTime time.Time
 }
 
 func newAbstractStream(id string, p SQLAdapter, tableName string, mode bulker.BulkMode, streamOptions ...bulker.StreamOption) (*AbstractSQLStream, error) {
@@ -56,6 +59,7 @@ func newAbstractStream(id string, p SQLAdapter, tableName string, mode bulker.Bu
 	//TODO: max column?
 	ps.state = bulker.State{Status: bulker.Active}
 	ps.customTypes = customFields
+	ps.startTime = time.Now()
 	return &ps, nil
 }
 

diff --git a/bulkerlib/implementations/sql/abstract_transactional.go b/bulkerlib/implementations/sql/abstract_transactional.go
@@ -107,6 +107,8 @@ func (ps *AbstractTransactionalSQLStream) postComplete(ctx context.Context, err
 			_ = ps.tx.Rollback()
 		}
 	} else {
+		sec := time.Since(ps.startTime).Seconds()
+		logging.Infof("[%s] Stream completed successfully in %.2f s. Avg Speed: %.2f events/sec.", ps.id, sec, float64(ps.state.SuccessfulRows)/sec)
 		if ps.tx != nil {
 			if ps.tmpTable != nil {
 				_ = ps.tx.Drop(ctx, ps.tmpTable, true)
@@ -141,6 +143,13 @@ func (ps *AbstractTransactionalSQLStream) flushBatchFile(ctx context.Context) (s
 		if err != nil {
 			return nil, errorj.Decorate(err, "failed to sync batch file")
 		}
+		stat, _ := ps.batchFile.Stat()
+		var batchSizeMb float64
+		if stat != nil {
+			batchSizeMb = float64(stat.Size()) / 1024 / 1024
+			sec := time.Since(ps.startTime).Seconds()
+			logging.Infof("[%s] Flushed %d events to batch file. Size: %.2f mb in %.2f s. Speed: %.2f mb/s", ps.id, ps.eventsInBatch, batchSizeMb, sec, batchSizeMb/sec)
+		}
 		workingFile := ps.batchFile
 		needToConvert := false
 		convertStart := time.Now()
@@ -202,7 +211,12 @@ func (ps *AbstractTransactionalSQLStream) flushBatchFile(ctx context.Context) (s
 			workingFile.Sync()
 		}
 		if needToConvert {
-			logging.Infof("[%s] Converted batch file from %s(%s) to %s(%s) in %s", ps.id, ps.marshaller.Format(), ps.marshaller.Compression(), ps.targetMarshaller.Format(), ps.targetMarshaller.Compression(), time.Now().Sub(convertStart))
+			stat, _ = workingFile.Stat()
+			var convertedSizeMb float64
+			if stat != nil {
+				convertedSizeMb = float64(stat.Size()) / 1024 / 1024
+			}
+			logging.Infof("[%s] Converted batch file from %s (%.2f mb) to %s (%.2f mb) in %.2f s.", ps.id, ps.marshaller.FileExtension(), batchSizeMb, ps.targetMarshaller.FileExtension(), convertedSizeMb, time.Since(convertStart).Seconds())
 		}
 		if ps.s3 != nil {
 			s3Config := s3BatchFileOption.Get(&ps.options)

diff --git a/bulkerlib/implementations/sql/bigquery.go b/bulkerlib/implementations/sql/bigquery.go
@@ -1082,6 +1082,7 @@ func (bq *BigQuery) RunJob(ctx context.Context, runner JobRunner, jobDescription
 	defer func() {
 		bq.logQuery(jobDescription, runner, err)
 	}()
+	startTime := time.Now()
 	state = &bulker.WarehouseState{}
 	var status *bigquery.JobStatus
 	var jobID string
@@ -1109,7 +1110,7 @@ func (bq *BigQuery) RunJob(ctx context.Context, runner JobRunner, jobDescription
 		}
 		return job, state, fmt.Errorf("Failed to %s.%s Completed with error: %v%s", jobDescription, jobID, err, bytesProcessed)
 	} else {
-		bq.Infof("Successfully %s.%s%s", jobDescription, jobID, bytesProcessed)
+		bq.Infof("Successfully %s.%s%s in %.2f s.", jobDescription, jobID, bytesProcessed, time.Since(startTime).Seconds())
 		return job, state, nil
 	}
 }
diff --git a/connectors/airbytecdk/go.mod b/connectors/airbytecdk/go.mod
@@ -1,3 +1,3 @@
 module github.com/jitsucom/bulker/airbytecdk
 
-go 1.21
+go 1.22
diff --git a/connectors/firebase/go.mod b/connectors/firebase/go.mod
@@ -1,6 +1,6 @@
 module github.com/jitsucom/bulker/connectors/firebase
 
-go 1.21
+go 1.22
 
 require (
 	cloud.google.com/go/firestore v1.14.0

diff --git a/eventslog/go.mod b/eventslog/go.mod
@@ -1,6 +1,6 @@
 module github.com/jitsucom/bulker/eventslog
 
-go 1.21
+go 1.22
 
 require (
 	github.com/ClickHouse/clickhouse-go/v2 v2.19.0

diff --git a/go.work b/go.work
@@ -1,4 +1,4 @@
-go 1.21
+go 1.22
 
 use (
 	./bulkerapp
-Original file line number
+Diff line change
@@ -1,4 +1,4 @@
-    go 1.21
+    go 1.22
     use (
     	./bulkerapp
@@ Expand Down @@