<commit_before>// Copyright 2017 syzkaller project authors. All rights reserved.
// Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.

package main

import (
	"bytes"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"path/filepath"
	"strings"
	"time"

	"github.com/google/syzkaller/dashboard/dashapi"
	"github.com/google/syzkaller/pkg/bisect"
	"github.com/google/syzkaller/pkg/build"
	"github.com/google/syzkaller/pkg/instance"
	"github.com/google/syzkaller/pkg/log"
	"github.com/google/syzkaller/pkg/mgrconfig"
	"github.com/google/syzkaller/pkg/osutil"
	"github.com/google/syzkaller/pkg/report"
	"github.com/google/syzkaller/pkg/vcs"
	"github.com/google/syzkaller/vm"
)

type JobProcessor struct {
	cfg             *Config
	name            string
	managers        []*Manager
	knownCommits    map[string]bool
	stop            chan struct{}
	shutdownPending chan struct{}
	dash            *dashapi.Dashboard
	syzkallerRepo   string
	syzkallerBranch string
}

func newJobProcessor(cfg *Config, managers []*Manager, stop, shutdownPending chan struct{}) *JobProcessor {
	return &JobProcessor{
		cfg:             cfg,
		name:            fmt.Sprintf("%v-job", cfg.Name),
		managers:        managers,
		knownCommits:    make(map[string]bool),
		stop:            stop,
		shutdownPending: shutdownPending,
		dash:            dashapi.New(cfg.DashboardClient, cfg.DashboardAddr, cfg.DashboardKey),
		syzkallerRepo:   cfg.SyzkallerRepo,
		syzkallerBranch: cfg.SyzkallerBranch,
	}
}

func (jp *JobProcessor) loop() {
	jobTicker := time.NewTicker(time.Duration(jp.cfg.JobPollPeriod) * time.Second)
	commitTicker := time.NewTicker(time.Duration(jp.cfg.CommitPollPeriod) * time.Second)
	defer jobTicker.Stop()
	defer commitTicker.Stop()
loop:
	for {
		// Check jp.stop separately first, otherwise if stop signal arrives during a job execution,
		// we can grab the next one with 50% probability.
		select {
		case <-jp.stop:
			break loop
		default:
		}
		select {
		case <-jobTicker.C:
			if len(kernelBuildSem) != 0 {
				// If normal kernel build is in progress (usually on start), don't query jobs.
				// Otherwise we claim a job, but can't start it for a while.
				continue loop
			}
			jp.pollJobs()
		case <-commitTicker.C:
			jp.pollCommits()
		case <-jp.stop:
			break loop
		}
	}
	log.Logf(0, "job loop stopped")
}

func (jp *JobProcessor) pollCommits() {
	for _, mgr := range jp.managers {
		if !mgr.mgrcfg.Jobs.PollCommits {
			continue
		}
		if err := jp.pollManagerCommits(mgr); err != nil {
			jp.Errorf("failed to poll commits on %v: %v", mgr.name, err)
		}
	}
}

func brokenRepo(url string) bool {
	// TODO(dvyukov): mmots contains weird squashed commits titled "linux-next" or "origin",
	// which contain hundreds of other commits. This makes fix attribution totally broken.
	return strings.Contains(url, "git.cmpxchg.org/linux-mmots")
}

