scottk/kubernetes
1
0
Fork 0
mirror of https://github.com/optim-enterprises-bv/kubernetes.git synced 2025-11-30 21:33:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Process plugins found before the ones watched

Browse Source 
Currently, the method `pluginwatcher.traversePluginDir` descends into
a directory adding filesystem watchers and creating synthetic `create`
events when it finds sockets files. However, a race condition might
happen when a recently-added watcher observes a `delete` event in a
socket file before `pluginwatcher.traversePlugindir` itself notices
this file.

This patch changes this behavior by registering watchers on
directories, enqueueing and processing `create` events from sockets
found, and only then processing the events from the registered watchers.
This commit is contained in:
Fabio Bertinatto
2019-03-11 13:05:07 +01:00
parent 90fbbee129
commit 817c9ca953
3 changed files with 43 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
pkg/kubelet/util/pluginwatcher/plugin_watcher_test.go
View File

@@ -27,6 +27,7 @@ import (
"github.com/stretchr/testify/require"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/klog"
registerapi "k8s.io/kubernetes/pkg/kubelet/apis/pluginregistration/v1"
)
@@ -173,9 +174,6 @@ func TestPluginRegistrationAtKubeletStart(t *testing.T) {
plugins[i] = p
}
w := newWatcherWithHandler(t, hdlr, false /* testDeprecatedDir */)
defer func() { require.NoError(t, w.Stop()) }()
var wg sync.WaitGroup
for i := 0; i < len(plugins); i++ {
wg.Add(1)
@@ -189,6 +187,9 @@ func TestPluginRegistrationAtKubeletStart(t *testing.T) {
}(plugins[i])
}
w := newWatcherWithHandler(t, hdlr, false /* testDeprecatedDir */)
defer func() { require.NoError(t, w.Stop()) }()
c := make(chan struct{})
go func() {
defer close(c)
@@ -198,7 +199,7 @@ func TestPluginRegistrationAtKubeletStart(t *testing.T) {
select {
case <-c:
return
case <-time.After(2 * time.Second):
case <-time.After(wait.ForeverTestTimeout):
t.Fatalf("Timeout while waiting for the plugin registration status")
}
}
@@ -238,11 +239,22 @@ func TestPlugiRegistrationFailureWithUnsupportedVersionAtKubeletStart(t *testing
hdlr := NewExampleHandler(supportedVersions, false /* permitDeprecatedDir */)
hdlr.AddPluginName(pluginName)
c := make(chan struct{})
go func() {
defer close(c)
require.True(t, waitForEvent(t, exampleEventValidate, hdlr.EventChan(p.pluginName)))
require.False(t, waitForPluginRegistrationStatus(t, p.registrationStatus))
}()
w := newWatcherWithHandler(t, hdlr, false /* testDeprecatedDir */)
defer func() { require.NoError(t, w.Stop()) }()
require.True(t, waitForEvent(t, exampleEventValidate, hdlr.EventChan(p.pluginName)))
require.False(t, waitForPluginRegistrationStatus(t, p.registrationStatus))
select {
case <-c:
return
case <-time.After(wait.ForeverTestTimeout):
t.Fatalf("Timeout while waiting for the plugin registration status")
}
}
func waitForPluginRegistrationStatus(t *testing.T, statusChan chan registerapi.RegistrationStatus) bool {
@@ -259,7 +271,7 @@ func waitForEvent(t *testing.T, expected examplePluginEvent, eventChan chan exam
select {
case event := <-eventChan:
return event == expected
case <-time.After(2 * time.Second):
case <-time.After(wait.ForeverTestTimeout):
t.Fatalf("Timed out while waiting for registration status %v", expected)
}