Skip to content

Commit

Permalink
refactor: add file watcher for TLS certificate files
Browse files Browse the repository at this point in the history 
- 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 <[email protected]>
  • Loading branch information
@afritzler
afritzler committed Jun 27, 2023
1 parent 6f42ff3 commit b18e618
Show file tree
Hide file tree
Showing 4 changed files with 185 additions and 32 deletions.
84 changes: 74 additions & 10 deletions cmd/agent/app/server.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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
}
Expand Down
129 changes: 107 additions & 22 deletions cmd/server/app/server.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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)
}
Expand Down Expand Up @@ -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
}
Expand Down
1 change: 1 addition & 0 deletions go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
3 changes: 3 additions & 0 deletions go.sum
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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=
Expand Down Expand Up @@ -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=
Expand Down

0 comments on commit b18e618

Please sign in to comment.