Improve Teleport reload behavior (#37773)

* clean up pidfile and reloads

* reduce log spam during graceful shutdowns

* hold an exclusive advisory lock on the pidfile

* do systemd reloads with the pidfile rather than $MAINPID

* prioritize signals over internal restarts

assets/aws/files/system/teleport-acm.service

@@ -11,6 +11,6 @@ Restart=always
 RestartSec=5
 RuntimeDirectory=teleport
 ExecStart=/usr/local/bin/teleport start --config=/etc/teleport.yaml --diag-addr=127.0.0.1:3000 --pid-file=/run/teleport/teleport.pid
-ExecReload=/bin/kill -HUP $MAINPID
+ExecReload=pkill -HUP -L -F /run/teleport/teleport.pid
 PIDFile=/run/teleport/teleport.pid
-LimitNOFILE=524288
+LimitNOFILE=524288

assets/aws/files/system/teleport-auth.service

@@ -11,9 +11,9 @@ Restart=always
 RestartSec=5
 RuntimeDirectory=teleport
 ExecStart=/usr/local/bin/teleport start --config=/etc/teleport.yaml --diag-addr=127.0.0.1:3000 --pid-file=/run/teleport/teleport.pid
-ExecReload=/bin/kill -HUP $MAINPID
+ExecReload=pkill -HUP -L -F /run/teleport/teleport.pid
 PIDFile=/run/teleport/teleport.pid
 LimitNOFILE=524288
 
 [Install]
-WantedBy=multi-user.target
+WantedBy=multi-user.target

assets/aws/files/system/teleport-node.service

@@ -12,9 +12,9 @@ RestartSec=5
 RuntimeDirectory=teleport
 ExecStartPre=/usr/local/bin/teleport-ssm-get-token
 ExecStart=/usr/local/bin/teleport start --config=/etc/teleport.yaml --diag-addr=127.0.0.1:3000 --pid-file=/run/teleport/teleport.pid
-ExecReload=/bin/kill -HUP $MAINPID
+ExecReload=pkill -HUP -L -F /run/teleport/teleport.pid
 PIDFile=/run/teleport/teleport.pid
 LimitNOFILE=524288
 
 [Install]
-WantedBy=multi-user.target
+WantedBy=multi-user.target

assets/aws/files/system/teleport-proxy-acm.service

@@ -13,9 +13,9 @@ RuntimeDirectory=teleport
 EnvironmentFile=/etc/teleport.d/conf
 ExecStartPre=/usr/local/bin/teleport-ssm-get-token
 ExecStart=/usr/local/bin/teleport start --config=/etc/teleport.yaml --diag-addr=127.0.0.1:3000 --pid-file=/run/teleport/teleport.pid
-ExecReload=/bin/kill -HUP $MAINPID
+ExecReload=pkill -HUP -L -F /run/teleport/teleport.pid
 PIDFile=/run/teleport/teleport.pid
 LimitNOFILE=524288
 
 [Install]
-WantedBy=multi-user.target
+WantedBy=multi-user.target

assets/aws/files/system/teleport-proxy.service

@@ -14,9 +14,9 @@ EnvironmentFile=/etc/teleport.d/conf
 ExecStartPre=/usr/local/bin/teleport-ssm-get-token
 ExecStartPre=/bin/aws s3 sync s3://${TELEPORT_S3_BUCKET}/live/${TELEPORT_DOMAIN_NAME} /var/lib/teleport
 ExecStart=/usr/local/bin/teleport start --config=/etc/teleport.yaml --diag-addr=127.0.0.1:3000 --pid-file=/run/teleport/teleport.pid
-ExecReload=/bin/kill -HUP $MAINPID
+ExecReload=pkill -HUP -L -F /run/teleport/teleport.pid
 PIDFile=/run/teleport/teleport.pid
 LimitNOFILE=524288
 
 [Install]
-WantedBy=multi-user.target
+WantedBy=multi-user.target

assets/aws/files/system/teleport.service

@@ -12,9 +12,9 @@ RestartSec=5
 RuntimeDirectory=teleport
 ExecStartPre=/usr/local/bin/teleport-all-pre-start
 ExecStart=/usr/local/bin/teleport start --config=/etc/teleport.yaml --diag-addr=127.0.0.1:3000 --pid-file=/run/teleport/teleport.pid
-ExecReload=/bin/kill -HUP $MAINPID
+ExecReload=pkill -HUP -L -F /run/teleport/teleport.pid
 PIDFile=/run/teleport/teleport.pid
 LimitNOFILE=524288
 
 [Install]
-WantedBy=multi-user.target
+WantedBy=multi-user.target

examples/systemd/fips/teleport.service

@@ -7,7 +7,7 @@ Type=simple
 Restart=on-failure
 EnvironmentFile=-/etc/default/teleport
 ExecStart=/usr/local/bin/teleport start --config /etc/teleport.yaml --fips --pid-file=/run/teleport.pid
-ExecReload=/bin/kill -HUP $MAINPID
+ExecReload=pkill -HUP -L -F /run/teleport.pid
 PIDFile=/run/teleport.pid
 LimitNOFILE=524288
 

examples/systemd/production/auth/teleport.service

@@ -11,9 +11,9 @@ Restart=on-failure
 # --roles='proxy,auth,node' is the default value
 # if none is set
 ExecStart=/usr/local/bin/teleport start --roles=auth --config=/etc/teleport.yaml --diag-addr=127.0.0.1:3000 --pid-file=/run/teleport.pid
-ExecReload=/bin/kill -HUP $MAINPID
+ExecReload=pkill -HUP -L -F /run/teleport.pid
 PIDFile=/run/teleport.pid
 LimitNOFILE=524288
 
 [Install]
-WantedBy=multi-user.target
+WantedBy=multi-user.target

examples/systemd/production/node/teleport.service

@@ -11,7 +11,7 @@ Restart=on-failure
 # --roles='proxy,auth,node' is the default value
 # if none is set
 ExecStart=/usr/local/bin/teleport start --roles=node --config=/etc/teleport.yaml --pid-file=/run/teleport.pid
-ExecReload=/bin/kill -HUP $MAINPID
+ExecReload=pkill -HUP -L -F /run/teleport.pid
 PIDFile=/run/teleport.pid
 LimitNOFILE=524288
 

examples/systemd/production/proxy/teleport.service

@@ -11,9 +11,9 @@ Restart=on-failure
 # --roles='proxy,auth,node' is the default value
 # if none is set
 ExecStart=/usr/local/bin/teleport start --roles=proxy --config=/etc/teleport.yaml --pid-file=/run/teleport.pid
-ExecReload=/bin/kill -HUP $MAINPID
+ExecReload=pkill -HUP -L -F /run/teleport.pid
 PIDFile=/run/teleport.pid
 LimitNOFILE=524288
 
 [Install]
-WantedBy=multi-user.target
+WantedBy=multi-user.target

examples/systemd/teleport.service

@@ -7,7 +7,7 @@ Type=simple
 Restart=on-failure
 EnvironmentFile=-/etc/default/teleport
 ExecStart=/usr/local/bin/teleport start --config /etc/teleport.yaml --pid-file=/run/teleport.pid
-ExecReload=/bin/kill -HUP $MAINPID
+ExecReload=pkill -HUP -L -F /run/teleport.pid
 PIDFile=/run/teleport.pid
 LimitNOFILE=524288
 

go.mod

@@ -100,6 +100,7 @@ require (
 	github.com/google/go-containerregistry v0.19.0
 	github.com/google/go-querystring v1.1.0
 	github.com/google/go-tpm-tools v0.4.2
+	github.com/google/renameio/v2 v2.0.0
 	github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510
 	github.com/google/uuid v1.6.0
 	github.com/googleapis/gax-go/v2 v2.12.0
@@ -343,7 +344,6 @@ require (
 	github.com/google/go-tpm v0.9.0 // indirect
 	github.com/google/go-tspi v0.3.0 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
-	github.com/google/renameio/v2 v2.0.0 // indirect
 	github.com/google/s2a-go v0.1.7 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect
 	github.com/gorilla/handlers v1.5.2 // indirect

lib/config/systemd.go

@@ -47,12 +47,13 @@ Type=simple
 Restart=on-failure
 EnvironmentFile=-{{ .EnvironmentFile }}
 ExecStart={{ .TeleportInstallationFile }} start --config {{ .TeleportConfigPath }} --pid-file={{ .PIDFile }}
-ExecReload=/bin/kill -HUP $MAINPID
+ExecReload=pkill -HUP -L -F "{{ .PIDFile }}"
 PIDFile={{ .PIDFile }}
 LimitNOFILE={{ .FileDescriptorLimit }}
 
 [Install]
-WantedBy=multi-user.target`))
+WantedBy=multi-user.target
+`))
 
 // SystemdFlags specifies configuration parameters for a systemd unit file.
 type SystemdFlags struct {

lib/config/testdata/TestWriteSystemdUnitFile.golden

@@ -7,9 +7,9 @@ Type=simple
 Restart=on-failure
 EnvironmentFile=-/custom/env/dir/teleport
 ExecStart=/custom/install/dir/teleport start --config /etc/teleport.yaml --pid-file=/custom/pid/dir/teleport.pid
-ExecReload=/bin/kill -HUP $MAINPID
+ExecReload=pkill -HUP -L -F "/custom/pid/dir/teleport.pid"
 PIDFile=/custom/pid/dir/teleport.pid
 LimitNOFILE=16384
 
 [Install]
-WantedBy=multi-user.target
+WantedBy=multi-user.target

lib/defaults/defaults.go

@@ -315,10 +315,6 @@ const (
 	// LowResPollingPeriod is a default low resolution polling period
 	LowResPollingPeriod = 600 * time.Second
 
-	// HighResReportingPeriod is a high resolution polling reporting
-	// period used in services
-	HighResReportingPeriod = 10 * time.Second
-
 	// SessionControlTimeout is the maximum amount of time a controlled session
 	// may persist after contact with the auth server is lost (sessctl semaphore
 	// leases are refreshed at a rate of ~1/2 this duration).

lib/service/service.go

@@ -46,6 +46,7 @@ import (
 	"time"
 
 	awscredentials "github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/google/renameio/v2"
 	"github.com/google/uuid"
 	"github.com/gravitational/roundtrip"
 	"github.com/gravitational/trace"
@@ -57,6 +58,7 @@ import (
 	"golang.org/x/crypto/acme"
 	"golang.org/x/crypto/acme/autocert"
 	"golang.org/x/crypto/ssh"
+	"golang.org/x/sys/unix"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/keepalive"
@@ -389,10 +391,9 @@ type TeleportProcess struct {
 	// closed all the listeners)
 	listenersClosed bool
 
-	// forkedPIDs is a collection of a teleport processes forked
-	// during restart used to collect their status in case if the
-	// child process crashed.
-	forkedPIDs []int
+	// forkedTeleportCount is the count of forked Teleport child processes
+	// currently active, as spawned by SIGHUP or SIGUSR2.
+	forkedTeleportCount atomic.Int32
 
 	// storage is a server local storage
 	storage *auth.ProcessStorage
@@ -738,6 +739,9 @@ func waitAndReload(ctx context.Context, cfg servicecfg.Config, srv Process, newT
 	}
 	newCfg := cfg
 	newCfg.FileDescriptors = fileDescriptors
+	// our PID hasn't changed as we reload in-process, and if we're no longer
+	// the "main" Teleport process we don't want to overwrite the PID file
+	newCfg.PIDFile = ""
 	newSrv, err := newTeleport(&newCfg)
 	if err != nil {
 		warnOnErr(srv.Close(), cfg.Log)
@@ -1252,13 +1256,8 @@ func NewTeleport(cfg *servicecfg.Config) (*TeleportProcess, error) {
 
 	// create the new pid file only after started successfully
 	if cfg.PIDFile != "" {
-		f, err := os.OpenFile(cfg.PIDFile, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o666)
-		if err != nil {
-			return nil, trace.ConvertSystemError(err)
-		}
-		_, err = fmt.Fprintf(f, "%v", os.Getpid())
-		if err = trace.NewAggregate(err, f.Close()); err != nil {
-			return nil, trace.Wrap(err)
+		if err := createLockedPIDFile(cfg.PIDFile); err != nil {
+			return nil, trace.Wrap(err, "creating pidfile")
 		}
 	}
 
@@ -5528,7 +5527,7 @@ func (process *TeleportProcess) StartShutdown(ctx context.Context) context.Conte
 	warnOnErr(process.stopListeners(), process.log)
 
 	// populate context values
-	if len(process.getForkedPIDs()) > 0 {
+	if process.forkedTeleportCount.Load() > 0 {
 		ctx = services.ProcessForkedContext(ctx)
 	}
 
@@ -6101,3 +6100,35 @@ func (process *TeleportProcess) newExternalAuditStorageConfigurator() (*external
 	statusService := local.NewStatusService(process.backend)
 	return externalauditstorage.NewConfigurator(process.ExitContext(), ecaSvc, integrationSvc, statusService)
 }
+
+// createLockedPIDFile creates a PID file in the path specified by pidFile
+// containing the current PID, atomically swapping it in the final place and
+// leaving it with an exclusive advisory lock that will get released when the
+// process ends, for the benefit of "pkill -L".
+func createLockedPIDFile(pidFile string) error {
+	pending, err := renameio.NewPendingFile(pidFile, renameio.WithPermissions(0o644))
+	if err != nil {
+		return trace.ConvertSystemError(err)
+	}
+	defer pending.Cleanup()
+	if _, err := fmt.Fprintf(pending, "%v\n", os.Getpid()); err != nil {
+		return trace.ConvertSystemError(err)
+	}
+
+	const minimumDupFD = 3 // skip stdio
+	locker, err := unix.FcntlInt(pending.Fd(), unix.F_DUPFD_CLOEXEC, minimumDupFD)
+	runtime.KeepAlive(pending)
+	if err != nil {
+		return trace.ConvertSystemError(err)
+	}
+	if err := unix.Flock(locker, unix.LOCK_EX|unix.LOCK_NB); err != nil {
+		_ = unix.Close(locker)
+		return trace.ConvertSystemError(err)
+	}
+	// deliberately leak the fd to hold the lock until the process dies
+
+	if err := pending.CloseAtomicallyReplace(); err != nil {
+		return trace.ConvertSystemError(err)
+	}
+	return nil
+}

lib/service/signals.go

@@ -22,6 +22,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net"
 	"os"
 	"os/exec"
@@ -40,13 +41,16 @@ import (
 
 // printShutdownStatus prints running services until shut down
 func (process *TeleportProcess) printShutdownStatus(ctx context.Context) {
-	t := time.NewTicker(defaults.HighResReportingPeriod)
+	statusInterval := defaults.HighResPollingPeriod
+	t := time.NewTimer(statusInterval)
 	defer t.Stop()
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		case <-t.C:
+			statusInterval = min(statusInterval*2, defaults.LowResPollingPeriod)
+			t.Reset(statusInterval)
 			process.log.Infof("Waiting for services: %v to finish.", process.Supervisor.Services())
 		}
 	}
@@ -64,7 +68,6 @@ func (process *TeleportProcess) WaitForSignals(ctx context.Context) error {
 		syscall.SIGUSR1, // log process diagnostic info
 		syscall.SIGUSR2, // initiate process restart procedure
 		syscall.SIGHUP,  // graceful restart procedure
-		syscall.SIGCHLD, // collect child status
 	)
 	defer signal.Stop(sigC)
 
@@ -124,12 +127,20 @@ func (process *TeleportProcess) WaitForSignals(ctx context.Context) error {
 				process.Shutdown(ctx)
 				process.log.Infof("All services stopped, exiting.")
 				return nil
-			case syscall.SIGCHLD:
-				process.collectStatuses()
 			default:
 				process.log.Infof("Ignoring %q.", signal)
 			}
 		case <-process.ReloadContext().Done():
+			// it's fine to signal.Stop the same channel multiple times, and
+			// after the function returns we're guaranteed to have restored the
+			// default handlers for the signals and that no more signals are
+			// pushed into the channel
+			signal.Stop(sigC)
+			if len(sigC) > 0 {
+				// exhaust all signals before the internal reload, so we don't
+				// miss signals to exit or to graceful restart instead
+				continue
+			}
 			process.log.Infof("Exiting signal handler: process has started internal reload.")
 			return ErrTeleportReloading
 		case <-process.ExitContext().Done():
@@ -160,8 +171,10 @@ func (process *TeleportProcess) WaitForSignals(ctx context.Context) error {
 	}
 }
 
-const defaultShutdownTimeout = time.Second * 3
-const maxShutdownTimeout = time.Minute * 10
+const (
+	defaultShutdownTimeout = time.Second * 3
+	maxShutdownTimeout     = time.Minute * 10
+)
 
 func getShutdownTimeout(log logrus.FieldLogger) time.Duration {
 	timeout := defaultShutdownTimeout
@@ -528,85 +541,33 @@ func (process *TeleportProcess) forkChild() error {
 	if err != nil {
 		return trace.ConvertSystemError(err)
 	}
-	process.pushForkedPID(p.Pid)
-	log.WithFields(logrus.Fields{"pid": p.Pid}).Infof("Forked new child process.")
+	log.WithField("pid", p.Pid).Infof("Forked new child process.")
+	log = process.log.WithField("pid", p.Pid)
 
-	messageReceived, cancel := context.WithCancel(context.TODO())
-	defer cancel()
+	process.forkedTeleportCount.Add(1)
 	go func() {
-		data := make([]byte, 1024)
-		len, err := readPipe.Read(data)
+		defer process.forkedTeleportCount.Add(-1)
+		state, err := p.Wait()
 		if err != nil {
-			log.Debug("Failed to read from pipe")
+			log.WithError(err).
+				Error("Failed waiting for forked Teleport process.")
 			return
 		}
-		log.Infof("Received message from pid %v: %v", p.Pid, string(data[:len]))
-		cancel()
+		log.WithField("status", state.String()).Warn("Forked Teleport process has exited.")
 	}()
 
-	select {
-	case <-time.After(signalPipeTimeout):
-		return trace.BadParameter("Failed waiting from process")
-	case <-messageReceived.Done():
-		log.WithFields(logrus.Fields{"pid": p.Pid}).Infof("Child process signals success.")
+	_ = writePipe.Close()
+	readPipe.SetReadDeadline(time.Now().Add(signalPipeTimeout))
+	buf := make([]byte, 1024)
+	// we require at least one byte from the child, otherwise we can't
+	// distinguish the child dying (and closing the pipe) and a deliberate close
+	// without data; conversely, we don't care if we get an I/O or timeout error
+	// if we know that the child has sent at least one byte
+	n, err := io.ReadAtLeast(readPipe, buf, 1)
+	if err != nil {
+		return trace.Wrap(err, "waiting for forked Teleport process to signal successful start")
 	}
+	log.WithField("data", string(buf[:n])).Infof("Forked Teleport process signaled successful start.")
 
 	return nil
 }
-
-// collectStatuses attempts to collect exit statuses from
-// forked teleport child processes.
-// If forked teleport process exited with an error during graceful
-// restart, parent process has to collect the child process status
-// otherwise the child process will become a zombie process.
-// Call Wait4(-1) is trying to collect status of any child
-// leads to warnings in logs, because other parts of the program could
-// have tried to collect the status of this process.
-// Instead this logic tries to collect statuses of the processes
-// forked during restart procedure.
-func (process *TeleportProcess) collectStatuses() {
-	pids := process.getForkedPIDs()
-	if len(pids) == 0 {
-		return
-	}
-	for _, pid := range pids {
-		var wait syscall.WaitStatus
-		rpid, err := syscall.Wait4(pid, &wait, syscall.WNOHANG, nil)
-		if err != nil {
-			process.log.Errorf("Wait call failed: %v.", err)
-			continue
-		}
-		if rpid == pid {
-			process.popForkedPID(pid)
-			process.log.Warningf("Forked teleport process %v has exited with status: %v.", pid, wait.ExitStatus())
-		}
-	}
-}
-
-func (process *TeleportProcess) pushForkedPID(pid int) {
-	process.Lock()
-	defer process.Unlock()
-	process.forkedPIDs = append(process.forkedPIDs, pid)
-}
-
-func (process *TeleportProcess) popForkedPID(pid int) {
-	process.Lock()
-	defer process.Unlock()
-	for i, p := range process.forkedPIDs {
-		if p == pid {
-			process.forkedPIDs = append(process.forkedPIDs[:i], process.forkedPIDs[i+1:]...)
-			return
-		}
-	}
-}
-
-func (process *TeleportProcess) getForkedPIDs() []int {
-	process.Lock()
-	defer process.Unlock()
-	if len(process.forkedPIDs) == 0 {
-		return nil
-	}
-	out := make([]int, len(process.forkedPIDs))
-	copy(out, process.forkedPIDs)
-	return out
-}

lib/sshutils/server.go

@@ -374,20 +374,30 @@ func (s *Server) Shutdown(ctx context.Context) error {
 	if activeConnections == 0 {
 		return err
 	}
+	minReportInterval := 10 * s.shutdownPollPeriod
+	maxReportInterval := 600 * s.shutdownPollPeriod
 	s.log.Infof("Shutdown: waiting for %v connections to finish.", activeConnections)
-	lastReport := time.Time{}
+	reportedConnections := activeConnections
+	lastReport := time.Now()
+	reportInterval := minReportInterval
 	ticker := time.NewTicker(s.shutdownPollPeriod)
 	defer ticker.Stop()
 	for {
 		select {
-		case <-ticker.C:
+		case now := <-ticker.C:
 			activeConnections = s.trackUserConnections(0)
 			if activeConnections == 0 {
 				return err
 			}
-			if time.Since(lastReport) > 10*s.shutdownPollPeriod {
+			if activeConnections != reportedConnections || now.Sub(lastReport) > reportInterval {
 				s.log.Infof("Shutdown: waiting for %v connections to finish.", activeConnections)
-				lastReport = time.Now()
+				lastReport = now
+				if activeConnections == reportedConnections {
+					reportInterval = min(reportInterval*2, maxReportInterval)
+				} else {
+					reportInterval = minReportInterval
+					reportedConnections = activeConnections
+				}
 			}
 		case <-ctx.Done():
 			s.log.Infof("Context canceled wait, returning.")