hanahmily commented on code in PR #642:
URL:
https://github.com/apache/skywalking-banyandb/pull/642#discussion_r2030597976
##########
banyand/liaison/http/server.go:
##########
@@ -265,3 +307,170 @@ func serveFileContents(file string, files
http.FileSystem) http.HandlerFunc {
http.ServeContent(w, r, fi.Name(), fi.ModTime(), index)
}
}
+
+// TLSReloader manages dynamic reloading of TLS certificates and keys for the
HTTP server.
+type TLSReloader struct {
Review Comment:
It duplicates the one in gRPC. Please move it to a pkg package to share
between gRPC and HTTP.
##########
banyand/liaison/grpc/server.go:
##########
@@ -323,6 +341,186 @@ func (s *server) GracefulStop() {
}
}
+// TLSReloader manages dynamic reloading of TLS certificates and keys for the
gRPC server.
+type TLSReloader struct {
+ watcher *fsnotify.Watcher // 8 bytes
+ cert *tls.Certificate // 8 bytes
+ log *logger.Logger // 8 bytes
+ stopCh chan struct{} // 8 bytes
Review Comment:
stopCh seems redundant since you can exit the loop by closing the watcher.
##########
banyand/liaison/grpc/server.go:
##########
@@ -323,6 +341,186 @@ func (s *server) GracefulStop() {
}
}
+// TLSReloader manages dynamic reloading of TLS certificates and keys for the
gRPC server.
+type TLSReloader struct {
+ watcher *fsnotify.Watcher // 8 bytes
+ cert *tls.Certificate // 8 bytes
+ log *logger.Logger // 8 bytes
+ stopCh chan struct{} // 8 bytes
+ certFile string // 16 bytes
+ keyFile string // 16 bytes
+ mu sync.RWMutex // 24 bytes
+}
+
+// NewTLSReloader creates a new TLSReloader instance for the gRPC server.
+func NewTLSReloader(certFile, keyFile string, log *logger.Logger)
(*TLSReloader, error) {
+ if certFile == "" || keyFile == "" {
+ return nil, errors.New("certFile and keyFile must be provided")
+ }
+ if log == nil {
+ return nil, errors.New("logger must not be nil")
+ }
+
+ watcher, err := fsnotify.NewWatcher()
+ if err != nil {
+ return nil, errors.Wrap(err, "failed to create fsnotify
watcher")
+ }
+
+ cert, err := tls.LoadX509KeyPair(certFile, keyFile)
+ if err != nil {
+ watcher.Close()
+ return nil, errors.Wrap(err, "failed to load initial TLS
certificate")
+ }
+
+ log.Info().Str("certFile", certFile).Str("keyFile",
keyFile).Msg("Successfully loaded initial TLS certificates for gRPC")
+
+ tr := &TLSReloader{
+ certFile: certFile,
+ keyFile: keyFile,
+ cert: &cert,
+ log: log,
+ watcher: watcher,
+ stopCh: make(chan struct{}),
+ }
+
+ return tr, nil
+}
+
+// Start begins monitoring the TLS certificate and key files for changes.
+func (tr *TLSReloader) Start() error {
+ tr.log.Info().Str("certFile", tr.certFile).Str("keyFile",
tr.keyFile).Msg("Starting TLS file monitoring for gRPC")
+ err := tr.watcher.Add(tr.certFile)
+ if err != nil {
+ return errors.Wrapf(err, "failed to watch cert file: %s",
tr.certFile)
+ }
+ err = tr.watcher.Add(tr.keyFile)
+ if err != nil {
+ return errors.Wrapf(err, "failed to watch key file: %s",
tr.keyFile)
+ }
+ go tr.watchFiles()
+ return nil
+}
+
+// watchFiles monitors the certificate and key files for changes and reloads
credentials.
+func (tr *TLSReloader) watchFiles() {
+ tr.log.Info().Msg("TLS file watcher loop started for gRPC")
+ for {
+ select {
+ case event, ok := <-tr.watcher.Events:
+ if !ok {
+ tr.log.Warn().Msg("Watcher events channel
closed unexpectedly")
+ return
+ }
+ if event.Op&fsnotify.Remove == fsnotify.Remove {
+ if event.Name == tr.certFile {
+ if err := tr.watcher.Add(tr.certFile);
err != nil {
+
tr.log.Error().Err(err).Str("file", tr.certFile).Msg("Failed to re-add cert
file to watcher")
+ }
+ } else if event.Name == tr.keyFile {
+ if err := tr.watcher.Add(tr.keyFile);
err != nil {
+
tr.log.Error().Err(err).Str("file", tr.keyFile).Msg("Failed to re-add key file
to watcher")
+ }
+ }
+ }
+ if event.Op&fsnotify.Write == fsnotify.Write ||
+ event.Op&fsnotify.Rename == fsnotify.Rename ||
+ event.Op&fsnotify.Create == fsnotify.Create {
+ tr.log.Info().Str("file",
event.Name).Msg("Detected uploaded/changed certificate")
+ if err := tr.reloadCertificate(); err != nil {
+ tr.log.Error().Err(err).Str("file",
event.Name).Msg("Failed to reload TLS certificate")
+ } else {
+ tr.log.Info().Str("file",
event.Name).Msg("Successfully updated TLS certificate")
+ }
+ }
+ case err, ok := <-tr.watcher.Errors:
+ if !ok {
+ tr.log.Warn().Msg("Watcher errors channel
closed unexpectedly")
+ return
+ }
+ tr.log.Error().Err(err).Msg("Error in file watcher")
+ case <-tr.stopCh:
+ tr.log.Info().Msg("Stopping TLS file watcher for gRPC")
+ return
+ }
+ }
+}
+
+// reloadCertificate reloads the TLS certificate from the certificate and key
files.
+func (tr *TLSReloader) reloadCertificate() error {
+ newCert, err := tls.LoadX509KeyPair(tr.certFile, tr.keyFile)
+ if err != nil {
+ return errors.Wrap(err, "failed to reload TLS certificate")
+ }
+ tr.mu.Lock()
+ defer tr.mu.Unlock()
+ tr.cert = &newCert
+ return nil
+}
+
+// GetCertificate returns the current TLS certificate.
+func (tr *TLSReloader) GetCertificate() *tls.Certificate {
+ tr.mu.RLock()
+ defer tr.mu.RUnlock()
+ return tr.cert
+}
+
+// Stop gracefully stops the TLS reloader.
+func (tr *TLSReloader) Stop() {
+ close(tr.stopCh)
+ if err := tr.watcher.Close(); err != nil {
+ tr.log.Error().Err(err).Msg("Failed to close fsnotify watcher")
+ }
+}
+
+// DynamicTLSCredentials implements credentials.TransportCredentials for
dynamic reloading.
+type DynamicTLSCredentials struct {
Review Comment:
I prefer using `tls.NewTLS` to initialize "TransportCredentials" instead of
a custom one. You can dynamically load certificates using "GetCertificate" in
`tls.Config`.
##########
banyand/liaison/grpc/server.go:
##########
@@ -323,6 +341,186 @@ func (s *server) GracefulStop() {
}
}
+// TLSReloader manages dynamic reloading of TLS certificates and keys for the
gRPC server.
+type TLSReloader struct {
+ watcher *fsnotify.Watcher // 8 bytes
+ cert *tls.Certificate // 8 bytes
+ log *logger.Logger // 8 bytes
+ stopCh chan struct{} // 8 bytes
+ certFile string // 16 bytes
+ keyFile string // 16 bytes
+ mu sync.RWMutex // 24 bytes
+}
+
+// NewTLSReloader creates a new TLSReloader instance for the gRPC server.
+func NewTLSReloader(certFile, keyFile string, log *logger.Logger)
(*TLSReloader, error) {
+ if certFile == "" || keyFile == "" {
+ return nil, errors.New("certFile and keyFile must be provided")
+ }
+ if log == nil {
+ return nil, errors.New("logger must not be nil")
+ }
+
+ watcher, err := fsnotify.NewWatcher()
+ if err != nil {
+ return nil, errors.Wrap(err, "failed to create fsnotify
watcher")
+ }
+
+ cert, err := tls.LoadX509KeyPair(certFile, keyFile)
+ if err != nil {
+ watcher.Close()
+ return nil, errors.Wrap(err, "failed to load initial TLS
certificate")
+ }
+
+ log.Info().Str("certFile", certFile).Str("keyFile",
keyFile).Msg("Successfully loaded initial TLS certificates for gRPC")
+
+ tr := &TLSReloader{
+ certFile: certFile,
+ keyFile: keyFile,
+ cert: &cert,
+ log: log,
+ watcher: watcher,
+ stopCh: make(chan struct{}),
+ }
+
+ return tr, nil
+}
+
+// Start begins monitoring the TLS certificate and key files for changes.
+func (tr *TLSReloader) Start() error {
+ tr.log.Info().Str("certFile", tr.certFile).Str("keyFile",
tr.keyFile).Msg("Starting TLS file monitoring for gRPC")
+ err := tr.watcher.Add(tr.certFile)
+ if err != nil {
+ return errors.Wrapf(err, "failed to watch cert file: %s",
tr.certFile)
+ }
+ err = tr.watcher.Add(tr.keyFile)
+ if err != nil {
+ return errors.Wrapf(err, "failed to watch key file: %s",
tr.keyFile)
+ }
+ go tr.watchFiles()
+ return nil
+}
+
+// watchFiles monitors the certificate and key files for changes and reloads
credentials.
+func (tr *TLSReloader) watchFiles() {
+ tr.log.Info().Msg("TLS file watcher loop started for gRPC")
+ for {
+ select {
+ case event, ok := <-tr.watcher.Events:
+ if !ok {
+ tr.log.Warn().Msg("Watcher events channel
closed unexpectedly")
+ return
+ }
+ if event.Op&fsnotify.Remove == fsnotify.Remove {
Review Comment:
Why do you watch the "remove" operation?
##########
banyand/liaison/grpc/server.go:
##########
@@ -323,6 +341,186 @@ func (s *server) GracefulStop() {
}
}
+// TLSReloader manages dynamic reloading of TLS certificates and keys for the
gRPC server.
+type TLSReloader struct {
+ watcher *fsnotify.Watcher // 8 bytes
+ cert *tls.Certificate // 8 bytes
+ log *logger.Logger // 8 bytes
+ stopCh chan struct{} // 8 bytes
+ certFile string // 16 bytes
+ keyFile string // 16 bytes
+ mu sync.RWMutex // 24 bytes
+}
+
+// NewTLSReloader creates a new TLSReloader instance for the gRPC server.
+func NewTLSReloader(certFile, keyFile string, log *logger.Logger)
(*TLSReloader, error) {
+ if certFile == "" || keyFile == "" {
+ return nil, errors.New("certFile and keyFile must be provided")
+ }
+ if log == nil {
+ return nil, errors.New("logger must not be nil")
+ }
+
+ watcher, err := fsnotify.NewWatcher()
+ if err != nil {
+ return nil, errors.Wrap(err, "failed to create fsnotify
watcher")
+ }
+
+ cert, err := tls.LoadX509KeyPair(certFile, keyFile)
+ if err != nil {
+ watcher.Close()
+ return nil, errors.Wrap(err, "failed to load initial TLS
certificate")
+ }
+
+ log.Info().Str("certFile", certFile).Str("keyFile",
keyFile).Msg("Successfully loaded initial TLS certificates for gRPC")
+
+ tr := &TLSReloader{
+ certFile: certFile,
+ keyFile: keyFile,
+ cert: &cert,
+ log: log,
+ watcher: watcher,
+ stopCh: make(chan struct{}),
+ }
+
+ return tr, nil
+}
+
+// Start begins monitoring the TLS certificate and key files for changes.
+func (tr *TLSReloader) Start() error {
+ tr.log.Info().Str("certFile", tr.certFile).Str("keyFile",
tr.keyFile).Msg("Starting TLS file monitoring for gRPC")
+ err := tr.watcher.Add(tr.certFile)
+ if err != nil {
+ return errors.Wrapf(err, "failed to watch cert file: %s",
tr.certFile)
+ }
+ err = tr.watcher.Add(tr.keyFile)
+ if err != nil {
+ return errors.Wrapf(err, "failed to watch key file: %s",
tr.keyFile)
+ }
+ go tr.watchFiles()
+ return nil
+}
+
+// watchFiles monitors the certificate and key files for changes and reloads
credentials.
+func (tr *TLSReloader) watchFiles() {
+ tr.log.Info().Msg("TLS file watcher loop started for gRPC")
+ for {
+ select {
+ case event, ok := <-tr.watcher.Events:
+ if !ok {
+ tr.log.Warn().Msg("Watcher events channel
closed unexpectedly")
+ return
+ }
+ if event.Op&fsnotify.Remove == fsnotify.Remove {
+ if event.Name == tr.certFile {
+ if err := tr.watcher.Add(tr.certFile);
err != nil {
+
tr.log.Error().Err(err).Str("file", tr.certFile).Msg("Failed to re-add cert
file to watcher")
+ }
+ } else if event.Name == tr.keyFile {
+ if err := tr.watcher.Add(tr.keyFile);
err != nil {
+
tr.log.Error().Err(err).Str("file", tr.keyFile).Msg("Failed to re-add key file
to watcher")
+ }
+ }
+ }
+ if event.Op&fsnotify.Write == fsnotify.Write ||
+ event.Op&fsnotify.Rename == fsnotify.Rename ||
+ event.Op&fsnotify.Create == fsnotify.Create {
+ tr.log.Info().Str("file",
event.Name).Msg("Detected uploaded/changed certificate")
+ if err := tr.reloadCertificate(); err != nil {
+ tr.log.Error().Err(err).Str("file",
event.Name).Msg("Failed to reload TLS certificate")
+ } else {
+ tr.log.Info().Str("file",
event.Name).Msg("Successfully updated TLS certificate")
+ }
+ }
+ case err, ok := <-tr.watcher.Errors:
+ if !ok {
+ tr.log.Warn().Msg("Watcher errors channel
closed unexpectedly")
+ return
+ }
+ tr.log.Error().Err(err).Msg("Error in file watcher")
+ case <-tr.stopCh:
+ tr.log.Info().Msg("Stopping TLS file watcher for gRPC")
+ return
+ }
+ }
+}
+
+// reloadCertificate reloads the TLS certificate from the certificate and key
files.
+func (tr *TLSReloader) reloadCertificate() error {
+ newCert, err := tls.LoadX509KeyPair(tr.certFile, tr.keyFile)
+ if err != nil {
+ return errors.Wrap(err, "failed to reload TLS certificate")
+ }
+ tr.mu.Lock()
+ defer tr.mu.Unlock()
+ tr.cert = &newCert
+ return nil
+}
+
+// GetCertificate returns the current TLS certificate.
+func (tr *TLSReloader) GetCertificate() *tls.Certificate {
+ tr.mu.RLock()
+ defer tr.mu.RUnlock()
+ return tr.cert
+}
+
+// Stop gracefully stops the TLS reloader.
+func (tr *TLSReloader) Stop() {
+ close(tr.stopCh)
+ if err := tr.watcher.Close(); err != nil {
+ tr.log.Error().Err(err).Msg("Failed to close fsnotify watcher")
+ }
+}
+
+// DynamicTLSCredentials implements credentials.TransportCredentials for
dynamic reloading.
+type DynamicTLSCredentials struct {
+ reloader *TLSReloader
+}
+
+// NewDynamicTLSCredentials creates a new DynamicTLSCredentials instance.
+func NewDynamicTLSCredentials(reloader *TLSReloader)
credentials.TransportCredentials {
+ return &DynamicTLSCredentials{reloader: reloader}
+}
+
+// ServerHandshake implements the server-side handshake for gRPC TLS.
+func (d *DynamicTLSCredentials) ServerHandshake(conn net.Conn) (net.Conn,
credentials.AuthInfo, error) {
+ tlsConfig := &tls.Config{
Review Comment:
The test fails because "h2" is not appended to "NextProtos."
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
