feat: potential SLA breach batch

Makefile

@@ -5,7 +5,7 @@ NAME = "github.com/goto/optimus"
 LAST_COMMIT := $(shell git rev-parse --short HEAD)
 LAST_TAG := "$(shell git rev-list --tags --max-count=1)"
 OPMS_VERSION := "$(shell git describe --tags ${LAST_TAG})-next"
-PROTON_COMMIT := "7089e46264df681494f82d58d8c964cd6167c9a6"
+PROTON_COMMIT := "2b7e601954cc0092716a36a81581d547b6bf4aeb"
 
 
 .PHONY: build test test-ci generate-proto unit-test-ci integration-test vet coverage clean install lint

core/scheduler/handler/v1beta1/job_run.go

@@ -40,6 +40,7 @@ const (
 
 type JobSLAPredictorService interface {
 	IdentifySLABreaches(ctx context.Context, projectName tenant.ProjectName, jobNames []scheduler.JobName, labels map[string]string, reqConfig service.JobSLAPredictorRequestConfig) (map[scheduler.JobName]map[scheduler.JobName]*service.JobState, error)
+	IdentifySLABreachesBatch(ctx context.Context, combos []scheduler.SLABreachCombo, reqConfig service.JobSLAPredictorRequestConfig) (map[string]*service.TargetBreach, error)
 }
 
 type JobExpectatorService interface {
@@ -512,25 +513,56 @@ func (h JobRunHandler) GetInterval(ctx context.Context, req *pb.GetIntervalReque
 	}, nil
 }
 
-// IdentifyPotentialSLABreach predicts potential SLA breaches for the given job targets at their targeted SLA time
-func (h JobRunHandler) IdentifyPotentialSLABreach(ctx context.Context, req *pb.IdentifyPotentialSLABreachRequest) (*pb.IdentifyPotentialSLABreachResponse, error) {
-	response := pb.IdentifyPotentialSLABreachResponse{}
+// buildIdentifySLABreachInputs resolves an IdentifyPotentialSLABreachRequest into
+// the combos (cross product of projects x label groups) to evaluate and the
+// shared predictor config. job_names only applies in legacy single-project mode,
+// where target selection isn't ambiguous.
+func (h JobRunHandler) buildIdentifySLABreachInputs(req *pb.IdentifyPotentialSLABreachRequest) ([]scheduler.SLABreachCombo, service.JobSLAPredictorRequestConfig, error) {
+	// resolve projects: prefer repeated project_names, fall back to path project_name
+	projectNameStrs := req.GetProjectNames()
+	if len(projectNameStrs) == 0 && req.GetProjectName() != "" {
+		projectNameStrs = []string{req.GetProjectName()}
+	}
+	if len(projectNameStrs) == 0 {
+		err := errors.InvalidArgument(scheduler.EntityJobRun, "no project provided")
+		h.l.Error("error identifying potential SLA breaches: %v", err)
+		return nil, service.JobSLAPredictorRequestConfig{}, errors.GRPCErr(err, "unable to identify potential SLA breaches")
+	}
+	projectNames := make([]tenant.ProjectName, 0, len(projectNameStrs))
+	for _, pn := range projectNameStrs {
+		projectName, err := tenant.ProjectNameFrom(pn)
+		if err != nil {
+			h.l.Error("error adapting project name [%s]: %v", pn, err)
+			return nil, service.JobSLAPredictorRequestConfig{}, errors.GRPCErr(err, "unable to adapt project name")
+		}
+		projectNames = append(projectNames, projectName)
+	}
 
 	jobNames := []scheduler.JobName{}
 	for _, jn := range req.GetJobNames() {
 		jobName, err := scheduler.JobNameFrom(jn)
 		if err != nil {
 			h.l.Error("error adapting job name [%s]: %v", jn, err)
-			return nil, errors.GRPCErr(err, "unable to adapt job name")
+			return nil, service.JobSLAPredictorRequestConfig{}, errors.GRPCErr(err, "unable to adapt job name")
 		}
 		jobNames = append(jobNames, jobName)
 	}
 
-	projectName, err := tenant.ProjectNameFrom(req.GetProjectName())
-	if err != nil {
-		h.l.Error("error adapting project name [%s]: %v", req.GetProjectName(), err)
-		return nil, errors.GRPCErr(err, "unable to adapt project name")
+	// resolve label groups: prefer repeated label_groups, fall back to job_labels.
+	// severity is a single request-level value (req.GetSeverity()), not per-group.
+	type labelGroup struct {
+		labels map[string]string
+		name   string
 	}
+	labelGroups := make([]labelGroup, 0, len(req.GetLabelGroups()))
+	for _, g := range req.GetLabelGroups() {
+		labelGroups = append(labelGroups, labelGroup{labels: g.GetJobLabels(), name: g.GetName()})
+	}
+	legacyMode := len(labelGroups) == 0
+	if legacyMode {
+		labelGroups = append(labelGroups, labelGroup{labels: req.GetJobLabels()})
+	}
+
 	// consider jobs with next schedule within next and before scheduleRangeInHours hours
 	scheduleRangeInHours := time.Duration(req.GetScheduledRangeInHours()) * time.Hour
 	referenceTime := time.Now().UTC()
@@ -546,16 +578,49 @@ func (h JobRunHandler) IdentifyPotentialSLABreach(ctx context.Context, req *pb.I
 		Severity:             req.GetSeverity(),
 		DamperCoeff:          float64(req.GetDamperCoeff()),
 	}
-	jobBreaches, err := h.jobSLAPredictorService.IdentifySLABreaches(ctx, projectName, jobNames, req.GetJobLabels(), reqConfig)
+
+	// build combos = projects x label groups.
+	combos := make([]scheduler.SLABreachCombo, 0, len(projectNames)*len(labelGroups))
+	for _, projectName := range projectNames {
+		for _, g := range labelGroups {
+			combo := scheduler.SLABreachCombo{
+				ProjectName: projectName,
+				Labels:      g.labels,
+				GroupName:   g.name,
+			}
+			if legacyMode && len(projectNames) == 1 {
+				combo.JobNames = jobNames
+			}
+			combos = append(combos, combo)
+		}
+	}
+
+	return combos, reqConfig, nil
+}
+
+// IdentifyPotentialSLABreach predicts potential SLA breaches for the given job targets at their targeted SLA time.
+// It accepts multiple projects and multiple label groups in a single request, evaluates the cross product, and
+// consolidates alerting into one message per team.
+func (h JobRunHandler) IdentifyPotentialSLABreach(ctx context.Context, req *pb.IdentifyPotentialSLABreachRequest) (*pb.IdentifyPotentialSLABreachResponse, error) {
+	response := pb.IdentifyPotentialSLABreachResponse{}
+
+	combos, reqConfig, err := h.buildIdentifySLABreachInputs(req)
+	if err != nil {
+		return nil, err
+	}
+
+	targetBreaches, err := h.jobSLAPredictorService.IdentifySLABreachesBatch(ctx, combos, reqConfig)
 	if err != nil {
 		h.l.Error("error identifying potential SLA breaches: %v", err)
 		return nil, errors.GRPCErr(err, "unable to identify potential SLA breaches")
 	}
 
 	jobs := make(map[string]*pb.UpstreamJobsStatus)
-	for jobNameTarget, upstreamJobStates := range jobBreaches {
+	// key is the project-qualified target identifier ("<project>/<job>"); breach
+	// holds that target's upstream cause jobs and their inferred SLA states.
+	for key, breach := range targetBreaches {
 		upstreamStatus := []*pb.UpstreamJobStatus{}
-		for upstreamJobName, upstreamJobState := range upstreamJobStates {
+		for upstreamJobName, upstreamJobState := range breach.Upstreams {
 			if upstreamJobState == nil || upstreamJobState.InferredSLA == nil {
 				continue
 			}
@@ -571,8 +636,10 @@ func (h JobRunHandler) IdentifyPotentialSLABreach(ctx context.Context, req *pb.I
 		if len(upstreamStatus) == 0 {
 			continue
 		}
-		jobs[jobNameTarget.String()] = &pb.UpstreamJobsStatus{
-			JobsStatus: upstreamStatus,
+		jobs[key] = &pb.UpstreamJobsStatus{
+			JobsStatus:        upstreamStatus,
+			TargetProjectName: breach.TargetProject,
+			TargetJobName:     breach.TargetJobName.String(),
 		}
 	}
 

core/scheduler/job_run.go

@@ -135,11 +135,140 @@ type UpstreamAttrs struct {
 	Status        string
 }
 
+// SLABreachTarget is a single SLA-bearing (target) job that may breach, along
+// with the upstream cause jobs owned by the alerted team.
+type SLABreachTarget struct {
+	JobName string
+	Causes  []UpstreamAttrs
+}
+
+// SLABreachGroup groups targets that share the same SLA target (label group),
+// carrying its own severity so different SLA buckets can be distinguished within
+// a single team's message.
+type SLABreachGroup struct {
+	Name     string
+	Severity string
+	Targets  []SLABreachTarget
+}
+
+// SLABreachProject groups the breaching targets by the target job's project.
+type SLABreachProject struct {
+	Name   string
+	Groups []SLABreachGroup
+}
+
+// PotentialSLABreachAttrs is a single consolidated alert for one team. It is
+// routed to the team owning the root-cause (upstream) jobs, and its body is
+// organized as project -> SLA group (with severity) -> target -> causes.
 type PotentialSLABreachAttrs struct {
-	ProjectName         string
-	TeamName            string
-	JobToUpstreamsCause map[string][]UpstreamAttrs
-	Severity            string
+	TeamName string
+	Projects []SLABreachProject
+}
+
+// SLABreachCombo is a single (project, label-group) unit of work. The batch
+// entrypoint evaluates the cross product of projects x label-groups as combos
+// and consolidates the results into one alert per team.
+type SLABreachCombo struct {
+	ProjectName tenant.ProjectName
+	JobNames    []JobName
+	Labels      map[string]string
+	GroupName   string // display name for the SLA group; derived from labels if empty
+}
+
+// TeamBreachAggregator accumulates breaches into a deterministic, insertion-
+// ordered team -> project -> group -> target -> causes structure and builds one
+// PotentialSLABreachAttrs per team.
+type TeamBreachAggregator struct {
+	teams     map[string]*teamAgg
+	teamOrder []string
+}
+
+type teamAgg struct {
+	name         string
+	projects     map[string]*projectAgg
+	projectOrder []string
+}
+
+type projectAgg struct {
+	name       string
+	groups     map[string]*groupAgg
+	groupOrder []string
+}
+
+type groupAgg struct {
+	name        string
+	severity    string
+	targets     map[string]*targetAgg
+	targetOrder []string
+}
+
+type targetAgg struct {
+	name       string
+	causes     map[string]UpstreamAttrs
+	causeOrder []string
+}
+
+func NewTeamBreachAggregator() *TeamBreachAggregator {
+	return &TeamBreachAggregator{teams: map[string]*teamAgg{}}
+}
+
+func (a *TeamBreachAggregator) Add(team, project, group, severity, target string, cause UpstreamAttrs) {
+	t, ok := a.teams[team]
+	if !ok {
+		t = &teamAgg{name: team, projects: map[string]*projectAgg{}}
+		a.teams[team] = t
+		a.teamOrder = append(a.teamOrder, team)
+	}
+	p, ok := t.projects[project]
+	if !ok {
+		p = &projectAgg{name: project, groups: map[string]*groupAgg{}}
+		t.projects[project] = p
+		t.projectOrder = append(t.projectOrder, project)
+	}
+	g, ok := p.groups[group]
+	if !ok {
+		g = &groupAgg{name: group, severity: severity, targets: map[string]*targetAgg{}}
+		p.groups[group] = g
+		p.groupOrder = append(p.groupOrder, group)
+	}
+	tg, ok := g.targets[target]
+	if !ok {
+		tg = &targetAgg{name: target, causes: map[string]UpstreamAttrs{}}
+		g.targets[target] = tg
+		g.targetOrder = append(g.targetOrder, target)
+	}
+	if _, ok := tg.causes[cause.JobName]; !ok {
+		tg.causes[cause.JobName] = cause
+		tg.causeOrder = append(tg.causeOrder, cause.JobName)
+	}
+}
+
+func (a *TeamBreachAggregator) Build() []*PotentialSLABreachAttrs {
+	out := make([]*PotentialSLABreachAttrs, 0, len(a.teamOrder))
+	for _, teamName := range a.teamOrder {
+		t := a.teams[teamName]
+		attr := &PotentialSLABreachAttrs{TeamName: t.name}
+		for _, projectName := range t.projectOrder {
+			p := t.projects[projectName]
+			project := SLABreachProject{Name: p.name}
+			for _, groupName := range p.groupOrder {
+				g := p.groups[groupName]
+				group := SLABreachGroup{Name: g.name, Severity: g.severity}
+				for _, targetName := range g.targetOrder {
+					tg := g.targets[targetName]
+					target := SLABreachTarget{JobName: tg.name}
+					for _, causeName := range tg.causeOrder {
+						target.Causes = append(target.Causes, tg.causes[causeName])
+					}
+					group.Targets = append(group.Targets, target)
+				}
+				project.Groups = append(project.Groups, group)
+			}
+			attr.Projects = append(attr.Projects, project)
+		}
+		out = append(out, attr)
+	}
+	return out
 }
 
 const (