Merge pull request #13259 from soltysh/job_controller_logic

cmd/kube-controller-manager/app/controllermanager.go

@@ -38,6 +38,7 @@ import (
 	"k8s.io/kubernetes/pkg/controller/daemon"
 	"k8s.io/kubernetes/pkg/controller/deployment"
 	"k8s.io/kubernetes/pkg/controller/endpoint"
+	"k8s.io/kubernetes/pkg/controller/job"
 	"k8s.io/kubernetes/pkg/controller/namespace"
 	"k8s.io/kubernetes/pkg/controller/node"
 	"k8s.io/kubernetes/pkg/controller/persistentvolume"
@@ -66,6 +67,7 @@ type CMServer struct {
 	ConcurrentEndpointSyncs           int
 	ConcurrentRCSyncs                 int
 	ConcurrentDSCSyncs                int
+	ConcurrentJobSyncs                int
 	ServiceSyncPeriod                 time.Duration
 	NodeSyncPeriod                    time.Duration
 	ResourceQuotaSyncPeriod           time.Duration
@@ -104,6 +106,7 @@ func NewCMServer() *CMServer {
 		ConcurrentEndpointSyncs:           5,
 		ConcurrentRCSyncs:                 5,
 		ConcurrentDSCSyncs:                2,
+		ConcurrentJobSyncs:                5,
 		ServiceSyncPeriod:                 5 * time.Minute,
 		NodeSyncPeriod:                    10 * time.Second,
 		ResourceQuotaSyncPeriod:           10 * time.Second,
@@ -238,6 +241,9 @@ func (s *CMServer) Run(_ []string) error {
 	go daemon.NewDaemonSetsController(kubeClient).
 		Run(s.ConcurrentDSCSyncs, util.NeverStop)
 
+	go job.NewJobManager(kubeClient).
+		Run(s.ConcurrentJobSyncs, util.NeverStop)
+
 	cloud, err := cloudprovider.InitCloudProvider(s.CloudProvider, s.CloudConfigFile)
 	if err != nil {
 		glog.Fatalf("Cloud provider could not be initialized: %v", err)

pkg/api/testing/fuzzer.go

@@ -139,6 +139,13 @@ func FuzzerFor(t *testing.T, version string, src rand.Source) *fuzz.Fuzzer {
 				j.RollingUpdate = &rollingUpdate
 			}
 		},
+		func(j *experimental.JobSpec, c fuzz.Continue) {
+			c.FuzzNoCustom(j) // fuzz self without calling this function again
+			completions := c.Rand.Int()
+			parallelism := c.Rand.Int()
+			j.Completions = &completions
+			j.Parallelism = &parallelism
+		},
 		func(j *api.List, c fuzz.Continue) {
 			c.FuzzNoCustom(j) // fuzz self without calling this function again
 			// TODO: uncomment when round trip starts from a versioned object

pkg/apis/experimental/v1/defaults.go

@@ -75,5 +75,15 @@ func addDefaultingFuncs() {
 				*obj.Spec.UniqueLabelKey = "deployment.kubernetes.io/podTemplateHash"
 			}
 		},
+		func(obj *Job) {
+			if obj.Spec.Completions == nil {
+				completions := 1
+				obj.Spec.Completions = &completions
+			}
+			if obj.Spec.Parallelism == nil {
+				parallelism := 2
+				obj.Spec.Parallelism = &parallelism
+			}
+		},
 	)
 }

pkg/apis/experimental/v1/defaults_test.go

@@ -189,6 +189,43 @@ func TestSetDefaultDeployment(t *testing.T) {
 	}
 }
 
+func TestSetDefaultJob(t *testing.T) {
+	expected := &Job{
+		Spec: JobSpec{
+			Completions: newInt(1),
+			Parallelism: newInt(2),
+		},
+	}
+	tests := []*Job{
+		{},
+		{
+			Spec: JobSpec{
+				Completions: newInt(1),
+			},
+		},
+		{
+			Spec: JobSpec{
+				Parallelism: newInt(2),
+			},
+		},
+	}
+
+	for _, original := range tests {
+		obj2 := roundTrip(t, runtime.Object(original))
+		got, ok := obj2.(*Job)
+		if !ok {
+			t.Errorf("unexpected object: %v", got)
+			t.FailNow()
+		}
+		if *got.Spec.Completions != *expected.Spec.Completions {
+			t.Errorf("got different completions than expected: %d %d", *got.Spec.Completions, *expected.Spec.Completions)
+		}
+		if *got.Spec.Parallelism != *expected.Spec.Parallelism {
+			t.Errorf("got different parallelism than expected: %d %d", *got.Spec.Parallelism, *expected.Spec.Parallelism)
+		}
+	}
+}
+
 func roundTrip(t *testing.T, obj runtime.Object) runtime.Object {
 	data, err := v1.Codec.Encode(obj)
 	if err != nil {

pkg/client/cache/listers.go

@@ -344,3 +344,55 @@ func (s *StoreToEndpointsLister) GetServiceEndpoints(svc *api.Service) (ep api.E
 	err = fmt.Errorf("Could not find endpoints for service: %v", svc.Name)
 	return
 }
+
+// StoreToJobLister gives a store List and Exists methods. The store must contain only Jobs.
+type StoreToJobLister struct {
+	Store
+}
+
+// Exists checks if the given job exists in the store.
+func (s *StoreToJobLister) Exists(job *experimental.Job) (bool, error) {
+	_, exists, err := s.Store.Get(job)
+	if err != nil {
+		return false, err
+	}
+	return exists, nil
+}
+
+// StoreToJobLister lists all jobs in the store.
+func (s *StoreToJobLister) List() (jobs []experimental.Job, err error) {
+	for _, c := range s.Store.List() {
+		jobs = append(jobs, *(c.(*experimental.Job)))
+	}
+	return jobs, nil
+}
+
+// GetPodControllers returns a list of jobs managing a pod. Returns an error only if no matching jobs are found.
+func (s *StoreToJobLister) GetPodJobs(pod *api.Pod) (jobs []experimental.Job, err error) {
+	var selector labels.Selector
+	var job experimental.Job
+
+	if len(pod.Labels) == 0 {
+		err = fmt.Errorf("No jobs found for pod %v because it has no labels", pod.Name)
+		return
+	}
+
+	for _, m := range s.Store.List() {
+		job = *m.(*experimental.Job)
+		if job.Namespace != pod.Namespace {
+			continue
+		}
+		labelSet := labels.Set(job.Spec.Selector)
+		selector = labels.Set(job.Spec.Selector).AsSelector()
+
+		// Job with a nil or empty selector match nothing
+		if labelSet.AsSelector().Empty() || !selector.Matches(labels.Set(pod.Labels)) {
+			continue
+		}
+		jobs = append(jobs, job)
+	}
+	if len(jobs) == 0 {
+		err = fmt.Errorf("Could not find jobs for pod %s in namespace %s with labels: %v", pod.Name, pod.Namespace, pod.Labels)
+	}
+	return
+}

pkg/client/unversioned/testclient/actions.go

@@ -100,6 +100,17 @@ func NewUpdateAction(resource, namespace string, object runtime.Object) UpdateAc
 	return action
 }
 
+func NewUpdateSubresourceAction(resource, subresource, namespace string, object runtime.Object) UpdateActionImpl {
+	action := UpdateActionImpl{}
+	action.Verb = "update"
+	action.Resource = resource
+	action.Subresource = subresource
+	action.Namespace = namespace
+	action.Object = object
+
+	return action
+}
+
 func NewRootDeleteAction(resource, name string) DeleteActionImpl {
 	action := DeleteActionImpl{}
 	action.Verb = "delete"

pkg/client/unversioned/testclient/fake_jobs.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+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 testclient
+
+import (
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/apis/experimental"
+	"k8s.io/kubernetes/pkg/fields"
+	"k8s.io/kubernetes/pkg/labels"
+	"k8s.io/kubernetes/pkg/watch"
+)
+
+// FakeJobs implements JobInterface. Meant to be embedded into a struct to get a default
+// implementation. This makes faking out just the method you want to test easier.
+type FakeJobs struct {
+	Fake      *FakeExperimental
+	Namespace string
+}
+
+func (c *FakeJobs) Get(name string) (*experimental.Job, error) {
+	obj, err := c.Fake.Invokes(NewGetAction("jobs", c.Namespace, name), &experimental.Job{})
+	if obj == nil {
+		return nil, err
+	}
+
+	return obj.(*experimental.Job), err
+}
+
+func (c *FakeJobs) List(label labels.Selector, fields fields.Selector) (*experimental.JobList, error) {
+	obj, err := c.Fake.Invokes(NewListAction("jobs", c.Namespace, label, nil), &experimental.JobList{})
+	if obj == nil {
+		return nil, err
+	}
+
+	return obj.(*experimental.JobList), err
+}
+
+func (c *FakeJobs) Create(job *experimental.Job) (*experimental.Job, error) {
+	obj, err := c.Fake.Invokes(NewCreateAction("jobs", c.Namespace, job), job)
+	if obj == nil {
+		return nil, err
+	}
+
+	return obj.(*experimental.Job), err
+}
+
+func (c *FakeJobs) Update(job *experimental.Job) (*experimental.Job, error) {
+	obj, err := c.Fake.Invokes(NewUpdateAction("jobs", c.Namespace, job), job)
+	if obj == nil {
+		return nil, err
+	}
+
+	return obj.(*experimental.Job), err
+}
+
+func (c *FakeJobs) Delete(name string, options *api.DeleteOptions) error {
+	_, err := c.Fake.Invokes(NewDeleteAction("jobs", c.Namespace, name), &experimental.Job{})
+	return err
+}
+
+func (c *FakeJobs) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
+	return c.Fake.InvokesWatch(NewWatchAction("jobs", c.Namespace, label, field, resourceVersion))
+}
+
+func (c *FakeJobs) UpdateStatus(job *experimental.Job) (result *experimental.Job, err error) {
+	obj, err := c.Fake.Invokes(NewUpdateSubresourceAction("jobs", "status", c.Namespace, job), job)
+	if obj == nil {
+		return nil, err
+	}
+
+	return obj.(*experimental.Job), err
+}

pkg/client/unversioned/testclient/fake_persistent_volume_claims.go

@@ -74,13 +74,7 @@ func (c *FakePersistentVolumeClaims) Watch(label labels.Selector, field fields.S
 }
 
 func (c *FakePersistentVolumeClaims) UpdateStatus(claim *api.PersistentVolumeClaim) (*api.PersistentVolumeClaim, error) {
-	action := UpdateActionImpl{}
-	action.Verb = "update"
-	action.Resource = "persistentvolumeclaims"
-	action.Subresource = "status"
-	action.Object = claim
-
-	obj, err := c.Fake.Invokes(action, claim)
+	obj, err := c.Fake.Invokes(NewUpdateSubresourceAction("persistentvolumeclaims", "status", c.Namespace, claim), claim)
 	if obj == nil {
 		return nil, err
 	}

pkg/client/unversioned/testclient/fake_pods.go

@@ -92,13 +92,7 @@ func (c *FakePods) Bind(binding *api.Binding) error {
 }
 
 func (c *FakePods) UpdateStatus(pod *api.Pod) (*api.Pod, error) {
-	action := UpdateActionImpl{}
-	action.Verb = "update"
-	action.Resource = "pods"
-	action.Subresource = "status"
-	action.Object = pod
-
-	obj, err := c.Fake.Invokes(action, pod)
+	obj, err := c.Fake.Invokes(NewUpdateSubresourceAction("pods", "status", c.Namespace, pod), pod)
 	if obj == nil {
 		return nil, err
 	}

pkg/client/unversioned/testclient/fake_resource_quotas.go

@@ -76,13 +76,7 @@ func (c *FakeResourceQuotas) Watch(label labels.Selector, field fields.Selector,
 }
 
 func (c *FakeResourceQuotas) UpdateStatus(resourceQuota *api.ResourceQuota) (*api.ResourceQuota, error) {
-	action := UpdateActionImpl{}
-	action.Verb = "update"
-	action.Resource = "resourcequotas"
-	action.Subresource = "status"
-	action.Object = resourceQuota
-
-	obj, err := c.Fake.Invokes(action, resourceQuota)
+	obj, err := c.Fake.Invokes(NewUpdateSubresourceAction("resourcequotas", "status", c.Namespace, resourceQuota), resourceQuota)
 	if obj == nil {
 		return nil, err
 	}

pkg/client/unversioned/testclient/testclient.go

@@ -304,5 +304,5 @@ func (c *FakeExperimental) Scales(namespace string) client.ScaleInterface {
 }
 
 func (c *FakeExperimental) Jobs(namespace string) client.JobInterface {
-	panic("unimplemented")
+	return &FakeJobs{Fake: c, Namespace: namespace}
 }

pkg/controller/controller_utils.go

@@ -26,7 +26,6 @@ import (
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/latest"
 	"k8s.io/kubernetes/pkg/api/validation"
-	"k8s.io/kubernetes/pkg/apis/experimental"
 	"k8s.io/kubernetes/pkg/client/cache"
 	"k8s.io/kubernetes/pkg/client/record"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
@@ -213,10 +212,10 @@ func NewControllerExpectations() *ControllerExpectations {
 // PodControlInterface is an interface that knows how to add or delete pods
 // created as an interface to allow testing.
 type PodControlInterface interface {
-	// CreateReplica creates new replicated pods according to the spec.
-	CreateReplica(namespace string, controller *api.ReplicationController) error
-	// CreateReplicaOnNode creates a new pod according to the spec on the specified node.
-	CreateReplicaOnNode(namespace string, ds *experimental.DaemonSet, nodeName string) error
+	// CreatePods creates new pods according to the spec.
+	CreatePods(namespace string, template *api.PodTemplateSpec, object runtime.Object) error
+	// CreatePodsOnNode creates a new pod accorting to the spec on the specified node.
+	CreatePodsOnNode(nodeName, namespace string, template *api.PodTemplateSpec, object runtime.Object) error
 	// DeletePod deletes the pod identified by podID.
 	DeletePod(namespace string, podID string) error
 }
@@ -227,7 +226,7 @@ type RealPodControl struct {
 	Recorder   record.EventRecorder
 }
 
-func getReplicaLabelSet(template *api.PodTemplateSpec) labels.Set {
+func getPodsLabelSet(template *api.PodTemplateSpec) labels.Set {
 	desiredLabels := make(labels.Set)
 	for k, v := range template.Labels {
 		desiredLabels[k] = v
@@ -235,7 +234,7 @@ func getReplicaLabelSet(template *api.PodTemplateSpec) labels.Set {
 	return desiredLabels
 }
 
-func getReplicaAnnotationSet(template *api.PodTemplateSpec, object runtime.Object) (labels.Set, error) {
+func getPodsAnnotationSet(template *api.PodTemplateSpec, object runtime.Object) (labels.Set, error) {
 	desiredAnnotations := make(labels.Set)
 	for k, v := range template.Annotations {
 		desiredAnnotations[k] = v
@@ -254,7 +253,7 @@ func getReplicaAnnotationSet(template *api.PodTemplateSpec, object runtime.Objec
 	return desiredAnnotations, nil
 }
 
-func getReplicaPrefix(controllerName string) string {
+func getPodsPrefix(controllerName string) string {
 	// use the dash (if the name isn't too long) to make the pod name a bit prettier
 	prefix := fmt.Sprintf("%s-", controllerName)
 	if ok, _ := validation.ValidatePodName(prefix, true); !ok {
@@ -263,44 +262,25 @@ func getReplicaPrefix(controllerName string) string {
 	return prefix
 }
 
-func (r RealPodControl) CreateReplica(namespace string, controller *api.ReplicationController) error {
-	desiredLabels := getReplicaLabelSet(controller.Spec.Template)
-	desiredAnnotations, err := getReplicaAnnotationSet(controller.Spec.Template, controller)
-	if err != nil {
-		return err
-	}
-	prefix := getReplicaPrefix(controller.Name)
+func (r RealPodControl) CreatePods(namespace string, template *api.PodTemplateSpec, object runtime.Object) error {
+	return r.createPods("", namespace, template, object)
+}
 
-	pod := &api.Pod{
-		ObjectMeta: api.ObjectMeta{
-			Labels:       desiredLabels,
-			Annotations:  desiredAnnotations,
-			GenerateName: prefix,
-		},
-	}
-	if err := api.Scheme.Convert(&controller.Spec.Template.Spec, &pod.Spec); err != nil {
-		return fmt.Errorf("unable to convert pod template: %v", err)
-	}
-	if labels.Set(pod.Labels).AsSelector().Empty() {
-		return fmt.Errorf("unable to create pod replica, no labels")
-	}
-	if newPod, err := r.KubeClient.Pods(namespace).Create(pod); err != nil {
-		r.Recorder.Eventf(controller, "FailedCreate", "Error creating: %v", err)
-		return fmt.Errorf("unable to create pod replica: %v", err)
-	} else {
-		glog.V(4).Infof("Controller %v created pod %v", controller.Name, newPod.Name)
-		r.Recorder.Eventf(controller, "SuccessfulCreate", "Created pod: %v", newPod.Name)
-	}
-	return nil
+func (r RealPodControl) CreatePodsOnNode(nodeName, namespace string, template *api.PodTemplateSpec, object runtime.Object) error {
+	return r.createPods(nodeName, namespace, template, object)
 }
 
-func (r RealPodControl) CreateReplicaOnNode(namespace string, ds *experimental.DaemonSet, nodeName string) error {
-	desiredLabels := getReplicaLabelSet(ds.Spec.Template)
-	desiredAnnotations, err := getReplicaAnnotationSet(ds.Spec.Template, ds)
+func (r RealPodControl) createPods(nodeName, namespace string, template *api.PodTemplateSpec, object runtime.Object) error {
+	desiredLabels := getPodsLabelSet(template)
+	desiredAnnotations, err := getPodsAnnotationSet(template, object)
 	if err != nil {
 		return err
 	}
-	prefix := getReplicaPrefix(ds.Name)
+	meta, err := api.ObjectMetaFor(object)
+	if err != nil {
+		return fmt.Errorf("object does not have ObjectMeta, %v", err)
+	}
+	prefix := getPodsPrefix(meta.Name)
 
 	pod := &api.Pod{
 		ObjectMeta: api.ObjectMeta{
@@ -309,22 +289,22 @@ func (r RealPodControl) CreateReplicaOnNode(namespace string, ds *experimental.D
 			GenerateName: prefix,
 		},
 	}
-	if err := api.Scheme.Convert(&ds.Spec.Template.Spec, &pod.Spec); err != nil {
+	if len(nodeName) != 0 {
+		pod.Spec.NodeName = nodeName
+	}
+	if err := api.Scheme.Convert(&template.Spec, &pod.Spec); err != nil {
 		return fmt.Errorf("unable to convert pod template: %v", err)
 	}
-	// if a pod does not have labels then it cannot be controlled by any controller
 	if labels.Set(pod.Labels).AsSelector().Empty() {
-		return fmt.Errorf("unable to create pod replica, no labels")
+		return fmt.Errorf("unable to create pods, no labels")
 	}
-	pod.Spec.NodeName = nodeName
 	if newPod, err := r.KubeClient.Pods(namespace).Create(pod); err != nil {
-		r.Recorder.Eventf(ds, "FailedCreate", "Error creating: %v", err)
-		return fmt.Errorf("unable to create pod replica: %v", err)
+		r.Recorder.Eventf(object, "FailedCreate", "Error creating: %v", err)
+		return fmt.Errorf("unable to create pods: %v", err)
 	} else {
-		glog.V(4).Infof("Controller %v created pod %v", ds.Name, newPod.Name)
-		r.Recorder.Eventf(ds, "SuccessfulCreate", "Created pod: %v", newPod.Name)
+		glog.V(4).Infof("Controller %v created pod %v", meta.Name, newPod.Name)
+		r.Recorder.Eventf(object, "SuccessfulCreate", "Created pod: %v", newPod.Name)
 	}
-
 	return nil
 }
 

pkg/controller/controller_utils_test.go

@@ -180,7 +180,7 @@ func TestControllerExpectations(t *testing.T) {
 	}
 }
 
-func TestCreateReplica(t *testing.T) {
+func TestCreatePods(t *testing.T) {
 	ns := api.NamespaceDefault
 	body := runtime.EncodeOrDie(testapi.Default.Codec(), &api.Pod{ObjectMeta: api.ObjectMeta{Name: "empty_pod"}})
 	fakeHandler := util.FakeHandler{
@@ -199,7 +199,7 @@ func TestCreateReplica(t *testing.T) {
 	controllerSpec := newReplicationController(1)
 
 	// Make sure createReplica sends a POST to the apiserver with a pod from the controllers pod template
-	podControl.CreateReplica(ns, controllerSpec)
+	podControl.CreatePods(ns, controllerSpec.Spec.Template, controllerSpec)
 
 	expectedPod := api.Pod{
 		ObjectMeta: api.ObjectMeta{

pkg/controller/daemon/controller.go

@@ -371,7 +371,7 @@ func (dsc *DaemonSetsController) manage(ds *experimental.DaemonSet) {
 
 	glog.V(4).Infof("Nodes needing daemon pods for daemon set %s: %+v", ds.Name, nodesNeedingDaemonPods)
 	for i := range nodesNeedingDaemonPods {
-		if err := dsc.podControl.CreateReplicaOnNode(ds.Namespace, ds, nodesNeedingDaemonPods[i]); err != nil {
+		if err := dsc.podControl.CreatePodsOnNode(nodesNeedingDaemonPods[i], ds.Namespace, ds.Spec.Template, ds); err != nil {
 			glog.V(2).Infof("Failed creation, decrementing expectations for set %q/%q", ds.Namespace, ds.Name)
 			dsc.expectations.CreationObserved(dsKey)
 			util.HandleError(err)

pkg/controller/daemon/controller_test.go

@@ -27,6 +27,7 @@ import (
 	"k8s.io/kubernetes/pkg/client/cache"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/controller"
+	"k8s.io/kubernetes/pkg/runtime"
 	"k8s.io/kubernetes/pkg/securitycontext"
 )
 
@@ -38,7 +39,7 @@ var (
 )
 
 type FakePodControl struct {
-	daemonSet     []experimental.DaemonSet
+	podSpec       []api.PodTemplateSpec
 	deletePodName []string
 	lock          sync.Mutex
 	err           error
@@ -48,17 +49,17 @@ func init() {
 	api.ForTesting_ReferencesAllowBlankSelfLinks = true
 }
 
-func (f *FakePodControl) CreateReplica(namespace string, spec *api.ReplicationController) error {
+func (f *FakePodControl) CreatePods(namespace string, spec *api.PodTemplateSpec, object runtime.Object) error {
 	return nil
 }
 
-func (f *FakePodControl) CreateReplicaOnNode(namespace string, ds *experimental.DaemonSet, nodeName string) error {
+func (f *FakePodControl) CreatePodsOnNode(nodeName, namespace string, spec *api.PodTemplateSpec, object runtime.Object) error {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 	if f.err != nil {
 		return f.err
 	}
-	f.daemonSet = append(f.daemonSet, *ds)
+	f.podSpec = append(f.podSpec, *spec)
 	return nil
 }
 
@@ -75,7 +76,7 @@ func (f *FakePodControl) clear() {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 	f.deletePodName = []string{}
-	f.daemonSet = []experimental.DaemonSet{}
+	f.podSpec = []api.PodTemplateSpec{}
 }
 
 func newDaemonSet(name string) *experimental.DaemonSet {
@@ -164,8 +165,8 @@ func newTestController() (*DaemonSetsController, *FakePodControl) {
 }
 
 func validateSyncDaemonSets(t *testing.T, fakePodControl *FakePodControl, expectedCreates, expectedDeletes int) {
-	if len(fakePodControl.daemonSet) != expectedCreates {
-		t.Errorf("Unexpected number of creates.  Expected %d, saw %d\n", expectedCreates, len(fakePodControl.daemonSet))
+	if len(fakePodControl.podSpec) != expectedCreates {
+		t.Errorf("Unexpected number of creates.  Expected %d, saw %d\n", expectedCreates, len(fakePodControl.podSpec))
 	}
 	if len(fakePodControl.deletePodName) != expectedDeletes {
 		t.Errorf("Unexpected number of deletes.  Expected %d, saw %d\n", expectedDeletes, len(fakePodControl.deletePodName))

pkg/controller/job/doc.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+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 job contains logic for watching and synchronizing jobs.
+package job

pkg/controller/replication/replication_controller.go

@@ -350,7 +350,7 @@ func (rm *ReplicationManager) manageReplicas(filteredPods []*api.Pod, rc *api.Re
 		for i := 0; i < diff; i++ {
 			go func() {
 				defer wait.Done()
-				if err := rm.podControl.CreateReplica(rc.Namespace, rc); err != nil {
+				if err := rm.podControl.CreatePods(rc.Namespace, rc.Spec.Template, rc); err != nil {
 					// Decrement the expected number of creates because the informer won't observe this pod
 					glog.V(2).Infof("Failed creation, decrementing expectations for controller %q/%q", rc.Namespace, rc.Name)
 					rm.expectations.CreationObserved(rcKey)

pkg/controller/replication/replication_controller_test.go

@@ -27,7 +27,6 @@ import (
 
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/testapi"
-	"k8s.io/kubernetes/pkg/apis/experimental"
 	"k8s.io/kubernetes/pkg/client/cache"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/client/unversioned/testclient"
@@ -42,7 +41,7 @@ import (
 )
 
 type FakePodControl struct {
-	controllerSpec []api.ReplicationController
+	controllerSpec []api.PodTemplateSpec
 	deletePodName  []string
 	lock           sync.Mutex
 	err            error
@@ -60,7 +59,7 @@ func init() {
 	api.ForTesting_ReferencesAllowBlankSelfLinks = true
 }
 
-func (f *FakePodControl) CreateReplica(namespace string, spec *api.ReplicationController) error {
+func (f *FakePodControl) CreatePods(namespace string, spec *api.PodTemplateSpec, object runtime.Object) error {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 	if f.err != nil {
@@ -70,7 +69,7 @@ func (f *FakePodControl) CreateReplica(namespace string, spec *api.ReplicationCo
 	return nil
 }
 
-func (f *FakePodControl) CreateReplicaOnNode(namespace string, daemon *experimental.DaemonSet, nodeName string) error {
+func (f *FakePodControl) CreatePodsOnNode(nodeName, namespace string, template *api.PodTemplateSpec, object runtime.Object) error {
 	return nil
 }
 
@@ -87,7 +86,7 @@ func (f *FakePodControl) clear() {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 	f.deletePodName = []string{}
-	f.controllerSpec = []api.ReplicationController{}
+	f.controllerSpec = []api.PodTemplateSpec{}
 }
 
 func getKey(rc *api.ReplicationController, t *testing.T) string {