mirror of https://github.com/k3s-io/k3s
Merge pull request #55706 from wackxu/bss
Automatic merge from submit-queue. If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>. use shared informers for BootstrapSigner controller **What this PR does / why we need it**: fix TODO: Switch to shared informers **Which issue(s) this PR fixes** *(optional, in `fixes #<issue number>(, fixes #<issue_number>, ...)` format, will close the issue(s) when PR gets merged)*: Fixes # **Special notes for your reviewer**: **Release note**: ```release-note NONE ```pull/6/head
Kubernetes Submit Queue
GitHub
commit
5043c7ecc5
|
|
@ -25,6 +25,8 @@ import (
|
|||
func startBootstrapSignerController(ctx ControllerContext) (bool, error) {
|
||||
bsc, err := bootstrap.NewBootstrapSigner(
|
||||
ctx.ClientBuilder.ClientGoClientOrDie("bootstrap-signer"),
|
||||
ctx.InformerFactory.Core().V1().Secrets(),
|
||||
ctx.InformerFactory.Core().V1().ConfigMaps(),
|
||||
bootstrap.DefaultBootstrapSignerOptions(),
|
||||
)
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -21,11 +21,14 @@ go_test(
|
|||
"//pkg/apis/core:go_default_library",
|
||||
"//pkg/apis/core/helper:go_default_library",
|
||||
"//pkg/bootstrap/api:go_default_library",
|
||||
"//pkg/controller:go_default_library",
|
||||
"//vendor/github.com/davecgh/go-spew/spew:go_default_library",
|
||||
"//vendor/github.com/stretchr/testify/assert:go_default_library",
|
||||
"//vendor/k8s.io/api/core/v1: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/client-go/informers:go_default_library",
|
||||
"//vendor/k8s.io/client-go/informers/core/v1:go_default_library",
|
||||
"//vendor/k8s.io/client-go/kubernetes/fake:go_default_library",
|
||||
"//vendor/k8s.io/client-go/testing:go_default_library",
|
||||
],
|
||||
|
|
@ -44,6 +47,7 @@ go_library(
|
|||
deps = [
|
||||
"//pkg/apis/core:go_default_library",
|
||||
"//pkg/bootstrap/api:go_default_library",
|
||||
"//pkg/controller:go_default_library",
|
||||
"//pkg/util/metrics:go_default_library",
|
||||
"//vendor/github.com/golang/glog:go_default_library",
|
||||
"//vendor/gopkg.in/square/go-jose.v2:go_default_library",
|
||||
|
|
@ -51,11 +55,14 @@ go_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/fields:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/labels:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/watch:go_default_library",
|
||||
"//vendor/k8s.io/client-go/informers/core/v1:go_default_library",
|
||||
"//vendor/k8s.io/client-go/kubernetes:go_default_library",
|
||||
"//vendor/k8s.io/client-go/listers/core/v1:go_default_library",
|
||||
"//vendor/k8s.io/client-go/tools/cache:go_default_library",
|
||||
"//vendor/k8s.io/client-go/util/workqueue:go_default_library",
|
||||
],
|
||||
|
|
|
|||
|
|
@ -22,25 +22,25 @@ import (
|
|||
|
||||
"github.com/golang/glog"
|
||||
|
||||
"fmt"
|
||||
"k8s.io/api/core/v1"
|
||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/fields"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/labels"
|
||||
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
|
||||
"k8s.io/apimachinery/pkg/util/wait"
|
||||
"k8s.io/apimachinery/pkg/watch"
|
||||
informers "k8s.io/client-go/informers/core/v1"
|
||||
clientset "k8s.io/client-go/kubernetes"
|
||||
corelisters "k8s.io/client-go/listers/core/v1"
|
||||
"k8s.io/client-go/tools/cache"
|
||||
"k8s.io/client-go/util/workqueue"
|
||||
api "k8s.io/kubernetes/pkg/apis/core"
|
||||
bootstrapapi "k8s.io/kubernetes/pkg/bootstrap/api"
|
||||
"k8s.io/kubernetes/pkg/controller"
|
||||
"k8s.io/kubernetes/pkg/util/metrics"
|
||||
)
|
||||
|
||||
// BootstrapSignerOptions contains options for the BootstrapSigner
|
||||
type BootstrapSignerOptions struct {
|
||||
|
||||
// ConfigMapNamespace is the namespace of the ConfigMap
|
||||
ConfigMapNamespace string
|
||||
|
||||
|
|
@ -73,86 +73,99 @@ func DefaultBootstrapSignerOptions() BootstrapSignerOptions {
|
|||
type BootstrapSigner struct {
|
||||
client clientset.Interface
|
||||
configMapKey string
|
||||
configMapName string
|
||||
configMapNamespace string
|
||||
secretNamespace string
|
||||
|
||||
configMaps cache.Store
|
||||
secrets cache.Store
|
||||
|
||||
// syncQueue handles synchronizing updates to the ConfigMap. We'll only ever
|
||||
// have one item (Named <ConfigMapName>) in this queue. We are using it
|
||||
// serializes and collapses updates as they can come from both the ConfigMap
|
||||
// and Secrets controllers.
|
||||
syncQueue workqueue.Interface
|
||||
syncQueue workqueue.RateLimitingInterface
|
||||
|
||||
// Since we join two objects, we'll watch both of them with controllers.
|
||||
configMapsController cache.Controller
|
||||
secretsController cache.Controller
|
||||
secretLister corelisters.SecretLister
|
||||
secretSynced cache.InformerSynced
|
||||
|
||||
configMapLister corelisters.ConfigMapLister
|
||||
configMapSynced cache.InformerSynced
|
||||
}
|
||||
|
||||
// NewBootstrapSigner returns a new *BootstrapSigner.
|
||||
//
|
||||
// TODO: Switch to shared informers
|
||||
func NewBootstrapSigner(cl clientset.Interface, options BootstrapSignerOptions) (*BootstrapSigner, error) {
|
||||
func NewBootstrapSigner(cl clientset.Interface, secrets informers.SecretInformer, configMaps informers.ConfigMapInformer, options BootstrapSignerOptions) (*BootstrapSigner, error) {
|
||||
e := &BootstrapSigner{
|
||||
client: cl,
|
||||
configMapKey: options.ConfigMapNamespace + "/" + options.ConfigMapName,
|
||||
configMapName: options.ConfigMapName,
|
||||
configMapNamespace: options.ConfigMapNamespace,
|
||||
secretNamespace: options.TokenSecretNamespace,
|
||||
syncQueue: workqueue.NewNamed("bootstrap_signer_queue"),
|
||||
secretLister: secrets.Lister(),
|
||||
secretSynced: secrets.Informer().HasSynced,
|
||||
configMapLister: configMaps.Lister(),
|
||||
configMapSynced: configMaps.Informer().HasSynced,
|
||||
syncQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "bootstrap_signer_queue"),
|
||||
}
|
||||
if cl.CoreV1().RESTClient().GetRateLimiter() != nil {
|
||||
if err := metrics.RegisterMetricAndTrackRateLimiterUsage("bootstrap_signer", cl.CoreV1().RESTClient().GetRateLimiter()); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
configMapSelector := fields.SelectorFromSet(map[string]string{api.ObjectNameField: options.ConfigMapName})
|
||||
e.configMaps, e.configMapsController = cache.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func(lo metav1.ListOptions) (runtime.Object, error) {
|
||||
lo.FieldSelector = configMapSelector.String()
|
||||
return e.client.CoreV1().ConfigMaps(options.ConfigMapNamespace).List(lo)
|
||||
|
||||
configMaps.Informer().AddEventHandlerWithResyncPeriod(
|
||||
cache.FilteringResourceEventHandler{
|
||||
FilterFunc: func(obj interface{}) bool {
|
||||
switch t := obj.(type) {
|
||||
case *v1.ConfigMap:
|
||||
return t.Name == options.ConfigMapName && t.Namespace == options.ConfigMapNamespace
|
||||
default:
|
||||
utilruntime.HandleError(fmt.Errorf("object passed to %T that is not expected: %T", e, obj))
|
||||
return false
|
||||
}
|
||||
},
|
||||
WatchFunc: func(lo metav1.ListOptions) (watch.Interface, error) {
|
||||
lo.FieldSelector = configMapSelector.String()
|
||||
return e.client.CoreV1().ConfigMaps(options.ConfigMapNamespace).Watch(lo)
|
||||
},
|
||||
},
|
||||
&v1.ConfigMap{},
|
||||
options.ConfigMapResync,
|
||||
cache.ResourceEventHandlerFuncs{
|
||||
Handler: cache.ResourceEventHandlerFuncs{
|
||||
AddFunc: func(_ interface{}) { e.pokeConfigMapSync() },
|
||||
UpdateFunc: func(_, _ interface{}) { e.pokeConfigMapSync() },
|
||||
},
|
||||
},
|
||||
options.ConfigMapResync,
|
||||
)
|
||||
|
||||
secretSelector := fields.SelectorFromSet(map[string]string{api.SecretTypeField: string(bootstrapapi.SecretTypeBootstrapToken)})
|
||||
e.secrets, e.secretsController = cache.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func(lo metav1.ListOptions) (runtime.Object, error) {
|
||||
lo.FieldSelector = secretSelector.String()
|
||||
return e.client.CoreV1().Secrets(e.secretNamespace).List(lo)
|
||||
secrets.Informer().AddEventHandlerWithResyncPeriod(
|
||||
cache.FilteringResourceEventHandler{
|
||||
FilterFunc: func(obj interface{}) bool {
|
||||
switch t := obj.(type) {
|
||||
case *v1.Secret:
|
||||
return t.Type == bootstrapapi.SecretTypeBootstrapToken && t.Namespace == e.secretNamespace
|
||||
default:
|
||||
utilruntime.HandleError(fmt.Errorf("object passed to %T that is not expected: %T", e, obj))
|
||||
return false
|
||||
}
|
||||
},
|
||||
WatchFunc: func(lo metav1.ListOptions) (watch.Interface, error) {
|
||||
lo.FieldSelector = secretSelector.String()
|
||||
return e.client.CoreV1().Secrets(e.secretNamespace).Watch(lo)
|
||||
},
|
||||
},
|
||||
&v1.Secret{},
|
||||
options.SecretResync,
|
||||
cache.ResourceEventHandlerFuncs{
|
||||
Handler: cache.ResourceEventHandlerFuncs{
|
||||
AddFunc: func(_ interface{}) { e.pokeConfigMapSync() },
|
||||
UpdateFunc: func(_, _ interface{}) { e.pokeConfigMapSync() },
|
||||
DeleteFunc: func(_ interface{}) { e.pokeConfigMapSync() },
|
||||
},
|
||||
},
|
||||
options.SecretResync,
|
||||
)
|
||||
|
||||
return e, nil
|
||||
}
|
||||
|
||||
// Run runs controller loops and returns when they are done
|
||||
func (e *BootstrapSigner) Run(stopCh <-chan struct{}) {
|
||||
go e.configMapsController.Run(stopCh)
|
||||
go e.secretsController.Run(stopCh)
|
||||
// Shut down queues
|
||||
defer utilruntime.HandleCrash()
|
||||
defer e.syncQueue.ShutDown()
|
||||
|
||||
if !controller.WaitForCacheSync("bootstrap_signer", stopCh, e.configMapSynced, e.secretSynced) {
|
||||
return
|
||||
}
|
||||
|
||||
glog.V(5).Infof("Starting workers")
|
||||
go wait.Until(e.serviceConfigMapQueue, 0, stopCh)
|
||||
<-stopCh
|
||||
glog.V(1).Infof("Shutting down")
|
||||
}
|
||||
|
||||
func (e *BootstrapSigner) pokeConfigMapSync() {
|
||||
|
|
@ -237,27 +250,32 @@ func (e *BootstrapSigner) updateConfigMap(cm *v1.ConfigMap) {
|
|||
|
||||
// getConfigMap gets the ConfigMap we are interested in
|
||||
func (e *BootstrapSigner) getConfigMap() *v1.ConfigMap {
|
||||
configMap, exists, err := e.configMaps.GetByKey(e.configMapKey)
|
||||
configMap, err := e.configMapLister.ConfigMaps(e.configMapNamespace).Get(e.configMapName)
|
||||
|
||||
// If we can't get the configmap just return nil. The resync will eventually
|
||||
// sync things up.
|
||||
if err != nil {
|
||||
if !apierrors.IsNotFound(err) {
|
||||
utilruntime.HandleError(err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
return configMap
|
||||
}
|
||||
|
||||
func (e *BootstrapSigner) listSecrets() []*v1.Secret {
|
||||
secrets, err := e.secretLister.Secrets(e.secretNamespace).List(labels.Everything())
|
||||
if err != nil {
|
||||
utilruntime.HandleError(err)
|
||||
return nil
|
||||
}
|
||||
|
||||
if exists {
|
||||
return configMap.(*v1.ConfigMap)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *BootstrapSigner) listSecrets() []*v1.Secret {
|
||||
secrets := e.secrets.List()
|
||||
|
||||
items := []*v1.Secret{}
|
||||
for _, obj := range secrets {
|
||||
items = append(items, obj.(*v1.Secret))
|
||||
for _, secret := range secrets {
|
||||
if secret.Type == bootstrapapi.SecretTypeBootstrapToken {
|
||||
items = append(items, secret)
|
||||
}
|
||||
}
|
||||
return items
|
||||
}
|
||||
|
|
|
|||
|
|
@ -24,10 +24,13 @@ import (
|
|||
"k8s.io/api/core/v1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/client-go/informers"
|
||||
coreinformers "k8s.io/client-go/informers/core/v1"
|
||||
"k8s.io/client-go/kubernetes/fake"
|
||||
core "k8s.io/client-go/testing"
|
||||
api "k8s.io/kubernetes/pkg/apis/core"
|
||||
bootstrapapi "k8s.io/kubernetes/pkg/bootstrap/api"
|
||||
"k8s.io/kubernetes/pkg/controller"
|
||||
)
|
||||
|
||||
func init() {
|
||||
|
|
@ -36,14 +39,17 @@ func init() {
|
|||
|
||||
const testTokenID = "abc123"
|
||||
|
||||
func newBootstrapSigner() (*BootstrapSigner, *fake.Clientset, error) {
|
||||
func newBootstrapSigner() (*BootstrapSigner, *fake.Clientset, coreinformers.SecretInformer, coreinformers.ConfigMapInformer, error) {
|
||||
options := DefaultBootstrapSignerOptions()
|
||||
cl := fake.NewSimpleClientset()
|
||||
bsc, err := NewBootstrapSigner(cl, options)
|
||||
informers := informers.NewSharedInformerFactory(fake.NewSimpleClientset(), controller.NoResyncPeriodFunc())
|
||||
secrets := informers.Core().V1().Secrets()
|
||||
configMaps := informers.Core().V1().ConfigMaps()
|
||||
bsc, err := NewBootstrapSigner(cl, secrets, configMaps, options)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
return nil, nil, nil, nil, err
|
||||
}
|
||||
return bsc, cl, nil
|
||||
return bsc, cl, secrets, configMaps, nil
|
||||
}
|
||||
|
||||
func newConfigMap(tokenID, signature string) *v1.ConfigMap {
|
||||
|
|
@ -64,7 +70,7 @@ func newConfigMap(tokenID, signature string) *v1.ConfigMap {
|
|||
}
|
||||
|
||||
func TestNoConfigMap(t *testing.T) {
|
||||
signer, cl, err := newBootstrapSigner()
|
||||
signer, cl, _, _, err := newBootstrapSigner()
|
||||
if err != nil {
|
||||
t.Fatalf("error creating BootstrapSigner: %v", err)
|
||||
}
|
||||
|
|
@ -73,17 +79,17 @@ func TestNoConfigMap(t *testing.T) {
|
|||
}
|
||||
|
||||
func TestSimpleSign(t *testing.T) {
|
||||
signer, cl, err := newBootstrapSigner()
|
||||
signer, cl, secrets, configMaps, err := newBootstrapSigner()
|
||||
if err != nil {
|
||||
t.Fatalf("error creating BootstrapSigner: %v", err)
|
||||
}
|
||||
|
||||
cm := newConfigMap("", "")
|
||||
signer.configMaps.Add(cm)
|
||||
configMaps.Informer().GetIndexer().Add(cm)
|
||||
|
||||
secret := newTokenSecret(testTokenID, "tokenSecret")
|
||||
addSecretSigningUsage(secret, "true")
|
||||
signer.secrets.Add(secret)
|
||||
secrets.Informer().GetIndexer().Add(secret)
|
||||
|
||||
signer.signConfigMap()
|
||||
|
||||
|
|
@ -97,17 +103,17 @@ func TestSimpleSign(t *testing.T) {
|
|||
}
|
||||
|
||||
func TestNoSignNeeded(t *testing.T) {
|
||||
signer, cl, err := newBootstrapSigner()
|
||||
signer, cl, secrets, configMaps, err := newBootstrapSigner()
|
||||
if err != nil {
|
||||
t.Fatalf("error creating BootstrapSigner: %v", err)
|
||||
}
|
||||
|
||||
cm := newConfigMap(testTokenID, "eyJhbGciOiJIUzI1NiIsImtpZCI6ImFiYzEyMyJ9..QSxpUG7Q542CirTI2ECPSZjvBOJURUW5a7XqFpNI958")
|
||||
signer.configMaps.Add(cm)
|
||||
configMaps.Informer().GetIndexer().Add(cm)
|
||||
|
||||
secret := newTokenSecret(testTokenID, "tokenSecret")
|
||||
addSecretSigningUsage(secret, "true")
|
||||
signer.secrets.Add(secret)
|
||||
secrets.Informer().GetIndexer().Add(secret)
|
||||
|
||||
signer.signConfigMap()
|
||||
|
||||
|
|
@ -115,17 +121,17 @@ func TestNoSignNeeded(t *testing.T) {
|
|||
}
|
||||
|
||||
func TestUpdateSignature(t *testing.T) {
|
||||
signer, cl, err := newBootstrapSigner()
|
||||
signer, cl, secrets, configMaps, err := newBootstrapSigner()
|
||||
if err != nil {
|
||||
t.Fatalf("error creating BootstrapSigner: %v", err)
|
||||
}
|
||||
|
||||
cm := newConfigMap(testTokenID, "old signature")
|
||||
signer.configMaps.Add(cm)
|
||||
configMaps.Informer().GetIndexer().Add(cm)
|
||||
|
||||
secret := newTokenSecret(testTokenID, "tokenSecret")
|
||||
addSecretSigningUsage(secret, "true")
|
||||
signer.secrets.Add(secret)
|
||||
secrets.Informer().GetIndexer().Add(secret)
|
||||
|
||||
signer.signConfigMap()
|
||||
|
||||
|
|
@ -139,13 +145,13 @@ func TestUpdateSignature(t *testing.T) {
|
|||
}
|
||||
|
||||
func TestRemoveSignature(t *testing.T) {
|
||||
signer, cl, err := newBootstrapSigner()
|
||||
signer, cl, _, configMaps, err := newBootstrapSigner()
|
||||
if err != nil {
|
||||
t.Fatalf("error creating BootstrapSigner: %v", err)
|
||||
}
|
||||
|
||||
cm := newConfigMap(testTokenID, "old signature")
|
||||
signer.configMaps.Add(cm)
|
||||
configMaps.Informer().GetIndexer().Add(cm)
|
||||
|
||||
signer.signConfigMap()
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue