Merge pull request kubernetes#120202 from sttts/sttts-controlplane-config-split

Step 2 – generic controlplane: split server

Author: k8s-ci-robot

cmd/kube-apiserver/app/server.go

@@ -27,7 +27,6 @@ import (
 	"os"
 
 	"github.com/spf13/cobra"
-
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	extensionsapiserver "k8s.io/apiextensions-apiserver/pkg/apiserver"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -185,7 +184,7 @@ func CreateServerChain(config CompletedConfig) (*aggregatorapiserver.APIAggregat
 	}
 
 	// aggregator comes last in the chain
-	aggregatorServer, err := createAggregatorServer(config.Aggregator, kubeAPIServer.GenericAPIServer, apiExtensionsServer.Informers, crdAPIEnabled)
+	aggregatorServer, err := createAggregatorServer(config.Aggregator, kubeAPIServer.ControlPlane.GenericAPIServer, apiExtensionsServer.Informers, crdAPIEnabled)
 	if err != nil {
 		// we don't need special handling for innerStopCh because the aggregator server doesn't create any go routines
 		return nil, err

pkg/controlplane/apiserver/apis.go

@@ -0,0 +1,91 @@
+/*
+Copyright 2023 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 apiserver
+
+import (
+	"fmt"
+
+	"k8s.io/apiserver/pkg/registry/generic"
+	genericapiserver "k8s.io/apiserver/pkg/server"
+	serverstorage "k8s.io/apiserver/pkg/server/storage"
+	"k8s.io/klog/v2"
+)
+
+// RESTStorageProvider is a factory type for REST storage.
+type RESTStorageProvider interface {
+	GroupName() string
+	NewRESTStorage(apiResourceConfigSource serverstorage.APIResourceConfigSource, restOptionsGetter generic.RESTOptionsGetter) (genericapiserver.APIGroupInfo, error)
+}
+
+// InstallAPIs will install the APIs for the restStorageProviders if they are enabled.
+func (s *Server) InstallAPIs(restStorageProviders ...RESTStorageProvider) error {
+	nonLegacy := []*genericapiserver.APIGroupInfo{}
+
+	// used later in the loop to filter the served resource by those that have expired.
+	resourceExpirationEvaluator, err := genericapiserver.NewResourceExpirationEvaluator(*s.GenericAPIServer.Version)
+	if err != nil {
+		return err
+	}
+
+	for _, restStorageBuilder := range restStorageProviders {
+		groupName := restStorageBuilder.GroupName()
+		apiGroupInfo, err := restStorageBuilder.NewRESTStorage(s.APIResourceConfigSource, s.RESTOptionsGetter)
+		if err != nil {
+			return fmt.Errorf("problem initializing API group %q: %w", groupName, err)
+		}
+		if len(apiGroupInfo.VersionedResourcesStorageMap) == 0 {
+			// If we have no storage for any resource configured, this API group is effectively disabled.
+			// This can happen when an entire API group, version, or development-stage (alpha, beta, GA) is disabled.
+			klog.Infof("API group %q is not enabled, skipping.", groupName)
+			continue
+		}
+
+		// Remove resources that serving kinds that are removed.
+		// We do this here so that we don't accidentally serve versions without resources or openapi information that for kinds we don't serve.
+		// This is a spot above the construction of individual storage handlers so that no sig accidentally forgets to check.
+		resourceExpirationEvaluator.RemoveDeletedKinds(groupName, apiGroupInfo.Scheme, apiGroupInfo.VersionedResourcesStorageMap)
+		if len(apiGroupInfo.VersionedResourcesStorageMap) == 0 {
+			klog.V(1).Infof("Removing API group %v because it is time to stop serving it because it has no versions per APILifecycle.", groupName)
+			continue
+		}
+
+		klog.V(1).Infof("Enabling API group %q.", groupName)
+
+		if postHookProvider, ok := restStorageBuilder.(genericapiserver.PostStartHookProvider); ok {
+			name, hook, err := postHookProvider.PostStartHook()
+			if err != nil {
+				return fmt.Errorf("error building PostStartHook: %w", err)
+			}
+			s.GenericAPIServer.AddPostStartHookOrDie(name, hook)
+		}
+
+		if len(groupName) == 0 {
+			// the legacy group for core APIs is special that it is installed into /api via this special install method.
+			if err := s.GenericAPIServer.InstallLegacyAPIGroup(genericapiserver.DefaultLegacyAPIPrefix, &apiGroupInfo); err != nil {
+				return fmt.Errorf("error in registering legacy API: %w", err)
+			}
+		} else {
+			// everything else goes to /apis
+			nonLegacy = append(nonLegacy, &apiGroupInfo)
+		}
+	}
+
+	if err := s.GenericAPIServer.InstallAPIGroups(nonLegacy...); err != nil {
+		return fmt.Errorf("error in registering group versions: %w", err)
+	}
+	return nil
+}

pkg/controlplane/apiserver/peer.go

@@ -18,6 +18,8 @@ package apiserver
 
 import (
 	"fmt"
+	"net"
+	"strconv"
 	"time"
 
 	"k8s.io/apimachinery/pkg/runtime"
@@ -87,3 +89,13 @@ func CreatePeerEndpointLeaseReconciler(c genericapiserver.Config, storageFactory
 	reconciler, err := reconcilers.NewPeerEndpointLeaseReconciler(config, "/peerserverleases/", ttl)
 	return reconciler, err
 }
+
+// utility function to get the apiserver address that is used by peer apiservers to proxy
+// requests to this apiserver in case the peer is incapable of serving the request
+func getPeerAddress(peerAdvertiseAddress reconcilers.PeerAdvertiseAddress, publicAddress net.IP, publicServicePort int) string {
+	if peerAdvertiseAddress.PeerAdvertiseIP != "" && peerAdvertiseAddress.PeerAdvertisePort != "" {
+		return net.JoinHostPort(peerAdvertiseAddress.PeerAdvertiseIP, peerAdvertiseAddress.PeerAdvertisePort)
+	} else {
+		return net.JoinHostPort(publicAddress.String(), strconv.Itoa(publicServicePort))
+	}
+}