Aggregate OpenAPI specs

cmd/kube-apiserver/app/aggregator.go

@@ -50,7 +50,6 @@ func createAggregatorConfig(kubeAPIServerConfig genericapiserver.Config, command
 
 	// the aggregator doesn't wire these up.  It just delegates them to the kubeapiserver
 	genericConfig.EnableSwaggerUI = false
-	genericConfig.OpenAPIConfig = nil
 	genericConfig.SwaggerConfig = nil
 
 	// copy the etcd options so we don't mutate originals.

staging/src/k8s.io/apiserver/pkg/server/genericapiserver.go

@@ -28,6 +28,7 @@ import (
 	"github.com/emicklei/go-restful-swagger12"
 	"github.com/golang/glog"
 
+	"github.com/go-openapi/spec"
 	"k8s.io/apimachinery/pkg/apimachinery"
 	"k8s.io/apimachinery/pkg/apimachinery/registered"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -43,6 +44,7 @@ import (
 	apirequest "k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/registry/rest"
 	"k8s.io/apiserver/pkg/server/healthz"
+	"k8s.io/apiserver/pkg/server/openapi"
 	"k8s.io/apiserver/pkg/server/routes"
 	restclient "k8s.io/client-go/rest"
 )
@@ -130,6 +132,9 @@ type GenericAPIServer struct {
 	swaggerConfig *swagger.Config
 	openAPIConfig *openapicommon.Config
 
+	// Enables updating OpenAPI spec using update method.
+	OpenAPIService *openapi.OpenAPIService
+
 	// PostStartHooks are each called after the server has started listening, in a separate go func for each
 	// with no guarantee of ordering between them.  The map key is a name used for error reporting.
 	// It may kill the process with a panic if it wishes to by returning an error.
@@ -165,6 +170,9 @@ type DelegationTarget interface {
 
 	// ListedPaths returns the paths for supporting an index
 	ListedPaths() []string
+
+	// OpenAPISpec returns the OpenAPI spec of the delegation target if exists, nil otherwise.
+	OpenAPISpec() *spec.Swagger
 }
 
 func (s *GenericAPIServer) UnprotectedHandler() http.Handler {
@@ -180,6 +188,9 @@ func (s *GenericAPIServer) HealthzChecks() []healthz.HealthzChecker {
 func (s *GenericAPIServer) ListedPaths() []string {
 	return s.listedPathProvider.ListedPaths()
 }
+func (s *GenericAPIServer) OpenAPISpec() *spec.Swagger {
+	return s.OpenAPIService.GetSpec()
+}
 
 var EmptyDelegate = emptyDelegate{
 	requestContextMapper: apirequest.NewRequestContextMapper(),
@@ -204,6 +215,9 @@ func (s emptyDelegate) ListedPaths() []string {
 func (s emptyDelegate) RequestContextMapper() apirequest.RequestContextMapper {
 	return s.requestContextMapper
 }
+func (s emptyDelegate) OpenAPISpec() *spec.Swagger {
+	return nil
+}
 
 func init() {
 	// Send correct mime type for .svg files.
@@ -233,17 +247,22 @@ func (s *GenericAPIServer) PrepareRun() preparedGenericAPIServer {
 	if s.swaggerConfig != nil {
 		routes.Swagger{Config: s.swaggerConfig}.Install(s.Handler.GoRestfulContainer)
 	}
-	if s.openAPIConfig != nil {
-		routes.OpenAPI{
-			Config: s.openAPIConfig,
-		}.Install(s.Handler.GoRestfulContainer, s.Handler.NonGoRestfulMux)
-	}
+	s.PrepareOpenAPIService()
 
 	s.installHealthz()
 
 	return preparedGenericAPIServer{s}
 }
 
+// PrepareOpenAPIService installs OpenAPI handler if it does not exists.
+func (s *GenericAPIServer) PrepareOpenAPIService() {
+	if s.openAPIConfig != nil && s.OpenAPIService == nil {
+		s.OpenAPIService = routes.OpenAPI{
+			Config: s.openAPIConfig,
+		}.Install(s.Handler.GoRestfulContainer, s.Handler.NonGoRestfulMux)
+	}
+}
+
 // Run spawns the secure http server. It only returns if stopCh is closed
 // or the secure port cannot be listened on initially.
 func (s preparedGenericAPIServer) Run(stopCh <-chan struct{}) error {

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

@@ -17,8 +17,11 @@ limitations under the License.
 package apiserver
 
 import (
+	"context"
+	"encoding/json"
 	"net/http"
 	"net/url"
+	"sync"
 	"time"
 
 	"k8s.io/apimachinery/pkg/apimachinery/announced"
@@ -37,6 +40,14 @@ import (
 	listersv1 "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/pkg/version"
 
+	"bytes"
+	"fmt"
+	"github.com/go-openapi/spec"
+	"github.com/golang/glog"
+	"github.com/pkg/errors"
+	"io"
+	"k8s.io/apiserver/pkg/server/openapi"
+	"k8s.io/client-go/transport"
 	"k8s.io/kube-aggregator/pkg/apis/apiregistration"
 	"k8s.io/kube-aggregator/pkg/apis/apiregistration/install"
 	"k8s.io/kube-aggregator/pkg/apis/apiregistration/v1beta1"
@@ -54,6 +65,10 @@ var (
 	Codecs               = serializer.NewCodecFactory(Scheme)
 )
 
+const (
+	LOAD_OPENAPI_SPEC_MAX_RETRIES = 10
+)
+
 func init() {
 	install.Install(groupFactoryRegistry, registry, Scheme)
 
@@ -118,6 +133,24 @@ type APIAggregator struct {
 	// handledGroups are the groups that already have routes
 	handledGroups sets.String
 
+	// Swagger spec for each api service
+	apiServiceSpecs map[string]*spec.Swagger
+
+	// List of the specs that needs to be loaded. When a spec is successfully loaded
+	// it will be removed from this list and added to apiServiceSpecs.
+	// Map values are retry counts. After a preset retries, it will stop
+	// trying.
+	toLoadAPISpec map[string]int
+
+	// protecting toLoadAPISpec and apiServiceSpecs
+	specMutex sync.Mutex
+
+	// rootSpec is the OpenAPI spec of the Aggregator server.
+	rootSpec *spec.Swagger
+
+	// delegationSpec is the delegation API Server's spec (most of API groups are in this spec).
+	delegationSpec *spec.Swagger
+
 	// lister is used to add group handling for /apis/<group> aggregator lookups based on
 	// controller state
 	lister listers.APIServiceLister
@@ -191,11 +224,14 @@ func (c completedConfig) NewWithDelegate(delegationTarget genericapiserver.Deleg
 	s := &APIAggregator{
 		GenericAPIServer: genericServer,
 		delegateHandler:  delegationTarget.UnprotectedHandler(),
+		delegationSpec:   delegationTarget.OpenAPISpec(),
 		contextMapper:    c.GenericConfig.RequestContextMapper,
 		proxyClientCert:  c.ProxyClientCert,
 		proxyClientKey:   c.ProxyClientKey,
 		proxyTransport:   proxyTransport,
 		proxyHandlers:    map[string]*proxyHandler{},
+		apiServiceSpecs:  map[string]*spec.Swagger{},
+		toLoadAPISpec:    map[string]int{},
 		handledGroups:    sets.String{},
 		lister:           informerFactory.Apiregistration().InternalVersion().APIServices().Lister(),
 		APIRegistrationInformers: informerFactory,
@@ -244,6 +280,20 @@ func (c completedConfig) NewWithDelegate(delegationTarget genericapiserver.Deleg
 		return nil
 	})
 
+	s.GenericAPIServer.PrepareOpenAPIService()
+
+	if s.GenericAPIServer.OpenAPIService != nil {
+		s.rootSpec = s.GenericAPIServer.OpenAPIService.GetSpec()
+		if err := s.updateOpenAPISpec(); err != nil {
+			return nil, err
+		}
+		s.GenericAPIServer.OpenAPIService.AddUpdateHook(func(r *http.Request) {
+			if s.tryLoadingOpenAPISpecs(r) {
+				s.updateOpenAPISpec()
+			}
+		})
+	}
+
 	return s, nil
 }
 
@@ -274,6 +324,9 @@ func (s *APIAggregator) AddAPIService(apiService *apiregistration.APIService) {
 	}
 	proxyHandler.updateAPIService(apiService)
 	s.proxyHandlers[apiService.Name] = proxyHandler
+
+	s.deferLoadAPISpec(apiService.Name)
+
 	s.GenericAPIServer.Handler.NonGoRestfulMux.Handle(proxyPath, proxyHandler)
 	s.GenericAPIServer.Handler.NonGoRestfulMux.UnlistedHandlePrefix(proxyPath+"/", proxyHandler)
 
@@ -302,6 +355,19 @@ func (s *APIAggregator) AddAPIService(apiService *apiregistration.APIService) {
 	s.handledGroups.Insert(apiService.Spec.Group)
 }
 
+func (s *APIAggregator) deferLoadAPISpec(name string) {
+	s.specMutex.Lock()
+	defer s.specMutex.Unlock()
+	s.toLoadAPISpec[name] = 0
+}
+
+func (s *APIAggregator) deleteApiSpec(name string) {
+	s.specMutex.Lock()
+	defer s.specMutex.Unlock()
+	delete(s.apiServiceSpecs, name)
+	delete(s.toLoadAPISpec, name)
+}
+
 // RemoveAPIService removes the APIService from being handled.  It is not thread-safe, so only call it on one thread at a time please.
 // It's a slow moving API, so its ok to run the controller on a single thread.
 func (s *APIAggregator) RemoveAPIService(apiServiceName string) {
@@ -315,7 +381,124 @@ func (s *APIAggregator) RemoveAPIService(apiServiceName string) {
 	s.GenericAPIServer.Handler.NonGoRestfulMux.Unregister(proxyPath)
 	s.GenericAPIServer.Handler.NonGoRestfulMux.Unregister(proxyPath + "/")
 	delete(s.proxyHandlers, apiServiceName)
+	s.deleteApiSpec(apiServiceName)
+	s.updateOpenAPISpec()
 
 	// TODO unregister group level discovery when there are no more versions for the group
 	// We don't need this right away because the handler properly delegates when no versions are present
 }
+
+func (_ *APIAggregator) loadOpenAPISpec(p *proxyHandler, r *http.Request) (*spec.Swagger, error) {
+	value := p.handlingInfo.Load()
+	if value == nil {
+		return nil, nil
+	}
+	handlingInfo := value.(proxyHandlingInfo)
+	if handlingInfo.local {
+		return nil, nil
+	}
+	loc, err := p.routing.ResolveEndpoint(handlingInfo.serviceNamespace, handlingInfo.serviceName)
+	if err != nil {
+		return nil, fmt.Errorf("missing route")
+	}
+	host := loc.Host
+
+	var w io.Reader
+	req, err := http.NewRequest("GET", "/swagger.json", w)
+	if err != nil {
+		return nil, err
+	}
+	req.URL.Scheme = "https"
+	req.URL.Host = host
+
+	req = req.WithContext(context.Background())
+	// Get user from the original request
+	ctx, ok := p.contextMapper.Get(r)
+	if !ok {
+		return nil, fmt.Errorf("missing context")
+	}
+	user, ok := genericapirequest.UserFrom(ctx)
+	if !ok {
+		return nil, fmt.Errorf("missing user")
+	}
+	proxyRoundTripper := transport.NewAuthProxyRoundTripper(user.GetName(), user.GetGroups(), user.GetExtra(), handlingInfo.proxyRoundTripper)
+	res, err := proxyRoundTripper.RoundTrip(req)
+	if err != nil {
+		return nil, err
+	}
+	if res.StatusCode != http.StatusOK {
+		return nil, errors.New(res.Status)
+	}
+	buf := new(bytes.Buffer)
+	buf.ReadFrom(res.Body)
+	bytes := buf.Bytes()
+	var s spec.Swagger
+	if err := json.Unmarshal(bytes, &s); err != nil {
+		return nil, err
+	}
+	return &s, nil
+}
+
+// Returns true if any Spec is loaded
+func (s *APIAggregator) tryLoadingOpenAPISpecs(r *http.Request) bool {
+	s.specMutex.Lock()
+	defer s.specMutex.Unlock()
+	if len(s.toLoadAPISpec) == 0 {
+		return false
+	}
+	loaded := false
+	newList := map[string]int{}
+	for name, retries := range s.toLoadAPISpec {
+		if retries >= LOAD_OPENAPI_SPEC_MAX_RETRIES {
+			continue
+		}
+		proxyHandler := s.proxyHandlers[name]
+		if spec, err := s.loadOpenAPISpec(proxyHandler, r); err != nil {
+			glog.Warningf("Failed to Load OpenAPI spec (try %d of %d) for %s, err=%s", retries+1, LOAD_OPENAPI_SPEC_MAX_RETRIES, name, err)
+			newList[name] = retries + 1
+		} else if spec != nil {
+			s.apiServiceSpecs[name] = spec
+			loaded = true
+		}
+		s.toLoadAPISpec = newList
+	}
+	return loaded
+}
+
+func (s *APIAggregator) updateOpenAPISpec() error {
+	s.specMutex.Lock()
+	defer s.specMutex.Unlock()
+	if s.GenericAPIServer.OpenAPIService == nil {
+		return nil
+	}
+	sp, err := openapi.CloneSpec(s.rootSpec)
+	if err != nil {
+		return err
+	}
+	openapi.FilterSpecByPaths(sp, []string{"/apis/apiregistration.k8s.io/"})
+	if _, found := sp.Paths.Paths["/version/"]; found {
+		return fmt.Errorf("Cleanup didn't work")
+	}
+	if err := openapi.MergeSpecs(sp, s.delegationSpec); err != nil {
+		return err
+	}
+
+	for k, v := range s.apiServiceSpecs {
+		version := apiregistration.APIServiceNameToGroupVersion(k)
+
+		proxyPath := "/apis/" + version.Group + "/"
+		// v1. is a special case for the legacy API.  It proxies to a wider set of endpoints.
+		if k == legacyAPIServiceName {
+			proxyPath = "/api/"
+		}
+		spc, err := openapi.CloneSpec(v)
+		if err != nil {
+			return err
+		}
+		openapi.FilterSpecByPaths(spc, []string{proxyPath})
+		if err := openapi.MergeSpecs(sp, spc); err != nil {
+			return err
+		}
+	}
+	return s.GenericAPIServer.OpenAPIService.UpdateSpec(sp)
+}