Implement systemd-notify protocol

contrib/systemd/gitea.service

@@ -56,7 +56,7 @@ After=network.target
 #LimitMEMLOCK=infinity
 #LimitNOFILE=65535
 RestartSec=2s
-Type=simple
+Type=notify
 User=git
 Group=git
 WorkingDirectory=/var/lib/gitea/

modules/graceful/manager_unix.go

@@ -12,6 +12,7 @@ import (
 	"os"
 	"os/signal"
 	"runtime/pprof"
+	"strconv"
 	"sync"
 	"syscall"
 	"time"
@@ -46,14 +47,49 @@ type Manager struct {
 
 func newGracefulManager(ctx context.Context) *Manager {
 	manager := &Manager{
-		isChild: len(os.Getenv(listenFDs)) > 0 && os.Getppid() > 1,
+		isChild: len(os.Getenv(listenFDsEnv)) > 0 && os.Getppid() > 1,
 		lock:    &sync.RWMutex{},
 	}
 	manager.createServerWaitGroup.Add(numberOfServersToCreate)
 	manager.start(ctx)
 	return manager
 }
 
+type systemdNotifyMsg string
+
+const (
+	readyMsg     systemdNotifyMsg = "READY=1"
+	stoppingMsg  systemdNotifyMsg = "STOPPING=1"
+	reloadingMsg systemdNotifyMsg = "RELOADING=1"
+	watchdogMsg  systemdNotifyMsg = "WATCHDOG=1"
+)
+
+func statusMsg(msg string) systemdNotifyMsg {
+	return systemdNotifyMsg("STATUS=" + msg)
+}
+
+func pidMsg() systemdNotifyMsg {
+	return systemdNotifyMsg("MAINPID=" + strconv.Itoa(os.Getpid()))
+}
+
+// Notify systemd of status via the notify protocol
+func (g *Manager) notify(msg systemdNotifyMsg) {
+	conn, err := getNotifySocket()
+	if err != nil {
+		// the err is logged in getNotifySocket
+		return
+	}
+	if conn == nil {
+		return
+	}
+	defer conn.Close()
+
+	if _, err = conn.Write([]byte(msg)); err != nil {
+		log.Warn("Failed to notify NOTIFY_SOCKET: %v", err)
+		return
+	}
+}
+
 func (g *Manager) start(ctx context.Context) {
 	// Make contexts
 	g.terminateCtx, g.terminateCtxCancel = context.WithCancel(ctx)
@@ -73,6 +109,8 @@ func (g *Manager) start(ctx context.Context) {
 
 	// Set the running state & handle signals
 	g.setState(stateRunning)
+	g.notify(statusMsg("Starting Gitea"))
+	g.notify(pidMsg())
 	go g.handleSignals(g.managerCtx)
 
 	// Handle clean up of unused provided listeners	and delayed start-up
@@ -85,6 +123,7 @@ func (g *Manager) start(ctx context.Context) {
 		// Ignore the error here there's not much we can do with it
 		// They're logged in the CloseProvidedListeners function
 		_ = CloseProvidedListeners()
+		g.notify(readyMsg)
 	}()
 	if setting.StartupTimeout > 0 {
 		go func() {
@@ -105,6 +144,8 @@ func (g *Manager) start(ctx context.Context) {
 				return
 			case <-time.After(setting.StartupTimeout):
 				log.Error("Startup took too long! Shutting down")
+				g.notify(statusMsg("Startup took too long! Shutting down"))
+				g.notify(stoppingMsg)
 				g.doShutdown()
 			}
 		}()
@@ -127,6 +168,13 @@ func (g *Manager) handleSignals(ctx context.Context) {
 		syscall.SIGTSTP,
 	)
 
+	watchdogTimeout := getWatchdogTimeout()
+	t := &time.Ticker{}
+	if watchdogTimeout != 0 {
+		g.notify(watchdogMsg)
+		t = time.NewTicker(watchdogTimeout / 2)
+	}
+
 	pid := syscall.Getpid()
 	for {
 		select {
@@ -137,6 +185,7 @@ func (g *Manager) handleSignals(ctx context.Context) {
 				g.DoGracefulRestart()
 			case syscall.SIGUSR1:
 				log.Warn("PID %d. Received SIGUSR1. Releasing and reopening logs", pid)
+				g.notify(statusMsg("Releasing and reopening logs"))
 				if err := log.ReleaseReopen(); err != nil {
 					log.Error("Error whilst releasing and reopening logs: %v", err)
 				}
@@ -154,6 +203,8 @@ func (g *Manager) handleSignals(ctx context.Context) {
 			default:
 				log.Info("PID %d. Received %v.", pid, sig)
 			}
+		case <-t.C:
+			g.notify(watchdogMsg)
 		case <-ctx.Done():
 			log.Warn("PID: %d. Background context for manager closed - %v - Shutting down...", pid, ctx.Err())
 			g.DoGracefulShutdown()
@@ -170,6 +221,9 @@ func (g *Manager) doFork() error {
 	}
 	g.forked = true
 	g.lock.Unlock()
+
+	g.notify(reloadingMsg)
+
 	// We need to move the file logs to append pids
 	setting.RestartLogsWithPIDSuffix()
 
@@ -192,18 +246,27 @@ func (g *Manager) DoGracefulRestart() {
 		}
 	} else {
 		log.Info("PID: %d. Not set restartable. Shutting down...", os.Getpid())
-
+		g.notify(stoppingMsg)
 		g.doShutdown()
 	}
 }
 
 // DoImmediateHammer causes an immediate hammer
 func (g *Manager) DoImmediateHammer() {
+	g.notify(statusMsg("Sending immediate hammer"))
 	g.doHammerTime(0 * time.Second)
 }
 
 // DoGracefulShutdown causes a graceful shutdown
 func (g *Manager) DoGracefulShutdown() {
+	g.lock.Lock()
+	if !g.forked {
+		g.lock.Unlock()
+		g.notify(stoppingMsg)
+	} else {
+		g.lock.Unlock()
+		g.notify(statusMsg("shutting down after fork"))
+	}
 	g.doShutdown()
 }
 

modules/graceful/net_unix.go

@@ -14,16 +14,20 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"time"
 
 	"code.gitea.io/gitea/modules/log"
 	"code.gitea.io/gitea/modules/setting"
 	"code.gitea.io/gitea/modules/util"
 )
 
 const (
-	listenFDs = "LISTEN_FDS"
-	startFD   = 3
-	unlinkFDs = "GITEA_UNLINK_FDS"
+	listenFDsEnv = "LISTEN_FDS"
+	startFD      = 3
+	unlinkFDsEnv = "GITEA_UNLINK_FDS"
+
+	notifySocketEnv    = "NOTIFY_SOCKET"
+	watchdogTimeoutEnv = "WATCHDOG_USEC"
 )
 
 // In order to keep the working directory the same as when we started we record
@@ -38,51 +42,87 @@ var (
 	activeListenersToUnlink   = []bool{}
 	providedListeners         = []net.Listener{}
 	activeListeners           = []net.Listener{}
+
+	notifySocketAddr string
+	watchdogTimeout  time.Duration
 )
 
 func getProvidedFDs() (savedErr error) {
 	// Only inherit the provided FDS once but we will save the error so that repeated calls to this function will return the same error
 	once.Do(func() {
 		mutex.Lock()
 		defer mutex.Unlock()
+		// now handle some additional systemd provided things
+		notifySocketAddr = os.Getenv(notifySocketEnv)
+		if notifySocketAddr != "" {
+			log.Debug("Systemd Notify Socket provided: %s", notifySocketAddr)
+			savedErr = os.Unsetenv(notifySocketEnv)
+			if savedErr != nil {
+				log.Warn("Unable to Unset the NOTIFY_SOCKET environment variable: %v", savedErr)
+				return
+			}
 
-		numFDs := os.Getenv(listenFDs)
-		if numFDs == "" {
-			return
-		}
-		n, err := strconv.Atoi(numFDs)
-		if err != nil {
-			savedErr = fmt.Errorf("%s is not a number: %s. Err: %v", listenFDs, numFDs, err)
-			return
-		}
+			// FIXME: We don't handle WATCHDOG_PID
+			timeoutStr := os.Getenv(watchdogTimeoutEnv)
+			if timeoutStr != "" {
+				savedErr = os.Unsetenv(watchdogTimeoutEnv)
+				if savedErr != nil {
+					log.Warn("Unable to Unset the WATCHDOG_USEC environment variable: %v", savedErr)
+					return
+				}
 
-		fdsToUnlinkStr := strings.Split(os.Getenv(unlinkFDs), ",")
-		providedListenersToUnlink = make([]bool, n)
-		for _, fdStr := range fdsToUnlinkStr {
-			i, err := strconv.Atoi(fdStr)
-			if err != nil || i < 0 || i >= n {
-				continue
+				s, err := strconv.ParseInt(timeoutStr, 10, 64)
+				if err != nil {
+					log.Error("Unable to parse the provided WATCHDOG_USEC: %v", err)
+					savedErr = fmt.Errorf("unable to parse the provided WATCHDOG_USEC: %w", err)
+					return
+				}
+				if s <= 0 {
+					log.Error("Unable to parse the provided WATCHDOG_USEC: %s should be a positive number", timeoutStr)
+					savedErr = fmt.Errorf("unable to parse the provided WATCHDOG_USEC: %s should be a positive number", timeoutStr)
+					return
+				}
+				watchdogTimeout = time.Duration(s) * time.Microsecond
 			}
-			providedListenersToUnlink[i] = true
+		} else {
+			log.Trace("No Systemd Notify Socket provided")
 		}
 
-		for i := startFD; i < n+startFD; i++ {
-			file := os.NewFile(uintptr(i), fmt.Sprintf("listener_FD%d", i))
+		if numFDs := os.Getenv(listenFDsEnv); numFDs != "" {
+			n, err := strconv.Atoi(numFDs)
+			if err != nil {
+				savedErr = fmt.Errorf("%s is not a number: %s. Err: %v", listenFDsEnv, numFDs, err)
+				return
+			}
 
-			l, err := net.FileListener(file)
-			if err == nil {
-				// Close the inherited file if it's a listener
-				if err = file.Close(); err != nil {
-					savedErr = fmt.Errorf("error closing provided socket fd %d: %s", i, err)
-					return
+			fdsToUnlinkStr := strings.Split(os.Getenv(unlinkFDsEnv), ",")
+			providedListenersToUnlink = make([]bool, n)
+			for _, fdStr := range fdsToUnlinkStr {
+				i, err := strconv.Atoi(fdStr)
+				if err != nil || i < 0 || i >= n {
+					continue
 				}
-				providedListeners = append(providedListeners, l)
-				continue
+				providedListenersToUnlink[i] = true
 			}
 
-			// If needed we can handle packetconns here.
-			savedErr = fmt.Errorf("Error getting provided socket fd %d: %v", i, err)
-			return
+			for i := startFD; i < n+startFD; i++ {
+				file := os.NewFile(uintptr(i), fmt.Sprintf("listener_FD%d", i))
+
+				l, err := net.FileListener(file)
+				if err == nil {
+					// Close the inherited file if it's a listener
+					if err = file.Close(); err != nil {
+						savedErr = fmt.Errorf("error closing provided socket fd %d: %s", i, err)
+						return
+					}
+					providedListeners = append(providedListeners, l)
+					continue
+				}
+
+				// If needed we can handle packetconns here.
+				savedErr = fmt.Errorf("Error getting provided socket fd %d: %v", i, err)
+				return
+			}
 		}
 	})
 	return savedErr
@@ -255,3 +295,36 @@ func getActiveListenersToUnlink() []bool {
 	copy(listenersToUnlink, activeListenersToUnlink)
 	return listenersToUnlink
 }
+
+func getNotifySocket() (*net.UnixConn, error) {
+	if err := getProvidedFDs(); err != nil {
+		// This error will be logged elsewhere
+		return nil, nil
+	}
+
+	if notifySocketAddr == "" {
+		return nil, nil
+	}
+
+	socketAddr := &net.UnixAddr{
+		Name: notifySocketAddr,
+		Net:  "unixgram",
+	}
+
+	notifySocket, err := net.DialUnix(socketAddr.Net, nil, socketAddr)
+	if err != nil {
+		log.Warn("failed to dial NOTIFY_SOCKET %s: %v", socketAddr, err)
+		return nil, err
+	}
+
+	return notifySocket, nil
+}
+
+func getWatchdogTimeout() time.Duration {
+	if err := getProvidedFDs(); err != nil {
+		// This error will be logged elsewhere
+		return 0
+	}
+
+	return watchdogTimeout
+}

modules/graceful/restart_unix.go

@@ -16,6 +16,7 @@ import (
 	"strings"
 	"sync"
 	"syscall"
+	"time"
 )
 
 var killParent sync.Once
@@ -70,11 +71,20 @@ func RestartProcess() (int, error) {
 	// Pass on the environment and replace the old count key with the new one.
 	var env []string
 	for _, v := range os.Environ() {
-		if !strings.HasPrefix(v, listenFDs+"=") {
+		if !strings.HasPrefix(v, listenFDsEnv+"=") {
 			env = append(env, v)
 		}
 	}
-	env = append(env, fmt.Sprintf("%s=%d", listenFDs, len(listeners)))
+	env = append(env, fmt.Sprintf("%s=%d", listenFDsEnv, len(listeners)))
+
+	if notifySocketAddr != "" {
+		env = append(env, fmt.Sprintf("%s=%s", notifySocketEnv, notifySocketAddr))
+	}
+
+	if watchdogTimeout != 0 {
+		watchdogStr := strconv.FormatInt(int64(watchdogTimeout/time.Millisecond), 10)
+		env = append(env, fmt.Sprintf("%s=%s", watchdogTimeoutEnv, watchdogStr))
+	}
 
 	sb := &strings.Builder{}
 	for i, unlink := range getActiveListenersToUnlink() {
@@ -87,7 +97,7 @@ func RestartProcess() (int, error) {
 	unlinkStr := sb.String()
 	if len(unlinkStr) > 0 {
 		unlinkStr = unlinkStr[:len(unlinkStr)-1]
-		env = append(env, fmt.Sprintf("%s=%s", unlinkFDs, unlinkStr))
+		env = append(env, fmt.Sprintf("%s=%s", unlinkFDsEnv, unlinkStr))
 	}
 
 	allFiles := append([]*os.File{os.Stdin, os.Stdout, os.Stderr}, files...)