Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Kubernetes CronJob Implementation in Detail #k8sjp

Shimpei Otsubo
September 28, 2018

Kubernetes CronJob Implementation in Detail #k8sjp

Shimpei Otsubo

September 28, 2018
Tweet

More Decks by Shimpei Otsubo

Other Decks in Programming

Transcript

  1. ©2018 Wantedly, Inc. CronJob Implementation in Detail ίʔυ͔ΒಡΈղ͘ CronJob ͷৄࡉ࢓༷

    Kubernetes Meetup Tokyo #13 28.Sep.2018 - Shimpei Otsubo - @potsbo
  2. ຊ൪Ϋϥελͷ$SPO+PCͷ஗ԆΛܭଌͯ͠Έͨ ඵ͸༨༟ඵ΋ฏؾͰ஗ΕΔ n = 475 $SPO+PC஗ԆώετάϥϜ 0 30 60 ஗Ԇ

    ඵ <5 <10 <15 <20 <25 <30 <35 <40 <45 <50 <55 <60 <65 <70 <75 <80 ݸ
 ਺ ˞4UBSUJOH%FBEMJOF4FDPOET͕ͷ΋ͷ ©2018 Wantedly, Inc.
  3. // Run the main goroutine responsible for watching and syncing

    jobs. func (jm *CronJobController) Run(stopCh <-chan struct{}) { defer utilruntime.HandleCrash() glog.Infof("Starting CronJob Manager") // Check things every 10 second. go wait.Until(jm.syncAll, 10*time.Second, stopCh) <-stopCh glog.Infof("Shutting down CronJob Manager") } $SPO+PC$POUSPMMFS͕ಈ͖࢝ΊΔGVOD ©2018 Wantedly, Inc.
  4. // Run the main goroutine responsible for watching and syncing

    jobs. func (jm *CronJobController) Run(stopCh <-chan struct{}) { defer utilruntime.HandleCrash() glog.Infof("Starting CronJob Manager") // Check things every 10 second. go wait.Until(jm.syncAll, 10*time.Second, stopCh) <-stopCh glog.Infof("Shutting down CronJob Manager") } FWFSZTFDPOEͱॻ͍ͯ͋Δ͕࣮͸શ෦ॲཧඵεϦʔϓ $SPO+PC͕ଟ͍ͱ෼୯ҐͰ஗ΕΔ͜ͱ΋͋Γͦ͏ ˞ҰճͷTZOD"MMʹຊ౰ʹͦΜͳʹ͕͔͔͍࣌ؒͬͯΔͷ͔͸ະܭଌ ©2018 Wantedly, Inc.
  5. func syncOne(sj *batchv1beta1.CronJob, js []batchv1.Job, now time.Time, jc jobControlInterface, sjc

    sjControlInterface, pc podControlInterface, recorder record.EventRecorder) { // 50+ lines omitted times, err := getRecentUnmetScheduleTimes(*sj, now) if err != nil { return } // 10+ lines omitted scheduledTime := times[len(times)-1] tooLate := false if sj.Spec.StartingDeadlineSeconds != nil { tooLate = scheduledTime.Add(time.Second * time.Duration(*sj.Spec.StartingDeadlineSeconds)).Before(now) } if tooLate { return } // 30 lines omitted jobReq, err := getJobFromTemplate(sj, scheduledTime) if err != nil { return } jobResp, err := jc.CreateJob(sj.Namespace, jobReq) if err != nil { return } // 20+ lines omitted return } ˞దٓ؆ུԽ
  6. func syncOne(sj *batchv1beta1.CronJob, js []batchv1.Job, now time.Time, jc jobControlInterface, sjc

    sjControlInterface, pc podControlInterface, recorder record.EventRecorder) { // 50+ lines omitted times, err := getRecentUnmetScheduleTimes(*sj, now) if err != nil { return } // 10+ lines omitted scheduledTime := times[len(times)-1] tooLate := false if sj.Spec.StartingDeadlineSeconds != nil { tooLate = scheduledTime.Add(time.Second * time.Duration(*sj.Spec.StartingDeadlineSeconds)).Before(now) } if tooLate { return } // 30 lines omitted jobReq, err := getJobFromTemplate(sj, scheduledTime) if err != nil { return } jobResp, err := jc.CreateJob(sj.Namespace, jobReq) if err != nil { return } // 20+ lines omitted return } ະ࣮ߦ࣌ࠁҰཡΛऔಘ ˞దٓ؆ུԽ
  7. func syncOne(sj *batchv1beta1.CronJob, js []batchv1.Job, now time.Time, jc jobControlInterface, sjc

    sjControlInterface, pc podControlInterface, recorder record.EventRecorder) { // 50+ lines omitted times, err := getRecentUnmetScheduleTimes(*sj, now) if err != nil { return } // 10+ lines omitted scheduledTime := times[len(times)-1] tooLate := false if sj.Spec.StartingDeadlineSeconds != nil { tooLate = scheduledTime.Add(time.Second * time.Duration(*sj.Spec.StartingDeadlineSeconds)).Before(now) } if tooLate { return } // 30 lines omitted jobReq, err := getJobFromTemplate(sj, scheduledTime) if err != nil { return } jobResp, err := jc.CreateJob(sj.Namespace, jobReq) if err != nil { return } // 20+ lines omitted return } ൃݟ͕஗͗ͨ͢ʁ൑ఆ ະ࣮ߦ࣌ࠁҰཡΛऔಘ ˞దٓ؆ུԽ
  8. func syncOne(sj *batchv1beta1.CronJob, js []batchv1.Job, now time.Time, jc jobControlInterface, sjc

    sjControlInterface, pc podControlInterface, recorder record.EventRecorder) { // 50+ lines omitted times, err := getRecentUnmetScheduleTimes(*sj, now) if err != nil { return } // 10+ lines omitted scheduledTime := times[len(times)-1] tooLate := false if sj.Spec.StartingDeadlineSeconds != nil { tooLate = scheduledTime.Add(time.Second * time.Duration(*sj.Spec.StartingDeadlineSeconds)).Before(now) } if tooLate { return } // 30 lines omitted jobReq, err := getJobFromTemplate(sj, scheduledTime) if err != nil { return } jobResp, err := jc.CreateJob(sj.Namespace, jobReq) if err != nil { return } // 20+ lines omitted return } +PCൃߦ ൃݟ͕஗͗ͨ͢ʁ൑ఆ ະ࣮ߦ࣌ࠁҰཡΛऔಘ ˞దٓ؆ུԽ
  9. func getRecentUnmetScheduleTimes(sj batchv1beta1.CronJob, now time.Time) ([]time.Time, error) { starts :=

    []time.Time{} sched, err := cron.ParseStandard(sj.Spec.Schedule) if err != nil { return starts, fmt.Errorf("Unparseable schedule: %s : %s", sj.Spec.Schedule, err) } var earliestTime time.Time if sj.Status.LastScheduleTime != nil { earliestTime = sj.Status.LastScheduleTime.Time } else { earliestTime = sj.ObjectMeta.CreationTimestamp.Time } if sj.Spec.StartingDeadlineSeconds != nil { // Controller is not going to schedule anything below this point schedulingDeadline := now.Add(-time.Second * time.Duration(*sj.Spec.StartingDeadlineSeconds)) if schedulingDeadline.After(earliestTime) { earliestTime = schedulingDeadline } } if earliestTime.After(now) { return []time.Time{}, nil } for t := sched.Next(earliestTime); !t.After(now); t = sched.Next(t) { starts = append(starts, t) } return starts, nil } ˞దٓ؆ུԽ
  10. func getRecentUnmetScheduleTimes(sj batchv1beta1.CronJob, now time.Time) ([]time.Time, error) { starts :=

    []time.Time{} sched, err := cron.ParseStandard(sj.Spec.Schedule) if err != nil { return starts, fmt.Errorf("Unparseable schedule: %s : %s", sj.Spec.Schedule, err) } var earliestTime time.Time if sj.Status.LastScheduleTime != nil { earliestTime = sj.Status.LastScheduleTime.Time } else { earliestTime = sj.ObjectMeta.CreationTimestamp.Time } if sj.Spec.StartingDeadlineSeconds != nil { // Controller is not going to schedule anything below this point schedulingDeadline := now.Add(-time.Second * time.Duration(*sj.Spec.StartingDeadlineSeconds)) if schedulingDeadline.After(earliestTime) { earliestTime = schedulingDeadline } } if earliestTime.After(now) { return []time.Time{}, nil } for t := sched.Next(earliestTime); !t.After(now); t = sched.Next(t) { starts = append(starts, t) } return starts, nil } Ͳͷఔ౓աڈ͔Β୳͔͢ʁ ˞దٓ؆ུԽ
  11. func getRecentUnmetScheduleTimes(sj batchv1beta1.CronJob, now time.Time) ([]time.Time, error) { starts :=

    []time.Time{} sched, err := cron.ParseStandard(sj.Spec.Schedule) if err != nil { return starts, fmt.Errorf("Unparseable schedule: %s : %s", sj.Spec.Schedule, err) } var earliestTime time.Time if sj.Status.LastScheduleTime != nil { earliestTime = sj.Status.LastScheduleTime.Time } else { earliestTime = sj.ObjectMeta.CreationTimestamp.Time } if sj.Spec.StartingDeadlineSeconds != nil { // Controller is not going to schedule anything below this point schedulingDeadline := now.Add(-time.Second * time.Duration(*sj.Spec.StartingDeadlineSeconds)) if schedulingDeadline.After(earliestTime) { earliestTime = schedulingDeadline } } if earliestTime.After(now) { return []time.Time{}, nil } for t := sched.Next(earliestTime); !t.After(now); t = sched.Next(t) { starts = append(starts, t) } return starts, nil } Ͳͷఔ౓աڈ͔Β୳͔͢ʁ ͦͷաڈ͔Βࠓ·ͰϚον͢Δ࣌ࠁΛϦετ ˞దٓ؆ུԽ
  12. $SPO+PCͷBQQMZ࣌ʹ͍͖ͳΓKPC͕૸Γग़͢ ໌೔͔ΒAM 3:00 ʹಈ͍ͯ΄͍͠ 1:00 AM 3:00 AM ୳͢ൣғ 3:00

    AM ະ࣮ߦ൑ఆ ?:?? ?? ҙਤ͠ͳ͍Job࣮ߦ ࠓ೔ͷAM 1:00 ͔Β apply ͷॠؒ·Ͱʹࠓ೔ͷAM 3:00 ͕ೖͬͯ͠·͏ apply ͷ௚ޙʮະ࣮ߦͷ+PCͩʯͱೝࣝ͞Εͯͦͷॠؒಈ͖ग़͢ apply StartingDeadlineSeconds͕খ͚͞Ε͹͜͜ͰTLJQͯ͠΋Β͑Δ ࠓ೔ ໌೔ ࡢ೔ ࠷ޙ 1:00 AM ©2018 Wantedly, Inc.