consumers count and experiments for logging

jirfag · jirfag · commit 5ec28b340eae · 2019-07-06T11:44:47.000+03:00
diff --git a/pkg/worker/analyze/analyzequeue/consumers/analyze_pr.go b/pkg/worker/analyze/analyzequeue/consumers/analyze_pr.go
@@ -15,6 +15,7 @@ import (
 
 	"github.com/golangci/golangci-api/pkg/worker/analytics"
 	"github.com/golangci/golangci-api/pkg/worker/analyze/processors"
+	"github.com/golangci/golangci-api/pkg/worker/lib/experiments"
 	"github.com/golangci/golangci-api/pkg/worker/lib/github"
 )
 
@@ -26,11 +27,12 @@ type AnalyzePR struct {
 	log        logutil.Log
 }
 
-func NewAnalyzePR(pf processors.PullProcessorFactory, log logutil.Log, errTracker apperrors.Tracker, cfg config.Config) *AnalyzePR {
+func NewAnalyzePR(pf processors.PullProcessorFactory, log logutil.Log, errTracker apperrors.Tracker, cfg config.Config, ec *experiments.Checker) *AnalyzePR {
 	return &AnalyzePR{
 		baseConsumer: baseConsumer{
 			eventName: analytics.EventPRChecked,
 			cfg:       cfg,
+			ec:        ec,
 		},
 		pf:         pf,
 		errTracker: errTracker,
diff --git a/pkg/worker/analyze/analyzequeue/consumers/analyze_pr_test.go b/pkg/worker/analyze/analyzequeue/consumers/analyze_pr_test.go
@@ -7,6 +7,8 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/golangci/golangci-api/pkg/worker/lib/experiments"
+
 	"github.com/golangci/golangci-api/internal/shared/apperrors"
 
 	"github.com/golangci/golangci-api/internal/shared/config"
@@ -41,8 +43,9 @@ func TestAnalyzeRepo(t *testing.T) {
 	log := logutil.NewStderrLog("")
 	cfg := config.NewEnvConfig(log)
 	errTracker := apperrors.NewNopTracker()
+	ec := experiments.NewChecker(cfg, log)
 
-	err := NewAnalyzePR(pf, log, errTracker, cfg).Consume(context.Background(), repoOwner, repoName,
+	err := NewAnalyzePR(pf, log, errTracker, cfg, ec).Consume(context.Background(), repoOwner, repoName,
 		false, cfg.GetString("TEST_GITHUB_TOKEN"), prNumber, "", userID, "test-guid", "commit-sha")
 	assert.NoError(t, err)
 }
diff --git a/pkg/worker/analyze/analyzequeue/consumers/analyze_repo.go b/pkg/worker/analyze/analyzequeue/consumers/analyze_repo.go
@@ -13,6 +13,7 @@ import (
 
 	"github.com/golangci/golangci-api/pkg/worker/analytics"
 	"github.com/golangci/golangci-api/pkg/worker/analyze/processors"
+	"github.com/golangci/golangci-api/pkg/worker/lib/experiments"
 	"github.com/golangci/golangci-api/pkg/worker/lib/github"
 	"github.com/pkg/errors"
 )
@@ -26,11 +27,12 @@ type AnalyzeRepo struct {
 	cfg        config.Config
 }
 
-func NewAnalyzeRepo(rpf *processors.RepoProcessorFactory, log logutil.Log, errTracker apperrors.Tracker, cfg config.Config) *AnalyzeRepo {
+func NewAnalyzeRepo(rpf *processors.RepoProcessorFactory, log logutil.Log, errTracker apperrors.Tracker, cfg config.Config, ec *experiments.Checker) *AnalyzeRepo {
 	return &AnalyzeRepo{
 		baseConsumer: baseConsumer{
 			eventName: analytics.EventRepoAnalyzed,
 			cfg:       cfg,
+			ec:        ec,
 		},
 		rpf:        rpf,
 		log:        log,
diff --git a/pkg/worker/analyze/analyzequeue/consumers/base_consumer.go b/pkg/worker/analyze/analyzequeue/consumers/base_consumer.go
@@ -7,6 +7,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/golangci/golangci-api/pkg/worker/lib/experiments"
+
 	"github.com/golangci/golangci-api/internal/shared/config"
 
 	"github.com/golangci/golangci-api/internal/shared/logutil"
@@ -16,6 +18,7 @@ import (
 type baseConsumer struct {
 	eventName analytics.EventName
 	cfg       config.Config
+	ec        *experiments.Checker
 }
 
 const statusOk = "ok"
@@ -28,22 +31,18 @@ func (c baseConsumer) prepareContext(ctx context.Context, trackingProps map[stri
 }
 
 func (c baseConsumer) getUnrecoverableErrorLogger(log logutil.Log, repoFullName string) logutil.Func {
-	ignoredRepos := c.cfg.GetStringList("LOG_AS_INFO_UNRECOVERABLE_ERRORS_FOR_REPOS")
-	for _, ignoredRepo := range ignoredRepos {
-		if strings.EqualFold(ignoredRepo, repoFullName) {
-			return log.Infof
-		}
+	repoParts := strings.SplitN(repoFullName, "/", 2)
+	if c.ec.IsActiveForRepo("LOG_AS_INFO_UNRECOVERABLE_ERRORS_FOR", repoParts[0], repoParts[1]) {
+		return log.Infof
 	}
 
 	return log.Warnf
 }
 
 func (c baseConsumer) getRecoverableErrorLogger(log logutil.Log, repoFullName string) logutil.Func {
-	ignoredRepos := c.cfg.GetStringList("LOG_AS_INFO_RECOVERABLE_ERRORS_FOR_REPOS")
-	for _, ignoredRepo := range ignoredRepos {
-		if strings.EqualFold(ignoredRepo, repoFullName) {
-			return log.Infof
-		}
+	repoParts := strings.SplitN(repoFullName, "/", 2)
+	if c.ec.IsActiveForRepo("LOG_AS_INFO_RECOVERABLE_ERRORS_FOR", repoParts[0], repoParts[1]) {
+		return log.Infof
 	}
 
 	return log.Errorf
diff --git a/pkg/worker/app/app.go b/pkg/worker/app/app.go
@@ -1,6 +1,11 @@
 package app
 
 import (
+	"fmt"
+	"sync"
+
+	"github.com/golangci/golangci-api/pkg/worker/lib/experiments"
+
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/session"
 	redigo "github.com/garyburd/redigo/redis"
@@ -25,6 +30,7 @@ type App struct {
 	trackedLog      logutil.Log
 	errTracker      apperrors.Tracker
 	cfg             config.Config
+	ec              *experiments.Checker
 	redisPool       *redigo.Pool
 	distLockFactory *redsync.Redsync
 	awsSess         *session.Session
@@ -73,17 +79,20 @@ func (a *App) buildDeps() {
 	if a.ppf == nil {
 		a.ppf = processors.NewBasicPullProcessorFactory(&processors.BasicPullConfig{})
 	}
+	if a.ec == nil {
+		a.ec = experiments.NewChecker(a.cfg, a.trackedLog)
+	}
 }
 
 func (a App) buildMultiplexer() *consumers.Multiplexer {
 	rpf := processors.NewRepoProcessorFactory(&processors.StaticRepoConfig{})
 
 	// it's important to use a.log, not a.trackedLog
-	repoAnalyzer := analyzesConsumers.NewAnalyzeRepo(rpf, a.log, a.errTracker, a.cfg)
+	repoAnalyzer := analyzesConsumers.NewAnalyzeRepo(rpf, a.log, a.errTracker, a.cfg, a.ec)
 	repoAnalyzesRunner := repoanalyzesqueue.NewConsumer(repoAnalyzer)
 
 	// it's important to use a.log, not a.trackedLog
-	pullAnalyzer := analyzesConsumers.NewAnalyzePR(a.ppf, a.log, a.errTracker, a.cfg)
+	pullAnalyzer := analyzesConsumers.NewAnalyzePR(a.ppf, a.log, a.errTracker, a.cfg, a.ec)
 	pullAnalyzesRunner := pullanalyzesqueue.NewConsumer(pullAnalyzer)
 
 	multiplexer := consumers.NewMultiplexer()
@@ -136,10 +145,26 @@ func (a App) BuildTestDeps() *TestDeps {
 
 func (a App) Run() {
 	consumerMultiplexer := a.buildMultiplexer()
-	analyzesSQS := sqs.NewQueue(a.cfg.GetString("SQS_ANALYZES_QUEUE_URL"),
-		a.awsSess, a.trackedLog, analyzesqueue.VisibilityTimeoutSec)
-	consumer := consumer.NewSQS(a.trackedLog, a.cfg, analyzesSQS,
-		consumerMultiplexer, "analyzes", analyzesqueue.VisibilityTimeoutSec)
 
-	consumer.Run()
+	consumersCount := a.cfg.GetInt("CONSUMERS_COUNT", 1)
+	a.log.Infof("Starting %d consumers...", consumersCount)
+
+	var wg sync.WaitGroup
+	wg.Add(consumersCount)
+	for i := 0; i < consumersCount; i++ {
+		go func(i int) {
+			defer wg.Done()
+
+			trackedLog := a.trackedLog.Child(fmt.Sprintf("consumer #%d", i))
+			analyzesSQS := sqs.NewQueue(a.cfg.GetString("SQS_ANALYZES_QUEUE_URL"),
+				a.awsSess, trackedLog, analyzesqueue.VisibilityTimeoutSec)
+			consumer := consumer.NewSQS(trackedLog, a.cfg, analyzesSQS,
+				consumerMultiplexer, "analyzes", analyzesqueue.VisibilityTimeoutSec)
+
+			consumer.Run()
+		}(i)
+	}
+
+	a.log.Infof("Started %d consumers", consumersCount)
+	wg.Wait()
 }
