Commit a4542ae5 authored by staebler's avatar staebler

Create an EventRateLimit admission control plug-in for the API Server.

The EventRateLimit plug-in limits the number of events that the API Server will accept in a given time period. It allows for server-wide, per-namespace, per-user,and per-source+object rate limiting.
parent 55cb667b
......@@ -26,6 +26,7 @@ go_library(
"//plugin/pkg/admission/antiaffinity:go_default_library",
"//plugin/pkg/admission/defaulttolerationseconds:go_default_library",
"//plugin/pkg/admission/deny:go_default_library",
"//plugin/pkg/admission/eventratelimit:go_default_library",
"//plugin/pkg/admission/exec:go_default_library",
"//plugin/pkg/admission/gc:go_default_library",
"//plugin/pkg/admission/imagepolicy:go_default_library",
......
......@@ -30,6 +30,7 @@ import (
"k8s.io/kubernetes/plugin/pkg/admission/antiaffinity"
"k8s.io/kubernetes/plugin/pkg/admission/defaulttolerationseconds"
"k8s.io/kubernetes/plugin/pkg/admission/deny"
"k8s.io/kubernetes/plugin/pkg/admission/eventratelimit"
"k8s.io/kubernetes/plugin/pkg/admission/exec"
"k8s.io/kubernetes/plugin/pkg/admission/gc"
"k8s.io/kubernetes/plugin/pkg/admission/imagepolicy"
......@@ -59,6 +60,7 @@ func RegisterAllAdmissionPlugins(plugins *admission.Plugins) {
antiaffinity.Register(plugins)
defaulttolerationseconds.Register(plugins)
deny.Register(plugins)
eventratelimit.Register(plugins)
exec.Register(plugins)
gc.Register(plugins)
imagepolicy.Register(plugins)
......
......@@ -467,6 +467,8 @@ pkg/volume/util
pkg/volume/vsphere_volume
plugin/cmd/kube-scheduler/app
plugin/pkg/admission/antiaffinity
plugin/pkg/admission/eventratelimit/apis/eventratelimit
plugin/pkg/admission/eventratelimit/apis/eventratelimit/v1alpha1
plugin/pkg/admission/initialization
plugin/pkg/admission/initialresources
plugin/pkg/admission/limitranger
......
......@@ -17,6 +17,7 @@ filegroup(
"//plugin/pkg/admission/antiaffinity:all-srcs",
"//plugin/pkg/admission/defaulttolerationseconds:all-srcs",
"//plugin/pkg/admission/deny:all-srcs",
"//plugin/pkg/admission/eventratelimit:all-srcs",
"//plugin/pkg/admission/exec:all-srcs",
"//plugin/pkg/admission/gc:all-srcs",
"//plugin/pkg/admission/imagepolicy:all-srcs",
......
package(default_visibility = ["//visibility:public"])
licenses(["notice"])
load(
"@io_bazel_rules_go//go:def.bzl",
"go_library",
"go_test",
)
go_test(
name = "go_default_test",
srcs = [
"admission_test.go",
"cache_test.go",
],
library = ":go_default_library",
tags = ["automanaged"],
deps = [
"//pkg/api:go_default_library",
"//plugin/pkg/admission/eventratelimit/apis/eventratelimit:go_default_library",
"//vendor/github.com/hashicorp/golang-lru:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/util/clock:go_default_library",
"//vendor/k8s.io/apiserver/pkg/admission:go_default_library",
"//vendor/k8s.io/apiserver/pkg/authentication/user:go_default_library",
"//vendor/k8s.io/client-go/util/flowcontrol:go_default_library",
],
)
go_library(
name = "go_default_library",
srcs = [
"admission.go",
"cache.go",
"clock.go",
"config.go",
"doc.go",
"limitenforcer.go",
],
tags = ["automanaged"],
deps = [
"//pkg/api:go_default_library",
"//plugin/pkg/admission/eventratelimit/apis/eventratelimit:go_default_library",
"//plugin/pkg/admission/eventratelimit/apis/eventratelimit/install:go_default_library",
"//plugin/pkg/admission/eventratelimit/apis/eventratelimit/v1alpha1:go_default_library",
"//plugin/pkg/admission/eventratelimit/apis/eventratelimit/validation:go_default_library",
"//vendor/github.com/hashicorp/golang-lru:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/apimachinery/announced:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/apimachinery/registered:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/runtime/serializer:go_default_library",
"//vendor/k8s.io/apiserver/pkg/admission:go_default_library",
"//vendor/k8s.io/client-go/util/flowcontrol:go_default_library",
],
)
filegroup(
name = "package-srcs",
srcs = glob(["**"]),
tags = ["automanaged"],
visibility = ["//visibility:private"],
)
filegroup(
name = "all-srcs",
srcs = [
":package-srcs",
"//plugin/pkg/admission/eventratelimit/apis/eventratelimit:all-srcs",
],
tags = ["automanaged"],
)
/*
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 eventratelimit
import (
"io"
"k8s.io/apiserver/pkg/admission"
"k8s.io/client-go/util/flowcontrol"
"k8s.io/kubernetes/pkg/api"
eventratelimitapi "k8s.io/kubernetes/plugin/pkg/admission/eventratelimit/apis/eventratelimit"
"k8s.io/kubernetes/plugin/pkg/admission/eventratelimit/apis/eventratelimit/validation"
)
// Register registers a plugin
func Register(plugins *admission.Plugins) {
plugins.Register("EventRateLimit",
func(config io.Reader) (admission.Interface, error) {
// load the configuration provided (if any)
configuration, err := LoadConfiguration(config)
if err != nil {
return nil, err
}
// validate the configuration (if any)
if configuration != nil {
if errs := validation.ValidateConfiguration(configuration); len(errs) != 0 {
return nil, errs.ToAggregate()
}
}
return newEventRateLimit(configuration, realClock{})
})
}
// eventRateLimitAdmission implements an admission controller that can enforce event rate limits
type eventRateLimitAdmission struct {
*admission.Handler
// limitEnforcers is the collection of limit enforcers. There is one limit enforcer for each
// active limit type. As there are 4 limit types, the length of the array will be at most 4.
// The array is read-only after construction.
limitEnforcers []*limitEnforcer
}
// newEventRateLimit configures an admission controller that can enforce event rate limits
func newEventRateLimit(config *eventratelimitapi.Configuration, clock flowcontrol.Clock) (admission.Interface, error) {
limitEnforcers := make([]*limitEnforcer, 0, len(config.Limits))
for _, limitConfig := range config.Limits {
enforcer, err := newLimitEnforcer(limitConfig, clock)
if err != nil {
return nil, err
}
limitEnforcers = append(limitEnforcers, enforcer)
}
eventRateLimitAdmission := &eventRateLimitAdmission{
Handler: admission.NewHandler(admission.Create, admission.Update),
limitEnforcers: limitEnforcers,
}
return eventRateLimitAdmission, nil
}
// Admit makes admission decisions while enforcing event rate limits
func (a *eventRateLimitAdmission) Admit(attr admission.Attributes) (err error) {
// ignore all operations that do not correspond to an Event kind
if attr.GetKind().GroupKind() != api.Kind("Event") {
return nil
}
var rejectionError error
// give each limit enforcer a chance to reject the event
for _, enforcer := range a.limitEnforcers {
if err := enforcer.accept(attr); err != nil {
rejectionError = err
}
}
return rejectionError
}
package(default_visibility = ["//visibility:public"])
licenses(["notice"])
load(
"@io_bazel_rules_go//go:def.bzl",
"go_library",
)
go_library(
name = "go_default_library",
srcs = [
"doc.go",
"register.go",
"types.go",
"zz_generated.deepcopy.go",
],
tags = ["automanaged"],
deps = [
"//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/conversion:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
],
)
filegroup(
name = "package-srcs",
srcs = glob(["**"]),
tags = ["automanaged"],
visibility = ["//visibility:private"],
)
filegroup(
name = "all-srcs",
srcs = [
":package-srcs",
"//plugin/pkg/admission/eventratelimit/apis/eventratelimit/install:all-srcs",
"//plugin/pkg/admission/eventratelimit/apis/eventratelimit/v1alpha1:all-srcs",
"//plugin/pkg/admission/eventratelimit/apis/eventratelimit/validation:all-srcs",
],
tags = ["automanaged"],
)
reviewers:
- deads2k
- derekwaynecarr
approvers:
- deads2k
- derekwaynecarr
- smarterclayton
\ No newline at end of file
/*
Copyright 2016 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.
*/
// +k8s:deepcopy-gen=package,register
package eventratelimit // import "k8s.io/kubernetes/plugin/pkg/admission/eventratelimit/apis/eventratelimit"
package(default_visibility = ["//visibility:public"])
licenses(["notice"])
load(
"@io_bazel_rules_go//go:def.bzl",
"go_library",
)
go_library(
name = "go_default_library",
srcs = ["install.go"],
tags = ["automanaged"],
deps = [
"//plugin/pkg/admission/eventratelimit/apis/eventratelimit:go_default_library",
"//plugin/pkg/admission/eventratelimit/apis/eventratelimit/v1alpha1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/apimachinery/announced:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/apimachinery/registered:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
],
)
filegroup(
name = "package-srcs",
srcs = glob(["**"]),
tags = ["automanaged"],
visibility = ["//visibility:private"],
)
filegroup(
name = "all-srcs",
srcs = [":package-srcs"],
tags = ["automanaged"],
)
/*
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 install installs the experimental API group, making it available as
// an option to all of the API encoding/decoding machinery.
package install
import (
"k8s.io/apimachinery/pkg/apimachinery/announced"
"k8s.io/apimachinery/pkg/apimachinery/registered"
"k8s.io/apimachinery/pkg/runtime"
internalapi "k8s.io/kubernetes/plugin/pkg/admission/eventratelimit/apis/eventratelimit"
versionedapi "k8s.io/kubernetes/plugin/pkg/admission/eventratelimit/apis/eventratelimit/v1alpha1"
)
// Install registers the API group and adds types to a scheme
func Install(groupFactoryRegistry announced.APIGroupFactoryRegistry, registry *registered.APIRegistrationManager, scheme *runtime.Scheme) {
if err := announced.NewGroupMetaFactory(
&announced.GroupMetaFactoryArgs{
GroupName: internalapi.GroupName,
VersionPreferenceOrder: []string{versionedapi.SchemeGroupVersion.Version},
AddInternalObjectsToScheme: internalapi.AddToScheme,
},
announced.VersionToSchemeFunc{
versionedapi.SchemeGroupVersion.Version: versionedapi.AddToScheme,
},
).Announce(groupFactoryRegistry).RegisterAndEnable(registry, scheme); err != nil {
panic(err)
}
}
/*
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 eventratelimit
import (
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
)
var (
SchemeBuilder = runtime.NewSchemeBuilder(addKnownTypes)
AddToScheme = SchemeBuilder.AddToScheme
)
// GroupName is the group name use in this package
const GroupName = "eventratelimit.admission.k8s.io"
// SchemeGroupVersion is group version used to register these objects
var SchemeGroupVersion = schema.GroupVersion{Group: GroupName, Version: runtime.APIVersionInternal}
// Kind takes an unqualified kind and returns a Group qualified GroupKind
func Kind(kind string) schema.GroupKind {
return SchemeGroupVersion.WithKind(kind).GroupKind()
}
// Resource takes an unqualified resource and returns a Group qualified GroupResource
func Resource(resource string) schema.GroupResource {
return SchemeGroupVersion.WithResource(resource).GroupResource()
}
func addKnownTypes(scheme *runtime.Scheme) error {
// TODO this will get cleaned up with the scheme types are fixed
scheme.AddKnownTypes(SchemeGroupVersion,
&Configuration{},
)
return nil
}
/*
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 eventratelimit
import metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
// LimitType is the type of the limit (e.g., per-namespace)
type LimitType string
const (
// ServerLimitType is a type of limit where there is one bucket shared by
// all of the event queries received by the API Server.
ServerLimitType LimitType = "Server"
// NamespaceLimitType is a type of limit where there is one bucket used by
// each namespace
NamespaceLimitType LimitType = "Namespace"
// UserLimitType is a type of limit where there is one bucket used by each
// user
UserLimitType LimitType = "User"
// SourceAndObjectLimitType is a type of limit where there is one bucket used
// by each combination of source and involved object of the event.
SourceAndObjectLimitType LimitType = "SourceAndObject"
)
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
// Configuration provides configuration for the EventRateLimit admission
// controller.
type Configuration struct {
metav1.TypeMeta `json:",inline"`
// limits are the limits to place on event queries received.
// Limits can be placed on events received server-wide, per namespace,
// per user, and per source+object.
// At least one limit is required.
Limits []Limit `json:"limits"`
}
// Limit is the configuration for a particular limit type
type Limit struct {
// type is the type of limit to which this configuration applies
Type LimitType `json:"type"`
// qps is the number of event queries per second that are allowed for this
// type of limit. The qps and burst fields are used together to determine if
// a particular event query is accepted. The qps determines how many queries
// are accepted once the burst amount of queries has been exhausted.
QPS int32 `json:"qps"`
// burst is the burst number of event queries that are allowed for this type
// of limit. The qps and burst fields are used together to determine if a
// particular event query is accepted. The burst determines the maximum size
// of the allowance granted for a particular bucket. For example, if the burst
// is 10 and the qps is 3, then the admission control will accept 10 queries
// before blocking any queries. Every second, 3 more queries will be allowed.
// If some of that allowance is not used, then it will roll over to the next
// second, until the maximum allowance of 10 is reached.
Burst int32 `json:"burst"`
// cacheSize is the size of the LRU cache for this type of limit. If a bucket
// is evicted from the cache, then the allowance for that bucket is reset. If
// more queries are later received for an evicted bucket, then that bucket
// will re-enter the cache with a clean slate, giving that bucket a full
// allowance of burst queries.
//
// The default cache size is 4096.
//
// If limitType is 'server', then cacheSize is ignored.
// +optional
CacheSize int32 `json:"cacheSize,omitempty"`
}
package(default_visibility = ["//visibility:public"])
licenses(["notice"])
load(
"@io_bazel_rules_go//go:def.bzl",
"go_library",
)
go_library(
name = "go_default_library",
srcs = [
"defaults.go",
"doc.go",
"register.go",
"types.go",
"zz_generated.conversion.go",
"zz_generated.deepcopy.go",
"zz_generated.defaults.go",
],
tags = ["automanaged"],
deps = [
"//plugin/pkg/admission/eventratelimit/apis/eventratelimit:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/conversion:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
],
)
filegroup(
name = "package-srcs",
srcs = glob(["**"]),
tags = ["automanaged"],
visibility = ["//visibility:private"],
)
filegroup(
name = "all-srcs",
srcs = [":package-srcs"],
tags = ["automanaged"],
)
/*
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 v1alpha1
import kruntime "k8s.io/apimachinery/pkg/runtime"
func addDefaultingFuncs(scheme *kruntime.Scheme) error {
return RegisterDefaults(scheme)
}
func SetDefaults_Configuration(obj *Configuration) {}
/*
Copyright 2016 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.
*/
// +k8s:deepcopy-gen=package,register
// +k8s:conversion-gen=k8s.io/kubernetes/plugin/pkg/admission/eventratelimit/apis/eventratelimit
// +k8s:defaulter-gen=TypeMeta
// Package v1alpha1 is the v1alpha1 version of the API.
// +groupName=eventratelimit.admission.k8s.io
package v1alpha1 // import "k8s.io/kubernetes/plugin/pkg/admission/eventratelimit/apis/eventratelimit/v1alpha1"
/*
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 v1alpha1
import (
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
)
// GroupName is the group name use in this package
const GroupName = "eventratelimit.admission.k8s.io"
// SchemeGroupVersion is group version used to register these objects
var SchemeGroupVersion = schema.GroupVersion{Group: GroupName, Version: "v1alpha1"}
var (
// TODO: move SchemeBuilder with zz_generated.deepcopy.go to k8s.io/api.
// localSchemeBuilder and AddToScheme will stay in k8s.io/kubernetes.
SchemeBuilder runtime.SchemeBuilder
localSchemeBuilder = &SchemeBuilder
AddToScheme = localSchemeBuilder.AddToScheme
)
func init() {
// We only register manually written functions here. The registration of the
// generated functions takes place in the generated files. The separation
// makes the code compile even when the generated files are missing.
localSchemeBuilder.Register(addKnownTypes, addDefaultingFuncs)
}
func addKnownTypes(scheme *runtime.Scheme) error {
scheme.AddKnownTypes(SchemeGroupVersion,
&Configuration{},
)
return nil
}
/*
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 v1alpha1
import metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
// LimitType is the type of the limit (e.g., per-namespace)
type LimitType string
const (
// ServerLimitType is a type of limit where there is one bucket shared by
// all of the event queries received by the API Server.
ServerLimitType LimitType = "Server"
// NamespaceLimitType is a type of limit where there is one bucket used by
// each namespace
NamespaceLimitType LimitType = "Namespace"
// UserLimitType is a type of limit where there is one bucket used by each
// user
UserLimitType LimitType = "User"
// SourceAndObjectLimitType is a type of limit where there is one bucket used
// by each combination of source and involved object of the event.
SourceAndObjectLimitType LimitType = "SourceAndObject"
)
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
// Configuration provides configuration for the EventRateLimit admission
// controller.
type Configuration struct {
metav1.TypeMeta `json:",inline"`
// limits are the limits to place on event queries received.
// Limits can be placed on events received server-wide, per namespace,
// per user, and per source+object.
// At least one limit is required.
Limits []Limit `json:"limits"`
}
// Limit is the configuration for a particular limit type
type Limit struct {
// type is the type of limit to which this configuration applies
Type LimitType `json:"type"`
// qps is the number of event queries per second that are allowed for this
// type of limit. The qps and burst fields are used together to determine if
// a particular event query is accepted. The qps determines how many queries
// are accepted once the burst amount of queries has been exhausted.
QPS int32 `json:"qps"`
// burst is the burst number of event queries that are allowed for this type
// of limit. The qps and burst fields are used together to determine if a
// particular event query is accepted. The burst determines the maximum size
// of the allowance granted for a particular bucket. For example, if the burst
// is 10 and the qps is 3, then the admission control will accept 10 queries
// before blocking any queries. Every second, 3 more queries will be allowed.
// If some of that allowance is not used, then it will roll over to the next
// second, until the maximum allowance of 10 is reached.
Burst int32 `json:"burst"`
// cacheSize is the size of the LRU cache for this type of limit. If a bucket
// is evicted from the cache, then the allowance for that bucket is reset. If
// more queries are later received for an evicted bucket, then that bucket
// will re-enter the cache with a clean slate, giving that bucket a full
// allowance of burst queries.
//
// The default cache size is 4096.
//
// If limitType is 'server', then cacheSize is ignored.
// +optional
CacheSize int32 `json:"cacheSize,omitempty"`
}
// +build !ignore_autogenerated
/*
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.
*/
// This file was autogenerated by conversion-gen. Do not edit it manually!
package v1alpha1
import (
conversion "k8s.io/apimachinery/pkg/conversion"
runtime "k8s.io/apimachinery/pkg/runtime"
eventratelimit "k8s.io/kubernetes/plugin/pkg/admission/eventratelimit/apis/eventratelimit"
unsafe "unsafe"
)
func init() {
localSchemeBuilder.Register(RegisterConversions)
}
// RegisterConversions adds conversion functions to the given scheme.
// Public to allow building arbitrary schemes.
func RegisterConversions(scheme *runtime.Scheme) error {
return scheme.AddGeneratedConversionFuncs(
Convert_v1alpha1_Configuration_To_eventratelimit_Configuration,
Convert_eventratelimit_Configuration_To_v1alpha1_Configuration,
Convert_v1alpha1_Limit_To_eventratelimit_Limit,
Convert_eventratelimit_Limit_To_v1alpha1_Limit,
)
}
func autoConvert_v1alpha1_Configuration_To_eventratelimit_Configuration(in *Configuration, out *eventratelimit.Configuration, s conversion.Scope) error {
out.Limits = *(*[]eventratelimit.Limit)(unsafe.Pointer(&in.Limits))
return nil
}
// Convert_v1alpha1_Configuration_To_eventratelimit_Configuration is an autogenerated conversion function.
func Convert_v1alpha1_Configuration_To_eventratelimit_Configuration(in *Configuration, out *eventratelimit.Configuration, s conversion.Scope) error {
return autoConvert_v1alpha1_Configuration_To_eventratelimit_Configuration(in, out, s)
}
func autoConvert_eventratelimit_Configuration_To_v1alpha1_Configuration(in *eventratelimit.Configuration, out *Configuration, s conversion.Scope) error {
out.Limits = *(*[]Limit)(unsafe.Pointer(&in.Limits))
return nil
}
// Convert_eventratelimit_Configuration_To_v1alpha1_Configuration is an autogenerated conversion function.
func Convert_eventratelimit_Configuration_To_v1alpha1_Configuration(in *eventratelimit.Configuration, out *Configuration, s conversion.Scope) error {
return autoConvert_eventratelimit_Configuration_To_v1alpha1_Configuration(in, out, s)
}
func autoConvert_v1alpha1_Limit_To_eventratelimit_Limit(in *Limit, out *eventratelimit.Limit, s conversion.Scope) error {
out.Type = eventratelimit.LimitType(in.Type)
out.QPS = in.QPS
out.Burst = in.Burst
out.CacheSize = in.CacheSize
return nil
}
// Convert_v1alpha1_Limit_To_eventratelimit_Limit is an autogenerated conversion function.
func Convert_v1alpha1_Limit_To_eventratelimit_Limit(in *Limit, out *eventratelimit.Limit, s conversion.Scope) error {
return autoConvert_v1alpha1_Limit_To_eventratelimit_Limit(in, out, s)
}
func autoConvert_eventratelimit_Limit_To_v1alpha1_Limit(in *eventratelimit.Limit, out *Limit, s conversion.Scope) error {
out.Type = LimitType(in.Type)
out.QPS = in.QPS
out.Burst = in.Burst
out.CacheSize = in.CacheSize
return nil
}
// Convert_eventratelimit_Limit_To_v1alpha1_Limit is an autogenerated conversion function.
func Convert_eventratelimit_Limit_To_v1alpha1_Limit(in *eventratelimit.Limit, out *Limit, s conversion.Scope) error {
return autoConvert_eventratelimit_Limit_To_v1alpha1_Limit(in, out, s)
}
// +build !ignore_autogenerated
/*
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.
*/
// This file was autogenerated by deepcopy-gen. Do not edit it manually!
package v1alpha1
import (
conversion "k8s.io/apimachinery/pkg/conversion"
runtime "k8s.io/apimachinery/pkg/runtime"
reflect "reflect"
)
func init() {
SchemeBuilder.Register(RegisterDeepCopies)
}
// RegisterDeepCopies adds deep-copy functions to the given scheme. Public
// to allow building arbitrary schemes.
//
// Deprecated: deepcopy registration will go away when static deepcopy is fully implemented.
func RegisterDeepCopies(scheme *runtime.Scheme) error {
return scheme.AddGeneratedDeepCopyFuncs(
conversion.GeneratedDeepCopyFunc{Fn: func(in interface{}, out interface{}, c *conversion.Cloner) error {
in.(*Configuration).DeepCopyInto(out.(*Configuration))
return nil
}, InType: reflect.TypeOf(&Configuration{})},
conversion.GeneratedDeepCopyFunc{Fn: func(in interface{}, out interface{}, c *conversion.Cloner) error {
in.(*Limit).DeepCopyInto(out.(*Limit))
return nil
}, InType: reflect.TypeOf(&Limit{})},
)
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *Configuration) DeepCopyInto(out *Configuration) {
*out = *in
out.TypeMeta = in.TypeMeta
if in.Limits != nil {
in, out := &in.Limits, &out.Limits
*out = make([]Limit, len(*in))
copy(*out, *in)
}
return
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Configuration.
func (in *Configuration) DeepCopy() *Configuration {
if in == nil {
return nil
}
out := new(Configuration)
in.DeepCopyInto(out)
return out
}
// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
func (in *Configuration) DeepCopyObject() runtime.Object {
if c := in.DeepCopy(); c != nil {
return c
} else {
return nil
}
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *Limit) DeepCopyInto(out *Limit) {
*out = *in
return
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Limit.
func (in *Limit) DeepCopy() *Limit {
if in == nil {
return nil
}
out := new(Limit)
in.DeepCopyInto(out)
return out
}
// +build !ignore_autogenerated
/*
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.
*/
// This file was autogenerated by defaulter-gen. Do not edit it manually!
package v1alpha1
import (
runtime "k8s.io/apimachinery/pkg/runtime"
)
// RegisterDefaults adds defaulters functions to the given scheme.
// Public to allow building arbitrary schemes.
// All generated defaulters are covering - they call all nested defaulters.
func RegisterDefaults(scheme *runtime.Scheme) error {
scheme.AddTypeDefaultingFunc(&Configuration{}, func(obj interface{}) { SetObjectDefaults_Configuration(obj.(*Configuration)) })
return nil
}
func SetObjectDefaults_Configuration(in *Configuration) {
SetDefaults_Configuration(in)
}
package(default_visibility = ["//visibility:public"])
licenses(["notice"])
load(
"@io_bazel_rules_go//go:def.bzl",
"go_library",
"go_test",
)
go_library(
name = "go_default_library",
srcs = ["validation.go"],
tags = ["automanaged"],
deps = [
"//plugin/pkg/admission/eventratelimit/apis/eventratelimit:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/util/validation/field:go_default_library",
],
)
filegroup(
name = "package-srcs",
srcs = glob(["**"]),
tags = ["automanaged"],
visibility = ["//visibility:private"],
)
filegroup(
name = "all-srcs",
srcs = [":package-srcs"],
tags = ["automanaged"],
)
go_test(
name = "go_default_test",
srcs = ["validation_test.go"],
library = ":go_default_library",
tags = ["automanaged"],
deps = ["//plugin/pkg/admission/eventratelimit/apis/eventratelimit:go_default_library"],
)
/*
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 validation
import (
"k8s.io/apimachinery/pkg/util/validation/field"
eventratelimitapi "k8s.io/kubernetes/plugin/pkg/admission/eventratelimit/apis/eventratelimit"
)
var limitTypes = map[eventratelimitapi.LimitType]bool{
eventratelimitapi.ServerLimitType: true,
eventratelimitapi.NamespaceLimitType: true,
eventratelimitapi.UserLimitType: true,
eventratelimitapi.SourceAndObjectLimitType: true,
}
// ValidateConfiguration validates the configuration.
func ValidateConfiguration(config *eventratelimitapi.Configuration) field.ErrorList {
allErrs := field.ErrorList{}
limitsPath := field.NewPath("limits")
if len(config.Limits) == 0 {
allErrs = append(allErrs, field.Invalid(limitsPath, config.Limits, "must not be empty"))
}
for i, limit := range config.Limits {
idxPath := limitsPath.Index(i)
if !limitTypes[limit.Type] {
allowedValues := make([]string, len(limitTypes))
i := 0
for limitType := range limitTypes {
allowedValues[i] = string(limitType)
i++
}
allErrs = append(allErrs, field.NotSupported(idxPath.Child("type"), limit.Type, allowedValues))
}
if limit.Burst <= 0 {
allErrs = append(allErrs, field.Invalid(idxPath.Child("burst"), limit.Burst, "must be positive"))
}
if limit.QPS <= 0 {
allErrs = append(allErrs, field.Invalid(idxPath.Child("qps"), limit.QPS, "must be positive"))
}
if limit.Type != eventratelimitapi.ServerLimitType {
if limit.CacheSize < 0 {
allErrs = append(allErrs, field.Invalid(idxPath.Child("cacheSize"), limit.CacheSize, "must not be negative"))
}
}
}
return allErrs
}
/*
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 validation
import (
"testing"
eventratelimitapi "k8s.io/kubernetes/plugin/pkg/admission/eventratelimit/apis/eventratelimit"
)
func TestValidateConfiguration(t *testing.T) {
cases := []struct {
name string
config eventratelimitapi.Configuration
expectedResult bool
}{
{
name: "valid server",
config: eventratelimitapi.Configuration{
Limits: []eventratelimitapi.Limit{
{
Type: "Server",
Burst: 5,
QPS: 1,
},
},
},
expectedResult: true,
},
{
name: "valid namespace",
config: eventratelimitapi.Configuration{
Limits: []eventratelimitapi.Limit{
{
Type: "Namespace",
Burst: 10,
QPS: 2,
CacheSize: 100,
},
},
},
expectedResult: true,
},
{
name: "valid user",
config: eventratelimitapi.Configuration{
Limits: []eventratelimitapi.Limit{
{
Type: "User",
Burst: 10,
QPS: 2,
CacheSize: 100,
},
},
},
expectedResult: true,
},
{
name: "valid source+object",
config: eventratelimitapi.Configuration{
Limits: []eventratelimitapi.Limit{
{
Type: "SourceAndObject",
Burst: 5,
QPS: 1,
CacheSize: 1000,
},
},
},
expectedResult: true,
},
{
name: "valid multiple",
config: eventratelimitapi.Configuration{
Limits: []eventratelimitapi.Limit{
{
Type: "Server",
Burst: 5,
QPS: 1,
},
{
Type: "Namespace",
Burst: 10,
QPS: 2,
CacheSize: 100,
},
{
Type: "SourceAndObject",
Burst: 25,
QPS: 10,
CacheSize: 1000,
},
},
},
expectedResult: true,
},
{
name: "missing limits",
config: eventratelimitapi.Configuration{},
expectedResult: false,
},
{
name: "missing type",
config: eventratelimitapi.Configuration{
Limits: []eventratelimitapi.Limit{
{
Burst: 25,
QPS: 10,
CacheSize: 1000,
},
},
},
expectedResult: false,
},
{
name: "invalid type",
config: eventratelimitapi.Configuration{
Limits: []eventratelimitapi.Limit{
{
Type: "unknown-type",
Burst: 25,
QPS: 10,
CacheSize: 1000,
},
},
},
expectedResult: false,
},
{
name: "missing burst",
config: eventratelimitapi.Configuration{
Limits: []eventratelimitapi.Limit{
{
Type: "Server",
QPS: 1,
},
},
},
expectedResult: false,
},
{
name: "missing qps",
config: eventratelimitapi.Configuration{
Limits: []eventratelimitapi.Limit{
{
Type: "Server",
Burst: 5,
},
},
},
expectedResult: false,
},
{
name: "negative cache size",
config: eventratelimitapi.Configuration{
Limits: []eventratelimitapi.Limit{
{
Type: "Namespace",
Burst: 10,
QPS: 2,
CacheSize: -1,
},
},
},
expectedResult: false,
},
}
for _, tc := range cases {
errs := ValidateConfiguration(&tc.config)
if e, a := tc.expectedResult, len(errs) == 0; e != a {
if e {
t.Errorf("%v: expected success: %v", tc.name, errs)
} else {
t.Errorf("%v: expected failure", tc.name)
}
}
}
}
// +build !ignore_autogenerated
/*
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.
*/
// This file was autogenerated by deepcopy-gen. Do not edit it manually!
package eventratelimit
import (
conversion "k8s.io/apimachinery/pkg/conversion"
runtime "k8s.io/apimachinery/pkg/runtime"
reflect "reflect"
)
func init() {
SchemeBuilder.Register(RegisterDeepCopies)
}
// RegisterDeepCopies adds deep-copy functions to the given scheme. Public
// to allow building arbitrary schemes.
//
// Deprecated: deepcopy registration will go away when static deepcopy is fully implemented.
func RegisterDeepCopies(scheme *runtime.Scheme) error {
return scheme.AddGeneratedDeepCopyFuncs(
conversion.GeneratedDeepCopyFunc{Fn: func(in interface{}, out interface{}, c *conversion.Cloner) error {
in.(*Configuration).DeepCopyInto(out.(*Configuration))
return nil
}, InType: reflect.TypeOf(&Configuration{})},
conversion.GeneratedDeepCopyFunc{Fn: func(in interface{}, out interface{}, c *conversion.Cloner) error {
in.(*Limit).DeepCopyInto(out.(*Limit))
return nil
}, InType: reflect.TypeOf(&Limit{})},
)
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *Configuration) DeepCopyInto(out *Configuration) {
*out = *in
out.TypeMeta = in.TypeMeta
if in.Limits != nil {
in, out := &in.Limits, &out.Limits
*out = make([]Limit, len(*in))
copy(*out, *in)
}
return
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Configuration.
func (in *Configuration) DeepCopy() *Configuration {
if in == nil {
return nil
}
out := new(Configuration)
in.DeepCopyInto(out)
return out
}
// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
func (in *Configuration) DeepCopyObject() runtime.Object {
if c := in.DeepCopy(); c != nil {
return c
} else {
return nil
}
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *Limit) DeepCopyInto(out *Limit) {
*out = *in
return
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Limit.
func (in *Limit) DeepCopy() *Limit {
if in == nil {
return nil
}
out := new(Limit)
in.DeepCopyInto(out)
return out
}
/*
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 eventratelimit
import (
"github.com/hashicorp/golang-lru"
"k8s.io/client-go/util/flowcontrol"
)
// cache is an interface for caching the limits of a particular type
type cache interface {
// get the rate limiter associated with the specified key
get(key interface{}) flowcontrol.RateLimiter
}
// singleCache is a cache that only stores a single, constant item
type singleCache struct {
// the single rate limiter held by the cache
rateLimiter flowcontrol.RateLimiter
}
func (c *singleCache) get(key interface{}) flowcontrol.RateLimiter {
return c.rateLimiter
}
// lruCache is a least-recently-used cache
type lruCache struct {
// factory to use to create new rate limiters
rateLimiterFactory func() flowcontrol.RateLimiter
// the actual LRU cache
cache *lru.Cache
}
func (c *lruCache) get(key interface{}) flowcontrol.RateLimiter {
value, found := c.cache.Get(key)
if !found {
rateLimter := c.rateLimiterFactory()
c.cache.Add(key, rateLimter)
return rateLimter
}
return value.(flowcontrol.RateLimiter)
}
/*
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 eventratelimit
import (
"testing"
"github.com/hashicorp/golang-lru"
"k8s.io/client-go/util/flowcontrol"
)
func TestSingleCache(t *testing.T) {
rateLimiter := flowcontrol.NewTokenBucketRateLimiter(1., 1)
cache := singleCache{
rateLimiter: rateLimiter,
}
cases := []interface{}{nil, "key1", "key2"}
for _, tc := range cases {
actual := cache.get(tc)
if e, a := rateLimiter, actual; e != a {
t.Errorf("unexpected entry in cache for key %v: expected %v, got %v", tc, e, a)
}
}
}
func TestLRUCache(t *testing.T) {
rateLimiters := []flowcontrol.RateLimiter{
flowcontrol.NewTokenBucketRateLimiter(1., 1),
flowcontrol.NewTokenBucketRateLimiter(2., 2),
flowcontrol.NewTokenBucketRateLimiter(3., 3),
flowcontrol.NewTokenBucketRateLimiter(4., 4),
}
nextRateLimiter := 0
rateLimiterFactory := func() flowcontrol.RateLimiter {
rateLimiter := rateLimiters[nextRateLimiter]
nextRateLimiter++
return rateLimiter
}
underlyingCache, err := lru.New(2)
if err != nil {
t.Fatalf("Could not create LRU cache: %v", err)
}
cache := lruCache{
rateLimiterFactory: rateLimiterFactory,
cache: underlyingCache,
}
cases := []struct {
name string
key int
expected flowcontrol.RateLimiter
}{
{
name: "first added",
key: 0,
expected: rateLimiters[0],
},
{
name: "first obtained",
key: 0,
expected: rateLimiters[0],
},
{
name: "second added",
key: 1,
expected: rateLimiters[1],
},
{
name: "second obtained",
key: 1,
expected: rateLimiters[1],
},
{
name: "first obtained second time",
key: 0,
expected: rateLimiters[0],
},
{
name: "third added",
key: 2,
expected: rateLimiters[2],
},
{
name: "third obtained",
key: 2,
expected: rateLimiters[2],
},
{
name: "first obtained third time",
key: 0,
expected: rateLimiters[0],
},
{
name: "second re-added after eviction",
key: 1,
expected: rateLimiters[3],
},
}
for _, tc := range cases {
actual := cache.get(tc.key)
if e, a := tc.expected, actual; e != a {
t.Errorf("%v: unexpected entry in cache for key %v: expected %v, got %v", tc.name, tc.key, e, a)
}
}
}
/*
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 eventratelimit
import (
"time"
)
// realClock implements flowcontrol.Clock in terms of standard time functions.
type realClock struct{}
// Now is identical to time.Now.
func (realClock) Now() time.Time {
return time.Now()
}
// Sleep is identical to time.Sleep.
func (realClock) Sleep(d time.Duration) {
time.Sleep(d)
}
/*
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 eventratelimit
import (
"fmt"
"io"
"io/ioutil"
"os"
"k8s.io/apimachinery/pkg/apimachinery/announced"
"k8s.io/apimachinery/pkg/apimachinery/registered"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/serializer"
eventratelimitapi "k8s.io/kubernetes/plugin/pkg/admission/eventratelimit/apis/eventratelimit"
"k8s.io/kubernetes/plugin/pkg/admission/eventratelimit/apis/eventratelimit/install"
eventratelimitv1alpha1 "k8s.io/kubernetes/plugin/pkg/admission/eventratelimit/apis/eventratelimit/v1alpha1"
)
var (
groupFactoryRegistry = make(announced.APIGroupFactoryRegistry)
registry = registered.NewOrDie(os.Getenv("KUBE_API_VERSIONS"))
scheme = runtime.NewScheme()
codecs = serializer.NewCodecFactory(scheme)
)
func init() {
install.Install(groupFactoryRegistry, registry, scheme)
}
// LoadConfiguration loads the provided configuration.
func LoadConfiguration(config io.Reader) (*eventratelimitapi.Configuration, error) {
// if no config is provided, return a default configuration
if config == nil {
externalConfig := &eventratelimitv1alpha1.Configuration{}
scheme.Default(externalConfig)
internalConfig := &eventratelimitapi.Configuration{}
if err := scheme.Convert(externalConfig, internalConfig, nil); err != nil {
return nil, err
}
return internalConfig, nil
}
// we have a config so parse it.
data, err := ioutil.ReadAll(config)
if err != nil {
return nil, err
}
decoder := codecs.UniversalDecoder()
decodedObj, err := runtime.Decode(decoder, data)
if err != nil {
return nil, err
}
resourceQuotaConfiguration, ok := decodedObj.(*eventratelimitapi.Configuration)
if !ok {
return nil, fmt.Errorf("unexpected type: %T", decodedObj)
}
return resourceQuotaConfiguration, nil
}
/*
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 eventratelimit contains an admission controller that enforces a rate limit on events
package eventratelimit // import "k8s.io/kubernetes/plugin/pkg/admission/eventratelimit"
/*
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 eventratelimit
import (
"fmt"
"strings"
"github.com/hashicorp/golang-lru"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apiserver/pkg/admission"
"k8s.io/client-go/util/flowcontrol"
"k8s.io/kubernetes/pkg/api"
eventratelimitapi "k8s.io/kubernetes/plugin/pkg/admission/eventratelimit/apis/eventratelimit"
)
const (
// cache size to use if the user did not specify a cache size
defaultCacheSize = 4096
)
// limitEnforcer enforces a single type of event rate limit, such as server, namespace, or source+object
type limitEnforcer struct {
// type of this limit
limitType eventratelimitapi.LimitType
// cache for holding the rate limiters
cache cache
// a keyFunc which is responsible for computing a single key based on input
keyFunc func(admission.Attributes) string
}
func newLimitEnforcer(config eventratelimitapi.Limit, clock flowcontrol.Clock) (*limitEnforcer, error) {
rateLimiterFactory := func() flowcontrol.RateLimiter {
return flowcontrol.NewTokenBucketRateLimiterWithClock(float32(config.QPS), int(config.Burst), clock)
}
if config.Type == eventratelimitapi.ServerLimitType {
return &limitEnforcer{
limitType: config.Type,
cache: &singleCache{
rateLimiter: rateLimiterFactory(),
},
keyFunc: getServerKey,
}, nil
}
cacheSize := int(config.CacheSize)
if cacheSize == 0 {
cacheSize = defaultCacheSize
}
underlyingCache, err := lru.New(cacheSize)
if err != nil {
return nil, fmt.Errorf("could not create lru cache: %v", err)
}
cache := &lruCache{
rateLimiterFactory: rateLimiterFactory,
cache: underlyingCache,
}
var keyFunc func(admission.Attributes) string
switch t := config.Type; t {
case eventratelimitapi.NamespaceLimitType:
keyFunc = getNamespaceKey
case eventratelimitapi.UserLimitType:
keyFunc = getUserKey
case eventratelimitapi.SourceAndObjectLimitType:
keyFunc = getSourceAndObjectKey
default:
return nil, fmt.Errorf("unknown event rate limit type: %v", t)
}
return &limitEnforcer{
limitType: config.Type,
cache: cache,
keyFunc: keyFunc,
}, nil
}
func (enforcer *limitEnforcer) accept(attr admission.Attributes) error {
key := enforcer.keyFunc(attr)
rateLimiter := enforcer.cache.get(key)
// ensure we have available rate
allow := rateLimiter.TryAccept()
if !allow {
return apierrors.NewTooManyRequestsError(fmt.Sprintf("limit reached on type %v for key %v", enforcer.limitType, key))
}
return nil
}
func getServerKey(attr admission.Attributes) string {
return ""
}
// getNamespaceKey returns a cache key that is based on the namespace of the event request
func getNamespaceKey(attr admission.Attributes) string {
return attr.GetNamespace()
}
// getUserKey returns a cache key that is based on the user of the event request
func getUserKey(attr admission.Attributes) string {
userInfo := attr.GetUserInfo()
if userInfo == nil {
return ""
}
return userInfo.GetName()
}
// getSourceAndObjectKey returns a cache key that is based on the source+object of the event
func getSourceAndObjectKey(attr admission.Attributes) string {
object := attr.GetObject()
if object == nil {
return ""
}
event, ok := object.(*api.Event)
if !ok {
return ""
}
return strings.Join([]string{
event.Source.Component,
event.Source.Host,
event.InvolvedObject.Kind,
event.InvolvedObject.Namespace,
event.InvolvedObject.Name,
string(event.InvolvedObject.UID),
event.InvolvedObject.APIVersion,
}, "")
}
......@@ -28,6 +28,12 @@ import (
"k8s.io/apimachinery/pkg/util/validation/field"
)
const (
// StatusTooManyRequests means the server experienced too many requests within a
// given window and that the client must wait to perform the action again.
StatusTooManyRequests = 429
)
// StatusError is an error intended for consumption by a REST API server; it can also be
// reconstructed by clients from a REST response. Public to allow easy type switches.
type StatusError struct {
......@@ -298,6 +304,18 @@ func NewTimeoutError(message string, retryAfterSeconds int) *StatusError {
}}
}
// NewTooManyRequestsError returns an error indicating that the request was rejected because
// the server has received too many requests. Client should wait and retry. But if the request
// is perishable, then the client should not retry the request.
func NewTooManyRequestsError(message string) *StatusError {
return &StatusError{metav1.Status{
Status: metav1.StatusFailure,
Code: StatusTooManyRequests,
Reason: metav1.StatusReasonTooManyRequests,
Message: fmt.Sprintf("Too many requests: %s", message),
}}
}
// NewGenericServerResponse returns a new error for server responses that are not in a recognizable form.
func NewGenericServerResponse(code int, verb string, qualifiedResource schema.GroupResource, name, serverMessage string, retryAfterSeconds int, isUnexpectedResponse bool) *StatusError {
reason := metav1.StatusReasonUnknown
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment