Matthias Andreas Benkard | 832a54e | 2019-01-29 09:27:38 +0100 | [diff] [blame^] | 1 | /* |
| 2 | Copyright 2014 The Kubernetes Authors. |
| 3 | |
| 4 | Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | you may not use this file except in compliance with the License. |
| 6 | You may obtain a copy of the License at |
| 7 | |
| 8 | http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | |
| 10 | Unless required by applicable law or agreed to in writing, software |
| 11 | distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | See the License for the specific language governing permissions and |
| 14 | limitations under the License. |
| 15 | */ |
| 16 | |
| 17 | package server |
| 18 | |
| 19 | import ( |
| 20 | "fmt" |
| 21 | "net/http" |
| 22 | "strings" |
| 23 | "sync" |
| 24 | "time" |
| 25 | |
| 26 | systemd "github.com/coreos/go-systemd/daemon" |
| 27 | "github.com/emicklei/go-restful-swagger12" |
| 28 | "github.com/golang/glog" |
| 29 | |
| 30 | "k8s.io/apimachinery/pkg/api/meta" |
| 31 | metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" |
| 32 | "k8s.io/apimachinery/pkg/runtime" |
| 33 | "k8s.io/apimachinery/pkg/runtime/schema" |
| 34 | "k8s.io/apimachinery/pkg/runtime/serializer" |
| 35 | "k8s.io/apimachinery/pkg/util/sets" |
| 36 | utilwaitgroup "k8s.io/apimachinery/pkg/util/waitgroup" |
| 37 | "k8s.io/apiserver/pkg/admission" |
| 38 | "k8s.io/apiserver/pkg/audit" |
| 39 | genericapi "k8s.io/apiserver/pkg/endpoints" |
| 40 | "k8s.io/apiserver/pkg/endpoints/discovery" |
| 41 | "k8s.io/apiserver/pkg/registry/rest" |
| 42 | "k8s.io/apiserver/pkg/server/healthz" |
| 43 | "k8s.io/apiserver/pkg/server/routes" |
| 44 | restclient "k8s.io/client-go/rest" |
| 45 | openapicommon "k8s.io/kube-openapi/pkg/common" |
| 46 | ) |
| 47 | |
| 48 | // Info about an API group. |
| 49 | type APIGroupInfo struct { |
| 50 | PrioritizedVersions []schema.GroupVersion |
| 51 | // Info about the resources in this group. It's a map from version to resource to the storage. |
| 52 | VersionedResourcesStorageMap map[string]map[string]rest.Storage |
| 53 | // OptionsExternalVersion controls the APIVersion used for common objects in the |
| 54 | // schema like api.Status, api.DeleteOptions, and metav1.ListOptions. Other implementors may |
| 55 | // define a version "v1beta1" but want to use the Kubernetes "v1" internal objects. |
| 56 | // If nil, defaults to groupMeta.GroupVersion. |
| 57 | // TODO: Remove this when https://github.com/kubernetes/kubernetes/issues/19018 is fixed. |
| 58 | OptionsExternalVersion *schema.GroupVersion |
| 59 | // MetaGroupVersion defaults to "meta.k8s.io/v1" and is the scheme group version used to decode |
| 60 | // common API implementations like ListOptions. Future changes will allow this to vary by group |
| 61 | // version (for when the inevitable meta/v2 group emerges). |
| 62 | MetaGroupVersion *schema.GroupVersion |
| 63 | |
| 64 | // Scheme includes all of the types used by this group and how to convert between them (or |
| 65 | // to convert objects from outside of this group that are accepted in this API). |
| 66 | // TODO: replace with interfaces |
| 67 | Scheme *runtime.Scheme |
| 68 | // NegotiatedSerializer controls how this group encodes and decodes data |
| 69 | NegotiatedSerializer runtime.NegotiatedSerializer |
| 70 | // ParameterCodec performs conversions for query parameters passed to API calls |
| 71 | ParameterCodec runtime.ParameterCodec |
| 72 | } |
| 73 | |
| 74 | // GenericAPIServer contains state for a Kubernetes cluster api server. |
| 75 | type GenericAPIServer struct { |
| 76 | // discoveryAddresses is used to build cluster IPs for discovery. |
| 77 | discoveryAddresses discovery.Addresses |
| 78 | |
| 79 | // LoopbackClientConfig is a config for a privileged loopback connection to the API server |
| 80 | LoopbackClientConfig *restclient.Config |
| 81 | |
| 82 | // minRequestTimeout is how short the request timeout can be. This is used to build the RESTHandler |
| 83 | minRequestTimeout time.Duration |
| 84 | |
| 85 | // ShutdownTimeout is the timeout used for server shutdown. This specifies the timeout before server |
| 86 | // gracefully shutdown returns. |
| 87 | ShutdownTimeout time.Duration |
| 88 | |
| 89 | // legacyAPIGroupPrefixes is used to set up URL parsing for authorization and for validating requests |
| 90 | // to InstallLegacyAPIGroup |
| 91 | legacyAPIGroupPrefixes sets.String |
| 92 | |
| 93 | // admissionControl is used to build the RESTStorage that backs an API Group. |
| 94 | admissionControl admission.Interface |
| 95 | |
| 96 | // SecureServingInfo holds configuration of the TLS server. |
| 97 | SecureServingInfo *SecureServingInfo |
| 98 | |
| 99 | // ExternalAddress is the address (hostname or IP and port) that should be used in |
| 100 | // external (public internet) URLs for this GenericAPIServer. |
| 101 | ExternalAddress string |
| 102 | |
| 103 | // Serializer controls how common API objects not in a group/version prefix are serialized for this server. |
| 104 | // Individual APIGroups may define their own serializers. |
| 105 | Serializer runtime.NegotiatedSerializer |
| 106 | |
| 107 | // "Outputs" |
| 108 | // Handler holds the handlers being used by this API server |
| 109 | Handler *APIServerHandler |
| 110 | |
| 111 | // listedPathProvider is a lister which provides the set of paths to show at / |
| 112 | listedPathProvider routes.ListedPathProvider |
| 113 | |
| 114 | // DiscoveryGroupManager serves /apis |
| 115 | DiscoveryGroupManager discovery.GroupManager |
| 116 | |
| 117 | // Enable swagger and/or OpenAPI if these configs are non-nil. |
| 118 | swaggerConfig *swagger.Config |
| 119 | openAPIConfig *openapicommon.Config |
| 120 | |
| 121 | // PostStartHooks are each called after the server has started listening, in a separate go func for each |
| 122 | // with no guarantee of ordering between them. The map key is a name used for error reporting. |
| 123 | // It may kill the process with a panic if it wishes to by returning an error. |
| 124 | postStartHookLock sync.Mutex |
| 125 | postStartHooks map[string]postStartHookEntry |
| 126 | postStartHooksCalled bool |
| 127 | disabledPostStartHooks sets.String |
| 128 | |
| 129 | preShutdownHookLock sync.Mutex |
| 130 | preShutdownHooks map[string]preShutdownHookEntry |
| 131 | preShutdownHooksCalled bool |
| 132 | |
| 133 | // healthz checks |
| 134 | healthzLock sync.Mutex |
| 135 | healthzChecks []healthz.HealthzChecker |
| 136 | healthzCreated bool |
| 137 | |
| 138 | // auditing. The backend is started after the server starts listening. |
| 139 | AuditBackend audit.Backend |
| 140 | |
| 141 | // enableAPIResponseCompression indicates whether API Responses should support compression |
| 142 | // if the client requests it via Accept-Encoding |
| 143 | enableAPIResponseCompression bool |
| 144 | |
| 145 | // delegationTarget is the next delegate in the chain. This is never nil. |
| 146 | delegationTarget DelegationTarget |
| 147 | |
| 148 | // HandlerChainWaitGroup allows you to wait for all chain handlers finish after the server shutdown. |
| 149 | HandlerChainWaitGroup *utilwaitgroup.SafeWaitGroup |
| 150 | } |
| 151 | |
| 152 | // DelegationTarget is an interface which allows for composition of API servers with top level handling that works |
| 153 | // as expected. |
| 154 | type DelegationTarget interface { |
| 155 | // UnprotectedHandler returns a handler that is NOT protected by a normal chain |
| 156 | UnprotectedHandler() http.Handler |
| 157 | |
| 158 | // PostStartHooks returns the post-start hooks that need to be combined |
| 159 | PostStartHooks() map[string]postStartHookEntry |
| 160 | |
| 161 | // PreShutdownHooks returns the pre-stop hooks that need to be combined |
| 162 | PreShutdownHooks() map[string]preShutdownHookEntry |
| 163 | |
| 164 | // HealthzChecks returns the healthz checks that need to be combined |
| 165 | HealthzChecks() []healthz.HealthzChecker |
| 166 | |
| 167 | // ListedPaths returns the paths for supporting an index |
| 168 | ListedPaths() []string |
| 169 | |
| 170 | // NextDelegate returns the next delegationTarget in the chain of delegations |
| 171 | NextDelegate() DelegationTarget |
| 172 | } |
| 173 | |
| 174 | func (s *GenericAPIServer) UnprotectedHandler() http.Handler { |
| 175 | // when we delegate, we need the server we're delegating to choose whether or not to use gorestful |
| 176 | return s.Handler.Director |
| 177 | } |
| 178 | func (s *GenericAPIServer) PostStartHooks() map[string]postStartHookEntry { |
| 179 | return s.postStartHooks |
| 180 | } |
| 181 | func (s *GenericAPIServer) PreShutdownHooks() map[string]preShutdownHookEntry { |
| 182 | return s.preShutdownHooks |
| 183 | } |
| 184 | func (s *GenericAPIServer) HealthzChecks() []healthz.HealthzChecker { |
| 185 | return s.healthzChecks |
| 186 | } |
| 187 | func (s *GenericAPIServer) ListedPaths() []string { |
| 188 | return s.listedPathProvider.ListedPaths() |
| 189 | } |
| 190 | |
| 191 | func (s *GenericAPIServer) NextDelegate() DelegationTarget { |
| 192 | return s.delegationTarget |
| 193 | } |
| 194 | |
| 195 | type emptyDelegate struct { |
| 196 | } |
| 197 | |
| 198 | func NewEmptyDelegate() DelegationTarget { |
| 199 | return emptyDelegate{} |
| 200 | } |
| 201 | |
| 202 | func (s emptyDelegate) UnprotectedHandler() http.Handler { |
| 203 | return nil |
| 204 | } |
| 205 | func (s emptyDelegate) PostStartHooks() map[string]postStartHookEntry { |
| 206 | return map[string]postStartHookEntry{} |
| 207 | } |
| 208 | func (s emptyDelegate) PreShutdownHooks() map[string]preShutdownHookEntry { |
| 209 | return map[string]preShutdownHookEntry{} |
| 210 | } |
| 211 | func (s emptyDelegate) HealthzChecks() []healthz.HealthzChecker { |
| 212 | return []healthz.HealthzChecker{} |
| 213 | } |
| 214 | func (s emptyDelegate) ListedPaths() []string { |
| 215 | return []string{} |
| 216 | } |
| 217 | func (s emptyDelegate) NextDelegate() DelegationTarget { |
| 218 | return nil |
| 219 | } |
| 220 | |
| 221 | // preparedGenericAPIServer is a private wrapper that enforces a call of PrepareRun() before Run can be invoked. |
| 222 | type preparedGenericAPIServer struct { |
| 223 | *GenericAPIServer |
| 224 | } |
| 225 | |
| 226 | // PrepareRun does post API installation setup steps. |
| 227 | func (s *GenericAPIServer) PrepareRun() preparedGenericAPIServer { |
| 228 | if s.swaggerConfig != nil { |
| 229 | routes.Swagger{Config: s.swaggerConfig}.Install(s.Handler.GoRestfulContainer) |
| 230 | } |
| 231 | if s.openAPIConfig != nil { |
| 232 | routes.OpenAPI{ |
| 233 | Config: s.openAPIConfig, |
| 234 | }.Install(s.Handler.GoRestfulContainer, s.Handler.NonGoRestfulMux) |
| 235 | } |
| 236 | |
| 237 | s.installHealthz() |
| 238 | |
| 239 | // Register audit backend preShutdownHook. |
| 240 | if s.AuditBackend != nil { |
| 241 | s.AddPreShutdownHook("audit-backend", func() error { |
| 242 | s.AuditBackend.Shutdown() |
| 243 | return nil |
| 244 | }) |
| 245 | } |
| 246 | |
| 247 | return preparedGenericAPIServer{s} |
| 248 | } |
| 249 | |
| 250 | // Run spawns the secure http server. It only returns if stopCh is closed |
| 251 | // or the secure port cannot be listened on initially. |
| 252 | func (s preparedGenericAPIServer) Run(stopCh <-chan struct{}) error { |
| 253 | err := s.NonBlockingRun(stopCh) |
| 254 | if err != nil { |
| 255 | return err |
| 256 | } |
| 257 | |
| 258 | <-stopCh |
| 259 | |
| 260 | err = s.RunPreShutdownHooks() |
| 261 | if err != nil { |
| 262 | return err |
| 263 | } |
| 264 | |
| 265 | // Wait for all requests to finish, which are bounded by the RequestTimeout variable. |
| 266 | s.HandlerChainWaitGroup.Wait() |
| 267 | |
| 268 | return nil |
| 269 | } |
| 270 | |
| 271 | // NonBlockingRun spawns the secure http server. An error is |
| 272 | // returned if the secure port cannot be listened on. |
| 273 | func (s preparedGenericAPIServer) NonBlockingRun(stopCh <-chan struct{}) error { |
| 274 | // Use an stop channel to allow graceful shutdown without dropping audit events |
| 275 | // after http server shutdown. |
| 276 | auditStopCh := make(chan struct{}) |
| 277 | |
| 278 | // Start the audit backend before any request comes in. This means we must call Backend.Run |
| 279 | // before http server start serving. Otherwise the Backend.ProcessEvents call might block. |
| 280 | if s.AuditBackend != nil { |
| 281 | if err := s.AuditBackend.Run(auditStopCh); err != nil { |
| 282 | return fmt.Errorf("failed to run the audit backend: %v", err) |
| 283 | } |
| 284 | } |
| 285 | |
| 286 | // Use an internal stop channel to allow cleanup of the listeners on error. |
| 287 | internalStopCh := make(chan struct{}) |
| 288 | |
| 289 | if s.SecureServingInfo != nil && s.Handler != nil { |
| 290 | if err := s.SecureServingInfo.Serve(s.Handler, s.ShutdownTimeout, internalStopCh); err != nil { |
| 291 | close(internalStopCh) |
| 292 | return err |
| 293 | } |
| 294 | } |
| 295 | |
| 296 | // Now that listener have bound successfully, it is the |
| 297 | // responsibility of the caller to close the provided channel to |
| 298 | // ensure cleanup. |
| 299 | go func() { |
| 300 | <-stopCh |
| 301 | close(internalStopCh) |
| 302 | s.HandlerChainWaitGroup.Wait() |
| 303 | close(auditStopCh) |
| 304 | }() |
| 305 | |
| 306 | s.RunPostStartHooks(stopCh) |
| 307 | |
| 308 | if _, err := systemd.SdNotify(true, "READY=1\n"); err != nil { |
| 309 | glog.Errorf("Unable to send systemd daemon successful start message: %v\n", err) |
| 310 | } |
| 311 | |
| 312 | return nil |
| 313 | } |
| 314 | |
| 315 | // installAPIResources is a private method for installing the REST storage backing each api groupversionresource |
| 316 | func (s *GenericAPIServer) installAPIResources(apiPrefix string, apiGroupInfo *APIGroupInfo) error { |
| 317 | for _, groupVersion := range apiGroupInfo.PrioritizedVersions { |
| 318 | if len(apiGroupInfo.VersionedResourcesStorageMap[groupVersion.Version]) == 0 { |
| 319 | glog.Warningf("Skipping API %v because it has no resources.", groupVersion) |
| 320 | continue |
| 321 | } |
| 322 | |
| 323 | apiGroupVersion := s.getAPIGroupVersion(apiGroupInfo, groupVersion, apiPrefix) |
| 324 | if apiGroupInfo.OptionsExternalVersion != nil { |
| 325 | apiGroupVersion.OptionsExternalVersion = apiGroupInfo.OptionsExternalVersion |
| 326 | } |
| 327 | |
| 328 | if err := apiGroupVersion.InstallREST(s.Handler.GoRestfulContainer); err != nil { |
| 329 | return fmt.Errorf("unable to setup API %v: %v", apiGroupInfo, err) |
| 330 | } |
| 331 | } |
| 332 | |
| 333 | return nil |
| 334 | } |
| 335 | |
| 336 | func (s *GenericAPIServer) InstallLegacyAPIGroup(apiPrefix string, apiGroupInfo *APIGroupInfo) error { |
| 337 | if !s.legacyAPIGroupPrefixes.Has(apiPrefix) { |
| 338 | return fmt.Errorf("%q is not in the allowed legacy API prefixes: %v", apiPrefix, s.legacyAPIGroupPrefixes.List()) |
| 339 | } |
| 340 | if err := s.installAPIResources(apiPrefix, apiGroupInfo); err != nil { |
| 341 | return err |
| 342 | } |
| 343 | |
| 344 | // setup discovery |
| 345 | apiVersions := []string{} |
| 346 | for _, groupVersion := range apiGroupInfo.PrioritizedVersions { |
| 347 | apiVersions = append(apiVersions, groupVersion.Version) |
| 348 | } |
| 349 | // Install the version handler. |
| 350 | // Add a handler at /<apiPrefix> to enumerate the supported api versions. |
| 351 | s.Handler.GoRestfulContainer.Add(discovery.NewLegacyRootAPIHandler(s.discoveryAddresses, s.Serializer, apiPrefix, apiVersions).WebService()) |
| 352 | |
| 353 | return nil |
| 354 | } |
| 355 | |
| 356 | // Exposes the given api group in the API. |
| 357 | func (s *GenericAPIServer) InstallAPIGroup(apiGroupInfo *APIGroupInfo) error { |
| 358 | // Do not register empty group or empty version. Doing so claims /apis/ for the wrong entity to be returned. |
| 359 | // Catching these here places the error much closer to its origin |
| 360 | if len(apiGroupInfo.PrioritizedVersions[0].Group) == 0 { |
| 361 | return fmt.Errorf("cannot register handler with an empty group for %#v", *apiGroupInfo) |
| 362 | } |
| 363 | if len(apiGroupInfo.PrioritizedVersions[0].Version) == 0 { |
| 364 | return fmt.Errorf("cannot register handler with an empty version for %#v", *apiGroupInfo) |
| 365 | } |
| 366 | |
| 367 | if err := s.installAPIResources(APIGroupPrefix, apiGroupInfo); err != nil { |
| 368 | return err |
| 369 | } |
| 370 | |
| 371 | // setup discovery |
| 372 | // Install the version handler. |
| 373 | // Add a handler at /apis/<groupName> to enumerate all versions supported by this group. |
| 374 | apiVersionsForDiscovery := []metav1.GroupVersionForDiscovery{} |
| 375 | for _, groupVersion := range apiGroupInfo.PrioritizedVersions { |
| 376 | // Check the config to make sure that we elide versions that don't have any resources |
| 377 | if len(apiGroupInfo.VersionedResourcesStorageMap[groupVersion.Version]) == 0 { |
| 378 | continue |
| 379 | } |
| 380 | apiVersionsForDiscovery = append(apiVersionsForDiscovery, metav1.GroupVersionForDiscovery{ |
| 381 | GroupVersion: groupVersion.String(), |
| 382 | Version: groupVersion.Version, |
| 383 | }) |
| 384 | } |
| 385 | preferredVersionForDiscovery := metav1.GroupVersionForDiscovery{ |
| 386 | GroupVersion: apiGroupInfo.PrioritizedVersions[0].String(), |
| 387 | Version: apiGroupInfo.PrioritizedVersions[0].Version, |
| 388 | } |
| 389 | apiGroup := metav1.APIGroup{ |
| 390 | Name: apiGroupInfo.PrioritizedVersions[0].Group, |
| 391 | Versions: apiVersionsForDiscovery, |
| 392 | PreferredVersion: preferredVersionForDiscovery, |
| 393 | } |
| 394 | |
| 395 | s.DiscoveryGroupManager.AddGroup(apiGroup) |
| 396 | s.Handler.GoRestfulContainer.Add(discovery.NewAPIGroupHandler(s.Serializer, apiGroup).WebService()) |
| 397 | |
| 398 | return nil |
| 399 | } |
| 400 | |
| 401 | func (s *GenericAPIServer) getAPIGroupVersion(apiGroupInfo *APIGroupInfo, groupVersion schema.GroupVersion, apiPrefix string) *genericapi.APIGroupVersion { |
| 402 | storage := make(map[string]rest.Storage) |
| 403 | for k, v := range apiGroupInfo.VersionedResourcesStorageMap[groupVersion.Version] { |
| 404 | storage[strings.ToLower(k)] = v |
| 405 | } |
| 406 | version := s.newAPIGroupVersion(apiGroupInfo, groupVersion) |
| 407 | version.Root = apiPrefix |
| 408 | version.Storage = storage |
| 409 | return version |
| 410 | } |
| 411 | |
| 412 | func (s *GenericAPIServer) newAPIGroupVersion(apiGroupInfo *APIGroupInfo, groupVersion schema.GroupVersion) *genericapi.APIGroupVersion { |
| 413 | return &genericapi.APIGroupVersion{ |
| 414 | GroupVersion: groupVersion, |
| 415 | MetaGroupVersion: apiGroupInfo.MetaGroupVersion, |
| 416 | |
| 417 | ParameterCodec: apiGroupInfo.ParameterCodec, |
| 418 | Serializer: apiGroupInfo.NegotiatedSerializer, |
| 419 | Creater: apiGroupInfo.Scheme, |
| 420 | Convertor: apiGroupInfo.Scheme, |
| 421 | UnsafeConvertor: runtime.UnsafeObjectConvertor(apiGroupInfo.Scheme), |
| 422 | Defaulter: apiGroupInfo.Scheme, |
| 423 | Typer: apiGroupInfo.Scheme, |
| 424 | Linker: runtime.SelfLinker(meta.NewAccessor()), |
| 425 | |
| 426 | Admit: s.admissionControl, |
| 427 | MinRequestTimeout: s.minRequestTimeout, |
| 428 | EnableAPIResponseCompression: s.enableAPIResponseCompression, |
| 429 | OpenAPIConfig: s.openAPIConfig, |
| 430 | } |
| 431 | } |
| 432 | |
| 433 | // NewDefaultAPIGroupInfo returns an APIGroupInfo stubbed with "normal" values |
| 434 | // exposed for easier composition from other packages |
| 435 | func NewDefaultAPIGroupInfo(group string, scheme *runtime.Scheme, parameterCodec runtime.ParameterCodec, codecs serializer.CodecFactory) APIGroupInfo { |
| 436 | return APIGroupInfo{ |
| 437 | PrioritizedVersions: scheme.PrioritizedVersionsForGroup(group), |
| 438 | VersionedResourcesStorageMap: map[string]map[string]rest.Storage{}, |
| 439 | // TODO unhardcode this. It was hardcoded before, but we need to re-evaluate |
| 440 | OptionsExternalVersion: &schema.GroupVersion{Version: "v1"}, |
| 441 | Scheme: scheme, |
| 442 | ParameterCodec: parameterCodec, |
| 443 | NegotiatedSerializer: codecs, |
| 444 | } |
| 445 | } |