Reconnect broken LDAP connections (#10183)

* Reconnect broken LDAP connections

Ensure that our LDAP client returns trace.ConnectionProblem errors
for LDAP error 200 (network error).

When this condition is detected, we re-initialize our LDAP client
to get a fresh connection.

Also clean up some of the errors in the reconciler so that they
include the resources kind and name without stringify-ing the
entire resource.

Lastly, add both a connect and request timeout for our LDAP client,
allowing us to detect connection issues sooner.

Fixes #9808

* Reschedule pending renewals after succesful LDAP cert renewals

Prior to this change, if we encountered an LDAP network error and
re-initialized the LDAP client, we would schedule a new renewal
but also leave the existing renewal in place. Since each succesful
renewal schedules a subsequent renewal, we would be renewing twice
as often as necessary after recovering from this network error.
Each subsequent network error that we recover from would create
yet another set of periodic renewals. In the worst case we could
ultimately exhaust system resources by doing nothing but handling
LDAP cert renewals.

Fix this by using a single *time.Timer that is used to handle the
next LDAP cert renewal. When it comes time to schedule a renewal,
we use timer.Reset() to bump the existing renewal out rather than
schedule an additional one.

This introduced the need for a mutex, so I got rid of the atomic
load/store for ldapInitialized and put that behind the same mutex.

* Add retries to LDAP client initialization

Prior to this change, we had retry logic on the initial setup and
on receiving a network error during a query. We did not have retry
logic in place if the re-initialization failed. This would put the
system in a state where desktop discovery stops running until the
cert approaches expiration (up to a maximum of 5-6 hours)

Now, we schedule the next re-initialization to be either:
- 10 minutes later, in the case of an error
- After 2/3 of the cert's validity has elapsed (5-6 hours),
  for the happy path

lib/services/reconciler.go

@@ -129,9 +129,9 @@ func (r *Reconciler) processRegisteredResource(ctx context.Context, newResources
 		return nil
 	}
 
-	r.log.Infof("%v removed, deleting.", registered)
+	r.log.Infof("%v %v removed, deleting.", registered.GetKind(), registered.GetName())
 	if err := r.cfg.OnDelete(ctx, registered); err != nil {
-		return trace.Wrap(err, "failed to delete %v", registered)
+		return trace.Wrap(err, "failed to delete  %v %v", registered.GetKind(), registered.GetName())
 	}
 
 	return nil
@@ -145,19 +145,19 @@ func (r *Reconciler) processNewResource(ctx context.Context, currentResources ty
 	registered := currentResources.Find(new.GetName())
 	if registered == nil {
 		if r.cfg.Matcher(new) {
-			r.log.Infof("%v %v matches, creating.", new.GetKind(), new.GetMetadata().Name)
+			r.log.Infof("%v %v matches, creating.", new.GetKind(), new.GetName())
 			if err := r.cfg.OnCreate(ctx, new); err != nil {
-				return trace.Wrap(err, "failed to create %v", new)
+				return trace.Wrap(err, "failed to create %v %v", new.GetKind(), new.GetName())
 			}
 			return nil
 		}
-		r.log.Debugf("%v %v doesn't match, not creating.", new.GetKind(), new.GetMetadata().Name)
+		r.log.Debugf("%v %v doesn't match, not creating.", new.GetKind(), new.GetName())
 		return nil
 	}
 
 	// Don't overwrite resource of a different origin.
 	if registered.Origin() != new.Origin() {
-		r.log.Debugf("%v has different origin (%v vs %v), not updating.", new.GetMetadata().Name,
+		r.log.Debugf("%v has different origin (%v vs %v), not updating.", new.GetName(),
 			new.Origin(), registered.Origin())
 		return nil
 	}
@@ -166,19 +166,19 @@ func (r *Reconciler) processNewResource(ctx context.Context, currentResources ty
 	// labels still match.
 	if CompareResources(new, registered) != Equal {
 		if r.cfg.Matcher(new) {
-			r.log.Infof("%v %v updated, updating.", new.GetKind(), new.GetMetadata().Name)
+			r.log.Infof("%v %v updated, updating.", new.GetKind(), new.GetName())
 			if err := r.cfg.OnUpdate(ctx, new); err != nil {
-				return trace.Wrap(err, "failed to update %v", new)
+				return trace.Wrap(err, "failed to update %v %v", new.GetKind(), new.GetName())
 			}
 			return nil
 		}
-		r.log.Infof("%v %v updated and no longer matches, deleting.", new.GetKind(), new.GetMetadata().Name)
+		r.log.Infof("%v %v updated and no longer matches, deleting.", new.GetKind(), new.GetName())
 		if err := r.cfg.OnDelete(ctx, registered); err != nil {
-			return trace.Wrap(err, "failed to delete %v", new)
+			return trace.Wrap(err, "failed to delete %v %v", new.GetKind(), new.GetName())
 		}
 		return nil
 	}
 
-	r.log.Debugf("%v %v is already registered.", new.GetKind(), new.GetMetadata().Name)
+	r.log.Debugf("%v %v is already registered.", new.GetKind(), new.GetName())
 	return nil
 }

lib/srv/desktop/discovery.go

@@ -18,7 +18,6 @@ import (
 	"context"
 	"fmt"
 	"strings"
-	"sync/atomic"
 	"time"
 
 	"github.com/go-ldap/ldap/v3"
@@ -67,7 +66,7 @@ func (s *WindowsService) startDesktopDiscovery() error {
 		GetCurrentResources: func() types.ResourcesWithLabels { return s.lastDiscoveryResults },
 		GetNewResources:     s.getDesktopsFromLDAP,
 		OnCreate:            s.upsertDesktop,
-		OnUpdate:            s.updateDesktop,
+		OnUpdate:            s.upsertDesktop,
 		OnDelete:            s.deleteDesktop,
 		Log:                 s.cfg.Log,
 	})
@@ -113,7 +112,7 @@ func (s *WindowsService) ldapSearchFilter() string {
 
 // getDesktopsFromLDAP discovers Windows hosts via LDAP
 func (s *WindowsService) getDesktopsFromLDAP() types.ResourcesWithLabels {
-	if atomic.LoadInt32(&s.ldapInitialized) == 0 {
+	if !s.ldapReady() {
 		s.cfg.Log.Warn("skipping desktop discovery: LDAP not yet initialized")
 		return nil
 	}
@@ -122,7 +121,17 @@ func (s *WindowsService) getDesktopsFromLDAP() types.ResourcesWithLabels {
 	s.cfg.Log.Debugf("searching for desktops with LDAP filter %v", filter)
 
 	entries, err := s.lc.readWithFilter(s.cfg.DiscoveryBaseDN, filter, computerAttribtes)
-	if err != nil {
+	if trace.IsConnectionProblem(err) {
+		// If the connection was broken, re-initialize the LDAP client so that it's
+		// ready for the next reconcile loop. Return the last known set of desktops
+		// in this case, so that the reconciler doesn't delete the desktops it already
+		// knows about.
+		s.cfg.Log.Info("LDAP connection error when searching for desktops, reinitializing client")
+		if err := s.initializeLDAP(); err != nil {
+			s.cfg.Log.Errorf("failed to reinitialize LDAP client, will retry on next reconcile: %v", err)
+		}
+		return s.lastDiscoveryResults
+	} else if err != nil {
 		s.cfg.Log.Warnf("could not discover Windows Desktops: %v", err)
 		return nil
 	}
@@ -148,19 +157,11 @@ func (s *WindowsService) getDesktopsFromLDAP() types.ResourcesWithLabels {
 func (s *WindowsService) upsertDesktop(ctx context.Context, r types.ResourceWithLabels) error {
 	d, ok := r.(types.WindowsDesktop)
 	if !ok {
-		return trace.Errorf("create: expected a WindowsDesktop, got %T", r)
+		return trace.Errorf("upsert: expected a WindowsDesktop, got %T", r)
 	}
 	return s.cfg.AuthClient.UpsertWindowsDesktop(ctx, d)
 }
 
-func (s *WindowsService) updateDesktop(ctx context.Context, r types.ResourceWithLabels) error {
-	d, ok := r.(types.WindowsDesktop)
-	if !ok {
-		return trace.Errorf("update: expected a WindowsDesktop, got %T", r)
-	}
-	return s.cfg.AccessPoint.UpdateWindowsDesktop(ctx, d)
-}
-
 func (s *WindowsService) deleteDesktop(ctx context.Context, r types.ResourceWithLabels) error {
 	return s.cfg.AuthClient.DeleteWindowsDesktop(ctx, r.GetName())
 }

lib/srv/desktop/ldap.go

@@ -67,11 +67,12 @@ func (c *ldapClient) readWithFilter(dn string, filter string, attrs []string) ([
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	res, err := c.client.Search(req)
-	if err != nil {
+	if ldap.IsErrorWithCode(err, ldap.ErrorNetwork) {
+		return nil, trace.ConnectionProblem(err, "fetching LDAP object %q", dn)
+	} else if err != nil {
 		return nil, trace.Wrap(err, "fetching LDAP object %q: %v", dn, err)
 	}
 	return res.Entries, nil
-
 }
 
 // read fetches an LDAP entry at path and its children, if any. Only
@@ -113,6 +114,8 @@ func (c *ldapClient) create(dn string, class string, attrs map[string][]string)
 				return trace.BadParameter("object constraint violation on %q: %v", dn, err)
 			case ldap.LDAPResultInsufficientAccessRights:
 				return trace.AccessDenied("insufficient permissions to create %q: %v", dn, err)
+			case ldap.ErrorNetwork:
+				return trace.ConnectionProblem(err, "network error creating %q", dn)
 			}
 		}
 		return trace.Wrap(err, "error creating LDAP object %q: %v", dn, err)
@@ -127,6 +130,8 @@ func (c *ldapClient) createContainer(dn string) error {
 	// Ignore the error if container already exists.
 	if trace.IsAlreadyExists(err) {
 		return nil
+	} else if ldap.IsErrorWithCode(err, ldap.ErrorNetwork) {
+		return trace.ConnectionProblem(err, "creating %v", dn)
 	}
 	return trace.Wrap(err)
 }
@@ -148,7 +153,9 @@ func (c *ldapClient) update(dn string, replaceAttrs map[string][]string) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
-	if err := c.client.Modify(req); err != nil {
+	if err := c.client.Modify(req); ldap.IsErrorWithCode(err, ldap.ErrorNetwork) {
+		return trace.ConnectionProblem(err, "updating %q", dn)
+	} else if err != nil {
 		return trace.Wrap(err, "updating %q: %v", dn, err)
 	}
 	return nil

lib/srv/desktop/windows_server.go

@@ -29,7 +29,7 @@ import (
 	"fmt"
 	"net"
 	"strings"
-	"sync/atomic"
+	"sync"
 	"time"
 
 	"github.com/go-ldap/ldap/v3"
@@ -60,6 +60,15 @@ const (
 	// when resolving Windows Desktop hostnames
 	dnsDialTimeout = 5 * time.Second
 
+	// ldapDialTimeout is the timeout for dialing the LDAP server
+	// when making an initial connection
+	ldapDialTimeout = 5 * time.Second
+
+	// ldapRequestTimeout is the timeout for making LDAP requests.
+	// It is larger than the dial timeout because LDAP queries in large
+	// Active Directory environments may take longer to complete.
+	ldapRequestTimeout = 20 * time.Second
+
 	// windowsDesktopCertTTL is the TTL for Teleport-issued Windows Certificates.
 	// Certificates are requested on each connection attempt, so the TTL is
 	// deliberately set to a small value to give enough time to establish a
@@ -72,6 +81,10 @@ const (
 	// not used for interactive login and is only used when issuing certs for
 	// a restrictive service account.
 	windowsDesktopServiceCertTTL = 8 * time.Hour
+
+	// windowsDesktopServiceCertRetryInterval indicates how often to retry
+	// issuing an LDAP certificate if the operation fails.
+	windowsDesktopServiceCertRetryInterval = 10 * time.Minute
 )
 
 // WindowsService implements the RDP-based Windows desktop access service.
@@ -83,11 +96,16 @@ type WindowsService struct {
 	cfg        WindowsServiceConfig
 	middleware *auth.Middleware
 
-	ldapInitialized int32
-	lc              *ldapClient
+	lc *ldapClient
+
+	mu              sync.Mutex // mu protects the fields that follow
+	ldapInitialized bool
+	ldapCertRenew   *time.Timer
 
 	// lastDisoveryResults stores the results of the most recent LDAP search
-	// when desktop discovery is enabled
+	// when desktop discovery is enabled.
+	// no synchronization is necessary because this is only read/written from
+	// the reconciler goroutine.
 	lastDiscoveryResults types.ResourcesWithLabels
 
 	// Windows hosts discovered via LDAP likely won't resolve with the
@@ -323,23 +341,12 @@ func NewWindowsService(cfg WindowsServiceConfig) (*WindowsService, error) {
 		close:       close,
 	}
 
-	// run LDAP initialization in a retry loop - this prevents the service
-	// from crashing and taking the entire Teleport process with it if
-	// we cannot connect to LDAP
-	l, err := utils.NewLinear(utils.LinearConfig{
-		First: 5 * time.Second,
-		Step:  30 * time.Second,
-		Max:   30 * time.Minute,
-		Clock: s.cfg.Clock,
-	})
-	if err != nil {
-		return nil, trace.Wrap(err)
+	// initialize LDAP - if this fails it will automatically schedule a retry.
+	// we don't want to return an error in this case, because failure to start
+	// the service brings down the entire Teleport process
+	if err := s.initializeLDAP(); err != nil {
+		s.cfg.Log.WithError(err).Error("initializing LDAP client, will retry")
 	}
-	go func() {
-		if retryErr := l.For(s.closeCtx, s.initializeLDAP); retryErr != nil {
-			s.cfg.Log.WithError(retryErr).Error("attempting to initialize LDAP client")
-		}
-	}()
 
 	// TODO(zmb3): session recording.
 
@@ -408,22 +415,38 @@ func (s *WindowsService) tlsConfigForLDAP() (*tls.Config, error) {
 	return tc, nil
 }
 
+// initializeLDAP requests a TLS certificate from the auth server to be used for
+// authenticating with the LDAP server. If the certificate is obtained, and
+// authentication with the LDAP server succeeds, it schedules a renewal to take
+// place before the certificate expires. If we are unable to obtain a certificate
+// and authenticate with the LDAP server, then the operation will be automatically
+// retried.
 func (s *WindowsService) initializeLDAP() error {
 	tc, err := s.tlsConfigForLDAP()
 	if trace.IsAccessDenied(err) && modules.GetModules().BuildType() == modules.BuildEnterprise {
 		s.cfg.Log.Warn("Could not generate certificate for LDAPS. Ensure that the auth server is licensed for desktop access.")
 	}
 	if err != nil {
-		atomic.StoreInt32(&s.ldapInitialized, 0)
+		s.mu.Lock()
+		s.ldapInitialized = false
+		// in the case where we're not licensed for desktop access, we retry less frequently,
+		// since this is likely not an intermittent error that will resolve itself quickly
+		s.scheduleNextLDAPCertRenewalLocked(windowsDesktopServiceCertRetryInterval * 3)
+		s.mu.Unlock()
 		return trace.Wrap(err)
 	}
 
-	conn, err := ldap.DialURL("ldaps://"+s.cfg.Addr, ldap.DialWithTLSConfig(tc))
+	conn, err := ldap.DialURL("ldaps://"+s.cfg.Addr,
+		ldap.DialWithTLSDialer(tc, &net.Dialer{Timeout: ldapDialTimeout}))
 	if err != nil {
-		atomic.StoreInt32(&s.ldapInitialized, 0)
+		s.mu.Lock()
+		s.ldapInitialized = false
+		s.scheduleNextLDAPCertRenewalLocked(windowsDesktopServiceCertRetryInterval)
+		s.mu.Unlock()
 		return trace.Wrap(err, "dial")
 	}
 
+	conn.SetTimeout(ldapRequestTimeout)
 	s.lc.setClient(conn)
 
 	// Note: admin still needs to import our CA into the Group Policy following
@@ -441,22 +464,29 @@ func (s *WindowsService) initializeLDAP() error {
 		return trace.Wrap(err)
 	}
 
-	atomic.StoreInt32(&s.ldapInitialized, 1)
+	s.mu.Lock()
+	s.ldapInitialized = true
+	s.scheduleNextLDAPCertRenewalLocked(windowsDesktopServiceCertTTL / 3)
+	s.mu.Unlock()
 
-	// if we were successful in initializing the client, schedule a renewal
-	// so that we get a new cert prior to expiration
-	go func() {
-		select {
-		case <-time.After(windowsDesktopServiceCertTTL / 3):
+	return nil
+}
+
+// scheduleNextLDAPCertRenewalLocked schedules a renewal of our LDAP credentials
+// after some amount of time has elapsed. If an existing renewal is already
+// scheduled, it is canceled and this new one takes its place.
+//
+// The lock on s.mu MUST be held.
+func (s *WindowsService) scheduleNextLDAPCertRenewalLocked(after time.Duration) {
+	if s.ldapCertRenew != nil {
+		s.ldapCertRenew.Reset(after)
+	} else {
+		s.ldapCertRenew = time.AfterFunc(after, func() {
 			if err := s.initializeLDAP(); err != nil {
 				s.cfg.Log.WithError(err).Error("couldn't renew certificate for LDAP auth")
 			}
-		case <-s.closeCtx.Done():
-			return
-		}
-	}()
-
-	return nil
+		})
+	}
 }
 
 func (s *WindowsService) startServiceHeartbeat() error {
@@ -519,6 +549,12 @@ func (s *WindowsService) startStaticHostHeartbeats() error {
 // Close instructs the server to stop accepting new connections and abort all
 // established ones. Close does not wait for the connections to be finished.
 func (s *WindowsService) Close() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.ldapCertRenew != nil {
+		s.ldapCertRenew.Stop()
+	}
 	s.close()
 	s.lc.close()
 	return nil
@@ -553,6 +589,12 @@ func (s *WindowsService) Serve(plainLis net.Listener) error {
 	}
 }
 
+func (s *WindowsService) ldapReady() bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.ldapInitialized
+}
+
 // handleConnection handles TLS connections from a Teleport proxy.
 // It authenticates and authorizes the connection, and then begins
 // translating the TDP messages from the proxy into native RDP.
@@ -564,13 +606,13 @@ func (s *WindowsService) handleConnection(proxyConn *tls.Conn) {
 	// Inline function to enforce that we are centralizing TDP Error sending in this function.
 	sendTdpError := func(message string) {
 		if err := tdpConn.OutputMessage(tdp.Error{Message: message}); err != nil {
-			s.cfg.Log.Errorf("Failed to send TDP error message %v: %v", tdp.Error{Message: message}, err)
+			log.Errorf("Failed to send TDP error message %v: %v", tdp.Error{Message: message}, err)
 		}
 	}
 
 	// don't handle connections until the LDAP initialization retry loop has succeeded
 	// (it would fail anyway, but this presents a better error to the user)
-	if atomic.LoadInt32(&s.ldapInitialized) != 1 {
+	if !s.ldapReady() {
 		// TODO(zmb3): send TDP error message
 		log.Error("This service cannot accept connections until LDAP initialization has completed.")
 		return