Supervisor conceptsupervisor

3 files changed, 114 insertions, 1 deletions

diff --git a/.gitignore b/.gitignore
index 0aba6d488..79f616dad 100644
--- a/.gitignore
+++ b/.gitignore
@@ -6,3 +6,4 @@
 /internal/service/smarthttp/testdata
 /internal/testhelper/testdata
 /config.toml
+/gitaly-supervisor
diff --git a/cmd/gitaly-supervisor/main.go b/cmd/gitaly-supervisor/main.go
new file mode 100644
index 000000000..960bfbfb5
--- /dev/null
+++ b/cmd/gitaly-supervisor/main.go
@@ -0,0 +1,47 @@
+package main
+
+import (
+	"log"
+	"os"
+	"os/exec"
+	"os/signal"
+	"syscall"
+	"time"
+)
+
+func main() {
+
+	pipeR, pipeW, err := os.Pipe()
+	if err != nil {
+		log.Fatal(err)
+	}
+	cmd := exec.Command(os.Args[1], os.Args[2])
+	cmd.Stderr = os.Stderr
+	cmd.Stdout = os.Stdout
+	cmd.ExtraFiles = []*os.File{pipeR}
+	if err := cmd.Start(); err != nil {
+		log.Fatal(err)
+	}
+	if err := pipeR.Close(); err != nil {
+		log.Fatal(err)
+	}
+	go func() {
+		time.Sleep(time.Second)
+		for range time.NewTicker(100 * time.Millisecond).C {
+			if _, err := pipeW.Write([]byte{0}); err != nil {
+				log.Fatal("heartbeat write failed: %v", err)
+			}
+		}
+	}()
+	ch := make(chan os.Signal, 1)
+	signal.Notify(ch, syscall.SIGTERM)
+	go func() {
+		<-ch
+		process := cmd.Process
+		if process == nil {
+			return
+		}
+		syscall.Kill(process.Pid, syscall.SIGHUP)
+	}()
+	log.Print(cmd.Wait())
+}
diff --git a/cmd/gitaly/main.go b/cmd/gitaly/main.go
index 3a2e9062f..ff2083479 100644
--- a/cmd/gitaly/main.go
+++ b/cmd/gitaly/main.go
@@ -5,6 +5,9 @@ import (
 	"net"
 	"net/http"
 	"os"
+	"os/signal"
+	"syscall"
+	"time"
 
 	log "github.com/Sirupsen/logrus"
 
@@ -146,7 +149,29 @@ func main() {
 		}()
 	}
 
-	log.Fatal(<-serverError)
+	hupCh := make(chan os.Signal, 1)
+	signal.Notify(hupCh, syscall.SIGHUP)
+	select {
+	case <-listenSupervisor():
+		log.Print("supervisor went away")
+	case err := <-serverError:
+		log.Printf("server error: %v", err)
+	case <-hupCh:
+		log.Printf("received SIGHUP")
+	}
+
+	stopDone := make(chan struct{})
+	go func() {
+		log.Print("starting graceful stop")
+		server.GracefulStop()
+		close(stopDone)
+	}()
+	select {
+	case <-time.After(10 * time.Minute):
+		log.Print("graceful stop timed out")
+	case <-stopDone:
+	}
+	log.Print("exiting")
 }
 
 func createUnixListener(socketPath string) (net.Listener, error) {
@@ -156,3 +181,43 @@ func createUnixListener(socketPath string) (net.Listener, error) {
 	l, err := net.Listen("unix", socketPath)
 	return connectioncounter.New("unix", l), err
 }
+
+func listenSupervisor() <-chan struct{} {
+	supervisor := os.NewFile(3, "supervisor")
+	if _, err := supervisor.Stat(); err != nil {
+		log.Fatal(err)
+	}
+
+	done := make(chan struct{})
+	go func() {
+		time.Sleep(time.Second)
+		heartbeat := make(chan struct{})
+		go func() {
+			data := make([]byte, 1)
+			for {
+				n, err := supervisor.Read(data)
+				if n > 0 && err == nil {
+					heartbeat <- struct{}{}
+				}
+			}
+		}()
+
+		t := time.NewTicker(100 * time.Millisecond)
+		ticksQuiet := 0
+		for {
+			select {
+			case <-t.C:
+				ticksQuiet += 1
+				if ticksQuiet == 10 {
+					log.Print("too many missed heartbeats from supervisor")
+					close(done)
+					return
+				}
+			case <-heartbeat:
+				ticksQuiet = 0
+			}
+		}
+	}()
+
+	return done
+}