func (jp *JobProcessor) pollManagerCommits(mgr *Manager) error {
	resp, err := mgr.dash.CommitPoll()
	if err != nil {
		return err
	}
	log.Logf(0, "polling commits for %v: repos %v, commits %v", mgr.name, len(resp.Repos), len(resp.Commits))
	if len(resp.Repos) == 0 {
		return fmt.Errorf("no repos")
	}
	commits := make(map[string]*vcs.Commit)
	for i, repo := range resp.Repos {
		if brokenRepo(repo.URL) {
			continue
		}
		if resp.ReportEmail != "" {
			commits1, err := jp.pollRepo(mgr, repo.URL, repo.Branch, resp.ReportEmail)
			if err != nil {
				jp.Errorf("failed to poll %v %v: %v", repo.URL, repo.Branch, err)
				continue
			}
			log.Logf(1, "got %v commits from %v/%v repo", len(commits1), repo.URL, repo.Branch)
			for _, com := range commits1 {
				// Only the "main" repo is the source of true hashes.
				if i != 0 {
					com.Hash = ""
				}
				// Not overwrite existing commits, in particular commit from the main repo with hash.
				if _, ok := commits[com.Title]; !ok && !jp.knownCommits[com.Title] && len(commits) < 100 {
					commits[com.Title] = com
					jp.knownCommits[com.Title] = true
				}
			}
		}
		if i == 0 && len(resp.Commits) != 0 {
			commits1, err := jp.getCommitInfo(mgr, repo.URL, repo.Branch, resp.Commits)
			if err != nil {
				jp.Errorf("failed to poll %v %v: %v", repo.URL, repo.Branch, err)
				continue
			}
			log.Logf(1, "got %v commit infos from %v/%v repo", len(commits1), repo.URL, repo.Branch)
			for _, com := range commits1 {
				// GetCommitByTitle does not accept ReportEmail and does not return tags,
				// so don't replace the existing commit.
				if _, ok := commits[com.Title]; !ok {
					commits[com.Title] = com
				}
			}
		}
	}
	results := make([]dashapi.Commit, 0, len(commits))
	for _, com := range commits {
		results = append(results, dashapi.Commit{
			Hash:   com.Hash,
			Title:  com.Title,
			Author: com.Author,
			BugIDs: com.Tags,
			Date:   com.Date,
		})
	}
	return mgr.dash.UploadCommits(results)
}

func (jp *JobProcessor) pollRepo(mgr *Manager, URL, branch, reportEmail string) ([]*vcs.Commit, error) {
	dir := osutil.Abs(filepath.Join("jobs", mgr.managercfg.TargetOS, "kernel"))
	repo, err := vcs.NewRepo(mgr.managercfg.TargetOS, mgr.managercfg.Type, dir)
	if err != nil {
		return nil, fmt.Errorf("failed to create kernel repo: %v", err)
	}
	if _, err = repo.CheckoutBranch(URL, branch); err != nil {
		return nil, fmt.Errorf("failed to checkout kernel repo %v/%v: %v", URL, branch, err)
	}
	return repo.ExtractFixTagsFromCommits("HEAD", reportEmail)
}

func (jp *JobProcessor) getCommitInfo(mgr *Manager, URL, branch string, commits []string) ([]*vcs.Commit, error) {
	dir := osutil.Abs(filepath.Join("jobs", mgr.managercfg.TargetOS, "kernel"))
	repo, err := vcs.NewRepo(mgr.managercfg.TargetOS, mgr.managercfg.Type, dir)
	if err != nil {
		return nil, fmt.Errorf("failed to create kernel repo: %v", err)
	}
	if _, err = repo.CheckoutBranch(URL, branch); err != nil {
		return nil, fmt.Errorf("failed to checkout kernel repo %v/%v: %v", URL, branch, err)
	}
	results, missing, err := repo.GetCommitsByTitles(commits)
	if err != nil {
		return nil, err
	}
	for _, title := range missing {
		log.Logf(0, "did not find commit %q", title)
	}
	return results, nil
}

