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 []service.SLABreachCombo, reqConfig service.JobSLAPredictorRequestConfig) (map[string]*service.TargetBreach, error)
 }
 
 type JobExpectatorService interface {
@@ -512,10 +513,32 @@ 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
+// 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{}
 
+	// 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, 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, 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)
@@ -526,11 +549,21 @@ func (h JobRunHandler) IdentifyPotentialSLABreach(ctx context.Context, req *pb.I
 		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 +579,34 @@ 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. job_names only applies in legacy
+	// single-project mode where target selection isn't ambiguous.
+	combos := make([]service.SLABreachCombo, 0, len(projectNames)*len(labelGroups))
+	for _, projectName := range projectNames {
+		for _, g := range labelGroups {
+			combo := service.SLABreachCombo{
+				ProjectName: projectName,
+				Labels:      g.labels,
+				GroupName:   g.name,
+			}
+			if legacyMode && len(projectNames) == 1 {
+				combo.JobNames = jobNames
+			}
+			combos = append(combos, combo)
+		}
+	}
+
+	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 {
+	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 +622,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,34 @@ 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
 }
 
 const (

core/scheduler/service/job_sla_predictor_aggregator_test.go

@@ -0,0 +1,59 @@
+package service // nolint: testpackage
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/goto/optimus/core/scheduler"
+)
+
+func TestTeamBreachAggregator(t *testing.T) {
+	t.Run("consolidates per team -> project -> group -> target and dedupes causes", func(t *testing.T) {
+		agg := newTeamBreachAggregator()
+
+		// team-a: p1/g1(CRITICAL)/t1 with cause c1 (added twice -> deduped) and c2
+		agg.add("team-a", "p1", "g1", "CRITICAL", "t1", scheduler.UpstreamAttrs{JobName: "c1", RelativeLevel: 2, Status: "NOT_STARTED"})
+		agg.add("team-a", "p1", "g1", "CRITICAL", "t1", scheduler.UpstreamAttrs{JobName: "c1", RelativeLevel: 2, Status: "NOT_STARTED"})
+		agg.add("team-a", "p1", "g1", "CRITICAL", "t1", scheduler.UpstreamAttrs{JobName: "c2", RelativeLevel: 1, Status: "RUNNING_LATE"})
+		// team-a: second group in same project
+		agg.add("team-a", "p1", "g2", "WARNING", "t2", scheduler.UpstreamAttrs{JobName: "c3"})
+		// team-a: second project
+		agg.add("team-a", "p2", "g1", "CRITICAL", "t3", scheduler.UpstreamAttrs{JobName: "c4"})
+		// team-b: separate team
+		agg.add("team-b", "p1", "g1", "CRITICAL", "t1", scheduler.UpstreamAttrs{JobName: "c5"})
+
+		out := agg.build()
+		assert.Len(t, out, 2)
+
+		// insertion order is preserved: team-a first
+		teamA := out[0]
+		assert.Equal(t, "team-a", teamA.TeamName)
+		assert.Len(t, teamA.Projects, 2)
+
+		p1 := teamA.Projects[0]
+		assert.Equal(t, "p1", p1.Name)
+		assert.Len(t, p1.Groups, 2)
+
+		g1 := p1.Groups[0]
+		assert.Equal(t, "g1", g1.Name)
+		assert.Equal(t, "CRITICAL", g1.Severity)
+		assert.Len(t, g1.Targets, 1)
+		assert.Equal(t, "t1", g1.Targets[0].JobName)
+		assert.Len(t, g1.Targets[0].Causes, 2) // c1 deduped, c1 + c2
+
+		g2 := p1.Groups[1]
+		assert.Equal(t, "g2", g2.Name)
+		assert.Equal(t, "WARNING", g2.Severity)
+
+		assert.Equal(t, "p2", teamA.Projects[1].Name)
+		assert.Equal(t, "team-b", out[1].TeamName)
+	})
+}
+
+func TestDeriveGroupName(t *testing.T) {
+	assert.Equal(t, "default", deriveGroupName(nil))
+	assert.Equal(t, "default", deriveGroupName(map[string]string{}))
+	// keys are sorted for stability
+	assert.Equal(t, "a=1, b=2", deriveGroupName(map[string]string{"b": "2", "a": "1"}))
+}

core/scheduler/service/job_sla_predictor_batch_test.go

@@ -0,0 +1,141 @@
+package service_test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/goto/salt/log"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+
+	"github.com/goto/optimus/config"
+	"github.com/goto/optimus/core/scheduler"
+	"github.com/goto/optimus/core/scheduler/service"
+	"github.com/goto/optimus/core/tenant"
+)
+
+func TestIdentifySLABreachesBatch(t *testing.T) {
+	ctx := context.Background()
+	l := log.NewNoop()
+	conf := config.PotentialSLABreachConfig{
+		DamperCoeff:             1.0,
+		EnablePersistentLogging: false,
+	}
+
+	scheduledChangeGetter := NewScheduledChangeGetter(t)
+	scheduledChangeGetter.On("GetRecentScheduleChange", ctx, mock.Anything, mock.Anything, mock.Anything).Return("", nil).Maybe()
+
+	t.Run("returns project-qualified breaches for a combo", func(t *testing.T) {
+		// job-C -> job-B -> job-A (target). job-C is running late so job-A may breach.
+		jobLineageFetcher := NewJobLineageFetcher(t)
+		durationEstimator := NewDurationEstimator(t)
+		jobDetailsGetter := NewJobDetailsGetter(t)
+		svc := service.NewJobSLAPredictorService(l, conf, nil, jobLineageFetcher, durationEstimator, jobDetailsGetter, nil, nil, scheduledChangeGetter)
+
+		projectName := tenant.ProjectName("project-a")
+		referenceTime := time.Now().UTC()
+		reqConfig := service.JobSLAPredictorRequestConfig{
+			ReferenceTime:        referenceTime,
+			ScheduleRangeInHours: 10 * time.Hour,
+			EnableAlert:          false,
+			DamperCoeff:          conf.DamperCoeff,
+		}
+
+		tnnt, _ := tenant.NewTenant("project-a", "team-a")
+		startDate := referenceTime.Add(-24 * time.Hour).Truncate(time.Hour)
+		scheduledAt := referenceTime.Add(1 * time.Minute).Truncate(time.Minute)
+		interval := fmt.Sprintf("%d %d * * *", scheduledAt.Minute(), scheduledAt.Hour())
+		jobA := &scheduler.JobWithDetails{
+			Name: "job-A",
+			Job:  &scheduler.Job{Tenant: tnnt, Name: "job-A"},
+			Schedule: &scheduler.Schedule{
+				StartDate: startDate,
+				Interval:  interval,
+			},
+			Alerts: []scheduler.Alert{
+				{On: scheduler.EventCategorySLAMiss, Config: map[string]string{"duration": "30m"}},
+			},
+		}
+
+		jobASchedule := &scheduler.JobSchedule{JobName: "job-A", ScheduledAt: scheduledAt}
+		jobALineage := &scheduler.JobLineageSummary{
+			JobName:          "job-A",
+			ScheduleInterval: interval,
+			IsEnabled:        true,
+			JobRuns:          map[scheduler.JobName]*scheduler.JobRunSummary{"job-A": {ScheduledAt: scheduledAt}},
+		}
+		jobBLineage := &scheduler.JobLineageSummary{
+			JobName:          "job-B",
+			ScheduleInterval: interval,
+			IsEnabled:        true,
+			JobRuns:          map[scheduler.JobName]*scheduler.JobRunSummary{"job-A": {ScheduledAt: scheduledAt.Add(-15 * time.Minute)}},
+		}
+		jobCTaskStartTime := scheduledAt.Add(-20 * time.Minute)
+		jobCLineage := &scheduler.JobLineageSummary{
+			JobName:          "job-C",
+			ScheduleInterval: interval,
+			IsEnabled:        true,
+			JobRuns: map[scheduler.JobName]*scheduler.JobRunSummary{
+				"job-A": {ScheduledAt: scheduledAt.Add(-25 * time.Minute), TaskStartTime: &jobCTaskStartTime},
+			},
+		}
+		jobALineage.Upstreams = []*scheduler.JobLineageSummary{jobBLineage}
+		jobBLineage.Upstreams = []*scheduler.JobLineageSummary{jobCLineage}
+
+		jobDetailsGetter.On("GetJobs", ctx, projectName, []string{"job-A"}).Return([]*scheduler.JobWithDetails{jobA}, nil).Once()
+		jobLineageFetcher.On("GetJobLineage", ctx, map[scheduler.JobName]*scheduler.JobSchedule{"job-A": jobASchedule}).
+			Return(map[scheduler.JobName]*scheduler.JobLineageSummary{"job-A": jobALineage}, nil).Once()
+		durationEstimator.On("GetPercentileDurationByJobNames", ctx, referenceTime, mock.MatchedBy(func(jobNames []scheduler.JobName) bool {
+			return assert.ElementsMatch(t, []scheduler.JobName{"job-A", "job-B", "job-C"}, jobNames)
+		})).Return(map[scheduler.JobName]*time.Duration{
+			"job-A": durationPtr(20 * time.Minute),
+			"job-B": durationPtr(15 * time.Minute),
+			"job-C": durationPtr(10 * time.Minute),
+		}, nil).Once()
+
+		combos := []service.SLABreachCombo{
+			{ProjectName: projectName, JobNames: []scheduler.JobName{"job-A"}, GroupName: "sla-8am"},
+		}
+
+		// when
+		res, err := svc.IdentifySLABreachesBatch(ctx, combos, reqConfig)
+
+		// then
+		assert.NoError(t, err)
+		assert.Len(t, res, 1)
+		tb, ok := res["project-a/job-A"]
+		assert.True(t, ok)
+		assert.Equal(t, "project-a", tb.TargetProject)
+		assert.Equal(t, scheduler.JobName("job-A"), tb.TargetJobName)
+		assert.Contains(t, tb.Upstreams, scheduler.JobName("job-C"))
+	})
+
+	t.Run("returns error only when every combo fails", func(t *testing.T) {
+		jobLineageFetcher := NewJobLineageFetcher(t)
+		durationEstimator := NewDurationEstimator(t)
+		jobDetailsGetter := NewJobDetailsGetter(t)
+		svc := service.NewJobSLAPredictorService(l, conf, nil, jobLineageFetcher, durationEstimator, jobDetailsGetter, nil, nil, scheduledChangeGetter)
+
+		projectName := tenant.ProjectName("project-a")
+		labels := map[string]string{"criticality": "critical"}
+		reqConfig := service.JobSLAPredictorRequestConfig{
+			ReferenceTime:        time.Now().UTC(),
+			ScheduleRangeInHours: 10 * time.Hour,
+			EnableAlert:          false,
+			DamperCoeff:          conf.DamperCoeff,
+		}
+
+		jobDetailsGetter.On("GetJobsByLabels", ctx, projectName, labels).Return(nil, assert.AnError).Once()
+
+		combos := []service.SLABreachCombo{{ProjectName: projectName, Labels: labels}}
+
+		res, err := svc.IdentifySLABreachesBatch(ctx, combos, reqConfig)
+
+		assert.Error(t, err)
+		assert.Len(t, res, 0)
+	})
+}
+
+func durationPtr(d time.Duration) *time.Duration { return &d }