fix: get job run from Optimus DB

core/scheduler/handler/v1beta1/job_run.go

@@ -11,12 +11,14 @@ import (
 	"github.com/goto/optimus/core/tenant"
 	"github.com/goto/optimus/internal/errors"
 	"github.com/goto/optimus/internal/lib/interval"
+	"github.com/goto/optimus/internal/utils/filter"
 	pb "github.com/goto/optimus/protos/gotocompany/optimus/core/v1beta1"
 )
 
 type JobRunService interface {
 	JobRunInput(context.Context, tenant.ProjectName, scheduler.JobName, scheduler.RunConfig) (*scheduler.ExecutorInput, error)
 	UpdateJobState(context.Context, *scheduler.Event) error
+	GetJobRunsByFilter(ctx context.Context, projectName tenant.ProjectName, jobName scheduler.JobName, filters ...filter.FilterOpt) ([]*scheduler.JobRun, error)
 	GetJobRuns(ctx context.Context, projectName tenant.ProjectName, jobName scheduler.JobName, criteria *scheduler.JobRunsCriteria) ([]*scheduler.JobRunStatus, error)
 	UploadToScheduler(ctx context.Context, projectName tenant.ProjectName) error
 	GetInterval(ctx context.Context, projectName tenant.ProjectName, jobName scheduler.JobName, referenceTime time.Time) (interval.Interval, error)
@@ -80,8 +82,51 @@ func (h JobRunHandler) JobRunInput(ctx context.Context, req *pb.JobRunInputReque
 	}, nil
 }
 
