Add concurrencyThread setting for SQS ingestion. (#100)

internal/config/config.go

@@ -98,6 +98,7 @@ type S3SQS struct {
 	WaitTimeout       int64  `json:"waitTimeout,omitempty" yaml:"waitTimeout" env:"WAITTIMEOUT"`                   // in seconds
 	VisibilityTimeout int64  `json:"visibilityTimeout,omitempty" yaml:"visibilityTimeout" env:"VISIBILITYTIMEOUT"` // in seconds
 	Retries           int    `json:"retries" yaml:"retries" env:"RETRIES"`
+	ConcurrencyThread int    `json:"concurrencyThread" yaml:"concurrencyThread" env:"concurrencyThread"`
 }
 
 // Presto represents the Presto configuration

internal/ingress/s3sqs/s3sqs.go

@@ -8,6 +8,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"log"
 	"net/url"
 	"runtime"
 	"time"
@@ -55,7 +56,10 @@ func New(conf *config.S3SQS, region string, monitor monitor.Monitor) (*Ingress,
 	if err != nil {
 		return nil, err
 	}
-
+	if conf.ConcurrencyThread != 0 {
+		concurrency = int64(runtime.NumCPU() * conf.ConcurrencyThread)
+		log.Printf("new sqs: use config ConcurrencyThread(%d) to set the threads per CPU, total threads are %d", conf.ConcurrencyThread, concurrency)
+	}
 	return NewWith(reader, loader, monitor), nil
 }
 

internal/storage/disk/disk.go

@@ -86,7 +86,7 @@ func (s *Storage) Open(dir string, options config.Badger) error {
 	case config.BadgerIngestion:
 		opts = opts.
 			WithBaseLevelSize(204800000).
-			WithMaxLevels(3).
+			WithMaxLevels(7).
 			WithSyncWrites(false)
 	}
 
@@ -96,15 +96,15 @@ func (s *Storage) Open(dir string, options config.Badger) error {
 	}
 
 	// max size of lsm tree in bytes after which data is propagated to disk.
-	// The default is 64 MB.
+	// The default is 2 MB.
 	if options.MaxTableSize != nil {
 		opts = opts.WithBaseTableSize(*options.MaxTableSize)
 	}
 
 	// ValueLogMaxEntries sets the maximum number of entries a value log file can hold approximately.
 	// A actual size limit of a value log file is the minimum of ValueLogFileSize and
 	// ValueLogMaxEntries.
-	// The default value of ValueLogMaxEntries is 1 Million
+	// The default value of ValueLogMaxEntries is one million (1000000)
 	if options.ValueLogMaxEntries != nil {
 		opts = opts.WithValueLogMaxEntries(*options.ValueLogMaxEntries)
 	}