From b18e618a515eb249dff03b5b85ce7ae10db51afa Mon Sep 17 00:00:00 2001 From: Andreas Fritzler Date: Tue, 27 Jun 2023 08:11:37 +0200 Subject: [PATCH] refactor: add file watcher for TLS certificate files - Add fsnotify dependency to go.mod - Add file watcher for certificate files in `cmd/agent/app/server.go` and `cmd/server/app/server.go` - Reload TLS config and credentials on certificate file change in `cmd/agent/app/server.go` and `cmd/server/app/server.go` - Refactor and clean up file watcher code in `cmd/agent/app/server.go` and `cmd/server/app/server.go` Signed-off-by: Andreas Fritzler --- cmd/agent/app/server.go | 84 ++++++++++++++++++++++--- cmd/server/app/server.go | 129 ++++++++++++++++++++++++++++++++------- go.mod | 1 + go.sum | 3 + 4 files changed, 185 insertions(+), 32 deletions(-) diff --git a/cmd/agent/app/server.go b/cmd/agent/app/server.go index d0a13a7ef..97a84f7ae 100644 --- a/cmd/agent/app/server.go +++ b/cmd/agent/app/server.go @@ -28,6 +28,7 @@ import ( "strconv" "time" + "github.com/fsnotify/fsnotify" "github.com/prometheus/client_golang/prometheus/promhttp" "github.com/spf13/cobra" "google.golang.org/grpc" @@ -83,19 +84,82 @@ func (a *Agent) run(o *options.GrpcProxyAgentOptions) error { func (a *Agent) runProxyConnection(o *options.GrpcProxyAgentOptions, stopCh <-chan struct{}) error { var tlsConfig *tls.Config var err error - if tlsConfig, err = util.GetClientTLSConfig(o.CaCert, o.AgentCert, o.AgentKey, o.ProxyServerHost, o.AlpnProtos); err != nil { + + watcher, err := fsnotify.NewWatcher() + if err != nil { + klog.Fatal(err) + } + defer func(watcher *fsnotify.Watcher) { + if err := watcher.Close(); err != nil { + klog.ErrorS(err, "failed to close watcher") + return + } + }(watcher) + + // Watch the certificate files + if err := watcher.Add(o.AgentCert); err != nil { return err } - dialOptions := []grpc.DialOption{ - grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig)), - grpc.WithKeepaliveParams(keepalive.ClientParameters{ - Time: o.KeepaliveTime, - PermitWithoutStream: true, - }), + if err := watcher.Add(o.AgentKey); err != nil { + return err } - cc := o.ClientSetConfig(dialOptions...) - cs := cc.NewAgentClientSet(stopCh) - cs.Serve() + if err := watcher.Add(o.CaCert); err != nil { + return err + } + + reload := make(chan bool) + + // Goroutine to watch for file changes + go func() { + for { + select { + case event, ok := <-watcher.Events: + if !ok { + return + } + if event.Op&fsnotify.Write == fsnotify.Write { + reload <- true + } + case err, ok := <-watcher.Errors: + if !ok { + return + } + klog.ErrorS(err, "failed to watch for file changes") + case <-stopCh: + // Handle graceful shutdown + return + } + } + }() + + // Goroutine to handle main logic + go func() { + for { + select { + case <-reload: + tlsConfig, err = util.GetClientTLSConfig(o.CaCert, o.AgentCert, o.AgentKey, o.ProxyServerHost, o.AlpnProtos) + if err != nil { + klog.ErrorS(err, "Failed to reload TLS config") + continue + } + + dialOptions := []grpc.DialOption{ + grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig)), + grpc.WithKeepaliveParams(keepalive.ClientParameters{ + Time: o.KeepaliveTime, + PermitWithoutStream: true, + }), + } + cc := o.ClientSetConfig(dialOptions...) + cs := cc.NewAgentClientSet(stopCh) + cs.Serve() + + case <-stopCh: + // Handle server shutdown + return + } + } + }() return nil } diff --git a/cmd/server/app/server.go b/cmd/server/app/server.go index 968a07450..e1514d8e2 100644 --- a/cmd/server/app/server.go +++ b/cmd/server/app/server.go @@ -35,6 +35,7 @@ import ( "syscall" "time" + "github.com/fsnotify/fsnotify" "github.com/prometheus/client_golang/prometheus/promhttp" "github.com/spf13/cobra" "google.golang.org/grpc" @@ -135,7 +136,7 @@ func (p *Proxy) run(o *options.ProxyRunOptions) error { } klog.V(1).Infoln("Starting agent server for tunnel connections.") - err = p.runAgentServer(o, server) + err = p.runAgentServer(o, server, ctx.Done()) if err != nil { return fmt.Errorf("failed to run the agent server: %v", err) } @@ -353,33 +354,117 @@ func (p *Proxy) runMTLSFrontendServer(ctx context.Context, o *options.ProxyRunOp return stop, nil } -func (p *Proxy) runAgentServer(o *options.ProxyRunOptions, server *server.ProxyServer) error { +func (p *Proxy) runAgentServer(o *options.ProxyRunOptions, server *server.ProxyServer, stopCh <-chan struct{}) error { var tlsConfig *tls.Config var err error - if tlsConfig, err = p.getTLSConfig(o.ClusterCaCert, o.ClusterCert, o.ClusterKey, o.CipherSuites); err != nil { - return err + + watcher, err := fsnotify.NewWatcher() + if err != nil { + klog.Fatal(err) } + defer func(watcher *fsnotify.Watcher) { + if err := watcher.Close(); err != nil { + klog.ErrorS(err, "failed to close watcher") + return + } + }(watcher) - addr := net.JoinHostPort(o.AgentBindAddress, strconv.Itoa(o.AgentPort)) - agentServerOptions := []grpc.ServerOption{ - grpc.Creds(credentials.NewTLS(tlsConfig)), - grpc.KeepaliveParams(keepalive.ServerParameters{Time: o.KeepaliveTime}), - grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{ - MinTime: 30 * time.Second, - PermitWithoutStream: true, - }), + // Watch the certificate files + if err := watcher.Add(o.ClusterCaCert); err != nil { + return err } - grpcServer := grpc.NewServer(agentServerOptions...) - agent.RegisterAgentServiceServer(grpcServer, server) - lis, err := net.Listen("tcp", addr) - if err != nil { - return fmt.Errorf("failed to listen on %s: %v", addr, err) + if err := watcher.Add(o.ClusterCert); err != nil { + return err } - labels := runpprof.Labels( - "core", "agentListener", - "port", strconv.FormatUint(uint64(o.AgentPort), 10), - ) - go runpprof.Do(context.Background(), labels, func(context.Context) { grpcServer.Serve(lis) }) + if err := watcher.Add(o.ClusterKey); err != nil { + return err + } + + reload := make(chan bool) + restartServer := make(chan bool) + + // Goroutine to watch for file changes + go func() { + for { + select { + case event, ok := <-watcher.Events: + if !ok { + return + } + if event.Op&fsnotify.Write == fsnotify.Write { + reload <- true + } + case err, ok := <-watcher.Errors: + if !ok { + return + } + klog.ErrorS(err, "failed to watch for file changes") + case <-stopCh: + // Handle graceful shutdown + return + } + } + }() + + addr := net.JoinHostPort(o.AgentBindAddress, strconv.Itoa(o.AgentPort)) + + go func() { + var grpcServer *grpc.Server + var lis net.Listener + + for { + select { + case <-reload: + if tlsConfig, err = p.getTLSConfig(o.ClusterCaCert, o.ClusterCert, o.ClusterKey, o.CipherSuites); err != nil { + klog.ErrorS(err, "Failed to reload TLS config:") + } + klog.V(1).Info("TLS config reloaded") + restartServer <- true + + case <-restartServer: + if grpcServer != nil { + grpcServer.GracefulStop() + klog.V(1).Info("gRPC server stopped") + } + + agentServerOptions := []grpc.ServerOption{ + grpc.Creds(credentials.NewTLS(tlsConfig)), + grpc.KeepaliveParams(keepalive.ServerParameters{Time: o.KeepaliveTime}), + grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{ + MinTime: 30 * time.Second, + PermitWithoutStream: true, + }), + } + grpcServer = grpc.NewServer(agentServerOptions...) + agent.RegisterAgentServiceServer(grpcServer, server) + + lis, err = net.Listen("tcp", addr) + if err != nil { + klog.Fatalf("failed to listen on %s: %v", addr, err) + } + + go func() { + if err := grpcServer.Serve(lis); err != nil { + klog.Fatalf("failed to serve: %v", err) + } + }() + + klog.V(1).Info("gRPC server restarted") + case <-stopCh: + // Handle graceful shutdown + if grpcServer != nil { + grpcServer.GracefulStop() + } + if lis != nil { + lis.Close() + } + return + } + } + }() + + // Initial restart to start the server + restartServer <- true return nil } diff --git a/go.mod b/go.mod index fae812f91..8caff4d5e 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module sigs.k8s.io/apiserver-network-proxy go 1.19 require ( + github.com/fsnotify/fsnotify v1.6.0 github.com/golang/mock v1.6.0 github.com/google/uuid v1.1.2 github.com/prometheus/client_golang v1.12.1 diff --git a/go.sum b/go.sum index f7621d91c..156cc6c7c 100644 --- a/go.sum +++ b/go.sum @@ -107,6 +107,8 @@ github.com/form3tech-oss/jwt-go v3.2.2+incompatible/go.mod h1:pbq4aXjuKjdthFRnoD github.com/form3tech-oss/jwt-go v3.2.3+incompatible/go.mod h1:pbq4aXjuKjdthFRnoDwaVPLA+WlJuPGy+QneDUgJi2k= github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo= github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ= +github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4HY= +github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw= github.com/getkin/kin-openapi v0.76.0/go.mod h1:660oXbgy5JFMKreazJaQTw7o+X00qeSyhcnluiMv+Xg= github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04= github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU= @@ -534,6 +536,7 @@ golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220209214540-3681064d5158/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.5.0 h1:MUK/U/4lj1t1oPg0HfuXDN/Z1wv31ZJ/YcPiGccS4DU= golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=