avoid serializing decryptKey() every 15mins (#16135)

if the certs are the same in an environment where the 
cert files are symlinks (e.g Kubernetes), then we resort
to reloading certs every 15mins - we can avoid reload
of the kes client instance. Ensure that the price to pay 
for contending with the lock must happen when necessary.

internal/kms/kes.go

@@ -18,6 +18,7 @@
 package kms
 
 import (
+	"bytes"
 	"context"
 	"crypto/tls"
 	"crypto/x509"
@@ -95,19 +96,35 @@ func NewWithConfig(config Config) (KMS, error) {
 	}
 	go func() {
 		for {
+			var prevCertificate tls.Certificate
 			select {
-			case certificate := <-config.ReloadCertEvents:
+			case certificate, ok := <-config.ReloadCertEvents:
-				client := kes.NewClientWithConfig("", &tls.Config{
+				if !ok {
-					MinVersion:         tls.VersionTLS12,
+					return
-					Certificates:       []tls.Certificate{certificate},
+				}
-					RootCAs:            config.RootCAs,
+				sameCert := true
-					ClientSessionCache: tls.NewLRUClientSessionCache(tlsClientSessionCacheSize),
+				for i, b := range certificate.Certificate {
-				})
+					if !bytes.Equal(b, prevCertificate.Certificate[i]) {
-				client.Endpoints = endpoints
+						sameCert = false
+						break
+					}
+				}
+				// Do not reload if its the same cert as before.
+				if !sameCert {
+					client := kes.NewClientWithConfig("", &tls.Config{
+						MinVersion:         tls.VersionTLS12,
+						Certificates:       []tls.Certificate{certificate},
+						RootCAs:            config.RootCAs,
+						ClientSessionCache: tls.NewLRUClientSessionCache(tlsClientSessionCacheSize),
+					})
+					client.Endpoints = endpoints
 
-				c.lock.Lock()
+					c.lock.Lock()
-				c.client = client
+					c.client = client
-				c.lock.Unlock()
+					c.lock.Unlock()
+
+					prevCertificate = certificate
+				}
 			}
 		}
 	}()