make new repo analysis under experiment

Author: jirfag

README.md

@@ -50,7 +50,8 @@ Configurate via `.env` file. Dev `.env` may be like this:
 REDIS_URL="redis://localhost:6379"
 API_URL="https://api.dev.golangci.com"
 WEB_ROOT="https://dev.golangci.com"
-USE_CONTAINER_EXECUTOR=1
+USE_CONTAINER_EXECUTOR_PERCENT=100
+USE_NEW_REPO_ANALYSIS_PERCENT=100
 ORCHESTRATOR_ADDR="http://127.0.0.1:8001"
 ORCHESTRATOR_TOKEN=secret_token
 ```

app/analyze/analyzequeue/consume.go

@@ -2,17 +2,32 @@ package analyzequeue
 
 import (
 	"fmt"
-	"log"
 
+	"github.com/golangci/golangci-shared/pkg/apperrors"
+	"github.com/golangci/golangci-shared/pkg/config"
+	"github.com/golangci/golangci-shared/pkg/logutil"
 	"github.com/golangci/golangci-worker/app/analyze/analyzequeue/consumers"
+	"github.com/golangci/golangci-worker/app/analyze/processors"
+	"github.com/golangci/golangci-worker/app/lib/experiments"
 	"github.com/golangci/golangci-worker/app/lib/queue"
 )
 
 func RegisterTasks() {
+	log := logutil.NewStderrLog("repo analysis")
+	log.SetLevel(logutil.LogLevelInfo)
+	cfg := config.NewEnvConfig(log)
+	et := apperrors.GetTracker(cfg, log, "worker")
+
+	trackedLog := apperrors.WrapLogWithTracker(log, nil, et)
+	ec := experiments.NewChecker(cfg, trackedLog)
+
+	rpf := processors.NewRepoProcessorFactory(&processors.StaticRepoConfig{}, trackedLog)
+	repoAnalyzer := consumers.NewAnalyzeRepo(ec, rpf)
+
 	server := queue.GetServer()
 	err := server.RegisterTasks(map[string]interface{}{
 		"analyzeV2":   consumers.NewAnalyzePR().Consume,
-		"analyzeRepo": consumers.NewAnalyzeRepo().Consume,
+		"analyzeRepo": repoAnalyzer.Consume,
 	})
 	if err != nil {
 		log.Fatalf("Can't register queue tasks: %s", err)

app/analyze/analyzequeue/consumers/analyze_repo.go

@@ -2,24 +2,32 @@ package consumers
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"os"
+	"strings"
 	"time"
 
 	"github.com/golangci/golangci-worker/app/analytics"
 	"github.com/golangci/golangci-worker/app/analyze/processors"
+	"github.com/golangci/golangci-worker/app/lib/experiments"
+	"github.com/golangci/golangci-worker/app/lib/github"
+	"github.com/pkg/errors"
 )
 
 type AnalyzeRepo struct {
 	baseConsumer
+
+	ec  *experiments.Checker
+	rpf *processors.RepoProcessorFactory
 }
 
-func NewAnalyzeRepo() *AnalyzeRepo {
+func NewAnalyzeRepo(ec *experiments.Checker, rpf *processors.RepoProcessorFactory) *AnalyzeRepo {
 	return &AnalyzeRepo{
 		baseConsumer: baseConsumer{
 			eventName: analytics.EventRepoAnalyzed,
 		},
+		ec:  ec,
+		rpf: rpf,
 	}
 }
 
@@ -36,7 +44,7 @@ func (c AnalyzeRepo) Consume(ctx context.Context, repoName, analysisGUID, branch
 		return errors.New("repo analysis is disabled")
 	}
 
-	_ = c.wrapConsuming(ctx, func() error {
+	return c.wrapConsuming(ctx, func() error {
 		var cancel context.CancelFunc
 		// If you change timeout value don't forget to change it
 		// in golangci-api stale analyzes checker
@@ -45,12 +53,35 @@ func (c AnalyzeRepo) Consume(ctx context.Context, repoName, analysisGUID, branch
 
 		return c.analyzeRepo(ctx, repoName, analysisGUID, branch)
 	})
-
-	// Don't return error to machinery: we will retry this task ourself from golangci-api
-	return nil
 }
 
 func (c AnalyzeRepo) analyzeRepo(ctx context.Context, repoName, analysisGUID, branch string) error {
+	parts := strings.Split(repoName, "/")
+	repo := &github.Repo{
+		Owner: parts[0],
+		Name:  parts[1],
+	}
+	if len(parts) != 2 {
+		return fmt.Errorf("invalid repo name %s", repoName)
+	}
+
+	if c.ec.IsActiveForAnalysis("use_new_repo_analysis", repo, false) {
+		repoCtx := &processors.RepoContext{
+			Ctx:          ctx,
+			AnalysisGUID: analysisGUID,
+			Branch:       branch,
+			Repo:         repo,
+		}
+		p, cleanup, err := c.rpf.BuildProcessor(repoCtx)
+		if err != nil {
+			return errors.Wrap(err, "failed to build repo processor")
+		}
+		defer cleanup()
+
+		p.Process(repoCtx)
+		return nil
+	}
+
 	p, err := processors.NewGithubGoRepo(ctx, processors.GithubGoRepoConfig{}, analysisGUID, repoName, branch)
 	if err != nil {
 		return fmt.Errorf("can't make github go repo processor: %s", err)

app/analyze/analyzequeue/consumers/base_consumer.go

@@ -27,6 +27,7 @@ func (c baseConsumer) wrapConsuming(ctx context.Context, f func() error) (err er
 	defer func() {
 		if r := recover(); r != nil {
 			err = fmt.Errorf("panic recovered: %v, %s, source is %s", r, debug.Stack(), err)
+			analytics.Log(ctx).Errorf("processing of %q task failed: %s", c.eventName, err)
 		}
 	}()
 

app/analyze/processors/errors.go

@@ -1,11 +1,17 @@
 package processors
 
 import (
+	"errors"
+	"os"
 	"strings"
 
 	"github.com/golangci/golangci-worker/app/lib/github"
 )
 
+var (
+	errNothingToAnalyze = errors.New("nothing to analyze")
+)
+
 type IgnoredError struct {
 	Status        github.Status
 	StatusDesc    string
@@ -24,3 +30,22 @@ func escapeErrorText(text string, secrets map[string]string) string {
 
 	return ret
 }
+
+func buildSecrets() map[string]string {
+	const hidden = "{hidden}"
+	ret := map[string]string{}
+
+	for _, kv := range os.Environ() {
+		parts := strings.Split(kv, "=")
+		if len(parts) != 2 {
+			continue
+		}
+
+		v := parts[1]
+		if len(v) >= 6 {
+			ret[v] = hidden
+		}
+	}
+
+	return ret
+}

app/analyze/processors/executor.go

@@ -3,80 +3,27 @@ package processors
 import (
 	"context"
 	"fmt"
-	"hash/fnv"
 	"os"
-	"strconv"
-	"strings"
 
+	"github.com/golangci/golangci-shared/pkg/config"
 	"github.com/golangci/golangci-shared/pkg/logutil"
 	"github.com/golangci/golangci-worker/app/lib/executors"
+	"github.com/golangci/golangci-worker/app/lib/experiments"
 	"github.com/golangci/golangci-worker/app/lib/github"
 	"github.com/pkg/errors"
 )
 
-func hash(s string) uint32 {
-	h := fnv.New32a()
-	_, _ = h.Write([]byte(s))
-	return h.Sum32()
-}
-
-func getEnabledRepoNamesForContainerExperiment() map[string]bool {
-	repos := os.Getenv("CONTAINER_EXECUTOR_EXPERIMENT_REPOS")
-	if repos == "" {
-		return map[string]bool{}
-	}
-
-	repoList := strings.Split(repos, ",")
-	ret := map[string]bool{}
-	for _, r := range repoList {
-		ret[r] = true
-	}
-
-	return ret
-}
-
-func isContainerExecutorExperimentEnabled(repo *github.Repo) bool {
-	enabledRepos := getEnabledRepoNamesForContainerExperiment()
-	if enabledRepos[repo.FullName()] {
-		return true
-	}
-
-	percentStr := os.Getenv("CONTAINER_EXECUTOR_EXPERIMENT_PERCENT")
-	if percentStr == "" {
-		return false
+func makeExecutor(ctx context.Context, repo *github.Repo, forPull bool, log logutil.Log, ec *experiments.Checker) (executors.Executor, error) {
+	if log == nil { // TODO: remove
+		log = logutil.NewStderrLog("executor")
+		log.SetLevel(logutil.LogLevelInfo)
 	}
-
-	percent, err := strconv.Atoi(percentStr)
-	if err != nil {
-		return false
-	}
-
-	if percent < 0 || percent > 100 {
-		return false
-	}
-
-	hash := hash(fmt.Sprintf("%s/%s", repo.Owner, repo.Name))
-	return uint32(percent) > (hash % 100)
-}
-
-func makeExecutor(ctx context.Context, repo *github.Repo) (executors.Executor, error) {
-	var exec executors.Executor
-	log := logutil.NewStderrLog("executor")
-	log.SetLevel(logutil.LogLevelInfo)
-
-	var useContainerExecutor bool
-	useCE := os.Getenv("USE_CONTAINER_EXECUTOR")
-	if useCE != "" { //nolint:gocritic
-		useContainerExecutor = useCE == "1"
-		log.Infof("Container executor is enabled by env var: %t", useContainerExecutor)
-	} else if isContainerExecutorExperimentEnabled(repo) {
-		useContainerExecutor = true
-		log.Infof("Container executor is enabled by experiment")
-	} else {
-		log.Infof("Container executor is disabled, use remote shell")
+	if ec == nil { // TODO: remove
+		cfg := config.NewEnvConfig(log)
+		ec = experiments.NewChecker(cfg, log)
 	}
 
-	if useContainerExecutor {
+	if ec.IsActiveForAnalysis("use_container_executor", repo, forPull) {
 		ce, err := executors.NewContainer(log)
 		if err != nil {
 			return nil, errors.Wrap(err, "can't build container executor")
@@ -85,19 +32,17 @@ func makeExecutor(ctx context.Context, repo *github.Repo) (executors.Executor, e
 		if err = ce.Setup(ctx); err != nil {
 			return nil, errors.Wrap(err, "failed to setup container executor")
 		}
-		exec = ce.WithWorkDir("/goapp")
-	} else {
-		s := executors.NewRemoteShell(
-			os.Getenv("REMOTE_SHELL_USER"),
-			os.Getenv("REMOTE_SHELL_HOST"),
-			os.Getenv("REMOTE_SHELL_KEY_FILE_PATH"),
-		)
-		if err := s.SetupTempWorkDir(ctx); err != nil {
-			return nil, fmt.Errorf("can't setup temp work dir: %s", err)
-		}
+		return ce.WithWorkDir("/goapp"), nil
+	}
 
-		exec = s
+	s := executors.NewRemoteShell(
+		os.Getenv("REMOTE_SHELL_USER"),
+		os.Getenv("REMOTE_SHELL_HOST"),
+		os.Getenv("REMOTE_SHELL_KEY_FILE_PATH"),
+	)
+	if err := s.SetupTempWorkDir(ctx); err != nil {
+		return nil, fmt.Errorf("can't setup temp work dir: %s", err)
 	}
 
-	return exec, nil
+	return s, nil
 }

app/analyze/processors/github_go_pr.go

@@ -9,7 +9,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/golangci/golangci-api/pkg/app/ensuredeps"
+	"github.com/golangci/golangci-api/pkg/goenv/ensuredeps"
 
 	"github.com/golangci/golangci-worker/app/analytics"
 	"github.com/golangci/golangci-worker/app/analyze/linters"
@@ -60,7 +60,7 @@ func newGithubGoPR(ctx context.Context, c *github.Context, cfg githubGoPRConfig,
 
 	if cfg.exec == nil {
 		var err error
-		cfg.exec, err = makeExecutor(ctx, &c.Repo)
+		cfg.exec, err = makeExecutor(ctx, &c.Repo, true, nil, nil)
 		if err != nil {
 			return nil, fmt.Errorf("can't make executor: %s", err)
 		}

app/analyze/processors/github_go_repo.go

@@ -8,7 +8,7 @@ import (
 	"runtime/debug"
 	"strings"
 
-	"github.com/golangci/golangci-api/pkg/app/ensuredeps"
+	"github.com/golangci/golangci-api/pkg/goenv/ensuredeps"
 	"github.com/golangci/golangci-worker/app/analytics"
 	"github.com/golangci/golangci-worker/app/analyze/linters"
 	"github.com/golangci/golangci-worker/app/analyze/linters/golinters"
@@ -55,7 +55,7 @@ func NewGithubGoRepo(ctx context.Context, cfg GithubGoRepoConfig, analysisGUID,
 
 	if cfg.exec == nil {
 		var err error
-		cfg.exec, err = makeExecutor(ctx, repo)
+		cfg.exec, err = makeExecutor(ctx, repo, true, nil, nil)
 		if err != nil {
 			return nil, fmt.Errorf("can't make executor: %s", err)
 		}