Repository: coreos/fleet
Pull request #1496 (2016-03-10 15:28:40)

server: fix panic on graceful shutdown
Retry of https://github.com/coreos/fleet/pull/1452 , because Semaphore.

antrik: Weird: Semaphore did check the latest commit, but it doesn't show up in the status box...

Patch:
diff --git a/agent/agent.go b/agent/agent.go
index be793efe..f2216566 100644
--- a/agent/agent.go
+++ b/agent/agent.go
@@ -57,11 +57,11 @@ func (a *Agent) MarshalJSON() ([]byte, error) {
 
 // Heartbeat updates the Registry periodically with an acknowledgement of the
 // Jobs this Agent is expected to be running.
-func (a *Agent) Heartbeat(stop chan bool) {
+func (a *Agent) Heartbeat(stop <-chan struct{}) {
 	a.heartbeatJobs(a.ttl, stop)
 }
 
-func (a *Agent) heartbeatJobs(ttl time.Duration, stop chan bool) {
+func (a *Agent) heartbeatJobs(ttl time.Duration, stop <-chan struct{}) {
 	heartbeat := func() {
 		machID := a.Machine.State().ID
 		launched := a.cache.launchedJobs()
diff --git a/agent/reconcile.go b/agent/reconcile.go
index ef092cbc..957adeea 100644
--- a/agent/reconcile.go
+++ b/agent/reconcile.go
@@ -48,7 +48,7 @@ type AgentReconciler struct {
 // Run periodically attempts to reconcile the provided Agent until the stop
 // channel is closed. Run will also reconcile in reaction to events on the
 // AgentReconciler's rStream.
-func (ar *AgentReconciler) Run(a *Agent, stop chan bool) {
+func (ar *AgentReconciler) Run(a *Agent, stop <-chan struct{}) {
 	reconcile := func() {
 		start := time.Now()
 		ar.Reconcile(a)
diff --git a/agent/unit_state.go b/agent/unit_state.go
index 2f765c51..88997140 100644
--- a/agent/unit_state.go
+++ b/agent/unit_state.go
@@ -70,7 +70,7 @@ type UnitStatePublisher struct {
 // Run caches all of the heartbeat objects from the provided channel, publishing
 // them to the Registry every 5s. Heartbeat objects are also published as they
 // are received on the channel.
-func (p *UnitStatePublisher) Run(beatchan <-chan *unit.UnitStateHeartbeat, stop chan bool) {
+func (p *UnitStatePublisher) Run(beatchan <-chan *unit.UnitStateHeartbeat, stop <-chan struct{}) {
 	go func() {
 		for {
 			select {
diff --git a/api/server.go b/api/server.go
index f2ca08d5..7b6eb028 100644
--- a/api/server.go
+++ b/api/server.go
@@ -57,7 +57,7 @@ func (s *Server) Serve() {
 // Available switches the Server's HTTP handler from a generic 503 Unavailable
 // response to the actual API. Once the provided channel is closed, the API is
 // torn back down and 503 responses are served.
-func (s *Server) Available(stop chan bool) {
+func (s *Server) Available(stop <-chan struct{}) {
 	s.cur = s.api
 	<-stop
 	s.cur = unavailable
diff --git a/engine/engine.go b/engine/engine.go
index bce3ebe3..dd50f4be 100644
--- a/engine/engine.go
+++ b/engine/engine.go
@@ -58,7 +58,7 @@ func New(reg *registry.EtcdRegistry, lManager lease.Manager, rStream pkg.EventSt
 	}
 }
 
-func (e *Engine) Run(ival time.Duration, stop chan bool) {
+func (e *Engine) Run(ival time.Duration, stop <-chan struct{}) {
 	leaseTTL := ival * 5
 	machID := e.machine.State().ID
 
diff --git a/fleetd/fleetd.go b/fleetd/fleetd.go
index 9ff5001b..ff972d61 100755
--- a/fleetd/fleetd.go
+++ b/fleetd/fleetd.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"os"
 	"os/signal"
+	"sync"
 	"syscall"
 
 	"github.com/coreos/fleet/Godeps/_workspace/src/github.com/rakyll/globalconf"
@@ -104,16 +105,21 @@ func main() {
 	}
 	srv.Run()
 
+	srvMutex := sync.Mutex{}
+
 	reconfigure := func() {
 		log.Infof("Reloading configuration from %s", *cfgPath)
 
+		srvMutex.Lock()
+		defer srvMutex.Unlock()
+
 		cfg, err := getConfig(cfgset, *cfgPath)
 		if err != nil {
 			log.Fatalf(err.Error())
 		}
 
 		log.Infof("Restarting server components")
-		srv.Stop()
+		srv.Kill()
 
 		srv, err = server.New(*cfg)
 		if err != nil {
@@ -124,7 +130,11 @@ func main() {
 
 	shutdown := func() {
 		log.Infof("Gracefully shutting down")
-		srv.Stop()
+
+		srvMutex.Lock()
+		defer srvMutex.Unlock()
+
+		srv.Kill()
 		srv.Purge()
 		os.Exit(0)
 	}
@@ -132,6 +142,9 @@ func main() {
 	writeState := func() {
 		log.Infof("Dumping server state")
 
+		srvMutex.Lock()
+		defer srvMutex.Unlock()
+
 		encoded, err := json.Marshal(srv)
 		if err != nil {
 			log.Errorf("Failed to dump server state: %v", err)
diff --git a/functional/util/config.go b/functional/util/config.go
index a2f36dc8..1c29ab2d 100644
--- a/functional/util/config.go
+++ b/functional/util/config.go
@@ -35,6 +35,7 @@ write_files:
     etcd_servers=[{{printf "%q" .EtcdEndpoint}}]
     etcd_key_prefix={{.EtcdKeyPrefix}}
     public_ip={{.IP}}
+    agent_ttl=3s
 
 ssh_authorized_keys:
  - {{printf "%q" .PublicKey}}
diff --git a/machine/coreos.go b/machine/coreos.go
index 1564dc62..4d9ed104 100644
--- a/machine/coreos.go
+++ b/machine/coreos.go
@@ -84,7 +84,7 @@ func (m *CoreOSMachine) Refresh() {
 
 // PeriodicRefresh updates the current state of the CoreOSMachine at the
 // interval indicated. Operation ceases when the provided channel is closed.
-func (m *CoreOSMachine) PeriodicRefresh(interval time.Duration, stop chan bool) {
+func (m *CoreOSMachine) PeriodicRefresh(interval time.Duration, stop <-chan struct{}) {
 	ticker := time.NewTicker(interval)
 	for {
 		select {
diff --git a/pkg/reconcile.go b/pkg/reconcile.go
index 4019a390..2406768e 100644
--- a/pkg/reconcile.go
+++ b/pkg/reconcile.go
@@ -32,7 +32,7 @@ type EventStream interface {
 }
 
 type PeriodicReconciler interface {
-	Run(stop chan bool)
+	Run(stop <-chan struct{})
 }
 
 // NewPeriodicReconciler creates a PeriodicReconciler that will run recFunc at least every
@@ -53,7 +53,7 @@ type reconciler struct {
 	clock   clockwork.Clock
 }
 
-func (r *reconciler) Run(stop chan bool) {
+func (r *reconciler) Run(stop <-chan struct{}) {
 	trigger := make(chan struct{})
 	go func() {
 		abort := make(chan struct{})
diff --git a/heart/monitor.go b/server/monitor.go
similarity index 71%
rename from heart/monitor.go
rename to server/monitor.go
index 87ffb30a..b1c3c1e4 100644
--- a/heart/monitor.go
+++ b/server/monitor.go
@@ -12,12 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package heart
+package server
 
 import (
 	"errors"
 	"time"
 
+	"github.com/coreos/fleet/heart"
 	"github.com/coreos/fleet/log"
 )
 
@@ -30,18 +31,19 @@ type Monitor struct {
 	ival time.Duration
 }
 
-// Monitor ensures a Heart is still beating until a channel is closed, returning
-// an error if the heartbeats fail.
-func (m *Monitor) Monitor(hrt Heart, stop chan bool) error {
+// Monitor periodically checks the given Heart to make sure it
+// beats successfully. If the heartbeat check fails for any
+// reason, an error is returned. If the supplied channel is
+// closed, Monitor returns true and a nil error.
+func (m *Monitor) Monitor(hrt heart.Heart, sdc <-chan struct{}) (bool, error) {
 	ticker := time.Tick(m.ival)
 	for {
 		select {
-		case <-stop:
-			log.Debug("Monitor exiting due to stop signal")
-			return nil
+		case <-sdc:
+			return true, nil
 		case <-ticker:
-			if _, err := m.check(hrt); err != nil {
-				return err
+			if _, err := check(hrt, m.TTL); err != nil {
+				return false, err
 			}
 		}
 	}
@@ -49,10 +51,10 @@ func (m *Monitor) Monitor(hrt Heart, stop chan bool) error {
 
 // check attempts to beat a Heart several times within a timeout, returning the
 // log index at which the beat succeeded or an error
-func (m *Monitor) check(hrt Heart) (idx uint64, err error) {
+func check(hrt heart.Heart, ttl time.Duration) (idx uint64, err error) {
 	// time out after a third of the machine presence TTL, attempting
 	// the heartbeat up to four times
-	timeout := m.TTL / 3
+	timeout := ttl / 3
 	interval := timeout / 4
 
 	tchan := time.After(timeout)
@@ -63,7 +65,7 @@ func (m *Monitor) check(hrt Heart) (idx uint64, err error) {
 			err = errors.New("Monitor timed out before successful heartbeat")
 			return
 		case <-next:
-			idx, err = hrt.Beat(m.TTL)
+			idx, err = hrt.Beat(ttl)
 			if err != nil {
 				log.Debugf("Monitor heartbeat function returned err, retrying in %v: %v", interval, err)
 			}
diff --git a/server/server.go b/server/server.go
index 565d202f..9a947e06 100644
--- a/server/server.go
+++ b/server/server.go
@@ -18,6 +18,7 @@ import (
 	"encoding/json"
 	"errors"
 	"net/http"
+	"sync"
 	"time"
 
 	etcd "github.com/coreos/fleet/Godeps/_workspace/src/github.com/coreos/etcd/client"
@@ -42,6 +43,8 @@ const (
 	// machineStateRefreshInterval is the amount of time the server will
 	// wait before each attempt to refresh the local machine state
 	machineStateRefreshInterval = time.Minute
+
+	shutdownTimeout = time.Minute
 )
 
 type Server struct {
@@ -52,13 +55,15 @@ type Server struct {
 	engine        *engine.Engine
 	mach          *machine.CoreOSMachine
 	hrt           heart.Heart
-	mon           *heart.Monitor
+	mon           *Monitor
 	api           *api.Server
 	disableEngine bool
 
 	engineReconcileInterval time.Duration
 
-	stop chan bool
+	killc chan struct{}  // used to signal monitor to shutdown server
+	stopc chan struct{}  // used to terminate all other goroutines
+	wg    sync.WaitGroup // used to co-ordinate shutdown
 }
 
 func New(cfg config.Config) (*Server, error) {
@@ -116,7 +121,7 @@ func New(cfg config.Config) (*Server, error) {
 	}
 
 	hrt := heart.New(reg, mach)
-	mon := heart.NewMonitor(agentTTL)
+	mon := NewMonitor(agentTTL)
 
 	apiServer := api.NewServer(listeners, api.NewServeMux(reg, cfg.TokenLimit))
 	apiServer.Serve()
@@ -133,7 +138,8 @@ func New(cfg config.Config) (*Server, error) {
 		hrt:         hrt,
 		mon:         mon,
 		api:         apiServer,
-		stop:        nil,
+		killc:       make(chan struct{}),
+		stopc:       nil,
 		engineReconcileInterval: eIval,
 		disableEngine:           cfg.DisableEngine,
 	}
@@ -170,40 +176,70 @@ func (s *Server) Run() {
 		time.Sleep(sleep)
 	}
 
-	log.Infof("Starting server components")
-
-	s.stop = make(chan bool)
+	go s.Supervise()
 
-	go s.Monitor()
-	go s.api.Available(s.stop)
-	go s.mach.PeriodicRefresh(machineStateRefreshInterval, s.stop)
-	go s.agent.Heartbeat(s.stop)
-	go s.aReconciler.Run(s.agent, s.stop)
+	log.Infof("Starting server components")
+	s.stopc = make(chan struct{})
+	s.wg = sync.WaitGroup{}
+	beatc := make(chan *unit.UnitStateHeartbeat)
+
+	components := []func(){
+		func() { s.api.Available(s.stopc) },
+		func() { s.mach.PeriodicRefresh(machineStateRefreshInterval, s.stopc) },
+		func() { s.agent.Heartbeat(s.stopc) },
+		func() { s.aReconciler.Run(s.agent, s.stopc) },
+		func() { s.usGen.Run(beatc, s.stopc) },
+		func() { s.usPub.Run(beatc, s.stopc) },
+	}
 	if s.disableEngine {
 		log.Info("Not starting engine; disable-engine is set")
 	} else {
-		go s.engine.Run(s.engineReconcileInterval, s.stop)
+		components = append(components, func() { s.engine.Run(s.engineReconcileInterval, s.stopc) })
+	}
+	for _, f := range components {
+		f := f
+		s.wg.Add(1)
+		go func() {
+			f()
+			s.wg.Done()
+		}()
 	}
-
-	beatchan := make(chan *unit.UnitStateHeartbeat)
-	go s.usGen.Run(beatchan, s.stop)
-	go s.usPub.Run(beatchan, s.stop)
 }
 
-// Monitor tracks the health of the Server. If the Server is ever deemed
-// unhealthy, the Server is restarted.
-func (s *Server) Monitor() {
-	err := s.mon.Monitor(s.hrt, s.stop)
-	if err != nil {
+// Supervise monitors the life of the Server and coordinates its shutdown.
+// A shutdown occurs when the monitor returns, either because a health check
+// fails or a user triggers a shutdown. If the shutdown is due to a health
+// check failure, the Server is restarted. Supervise will block shutdown until
+// all components have finished shutting down or a timeout occurs; if this
+// happens, the Server will not automatically be restarted.
+func (s *Server) Supervise() {
+	sd, err := s.mon.Monitor(s.hrt, s.killc)
+	if sd {
+		log.Infof("Server monitor triggered: told to shut down")
+	} else {
 		log.Errorf("Server monitor triggered: %v", err)
-
-		s.Stop()
+	}
+	close(s.stopc)
+	done := make(chan struct{})
+	go func() {
+		s.wg.Wait()
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(shutdownTimeout):
+		log.Errorf("Timed out waiting for server to shut down")
+		sd = true
+	}
+	if !sd {
+		log.Infof("Restarting server")
 		s.Run()
 	}
 }
 
-func (s *Server) Stop() {
-	close(s.stop)
+// Kill is used to gracefully terminate the server by triggering the Monitor to shut down
+func (s *Server) Kill() {
+	close(s.killc)
 }
 
 func (s *Server) Purge() {
diff --git a/unit/generator.go b/unit/generator.go
index a2d28e76..045c5c44 100644
--- a/unit/generator.go
+++ b/unit/generator.go
@@ -53,7 +53,7 @@ func (g *UnitStateGenerator) MarshalJSON() ([]byte, error) {
 
 // Run periodically calls Generate and sends received *UnitStateHeartbeat
 // objects to the provided channel.
-func (g *UnitStateGenerator) Run(receiver chan<- *UnitStateHeartbeat, stop chan bool) {
+func (g *UnitStateGenerator) Run(receiver chan<- *UnitStateHeartbeat, stop <-chan struct{}) {
 	tick := time.Tick(time.Second)
 	for {
 		select {


Test patch:
diff --git a/agent/unit_state_test.go b/agent/unit_state_test.go
index 3c58c4d4..cc765a18 100644
--- a/agent/unit_state_test.go
+++ b/agent/unit_state_test.go
@@ -320,7 +320,7 @@ func TestUnitStatePublisherRunTiming(t *testing.T) {
 	}
 
 	bc := make(chan *unit.UnitStateHeartbeat)
-	sc := make(chan bool)
+	sc := make(chan struct{})
 	go func() {
 		usp.Run(bc, sc)
 	}()
@@ -466,7 +466,7 @@ func TestUnitStatePublisherRunQueuing(t *testing.T) {
 		clock:           clockwork.NewFakeClock(),
 	}
 	bc := make(chan *unit.UnitStateHeartbeat)
-	sc := make(chan bool)
+	sc := make(chan struct{})
 	go func() {
 		usp.Run(bc, sc)
 	}()
@@ -599,7 +599,7 @@ func TestUnitStatePublisherRunWithDelays(t *testing.T) {
 	}
 
 	bc := make(chan *unit.UnitStateHeartbeat)
-	sc := make(chan bool)
+	sc := make(chan struct{})
 
 	wgs.Add(numPublishers)
 	wgf.Add(numPublishers)
diff --git a/functional/shutdown_test.go b/functional/shutdown_test.go
new file mode 100644
index 00000000..760c29eb
--- /dev/null
+++ b/functional/shutdown_test.go
@@ -0,0 +1,102 @@
+// Copyright 2016 CoreOS, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package functional
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/coreos/fleet/functional/platform"
+)
+
+// Check clean shutdown of fleetd under normal circumstances
+func TestShutdown(t *testing.T) {
+	cluster, err := platform.NewNspawnCluster("smoke")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cluster.Destroy()
+
+	m0, err := cluster.CreateMember()
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = cluster.WaitForNMachines(m0, 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Stop the fleet process.
+	if _, err = cluster.MemberCommand(m0, "sudo", "systemctl", "stop", "fleet"); err != nil {
+		t.Fatal(err)
+	}
+
+	// Check expected state after stop.
+	stdout, _ := cluster.MemberCommand(m0, "sudo", "systemctl", "show", "--property=ActiveState", "fleet")
+	if strings.TrimSpace(stdout) != "ActiveState=inactive" {
+		t.Fatalf("Fleet unit not reported as inactive: %s", stdout)
+	}
+	stdout, _ = cluster.MemberCommand(m0, "sudo", "systemctl", "show", "--property=Result", "fleet")
+	if strings.TrimSpace(stdout) != "Result=success" {
+		t.Fatalf("Result for fleet unit not reported as success: %s", stdout)
+	}
+}
+
+// Check clean shutdown of fleetd while automatic restart (after failed health check) is in progress
+func TestShutdownVsMonitor(t *testing.T) {
+	cluster, err := platform.NewNspawnCluster("smoke")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cluster.Destroy()
+
+	m0, err := cluster.CreateMember()
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = cluster.WaitForNMachines(m0, 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Cut connection to etcd.
+	//
+	// This will result in a failed health check, and consequently the monitor will attempt a restart.
+	if _, err = cluster.MemberCommand(m0, "sudo", "iptables", "-I", "OUTPUT", "-p", "tcp", "-m", "multiport", "--dports=2379,4001", "-j", "DROP"); err != nil {
+		t.Fatal(err)
+	}
+
+	// Wait for the monitor to trigger the restart.
+	//
+	// This will never complete, as long as there is no connectivity.
+	if _, err = cluster.MemberCommand(m0, "sudo", "sh", "-c", `'until journalctl -u fleet | grep -q "Server monitor triggered: Monitor timed out before successful heartbeat"; do sleep 1; done'`); err != nil {
+		t.Fatal(err)
+	}
+
+	// Stop fleetd while the restart is still in progress.
+	if _, err = cluster.MemberCommand(m0, "sudo", "systemctl", "stop", "fleet"); err != nil {
+		t.Fatal(err)
+	}
+
+	// Verify that fleetd was shut down cleanly in spite of the concurrent restart.
+	stdout, _ := cluster.MemberCommand(m0, "sudo", "systemctl", "show", "--property=ActiveState", "fleet")
+	if strings.TrimSpace(stdout) != "ActiveState=inactive" {
+		t.Fatalf("Fleet unit not reported as inactive: %s", stdout)
+	}
+	stdout, _ = cluster.MemberCommand(m0, "sudo", "systemctl", "show", "--property=Result", "fleet")
+	if strings.TrimSpace(stdout) != "Result=success" {
+		t.Fatalf("Result for fleet unit not reported as success: %s", stdout)
+	}
+}
diff --git a/pkg/reconcile_test.go b/pkg/reconcile_test.go
index 42f5c91e..3c4807a9 100644
--- a/pkg/reconcile_test.go
+++ b/pkg/reconcile_test.go
@@ -54,11 +54,11 @@ func TestPeriodicReconcilerRun(t *testing.T) {
 		clock:   fclock,
 	}
 	// launch the PeriodicReconciler in the background
-	prDone := make(chan bool)
-	stop := make(chan bool)
+	prDone := make(chan struct{})
+	stop := make(chan struct{})
 	go func() {
 		pr.Run(stop)
-		prDone <- true
+		close(prDone)
 	}()
 	// reconcile should have occurred once at start-up
 	select {