diff --git a/probe/kubernetes/client.go b/probe/kubernetes/client.go index 69fd1aa75..4f2af1d01 100644 --- a/probe/kubernetes/client.go +++ b/probe/kubernetes/client.go @@ -11,6 +11,7 @@ import ( log "github.com/Sirupsen/logrus" apiappsv1beta1 "k8s.io/api/apps/v1beta1" apibatchv1 "k8s.io/api/batch/v1" + apibatchv1beta1 "k8s.io/api/batch/v1beta1" apibatchv2alpha1 "k8s.io/api/batch/v2alpha1" apiv1 "k8s.io/api/core/v1" apiextensionsv1beta1 "k8s.io/api/extensions/v1beta1" @@ -133,6 +134,11 @@ func NewClient(config ClientConfig) (Client, error) { client: c, } + result.cronJobStore, err = result.setupCronjobStore() + if err != nil { + return nil, err + } + podStore := NewEventStore(result.triggerPodWatches, cache.MetaNamespaceKeyFunc) result.podStore = result.setupStore(c.CoreV1().RESTClient(), "pods", &apiv1.Pod{}, podStore) result.serviceStore = result.setupStore(c.CoreV1().RESTClient(), "services", &apiv1.Service{}, nil) @@ -141,7 +147,6 @@ func NewClient(config ClientConfig) (Client, error) { result.deploymentStore = result.setupStore(c.ExtensionsV1beta1().RESTClient(), "deployments", &apiextensionsv1beta1.Deployment{}, nil) result.daemonSetStore = result.setupStore(c.ExtensionsV1beta1().RESTClient(), "daemonsets", &apiextensionsv1beta1.DaemonSet{}, nil) result.jobStore = result.setupStore(c.BatchV1().RESTClient(), "jobs", &apibatchv1.Job{}, nil) - result.cronJobStore = result.setupStore(c.BatchV2alpha1().RESTClient(), "cronjobs", &apibatchv2alpha1.CronJob{}, nil) result.statefulSetStore = result.setupStore(c.AppsV1beta1().RESTClient(), "statefulsets", &apiappsv1beta1.StatefulSet{}, nil) return result, nil @@ -175,6 +180,20 @@ func (c *client) setupStore(kclient rest.Interface, resource string, itemType in return store } +func (c *client) setupCronjobStore() (cache.Store, error) { + const resource = "cronjobs" + ok, err := c.isResourceSupported(c.client.BatchV1beta1().RESTClient().APIVersion(), resource) + if err != nil { + return nil, err + } + if ok { + // kubernetes >= 1.8 + return c.setupStore(c.client.BatchV1beta1().RESTClient(), resource, &apibatchv1beta1.CronJob{}, nil), nil + } + // kubernetes < 1.8 + return c.setupStore(c.client.BatchV2alpha1().RESTClient(), resource, &apibatchv2alpha1.CronJob{}, nil), nil +} + // runReflectorUntil runs cache.Reflector#ListAndWatch in an endless loop, after checking that the resource is supported by kubernetes. // Errors are logged and retried with exponential backoff. func (c *client) runReflectorUntil(r *cache.Reflector, groupVersion schema.GroupVersion, resource string) { @@ -287,8 +306,7 @@ func (c *client) WalkCronJobs(f func(CronJob) error) error { jobs[j.UID] = j } for _, m := range c.cronJobStore.List() { - cj := m.(*apibatchv2alpha1.CronJob) - if err := f(NewCronJob(cj, jobs)); err != nil { + if err := f(NewCronJob(m, jobs)); err != nil { return err } } diff --git a/probe/kubernetes/cronjob.go b/probe/kubernetes/cronjob.go index b0e58a548..5b8177b5e 100644 --- a/probe/kubernetes/cronjob.go +++ b/probe/kubernetes/cronjob.go @@ -5,6 +5,7 @@ import ( "time" batchv1 "k8s.io/api/batch/v1" + batchv1beta1 "k8s.io/api/batch/v1beta1" batchv2alpha1 "k8s.io/api/batch/v2alpha1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" @@ -29,14 +30,25 @@ type CronJob interface { } type cronJob struct { - *batchv2alpha1.CronJob + *batchv1beta1.CronJob Meta jobs []*batchv1.Job } // NewCronJob creates a new cron job. jobs should be all jobs, which will be filtered // for those matching this cron job. -func NewCronJob(cj *batchv2alpha1.CronJob, jobs map[types.UID]*batchv1.Job) CronJob { +func NewCronJob(cji interface{}, jobs map[types.UID]*batchv1.Job) CronJob { + switch cj := cji.(type) { + case *batchv2alpha1.CronJob: + return newCronJob(upgradeCronJob(cj), jobs) + case *batchv1beta1.CronJob: + return newCronJob(cj, jobs) + default: + panic(fmt.Sprintf("interface conversion: interface{} is %T, not *batchv2alpha1.CronJob or *batchv1beta1.CronJob", cj)) + } +} + +func newCronJob(cj *batchv1beta1.CronJob, jobs map[types.UID]*batchv1.Job) CronJob { myJobs := []*batchv1.Job{} for _, o := range cj.Status.Active { if j, ok := jobs[o.UID]; ok { @@ -74,3 +86,29 @@ func (cj *cronJob) GetNode() report.Node { } return cj.MetaNode(report.MakeCronJobNodeID(cj.UID())).WithLatests(latest) } + +func upgradeCronJob(legacy *batchv2alpha1.CronJob) *batchv1beta1.CronJob { + jobTemplate := batchv1beta1.JobTemplateSpec{ + ObjectMeta: legacy.Spec.JobTemplate.ObjectMeta, + Spec: legacy.Spec.JobTemplate.Spec, + } + spec := batchv1beta1.CronJobSpec{ + Schedule: legacy.Spec.Schedule, + StartingDeadlineSeconds: legacy.Spec.StartingDeadlineSeconds, + ConcurrencyPolicy: batchv1beta1.ConcurrencyPolicy(legacy.Spec.ConcurrencyPolicy), + Suspend: legacy.Spec.Suspend, + JobTemplate: jobTemplate, + SuccessfulJobsHistoryLimit: legacy.Spec.SuccessfulJobsHistoryLimit, + FailedJobsHistoryLimit: legacy.Spec.FailedJobsHistoryLimit, + } + status := batchv1beta1.CronJobStatus{ + Active: legacy.Status.Active, + LastScheduleTime: legacy.Status.LastScheduleTime, + } + return &batchv1beta1.CronJob{ + TypeMeta: legacy.TypeMeta, + ObjectMeta: legacy.ObjectMeta, + Spec: spec, + Status: status, + } +}