pinpoint/go/workflows/internal/cbb_runner.go

// Workflow to run all CBB benchmarks on a particular browser / device

package internal

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/url"
	"path"
	"strings"
	"time"

	"go.temporal.io/sdk/temporal"
	"go.temporal.io/sdk/workflow"
	"golang.org/x/text/cases"
	"golang.org/x/text/language"

	"go.skia.org/infra/go/skerr"
	"go.skia.org/infra/go/sklog"
	"go.skia.org/infra/perf/go/ingest/format"
	"go.skia.org/infra/perf/go/perfresults"
	"go.skia.org/infra/pinpoint/go/common"
	"go.skia.org/infra/pinpoint/go/workflows"
)

// CbbRunnerParams defines the parameters for CbbRunnerWorkflow.
type CbbRunnerParams struct {
	// The commit to run the benchmarks on. For CBB, this should be a commit in
	// the Chromium main branch, with updated CBB marker file.
	Commit *common.CombinedCommit

	// Name of the device configuration, e.g. "mac-m3-pro-perf-cbb".
	BotConfig string

	// Name of the Browser to test. Supports "chrome", "safari", and "edge".
	Browser string

	// Browser channel to test. All browsers support "stable". Chrome and Edge
	// also support "dev", while Safari also supports "technology-preview".
	Channel string

	// Name of the Google cloud storage bucket to upload results to.
	Bucket string

	// The set of benchmarks to run, and the number of iterations for each.
	// If nil, use default.
	Benchmarks []BenchmarkRunConfig

	// SkipFinch is a bool flag used to disable certain Finch variation control.
	// Only used for Chrome on desktop platforms.
	// For Chrome Dev, it means passing --disable-field-trial-config to Chrome.
	// For Chrome Stable, it means not using Finch top variations seed file.
	SkipFinch bool
}

// CbbRunnerError is a custom error type for CbbRunnerWorkflow.
type CbbRunnerError struct {
	// SwarmingLinks maps benchmark names to their corresponding swarming task list links.
	SwarmingLinks       map[string]string
	Err                 string
	WorkflowLink        string
	TotalBenchmarkCount int
}

func (e *CbbRunnerError) Error() string {
	return e.Err
}

// Configuration for a particular benchmark.
type BenchmarkRunConfig struct {
	Benchmark  string
	Iterations int32
}

func validateParameters(cbb *CbbRunnerParams) error {
	switch cbb.Browser {
	case "chrome", "edge":
		if cbb.Channel != "stable" && cbb.Channel != "dev" {
			return skerr.Fmt(
				"Unrecognized browser channel %s, %s only supports stable and dev",
				cbb.Channel, cbb.Browser)
		}
	case "safari":
		if cbb.Channel != "stable" && cbb.Channel != "technology-preview" {
			return skerr.Fmt(
				"Unrecognized browser channel %s, %s only supports stable and technology-preview",
				cbb.Channel, cbb.Browser)
		}
	default:
		return skerr.Fmt(
			"Unrecognized browser %s, only chrome, safari, and edge are supported",
			cbb.Browser)
	}
	return nil
}

func setupBenchmarks(cbb *CbbRunnerParams) []BenchmarkRunConfig {
	if cbb.Benchmarks != nil {
		return cbb.Benchmarks
	}

	botConfig := cbb.BotConfig
	var benchmarks []BenchmarkRunConfig
	if strings.HasPrefix(botConfig, "mac-") {
		benchmarks = append(benchmarks, BenchmarkRunConfig{"speedometer3", 150})
	} else {
		benchmarks = append(benchmarks, BenchmarkRunConfig{"speedometer3", 22})
	}

	benchmarks = append(benchmarks,
		BenchmarkRunConfig{"jetstream2", 22},
		BenchmarkRunConfig{"jetstream3", 22},
		BenchmarkRunConfig{"motionmark1.3", 22},
	)
	if strings.HasPrefix(botConfig, "android-") {
		// Loadline2 is only available on Android. It also takes considerably
		// longer than other benchmarks, having 50 iterations built-in,
		// so give it a lower repetition count here.
		switch botConfig {
		case "android-pixel-tangor-perf-cbb":
			// CBB has 4 Tangor devices. Run 2 iterations on each device, for a total of 8.
			benchmarks = append(benchmarks, BenchmarkRunConfig{"loadline2_tablet", 8})
		case "android-pixel10-perf-cbb":
			// CBB has 21 Pixel 10 devices. Use two-thirds of these, so that each device
			// only needs to run a single iteration, even when some devices are dead.
			benchmarks = append(benchmarks, BenchmarkRunConfig{"loadline2_phone", 14})
		default:
			// Don't know whether to run loadline2_tablet or loadline2_phone.
			sklog.Errorf("Unknown Android config %s, please add it to setupBenchmarks function in cbb_runner.go", botConfig)
		}
	}

	return benchmarks
}

