github/kubernetes
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #104102 from tnqn/dynamic-file

Browse Source 
Improve dynamic cert file change detection
This commit is contained in:
Kubernetes Prow Robot
2021-08-05 16:36:26 -07:00
committed by GitHub
parent 8d7a797eb7 3cfe3d048f
commit 98bd200217
4 changed files with 155 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
staging/src/k8s.io/apiserver/go.mod
View File

@@ -11,6 +11,7 @@ require (
github.com/davecgh/go-spew v1.1.1
github.com/emicklei/go-restful v2.9.5+incompatible
github.com/evanphx/json-patch v4.11.0+incompatible
github.com/fsnotify/fsnotify v1.4.9
github.com/go-openapi/jsonpointer v0.19.5 // indirect
github.com/go-openapi/jsonreference v0.19.5 // indirect
github.com/go-openapi/swag v0.19.14 // indirect

64
staging/src/k8s.io/apiserver/pkg/server/dynamiccertificates/dynamic_cafile_content.go
View File

@@ -24,6 +24,7 @@ import (
"sync/atomic"
"time"
"github.com/fsnotify/fsnotify"
"k8s.io/client-go/util/cert"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -44,7 +45,7 @@ type ControllerRunner interface {
Run(workers int, stopCh <-chan struct{})
}
// DynamicFileCAContent provies a CAContentProvider that can dynamically react to new file content
// DynamicFileCAContent provides a CAContentProvider that can dynamically react to new file content
// It also fulfills the authenticator interface to provide verifyoptions
type DynamicFileCAContent struct {
name string
@@ -147,7 +148,7 @@ func (c *DynamicFileCAContent) RunOnce() error {
return c.loadCABundle()
}
// Run starts the kube-apiserver and blocks until stopCh is closed.
// Run starts the controller and blocks until stopCh is closed.
func (c *DynamicFileCAContent) Run(workers int, stopCh <-chan struct{}) {
defer utilruntime.HandleCrash()
defer c.queue.ShutDown()
@@ -158,17 +159,62 @@ func (c *DynamicFileCAContent) Run(workers int, stopCh <-chan struct{}) {
// doesn't matter what workers say, only start one.
go wait.Until(c.runWorker, time.Second, stopCh)
// start timer that rechecks every minute, just in case. this also serves to prime the controller quickly.
go wait.PollImmediateUntil(FileRefreshDuration, func() (bool, error) {
c.queue.Add(workItemKey)
return false, nil
}, stopCh)
// TODO this can be wired to an fsnotifier as well.
// start the loop that watches the CA file until stopCh is closed.
go wait.Until(func() {
if err := c.watchCAFile(stopCh); err != nil {
klog.ErrorS(err, "Failed to watch CA file, will retry later")
}
}, time.Minute, stopCh)
<-stopCh
}
func (c *DynamicFileCAContent) watchCAFile(stopCh <-chan struct{}) error {
// Trigger a check here to ensure the content will be checked periodically even if the following watch fails.
c.queue.Add(workItemKey)
w, err := fsnotify.NewWatcher()
if err != nil {
return fmt.Errorf("error creating fsnotify watcher: %v", err)
}
defer w.Close()
if err = w.Add(c.filename); err != nil {
return fmt.Errorf("error adding watch for file %s: %v", c.filename, err)
}
// Trigger a check in case the file is updated before the watch starts.
c.queue.Add(workItemKey)
for {
select {
case e := <-w.Events:
if err := c.handleWatchEvent(e, w); err != nil {
return err
}
case err := <-w.Errors:
return fmt.Errorf("received fsnotify error: %v", err)
case <-stopCh:
return nil
}
}
}
// handleWatchEvent triggers reloading the CA file, and restarts a new watch if it's a Remove or Rename event.
func (c *DynamicFileCAContent) handleWatchEvent(e fsnotify.Event, w *fsnotify.Watcher) error {
// This should be executed after restarting the watch (if applicable) to ensure no file event will be missing.
defer c.queue.Add(workItemKey)
if e.Op&(fsnotify.Remove|fsnotify.Rename) == 0 {
return nil
}
if err := w.Remove(c.filename); err != nil {
klog.InfoS("Failed to remove file watch, it may have been deleted", "file", c.filename, "err", err)
}
if err := w.Add(c.filename); err != nil {
return fmt.Errorf("error adding watch for file %s: %v", c.filename, err)
}
return nil
}
func (c *DynamicFileCAContent) runWorker() {
for c.processNextWorkItem() {
}

71
staging/src/k8s.io/apiserver/pkg/server/dynamiccertificates/dynamic_serving_content.go
View File

@@ -23,6 +23,8 @@ import (
"sync/atomic"
"time"
"github.com/fsnotify/fsnotify"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/util/workqueue"
@@ -38,7 +40,7 @@ type DynamicCertKeyPairContent struct {
// keyFile is the name of the key file to read.
keyFile string
// servingCert is a certKeyContent that contains the last read, non-zero length content of the key and cert
// certKeyPair is a certKeyContent that contains the last read, non-zero length content of the key and cert
certKeyPair atomic.Value
listeners []Listener
@@ -75,7 +77,7 @@ func (c *DynamicCertKeyPairContent) AddListener(listener Listener) {
c.listeners = append(c.listeners, listener)
}
// loadServingCert determines the next set of content for the file.
// loadCertKeyPair determines the next set of content for the file.
func (c *DynamicCertKeyPairContent) loadCertKeyPair() error {
cert, err := ioutil.ReadFile(c.certFile)
if err != nil {
@@ -132,17 +134,68 @@ func (c *DynamicCertKeyPairContent) Run(workers int, stopCh <-chan struct{}) {
// doesn't matter what workers say, only start one.
go wait.Until(c.runWorker, time.Second, stopCh)
// start timer that rechecks every minute, just in case. this also serves to prime the controller quickly.
go wait.PollImmediateUntil(FileRefreshDuration, func() (bool, error) {
c.queue.Add(workItemKey)
return false, nil
}, stopCh)
// TODO this can be wired to an fsnotifier as well.
// start the loop that watches the cert and key files until stopCh is closed.
go wait.Until(func() {
if err := c.watchCertKeyFile(stopCh); err != nil {
klog.ErrorS(err, "Failed to watch cert and key file, will retry later")
}
}, time.Minute, stopCh)
<-stopCh
}
func (c *DynamicCertKeyPairContent) watchCertKeyFile(stopCh <-chan struct{}) error {
// Trigger a check here to ensure the content will be checked periodically even if the following watch fails.
c.queue.Add(workItemKey)
w, err := fsnotify.NewWatcher()
if err != nil {
return fmt.Errorf("error creating fsnotify watcher: %v", err)
}
defer w.Close()
if err := w.Add(c.certFile); err != nil {
return fmt.Errorf("error adding watch for file %s: %v", c.certFile, err)
}
if err := w.Add(c.keyFile); err != nil {
return fmt.Errorf("error adding watch for file %s: %v", c.keyFile, err)
}
// Trigger a check in case the file is updated before the watch starts.
c.queue.Add(workItemKey)
for {
select {
case e := <-w.Events:
if err := c.handleWatchEvent(e, w); err != nil {
return err
}
case err := <-w.Errors:
return fmt.Errorf("received fsnotify error: %v", err)
case <-stopCh:
return nil
}
}
}
// handleWatchEvent triggers reloading the cert and key file, and restarts a new watch if it's a Remove or Rename event.
// If one file is updated before the other, the loadCertKeyPair method will catch the mismatch and will not apply the
// change. When an event of the other file is received, it will trigger reloading the files again and the new content
// will be loaded and used.
func (c *DynamicCertKeyPairContent) handleWatchEvent(e fsnotify.Event, w *fsnotify.Watcher) error {
// This should be executed after restarting the watch (if applicable) to ensure no file event will be missing.
defer c.queue.Add(workItemKey)
if e.Op&(fsnotify.Remove|fsnotify.Rename) == 0 {
return nil
}
if err := w.Remove(e.Name); err != nil {
klog.InfoS("Failed to remove file watch, it may have been deleted", "file", e.Name, "err", err)
}
if err := w.Add(e.Name); err != nil {
return fmt.Errorf("error adding watch for file %s: %v", e.Name, err)
}
return nil
}
func (c *DynamicCertKeyPairContent) runWorker() {
for c.processNextWorkItem() {
}