mirror of https://github.com/k3s-io/k3s
[Federation] Convert Daemonset to use the generic sync controller
Christian Bell
parent
5336f2de05
commit
15e81959e9
|
|
@ -24,7 +24,6 @@ go_library(
|
|||
"//federation/pkg/dnsprovider/providers/google/clouddns:go_default_library",
|
||||
"//federation/pkg/federatedtypes:go_default_library",
|
||||
"//federation/pkg/federation-controller/cluster:go_default_library",
|
||||
"//federation/pkg/federation-controller/daemonset:go_default_library",
|
||||
"//federation/pkg/federation-controller/deployment:go_default_library",
|
||||
"//federation/pkg/federation-controller/ingress:go_default_library",
|
||||
"//federation/pkg/federation-controller/namespace:go_default_library",
|
||||
|
|
|
|||
|
|
@ -38,7 +38,6 @@ import (
|
|||
"k8s.io/kubernetes/federation/pkg/dnsprovider"
|
||||
"k8s.io/kubernetes/federation/pkg/federatedtypes"
|
||||
clustercontroller "k8s.io/kubernetes/federation/pkg/federation-controller/cluster"
|
||||
daemonsetcontroller "k8s.io/kubernetes/federation/pkg/federation-controller/daemonset"
|
||||
deploymentcontroller "k8s.io/kubernetes/federation/pkg/federation-controller/deployment"
|
||||
ingresscontroller "k8s.io/kubernetes/federation/pkg/federation-controller/ingress"
|
||||
namespacecontroller "k8s.io/kubernetes/federation/pkg/federation-controller/namespace"
|
||||
|
|
@ -165,12 +164,6 @@ func StartControllers(s *options.CMServer, restClientCfg *restclient.Config) err
|
|||
}
|
||||
}
|
||||
|
||||
if controllerEnabled(s.Controllers, serverResources, daemonsetcontroller.ControllerName, daemonsetcontroller.RequiredResources, true) {
|
||||
daemonsetcontrollerClientset := federationclientset.NewForConfigOrDie(restclient.AddUserAgent(restClientCfg, "daemonset-controller"))
|
||||
daemonsetcontroller := daemonsetcontroller.NewDaemonSetController(daemonsetcontrollerClientset)
|
||||
daemonsetcontroller.Run(wait.NeverStop)
|
||||
}
|
||||
|
||||
if controllerEnabled(s.Controllers, serverResources, replicasetcontroller.ControllerName, replicasetcontroller.RequiredResources, true) {
|
||||
replicaSetClientset := federationclientset.NewForConfigOrDie(restclient.AddUserAgent(restClientCfg, replicasetcontroller.UserAgentName))
|
||||
replicaSetController := replicasetcontroller.NewReplicaSetController(replicaSetClientset)
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@ go_library(
|
|||
srcs = [
|
||||
"adapter.go",
|
||||
"configmap.go",
|
||||
"daemonset.go",
|
||||
"registry.go",
|
||||
"secret.go",
|
||||
],
|
||||
|
|
@ -20,6 +21,7 @@ go_library(
|
|||
"//federation/client/clientset_generated/federation_clientset:go_default_library",
|
||||
"//federation/pkg/federation-controller/util:go_default_library",
|
||||
"//pkg/api/v1:go_default_library",
|
||||
"//pkg/apis/extensions/v1beta1:go_default_library",
|
||||
"//pkg/client/clientset_generated/clientset:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
|
||||
|
|
|
|||
|
|
@ -0,0 +1,163 @@
|
|||
/*
|
||||
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 federatedtypes
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
pkgruntime "k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/types"
|
||||
"k8s.io/apimachinery/pkg/watch"
|
||||
federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset"
|
||||
"k8s.io/kubernetes/federation/pkg/federation-controller/util"
|
||||
"k8s.io/kubernetes/pkg/api/v1"
|
||||
extensionsv1 "k8s.io/kubernetes/pkg/apis/extensions/v1beta1"
|
||||
kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
|
||||
)
|
||||
|
||||
const (
|
||||
DaemonSetKind = "daemonset"
|
||||
DaemonSetControllerName = "daemonsets"
|
||||
)
|
||||
|
||||
func init() {
|
||||
RegisterFederatedType(DaemonSetKind, DaemonSetControllerName, []schema.GroupVersionResource{extensionsv1.SchemeGroupVersion.WithResource(DaemonSetControllerName)}, NewDaemonSetAdapter)
|
||||
}
|
||||
|
||||
type DaemonSetAdapter struct {
|
||||
client federationclientset.Interface
|
||||
}
|
||||
|
||||
func NewDaemonSetAdapter(client federationclientset.Interface) FederatedTypeAdapter {
|
||||
return &DaemonSetAdapter{client: client}
|
||||
}
|
||||
|
||||
func (a *DaemonSetAdapter) Kind() string {
|
||||
return DaemonSetKind
|
||||
}
|
||||
|
||||
func (a *DaemonSetAdapter) ObjectType() pkgruntime.Object {
|
||||
return &extensionsv1.DaemonSet{}
|
||||
}
|
||||
|
||||
func (a *DaemonSetAdapter) IsExpectedType(obj interface{}) bool {
|
||||
_, ok := obj.(*extensionsv1.DaemonSet)
|
||||
return ok
|
||||
}
|
||||
|
||||
func (a *DaemonSetAdapter) Copy(obj pkgruntime.Object) pkgruntime.Object {
|
||||
daemonset := obj.(*extensionsv1.DaemonSet)
|
||||
return &extensionsv1.DaemonSet{
|
||||
ObjectMeta: util.DeepCopyRelevantObjectMeta(daemonset.ObjectMeta),
|
||||
Spec: *(util.DeepCopyApiTypeOrPanic(&daemonset.Spec).(*extensionsv1.DaemonSetSpec)),
|
||||
}
|
||||
}
|
||||
|
||||
func (a *DaemonSetAdapter) Equivalent(obj1, obj2 pkgruntime.Object) bool {
|
||||
daemonset1 := obj1.(*extensionsv1.DaemonSet)
|
||||
daemonset2 := obj2.(*extensionsv1.DaemonSet)
|
||||
return util.ObjectMetaEquivalent(daemonset1.ObjectMeta, daemonset2.ObjectMeta) && reflect.DeepEqual(daemonset1.Spec, daemonset2.Spec)
|
||||
}
|
||||
|
||||
func (a *DaemonSetAdapter) NamespacedName(obj pkgruntime.Object) types.NamespacedName {
|
||||
daemonset := obj.(*extensionsv1.DaemonSet)
|
||||
return types.NamespacedName{Namespace: daemonset.Namespace, Name: daemonset.Name}
|
||||
}
|
||||
|
||||
func (a *DaemonSetAdapter) ObjectMeta(obj pkgruntime.Object) *metav1.ObjectMeta {
|
||||
return &obj.(*extensionsv1.DaemonSet).ObjectMeta
|
||||
}
|
||||
|
||||
func (a *DaemonSetAdapter) FedCreate(obj pkgruntime.Object) (pkgruntime.Object, error) {
|
||||
daemonset := obj.(*extensionsv1.DaemonSet)
|
||||
return a.client.Extensions().DaemonSets(daemonset.Namespace).Create(daemonset)
|
||||
}
|
||||
|
||||
func (a *DaemonSetAdapter) FedDelete(namespacedName types.NamespacedName, options *metav1.DeleteOptions) error {
|
||||
return a.client.Extensions().DaemonSets(namespacedName.Namespace).Delete(namespacedName.Name, options)
|
||||
}
|
||||
|
||||
func (a *DaemonSetAdapter) FedGet(namespacedName types.NamespacedName) (pkgruntime.Object, error) {
|
||||
return a.client.Extensions().DaemonSets(namespacedName.Namespace).Get(namespacedName.Name, metav1.GetOptions{})
|
||||
}
|
||||
|
||||
func (a *DaemonSetAdapter) FedList(namespace string, options metav1.ListOptions) (pkgruntime.Object, error) {
|
||||
return a.client.Extensions().DaemonSets(namespace).List(options)
|
||||
}
|
||||
|
||||
func (a *DaemonSetAdapter) FedUpdate(obj pkgruntime.Object) (pkgruntime.Object, error) {
|
||||
daemonset := obj.(*extensionsv1.DaemonSet)
|
||||
return a.client.Extensions().DaemonSets(daemonset.Namespace).Update(daemonset)
|
||||
}
|
||||
|
||||
func (a *DaemonSetAdapter) FedWatch(namespace string, options metav1.ListOptions) (watch.Interface, error) {
|
||||
return a.client.Extensions().DaemonSets(namespace).Watch(options)
|
||||
}
|
||||
|
||||
func (a *DaemonSetAdapter) ClusterCreate(client kubeclientset.Interface, obj pkgruntime.Object) (pkgruntime.Object, error) {
|
||||
daemonset := obj.(*extensionsv1.DaemonSet)
|
||||
return client.Extensions().DaemonSets(daemonset.Namespace).Create(daemonset)
|
||||
}
|
||||
|
||||
func (a *DaemonSetAdapter) ClusterDelete(client kubeclientset.Interface, nsName types.NamespacedName, options *metav1.DeleteOptions) error {
|
||||
return client.Extensions().DaemonSets(nsName.Namespace).Delete(nsName.Name, options)
|
||||
}
|
||||
|
||||
func (a *DaemonSetAdapter) ClusterGet(client kubeclientset.Interface, namespacedName types.NamespacedName) (pkgruntime.Object, error) {
|
||||
return client.Extensions().DaemonSets(namespacedName.Namespace).Get(namespacedName.Name, metav1.GetOptions{})
|
||||
}
|
||||
|
||||
func (a *DaemonSetAdapter) ClusterList(client kubeclientset.Interface, namespace string, options metav1.ListOptions) (pkgruntime.Object, error) {
|
||||
return client.Extensions().DaemonSets(namespace).List(options)
|
||||
}
|
||||
|
||||
func (a *DaemonSetAdapter) ClusterUpdate(client kubeclientset.Interface, obj pkgruntime.Object) (pkgruntime.Object, error) {
|
||||
daemonset := obj.(*extensionsv1.DaemonSet)
|
||||
return client.Extensions().DaemonSets(daemonset.Namespace).Update(daemonset)
|
||||
}
|
||||
|
||||
func (a *DaemonSetAdapter) ClusterWatch(client kubeclientset.Interface, namespace string, options metav1.ListOptions) (watch.Interface, error) {
|
||||
return client.Extensions().DaemonSets(namespace).Watch(options)
|
||||
}
|
||||
|
||||
func (a *DaemonSetAdapter) NewTestObject(namespace string) pkgruntime.Object {
|
||||
return &extensionsv1.DaemonSet{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
GenerateName: "test-daemonset-",
|
||||
Namespace: namespace,
|
||||
Labels: map[string]string{"app": "test-daemonset"},
|
||||
},
|
||||
Spec: extensionsv1.DaemonSetSpec{
|
||||
Template: v1.PodTemplateSpec{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Labels: map[string]string{"name": "test-pod"},
|
||||
},
|
||||
Spec: v1.PodSpec{
|
||||
Containers: []v1.Container{
|
||||
{
|
||||
Name: "test-daemonset",
|
||||
Image: "images/test-daemonset",
|
||||
Ports: []v1.ContainerPort{{ContainerPort: 9376}},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
|
@ -25,7 +25,6 @@ filegroup(
|
|||
srcs = [
|
||||
":package-srcs",
|
||||
"//federation/pkg/federation-controller/cluster:all-srcs",
|
||||
"//federation/pkg/federation-controller/daemonset:all-srcs",
|
||||
"//federation/pkg/federation-controller/deployment:all-srcs",
|
||||
"//federation/pkg/federation-controller/ingress:all-srcs",
|
||||
"//federation/pkg/federation-controller/namespace:all-srcs",
|
||||
|
|
|
|||
|
|
@ -1,72 +0,0 @@
|
|||
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 = ["daemonset_controller.go"],
|
||||
tags = ["automanaged"],
|
||||
deps = [
|
||||
"//federation/apis/federation/v1beta1:go_default_library",
|
||||
"//federation/client/clientset_generated/federation_clientset:go_default_library",
|
||||
"//federation/pkg/federation-controller/util:go_default_library",
|
||||
"//federation/pkg/federation-controller/util/deletionhelper:go_default_library",
|
||||
"//federation/pkg/federation-controller/util/eventsink:go_default_library",
|
||||
"//pkg/api:go_default_library",
|
||||
"//pkg/apis/extensions/v1beta1:go_default_library",
|
||||
"//pkg/client/clientset_generated/clientset:go_default_library",
|
||||
"//pkg/controller:go_default_library",
|
||||
"//vendor/github.com/golang/glog: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/types:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/watch:go_default_library",
|
||||
"//vendor/k8s.io/client-go/pkg/api/v1:go_default_library",
|
||||
"//vendor/k8s.io/client-go/tools/cache:go_default_library",
|
||||
"//vendor/k8s.io/client-go/tools/record:go_default_library",
|
||||
"//vendor/k8s.io/client-go/util/flowcontrol:go_default_library",
|
||||
],
|
||||
)
|
||||
|
||||
go_test(
|
||||
name = "go_default_test",
|
||||
srcs = ["daemonset_controller_test.go"],
|
||||
library = ":go_default_library",
|
||||
tags = ["automanaged"],
|
||||
deps = [
|
||||
"//federation/apis/federation/v1beta1:go_default_library",
|
||||
"//federation/client/clientset_generated/federation_clientset/fake:go_default_library",
|
||||
"//federation/pkg/federation-controller/util:go_default_library",
|
||||
"//federation/pkg/federation-controller/util/deletionhelper:go_default_library",
|
||||
"//federation/pkg/federation-controller/util/test:go_default_library",
|
||||
"//pkg/api/v1:go_default_library",
|
||||
"//pkg/apis/extensions/v1beta1:go_default_library",
|
||||
"//pkg/client/clientset_generated/clientset:go_default_library",
|
||||
"//pkg/client/clientset_generated/clientset/fake:go_default_library",
|
||||
"//vendor/github.com/stretchr/testify/assert: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"],
|
||||
)
|
||||
|
|
@ -1,436 +0,0 @@
|
|||
/*
|
||||
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.
|
||||
*/
|
||||
|
||||
package daemonset
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"reflect"
|
||||
"time"
|
||||
|
||||
"k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
pkgruntime "k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/types"
|
||||
"k8s.io/apimachinery/pkg/watch"
|
||||
clientv1 "k8s.io/client-go/pkg/api/v1"
|
||||
"k8s.io/client-go/tools/cache"
|
||||
"k8s.io/client-go/tools/record"
|
||||
"k8s.io/client-go/util/flowcontrol"
|
||||
federationapi "k8s.io/kubernetes/federation/apis/federation/v1beta1"
|
||||
federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset"
|
||||
"k8s.io/kubernetes/federation/pkg/federation-controller/util"
|
||||
"k8s.io/kubernetes/federation/pkg/federation-controller/util/deletionhelper"
|
||||
"k8s.io/kubernetes/federation/pkg/federation-controller/util/eventsink"
|
||||
"k8s.io/kubernetes/pkg/api"
|
||||
extensionsv1 "k8s.io/kubernetes/pkg/apis/extensions/v1beta1"
|
||||
kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
|
||||
"k8s.io/kubernetes/pkg/controller"
|
||||
|
||||
"github.com/golang/glog"
|
||||
)
|
||||
|
||||
const (
|
||||
allClustersKey = "ALL_CLUSTERS"
|
||||
ControllerName = "daemonsets"
|
||||
)
|
||||
|
||||
var (
|
||||
RequiredResources = []schema.GroupVersionResource{extensionsv1.SchemeGroupVersion.WithResource("daemonsets")}
|
||||
)
|
||||
|
||||
type DaemonSetController struct {
|
||||
// For triggering single daemonset reconciliation. This is used when there is an
|
||||
// add/update/delete operation on a daemonset in either federated API server or
|
||||
// in some member of the federation.
|
||||
daemonsetDeliverer *util.DelayingDeliverer
|
||||
|
||||
// For triggering all daemonsets reconciliation. This is used when
|
||||
// a new cluster becomes available.
|
||||
clusterDeliverer *util.DelayingDeliverer
|
||||
|
||||
// Contains daemonsets present in members of federation.
|
||||
daemonsetFederatedInformer util.FederatedInformer
|
||||
// For updating members of federation.
|
||||
federatedUpdater util.FederatedUpdater
|
||||
// Definitions of daemonsets that should be federated.
|
||||
daemonsetInformerStore cache.Store
|
||||
// Informer controller for daemonsets that should be federated.
|
||||
daemonsetInformerController cache.Controller
|
||||
|
||||
// Client to federated api server.
|
||||
federatedApiClient federationclientset.Interface
|
||||
|
||||
// Backoff manager for daemonsets
|
||||
daemonsetBackoff *flowcontrol.Backoff
|
||||
|
||||
// For events
|
||||
eventRecorder record.EventRecorder
|
||||
|
||||
deletionHelper *deletionhelper.DeletionHelper
|
||||
|
||||
daemonsetReviewDelay time.Duration
|
||||
clusterAvailableDelay time.Duration
|
||||
smallDelay time.Duration
|
||||
updateTimeout time.Duration
|
||||
}
|
||||
|
||||
// NewDaemonSetController returns a new daemonset controller
|
||||
func NewDaemonSetController(client federationclientset.Interface) *DaemonSetController {
|
||||
broadcaster := record.NewBroadcaster()
|
||||
broadcaster.StartRecordingToSink(eventsink.NewFederatedEventSink(client))
|
||||
recorder := broadcaster.NewRecorder(api.Scheme, clientv1.EventSource{Component: "federated-daemonset-controller"})
|
||||
|
||||
daemonsetcontroller := &DaemonSetController{
|
||||
federatedApiClient: client,
|
||||
daemonsetReviewDelay: time.Second * 10,
|
||||
clusterAvailableDelay: time.Second * 20,
|
||||
smallDelay: time.Second * 3,
|
||||
updateTimeout: time.Second * 30,
|
||||
daemonsetBackoff: flowcontrol.NewBackOff(5*time.Second, time.Minute),
|
||||
eventRecorder: recorder,
|
||||
}
|
||||
|
||||
// Build deliverers for triggering reconciliations.
|
||||
daemonsetcontroller.daemonsetDeliverer = util.NewDelayingDeliverer()
|
||||
daemonsetcontroller.clusterDeliverer = util.NewDelayingDeliverer()
|
||||
|
||||
// Start informer in federated API servers on daemonsets that should be federated.
|
||||
daemonsetcontroller.daemonsetInformerStore, daemonsetcontroller.daemonsetInformerController = cache.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func(options metav1.ListOptions) (pkgruntime.Object, error) {
|
||||
return client.Extensions().DaemonSets(metav1.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
|
||||
return client.Extensions().DaemonSets(metav1.NamespaceAll).Watch(options)
|
||||
},
|
||||
},
|
||||
&extensionsv1.DaemonSet{},
|
||||
controller.NoResyncPeriodFunc(),
|
||||
util.NewTriggerOnAllChanges(func(obj pkgruntime.Object) { daemonsetcontroller.deliverDaemonSetObj(obj, 0, false) }))
|
||||
|
||||
// Federated informer on daemonsets in members of federation.
|
||||
daemonsetcontroller.daemonsetFederatedInformer = util.NewFederatedInformer(
|
||||
client,
|
||||
func(cluster *federationapi.Cluster, targetClient kubeclientset.Interface) (cache.Store, cache.Controller) {
|
||||
return cache.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func(options metav1.ListOptions) (pkgruntime.Object, error) {
|
||||
return targetClient.Extensions().DaemonSets(metav1.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
|
||||
return targetClient.Extensions().DaemonSets(metav1.NamespaceAll).Watch(options)
|
||||
},
|
||||
},
|
||||
&extensionsv1.DaemonSet{},
|
||||
controller.NoResyncPeriodFunc(),
|
||||
// Trigger reconciliation whenever something in federated cluster is changed. In most cases it
|
||||
// would be just confirmation that some daemonset operation succeeded.
|
||||
util.NewTriggerOnAllChanges(
|
||||
func(obj pkgruntime.Object) {
|
||||
daemonsetcontroller.deliverDaemonSetObj(obj, daemonsetcontroller.daemonsetReviewDelay, false)
|
||||
},
|
||||
))
|
||||
},
|
||||
|
||||
&util.ClusterLifecycleHandlerFuncs{
|
||||
ClusterAvailable: func(cluster *federationapi.Cluster) {
|
||||
// When new cluster becomes available process all the daemonsets again.
|
||||
daemonsetcontroller.clusterDeliverer.DeliverAt(allClustersKey, nil, time.Now().Add(daemonsetcontroller.clusterAvailableDelay))
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
// Federated updater along with Create/Update/Delete operations.
|
||||
daemonsetcontroller.federatedUpdater = util.NewFederatedUpdater(daemonsetcontroller.daemonsetFederatedInformer,
|
||||
func(client kubeclientset.Interface, obj pkgruntime.Object) error {
|
||||
daemonset := obj.(*extensionsv1.DaemonSet)
|
||||
glog.V(4).Infof("Attempting to create daemonset: %s/%s", daemonset.Namespace, daemonset.Name)
|
||||
_, err := client.Extensions().DaemonSets(daemonset.Namespace).Create(daemonset)
|
||||
if err != nil {
|
||||
glog.Errorf("Error creating daemonset %s/%s/: %v", daemonset.Namespace, daemonset.Name, err)
|
||||
} else {
|
||||
glog.V(4).Infof("Successfully created daemonset %s/%s", daemonset.Namespace, daemonset.Name)
|
||||
}
|
||||
return err
|
||||
},
|
||||
func(client kubeclientset.Interface, obj pkgruntime.Object) error {
|
||||
daemonset := obj.(*extensionsv1.DaemonSet)
|
||||
glog.V(4).Infof("Attempting to update daemonset: %s/%s", daemonset.Namespace, daemonset.Name)
|
||||
_, err := client.Extensions().DaemonSets(daemonset.Namespace).Update(daemonset)
|
||||
if err != nil {
|
||||
glog.Errorf("Error updating daemonset %s/%s/: %v", daemonset.Namespace, daemonset.Name, err)
|
||||
} else {
|
||||
glog.V(4).Infof("Successfully updating daemonset %s/%s", daemonset.Namespace, daemonset.Name)
|
||||
}
|
||||
return err
|
||||
},
|
||||
func(client kubeclientset.Interface, obj pkgruntime.Object) error {
|
||||
daemonset := obj.(*extensionsv1.DaemonSet)
|
||||
glog.V(4).Infof("Attempting to delete daemonset: %s/%s", daemonset.Namespace, daemonset.Name)
|
||||
orphanDependents := false
|
||||
err := client.Extensions().DaemonSets(daemonset.Namespace).Delete(daemonset.Name, &metav1.DeleteOptions{OrphanDependents: &orphanDependents})
|
||||
if err != nil {
|
||||
glog.Errorf("Error deleting daemonset %s/%s/: %v", daemonset.Namespace, daemonset.Name, err)
|
||||
} else {
|
||||
glog.V(4).Infof("Successfully deleting daemonset %s/%s", daemonset.Namespace, daemonset.Name)
|
||||
}
|
||||
return err
|
||||
})
|
||||
|
||||
daemonsetcontroller.deletionHelper = deletionhelper.NewDeletionHelper(
|
||||
daemonsetcontroller.updateDaemonSet,
|
||||
// objNameFunc
|
||||
func(obj pkgruntime.Object) string {
|
||||
daemonset := obj.(*extensionsv1.DaemonSet)
|
||||
return daemonset.Name
|
||||
},
|
||||
daemonsetcontroller.updateTimeout,
|
||||
daemonsetcontroller.eventRecorder,
|
||||
daemonsetcontroller.daemonsetFederatedInformer,
|
||||
daemonsetcontroller.federatedUpdater,
|
||||
)
|
||||
|
||||
return daemonsetcontroller
|
||||
}
|
||||
|
||||
// Sends the given updated object to apiserver.
|
||||
// Assumes that the given object is a daemonset.
|
||||
func (daemonsetcontroller *DaemonSetController) updateDaemonSet(obj pkgruntime.Object) (pkgruntime.Object, error) {
|
||||
daemonset := obj.(*extensionsv1.DaemonSet)
|
||||
return daemonsetcontroller.federatedApiClient.Extensions().DaemonSets(daemonset.Namespace).Update(daemonset)
|
||||
}
|
||||
|
||||
func (daemonsetcontroller *DaemonSetController) Run(stopChan <-chan struct{}) {
|
||||
glog.V(1).Infof("Starting daemonset controller")
|
||||
go daemonsetcontroller.daemonsetInformerController.Run(stopChan)
|
||||
|
||||
glog.V(1).Infof("Starting daemonset federated informer")
|
||||
daemonsetcontroller.daemonsetFederatedInformer.Start()
|
||||
go func() {
|
||||
<-stopChan
|
||||
daemonsetcontroller.daemonsetFederatedInformer.Stop()
|
||||
}()
|
||||
glog.V(1).Infof("Starting daemonset deliverers")
|
||||
daemonsetcontroller.daemonsetDeliverer.StartWithHandler(func(item *util.DelayingDelivererItem) {
|
||||
daemonset := item.Value.(*types.NamespacedName)
|
||||
glog.V(4).Infof("Trigerring reconciliation of daemonset %s", daemonset.String())
|
||||
daemonsetcontroller.reconcileDaemonSet(daemonset.Namespace, daemonset.Name)
|
||||
})
|
||||
daemonsetcontroller.clusterDeliverer.StartWithHandler(func(_ *util.DelayingDelivererItem) {
|
||||
glog.V(4).Infof("Triggering reconciliation of all daemonsets")
|
||||
daemonsetcontroller.reconcileDaemonSetsOnClusterChange()
|
||||
})
|
||||
util.StartBackoffGC(daemonsetcontroller.daemonsetBackoff, stopChan)
|
||||
}
|
||||
|
||||
func getDaemonSetKey(namespace, name string) string {
|
||||
return types.NamespacedName{
|
||||
Namespace: namespace,
|
||||
Name: name,
|
||||
}.String()
|
||||
}
|
||||
|
||||
func (daemonsetcontroller *DaemonSetController) deliverDaemonSetObj(obj interface{}, delay time.Duration, failed bool) {
|
||||
daemonset := obj.(*extensionsv1.DaemonSet)
|
||||
daemonsetcontroller.deliverDaemonSet(daemonset.Namespace, daemonset.Name, delay, failed)
|
||||
}
|
||||
|
||||
// Adds backoff to delay if this delivery is related to some failure. Resets backoff if there was no failure.
|
||||
func (daemonsetcontroller *DaemonSetController) deliverDaemonSet(namespace string, name string, delay time.Duration, failed bool) {
|
||||
key := getDaemonSetKey(namespace, name)
|
||||
if failed {
|
||||
daemonsetcontroller.daemonsetBackoff.Next(key, time.Now())
|
||||
delay = delay + daemonsetcontroller.daemonsetBackoff.Get(key)
|
||||
} else {
|
||||
daemonsetcontroller.daemonsetBackoff.Reset(key)
|
||||
}
|
||||
daemonsetcontroller.daemonsetDeliverer.DeliverAfter(key,
|
||||
&types.NamespacedName{Namespace: namespace, Name: name}, delay)
|
||||
}
|
||||
|
||||
// Check whether all data stores are in sync. False is returned if any of the informer/stores is not yet
|
||||
// synced with the corresponding api server.
|
||||
func (daemonsetcontroller *DaemonSetController) isSynced() bool {
|
||||
if !daemonsetcontroller.daemonsetFederatedInformer.ClustersSynced() {
|
||||
glog.V(2).Infof("Cluster list not synced")
|
||||
return false
|
||||
}
|
||||
clusters, err := daemonsetcontroller.daemonsetFederatedInformer.GetReadyClusters()
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to get ready clusters: %v", err)
|
||||
return false
|
||||
}
|
||||
if !daemonsetcontroller.daemonsetFederatedInformer.GetTargetStore().ClustersSynced(clusters) {
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// The function triggers reconciliation of all federated daemonsets.
|
||||
func (daemonsetcontroller *DaemonSetController) reconcileDaemonSetsOnClusterChange() {
|
||||
if !daemonsetcontroller.isSynced() {
|
||||
daemonsetcontroller.clusterDeliverer.DeliverAt(allClustersKey, nil, time.Now().Add(daemonsetcontroller.clusterAvailableDelay))
|
||||
}
|
||||
for _, obj := range daemonsetcontroller.daemonsetInformerStore.List() {
|
||||
daemonset := obj.(*extensionsv1.DaemonSet)
|
||||
daemonsetcontroller.deliverDaemonSet(daemonset.Namespace, daemonset.Name, daemonsetcontroller.smallDelay, false)
|
||||
}
|
||||
}
|
||||
|
||||
func (daemonsetcontroller *DaemonSetController) reconcileDaemonSet(namespace string, daemonsetName string) {
|
||||
glog.V(4).Infof("Reconciling daemonset %s/%s", namespace, daemonsetName)
|
||||
|
||||
if !daemonsetcontroller.isSynced() {
|
||||
glog.V(4).Infof("Daemonset controller is not synced")
|
||||
daemonsetcontroller.deliverDaemonSet(namespace, daemonsetName, daemonsetcontroller.clusterAvailableDelay, false)
|
||||
return
|
||||
}
|
||||
|
||||
key := getDaemonSetKey(namespace, daemonsetName)
|
||||
baseDaemonSetObjFromStore, exist, err := daemonsetcontroller.daemonsetInformerStore.GetByKey(key)
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to query main daemonset store for %v: %v", key, err)
|
||||
daemonsetcontroller.deliverDaemonSet(namespace, daemonsetName, 0, true)
|
||||
return
|
||||
}
|
||||
|
||||
if !exist {
|
||||
glog.V(4).Infof("Skipping daemonset %s/%s - not federated", namespace, daemonsetName)
|
||||
// Not federated daemonset, ignoring.
|
||||
return
|
||||
}
|
||||
baseDaemonSetObj, err := api.Scheme.DeepCopy(baseDaemonSetObjFromStore)
|
||||
baseDaemonSet, ok := baseDaemonSetObj.(*extensionsv1.DaemonSet)
|
||||
if err != nil || !ok {
|
||||
glog.Errorf("Error in retrieving obj %s from store: %v, %v", daemonsetName, ok, err)
|
||||
daemonsetcontroller.deliverDaemonSet(namespace, daemonsetName, 0, true)
|
||||
return
|
||||
}
|
||||
if baseDaemonSet.DeletionTimestamp != nil {
|
||||
if err := daemonsetcontroller.delete(baseDaemonSet); err != nil {
|
||||
glog.Errorf("Failed to delete %s: %v", daemonsetName, err)
|
||||
daemonsetcontroller.eventRecorder.Eventf(baseDaemonSet, api.EventTypeWarning, "DeleteFailed",
|
||||
"DaemonSet delete failed: %v", err)
|
||||
daemonsetcontroller.deliverDaemonSet(namespace, daemonsetName, 0, true)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
glog.V(3).Infof("Ensuring delete object from underlying clusters finalizer for daemonset: %s",
|
||||
baseDaemonSet.Name)
|
||||
// Add the required finalizers before creating a daemonset in underlying clusters.
|
||||
updatedDaemonSetObj, err := daemonsetcontroller.deletionHelper.EnsureFinalizers(baseDaemonSet)
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to ensure delete object from underlying clusters finalizer in daemonset %s: %v",
|
||||
baseDaemonSet.Name, err)
|
||||
daemonsetcontroller.deliverDaemonSet(namespace, daemonsetName, 0, false)
|
||||
return
|
||||
}
|
||||
baseDaemonSet = updatedDaemonSetObj.(*extensionsv1.DaemonSet)
|
||||
|
||||
glog.V(3).Infof("Syncing daemonset %s in underlying clusters", baseDaemonSet.Name)
|
||||
|
||||
clusters, err := daemonsetcontroller.daemonsetFederatedInformer.GetReadyClusters()
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to get cluster list: %v", err)
|
||||
daemonsetcontroller.deliverDaemonSet(namespace, daemonsetName, daemonsetcontroller.clusterAvailableDelay, false)
|
||||
return
|
||||
}
|
||||
|
||||
operations := make([]util.FederatedOperation, 0)
|
||||
for _, cluster := range clusters {
|
||||
clusterDaemonSetObj, found, err := daemonsetcontroller.daemonsetFederatedInformer.GetTargetStore().GetByKey(cluster.Name, key)
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to get %s from %s: %v", key, cluster.Name, err)
|
||||
daemonsetcontroller.deliverDaemonSet(namespace, daemonsetName, 0, true)
|
||||
return
|
||||
}
|
||||
|
||||
// Do not modify. Otherwise make a deep copy.
|
||||
desiredDaemonSet := &extensionsv1.DaemonSet{
|
||||
ObjectMeta: util.DeepCopyRelevantObjectMeta(baseDaemonSet.ObjectMeta),
|
||||
Spec: *(util.DeepCopyApiTypeOrPanic(&baseDaemonSet.Spec).(*extensionsv1.DaemonSetSpec)),
|
||||
}
|
||||
|
||||
if !found {
|
||||
glog.V(4).Infof("Creating daemonset %s/%s in cluster %s", namespace, daemonsetName, cluster.Name)
|
||||
|
||||
daemonsetcontroller.eventRecorder.Eventf(baseDaemonSet, api.EventTypeNormal, "CreateInCluster",
|
||||
"Creating daemonset in cluster %s", cluster.Name)
|
||||
|
||||
operations = append(operations, util.FederatedOperation{
|
||||
Type: util.OperationTypeAdd,
|
||||
Obj: desiredDaemonSet,
|
||||
ClusterName: cluster.Name,
|
||||
})
|
||||
} else {
|
||||
clusterDaemonSet := clusterDaemonSetObj.(*extensionsv1.DaemonSet)
|
||||
|
||||
// Update existing daemonset, if needed.
|
||||
if !util.ObjectMetaEquivalent(desiredDaemonSet.ObjectMeta, clusterDaemonSet.ObjectMeta) ||
|
||||
!reflect.DeepEqual(desiredDaemonSet.Spec, clusterDaemonSet.Spec) {
|
||||
|
||||
glog.V(4).Infof("Upadting daemonset %s/%s in cluster %s", namespace, daemonsetName, cluster.Name)
|
||||
daemonsetcontroller.eventRecorder.Eventf(baseDaemonSet, api.EventTypeNormal, "UpdateInCluster",
|
||||
"Updating daemonset in cluster %s", cluster.Name)
|
||||
operations = append(operations, util.FederatedOperation{
|
||||
Type: util.OperationTypeUpdate,
|
||||
Obj: desiredDaemonSet,
|
||||
ClusterName: cluster.Name,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if len(operations) == 0 {
|
||||
glog.V(4).Infof("No operation needed for %s/%s", namespace, daemonsetName)
|
||||
// Everything is in order
|
||||
return
|
||||
}
|
||||
err = daemonsetcontroller.federatedUpdater.UpdateWithOnError(operations, daemonsetcontroller.updateTimeout,
|
||||
func(op util.FederatedOperation, operror error) {
|
||||
daemonsetcontroller.eventRecorder.Eventf(baseDaemonSet, api.EventTypeWarning, "UpdateInClusterFailed",
|
||||
"DaemonSet update in cluster %s failed: %v", op.ClusterName, operror)
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to execute updates for %s: %v, retrying shortly", key, err)
|
||||
daemonsetcontroller.deliverDaemonSet(namespace, daemonsetName, 0, true)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// delete deletes the given daemonset or returns error if the deletion was not complete.
|
||||
func (daemonsetcontroller *DaemonSetController) delete(daemonset *extensionsv1.DaemonSet) error {
|
||||
glog.V(3).Infof("Handling deletion of daemonset: %v", *daemonset)
|
||||
_, err := daemonsetcontroller.deletionHelper.HandleObjectInUnderlyingClusters(daemonset)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = daemonsetcontroller.federatedApiClient.Extensions().DaemonSets(daemonset.Namespace).Delete(daemonset.Name, nil)
|
||||
if err != nil {
|
||||
// Its all good if the error is not found error. That means it is deleted already and we do not have to do anything.
|
||||
// This is expected when we are processing an update as a result of daemonset finalizer deletion.
|
||||
// The process that deleted the last finalizer is also going to delete the daemonset and we do not have to do anything.
|
||||
if !errors.IsNotFound(err) {
|
||||
return fmt.Errorf("failed to delete daemonset: %v", err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
|
@ -53,6 +53,7 @@ go_test(
|
|||
name = "go_default_test",
|
||||
srcs = [
|
||||
"configmap_controller_test.go",
|
||||
"daemonset_controller_test.go",
|
||||
"secret_controller_test.go",
|
||||
],
|
||||
library = ":go_default_library",
|
||||
|
|
@ -65,6 +66,7 @@ go_test(
|
|||
"//federation/pkg/federation-controller/util/deletionhelper:go_default_library",
|
||||
"//federation/pkg/federation-controller/util/test:go_default_library",
|
||||
"//pkg/api/v1:go_default_library",
|
||||
"//pkg/apis/extensions/v1beta1:go_default_library",
|
||||
"//pkg/client/clientset_generated/clientset:go_default_library",
|
||||
"//pkg/client/clientset_generated/clientset/fake:go_default_library",
|
||||
"//vendor/github.com/golang/glog:go_default_library",
|
||||
|
|
|
|||
|
|
@ -14,19 +14,20 @@ See the License for the specific language governing permissions and
|
|||
limitations under the License.
|
||||
*/
|
||||
|
||||
package daemonset
|
||||
package sync
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/types"
|
||||
"k8s.io/apimachinery/pkg/util/wait"
|
||||
federationapi "k8s.io/kubernetes/federation/apis/federation/v1beta1"
|
||||
fakefedclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset/fake"
|
||||
"k8s.io/kubernetes/federation/pkg/federatedtypes"
|
||||
"k8s.io/kubernetes/federation/pkg/federation-controller/util"
|
||||
"k8s.io/kubernetes/federation/pkg/federation-controller/util/deletionhelper"
|
||||
. "k8s.io/kubernetes/federation/pkg/federation-controller/util/test"
|
||||
|
|
@ -38,12 +39,10 @@ import (
|
|||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
const (
|
||||
daemonsets string = "daemonsets"
|
||||
clusters string = "clusters"
|
||||
)
|
||||
|
||||
func TestDaemonSetController(t *testing.T) {
|
||||
daemonsets := "daemonsets"
|
||||
clusters := "clusters"
|
||||
|
||||
cluster1 := NewCluster("cluster1", apiv1.ConditionTrue)
|
||||
cluster2 := NewCluster("cluster2", apiv1.ConditionTrue)
|
||||
|
||||
|
|
@ -65,8 +64,8 @@ func TestDaemonSetController(t *testing.T) {
|
|||
RegisterFakeList(daemonsets, &cluster2Client.Fake, &extensionsv1.DaemonSetList{Items: []extensionsv1.DaemonSet{}})
|
||||
cluster2CreateChan := RegisterFakeCopyOnCreate(daemonsets, &cluster2Client.Fake, cluster2Watch)
|
||||
|
||||
daemonsetController := NewDaemonSetController(fakeClient)
|
||||
informer := ToFederatedInformerForTestOnly(daemonsetController.daemonsetFederatedInformer)
|
||||
daemonsetController := newFederationSyncController(fakeClient, federatedtypes.NewDaemonSetAdapter(fakeClient))
|
||||
informer := ToFederatedInformerForTestOnly(daemonsetController.informer)
|
||||
informer.SetClientFactory(func(cluster *federationapi.Cluster) (kubeclientset.Interface, error) {
|
||||
switch cluster.Name {
|
||||
case cluster1.Name:
|
||||
|
|
@ -78,10 +77,7 @@ func TestDaemonSetController(t *testing.T) {
|
|||
}
|
||||
})
|
||||
|
||||
daemonsetController.clusterAvailableDelay = time.Second
|
||||
daemonsetController.daemonsetReviewDelay = 50 * time.Millisecond
|
||||
daemonsetController.smallDelay = 20 * time.Millisecond
|
||||
daemonsetController.updateTimeout = 5 * time.Second
|
||||
daemonsetController.minimizeLatency()
|
||||
|
||||
stop := make(chan struct{})
|
||||
daemonsetController.Run(stop)
|
||||
|
|
@ -94,7 +90,9 @@ func TestDaemonSetController(t *testing.T) {
|
|||
},
|
||||
Spec: extensionsv1.DaemonSetSpec{
|
||||
Selector: &metav1.LabelSelector{
|
||||
MatchLabels: make(map[string]string),
|
||||
MatchLabels: map[string]string{
|
||||
"A": "xyz",
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
|
@ -115,10 +113,14 @@ func TestDaemonSetController(t *testing.T) {
|
|||
assert.True(t, daemonsetsEqual(daemonset1, *createdDaemonSet),
|
||||
fmt.Sprintf("expected: %v, actual: %v", daemonset1, *createdDaemonSet))
|
||||
|
||||
daemonsetKey := types.NamespacedName{
|
||||
Namespace: daemonset1.Namespace,
|
||||
Name: daemonset1.Name,
|
||||
}.String()
|
||||
// Wait for the daemonset to appear in the informer store
|
||||
err := WaitForStoreUpdate(
|
||||
daemonsetController.daemonsetFederatedInformer.GetTargetStore(),
|
||||
cluster1.Name, getDaemonSetKey(daemonset1.Namespace, daemonset1.Name), wait.ForeverTestTimeout)
|
||||
daemonsetController.informer.GetTargetStore(),
|
||||
cluster1.Name, daemonsetKey, wait.ForeverTestTimeout)
|
||||
assert.Nil(t, err, "daemonset should have appeared in the informer store")
|
||||
|
||||
// TODO: Re-enable this when we have fixed these flaky tests: https://github.com/kubernetes/kubernetes/issues/36540.
|
||||
|
|
@ -600,7 +600,6 @@ k8s.io/kubernetes/federation/pkg/dnsprovider/providers/aws/route53,cjcullen,1,
|
|||
k8s.io/kubernetes/federation/pkg/dnsprovider/providers/coredns,brendandburns,0,
|
||||
k8s.io/kubernetes/federation/pkg/dnsprovider/providers/google/clouddns,madhusudancs,1,
|
||||
k8s.io/kubernetes/federation/pkg/federation-controller/cluster,nikhiljindal,0,
|
||||
k8s.io/kubernetes/federation/pkg/federation-controller/daemonset,childsb,1,
|
||||
k8s.io/kubernetes/federation/pkg/federation-controller/deployment,zmerlynn,1,
|
||||
k8s.io/kubernetes/federation/pkg/federation-controller/ingress,vishh,1,
|
||||
k8s.io/kubernetes/federation/pkg/federation-controller/namespace,rrati,0,
|
||||
|
|
|
|||
|
Loading…
Reference in New Issue