-
Notifications
You must be signed in to change notification settings - Fork 4
/
scheduler.go
254 lines (204 loc) · 8.77 KB
/
scheduler.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
/*
Copyright 2021-2023 ICS-FORTH.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package scheduler
import (
"fmt"
"time"
"github.com/carv-ics-forth/frisbee/api/v1alpha1"
"github.com/carv-ics-forth/frisbee/pkg/expressions"
"github.com/carv-ics-forth/frisbee/pkg/lifecycle"
"github.com/go-logr/logr"
"github.com/pkg/errors"
"github.com/robfig/cron/v3"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"sigs.k8s.io/controller-runtime/pkg/client"
)
type Parameters struct {
// State is the real state of the system.
State lifecycle.Classifier
// LastScheduleTime is the time the controller last scheduled an object.
LastScheduleTime metav1.Time
// ScheduleSpec is the scheduling options
ScheduleSpec *v1alpha1.TaskSchedulerSpec
//
// Parameters Used for Timeline mode
//
// ExpectedTime is the evaluation of a timeline distribution defined in the ScheduleSpec.
ExpectedTimeline v1alpha1.Timeline
//
// Parameters Used for Sequential mode
//
// JobName is used a prefix for finding children tasks.
// We assume the standard naming pattern Job-Task.
JobName string
// ScheduledJobs normally points to the next QueuedJobs.
// In this case, we use it in conjunction with JobName to find the children's name,
// and from that to extract the status of the child.
ScheduledJobs int `json:"scheduledJobs,omitempty"`
}
// Schedule calculate the next scheduled run, and whether we've got a run that we haven't processed yet (or anything we missed).
// If we've missed a run, and we're still within the deadline to start it, we'll need to run a job.
// time-based and event-driven scheduling can be used in conjunction.
func Schedule(log logr.Logger, obj client.Object, params Parameters) (goToNextJob bool, nextTick time.Time, err error) {
// no scheduling constraint.
if params.ScheduleSpec == nil {
// no constraints. start everything as fast as possible.
return true, time.Time{}, nil
}
// logrus.Warn("Scheduling Info", params.State.ListAll())
// Sequential scheduling
if params.ScheduleSpec.Sequential != nil {
// if nothing is running, start a new job
if params.ScheduledJobs == -1 {
return true, time.Time{}, nil
}
// if a job is running, make sure that it is complete, where complete means
// that it is either successful or failed.
lastJob := fmt.Sprintf("%s-%d", params.JobName, params.ScheduledJobs+1)
if params.State.IsSuccessful(lastJob) || params.State.IsFailed(lastJob) {
return true, time.Time{}, nil
}
// otherwise, do not schedule anything, but requeue the request
return false, time.Time{}, nil
}
// Cron-based scheduling
if params.ScheduleSpec.Cron != nil {
missed, cronTick, err := cronWithDeadline(log, obj, params)
return !missed.IsZero(), cronTick, err
}
// Timeline-based scheduling
if params.ScheduleSpec.Timeline != nil {
missed, fixedTick, err := timelineWithDeadline(log, obj, params)
return !missed.IsZero(), fixedTick, err
}
// Event-based scheduling
if !params.ScheduleSpec.Event.IsZero() {
eval := expressions.Condition{Expr: params.ScheduleSpec.Event}
return eval.IsTrue(¶ms.State, obj), time.Time{}, nil
}
panic("this should never happen")
}
func cronWithDeadline(_ logr.Logger, obj client.Object, params Parameters) (lastMissed time.Time, next time.Time, err error) {
timeline, err := cron.ParseStandard(*params.ScheduleSpec.Cron)
if err != nil {
return time.Time{}, time.Time{}, errors.Wrapf(err, "unparseable timeline %q", *params.ScheduleSpec.Cron)
}
lastMissed, next, err = getNextScheduleTime(obj.GetCreationTimestamp().Time, timeline, params)
if err != nil {
return lastMissed, next, errors.Wrapf(err, "scheduling error")
}
/*
deadline := params.ScheduleSpec.StartingDeadlineSeconds
if !lastMissed.IsZero() && !honorDeadline(log, lastMissed, deadline) {
return lastMissed, next, errors.Errorf("scheduling violation. deadline of '%d' seconds is too strict.", *deadline)
}
*/
return lastMissed, next, nil
}
func timelineWithDeadline(_ logr.Logger, obj client.Object, params Parameters) (lastMissed time.Time, next time.Time, err error) {
timeline := params.ExpectedTimeline
lastMissed, next, err = getNextScheduleTime(obj.GetCreationTimestamp().Time, timeline, params)
if err != nil {
return lastMissed, next, errors.Wrapf(err, "timeline error")
}
/*
deadline := params.ScheduleSpec.StartingDeadlineSeconds
if !lastMissed.IsZero() && !honorDeadline(log, lastMissed, deadline) {
return lastMissed, next, errors.Errorf("scheduling violation. deadline of '%d' seconds is too strict.", *deadline)
}
*/
return lastMissed, next, nil
}
// Timeline describes a job's duty cycle.
type Timeline interface {
// Next returns the next activation time, later than the given time.
// Next is invoked initially, and then each time the job is run.
Next(time.Time) time.Time
}
// getNextScheduleTime figure out the next times that we need to create jobs at (or anything we missed).
//
// We'll start calculating appropriate times from our last run, or the creation
// of the CronJob if we can't find a last run. This gets the time of next schedule
// after last scheduled and before now.
//
// If there are too many missed runs, and we don't have any deadlines set, we'll
// bail so that we don't cause issues on controller restarts or wedges.
// Otherwise, we'll just return the missed runs (of which we'll just use the latest),
// and the next run, so that we can know when it's time to reconcile again.
func getNextScheduleTime(earliest time.Time, timeline Timeline, params Parameters) (lastMissed time.Time, next time.Time, err error) {
now := time.Now()
var earliestTime time.Time
if params.LastScheduleTime.IsZero() {
// If none found, then this is either a recently created cronJob,
// or the active/completed info was somehow lost (contract for status
// in kubernetes says it may need to be recreated), or that we have
// started a job, but have not noticed it yet (distributed systems can
// have arbitrary delays). In any case, use the creation time of the
// object as last known start time.
earliestTime = earliest
} else {
// for optimization purposes, cheat a bit and start from our last observed run time
// we could reconstitute this here, but there's not much point, since we've
// just updated it.
earliestTime = params.LastScheduleTime.Time
}
if params.ScheduleSpec.StartingDeadlineSeconds != nil {
// controller is not going to schedule anything below this point
schedulingDeadline := now.Add(-time.Second * time.Duration(*params.ScheduleSpec.StartingDeadlineSeconds))
if schedulingDeadline.After(earliestTime) {
earliestTime = schedulingDeadline
}
}
if earliestTime.After(now) {
// the earliest time is later than now.
// return the next activation time (used for re-queuing the request)
return time.Time{}, timeline.Next(now), nil
}
starts := 0
for t := timeline.Next(earliestTime); !t.After(now); t = timeline.Next(t) {
lastMissed = t
// An object might miss several starts. For example, if
// controller gets wedged on Friday at 5:01pm when everyone has
// gone home, and someone comes in on Tuesday AM and discovers
// the problem and restarts the controller, then all the hourly
// jobs, more than 80 of them for one hourly scheduledJob, should
// all start running with no further intervention (if the scheduledJob
// allows concurrency and late starts).
//
// However, if there is a bug somewhere, or incorrect clock
// on controller's server or apiservers (for setting creationTimestamp)
// then there could be so many missed start times (it could be off
// by decades or more), that it would eat up all the CPU and memory
// of this controller. In that case, we want to not try to list
// all the missed start times.
starts++
if starts > 100 {
// We can't get the most recent times so just return an empty slice
return time.Time{}, time.Time{},
errors.New("too many missed start times (> 100). Set or decrease .spec.startingDeadlineSeconds or check clock skew")
}
}
return lastMissed, timeline.Next(now), nil
}
/*
func honorDeadline(log logr.Logger, lastMissed time.Time, deadline *int64) bool {
// if there is a missed run, make sure we're not too late to start the run
tooLate := false
if deadline != nil {
skew := lastMissed.Add(time.Duration(*deadline) * time.Second)
log.Info("MissedSchedule", "skew", skew)
tooLate = skew.Before(time.Now())
}
return tooLate
}
*/