Merge pull request #35943 from jianhuiz/federation-apiserver-batch-job

federation/cmd/federation-apiserver/app/BUILD

@@ -10,6 +10,7 @@ load(
 go_library(
     name = "go_default_library",
     srcs = [
+        "batch.go",
         "core.go",
         "extensions.go",
         "federation.go",
@@ -30,6 +31,8 @@ go_library(
         "//pkg/api/install:go_default_library",
         "//pkg/api/rest:go_default_library",
         "//pkg/apimachinery/registered:go_default_library",
+        "//pkg/apis/batch:go_default_library",
+        "//pkg/apis/batch/install:go_default_library",
         "//pkg/apis/extensions:go_default_library",
         "//pkg/apis/extensions/install:go_default_library",
         "//pkg/apiserver/authenticator:go_default_library",
@@ -40,6 +43,7 @@ go_library(
         "//pkg/genericapiserver:go_default_library",
         "//pkg/genericapiserver/authorizer:go_default_library",
         "//pkg/genericapiserver/filters:go_default_library",
+        "//pkg/registry/batch/job/etcd:go_default_library",
         "//pkg/registry/cachesize:go_default_library",
         "//pkg/registry/core/configmap/etcd:go_default_library",
         "//pkg/registry/core/event/etcd:go_default_library",

federation/cmd/federation-apiserver/app/batch.go

+/*
+Copyright 2016 The Kubernetes Authors.
+
+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 app
+
+import (
+	"github.com/golang/glog"
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/api/rest"
+	"k8s.io/kubernetes/pkg/apimachinery/registered"
+	"k8s.io/kubernetes/pkg/apis/batch"
+	_ "k8s.io/kubernetes/pkg/apis/batch/install"
+	"k8s.io/kubernetes/pkg/genericapiserver"
+	jobetcd "k8s.io/kubernetes/pkg/registry/batch/job/etcd"
+	"k8s.io/kubernetes/pkg/registry/generic"
+)
+
+func installBatchAPIs(g *genericapiserver.GenericAPIServer, optsGetter generic.RESTOptionsGetter) {
+	jobStorage := jobetcd.NewStorage(optsGetter)
+
+	batchResources := map[string]rest.Storage{
+		"jobs":        jobStorage.Job,
+		"jobs/status": jobStorage.Status,
+	}
+	batchGroupMeta := registered.GroupOrDie(batch.GroupName)
+	apiGroupInfo := genericapiserver.APIGroupInfo{
+		GroupMeta: *batchGroupMeta,
+		VersionedResourcesStorageMap: map[string]map[string]rest.Storage{
+			"v1": batchResources,
+		},
+		OptionsExternalVersion: &registered.GroupOrDie(api.GroupName).GroupVersion,
+		Scheme:                 api.Scheme,
+		ParameterCodec:         api.ParameterCodec,
+		NegotiatedSerializer:   api.Codecs,
+	}
+	if err := g.InstallAPIGroup(&apiGroupInfo); err != nil {
+		glog.Fatalf("Error in registering group versions: %v", err)
+	}
+}

federation/cmd/federation-apiserver/app/server.go

@@ -207,6 +207,7 @@ func Run(s *options.ServerRunOptions) error {
 	installFederationAPIs(m, restOptionsFactory)
 	installCoreAPIs(s, m, restOptionsFactory)
 	installExtensionsAPIs(m, restOptionsFactory)
+	installBatchAPIs(m, restOptionsFactory)
 
 	sharedInformers.Start(wait.NeverStop)
 	m.PrepareRun().Run(wait.NeverStop)

pkg/registry/batch/job/etcd/etcd.go

@@ -27,6 +27,21 @@ import (
 	"k8s.io/kubernetes/pkg/runtime"
 )
 
+// JobStorage includes dummy storage for Job.
+type JobStorage struct {
+	Job    *REST
+	Status *StatusREST
+}
+
+func NewStorage(optsGetter generic.RESTOptionsGetter) JobStorage {
+	jobRest, jobStatusRest := NewREST(optsGetter)
+
+	return JobStorage{
+		Job:    jobRest,
+		Status: jobStatusRest,
+	}
+}
+
 // REST implements a RESTStorage for jobs against etcd
 type REST struct {
 	*genericregistry.Store

pkg/registry/batch/job/etcd/etcd_test.go

@@ -30,7 +30,7 @@ import (
 	etcdtesting "k8s.io/kubernetes/pkg/storage/etcd/testing"
 )
 
-func newStorage(t *testing.T) (*REST, *StatusREST, *etcdtesting.EtcdTestServer) {
+func newStorage(t *testing.T) (*JobStorage, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, batch.GroupName)
 	restOptions := generic.RESTOptions{
 		StorageConfig:           etcdStorage,
@@ -38,8 +38,8 @@ func newStorage(t *testing.T) (*REST, *StatusREST, *etcdtesting.EtcdTestServer)
 		DeleteCollectionWorkers: 1,
 		ResourcePrefix:          "jobs",
 	}
-	jobStorage, statusStorage := NewREST(restOptions)
-	return jobStorage, statusStorage, server
+	jobStorage := NewStorage(restOptions)
+	return &jobStorage, server
 }
 
 func validNewJob() *batch.Job {
@@ -78,10 +78,10 @@ func validNewJob() *batch.Job {
 }
 
 func TestCreate(t *testing.T) {
-	storage, _, server := newStorage(t)
+	storage, server := newStorage(t)
 	defer server.Terminate(t)
-	defer storage.Store.DestroyFunc()
-	test := registrytest.New(t, storage.Store)
+	defer storage.Job.Store.DestroyFunc()
+	test := registrytest.New(t, storage.Job.Store)
 	validJob := validNewJob()
 	validJob.ObjectMeta = api.ObjectMeta{}
 	test.TestCreate(
@@ -99,10 +99,10 @@ func TestCreate(t *testing.T) {
 }
 
 func TestUpdate(t *testing.T) {
-	storage, _, server := newStorage(t)
+	storage, server := newStorage(t)
 	defer server.Terminate(t)
-	defer storage.Store.DestroyFunc()
-	test := registrytest.New(t, storage.Store)
+	defer storage.Job.Store.DestroyFunc()
+	test := registrytest.New(t, storage.Job.Store)
 	two := int32(2)
 	test.TestUpdate(
 		// valid
@@ -128,34 +128,34 @@ func TestUpdate(t *testing.T) {
 }
 
 func TestDelete(t *testing.T) {
-	storage, _, server := newStorage(t)
+	storage, server := newStorage(t)
 	defer server.Terminate(t)
-	defer storage.Store.DestroyFunc()
-	test := registrytest.New(t, storage.Store)
+	defer storage.Job.Store.DestroyFunc()
+	test := registrytest.New(t, storage.Job.Store)
 	test.TestDelete(validNewJob())
 }
 
 func TestGet(t *testing.T) {
-	storage, _, server := newStorage(t)
+	storage, server := newStorage(t)
 	defer server.Terminate(t)
-	defer storage.Store.DestroyFunc()
-	test := registrytest.New(t, storage.Store)
+	defer storage.Job.Store.DestroyFunc()
+	test := registrytest.New(t, storage.Job.Store)
 	test.TestGet(validNewJob())
 }
 
 func TestList(t *testing.T) {
-	storage, _, server := newStorage(t)
+	storage, server := newStorage(t)
 	defer server.Terminate(t)
-	defer storage.Store.DestroyFunc()
-	test := registrytest.New(t, storage.Store)
+	defer storage.Job.Store.DestroyFunc()
+	test := registrytest.New(t, storage.Job.Store)
 	test.TestList(validNewJob())
 }
 
 func TestWatch(t *testing.T) {
-	storage, _, server := newStorage(t)
+	storage, server := newStorage(t)
 	defer server.Terminate(t)
-	defer storage.Store.DestroyFunc()
-	test := registrytest.New(t, storage.Store)
+	defer storage.Job.Store.DestroyFunc()
+	test := registrytest.New(t, storage.Job.Store)
 	test.TestWatch(
 		validNewJob(),
 		// matching labels

test/integration/federation/BUILD

@@ -16,6 +16,7 @@ go_test(
         "//federation/cmd/federation-apiserver/app:go_default_library",
         "//federation/cmd/federation-apiserver/app/options:go_default_library",
         "//pkg/api/v1:go_default_library",
+        "//pkg/apis/batch/v1:go_default_library",
         "//pkg/apis/extensions/v1beta1:go_default_library",
         "//pkg/apis/meta/v1:go_default_library",
         "//pkg/runtime/schema:go_default_library",

test/integration/federation/server_test.go

@@ -31,6 +31,7 @@ import (
 	"k8s.io/kubernetes/federation/cmd/federation-apiserver/app"
 	"k8s.io/kubernetes/federation/cmd/federation-apiserver/app/options"
 	"k8s.io/kubernetes/pkg/api/v1"
+	batch_v1 "k8s.io/kubernetes/pkg/apis/batch/v1"
 	ext_v1b1 "k8s.io/kubernetes/pkg/apis/extensions/v1beta1"
 	metav1 "k8s.io/kubernetes/pkg/apis/meta/v1"
 	"k8s.io/kubernetes/pkg/runtime/schema"
@@ -42,6 +43,7 @@ var serverIP = fmt.Sprintf("http://localhost:%v", insecurePort)
 var groupVersions = []schema.GroupVersion{
 	fed_v1b1.SchemeGroupVersion,
 	ext_v1b1.SchemeGroupVersion,
+	batch_v1.SchemeGroupVersion,
 }
 
 func TestRun(t *testing.T) {
@@ -211,6 +213,7 @@ func testAPIResourceList(t *testing.T) {
 	testFederationResourceList(t)
 	testCoreResourceList(t)
 	testExtensionsResourceList(t)
+	testBatchResourceList(t)
 }
 
 func testFederationResourceList(t *testing.T) {
@@ -344,3 +347,29 @@ func testExtensionsResourceList(t *testing.T) {
 	assert.True(t, found.Namespaced)
 	found = findResource(apiResourceList.APIResources, "deployments/rollback")
 }
+
+func testBatchResourceList(t *testing.T) {
+	serverURL := serverIP + "/apis/" + batch_v1.SchemeGroupVersion.String()
+	contents, err := readResponse(serverURL)
+	if err != nil {
+		t.Fatalf("%v", err)
+	}
+	var apiResourceList metav1.APIResourceList
+	err = json.Unmarshal(contents, &apiResourceList)
+	if err != nil {
+		t.Fatalf("Error in unmarshalling response from server %s: %v", serverURL, err)
+	}
+	// empty APIVersion for extensions group
+	assert.Equal(t, "v1", apiResourceList.APIVersion)
+	assert.Equal(t, batch_v1.SchemeGroupVersion.String(), apiResourceList.GroupVersion)
+	// Assert that there are exactly this number of resources.
+	assert.Equal(t, 2, len(apiResourceList.APIResources))
+
+	// Verify jobs
+	found := findResource(apiResourceList.APIResources, "jobs")
+	assert.NotNil(t, found)
+	assert.True(t, found.Namespaced)
+	found = findResource(apiResourceList.APIResources, "jobs/status")
+	assert.NotNil(t, found)
+	assert.True(t, found.Namespaced)
+}