fix(boot.go,pkg): switching to a global mutex for the cleaner to use

Author: Aaron Schlesinger

boot.go

@@ -49,7 +49,7 @@ func main() {
 					os.Exit(1)
 				}
 				pushLock := sshd.NewInMemoryRepositoryLock()
-				deleteLock := sshd.NewInMemoryRepositoryLock()
+				cleanerRef := cleaner.NewRef()
 				circ := sshd.NewCircuit()
 
 				s3Client, err := storage.GetClient(cnf.HealthSrvTestStorageRegion)
@@ -72,15 +72,15 @@ func main() {
 				log.Printf("Starting deleted app cleaner")
 				cleanerErrCh := make(chan error)
 				go func() {
-					if err := cleaner.Run(gitHomeDir, kubeClient.Namespaces(), deleteLock, cnf.CleanerPollSleepDuration()); err != nil {
+					if err := cleanerRef.Run(gitHomeDir, kubeClient.Namespaces(), cleanerRef, cnf.CleanerPollSleepDuration()); err != nil {
 						cleanerErrCh <- err
 					}
 				}()
 
 				log.Printf("Starting SSH server on %s:%d", cnf.SSHHostIP, cnf.SSHHostPort)
 				sshCh := make(chan int)
 				go func() {
-					sshCh <- pkg.RunBuilder(cnf.SSHHostIP, cnf.SSHHostPort, gitHomeDir, circ, pushLock, deleteLock)
+					sshCh <- pkg.RunBuilder(cnf.SSHHostIP, cnf.SSHHostPort, gitHomeDir, circ, pushLock, cleanerRef)
 				}()
 
 				select {

pkg/builder.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/Masterminds/cookoo"
 	clog "github.com/Masterminds/cookoo/log"
+	"github.com/deis/builder/pkg/cleaner"
 	"github.com/deis/builder/pkg/sshd"
 )
 
@@ -27,7 +28,7 @@ const (
 // Git.
 //
 // Run returns on of the Status* status code constants.
-func RunBuilder(sshHostIP string, sshHostPort int, gitHomeDir string, sshServerCircuit *sshd.Circuit, pushLock sshd.RepositoryLock, deleteLock sshd.RepositoryLock) int {
+func RunBuilder(sshHostIP string, sshHostPort int, gitHomeDir string, sshServerCircuit *sshd.Circuit, pushLock sshd.RepositoryLock, cleanerRef cleaner.Ref) int {
 	reg, router, ocxt := cookoo.Cookoo()
 	log.SetFlags(0) // Time is captured elsewhere.
 
@@ -58,7 +59,7 @@ func RunBuilder(sshHostIP string, sshHostPort int, gitHomeDir string, sshServerC
 	// Start the SSH service.
 	// TODO: We could refactor Serve to be a command, and then run this as
 	// a route.
-	if err := sshd.Serve(reg, router, sshServerCircuit, gitHomeDir, pushLock, deleteLock, cxt); err != nil {
+	if err := sshd.Serve(reg, router, sshServerCircuit, gitHomeDir, pushLock, cleanerRef, cxt); err != nil {
 		clog.Errf(cxt, "SSH server failed: %s", err)
 		return StatusLocalError
 	}

pkg/cleaner/cleaner.go

@@ -7,10 +7,10 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/deis/builder/pkg/k8s"
-	"github.com/deis/builder/pkg/sshd"
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/fields"
 	"k8s.io/kubernetes/pkg/labels"
@@ -20,6 +20,22 @@ const (
 	dotGitSuffix = ".git"
 )
 
+type Ref struct {
+	mut *sync.Mutex
+}
+
+func NewRef() Ref {
+	return Ref{mut: new(sync.Mutex)}
+}
+
+func (c Ref) Lock() {
+	c.mut.Lock()
+}
+
+func (c Ref) Unlock() {
+	c.mut.Unlock()
+}
+
 // localDirs returns all of the local directories immediately under gitHome that filter returns true for. filter will receive only the names of each of the top level directories (not their fully qualified paths), and should return true if it should be included in the output
 func localDirs(gitHome string, filter func(string) bool) ([]string, error) {
 	fileInfos, err := ioutil.ReadDir(gitHome)
@@ -79,7 +95,7 @@ func dirHasGitSuffix(dir string) bool {
 }
 
 // Run starts the deleted app cleaner. Every pollSleepDuration, it compares the result of nsLister.List with the directories in the top level of gitHome on the local file system. On any error, it uses log messages to output a human readable description of what happened.
-func Run(gitHome string, nsLister k8s.NamespaceLister, repoLock sshd.RepositoryLock, pollSleepDuration time.Duration) error {
+func (c Ref) Run(gitHome string, nsLister k8s.NamespaceLister, ref Ref, pollSleepDuration time.Duration) error {
 	for {
 		nsList, err := nsLister.List(labels.Everything(), fields.Everything())
 		if err != nil {
@@ -104,21 +120,11 @@ func Run(gitHome string, nsLister k8s.NamespaceLister, repoLock sshd.RepositoryL
 
 		for _, appToDelete := range appsToDelete {
 			dirToDelete := appToDelete + dotGitSuffix
-			// this value must be the same as what's in sshd/server.go. search for repoName := parts[0]
-			lockValue := "/" + filepath.Base(dirToDelete)
-			if err := repoLock.Lock(lockValue, time.Duration(0)); err != nil {
-				log.Printf("Cleaner error locking directory %s for deletion (%s)", dirToDelete, err)
-				continue
-			}
-
+			ref.Lock()
 			if err := os.RemoveAll(dirToDelete); err != nil {
 				log.Printf("Cleaner error removing deleted app %s (%s)", dirToDelete, err)
 			}
-
-			if err := repoLock.Unlock(lockValue, time.Duration(0)); err != nil {
-				log.Printf("Cleaner error unlocking directory %s for deletion (%s)", dirToDelete, err)
-				continue
-			}
+			ref.Unlock()
 		}
 
 		time.Sleep(pollSleepDuration)

pkg/sshd/lock.go

@@ -20,12 +20,27 @@ type RepositoryLock interface {
 	Unlock(repoName string, timeout time.Duration) error
 }
 
+type mutexLock struct {
+	mut *sync.Mutex
+}
+
+func (m *mutexLock) Lock(string, time.Duration) error {
+	m.mut.Lock()
+	return nil
+}
+
+func (m *mutexLock) Unlock(string, time.Duration) error {
+	m.mut.Unlock()
+	return nil
+}
+
 // NewInMemoryRepositoryLock returns a new instance of a RepositoryLock
 func NewInMemoryRepositoryLock() RepositoryLock {
-	return &inMemoryRepoLock{
-		mutex:   &sync.RWMutex{},
-		dataMap: make(map[string]bool),
-	}
+	return &mutexLock{mut: new(sync.Mutex)}
+	// return &inMemoryRepoLock{
+	// 	mutex:   &sync.RWMutex{},
+	// 	dataMap: make(map[string]bool),
+	// }
 }
 
 type inMemoryRepoLock struct {

pkg/sshd/server.go

@@ -17,6 +17,7 @@ import (
 	"github.com/Masterminds/cookoo"
 	"github.com/Masterminds/cookoo/log"
 	"github.com/Masterminds/cookoo/safely"
+	"github.com/deis/builder/pkg/cleaner"
 	"golang.org/x/crypto/ssh"
 )
 
@@ -50,7 +51,15 @@ const (
 //
 // This puts the following variables into the context:
 // 	- ssh.Closer (chan interface{}): Send a message to this to shutdown the server.
-func Serve(reg *cookoo.Registry, router *cookoo.Router, serverCircuit *Circuit, gitHomeDir string, concurrentPushLock RepositoryLock, concurrentDeleteLock RepositoryLock, c cookoo.Context) cookoo.Interrupt {
+func Serve(
+	reg *cookoo.Registry,
+	router *cookoo.Router,
+	serverCircuit *Circuit,
+	gitHomeDir string,
+	concurrentPushLock RepositoryLock,
+	cleanerRef cleaner.Ref,
+	c cookoo.Context) cookoo.Interrupt {
+
 	hostkeys := c.Get(HostKeys, []ssh.Signer{}).([]ssh.Signer)
 	addr := c.Get(Address, "0.0.0.0:2223").(string)
 	cfg := c.Get(ServerConfig, &ssh.ServerConfig{}).(*ssh.ServerConfig)
@@ -69,7 +78,7 @@ func Serve(reg *cookoo.Registry, router *cookoo.Router, serverCircuit *Circuit,
 		c:          c,
 		gitHome:    gitHomeDir,
 		pushLock:   concurrentPushLock,
-		deleteLock: concurrentDeleteLock,
+		cleanerRef: cleanerRef,
 	}
 
 	closer := make(chan interface{}, 1)
@@ -87,7 +96,7 @@ type server struct {
 	c          cookoo.Context
 	gitHome    string
 	pushLock   RepositoryLock
-	deleteLock RepositoryLock
+	cleanerRef cleaner.Ref
 }
 
 // listen handles accepting and managing connections. However, since closer
@@ -223,16 +232,8 @@ func (s *server) answer(channel ssh.Channel, requests <-chan *ssh.Request, sshCo
 				}
 
 				repoName := parts[1]
-				if err := s.deleteLock.Lock(repoName, time.Duration(0)); err != nil {
-					log.Errf(s.c, inProgressDelete)
-					// The error must be in git format
-					if err := gitPktLine(channel, fmt.Sprintf("ERR %v\n", inProgressDelete)); err != nil {
-						log.Errf(s.c, "Failed to write to channel: %s", err)
-					}
-					sendExitStatus(1, channel)
-					req.Reply(false, nil)
-					return nil
-				}
+				fmt.Printf("Server locking %s for delete\n", repoName)
+				s.cleanerRef.Lock()
 				if err := s.pushLock.Lock(repoName, time.Duration(0)); err != nil {
 					log.Errf(s.c, multiplePush)
 					// The error must be in git format
@@ -241,6 +242,7 @@ func (s *server) answer(channel ssh.Channel, requests <-chan *ssh.Request, sshCo
 					}
 					sendExitStatus(1, channel)
 					req.Reply(false, nil)
+					s.cleanerRef.Unlock()
 					return nil
 				}
 
@@ -257,11 +259,8 @@ func (s *server) answer(channel ssh.Channel, requests <-chan *ssh.Request, sshCo
 					// TODO: this is an important error case that needs to be covered
 					// Probably the best solution is to change the lock into a lease so that even on unlock failures, RepositoryLock will eventually yield
 				}
-				if err := s.deleteLock.Unlock(repoName, time.Duration(0)); err != nil {
-					log.Errf(s.c, "unable to unlock delete lock for %s (%s)", repoName, err)
-					// TODO: this is an important error case that needs to be covered
-					// Probably the best solution is to change the lock into a lease so that even on unlock failures, RepositoryLock will eventually yield
-				}
+				s.cleanerRef.Unlock()
+				fmt.Printf("Server unlocked %s for delete\n", repoName)
 				var xs uint32
 				if err != nil {
 					log.Errf(s.c, "Failed git receive: %v", err)