// Info about the browser we are testing. Retrieved from CBB info file in chromium repo.
type browserInfo struct {
	Browser                    string `json:"browser"`
	Channel                    string `json:"channel"`
	Platform                   string `json:"platform"`
	Version                    string `json:"version"`
	ChromiumMainBranchPosition int    `json:"chromium_main_branch_position"`
}

func getBrowserInfo(ctx workflow.Context, cbb *CbbRunnerParams) (*browserInfo, error) {
	var platformName string
	switch {
	case strings.HasPrefix(cbb.BotConfig, "mac-"):
		platformName = "mac"
	case strings.HasPrefix(cbb.BotConfig, "win-"):
		platformName = "windows"
	case strings.HasPrefix(cbb.BotConfig, "android-"):
		platformName = "android"
	default:
		return nil, skerr.Fmt("Unable to determine platform for bot %s", cbb.BotConfig)
	}
	gitPath := fmt.Sprintf("testing/perf/cbb_ref_info/%s/%s/%s.json", cbb.Browser, cbb.Channel, platformName)

	var content []byte
	if err := workflow.ExecuteActivity(ctx, ReadGitFileActivity, cbb.Commit, gitPath).Get(ctx, &content); err != nil {
		return nil, skerr.Wrapf(err, "Unable to fetch CBB info file %s", gitPath)
	}

	var bi browserInfo
	if err := json.Unmarshal(content, &bi); err != nil {
		return nil, skerr.Wrapf(err, "Unable to parse contents of CBB info file %s", gitPath)
	}
	sklog.Infof("CBB browser info: %v", bi)

	return &bi, nil
}

func setupBrowser(cbb *CbbRunnerParams, bi *browserInfo) string {
	browser := fmt.Sprintf("--official-browser=%s-%s", cbb.Browser, cbb.Channel)
	if cbb.Browser == "chrome" {
		browser = fmt.Sprintf("%s-%s", browser, bi.Version)
	}
	return browser
}

// Generate Pinpoint job ID. Uses abbreviations to keep the job ID short.
func genJobId(bi *browserInfo, cbb *CbbRunnerParams, benchmark string) string {
	browser := getShortBrowserName(bi.Browser, bi.Channel)

	if cbb.SkipFinch {
		browser += " sf"
	}

	version := getShortBrowserVersion(bi.Version, bi.Browser, bi.Channel)
	bot := getShortBotName(cbb.BotConfig)

	// Shorten benchmark name.
	switch {
	case strings.HasPrefix(benchmark, "speedometer"):
		benchmark = "SP"
	case strings.HasPrefix(benchmark, "jetstream2"):
		benchmark = "JS2"
	case strings.HasPrefix(benchmark, "jetstream3"):
		benchmark = "JS3"
	case strings.HasPrefix(benchmark, "loadline2"):
		benchmark = "LL2"
	case strings.HasPrefix(benchmark, "motionmark"):
		benchmark = "MM"
	}

	return fmt.Sprintf("CBB %s %s %s %s", browser, version, bot, benchmark)
}

// getShortBrowserName returns a shortened version of the browser name,
// together with the channel name if it is not Stable.
func getShortBrowserName(browser, channel string) string {
	// * Browser name is shortened to 3 characters.
	// * Channel name is omitted if it is "stable".
	// * Special case: safari technology-preview is abbreviated as "stp".
	if browser == "safari" && channel == "technology-preview" {
		return "stp"
	}
	browser = browser[:3]
	if channel != "stable" {
		browser += " " + channel
	}
	return browser
}

// getShortBrowserVersion returns a shortened browser version number, when appropriate.
func getShortBrowserVersion(version, browser, channel string) string {
	// Chrome and Edge versions are used as-is, but Safari versions are too long.
	// * Safari Stable version looks like "1.2.3 (12345.6.7.8)". Truncate at the space.
	// * STP version looks like "1.2.3 (Release 123, 12345.6.7.8)". Keep only the release number.
	if browser == "safari" {
		if channel == "stable" {
			space := strings.Index(version, " ")
			if space != -1 {
				version = version[:space]
			}
		} else {
			release := strings.Index(version, "Release ")
			if release != -1 {
				version = version[release+8:]
			}
			comma := strings.Index(version, ",")
			if comma != -1 {
				version = version[:comma]
			}
		}
	}

	return version
}

// getShortBotName returns a shortened version of the bot name.
func getShortBotName(bot string) string {
	switch bot {
	case "android-pixel-tangor-perf-cbb":
		return "tang"
	case "android-pixel10-perf-cbb":
		return "p10"
	case "mac-m3-pro-perf-cbb":
		return "m3"
	case "mac-m4-mini-perf-cbb":
		return "m4"
	case "mac-m5-pro-perf-cbb":
		return "m5"
	case "win-victus-perf-cbb":
		return "vic"
	case "win-arm64-snapdragon-elite-perf-cbb":
		return "elite"
	}
	return bot
}

// Workflow to run all CBB benchmarks on a particular browser / bot config and upload the results.
func CbbRunnerWorkflow(ctx workflow.Context, cbb *CbbRunnerParams) (map[string]*format.Format, error) {
	startTime := workflow.Now(ctx)

	ctx = workflow.WithActivityOptions(ctx, regularActivityOptions)

	err := validateParameters(cbb)
	if err != nil {
		return nil, skerr.Wrap(err)
	}

	bi, err := getBrowserInfo(ctx, cbb)
	if err != nil {
		return nil, skerr.Wrap(err)
	}

	benchmarks := setupBenchmarks(cbb)
	browser := setupBrowser(cbb, bi)

	extra_args := []string{browser}
	if bi.Browser == "chrome" && bi.Platform != "android" {
		switch bi.Channel {
		case "stable":
			if !cbb.SkipFinch {
				extra_args = append(
					extra_args,
					"--variations-test-seed-path=../../components/variations/test_data/cipd/single_group_per_study_prefer_existing_behavior/seed.json")
			}
		case "dev":
			if cbb.SkipFinch {
				extra_args = append(extra_args, "--disable-field-trial-config")
			}
		}
	}

	// Due to b/433537961, we need to reinstall browser APK before each
	// benchmark run on Pixel devices.
	if strings.HasPrefix(cbb.BotConfig, "android-pixel") {
		extra_args = append(extra_args, "--reinstall")
	}

	results := map[string]*format.Format{}

	var errs []error
	jobIds := map[string]string{}
	for _, b := range benchmarks {
		jobId := genJobId(bi, cbb, b.Benchmark)
		jobIds[b.Benchmark] = jobId

		p := &SingleCommitRunnerParams{
			PinpointJobID:  jobId,
			BotConfig:      cbb.BotConfig,
			Benchmark:      b.Benchmark,
			Story:          "default",
			CombinedCommit: cbb.Commit,
			Iterations:     b.Iterations,
			ExtraArgs:      extra_args,
		}

		if !strings.HasSuffix(p.Benchmark, ".crossbench") {
			p.Benchmark += ".crossbench"
		}

		options := cbbRunnerChildWorkflowOptions
		options.WorkflowID = p.PinpointJobID
		ctx = workflow.WithChildOptions(ctx, options)
		var cr *CommitRun
		if err := workflow.ExecuteChildWorkflow(ctx, workflows.SingleCommitRunner, p).Get(ctx, &cr); err != nil {
			errs = append(errs, skerr.Wrap(err))
			continue
		}

		// Check the success rate of swarming tasks. We require at least 80% of
		// the tasks to succeed in order to accept the results.
		numSuccess := 0
		for _, run := range cr.Runs {
			if run.Status.IsTaskSuccessful() && len(run.Values) > 0 {
				numSuccess++
			}
		}
		successRate := float64(numSuccess) / float64(b.Iterations)
		requiredSuccessRate := 0.8
		if successRate < requiredSuccessRate {
			errs = append(errs, skerr.Fmt(
				"Benchmark %s on %s only had success rate of %.0f%%, rejecting the results",
				b.Benchmark, cbb.BotConfig, successRate*100))
			continue
		}

		r := formatResult(ctx, cr, cbb.BotConfig, p.Benchmark, bi, cbb.SkipFinch, startTime, p.PinpointJobID)
		results[b.Benchmark] = r

		if cbb.Bucket != "" {
			var swc StringWriterCloser = StringWriterCloser{
				builder: new(strings.Builder),
			}
			if err := r.Write(swc); err != nil {
				errs = append(errs, skerr.Wrap(err))
				continue
			}

			var gsPath string
			if err := workflow.ExecuteActivity(
				ctx, UploadCbbResultsActivity, startTime, cbb, p.Benchmark, swc.builder.String()).Get(ctx, &gsPath); err != nil {
				errs = append(errs, skerr.Wrap(err))
				continue
			}
			sklog.Infof("Uploaded results to %s", gsPath)
		} else {
			sklog.Warning("No GS bucket provided to upload results")
		}
	}

	if len(errs) > 0 {
		swarmingLinks := map[string]string{}
		for benchmark, jobId := range common.SortedRange(jobIds) {
			swarmingLinks[benchmark] = getSwarmingLink(ctx, startTime, jobId)
		}

		err := CbbRunnerError{
			Err:                 errors.Join(errs...).Error(),
			WorkflowLink:        getWorkflowLink(ctx),
			TotalBenchmarkCount: len(benchmarks),
			SwarmingLinks:       swarmingLinks,
		}
		// The CbbRunnerError object must be wrapped in an ApplicationError in
		// order to be successfully passed back to the parent workflow.
		//nolint:wrapcheck // temporal error must be propagated raw
		return nil, temporal.NewApplicationError(err.Err, "CbbRuntimeError", err)
	}

	return results, nil
}

