Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Remove watch timeout to allow call staggering #1296

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
+33 −15
Open

Remove watch timeout to allow call staggering #1296

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2286fba
Remove watch timeout to allow call staggering
michaely-cb Jun 11, 2024
c4db076
fix tests
michaely-cb Jun 21, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ require (
gopkg.in/natefinch/lumberjack.v2 v2.0.0
k8s.io/api v0.29.0
k8s.io/apimachinery v0.29.0
k8s.io/client-go v1.5.2
k8s.io/client-go v0.29.0
michaely-cb marked this conversation as resolved.
Show resolved Hide resolved
k8s.io/klog v1.0.0
k8s.io/klog/v2 v2.110.1
k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00 // indirect
Expand Down
2 changes: 2 additions & 0 deletions pkg/k8sclient/k8sclient.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,9 @@ type NoK8sNetworkError struct {
// ClientInfo contains information given from k8s client
type ClientInfo struct {
Client kubernetes.Interface
WatchClient kubernetes.Interface
NetClient netclient.Interface
NetWatchClient netclient.Interface
EventBroadcaster record.EventBroadcaster
EventRecorder record.EventRecorder

Expand Down
17 changes: 15 additions & 2 deletions pkg/k8sclient/kubeconfig.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -228,8 +228,19 @@ func newClientInfo(config *rest.Config) (*ClientInfo, error) {
if err != nil {
return nil, err
}
netClient, err := netclient.NewForConfig(config)
if err != nil {
return nil, err
}

netclient, err := netclient.NewForConfig(config)
watchConfig := rest.CopyConfig(config)
// Do not set timeout for watches and delegate timeout to client-go
adrianchiris marked this conversation as resolved.
Show resolved Hide resolved
watchConfig.Timeout = 0
watchClient, err := kubernetes.NewForConfig(watchConfig)
if err != nil {
return nil, err
}
netWatchClient, err := netclient.NewForConfig(watchConfig)
if err != nil {
return nil, err
}
Expand All @@ -240,7 +251,9 @@ func newClientInfo(config *rest.Config) (*ClientInfo, error) {
recorder := broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "multus"})
return &ClientInfo{
Client: client,
NetClient: netclient,
WatchClient: watchClient,
NetClient: netClient,
NetWatchClient: netWatchClient,
EventBroadcaster: broadcaster,
EventRecorder: recorder,
}, nil
Expand Down
1 change: 1 addition & 0 deletions pkg/multus/multus_cni100_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ import (
cni100 "github.com/containernetworking/cni/pkg/types/100"
"github.com/containernetworking/plugins/pkg/ns"
"github.com/containernetworking/plugins/pkg/testutils"

"gopkg.in/k8snetworkplumbingwg/multus-cni.v4/pkg/k8sclient"
"gopkg.in/k8snetworkplumbingwg/multus-cni.v4/pkg/logging"
testhelpers "gopkg.in/k8snetworkplumbingwg/multus-cni.v4/pkg/testing"
Expand Down
5 changes: 4 additions & 1 deletion pkg/multus/multus_suite_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,10 +31,11 @@ import (
cni100 "github.com/containernetworking/cni/pkg/types/100"
cniversion "github.com/containernetworking/cni/pkg/version"
netfake "github.com/k8snetworkplumbingwg/network-attachment-definition-client/pkg/client/clientset/versioned/fake"
"gopkg.in/k8snetworkplumbingwg/multus-cni.v4/pkg/k8sclient"
"k8s.io/client-go/kubernetes/fake"
"k8s.io/client-go/tools/record"

"gopkg.in/k8snetworkplumbingwg/multus-cni.v4/pkg/k8sclient"

. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
)
Expand Down Expand Up @@ -225,6 +226,8 @@ func (f *fakeExec) FindInPath(plugin string, paths []string) (string, error) {
// NewFakeClientInfo returns fake client (just for testing)
func NewFakeClientInfo() *k8sclient.ClientInfo {
return &k8sclient.ClientInfo{
// We use watch clients to avoid reconnections in fixed intervals in production, though we do not
// distinguish between non-watch clients and watch clients in tests for simplicity
Client: fake.NewSimpleClientset(),
NetClient: netfake.NewSimpleClientset(),
EventRecorder: record.NewFakeRecorder(10),
Expand Down
17 changes: 8 additions & 9 deletions pkg/server/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -158,10 +158,10 @@ func informerObjectTrim(obj interface{}) (interface{}, error) {
return obj, nil
}

func newNetDefInformer(netdefClient netdefclient.Interface) (netdefinformer.SharedInformerFactory, cache.SharedIndexInformer) {
func newNetDefInformer(netWatchClient netdefclient.Interface) (netdefinformer.SharedInformerFactory, cache.SharedIndexInformer) {
const resyncInterval time.Duration = 1 * time.Second

informerFactory := netdefinformer.NewSharedInformerFactoryWithOptions(netdefClient, resyncInterval)
informerFactory := netdefinformer.NewSharedInformerFactoryWithOptions(netWatchClient, resyncInterval)
netdefInformer := informerFactory.InformerFor(&netdefv1.NetworkAttachmentDefinition{}, func(client netdefclient.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
return netdefinformerv1.NewNetworkAttachmentDefinitionInformer(
client,
Expand All @@ -173,7 +173,7 @@ func newNetDefInformer(netdefClient netdefclient.Interface) (netdefinformer.Shar
return informerFactory, netdefInformer
}

func newPodInformer(kubeClient kubernetes.Interface, nodeName string) (internalinterfaces.SharedInformerFactory, cache.SharedIndexInformer) {
func newPodInformer(watchClient kubernetes.Interface, nodeName string) (internalinterfaces.SharedInformerFactory, cache.SharedIndexInformer) {
var tweakFunc internalinterfaces.TweakListOptionsFunc
if nodeName != "" {
logging.Verbosef("Filtering pod watch for node %q", nodeName)
Expand All @@ -185,7 +185,7 @@ func newPodInformer(kubeClient kubernetes.Interface, nodeName string) (internali

const resyncInterval time.Duration = 1 * time.Second

informerFactory := informerfactory.NewSharedInformerFactoryWithOptions(kubeClient, resyncInterval, informerfactory.WithTransform(informerObjectTrim))
informerFactory := informerfactory.NewSharedInformerFactoryWithOptions(watchClient, resyncInterval, informerfactory.WithTransform(informerObjectTrim))
podInformer := informerFactory.InformerFor(&kapi.Pod{}, func(c kubernetes.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
return v1coreinformers.NewFilteredPodInformer(
c,
Expand All @@ -194,7 +194,6 @@ func newPodInformer(kubeClient kubernetes.Interface, nodeName string) (internali
cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
tweakFunc)
})

return informerFactory, podInformer
}

Expand Down Expand Up @@ -255,8 +254,8 @@ func NewCNIServer(daemonConfig *ControllerNetConf, serverConfig []byte, ignoreRe
}

func newCNIServer(rundir string, kubeClient *k8s.ClientInfo, exec invoke.Exec, servConfig []byte, ignoreReadinessIndicator bool) (*Server, error) {
informerFactory, podInformer := newPodInformer(kubeClient.Client, os.Getenv("MULTUS_NODE_NAME"))
netdefInformerFactory, netdefInformer := newNetDefInformer(kubeClient.NetClient)
podInformerFactory, podInformer := newPodInformer(kubeClient.WatchClient, os.Getenv("MULTUS_NODE_NAME"))
netdefInformerFactory, netdefInformer := newNetDefInformer(kubeClient.NetWatchClient)
kubeClient.SetK8sClientInformers(podInformer, netdefInformer)

router := http.NewServeMux()
Expand All @@ -277,7 +276,7 @@ func newCNIServer(rundir string, kubeClient *k8s.ClientInfo, exec invoke.Exec, s
[]string{"handler", "code", "method"},
),
},
informerFactory: informerFactory,
podInformerFactory: podInformerFactory,
podInformer: podInformer,
netdefInformerFactory: netdefInformerFactory,
netdefInformer: netdefInformer,
Expand Down Expand Up @@ -356,7 +355,7 @@ func newCNIServer(rundir string, kubeClient *k8s.ClientInfo, exec invoke.Exec, s

// Start starts the server and begins serving on the given listener
func (s *Server) Start(ctx context.Context, l net.Listener) {
s.informerFactory.Start(ctx.Done())
s.podInformerFactory.Start(ctx.Done())
s.netdefInformerFactory.Start(ctx.Done())

// Give the initial sync some time to complete in large clusters, but
Expand Down
2 changes: 1 addition & 1 deletion pkg/server/types.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ type Server struct {
exec invoke.Exec
serverConfig []byte
metrics *Metrics
informerFactory internalinterfaces.SharedInformerFactory
podInformerFactory internalinterfaces.SharedInformerFactory
podInformer cache.SharedIndexInformer
netdefInformerFactory netdefinformer.SharedInformerFactory
netdefInformer cache.SharedIndexInformer
Expand Down
2 changes: 1 addition & 1 deletion vendor/modules.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -478,7 +478,7 @@ k8s.io/apimachinery/pkg/version
k8s.io/apimachinery/pkg/watch
k8s.io/apimachinery/third_party/forked/golang/json
k8s.io/apimachinery/third_party/forked/golang/reflect
# k8s.io/client-go v1.5.2 => k8s.io/client-go v0.29.0
# k8s.io/client-go v0.29.0
## explicit; go 1.21
k8s.io/client-go/applyconfigurations/admissionregistration/v1
k8s.io/client-go/applyconfigurations/admissionregistration/v1alpha1
Expand Down