move initializer to the generic apiserver

move k8s.io/kubernetes/plugin/pkg/admission/initialization to
k8s.io/apiserver/pkg/admission/plugin/initialization/initialization.go;
move k8s.io/kubernetes/pkg/kubeapiserver/admission/configuration to
k8s.io/apiserver/pkg/admission/configuration.

Kubernetes-commit: 89a0511fcb22caf23427587c026952b2a387f293

pkg/admission/configuration/BUILD

@@ -0,0 +1,55 @@
+package(default_visibility = ["//visibility:public"])
+
+load(
+    "@io_bazel_rules_go//go:def.bzl",
+    "go_library",
+    "go_test",
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = [
+        "configuration_manager_test.go",
+        "external_admission_hook_manager_test.go",
+        "initializer_manager_test.go",
+    ],
+    library = ":go_default_library",
+    deps = [
+        "//vendor/k8s.io/api/admissionregistration/v1alpha1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+    ],
+)
+
+go_library(
+    name = "go_default_library",
+    srcs = [
+        "configuration_manager.go",
+        "external_admission_hook_manager.go",
+        "initializer_manager.go",
+    ],
+    deps = [
+        "//vendor/github.com/golang/glog:go_default_library",
+        "//vendor/k8s.io/api/admissionregistration/v1alpha1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+    ],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+)

pkg/admission/configuration/configuration_manager.go

@@ -0,0 +1,165 @@
+/*
+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 configuration
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+)
+
+const (
+	defaultInterval             = 1 * time.Second
+	defaultFailureThreshold     = 5
+	defaultBootstrapRetries     = 5
+	defaultBootstrapGraceperiod = 5 * time.Second
+)
+
+var (
+	ErrNotReady = fmt.Errorf("configuration is not ready")
+	ErrDisabled = fmt.Errorf("disabled")
+)
+
+type getFunc func() (runtime.Object, error)
+
+// When running, poller calls `get` every `interval`. If `get` is
+// successful, `Ready()` returns ready and `configuration()` returns the
+// `mergedConfiguration`; if `get` has failed more than `failureThreshold ` times,
+// `Ready()` returns not ready and `configuration()` returns nil configuration.
+// In an HA setup, the poller is consistent only if the `get` is
+// doing consistent read.
+type poller struct {
+	// a function to consistently read the latest configuration
+	get getFunc
+	// consistent read interval
+	// read-only
+	interval time.Duration
+	// if the number of consecutive read failure equals or exceeds the failureThreshold , the
+	// configuration is regarded as not ready.
+	// read-only
+	failureThreshold int
+	// number of consecutive failures so far.
+	failures int
+	// If the poller has passed the bootstrap phase. The poller is considered
+	// bootstrapped either bootstrapGracePeriod after the first call of
+	// configuration(), or when setConfigurationAndReady() is called, whichever
+	// comes first.
+	bootstrapped bool
+	// configuration() retries bootstrapRetries times if poller is not bootstrapped
+	// read-only
+	bootstrapRetries int
+	// Grace period for bootstrapping
+	// read-only
+	bootstrapGracePeriod time.Duration
+	once                 sync.Once
+	// if the configuration is regarded as ready.
+	ready               bool
+	mergedConfiguration runtime.Object
+	lastErr             error
+	// lock must be hold when reading/writing the data fields of poller.
+	lock sync.RWMutex
+}
+
+func newPoller(get getFunc) *poller {
+	p := poller{
+		get:                  get,
+		interval:             defaultInterval,
+		failureThreshold:     defaultFailureThreshold,
+		bootstrapRetries:     defaultBootstrapRetries,
+		bootstrapGracePeriod: defaultBootstrapGraceperiod,
+	}
+	return &p
+}
+
+func (a *poller) lastError(err error) {
+	a.lock.Lock()
+	defer a.lock.Unlock()
+	a.lastErr = err
+}
+
+func (a *poller) notReady() {
+	a.lock.Lock()
+	defer a.lock.Unlock()
+	a.ready = false
+}
+
+func (a *poller) bootstrapping() {
+	// bootstrapGracePeriod is read-only, so no lock is required
+	timer := time.NewTimer(a.bootstrapGracePeriod)
+	go func() {
+		<-timer.C
+		a.lock.Lock()
+		defer a.lock.Unlock()
+		a.bootstrapped = true
+	}()
+}
+
+// If the poller is not bootstrapped yet, the configuration() gets a few chances
+// to retry. This hides transient failures during system startup.
+func (a *poller) configuration() (runtime.Object, error) {
+	a.once.Do(a.bootstrapping)
+	a.lock.RLock()
+	defer a.lock.RUnlock()
+	retries := 1
+	if !a.bootstrapped {
+		retries = a.bootstrapRetries
+	}
+	for count := 0; count < retries; count++ {
+		if count > 0 {
+			a.lock.RUnlock()
+			time.Sleep(a.interval)
+			a.lock.RLock()
+		}
+		if a.ready {
+			return a.mergedConfiguration, nil
+		}
+	}
+	if a.lastErr != nil {
+		return nil, a.lastErr
+	}
+	return nil, ErrNotReady
+}
+
+func (a *poller) setConfigurationAndReady(value runtime.Object) {
+	a.lock.Lock()
+	defer a.lock.Unlock()
+	a.bootstrapped = true
+	a.mergedConfiguration = value
+	a.ready = true
+	a.lastErr = nil
+}
+
+func (a *poller) Run(stopCh <-chan struct{}) {
+	go wait.Until(a.sync, a.interval, stopCh)
+}
+
+func (a *poller) sync() {
+	configuration, err := a.get()
+	if err != nil {
+		a.failures++
+		a.lastError(err)
+		if a.failures >= a.failureThreshold {
+			a.notReady()
+		}
+		return
+	}
+	a.failures = 0
+	a.setConfigurationAndReady(configuration)
+}

pkg/admission/configuration/configuration_manager_test.go

@@ -0,0 +1,93 @@
+/*
+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 configuration
+
+import (
+	"fmt"
+	"math"
+	"sync"
+	"testing"
+	"time"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+)
+
+func TestTolerateBootstrapFailure(t *testing.T) {
+	var fakeGetSucceed bool
+	var fakeGetSucceedLock sync.RWMutex
+	fakeGetFn := func() (runtime.Object, error) {
+		fakeGetSucceedLock.RLock()
+		defer fakeGetSucceedLock.RUnlock()
+		if fakeGetSucceed {
+			return nil, nil
+		} else {
+			return nil, fmt.Errorf("this error shouldn't be exposed to caller")
+		}
+	}
+	poller := newPoller(fakeGetFn)
+	poller.bootstrapGracePeriod = 100 * time.Second
+	poller.bootstrapRetries = math.MaxInt32
+	// set failureThreshold to 0 so that one single failure will set "ready" to false.
+	poller.failureThreshold = 0
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+	go poller.Run(stopCh)
+	go func() {
+		// The test might have false negative, but won't be flaky
+		timer := time.NewTimer(2 * time.Second)
+		<-timer.C
+		fakeGetSucceedLock.Lock()
+		defer fakeGetSucceedLock.Unlock()
+		fakeGetSucceed = true
+	}()
+
+	done := make(chan struct{})
+	go func(t *testing.T) {
+		_, err := poller.configuration()
+		if err != nil {
+			t.Errorf("unexpected error: %v", err)
+		}
+		close(done)
+	}(t)
+	<-done
+}
+
+func TestNotTolerateNonbootstrapFailure(t *testing.T) {
+	fakeGetFn := func() (runtime.Object, error) {
+		return nil, fmt.Errorf("this error should be exposed to caller")
+	}
+	poller := newPoller(fakeGetFn)
+	poller.bootstrapGracePeriod = 1 * time.Second
+	poller.interval = 1 * time.Millisecond
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+	go poller.Run(stopCh)
+	// to kick the bootstrap timer
+	go poller.configuration()
+
+	wait.PollInfinite(1*time.Second, func() (bool, error) {
+		poller.lock.Lock()
+		defer poller.lock.Unlock()
+		return poller.bootstrapped, nil
+	})
+
+	_, err := poller.configuration()
+	if err == nil {
+		t.Errorf("unexpected no error")
+	}
+}

pkg/admission/configuration/external_admission_hook_manager.go

@@ -0,0 +1,83 @@
+/*
+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 configuration
+
+import (
+	"fmt"
+	"reflect"
+
+	"github.com/golang/glog"
+
+	"k8s.io/api/admissionregistration/v1alpha1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+)
+
+type ExternalAdmissionHookConfigurationLister interface {
+	List(opts metav1.ListOptions) (*v1alpha1.ExternalAdmissionHookConfigurationList, error)
+}
+
+type ExternalAdmissionHookConfigurationManager struct {
+	*poller
+}
+
+func NewExternalAdmissionHookConfigurationManager(c ExternalAdmissionHookConfigurationLister) *ExternalAdmissionHookConfigurationManager {
+	getFn := func() (runtime.Object, error) {
+		list, err := c.List(metav1.ListOptions{})
+		if err != nil {
+			if errors.IsNotFound(err) || errors.IsForbidden(err) {
+				glog.V(5).Infof("ExternalAdmissionHookConfiguration are disabled due to an error: %v", err)
+				return nil, ErrDisabled
+			}
+			return nil, err
+		}
+		return mergeExternalAdmissionHookConfigurations(list), nil
+	}
+
+	return &ExternalAdmissionHookConfigurationManager{
+		newPoller(getFn),
+	}
+}
+
+// ExternalAdmissionHooks returns the merged ExternalAdmissionHookConfiguration.
+func (im *ExternalAdmissionHookConfigurationManager) ExternalAdmissionHooks() (*v1alpha1.ExternalAdmissionHookConfiguration, error) {
+	configuration, err := im.poller.configuration()
+	if err != nil {
+		return nil, err
+	}
+	externalAdmissionHookConfiguration, ok := configuration.(*v1alpha1.ExternalAdmissionHookConfiguration)
+	if !ok {
+		return nil, fmt.Errorf("expected type %v, got type %v", reflect.TypeOf(externalAdmissionHookConfiguration), reflect.TypeOf(configuration))
+	}
+	return externalAdmissionHookConfiguration, nil
+}
+
+func (im *ExternalAdmissionHookConfigurationManager) Run(stopCh <-chan struct{}) {
+	im.poller.Run(stopCh)
+}
+
+func mergeExternalAdmissionHookConfigurations(
+	list *v1alpha1.ExternalAdmissionHookConfigurationList,
+) *v1alpha1.ExternalAdmissionHookConfiguration {
+	configurations := list.Items
+	var ret v1alpha1.ExternalAdmissionHookConfiguration
+	for _, c := range configurations {
+		ret.ExternalAdmissionHooks = append(ret.ExternalAdmissionHooks, c.ExternalAdmissionHooks...)
+	}
+	return &ret
+}

pkg/admission/configuration/external_admission_hook_manager_test.go

@@ -0,0 +1,40 @@
+/*
+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 configuration
+
+import (
+	"testing"
+
+	"k8s.io/api/admissionregistration/v1alpha1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+type disabledWebhookConfigLister struct{}
+
+func (l *disabledWebhookConfigLister) List(options metav1.ListOptions) (*v1alpha1.ExternalAdmissionHookConfigurationList, error) {
+	return nil, errors.NewNotFound(schema.GroupResource{Group: "admissionregistration", Resource: "externalAdmissionHookConfigurations"}, "")
+}
+func TestWebhookConfigDisabled(t *testing.T) {
+	manager := NewExternalAdmissionHookConfigurationManager(&disabledWebhookConfigLister{})
+	manager.sync()
+	_, err := manager.ExternalAdmissionHooks()
+	if err.Error() != ErrDisabled.Error() {
+		t.Errorf("expected %v, got %v", ErrDisabled, err)
+	}
+}

pkg/admission/configuration/initializer_manager.go

@@ -0,0 +1,88 @@
+/*
+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 configuration
+
+import (
+	"fmt"
+	"reflect"
+	"sort"
+
+	"github.com/golang/glog"
+
+	"k8s.io/api/admissionregistration/v1alpha1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+)
+
+type InitializerConfigurationLister interface {
+	List(opts metav1.ListOptions) (*v1alpha1.InitializerConfigurationList, error)
+}
+
+type InitializerConfigurationManager struct {
+	*poller
+}
+
+func NewInitializerConfigurationManager(c InitializerConfigurationLister) *InitializerConfigurationManager {
+	getFn := func() (runtime.Object, error) {
+		list, err := c.List(metav1.ListOptions{})
+		if err != nil {
+			if errors.IsNotFound(err) || errors.IsForbidden(err) {
+				glog.V(5).Infof("Initializers are disabled due to an error: %v", err)
+				return nil, ErrDisabled
+			}
+			return nil, err
+		}
+		return mergeInitializerConfigurations(list), nil
+	}
+	return &InitializerConfigurationManager{
+		newPoller(getFn),
+	}
+}
+
+// Initializers returns the merged InitializerConfiguration.
+func (im *InitializerConfigurationManager) Initializers() (*v1alpha1.InitializerConfiguration, error) {
+	configuration, err := im.poller.configuration()
+	if err != nil {
+		return nil, err
+	}
+	initializerConfiguration, ok := configuration.(*v1alpha1.InitializerConfiguration)
+	if !ok {
+		return nil, fmt.Errorf("expected type %v, got type %v", reflect.TypeOf(initializerConfiguration), reflect.TypeOf(configuration))
+	}
+	return initializerConfiguration, nil
+}
+
+func (im *InitializerConfigurationManager) Run(stopCh <-chan struct{}) {
+	im.poller.Run(stopCh)
+}
+
+func mergeInitializerConfigurations(initializerConfigurationList *v1alpha1.InitializerConfigurationList) *v1alpha1.InitializerConfiguration {
+	configurations := initializerConfigurationList.Items
+	sort.SliceStable(configurations, InitializerConfigurationSorter(configurations).ByName)
+	var ret v1alpha1.InitializerConfiguration
+	for _, c := range configurations {
+		ret.Initializers = append(ret.Initializers, c.Initializers...)
+	}
+	return &ret
+}
+
+type InitializerConfigurationSorter []v1alpha1.InitializerConfiguration
+
+func (a InitializerConfigurationSorter) ByName(i, j int) bool {
+	return a[i].Name < a[j].Name
+}

pkg/admission/configuration/initializer_manager_test.go

@@ -0,0 +1,182 @@
+/*
+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 configuration
+
+import (
+	"fmt"
+	"reflect"
+	"testing"
+	"time"
+
+	"k8s.io/api/admissionregistration/v1alpha1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+type mockLister struct {
+	invoked           int
+	successes         int
+	failures          int
+	configurationList v1alpha1.InitializerConfigurationList
+	t                 *testing.T
+}
+
+func newMockLister(successes, failures int, configurationList v1alpha1.InitializerConfigurationList, t *testing.T) *mockLister {
+	return &mockLister{
+		failures:          failures,
+		successes:         successes,
+		configurationList: configurationList,
+		t:                 t,
+	}
+}
+
+// The first List will be successful; the next m.failures List will
+// fail; the next m.successes List will be successful
+// List should only be called 1+m.failures+m.successes times.
+func (m *mockLister) List(options metav1.ListOptions) (*v1alpha1.InitializerConfigurationList, error) {
+	m.invoked++
+	if m.invoked == 1 {
+		return &m.configurationList, nil
+	}
+	if m.invoked <= 1+m.failures {
+		return nil, fmt.Errorf("some error")
+	}
+	if m.invoked <= 1+m.failures+m.successes {
+		return &m.configurationList, nil
+	}
+	m.t.Fatalf("unexpected call to List, should only be called %d times", 1+m.successes+m.failures)
+	return nil, nil
+}
+
+var _ InitializerConfigurationLister = &mockLister{}
+
+func TestConfiguration(t *testing.T) {
+	cases := []struct {
+		name     string
+		failures int
+		// note that the first call to mockLister is always a success.
+		successes   int
+		expectReady bool
+	}{
+		{
+			name:        "number of failures hasn't reached failureThreshold",
+			failures:    defaultFailureThreshold - 1,
+			expectReady: true,
+		},
+		{
+			name:        "number of failures just reaches failureThreshold",
+			failures:    defaultFailureThreshold,
+			expectReady: false,
+		},
+		{
+			name:        "number of failures exceeds failureThreshold",
+			failures:    defaultFailureThreshold + 1,
+			expectReady: false,
+		},
+		{
+			name:        "number of failures exceeds failureThreshold, but then get another success",
+			failures:    defaultFailureThreshold + 1,
+			successes:   1,
+			expectReady: true,
+		},
+	}
+	for _, c := range cases {
+		mock := newMockLister(c.successes, c.failures, v1alpha1.InitializerConfigurationList{}, t)
+		manager := NewInitializerConfigurationManager(mock)
+		manager.interval = 1 * time.Millisecond
+		for i := 0; i < 1+c.successes+c.failures; i++ {
+			manager.sync()
+		}
+		_, err := manager.Initializers()
+		if err != nil && c.expectReady {
+			t.Errorf("case %s, expect ready, got: %v", c.name, err)
+		}
+		if err == nil && !c.expectReady {
+			t.Errorf("case %s, expect not ready", c.name)
+		}
+	}
+}
+
+func TestMergeInitializerConfigurations(t *testing.T) {
+	configurationsList := v1alpha1.InitializerConfigurationList{
+		Items: []v1alpha1.InitializerConfiguration{
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "provider_2",
+				},
+				Initializers: []v1alpha1.Initializer{
+					{
+						Name: "initializer_a",
+					},
+					{
+						Name: "initializer_b",
+					},
+				},
+			},
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "provider_1",
+				},
+				Initializers: []v1alpha1.Initializer{
+					{
+						Name: "initializer_c",
+					},
+					{
+						Name: "initializer_d",
+					},
+				},
+			},
+		},
+	}
+
+	expected := &v1alpha1.InitializerConfiguration{
+		Initializers: []v1alpha1.Initializer{
+			{
+				Name: "initializer_c",
+			},
+			{
+				Name: "initializer_d",
+			},
+			{
+				Name: "initializer_a",
+			},
+			{
+				Name: "initializer_b",
+			},
+		},
+	}
+
+	got := mergeInitializerConfigurations(&configurationsList)
+	if !reflect.DeepEqual(got, expected) {
+		t.Errorf("expected: %#v, got: %#v", expected, got)
+	}
+}
+
+type disabledInitializerConfigLister struct{}
+
+func (l *disabledInitializerConfigLister) List(options metav1.ListOptions) (*v1alpha1.InitializerConfigurationList, error) {
+	return nil, errors.NewNotFound(schema.GroupResource{Group: "admissionregistration", Resource: "initializerConfigurations"}, "")
+}
+func TestInitializerConfigDisabled(t *testing.T) {
+	manager := NewInitializerConfigurationManager(&disabledInitializerConfigLister{})
+	manager.sync()
+	_, err := manager.Initializers()
+	if err.Error() != ErrDisabled.Error() {
+		t.Errorf("expected %v, got %v", ErrDisabled, err)
+	}
+}

pkg/admission/plugin/initialization/BUILD

@@ -0,0 +1,59 @@
+package(default_visibility = ["//visibility:public"])
+
+load(
+    "@io_bazel_rules_go//go:def.bzl",
+    "go_library",
+    "go_test",
+)
+
+go_library(
+    name = "go_default_library",
+    srcs = ["initialization.go"],
+    deps = [
+        "//pkg/api:go_default_library",
+        "//pkg/kubeapiserver/admission/configuration:go_default_library",
+        "//vendor/github.com/golang/glog:go_default_library",
+        "//vendor/k8s.io/api/admissionregistration/v1alpha1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/api/meta:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/api/validation:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/util/validation/field:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/admission:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/authorization/authorizer:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/features:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library",
+        "//vendor/k8s.io/client-go/kubernetes: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 = ["initialization_test.go"],
+    library = ":go_default_library",
+    deps = [
+        "//vendor/k8s.io/api/admissionregistration/v1alpha1:go_default_library",
+        "//vendor/k8s.io/api/core/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/api/meta:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/admission:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/authorization/authorizer:go_default_library",
+    ],
+)

pkg/admission/plugin/initialization/initialization.go

@@ -0,0 +1,363 @@
+/*
+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 initialization
+
+import (
+	"fmt"
+	"io"
+	"strings"
+
+	"github.com/golang/glog"
+
+	"k8s.io/api/admissionregistration/v1alpha1"
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/api/validation"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/validation/field"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/apiserver/pkg/admission"
+	"k8s.io/apiserver/pkg/admission/configuration"
+	"k8s.io/apiserver/pkg/authorization/authorizer"
+	"k8s.io/apiserver/pkg/features"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	clientset "k8s.io/client-go/kubernetes"
+)
+
+const (
+	// Name of admission plug-in
+	PluginName = "Initializers"
+)
+
+// Register registers a plugin
+func Register(plugins *admission.Plugins) {
+	plugins.Register(PluginName, func(config io.Reader) (admission.Interface, error) {
+		return NewInitializer(), nil
+	})
+}
+
+type initializerOptions struct {
+	Initializers []string
+}
+
+type InitializationConfig interface {
+	Run(stopCh <-chan struct{})
+	Initializers() (*v1alpha1.InitializerConfiguration, error)
+}
+
+type initializer struct {
+	config     InitializationConfig
+	authorizer authorizer.Authorizer
+}
+
+// NewInitializer creates a new initializer plugin which assigns newly created resources initializers
+// based on configuration loaded from the admission API group.
+// FUTURE: this may be moved to the storage layer of the apiserver, but for now this is an alpha feature
+//   that can be disabled.
+func NewInitializer() admission.Interface {
+	return &initializer{}
+}
+
+func (i *initializer) Validate() error {
+	if i.config == nil {
+		return fmt.Errorf("the Initializer admission plugin requires a Kubernetes client to be provided")
+	}
+	if i.authorizer == nil {
+		return fmt.Errorf("the Initializer admission plugin requires an authorizer to be provided")
+	}
+
+	if !utilfeature.DefaultFeatureGate.Enabled(features.Initializers) {
+		if err := utilfeature.DefaultFeatureGate.Set(string(features.Initializers) + "=true"); err != nil {
+			glog.Errorf("error enabling Initializers feature as part of admission plugin setup: %v", err)
+		} else {
+			glog.Infof("enabled Initializers feature as part of admission plugin setup")
+		}
+	}
+
+	i.config.Run(wait.NeverStop)
+	return nil
+}
+
+func (i *initializer) SetExternalKubeClientSet(client clientset.Interface) {
+	i.config = configuration.NewInitializerConfigurationManager(client.Admissionregistration().InitializerConfigurations())
+}
+
+func (i *initializer) SetAuthorizer(a authorizer.Authorizer) {
+	i.authorizer = a
+}
+
+var initializerFieldPath = field.NewPath("metadata", "initializers")
+
+// readConfig holds requests instead of failing them if the server is not yet initialized
+// or is unresponsive. It formats the returned error for client use if necessary.
+func (i *initializer) readConfig(a admission.Attributes) (*v1alpha1.InitializerConfiguration, error) {
+	// read initializers from config
+	config, err := i.config.Initializers()
+	if err == nil {
+		return config, nil
+	}
+
+	// if initializer configuration is disabled, fail open
+	if err == configuration.ErrDisabled {
+		return &v1alpha1.InitializerConfiguration{}, nil
+	}
+
+	e := errors.NewServerTimeout(a.GetResource().GroupResource(), "create", 1)
+	if err == configuration.ErrNotReady {
+		e.ErrStatus.Message = fmt.Sprintf("Waiting for initialization configuration to load: %v", err)
+		e.ErrStatus.Reason = "LoadingConfiguration"
+		e.ErrStatus.Details.Causes = append(e.ErrStatus.Details.Causes, metav1.StatusCause{
+			Type:    "InitializerConfigurationPending",
+			Message: "The server is waiting for the initializer configuration to be loaded.",
+		})
+	} else {
+		e.ErrStatus.Message = fmt.Sprintf("Unable to refresh the initializer configuration: %v", err)
+		e.ErrStatus.Reason = "LoadingConfiguration"
+		e.ErrStatus.Details.Causes = append(e.ErrStatus.Details.Causes, metav1.StatusCause{
+			Type:    "InitializerConfigurationFailure",
+			Message: "An error has occurred while refreshing the initializer configuration, no resources can be created until a refresh succeeds.",
+		})
+	}
+	return nil, e
+}
+
+// Admit checks for create requests to add initializers, or update request to enforce invariants.
+// The admission controller fails open if the object doesn't have ObjectMeta (can't be initialized).
+// A client with sufficient permission ("initialize" verb on resource) can specify its own initializers
+// or an empty initializers struct (which bypasses initialization). Only clients with the initialize verb
+// can update objects that have not completed initialization. Sub resources can still be modified on
+// resources that are undergoing initialization.
+// TODO: once this logic is ready for beta, move it into the REST storage layer.
+func (i *initializer) Admit(a admission.Attributes) (err error) {
+	switch a.GetOperation() {
+	case admission.Create, admission.Update:
+	default:
+		return nil
+	}
+
+	// TODO: should sub-resource action should be denied until the object is initialized?
+	if len(a.GetSubresource()) > 0 {
+		return nil
+	}
+
+	switch a.GetOperation() {
+	case admission.Create:
+		accessor, err := meta.Accessor(a.GetObject())
+		if err != nil {
+			// objects without meta accessor cannot be checked for initialization, and it is possible to make calls
+			// via our API that don't have ObjectMeta
+			return nil
+		}
+		existing := accessor.GetInitializers()
+		if existing != nil {
+			glog.V(5).Infof("Admin bypassing initialization for %s", a.GetResource())
+
+			// it must be possible for some users to bypass initialization - for now, check the initialize operation
+			if err := i.canInitialize(a, "create with initializers denied"); err != nil {
+				return err
+			}
+			// allow administrators to bypass initialization by setting an empty initializers struct
+			if len(existing.Pending) == 0 && existing.Result == nil {
+				accessor.SetInitializers(nil)
+				return nil
+			}
+		} else {
+			glog.V(5).Infof("Checking initialization for %s", a.GetResource())
+
+			config, err := i.readConfig(a)
+			if err != nil {
+				return err
+			}
+
+			// Mirror pods are exempt from initialization because they are created and initialized
+			// on the Kubelet before they appear in the API.
+			// TODO: once this moves to REST storage layer, this becomes a pod specific concern
+			if a.GetKind().GroupKind() == v1.SchemeGroupVersion.WithKind("Pod").GroupKind() {
+				accessor, err := meta.Accessor(a.GetObject())
+				if err != nil {
+					return err
+				}
+				annotations := accessor.GetAnnotations()
+				if _, isMirror := annotations[v1.MirrorPodAnnotationKey]; isMirror {
+					return nil
+				}
+			}
+
+			names := findInitializers(config, a.GetResource())
+			if len(names) == 0 {
+				glog.V(5).Infof("No initializers needed")
+				return nil
+			}
+
+			glog.V(5).Infof("Found initializers for %s: %v", a.GetResource(), names)
+			accessor.SetInitializers(newInitializers(names))
+		}
+
+	case admission.Update:
+		accessor, err := meta.Accessor(a.GetObject())
+		if err != nil {
+			// objects without meta accessor cannot be checked for initialization, and it is possible to make calls
+			// via our API that don't have ObjectMeta
+			return nil
+		}
+		updated := accessor.GetInitializers()
+
+		// controllers deployed with an empty initializers.pending have their initializers set to nil
+		// but should be able to update without changing their manifest
+		if updated != nil && len(updated.Pending) == 0 && updated.Result == nil {
+			accessor.SetInitializers(nil)
+			updated = nil
+		}
+
+		existingAccessor, err := meta.Accessor(a.GetOldObject())
+		if err != nil {
+			// if the old object does not have an accessor, but the new one does, error out
+			return fmt.Errorf("initialized resources must be able to set initializers (%T): %v", a.GetOldObject(), err)
+		}
+		existing := existingAccessor.GetInitializers()
+
+		// updates on initialized resources are allowed
+		if updated == nil && existing == nil {
+			return nil
+		}
+
+		glog.V(5).Infof("Modifying uninitialized resource %s", a.GetResource())
+
+		// because we are called before validation, we need to ensure the update transition is valid.
+		if errs := validation.ValidateInitializersUpdate(updated, existing, initializerFieldPath); len(errs) > 0 {
+			return errors.NewInvalid(a.GetKind().GroupKind(), a.GetName(), errs)
+		}
+
+		// caller must have the ability to mutate un-initialized resources
+		if err := i.canInitialize(a, "update to uninitialized resource denied"); err != nil {
+			return err
+		}
+
+		// TODO: restrict initialization list changes to specific clients?
+	}
+
+	return nil
+}
+
+func (i *initializer) canInitialize(a admission.Attributes, message string) error {
+	// caller must have the ability to mutate un-initialized resources
+	authorized, reason, err := i.authorizer.Authorize(authorizer.AttributesRecord{
+		Name:            a.GetName(),
+		ResourceRequest: true,
+		User:            a.GetUserInfo(),
+		Verb:            "initialize",
+		Namespace:       a.GetNamespace(),
+		APIGroup:        a.GetResource().Group,
+		APIVersion:      a.GetResource().Version,
+		Resource:        a.GetResource().Resource,
+	})
+	if err != nil {
+		return err
+	}
+	if !authorized {
+		return errors.NewForbidden(a.GetResource().GroupResource(), a.GetName(), fmt.Errorf("%s: %s", message, reason))
+	}
+	return nil
+}
+
+func (i *initializer) Handles(op admission.Operation) bool {
+	return op == admission.Create || op == admission.Update
+}
+
+// newInitializers populates an Initializers struct.
+func newInitializers(names []string) *metav1.Initializers {
+	if len(names) == 0 {
+		return nil
+	}
+	var init []metav1.Initializer
+	for _, name := range names {
+		init = append(init, metav1.Initializer{Name: name})
+	}
+	return &metav1.Initializers{
+		Pending: init,
+	}
+}
+
+// findInitializers returns the list of initializer names that apply to a config. It returns an empty list
+// if no initializers apply.
+func findInitializers(initializers *v1alpha1.InitializerConfiguration, gvr schema.GroupVersionResource) []string {
+	var names []string
+	for _, init := range initializers.Initializers {
+		if !matchRule(init.Rules, gvr) {
+			continue
+		}
+		names = append(names, init.Name)
+	}
+	return names
+}
+
+// matchRule returns true if any rule matches the provided group version resource.
+func matchRule(rules []v1alpha1.Rule, gvr schema.GroupVersionResource) bool {
+	for _, rule := range rules {
+		if !hasGroup(rule.APIGroups, gvr.Group) {
+			return false
+		}
+		if !hasVersion(rule.APIVersions, gvr.Version) {
+			return false
+		}
+		if !hasResource(rule.Resources, gvr.Resource) {
+			return false
+		}
+	}
+	return len(rules) > 0
+}
+
+func hasGroup(groups []string, group string) bool {
+	if groups[0] == "*" {
+		return true
+	}
+	for _, g := range groups {
+		if g == group {
+			return true
+		}
+	}
+	return false
+}
+
+func hasVersion(versions []string, version string) bool {
+	if versions[0] == "*" {
+		return true
+	}
+	for _, v := range versions {
+		if v == version {
+			return true
+		}
+	}
+	return false
+}
+
+func hasResource(resources []string, resource string) bool {
+	if resources[0] == "*" || resources[0] == "*/*" {
+		return true
+	}
+	for _, r := range resources {
+		if strings.Contains(r, "/") {
+			continue
+		}
+		if r == resource {
+			return true
+		}
+	}
+	return false
+}

pkg/admission/plugin/initialization/initialization_test.go

@@ -0,0 +1,194 @@
+/*
+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 initialization
+
+import (
+	"reflect"
+	"strings"
+	"testing"
+
+	"k8s.io/api/admissionregistration/v1alpha1"
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apiserver/pkg/admission"
+	"k8s.io/apiserver/pkg/authorization/authorizer"
+)
+
+func newInitializer(name string, rules ...v1alpha1.Rule) *v1alpha1.InitializerConfiguration {
+	return addInitializer(&v1alpha1.InitializerConfiguration{}, name, rules...)
+}
+
+func addInitializer(base *v1alpha1.InitializerConfiguration, name string, rules ...v1alpha1.Rule) *v1alpha1.InitializerConfiguration {
+	base.Initializers = append(base.Initializers, v1alpha1.Initializer{
+		Name:  name,
+		Rules: rules,
+	})
+	return base
+}
+
+func TestFindInitializers(t *testing.T) {
+	type args struct {
+		initializers *v1alpha1.InitializerConfiguration
+		gvr          schema.GroupVersionResource
+	}
+	tests := []struct {
+		name string
+		args args
+		want []string
+	}{
+		{
+			name: "empty",
+			args: args{
+				gvr:          schema.GroupVersionResource{},
+				initializers: newInitializer("1"),
+			},
+		},
+		{
+			name: "everything",
+			args: args{
+				gvr:          schema.GroupVersionResource{},
+				initializers: newInitializer("1", v1alpha1.Rule{APIGroups: []string{"*"}, APIVersions: []string{"*"}, Resources: []string{"*"}}),
+			},
+			want: []string{"1"},
+		},
+		{
+			name: "empty group",
+			args: args{
+				gvr:          schema.GroupVersionResource{},
+				initializers: newInitializer("1", v1alpha1.Rule{APIGroups: []string{""}, APIVersions: []string{"*"}, Resources: []string{"*"}}),
+			},
+			want: []string{"1"},
+		},
+		{
+			name: "pod",
+			args: args{
+				gvr: schema.GroupVersionResource{Resource: "pods"},
+				initializers: addInitializer(
+					newInitializer("1", v1alpha1.Rule{APIGroups: []string{""}, APIVersions: []string{"*"}, Resources: []string{"pods"}}),
+					"2", v1alpha1.Rule{APIGroups: []string{""}, APIVersions: []string{"*"}, Resources: []string{"pods"}},
+				),
+			},
+			want: []string{"1", "2"},
+		},
+		{
+			name: "multiple matches",
+			args: args{
+				gvr:          schema.GroupVersionResource{Resource: "pods"},
+				initializers: newInitializer("1", v1alpha1.Rule{APIGroups: []string{""}, APIVersions: []string{"*"}, Resources: []string{"pods"}}),
+			},
+			want: []string{"1"},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := findInitializers(tt.args.initializers, tt.args.gvr); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("findInitializers() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+type fakeAuthorizer struct {
+	accept bool
+}
+
+func (f *fakeAuthorizer) Authorize(a authorizer.Attributes) (bool, string, error) {
+	if f.accept {
+		return true, "", nil
+	}
+	return false, "denied", nil
+}
+
+func TestAdmitUpdate(t *testing.T) {
+	tests := []struct {
+		name             string
+		oldInitializers  *metav1.Initializers
+		newInitializers  *metav1.Initializers
+		verifyUpdatedObj func(runtime.Object) (pass bool, reason string)
+		err              string
+	}{
+		{
+			name:            "updates on initialized resources are allowed",
+			oldInitializers: nil,
+			newInitializers: nil,
+			err:             "",
+		},
+		{
+			name:            "updates on initialized resources are allowed",
+			oldInitializers: &metav1.Initializers{Pending: []metav1.Initializer{{Name: "init.k8s.io"}}},
+			newInitializers: &metav1.Initializers{},
+			verifyUpdatedObj: func(obj runtime.Object) (bool, string) {
+				accessor, err := meta.Accessor(obj)
+				if err != nil {
+					return false, "cannot get accessor"
+				}
+				if accessor.GetInitializers() != nil {
+					return false, "expect nil initializers"
+				}
+				return true, ""
+			},
+			err: "",
+		},
+		{
+			name:            "initializers may not be set once initialized",
+			oldInitializers: nil,
+			newInitializers: &metav1.Initializers{Pending: []metav1.Initializer{{Name: "init.k8s.io"}}},
+			err:             "field is immutable once initialization has completed",
+		},
+		{
+			name:            "empty initializer list is treated as nil initializer",
+			oldInitializers: nil,
+			newInitializers: &metav1.Initializers{},
+			verifyUpdatedObj: func(obj runtime.Object) (bool, string) {
+				accessor, err := meta.Accessor(obj)
+				if err != nil {
+					return false, "cannot get accessor"
+				}
+				if accessor.GetInitializers() != nil {
+					return false, "expect nil initializers"
+				}
+				return true, ""
+			},
+			err: "",
+		},
+	}
+
+	plugin := initializer{
+		config:     nil,
+		authorizer: &fakeAuthorizer{true},
+	}
+	for _, tc := range tests {
+		oldObj := &v1.Pod{}
+		oldObj.Initializers = tc.oldInitializers
+		newObj := &v1.Pod{}
+		newObj.Initializers = tc.newInitializers
+		a := admission.NewAttributesRecord(newObj, oldObj, schema.GroupVersionKind{}, "", "foo", schema.GroupVersionResource{}, "", admission.Update, nil)
+		err := plugin.Admit(a)
+		switch {
+		case tc.err == "" && err != nil:
+			t.Errorf("%q: unexpected error: %v", tc.name, err)
+		case tc.err != "" && err == nil:
+			t.Errorf("%q: unexpected no error, expected %s", tc.name, tc.err)
+		case tc.err != "" && err != nil && !strings.Contains(err.Error(), tc.err):
+			t.Errorf("%q: expected %s, got %v", tc.name, tc.err, err)
+		}
+	}
+
+}

pkg/server/options/admission.go

@@ -23,6 +23,7 @@ import (
 	"github.com/spf13/pflag"
 	"k8s.io/apiserver/pkg/admission"
 	"k8s.io/apiserver/pkg/admission/initializer"
+	"k8s.io/apiserver/pkg/admission/plugin/initialization"
 	"k8s.io/apiserver/pkg/admission/plugin/namespace/lifecycle"
 	"k8s.io/apiserver/pkg/server"
 	"k8s.io/client-go/informers"
@@ -53,7 +54,8 @@ func NewAdmissionOptions() *AdmissionOptions {
 	options := &AdmissionOptions{
 		Plugins:                &admission.Plugins{},
 		PluginNames:            []string{},
-		RecommendedPluginOrder: []string{lifecycle.PluginName},
+		RecommendedPluginOrder: []string{lifecycle.PluginName, initialization.PluginName},
+		DefaultOffPlugins:      []string{initialization.PluginName},
 	}
 	server.RegisterAllAdmissionPlugins(options.Plugins)
 	return options

pkg/server/options/admission_test.go

@@ -56,7 +56,7 @@ func TestEnabledPluginNamesMethod(t *testing.T) {
 			actualPluginNames := target.enabledPluginNames()
 
 			if len(actualPluginNames) != len(scenario.expectedPluginNames) {
-				t.Errorf("incorrect number of items, got %d, expected = %d", len(actualPluginNames), len(scenario.expectedPluginNames))
+				t.Fatalf("incorrect number of items, got %d, expected = %d", len(actualPluginNames), len(scenario.expectedPluginNames))
 			}
 			for i := range actualPluginNames {
 				if scenario.expectedPluginNames[i] != actualPluginNames[i] {

pkg/server/plugins.go

@@ -19,10 +19,12 @@ package server
 // This file exists to force the desired plugin implementations to be linked into genericapi pkg.
 import (
 	"k8s.io/apiserver/pkg/admission"
+	"k8s.io/apiserver/pkg/admission/plugin/initialization"
 	"k8s.io/apiserver/pkg/admission/plugin/namespace/lifecycle"
 )
 
 // RegisterAllAdmissionPlugins registers all admission plugins
 func RegisterAllAdmissionPlugins(plugins *admission.Plugins) {
 	lifecycle.Register(plugins)
+	initialization.Register(plugins)
 }