pinpoint/go/workflows/internal/pairwise.go

package internal

import (
	"github.com/google/shlex"
	swarming_pb "go.chromium.org/luci/swarming/proto/api_v2"

	"go.skia.org/infra/go/skerr"
	"go.skia.org/infra/go/sklog"
	"go.skia.org/infra/pinpoint/go/common"
	"go.skia.org/infra/pinpoint/go/compare"
	"go.skia.org/infra/pinpoint/go/sql/schema"
	"go.skia.org/infra/pinpoint/go/workflows"

	"go.temporal.io/sdk/temporal"
	"go.temporal.io/sdk/workflow"

	pinpoint_proto "go.skia.org/infra/pinpoint/proto/v1"
)

const (
	failed    = "Failed"
	completed = "Completed"
	canceled  = "Canceled"
	running   = "Running"
)

func PairwiseWorkflow(ctx workflow.Context, p *workflows.PairwiseParams) (
	pe *pinpoint_proto.PairwiseExecution, finalError error,
) {
	if p.Request.StartBuild == nil && p.Request.StartCommit == nil {
		return nil, skerr.Fmt("Base build and commit are empty.")
	}
	if p.Request.EndBuild == nil && p.Request.EndCommit == nil {
		return nil, skerr.Fmt("Experiment build and commit are empty.")
	}

	leftCas, err := convertCas(p.Request.StartBuild)
	if err != nil {
		return nil, skerr.Wrapf(err, "start build is invalid")
	}

	rightCas, err := convertCas(p.Request.EndBuild)
	if err != nil {
		return nil, skerr.Wrapf(err, "end build is invalid")
	}

	ctx = workflow.WithChildOptions(ctx, childWorkflowOptions)
	ctx = workflow.WithActivityOptions(ctx, regularActivityOptions)

	dbActivityOptions := regularActivityOptions
	// For local development/testing (when Project is empty or "local"), set database retries to 1 (fail-fast)
	if p.Request.Project == "" || p.Request.Project == "local" {
		dbActivityOptions.RetryPolicy = &temporal.RetryPolicy{
			MaximumAttempts: 1,
		}
	}
	dbCtx := workflow.WithActivityOptions(ctx, dbActivityOptions)

	jobID := workflow.GetInfo(ctx).WorkflowExecution.ID
	workflowStartTime := workflow.Now(ctx)
	if err := workflow.ExecuteActivity(dbCtx, AddInitialJob, p.Request, jobID).Get(dbCtx, nil); err != nil {
		// TODO(b/439651386) Convert subsequent uses of sklog to full errors once Job Store activities are
		// more stable and fully integrated
		sklog.Errorf("failed to add initial job info to Spanner: %s", err)
	}

	// Benchmark runs can sometimes generate an inconsistent number of data points.
	// So even if all benchmark runs were successful, the number of data values
	// generated by commit A vs B will be inconsistent. This can fail the balancing
	// requirement of the statistical test. It can pair up data incorrectly i.e.
	// commit A: [1], [2, 3]
	// commit B: [4, 5], [6]
	// So the analysis will pair up [2,5] together, which are from different runs,
	// violating pairwise analysis
	if p.Request.AggregationMethod == "" {
		p.Request.AggregationMethod = "mean"
	}

	leftExtraArgs, err := splitExtraArgs(p.Request.BaseExtraArgs)
	if err != nil {
		return nil, skerr.Wrap(err)
	}
	rightExtraArgs, err := splitExtraArgs(p.Request.ExperimentExtraArgs)
	if err != nil {
		return nil, skerr.Wrap(err)
	}

	pairwiseRunnerParams := PairwiseCommitsRunnerParams{
		SingleCommitRunnerParams: SingleCommitRunnerParams{
			PinpointJobID:     jobID,
			BotConfig:         p.Request.Configuration,
			Benchmark:         p.Request.Benchmark,
			Chart:             p.Request.Chart,
			Story:             p.Request.Story,
			StoryTags:         p.Request.StoryTags,
			AggregationMethod: p.Request.AggregationMethod,
			Iterations:        p.GetInitialAttempt(),
		},
		LeftCAS:        leftCas,
		RightCAS:       rightCas,
		LeftCommit:     (*common.CombinedCommit)(p.Request.StartCommit),
		RightCommit:    (*common.CombinedCommit)(p.Request.EndCommit),
		LeftExtraArgs:  leftExtraArgs,
		RightExtraArgs: rightExtraArgs,
	}

	mh := workflow.GetMetricsHandler(ctx).WithTags(map[string]string{
		"job_id":    jobID,
		"benchmark": p.Request.Benchmark,
		"config":    p.Request.Configuration,
		"story":     p.Request.Story,
	})
	protoResults := map[string]*pinpoint_proto.PairwiseExecution_WilcoxonResult{}

	defer func() {
		durationTime := workflow.Now(ctx).Sub(workflowStartTime)
		mh.Timer("pairwise_duration").Record(durationTime)
		durationInSeconds := durationTime.Seconds()

		// Create a new disconnected context for cleanup activities.
		// This ensures that even if the workflow context is canceled,
		// these final activities can still be scheduled and run.
		disconnectedCtx, _ := workflow.NewDisconnectedContext(ctx)
		disconnectedDbCtx := workflow.WithActivityOptions(disconnectedCtx, dbActivityOptions)

		// Final writebacks to Spanner before end of Pairwise workflow
		if finalError != nil {
			if temporal.IsCanceledError(finalError) {
				if err := workflow.ExecuteActivity(disconnectedDbCtx, UpdateJobStatus, jobID, canceled, durationInSeconds).Get(disconnectedDbCtx, nil); err != nil {
					sklog.Errorf("couldn't update status for canceled pairwise job with this ID: %s", jobID)
				}
			} else {
				// Pairwise job failed for other reasons.
				if err := workflow.ExecuteActivity(disconnectedDbCtx, SetErrors, jobID, finalError.Error()).Get(disconnectedDbCtx, nil); err != nil {
					sklog.Errorf("couldn't add error for pairwise job with this ID: %s", jobID)
				}
				if err := workflow.ExecuteActivity(disconnectedDbCtx, UpdateJobStatus, jobID, failed, durationInSeconds).Get(disconnectedDbCtx, nil); err != nil {
					sklog.Errorf("couldn't update status for pairwise job with this ID: %s", jobID)
				}
			}
		} else {
			if err := workflow.ExecuteActivity(disconnectedDbCtx, UpdateJobStatus, jobID, completed, durationInSeconds).Get(disconnectedDbCtx, nil); err != nil {
				sklog.Errorf("couldn't update status for pairwise job with this ID: %s", jobID)
			}
			// Write back to database the results of the comparison through the job store object
			if err := workflow.ExecuteActivity(disconnectedDbCtx, AddResults, jobID, protoResults).Get(disconnectedDbCtx, nil); err != nil {
				sklog.Errorf("couldn't add results for pairwise job with this ID: %s", jobID)
			}
		}
	}()

	// Update status to running
	if err := workflow.ExecuteActivity(dbCtx, UpdateJobStatus, jobID, running, int64(0)).Get(dbCtx, nil); err != nil {
		sklog.Errorf("couldn't update status for running pairwise job with this ID: %s", jobID)
	}

	var pr *PairwiseRun
	if err := workflow.ExecuteChildWorkflow(ctx, workflows.PairwiseCommitsRunner, &pairwiseRunnerParams).Get(ctx, &pr); err != nil {
		return nil, skerr.Wrap(err)
	}

	// Store details of commit buids and test runs
	if pr != nil {
		leftData := &schema.CommitRunData{
			Build: pr.Left.Build,
			Runs:  pr.Left.Runs,
		}
		rightData := &schema.CommitRunData{
			Build: pr.Right.Build,
			Runs:  pr.Right.Runs,
		}
		if err := workflow.ExecuteActivity(dbCtx, AddCommitRuns, jobID, leftData, rightData).Get(dbCtx, nil); err != nil {
			sklog.Errorf("couldn't add commit runs for pairwise job with this ID: %s", jobID)
		}
	}

	results, err := comparePairwiseRuns(ctx, pr, compare.UnknownDir)
	if err != nil {
		return nil, skerr.Wrapf(err, "failed to compare pairwise runs")
	}

	// we only return the "culprit" if this is a culprit-verification run.
	// Culprit verification workflows are run in parallel and returned via
	// channels. So the parent culprit_finder workflow is unable to determine
	// which completed child workflow belongs to which set of inputs. We
	// explicitly return this commit to work around this limitation.
	var culpritCandidate *pinpoint_proto.CombinedCommit
	if p.CulpritVerify {
		culpritCandidate = (*pinpoint_proto.CombinedCommit)(pairwiseRunnerParams.RightCommit)
	}

	for chart, res := range common.SortedRange(results) {
		protoResults[chart] = &pinpoint_proto.PairwiseExecution_WilcoxonResult{
			// Significant is used in CulpritFinder to determine whether to bisect.
			// Significant = true means that there's indeed a regression and it should
			// be investigated. If significant is not explicitly set to False, we see
			// Temporal workflows with Significant and Culprit omitted because the resolve
			// to nil.
			Significant:              res.Verdict == compare.Different,
			PValue:                   res.PValue,
			ConfidenceIntervalLower:  res.LowerCi,
			ConfidenceIntervalHigher: res.UpperCi,
			ControlMedian:            res.XMedian,
			TreatmentMedian:          res.YMedian,
		}
	}

	return &pinpoint_proto.PairwiseExecution{
		JobId:               jobID,
		CulpritCandidate:    culpritCandidate,
		Results:             protoResults,
		LeftSwarmingStatus:  pr.Left.GetSwarmingStatus(),
		RightSwarmingStatus: pr.Right.GetSwarmingStatus(),
	}, nil
}

func convertCas(pinpointCas *pinpoint_proto.CASReference) (*swarming_pb.CASReference, error) {
	if pinpointCas == nil {
		return nil, nil
	}
	if pinpointCas.Digest == nil {
		return nil, skerr.Fmt("cas digest cannot be nil: %v", pinpointCas)
	}
	return &swarming_pb.CASReference{
		CasInstance: pinpointCas.CasInstance,
		Digest: &swarming_pb.Digest{
			Hash:      pinpointCas.Digest.Hash,
			SizeBytes: pinpointCas.Digest.SizeBytes,
		},
	}, nil
}

func splitExtraArgs(args string) ([]string, error) {
	if args == "" {
		return nil, nil
	}
	parsed, err := shlex.Split(args)
	if err != nil {
		return nil, skerr.Wrapf(err, "failed to parse extra arguments %q", args)
	}
	return parsed, nil
}