Enable aggregator apiserver resyncing openapi spec

from delegation apiservers

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

@@ -53,12 +53,19 @@ type specAggregator struct {
 	// provided for dynamic OpenAPI spec
 	openAPIService          *handler.OpenAPIService
 	openAPIVersionedService *handler.OpenAPIService
+
+	// initialized is set to be true at the end of startup. All local specs
+	// must be registered before initialized is set, we panic otherwise.
+	initialized bool
 }
 
 var _ AggregationManager = &specAggregator{}
 
 // This function is not thread safe as it only being called on startup.
 func (s *specAggregator) addLocalSpec(spec *spec.Swagger, localHandler http.Handler, name, etag string) {
+	if s.initialized {
+		panic("Local spec must not be added after startup")
+	}
 	localAPIService := apiregistration.APIService{}
 	localAPIService.Name = name
 	s.openAPISpecs[name] = &openAPISpecInfo{
@@ -69,6 +76,17 @@ func (s *specAggregator) addLocalSpec(spec *spec.Swagger, localHandler http.Hand
 	}
 }
 
+// GetAPIServicesName returns the names of APIServices recorded in specAggregator.openAPISpecs.
+// We use this function to pass the names of local APIServices to the controller in this package,
+// so that the controller can periodically sync the OpenAPI spec from delegation API servers.
+func (s *specAggregator) GetAPIServiceNames() []string {
+	names := make([]string, len(s.openAPISpecs))
+	for key := range s.openAPISpecs {
+		names = append(names, key)
+	}
+	return names
+}
+
 // BuildAndRegisterAggregator registered OpenAPI aggregator handler. This function is not thread safe as it only being called on startup.
 func BuildAndRegisterAggregator(downloader *Downloader, delegationTarget server.DelegationTarget, webServices []*restful.WebService,
 	config *common.Config, pathHandler common.PathHandler) (AggregationManager, error) {
@@ -124,6 +142,9 @@ func BuildAndRegisterAggregator(downloader *Downloader, delegationTarget server.
 		return nil, err
 	}
 
+	// We set initialized to be true to forbid any future local spec addition
+	s.initialized = true
+
 	return s, nil
 }
 

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

@@ -49,6 +49,9 @@ type AggregationManager interface {
 	UpdateAPIServiceSpec(apiServiceName string, spec *spec.Swagger, etag string) error
 	RemoveAPIServiceSpec(apiServiceName string) error
 	GetAPIServiceInfo(apiServiceName string) (handler http.Handler, etag string, exists bool)
+
+	// GetAPIServicesName returns the names of APIServices recorded in AggregationManager.
+	GetAPIServiceNames() []string
 }
 
 // AggregationController periodically check for changes in OpenAPI specs of APIServices and update/remove
@@ -72,6 +75,18 @@ func NewAggregationController(downloader *Downloader, openAPIAggregationManager
 	}
 
 	c.syncHandler = c.sync
+	// During initialization, openAPIAggregationManager only has record of local APIServices. There must be
+	// no aggregated APIService recorded, because aggregated APIServices only get added to openAPIAggregationManager
+	// by calling AggregationController.AddAPIService or AggregationController.UpdateAPIService after the
+	// controller is initialized.
+	// Here we add delegation target API services to queue, to periodically sync dynamic OpenAPI spec from
+	// delegation target.
+	// NOTE: openAPIAggregationManager.GetAPIServiceNames() will also return the APIService of non-name spec
+	// for aggregator, which has no http.Handler. The first time sync (when popping off from queue) for
+	// this APIService will be a no-op, and the controller will drop the APIService from queue.
+	for _, name := range openAPIAggregationManager.GetAPIServiceNames() {
+		c.queue.AddAfter(name, time.Second)
+	}
 
 	return c
 }