// Provide a StringWriterCloser interface wrapper around strings.Builder,
// required for serializing the results to a string.
type StringWriterCloser struct {
	builder *strings.Builder
}

func (swc StringWriterCloser) Write(p []byte) (int, error) {
	n, err := swc.builder.Write(p)
	if err != nil {
		return n, skerr.Wrap(err)
	}
	return n, nil
}

func (StringWriterCloser) Close() error {
	return nil
}

func getWorkflowLink(ctx workflow.Context) string {
	isDev := strings.HasPrefix(workflow.GetActivityOptions(ctx).TaskQueue, "localhost.")
	var temporalHost string
	if isDev {
		temporalHost = "temporal-ui-dev.corp.goog"
	} else {
		temporalHost = "skia-temporal-ui.corp.goog"
	}
	exec := workflow.GetInfo(ctx).WorkflowExecution
	return fmt.Sprintf(
		"https://%s/namespaces/perf-internal/workflows/%s/%s/history",
		temporalHost, exec.ID, exec.RunID)
}

func getSwarmingLink(ctx workflow.Context, startTime time.Time, jobId string) string {
	// Create a link to all swarming tasks associated with a CBB benchmark run.
	// The link has these parameters:
	// * c=... selects the columns to display.
	// * st=%d Start time of the display.
	// * et=%d End time of the display. Set to the current time, as the
	//   benchmark runs have just ended. Since the link is saved with the test
	//   results, some additional Pinpoint jobs with the same name might have
	//   run by the time a user clicks on the link, in which case we don't want
	//   the future tasks to be included. Having a time range should also speed
	//   up retrieving the task list.
	// * n=false Sets the "Now" flag to false. Without this, the end time would be ignored.
	// * f=... Filter to select the tasks with the expected Pinpoint job ID.
	return fmt.Sprintf(
		"https://chrome-swarming.appspot.com/tasklist?c=name&c=duration&c=bot&c=state&st=%d&et=%d&n=false&f=pinpoint_job_id-tag%%3A%s",
		startTime.UnixMilli(), workflow.Now(ctx).UnixMilli(), url.PathEscape(jobId))
}

