wire in aggregator

Author: deads2k

pkg/cmd/server/api/validation/master.go

@@ -194,7 +194,7 @@ func ValidateMasterConfig(config *api.MasterConfig, fldPath *field.Path) Validat
 	validationResults.Append(ValidateControllerConfig(config.ControllerConfig, fldPath.Child("controllerConfig")))
 	validationResults.Append(ValidateAuditConfig(config.AuditConfig, fldPath.Child("auditConfig")))
 	validationResults.Append(ValidateMasterAuthConfig(config.AuthConfig, fldPath.Child("authConfig")))
-	validationResults.Append(ValidateAggregatorConfig(config.AuthConfig, fldPath.Child("aggregatorConfig")))
+	validationResults.Append(ValidateAggregatorConfig(config.AggregatorConfig, fldPath.Child("aggregatorConfig")))
 
 	return validationResults
 }

pkg/cmd/server/origin/aggregator.go

@@ -0,0 +1,171 @@
+/*
+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 app does all of the work necessary to create a Kubernetes
+// APIServer by binding together the API, master and APIServer infrastructure.
+// It can be configured and called directly or via the hyperkube framework.
+package origin
+
+import (
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"strings"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	genericapiserver "k8s.io/apiserver/pkg/server"
+	"k8s.io/apiserver/pkg/server/healthz"
+	kubeclientset "k8s.io/client-go/kubernetes"
+	"k8s.io/kube-aggregator/pkg/apis/apiregistration"
+	"k8s.io/kube-aggregator/pkg/apis/apiregistration/install"
+	aggregatorapiserver "k8s.io/kube-aggregator/pkg/apiserver"
+	apiregistrationclient "k8s.io/kube-aggregator/pkg/client/clientset_generated/internalclientset/typed/apiregistration/internalversion"
+	"k8s.io/kube-aggregator/pkg/controllers/autoregister"
+	kapi "k8s.io/kubernetes/pkg/api"
+	informers "k8s.io/kubernetes/pkg/client/informers/informers_generated/internalversion"
+)
+
+func (c *MasterConfig) createAggregatorConfig(kubeAPIServerConfig genericapiserver.Config) (*aggregatorapiserver.Config, error) {
+	// make a shallow copy to let us twiddle a few things
+	// most of the config actually remains the same.  We only need to mess with a couple items related to the particulars of the aggregator
+	genericConfig := kubeAPIServerConfig
+
+	// the aggregator doesn't wire these up.  It just delegates them to the kubeapiserver
+	genericConfig.EnableSwaggerUI = false
+	genericConfig.OpenAPIConfig = nil
+	genericConfig.SwaggerConfig = nil
+	genericConfig.FallThroughHandler = nil
+
+	// install our types into the scheme so that "normal" RESTOptionsGetters can work for us
+	install.Install(kapi.GroupFactoryRegistry, kapi.Registry, kapi.Scheme)
+
+	client, err := kubeclientset.NewForConfig(genericConfig.LoopbackClientConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	certBytes, err := ioutil.ReadFile(c.Options.AggregatorConfig.ProxyClientInfo.CertFile)
+	if err != nil {
+		return nil, err
+	}
+	keyBytes, err := ioutil.ReadFile(c.Options.AggregatorConfig.ProxyClientInfo.KeyFile)
+	if err != nil {
+		return nil, err
+	}
+	return &aggregatorapiserver.Config{
+		GenericConfig:       &genericConfig,
+		CoreAPIServerClient: client,
+		ProxyClientCert:     certBytes,
+		ProxyClientKey:      keyBytes,
+	}, nil
+}
+
+func createAggregatorServer(aggregatorConfig *aggregatorapiserver.Config, delegateAPIServer genericapiserver.DelegationTarget, sharedInformers informers.SharedInformerFactory, stopCh <-chan struct{}) (*aggregatorapiserver.APIAggregator, error) {
+	aggregatorServer, err := aggregatorConfig.Complete().NewWithDelegate(delegateAPIServer, stopCh)
+	if err != nil {
+		return nil, err
+	}
+
+	// create controllers for auto-registration
+	apiRegistrationClient, err := apiregistrationclient.NewForConfig(aggregatorConfig.GenericConfig.LoopbackClientConfig)
+	if err != nil {
+		return nil, err
+	}
+	autoRegistrationController := autoregister.NewAutoRegisterController(aggregatorServer.APIRegistrationInformers.Apiregistration().InternalVersion().APIServices(), apiRegistrationClient)
+	apiServices := apiServicesToRegister(delegateAPIServer, autoRegistrationController)
+
+	aggregatorServer.GenericAPIServer.AddPostStartHook("kube-apiserver-autoregistration", func(context genericapiserver.PostStartHookContext) error {
+		go autoRegistrationController.Run(5, stopCh)
+		return nil
+	})
+	aggregatorServer.GenericAPIServer.AddHealthzChecks(healthz.NamedCheck("autoregister-completion", func(r *http.Request) error {
+		items, err := aggregatorServer.APIRegistrationInformers.Apiregistration().InternalVersion().APIServices().Lister().List(labels.Everything())
+		if err != nil {
+			return err
+		}
+
+		missing := []apiregistration.APIService{}
+		for _, apiService := range apiServices {
+			found := false
+			for _, item := range items {
+				if item.Name == apiService.Name {
+					found = true
+					break
+				}
+			}
+
+			if !found {
+				missing = append(missing, *apiService)
+			}
+		}
+
+		if len(missing) > 0 {
+			return fmt.Errorf("missing APIService: %v", missing)
+		}
+		return nil
+	}))
+
+	return aggregatorServer, nil
+}
+
+func makeAPIService(gv schema.GroupVersion) *apiregistration.APIService {
+	return &apiregistration.APIService{
+		ObjectMeta: metav1.ObjectMeta{Name: gv.Version + "." + gv.Group},
+		Spec: apiregistration.APIServiceSpec{
+			Group:    gv.Group,
+			Version:  gv.Version,
+			Priority: 100,
+		},
+	}
+}
+
+func apiServicesToRegister(delegateAPIServer genericapiserver.DelegationTarget, registration autoregister.AutoAPIServiceRegistration) []*apiregistration.APIService {
+	apiServices := []*apiregistration.APIService{}
+
+	for _, curr := range delegateAPIServer.ListedPaths() {
+		if curr == "/api/v1" {
+			apiService := makeAPIService(schema.GroupVersion{Group: "", Version: "v1"})
+			registration.AddAPIServiceToSync(apiService)
+			apiServices = append(apiServices, apiService)
+			continue
+		}
+
+		if !strings.HasPrefix(curr, "/apis/") {
+			continue
+		}
+		// this comes back in a list that looks like /apis/rbac.authorization.k8s.io/v1alpha1
+		tokens := strings.Split(curr, "/")
+		if len(tokens) != 4 {
+			continue
+		}
+
+		apiService := makeAPIService(schema.GroupVersion{Group: tokens[2], Version: tokens[3]})
+
+		// TODO this is probably an indication that we need explicit and precise control over the discovery chain
+		// but for now its a special case
+		// apps has to come last for compatibility with 1.5 kubectl clients
+		if apiService.Spec.Group == "apps" {
+			apiService.Spec.Priority = 110
+		}
+
+		registration.AddAPIServiceToSync(apiService)
+		apiServices = append(apiServices, apiService)
+	}
+
+	return apiServices
+}

pkg/cmd/server/origin/master.go

@@ -172,7 +172,7 @@ var excludedV1Types = sets.NewString()
 
 // Run launches the OpenShift master by creating a kubernetes master, installing
 // OpenShift APIs into it and then running it.
-func (c *MasterConfig) Run(kc *kubernetes.MasterConfig, assetConfig *AssetConfig) {
+func (c *MasterConfig) Run(kc *kubernetes.MasterConfig, assetConfig *AssetConfig, stopCh <-chan struct{}) {
 	var (
 		messages []string
 		err      error
@@ -193,7 +193,28 @@ func (c *MasterConfig) Run(kc *kubernetes.MasterConfig, assetConfig *AssetConfig
 	for _, s := range messages {
 		glog.Infof(s, c.Options.ServingInfo.BindAddress)
 	}
-	go kmaster.GenericAPIServer.PrepareRun().Run(utilwait.NeverStop)
+
+	apiserver := kmaster.GenericAPIServer.PrepareRun()
+
+	// presence of the key indicates whether or not to enable the aggregator
+	if len(c.Options.AggregatorConfig.ProxyClientInfo.KeyFile) == 0 {
+		go apiserver.Run(utilwait.NeverStop)
+
+		// Attempt to verify the server came up for 20 seconds (100 tries * 100ms, 100ms timeout per try)
+		cmdutil.WaitForSuccessfulDial(c.TLS, c.Options.ServingInfo.BindNetwork, c.Options.ServingInfo.BindAddress, 100*time.Millisecond, 100*time.Millisecond, 100)
+		return
+	}
+
+	aggregatorConfig, err := c.createAggregatorConfig(*kc.Master.GenericConfig)
+	if err != nil {
+		glog.Fatalf("Failed to launch master: %v", err)
+	}
+	aggregatorServer, err := createAggregatorServer(aggregatorConfig, apiserver.GenericAPIServer, kc.Informers.InternalKubernetesInformers(), stopCh)
+	if err != nil {
+		// we don't need special handling for innerStopCh because the aggregator server doesn't create any go routines
+		glog.Fatalf("Failed to launch master: %v", err)
+	}
+	go aggregatorServer.GenericAPIServer.PrepareRun().Run(stopCh)
 
 	// Attempt to verify the server came up for 20 seconds (100 tries * 100ms, 100ms timeout per try)
 	cmdutil.WaitForSuccessfulDial(c.TLS, c.Options.ServingInfo.BindNetwork, c.Options.ServingInfo.BindAddress, 100*time.Millisecond, 100*time.Millisecond, 100)

pkg/cmd/server/start/start_master.go

@@ -507,7 +507,7 @@ func StartAPI(oc *origin.MasterConfig, kc *kubernetes.MasterConfig) error {
 		}
 	}
 
-	oc.Run(kc, embeddedAssetConfig)
+	oc.Run(kc, embeddedAssetConfig, utilwait.NeverStop)
 
 	// start DNS before the informers are started because it adds a ClusterIP index.
 	if oc.Options.DNSConfig != nil {