-// JobRun currently gets the job runs from scheduler based on the criteria
-// TODO: later should collect the job runs from optimus
+// GetJobRun gets job runs from optimus DB based on the criteria
+func (h JobRunHandler) GetJobRun(ctx context.Context, req *pb.GetJobRunsRequest) (*pb.GetJobRunsResponse, error) {
+	projectName, err := tenant.ProjectNameFrom(req.GetProjectName())
+	if err != nil {
+		h.l.Error("error adapting project name [%s]: %s", req.GetProjectName(), err)
+		return nil, errors.GRPCErr(err, "unable to get job run for "+req.GetJobName())
+	}
+
+	jobName, err := scheduler.JobNameFrom(req.GetJobName())
+	if err != nil {
+		h.l.Error("error adapting job name [%s]: %s", req.GetJobName(), err)
+		return nil, errors.GRPCErr(err, "unable to get job run for "+req.GetJobName())
+	}
+
+	if len(req.GetState()) > 0 {
+		_, err := scheduler.StateFromString(req.GetState())
+		if err != nil {
+			h.l.Error("error adapting job run state [%s]: %s", req.GetState(), err)
+			return nil, errors.GRPCErr(err, "invalid job run state: "+req.GetSince().String())
+		}
+	}
+
+	var jobRuns []*scheduler.JobRun
+	jobRuns, err = h.service.GetJobRunsByFilter(ctx, projectName, jobName,
+		filter.WithString(filter.RunState, req.GetState()),
+		filter.WithTime(filter.StartDate, req.GetSince().AsTime()),
+		filter.WithTime(filter.EndDate, req.GetUntil().AsTime()),
+	)
+	if err != nil {
+		h.l.Error("error getting job runs: %s", err)
+		return nil, errors.GRPCErr(err, "unable to get job run for "+req.GetJobName())
+	}
+
+	var runs []*pb.JobRun
+	for _, run := range jobRuns {
+		ts := timestamppb.New(run.ScheduledAt)
+		runs = append(runs, &pb.JobRun{
+			State:       run.State.String(),
+			ScheduledAt: ts,
+		})
+	}
+	return &pb.GetJobRunsResponse{JobRuns: runs}, nil
+}
+
+// JobRun gets the job runs from scheduler based on the criteria
 func (h JobRunHandler) JobRun(ctx context.Context, req *pb.JobRunRequest) (*pb.JobRunResponse, error) {
 	projectName, err := tenant.ProjectNameFrom(req.GetProjectName())
 	if err != nil {

core/scheduler/handler/v1beta1/job_run_test.go

@@ -18,6 +18,7 @@ import (
 	"github.com/goto/optimus/core/tenant"
 	"github.com/goto/optimus/internal/lib/interval"
 	"github.com/goto/optimus/internal/lib/window"
+	"github.com/goto/optimus/internal/utils/filter"
 	pb "github.com/goto/optimus/protos/gotocompany/optimus/core/v1beta1"
 )
 
@@ -716,6 +717,11 @@ func (m *mockJobRunService) UploadToScheduler(ctx context.Context, projectName t
 	return args.Error(0)
 }
 
+func (m *mockJobRunService) GetJobRunsByFilter(ctx context.Context, projectName tenant.ProjectName, jobName scheduler.JobName, filters ...filter.FilterOpt) ([]*scheduler.JobRun, error) {
+	args := m.Called(ctx, projectName, jobName, filters)
+	return args.Get(0).([]*scheduler.JobRun), args.Error(1)
+}
+
 func (m *mockJobRunService) GetJobRuns(ctx context.Context, projectName tenant.ProjectName, jobName scheduler.JobName, criteria *scheduler.JobRunsCriteria) ([]*scheduler.JobRunStatus, error) {
 	args := m.Called(ctx, projectName, jobName, criteria)
 	if args.Get(0) == nil {

core/scheduler/service/job_run_service.go

@@ -22,6 +22,7 @@ import (
 	"github.com/goto/optimus/internal/lib/interval"
 	"github.com/goto/optimus/internal/lib/window"
 	"github.com/goto/optimus/internal/models"
+	"github.com/goto/optimus/internal/utils/filter"
 )
 
 type metricType string
@@ -54,6 +55,8 @@ type JobRepository interface {
 type JobRunRepository interface {
 	GetByID(ctx context.Context, id scheduler.JobRunID) (*scheduler.JobRun, error)
 	GetByScheduledAt(ctx context.Context, tenant tenant.Tenant, name scheduler.JobName, scheduledAt time.Time) (*scheduler.JobRun, error)
+	GetLatestRun(ctx context.Context, project tenant.ProjectName, name scheduler.JobName, status *scheduler.State) (*scheduler.JobRun, error)
+	GetRunsByTimeRange(ctx context.Context, project tenant.ProjectName, jobName scheduler.JobName, status *scheduler.State, since, until time.Time) ([]*scheduler.JobRun, error)
 	GetByScheduledTimes(ctx context.Context, tenant tenant.Tenant, jobName scheduler.JobName, scheduledTimes []time.Time) ([]*scheduler.JobRun, error)
 	Create(ctx context.Context, tenant tenant.Tenant, name scheduler.JobName, scheduledAt time.Time, slaDefinitionInSec int64) error
 	Update(ctx context.Context, jobRunID uuid.UUID, endTime time.Time, jobRunStatus scheduler.State) error
@@ -146,6 +149,27 @@ func (s *JobRunService) JobRunInput(ctx context.Context, projectName tenant.Proj
 	return s.compiler.Compile(ctx, details, config, executedAt)
 }
 
+func (s *JobRunService) GetJobRunsByFilter(ctx context.Context, projectName tenant.ProjectName, jobName scheduler.JobName, filters ...filter.FilterOpt) ([]*scheduler.JobRun, error) {
+	f := filter.NewFilter(filters...)
+	var runState *scheduler.State
+	state, err := scheduler.StateFromString(f.GetStringValue(filter.RunState))
+	if err == nil {
+		runState = &state
+	}
+
+	if f.Contains(filter.StartDate) && f.Contains(filter.EndDate) {
+		//	get job run by scheduled at between start date and end date, filter by runState if applicable
+		return s.repo.GetRunsByTimeRange(ctx, projectName, jobName, runState,
+			f.GetTimeValue(filter.StartDate), f.GetTimeValue(filter.EndDate))
+	}
+
+	jobRun, err := s.repo.GetLatestRun(ctx, projectName, jobName, runState)
+	if err != nil {
+		return nil, err
+	}
+	return []*scheduler.JobRun{jobRun}, nil
+}
+
 func (s *JobRunService) GetJobRuns(ctx context.Context, projectName tenant.ProjectName, jobName scheduler.JobName, criteria *scheduler.JobRunsCriteria) ([]*scheduler.JobRunStatus, error) {
 	jobWithDetails, err := s.jobRepo.GetJobDetails(ctx, projectName, jobName)
 	if err != nil {

core/scheduler/service/job_run_service_test.go

@@ -1522,6 +1522,16 @@ func (m *mockJobRunRepository) Create(ctx context.Context, tenant tenant.Tenant,
 	return args.Error(0)
 }
 
+func (m *mockJobRunRepository) GetLatestRun(ctx context.Context, project tenant.ProjectName, jobName scheduler.JobName, runState *scheduler.State) (*scheduler.JobRun, error) {
+	args := m.Called(ctx, project, jobName, runState)
+	return args.Get(0).(*scheduler.JobRun), args.Error(1)
+}
+
+func (m *mockJobRunRepository) GetRunsByTimeRange(ctx context.Context, project tenant.ProjectName, jobName scheduler.JobName, runState *scheduler.State, since, until time.Time) ([]*scheduler.JobRun, error) {
+	args := m.Called(ctx, project, jobName, runState, since, until)
+	return args.Get(0).([]*scheduler.JobRun), args.Error(1)
+}
+
 func (m *mockJobRunRepository) Update(ctx context.Context, jobRunID uuid.UUID, endTime time.Time, jobRunStatus scheduler.State) error {
 	args := m.Called(ctx, jobRunID, endTime, jobRunStatus)
 	return args.Error(0)

internal/store/postgres/scheduler/job_run_repository.go

@@ -3,6 +3,7 @@ package scheduler
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"strings"
 	"time"
 
@@ -90,6 +91,56 @@ func (j *JobRunRepository) GetByID(ctx context.Context, id scheduler.JobRunID) (
 	return jr.toJobRun()
 }
 
+func (j *JobRunRepository) GetLatestRun(ctx context.Context, project tenant.ProjectName, jobName scheduler.JobName, runState *scheduler.State) (*scheduler.JobRun, error) {
+	var jr jobRun
+	var stateClause string
+	if runState != nil {
+		stateClause = fmt.Sprintf(" and status = '%s'", runState)
+	}
+	getLatestRun := fmt.Sprintf("SELECT %s, created_at FROM job_run j where project_name = $1 and job_name = $2 %s order by scheduled_at desc limit 1", jobRunColumns, stateClause)
+	err := j.db.QueryRow(ctx, getLatestRun, project, jobName).
+		Scan(&jr.ID, &jr.JobName, &jr.NamespaceName, &jr.ProjectName, &jr.ScheduledAt, &jr.StartTime, &jr.EndTime,
+			&jr.Status, &jr.SLADefinition, &jr.SLAAlert, &jr.Monitoring, &jr.CreatedAt)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, errors.NotFound(scheduler.EntityJobRun, "no record for job:"+jobName.String()+stateClause)
+		}
+		return nil, errors.Wrap(scheduler.EntityJobRun, "error while getting run", err)
+	}
+	return jr.toJobRun()
+}
+
+func (j *JobRunRepository) GetRunsByTimeRange(ctx context.Context, project tenant.ProjectName, jobName scheduler.JobName, runState *scheduler.State, since, until time.Time) ([]*scheduler.JobRun, error) {
+	var jobRunList []*scheduler.JobRun
+	var stateClause string
+	if runState != nil {
+		stateClause = fmt.Sprintf(" and status = '%s'", runState)
+	}
+	getLatestRun := fmt.Sprintf("SELECT %s, created_at FROM job_run j where project_name = $1 and job_name = $2 and scheduled_at >= $3 and scheduled_at <= $4 %s", jobRunColumns, stateClause)
+	rows, err := j.db.Query(ctx, getLatestRun, project, jobName, since, until)
+	if err != nil {
+		return nil, errors.Wrap(scheduler.EntityJobRun, "error while getting job runs", err)
+	}
+	for rows.Next() {
+		var jr jobRun
+		err := rows.Scan(&jr.ID, &jr.JobName, &jr.NamespaceName, &jr.ProjectName, &jr.ScheduledAt, &jr.StartTime, &jr.EndTime,
+			&jr.Status, &jr.SLADefinition, &jr.SLAAlert, &jr.Monitoring, &jr.CreatedAt)
+		if err != nil {
+			if errors.Is(err, pgx.ErrNoRows) {
+				errMsg := fmt.Sprintf("no record for job: %s, scheduled between : %s -> %s, %s", jobName, since, until, stateClause)
+				return nil, errors.NotFound(scheduler.EntityJobRun, errMsg)
+			}
+			return nil, errors.Wrap(scheduler.EntityJobRun, "error while getting run", err)
+		}
+		jobRun, err := jr.toJobRun()
+		if err != nil {
+			return nil, errors.Wrap(scheduler.EntityJobRun, "error while getting job runs", err)
+		}
+		jobRunList = append(jobRunList, jobRun)
+	}
+	return jobRunList, nil
+}
+
 func (j *JobRunRepository) GetByScheduledAt(ctx context.Context, t tenant.Tenant, jobName scheduler.JobName, scheduledAt time.Time) (*scheduler.JobRun, error) {
 	var jr jobRun
 	// todo: check if `order by created_at desc limit 1` is required

internal/utils/filter/filter_opt.go

@@ -17,6 +17,9 @@ const (
 	bitOnReplayStatus        uint64 = 1 << 6
 	bitOnScheduledAt         uint64 = 1 << 7
 	bitOnReplayID            uint64 = 1 << 8
+	bitOnRunState            uint64 = 1 << 9
+	bitOnStartDate           uint64 = 1 << 10
+	bitOnEndDate             uint64 = 1 << 11
 )
 
 const (
@@ -29,6 +32,9 @@ const (
 	ReplayStatus        = Operand(bitOnReplayStatus)
 	ScheduledAt         = Operand(bitOnScheduledAt)
 	ReplayID            = Operand(bitOnReplayID)
+	RunState            = Operand(bitOnRunState)
+	StartDate           = Operand(bitOnStartDate)
+	EndDate             = Operand(bitOnEndDate)
 )
 
 func WithTime(operand Operand, value time.Time) FilterOpt {