// Taking all swarming task results for one benchmark on one bot config,
// and convert the results into the format required by the perf dashboard.
func formatResult(ctx workflow.Context, cr *CommitRun, bot, benchmark string, bi *browserInfo, skipFinch bool, startTime time.Time, jobId string) *format.Format {
	// Create a link to the Temporal workflow that is currently running.
	// This link will be included in the pop-up for the result, to help debugging.
	workflowLink := getWorkflowLink(ctx)

	// Create a link to all swarming tasks associated with these results.
	swarmingLink := getSwarmingLink(ctx, startTime, jobId)

	data := format.Format{
		Version: 1,
		GitHash: fmt.Sprintf("CP:%d", cr.Build.Commit.Main.CommitPosition),
		Key: map[string]string{
			"master":    "ChromiumPerf",
			"bot":       bot,
			"benchmark": benchmark,
		},
		Links: map[string]string{
			"Browser Version": bi.Version,
			"Workflow":        workflowLink,
			"Swarming Tasks":  swarmingLink,
		},
	}

	values := map[string][]float64{}
	units := map[string]string{}
	for _, run := range cr.Runs {
		for c, v := range common.SortedRange(run.Values) {
			values[c] = append(values[c], v...)
		}
		for c, u := range common.SortedRange(run.Units) {
			units[c] = u
		}
	}

	// Form a human-readable browser name and channel string, such as "Chrome Stable".
	browser_id := fmt.Sprintf("%s %s", bi.Browser, bi.Channel)
	browser_id = strings.ReplaceAll(browser_id, "-", " ")
	browser_id = cases.Title(language.English).String(browser_id)
	if bi.Browser == "chrome" && bi.Platform != "android" {
		if bi.Channel == "stable" && !skipFinch {
			browser_id += " (Top Finch Variations)"
		} else if bi.Channel == "dev" && skipFinch {
			browser_id += " (Disable Field Trial)"
		}
	}

	for c, v := range common.SortedRange(values) {
		h := perfresults.Histogram{
			SampleValues: v,
		}
		u := units[c]
		r := format.Result{
			Key: map[string]string{
				"test":      c,
				"unit":      u,
				"subtest_1": browser_id,
			},
			Measurements: map[string][]format.SingleMeasurement{
				// Following the convention used in CBB v2 (bench-o-matic) and v3 (swarming-in-google3),
				// we use median (instead of mean) as the data value. We also return standard deviation
				// as a measurement of noise. Other statistics can be added in the future when needed.
				"stat": {
					{
						Value:       "value",
						Measurement: float32(h.Median()),
					},
					{
						Value:       "error",
						Measurement: float32(h.Stddev()),
					},
				},
			},
		}
		if strings.HasSuffix(units[c], "_smallerIsBetter") {
			r.Key["improvement_direction"] = "down"
		} else if strings.HasSuffix(units[c], "_biggerIsBetter") {
			r.Key["improvement_direction"] = "up"
		}
		data.Results = append(data.Results, r)
	}

	return &data
}

// Activity to upload CBB results to cloud storage, for import into perf dashboard.
func UploadCbbResultsActivity(ctx context.Context, t time.Time, cbb *CbbRunnerParams, benchmark, results string) (string, error) {
	if cbb.Bucket == "" {
		return "", nil
	}
	store, err := NewStore(ctx, cbb.Bucket, false)
	if err != nil {
		return "", skerr.Wrap(err)
	}

	// The file path inside the GS bucket uses a pattern similar to the one used by perf waterfall.
	datePath := fmt.Sprintf("%04d/%02d/%02d", t.Year(), t.Month(), t.Day())
	timestamp := fmt.Sprintf("%02d%02d%02d", t.Hour(), t.Minute(), t.Second())
	filename := fmt.Sprintf(
		"cbb_results_%s_%s_%s_%s_%04d_%02d_%02d_%02d_%02d_%02d.json",
		benchmark, cbb.BotConfig, cbb.Browser, cbb.Channel,
		t.Year(), t.Month(), t.Day(), t.Hour(), t.Minute(), t.Second(),
	)
	storeFilePath := path.Join(
		"ingest", datePath, "ChromiumPerf", cbb.BotConfig, timestamp, benchmark, filename)

	err = store.WriteFile(storeFilePath, results)
	if err != nil {
		return "", skerr.Wrap(err)
	}

	gsPath := fmt.Sprintf("gs://%s/%s", cbb.Bucket, storeFilePath)
	return gsPath, nil
}