@@ -24,7 +24,10 @@ import (
2424 "os"
2525 "time"
2626
27+ "github.com/fsnotify/fsnotify"
2728 "github.com/sirupsen/logrus"
29+
30+ "github.com/projectcalico/calico/lib/std/chanutil"
2831)
2932
3033const ContentTypeMultilineJSON = "application/x-ndjson"
@@ -71,20 +74,96 @@ func newHTTPClient(caCert, clientKey, clientCert, serverName string) (*http.Clie
7174}
7275
7376func newEmitterClient (url , caCert , clientKey , clientCert , serverName string ) (* emitterClient , error ) {
77+ // Create an initial HTTP client, and a function to help encapsualte the reload logic.
7478 client , err := newHTTPClient (caCert , clientKey , clientCert , serverName )
7579 if err != nil {
7680 return nil , err
7781 }
78- return & emitterClient {url : url , client : client }, nil
82+
83+ updChan := make (chan struct {})
84+ getClient := func () (* http.Client , error ) {
85+ select {
86+ case _ , ok := <- updChan :
87+ if ok {
88+ // Only reload the client if the channel is still open. If the filewatcher
89+ // has been closed, we'll just continue using the existing client as best-effort.
90+ logrus .Info ("Reloading client after certificate change" )
91+ client , err = newHTTPClient (caCert , clientKey , clientCert , serverName )
92+ if err != nil {
93+ return nil , fmt .Errorf ("error reloading CA cert: %s" , err )
94+ }
95+ }
96+ default :
97+ // No change, return the existing client.
98+ }
99+ return client , nil
100+ }
101+
102+ if caCert != "" || clientKey != "" || clientCert != "" {
103+ // Start a goroutine to watch for changes to the CA cert file and feed
104+ // them into the update channel.
105+ monitorFn , err := watchFiles (updChan , caCert , clientCert , clientKey )
106+ if err != nil {
107+ return nil , fmt .Errorf ("error setting up CA cert file watcher: %s" , err )
108+ }
109+ go monitorFn ()
110+ }
111+
112+ return & emitterClient {
113+ url : url ,
114+ getClient : getClient ,
115+ }, nil
116+ }
117+
118+ func watchFiles (updChan chan struct {}, files ... string ) (func (), error ) {
119+ fileWatcher , err := fsnotify .NewWatcher ()
120+ if err != nil {
121+ return nil , fmt .Errorf ("error creating file watcher: %s" , err )
122+ }
123+ for _ , file := range files {
124+ if err := fileWatcher .Add (file ); err != nil {
125+ logrus .WithError (err ).Warn ("Error watching file for changes" )
126+ continue
127+ }
128+ logrus .WithField ("file" , file ).Debug ("Watching file for changes" )
129+ }
130+
131+ return func () {
132+ // If we exit this function, make sure to close the file watcher and update channel.
133+ defer fileWatcher .Close ()
134+ defer close (updChan )
135+ defer logrus .Info ("File watcher closed" )
136+ for {
137+ select {
138+ case event , ok := <- fileWatcher .Events :
139+ if ! ok {
140+ return
141+ }
142+ if event .Op & fsnotify .Write == fsnotify .Write {
143+ logrus .WithField ("file" , event .Name ).Info ("File changed, triggering update" )
144+ _ = chanutil .WriteNonBlocking (updChan , struct {}{})
145+ }
146+ case err , ok := <- fileWatcher .Errors :
147+ if ! ok {
148+ return
149+ }
150+ logrus .Errorf ("error watching CA cert file: %s" , err )
151+ }
152+ }
153+ }, nil
79154}
80155
81156type emitterClient struct {
82- url string
83- client * http.Client
157+ url string
158+ getClient func () ( * http.Client , error )
84159}
85160
86161func (e * emitterClient ) Post (body io.Reader ) error {
87- resp , err := e .client .Post (e .url , ContentTypeMultilineJSON , body )
162+ client , err := e .getClient ()
163+ if err != nil {
164+ return err
165+ }
166+ resp , err := client .Post (e .url , ContentTypeMultilineJSON , body )
88167 if err != nil {
89168 return err
90169 }
0 commit comments