Consolidate local OpenAPI specs and APIServices' spec into one data structure

cmd/kube-apiserver/app/server.go

@@ -165,6 +165,9 @@ func CreateServerChain(runOptions *options.ServerRunOptions, stopCh <-chan struc
 	// this wires up openapi
 	kubeAPIServer.GenericAPIServer.PrepareRun()
 
+	// This will wire up openapi for extension api server
+	apiExtensionsServer.GenericAPIServer.PrepareRun()
+
 	// aggregator comes last in the chain
 	aggregatorConfig, err := createAggregatorConfig(*kubeAPIServerConfig.GenericConfig, runOptions, versionedInformers, serviceResolver, proxyTransport)
 	if err != nil {

staging/src/k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1/doc.go

@@ -20,4 +20,5 @@ limitations under the License.
 
 // Package v1beta1 is the v1beta1 version of the API.
 // +groupName=apiextensions.k8s.io
+// +k8s:openapi-gen=true
 package v1beta1 // import "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"

staging/src/k8s.io/kube-aggregator/pkg/apiserver/apiserver.go

@@ -17,7 +17,6 @@ limitations under the License.
 package apiserver
 
 import (
-	"fmt"
 	"net/http"
 	"time"
 
@@ -27,7 +26,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/runtime/serializer"
-	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/registry/rest"
@@ -41,6 +39,7 @@ import (
 	"k8s.io/kube-aggregator/pkg/client/clientset_generated/internalclientset"
 	informers "k8s.io/kube-aggregator/pkg/client/informers/internalversion"
 	listers "k8s.io/kube-aggregator/pkg/client/listers/apiregistration/internalversion"
+	openapicontroller "k8s.io/kube-aggregator/pkg/controllers/openapi"
 	statuscontrollers "k8s.io/kube-aggregator/pkg/controllers/status"
 	apiservicestorage "k8s.io/kube-aggregator/pkg/registry/apiservice/etcd"
 )
@@ -119,7 +118,7 @@ type APIAggregator struct {
 	// Information needed to determine routing for the aggregator
 	serviceResolver ServiceResolver
 
-	openAPIAggregator *openAPIAggregator
+	openAPIAggregationController *openapicontroller.AggregationController
 }
 
 type completedConfig struct {
@@ -222,15 +221,22 @@ func (c completedConfig) NewWithDelegate(delegationTarget genericapiserver.Deleg
 	})
 
 	if openApiConfig != nil {
-		s.openAPIAggregator, err = buildAndRegisterOpenAPIAggregator(
-			s.delegateHandler,
+		specDownloader := openapicontroller.NewDownloader(s.contextMapper)
+		openAPIAggregator, err := openapicontroller.BuildAndRegisterAggregator(
+			&specDownloader,
+			delegationTarget,
 			s.GenericAPIServer.Handler.GoRestfulContainer.RegisteredWebServices(),
 			openApiConfig,
-			s.GenericAPIServer.Handler.NonGoRestfulMux,
-			s.contextMapper)
+			s.GenericAPIServer.Handler.NonGoRestfulMux)
 		if err != nil {
 			return nil, err
 		}
+		s.openAPIAggregationController = openapicontroller.NewAggregationController(&specDownloader, openAPIAggregator)
+
+		s.GenericAPIServer.AddPostStartHook("apiservice-openapi-controller", func(context genericapiserver.PostStartHookContext) error {
+			go s.openAPIAggregationController.Run(context.StopCh)
+			return nil
+		})
 	}
 
 	return s, nil
@@ -243,7 +249,10 @@ func (s *APIAggregator) AddAPIService(apiService *apiregistration.APIService) er
 	// since they are wired against listers because they require multiple resources to respond
 	if proxyHandler, exists := s.proxyHandlers[apiService.Name]; exists {
 		proxyHandler.updateAPIService(apiService)
-		return s.openAPIAggregator.loadApiServiceSpec(proxyHandler, apiService)
+		if s.openAPIAggregationController != nil {
+			s.openAPIAggregationController.UpdateAPIService(proxyHandler, apiService)
+		}
+		return nil
 	}
 
 	proxyPath := "/apis/" + apiService.Spec.Group + "/" + apiService.Spec.Version
@@ -262,8 +271,8 @@ func (s *APIAggregator) AddAPIService(apiService *apiregistration.APIService) er
 		serviceResolver: s.serviceResolver,
 	}
 	proxyHandler.updateAPIService(apiService)
-	if err := s.openAPIAggregator.loadApiServiceSpec(proxyHandler, apiService); err != nil {
-		utilruntime.HandleError(fmt.Errorf("unable to load OpenAPI spec for API service %s: %v", apiService.Name, err))
+	if s.openAPIAggregationController != nil {
+		s.openAPIAggregationController.AddAPIService(proxyHandler, apiService)
 	}
 	s.proxyHandlers[apiService.Name] = proxyHandler
 	s.GenericAPIServer.Handler.NonGoRestfulMux.Handle(proxyPath, proxyHandler)
@@ -307,6 +316,9 @@ func (s *APIAggregator) RemoveAPIService(apiServiceName string) {
 	}
 	s.GenericAPIServer.Handler.NonGoRestfulMux.Unregister(proxyPath)
 	s.GenericAPIServer.Handler.NonGoRestfulMux.Unregister(proxyPath + "/")
+	if s.openAPIAggregationController != nil {
+		s.openAPIAggregationController.RemoveAPIService(apiServiceName)
+	}
 	delete(s.proxyHandlers, apiServiceName)
 
 	// TODO unregister group level discovery when there are no more versions for the group

staging/src/k8s.io/kube-aggregator/pkg/apiserver/openapi_aggregator_test.go → staging/src/k8s.io/kube-aggregator/pkg/controllers/openapi/aggregator_test.go

@@ -14,22 +14,27 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package apiserver
+package openapi
 
 import (
+	"fmt"
+	"net/http"
 	"reflect"
 	"testing"
 
 	"github.com/go-openapi/spec"
+	"github.com/stretchr/testify/assert"
 
+	"k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/kube-aggregator/pkg/apis/apiregistration"
 )
 
-func newApiServiceForTest(name, group string, minGroupPriority, versionPriority int32) apiregistration.APIService {
+func newAPIServiceForTest(name, group string, minGroupPriority, versionPriority int32) apiregistration.APIService {
 	r := apiregistration.APIService{}
 	r.Spec.Group = group
 	r.Spec.GroupPriorityMinimum = minGroupPriority
 	r.Spec.VersionPriority = versionPriority
+	r.Spec.Service = &apiregistration.ServiceReference{}
 	r.Name = name
 	return r
 }
@@ -44,25 +49,87 @@ func assertSortedServices(t *testing.T, actual []openAPISpecInfo, expectedNames
 	}
 }
 
-func TestApiServiceSort(t *testing.T) {
+func TestAPIServiceSort(t *testing.T) {
 	list := []openAPISpecInfo{
 		{
-			apiService: newApiServiceForTest("FirstService", "Group1", 10, 5),
+			apiService: newAPIServiceForTest("FirstService", "Group1", 10, 5),
 			spec:       &spec.Swagger{},
 		},
 		{
-			apiService: newApiServiceForTest("SecondService", "Group2", 15, 3),
+			apiService: newAPIServiceForTest("SecondService", "Group2", 15, 3),
 			spec:       &spec.Swagger{},
 		},
 		{
-			apiService: newApiServiceForTest("FirstServiceInternal", "Group1", 16, 3),
+			apiService: newAPIServiceForTest("FirstServiceInternal", "Group1", 16, 3),
 			spec:       &spec.Swagger{},
 		},
 		{
-			apiService: newApiServiceForTest("ThirdService", "Group3", 15, 3),
+			apiService: newAPIServiceForTest("ThirdService", "Group3", 15, 3),
 			spec:       &spec.Swagger{},
 		},
 	}
 	sortByPriority(list)
 	assertSortedServices(t, list, []string{"FirstService", "FirstServiceInternal", "SecondService", "ThirdService"})
 }
+
+type handlerTest struct {
+	etag string
+	data []byte
+}
+
+var _ http.Handler = handlerTest{}
+
+func (h handlerTest) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	if len(h.etag) > 0 {
+		w.Header().Add("Etag", h.etag)
+	}
+	ifNoneMatches := r.Header["If-None-Match"]
+	for _, match := range ifNoneMatches {
+		if match == h.etag {
+			w.WriteHeader(http.StatusNotModified)
+			return
+		}
+	}
+	w.Write(h.data)
+}
+
+func assertDownloadedSpec(actualSpec *spec.Swagger, actualEtag string, err error,
+	expectedSpecID string, expectedEtag string) error {
+	if err != nil {
+		return fmt.Errorf("downloadOpenAPISpec failed : %s", err)
+	}
+	if expectedSpecID == "" && actualSpec != nil {
+		return fmt.Errorf("expected Not Modified, actual ID %s", actualSpec.ID)
+	}
+	if actualSpec != nil && actualSpec.ID != expectedSpecID {
+		return fmt.Errorf("expected ID %s, actual ID %s", expectedSpecID, actualSpec.ID)
+	}
+	if actualEtag != expectedEtag {
+		return fmt.Errorf("expected ETag '%s', actual ETag '%s'", expectedEtag, actualEtag)
+	}
+	return nil
+}
+
+func TestDownloadOpenAPISpec(t *testing.T) {
+
+	s := Downloader{contextMapper: request.NewRequestContextMapper()}
+
+	// Test with no eTag
+	actualSpec, actualEtag, _, err := s.Download(handlerTest{data: []byte("{\"id\": \"test\"}")}, "")
+	assert.NoError(t, assertDownloadedSpec(actualSpec, actualEtag, err, "test", "\"6E8F849B434D4B98A569B9D7718876E9-356ECAB19D7FBE1336BABB1E70F8F3025050DE218BE78256BE81620681CFC9A268508E542B8B55974E17B2184BBFC8FFFAA577E51BE195D32B3CA2547818ABE4\""))
+
+	// Test with eTag
+	actualSpec, actualEtag, _, err = s.Download(
+		handlerTest{data: []byte("{\"id\": \"test\"}"), etag: "etag_test"}, "")
+	assert.NoError(t, assertDownloadedSpec(actualSpec, actualEtag, err, "test", "etag_test"))
+
+	// Test not modified
+	actualSpec, actualEtag, _, err = s.Download(
+		handlerTest{data: []byte("{\"id\": \"test\"}"), etag: "etag_test"}, "etag_test")
+	assert.NoError(t, assertDownloadedSpec(actualSpec, actualEtag, err, "", "etag_test"))
+
+	// Test different eTags
+	actualSpec, actualEtag, _, err = s.Download(
+		handlerTest{data: []byte("{\"id\": \"test\"}"), etag: "etag_test1"}, "etag_test2")
+	assert.NoError(t, assertDownloadedSpec(actualSpec, actualEtag, err, "test", "etag_test1"))
+}

staging/src/k8s.io/kube-aggregator/pkg/controllers/openapi/controller.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 openapi
+
+import (
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/go-openapi/spec"
+	"github.com/golang/glog"
+
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/kube-aggregator/pkg/apis/apiregistration"
+)
+
+const (
+	successfulUpdateDelay   = time.Minute
+	failedUpdateMaxExpDelay = time.Hour
+)
+
+type syncAction int
+
+const (
+	syncRequeue syncAction = iota
+	syncRequeueRateLimited
+	syncNothing
+)
+
+// AggregationManager is the interface between this controller and OpenAPI Aggregator service.
+type AggregationManager interface {
+	AddUpdateAPIService(handler http.Handler, apiService *apiregistration.APIService) error
+	UpdateAPIServiceSpec(apiServiceName string, spec *spec.Swagger, etag string) error
+	RemoveAPIServiceSpec(apiServiceName string) error
+	GetAPIServiceInfo(apiServiceName string) (handler http.Handler, etag string, exists bool)
+}
+
+// AggregationController periodically check for changes in OpenAPI specs of APIServices and update/remove
+// them if necessary.
+type AggregationController struct {
+	openAPIAggregationManager AggregationManager
+	queue                     workqueue.RateLimitingInterface
+	downloader                *Downloader
+
+	// To allow injection for testing.
+	syncHandler func(key string) (syncAction, error)
+}
+
+// NewAggregationController creates new OpenAPI aggregation controller.
+func NewAggregationController(downloader *Downloader, openAPIAggregationManager AggregationManager) *AggregationController {
+	c := &AggregationController{
+		openAPIAggregationManager: openAPIAggregationManager,
+		queue: workqueue.NewNamedRateLimitingQueue(
+			workqueue.NewItemExponentialFailureRateLimiter(successfulUpdateDelay, failedUpdateMaxExpDelay), "APIServiceOpenAPIAggregationControllerQueue1"),
+		downloader: downloader,
+	}
+
+	c.syncHandler = c.sync
+
+	return c
+}
+
+// Run starts OpenAPI AggregationController
+func (c *AggregationController) Run(stopCh <-chan struct{}) {
+	defer utilruntime.HandleCrash()
+	defer c.queue.ShutDown()
+
+	glog.Infof("Starting OpenAPI AggregationController")
+	defer glog.Infof("Shutting down OpenAPI AggregationController")
+
+	go wait.Until(c.runWorker, time.Second, stopCh)
+
+	<-stopCh
+}
+
+func (c *AggregationController) runWorker() {
+	for c.processNextWorkItem() {
+	}
+}
+
+// processNextWorkItem deals with one key off the queue.  It returns false when it's time to quit.
+func (c *AggregationController) processNextWorkItem() bool {
+	key, quit := c.queue.Get()
+	defer c.queue.Done(key)
+	if quit {
+		return false
+	}
+
+	glog.Infof("OpenAPI AggregationController: Processing item %s", key)
+
+	action, err := c.syncHandler(key.(string))
+	if err == nil {
+		c.queue.Forget(key)
+	} else {
+		utilruntime.HandleError(fmt.Errorf("loading OpenAPI spec for %q failed with: %v", key, err))
+	}
+
+	switch action {
+	case syncRequeue:
+		glog.Infof("OpenAPI AggregationController: action for item %s: Requeue.", key)
+		c.queue.AddAfter(key, successfulUpdateDelay)
+	case syncRequeueRateLimited:
+		glog.Infof("OpenAPI AggregationController: action for item %s: Rate Limited Requeue.", key)
+		c.queue.AddRateLimited(key)
+	case syncNothing:
+		glog.Infof("OpenAPI AggregationController: action for item %s: Nothing (removed from the queue).", key)
+	}
+
+	return true
+}
+
+func (c *AggregationController) sync(key string) (syncAction, error) {
+	handler, etag, exists := c.openAPIAggregationManager.GetAPIServiceInfo(key)
+	if !exists || handler == nil {
+		return syncNothing, nil
+	}
+	returnSpec, newEtag, httpStatus, err := c.downloader.Download(handler, etag)
+	switch {
+	case err != nil:
+		return syncRequeueRateLimited, err
+	case httpStatus == http.StatusNotModified:
+	case httpStatus == http.StatusNotFound || returnSpec == nil:
+		return syncRequeueRateLimited, fmt.Errorf("OpenAPI spec does not exists")
+	case httpStatus == http.StatusOK:
+		if err := c.openAPIAggregationManager.UpdateAPIServiceSpec(key, returnSpec, newEtag); err != nil {
+			return syncRequeueRateLimited, err
+		}
+	}
+	return syncRequeue, nil
+}
+
+// AddAPIService adds a new API Service to OpenAPI Aggregation.
+func (c *AggregationController) AddAPIService(handler http.Handler, apiService *apiregistration.APIService) {
+	if apiService.Spec.Service == nil {
+		return
+	}
+	if err := c.openAPIAggregationManager.AddUpdateAPIService(handler, apiService); err != nil {
+		utilruntime.HandleError(fmt.Errorf("adding %q to AggregationController failed with: %v", apiService.Name, err))
+	}
+	c.queue.AddAfter(apiService.Name, time.Second)
+}
+
+// UpdateAPIService updates API Service's info and handler.
+func (c *AggregationController) UpdateAPIService(handler http.Handler, apiService *apiregistration.APIService) {
+	if apiService.Spec.Service == nil {
+		return
+	}
+	if err := c.openAPIAggregationManager.AddUpdateAPIService(handler, apiService); err != nil {
+		utilruntime.HandleError(fmt.Errorf("updating %q to AggregationController failed with: %v", apiService.Name, err))
+	}
+	key := apiService.Name
+	if c.queue.NumRequeues(key) > 0 {
+		// The item has failed before. Remove it from failure queue and
+		// update it in a second
+		c.queue.Forget(key)
+		c.queue.AddAfter(key, time.Second)
+	}
+	// Else: The item has been succeeded before and it will be updated soon (after successfulUpdateDelay)
+	// we don't add it again as it will cause a duplication of items.
+}
+
+// RemoveAPIService removes API Service from OpenAPI Aggregation Controller.
+func (c *AggregationController) RemoveAPIService(apiServiceName string) {
+	if err := c.openAPIAggregationManager.RemoveAPIServiceSpec(apiServiceName); err != nil {
+		utilruntime.HandleError(fmt.Errorf("removing %q from AggregationController failed with: %v", apiServiceName, err))
+	}
+	// This will only remove it if it was failing before. If it was successful, processNextWorkItem will figure it out
+	// and will not add it again to the queue.
+	c.queue.Forget(apiServiceName)
+}

staging/src/k8s.io/kube-aggregator/pkg/controllers/openapi/downloader.go

+/*
+Copyright 2017 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 openapi
+
+import (
+	"crypto/sha512"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strings"
+
+	"github.com/go-openapi/spec"
+
+	"k8s.io/apiserver/pkg/authentication/user"
+	"k8s.io/apiserver/pkg/endpoints/request"
+)
+
+// Downloader is the OpenAPI downloader type. It will try to download spec from /swagger.json endpoint.
+type Downloader struct {
+	contextMapper request.RequestContextMapper
+}
+
+// NewDownloader creates a new OpenAPI Downloader.
+func NewDownloader(contextMapper request.RequestContextMapper) Downloader {
+	return Downloader{contextMapper}
+}
+
+// inMemoryResponseWriter is a http.Writer that keep the response in memory.
+type inMemoryResponseWriter struct {
+	writeHeaderCalled bool
+	header            http.Header
+	respCode          int
+	data              []byte
+}
+
+func newInMemoryResponseWriter() *inMemoryResponseWriter {
+	return &inMemoryResponseWriter{header: http.Header{}}
+}
+
+func (r *inMemoryResponseWriter) Header() http.Header {
+	return r.header
+}
+
+func (r *inMemoryResponseWriter) WriteHeader(code int) {
+	r.writeHeaderCalled = true
+	r.respCode = code
+}
+
+func (r *inMemoryResponseWriter) Write(in []byte) (int, error) {
+	if !r.writeHeaderCalled {
+		r.WriteHeader(http.StatusOK)
+	}
+	r.data = append(r.data, in...)
+	return len(in), nil
+}
+
+func (r *inMemoryResponseWriter) String() string {
+	s := fmt.Sprintf("ResponseCode: %d", r.respCode)
+	if r.data != nil {
+		s += fmt.Sprintf(", Body: %s", string(r.data))
+	}
+	if r.header != nil {
+		s += fmt.Sprintf(", Header: %s", r.header)
+	}
+	return s
+}
+
+func (s *Downloader) handlerWithUser(handler http.Handler, info user.Info) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		if ctx, ok := s.contextMapper.Get(req); ok {
+			s.contextMapper.Update(req, request.WithUser(ctx, info))
+		}
+		handler.ServeHTTP(w, req)
+	})
+}
+
+func etagFor(data []byte) string {
+	return fmt.Sprintf("%s%X\"", locallyGeneratedEtagPrefix, sha512.Sum512(data))
+}
+
+// Download downloads openAPI spec from /swagger.json endpoint of the given handler.
+// httpStatus is only valid if err == nil
+func (s *Downloader) Download(handler http.Handler, etag string) (returnSpec *spec.Swagger, newEtag string, httpStatus int, err error) {
+	handler = s.handlerWithUser(handler, &user.DefaultInfo{Name: aggregatorUser})
+	handler = request.WithRequestContext(handler, s.contextMapper)
+	handler = http.TimeoutHandler(handler, specDownloadTimeout, "request timed out")
+
+	req, err := http.NewRequest("GET", "/swagger.json", nil)
+	if err != nil {
+		return nil, "", 0, err
+	}
+
+	// Only pass eTag if it is not generated locally
+	if len(etag) > 0 && !strings.HasPrefix(etag, locallyGeneratedEtagPrefix) {
+		req.Header.Add("If-None-Match", etag)
+	}
+
+	writer := newInMemoryResponseWriter()
+	handler.ServeHTTP(writer, req)
+
+	switch writer.respCode {
+	case http.StatusNotModified:
+		if len(etag) == 0 {
+			return nil, etag, http.StatusNotModified, fmt.Errorf("http.StatusNotModified is not allowed in absence of etag")
+		}
+		return nil, etag, http.StatusNotModified, nil
+	case http.StatusNotFound:
+		// Gracefully skip 404, assuming the server won't provide any spec
+		return nil, "", http.StatusNotFound, nil
+	case http.StatusOK:
+		openAPISpec := &spec.Swagger{}
+		if err := json.Unmarshal(writer.data, openAPISpec); err != nil {
+			return nil, "", 0, err
+		}
+		newEtag = writer.Header().Get("Etag")
+		if len(newEtag) == 0 {
+			newEtag = etagFor(writer.data)
+			if len(etag) > 0 && strings.HasPrefix(etag, locallyGeneratedEtagPrefix) {
+				// The function call with an etag and server does not report an etag.
+				// That means this server does not support etag and the etag that passed
+				// to the function generated previously by us. Just compare etags and
+				// return StatusNotModified if they are the same.
+				if etag == newEtag {
+					return nil, etag, http.StatusNotModified, nil
+				}
+			}
+		}
+		return openAPISpec, newEtag, http.StatusOK, nil
+	default:
+		return nil, "", 0, fmt.Errorf("failed to retrieve openAPI spec, http error: %s", writer.String())
+	}
+}