func (jp *JobProcessor) pollJobs() {
	poll := &dashapi.JobPollReq{
		Managers: make(map[string]dashapi.ManagerJobs),
	}
	for _, mgr := range jp.managers {
		if !mgr.mgrcfg.Jobs.TestPatches &&
			!mgr.mgrcfg.Jobs.BisectCause &&
			!mgr.mgrcfg.Jobs.BisectFix {
			continue<commit_msg>syz-ci/jobs: improve log message for non-found git commits<commit_after>// Copyright 2017 syzkaller project authors. All rights reserved.
// Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.

package main

import (
	"bytes"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"path/filepath"
	"strings"
	"time"

	"github.com/google/syzkaller/dashboard/dashapi"
	"github.com/google/syzkaller/pkg/bisect"
	"github.com/google/syzkaller/pkg/build"
	"github.com/google/syzkaller/pkg/instance"
	"github.com/google/syzkaller/pkg/log"
	"github.com/google/syzkaller/pkg/mgrconfig"
	"github.com/google/syzkaller/pkg/osutil"
	"github.com/google/syzkaller/pkg/report"
	"github.com/google/syzkaller/pkg/vcs"
	"github.com/google/syzkaller/vm"
)

type JobProcessor struct {
	cfg             *Config
	name            string
	managers        []*Manager
	knownCommits    map[string]bool
	stop            chan struct{}
	shutdownPending chan struct{}
	dash            *dashapi.Dashboard
	syzkallerRepo   string
	syzkallerBranch string
}

func newJobProcessor(cfg *Config, managers []*Manager, stop, shutdownPending chan struct{}) *JobProcessor {
	return &JobProcessor{
		cfg:             cfg,
		name:            fmt.Sprintf("%v-job", cfg.Name),
		managers:        managers,
		knownCommits:    make(map[string]bool),
		stop:            stop,
		shutdownPending: shutdownPending,
		dash:            dashapi.New(cfg.DashboardClient, cfg.DashboardAddr, cfg.DashboardKey),
		syzkallerRepo:   cfg.SyzkallerRepo,
		syzkallerBranch: cfg.SyzkallerBranch,
	}
}

func (jp *JobProcessor) loop() {
	jobTicker := time.NewTicker(time.Duration(jp.cfg.JobPollPeriod) * time.Second)
	commitTicker := time.NewTicker(time.Duration(jp.cfg.CommitPollPeriod) * time.Second)
	defer jobTicker.Stop()
	defer commitTicker.Stop()
loop:
	for {
		// Check jp.stop separately first, otherwise if stop signal arrives during a job execution,
		// we can grab the next one with 50% probability.
		select {
		case <-jp.stop:
			break loop
		default:
		}
		select {
		case <-jobTicker.C:
			if len(kernelBuildSem) != 0 {
				// If normal kernel build is in progress (usually on start), don't query jobs.
				// Otherwise we claim a job, but can't start it for a while.
				continue loop
			}
			jp.pollJobs()
		case <-commitTicker.C:
			jp.pollCommits()
		case <-jp.stop:
			break loop
		}
	}
	log.Logf(0, "job loop stopped")
}

func (jp *JobProcessor) pollCommits() {
	for _, mgr := range jp.managers {
		if !mgr.mgrcfg.Jobs.PollCommits {
			continue
		}
		if err := jp.pollManagerCommits(mgr); err != nil {
			jp.Errorf("failed to poll commits on %v: %v", mgr.name, err)
		}
	}
}

func brokenRepo(url string) bool {
	// TODO(dvyukov): mmots contains weird squashed commits titled "linux-next" or "origin",
	// which contain hundreds of other commits. This makes fix attribution totally broken.
	return strings.Contains(url, "git.cmpxchg.org/linux-mmots")
}

func (jp *JobProcessor) pollManagerCommits(mgr *Manager) error {
	resp, err := mgr.dash.CommitPoll()
	if err != nil {
		return err
	}
	log.Logf(0, "polling commits for %v: repos %v, commits %v", mgr.name, len(resp.Repos), len(resp.Commits))
	if len(resp.Repos) == 0 {
		return fmt.Errorf("no repos")
	}
	commits := make(map[string]*vcs.Commit)
	for i, repo := range resp.Repos {
		if brokenRepo(repo.URL) {
			continue
		}
		if resp.ReportEmail != "" {
			commits1, err := jp.pollRepo(mgr, repo.URL, repo.Branch, resp.ReportEmail)
			if err != nil {
				jp.Errorf("failed to poll %v %v: %v", repo.URL, repo.Branch, err)
				continue
			}
			log.Logf(1, "got %v commits from %v/%v repo", len(commits1), repo.URL, repo.Branch)
			for _, com := range commits1 {
				// Only the "main" repo is the source of true hashes.
				if i != 0 {
					com.Hash = ""
				}
				// Not overwrite existing commits, in particular commit from the main repo with hash.
				if _, ok := commits[com.Title]; !ok && !jp.knownCommits[com.Title] && len(commits) < 100 {
					commits[com.Title] = com
					jp.knownCommits[com.Title] = true
				}
			}
		}
		if i == 0 && len(resp.Commits) != 0 {
			commits1, err := jp.getCommitInfo(mgr, repo.URL, repo.Branch, resp.Commits)
			if err != nil {
				jp.Errorf("failed to poll %v %v: %v", repo.URL, repo.Branch, err)
				continue
			}
			log.Logf(1, "got %v commit infos from %v/%v repo", len(commits1), repo.URL, repo.Branch)
			for _, com := range commits1 {
				// GetCommitByTitle does not accept ReportEmail and does not return tags,
				// so don't replace the existing commit.
				if _, ok := commits[com.Title]; !ok {
					commits[com.Title] = com
				}
			}
		}
	}
	results := make([]dashapi.Commit, 0, len(commits))
	for _, com := range commits {
		results = append(results, dashapi.Commit{
			Hash:   com.Hash,
			Title:  com.Title,
			Author: com.Author,
			BugIDs: com.Tags,
			Date:   com.Date,
		})
	}
	return mgr.dash.UploadCommits(results)
}

func (jp *JobProcessor) pollRepo(mgr *Manager, URL, branch, reportEmail string) ([]*vcs.Commit, error) {
	dir := osutil.Abs(filepath.Join("jobs", mgr.managercfg.TargetOS, "kernel"))
	repo, err := vcs.NewRepo(mgr.managercfg.TargetOS, mgr.managercfg.Type, dir)
	if err != nil {
		return nil, fmt.Errorf("failed to create kernel repo: %v", err)
	}
	if _, err = repo.CheckoutBranch(URL, branch); err != nil {
		return nil, fmt.Errorf("failed to checkout kernel repo %v/%v: %v", URL, branch, err)
	}
	return repo.ExtractFixTagsFromCommits("HEAD", reportEmail)
}

func (jp *JobProcessor) getCommitInfo(mgr *Manager, URL, branch string, commits []string) ([]*vcs.Commit, error) {
	dir := osutil.Abs(filepath.Join("jobs", mgr.managercfg.TargetOS, "kernel"))
	repo, err := vcs.NewRepo(mgr.managercfg.TargetOS, mgr.managercfg.Type, dir)
	if err != nil {
		return nil, fmt.Errorf("failed to create kernel repo: %v", err)
	}
	if _, err = repo.CheckoutBranch(URL, branch); err != nil {
		return nil, fmt.Errorf("failed to checkout kernel repo %v/%v: %v", URL, branch, err)
	}
	results, missing, err := repo.GetCommitsByTitles(commits)
	if err != nil {
		return nil, err
	}
	for _, title := range missing {
		log.Logf(0, "did not find commit %q in kernel repo %v/%v", title, URL, branch)
	}
	return results, nil
}

func (jp *JobProcessor) pollJobs() {
	poll := &dashapi.JobPollReq{
		Managers: make(map[string]dashapi.ManagerJobs),
	}
	for _, mgr := range jp.managers {
		if !mgr.mgrcfg.Jobs.TestPatches &&
			!mgr.mgrcfg.Jobs.BisectCause &&
			!mgr.mgrcfg.Jobs.BisectFix {
			continue
{'fraction_non_alphanumeric': 0.13846392793352488, 'fraction_numerical': 0.004193523336180787, 'mean_word_length': 3.4437543133195305, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '177369', 'n_tokens_mistral': 5011, 'n_tokens_neox': 4708, 'n_words': 1196}