ref(pkg/sshd,pkg/cleaner,pkg/builder.go,boot.go): surfacing the repository push lock, adding a repo delete lock

The deleted app cleaner uses the repo delete lock, and this commit
surfaces the concurrent push lock for symmetry with the delete lock

Author: Aaron Schlesinger

boot.go

@@ -48,6 +48,8 @@ func main() {
 					pkglog.Err("getting config for %s [%s]", serverConfAppName, err)
 					os.Exit(1)
 				}
+				pushLock := sshd.NewInMemoryRepositoryLock()
+				deleteLock := sshd.NewInMemoryRepositoryLock()
 				circ := sshd.NewCircuit()
 
 				s3Client, err := storage.GetClient(cnf.HealthSrvTestStorageRegion)
@@ -70,15 +72,15 @@ func main() {
 				log.Printf("Starting deleted app cleaner")
 				cleanerErrCh := make(chan error)
 				go func() {
-					if err := cleaner.Run(gitHomeDir, kubeClient.Namespaces(), cnf.CleanerPollSleepDuration); err != nil {
+					if err := cleaner.Run(gitHomeDir, kubeClient.Namespaces(), deleteLock, 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)
+					sshCh <- pkg.RunBuilder(cnf.SSHHostIP, cnf.SSHHostPort, gitHomeDir, circ, pushLock, deleteLock)
 				}()
 
 				select {

pkg/builder.go

@@ -27,7 +27,7 @@ const (
 // Git.
 //
 // Run returns on of the Status* status code constants.
-func RunBuilder(sshHostIP string, sshHostPort int, gitHomeDir string, sshServerCircuit *sshd.Circuit) int {
+func RunBuilder(sshHostIP string, sshHostPort int, gitHomeDir string, sshServerCircuit *sshd.Circuit, pushLock sshd.RepositoryLock, deleteLock sshd.RepositoryLock) int {
 	reg, router, ocxt := cookoo.Cookoo()
 	log.SetFlags(0) // Time is captured elsewhere.
 
@@ -58,7 +58,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, cxt); err != nil {
+	if err := sshd.Serve(reg, router, sshServerCircuit, gitHomeDir, pushLock, deleteLock, cxt); err != nil {
 		clog.Errf(cxt, "SSH server failed: %s", err)
 		return StatusLocalError
 	}

pkg/cleaner/cleaner.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/deis/builder/pkg/k8s"
+	"github.com/deis/builder/pkg/sshd"
 	"github.com/deis/pkg/log"
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/fields"
@@ -70,8 +71,7 @@ func getDisjunction(namespaceList []api.Namespace, dirs []string) []string {
 }
 
 // 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.Debug to output a human readable description of what happened.
-// TODO: locking mechanism on repositories. Nobody should be able to push to a repo while one is being deleted
-func Run(gitHome string, nsLister k8s.NamespaceLister, pollSleepDuration time.Duration) error {
+func Run(gitHome string, nsLister k8s.NamespaceLister, repoLock sshd.RepositoryLock, pollSleepDuration time.Duration) error {
 	for {
 		nsList, err := nsLister.List(labels.Everything(), fields.Everything())
 		if err != nil {
@@ -86,9 +86,17 @@ func Run(gitHome string, nsLister k8s.NamespaceLister, pollSleepDuration time.Du
 
 		disjunctions := getDisjunction(nsList.Items, gitDirs)
 		for _, disj := range disjunctions {
+			if err := repoLock.Lock(disj, time.Duration(0)); err != nil {
+				log.Debug("Cleaner error locking repository %s for deletion (%s)", disj, err)
+				continue
+			}
 			if err := os.RemoveAll(disj); err != nil {
 				log.Debug("Cleaner error removing deleted app %s (%s)", disj, err)
 			}
+			if err := repoLock.Unlock(disj, time.Duration(0)); err != nil {
+				log.Debug("Cleaner error unlocking repository %s for deletion (%s)", disj, err)
+				continue
+			}
 		}
 
 		time.Sleep(pollSleepDuration)

pkg/sshd/server.go

@@ -12,8 +12,6 @@ import (
 	"io"
 	"net"
 	"strings"
-	"sync"
-	"text/template"
 	"time"
 
 	"github.com/Masterminds/cookoo"
@@ -31,10 +29,8 @@ const (
 	ServerConfig string = "ssh.ServerConfig"
 
 	multiplePush string = "Another git push is ongoing"
-)
 
-var (
-	buildingRepos = NewInMemoryRepositoryLock()
+	inProgressDelete string = "This app was deleted and is being cleaned up. Please re-create it with 'deis create your_app'"
 )
 
 // Serve starts a native SSH server.
@@ -54,7 +50,7 @@ var (
 //
 // 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, c cookoo.Context) cookoo.Interrupt {
+func Serve(reg *cookoo.Registry, router *cookoo.Router, serverCircuit *Circuit, gitHomeDir string, concurrentPushLock RepositoryLock, concurrentDeleteLock RepositoryLock, 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)
@@ -70,8 +66,10 @@ func Serve(reg *cookoo.Registry, router *cookoo.Router, serverCircuit *Circuit,
 	}
 
 	srv := &server{
-		c:       c,
-		gitHome: gitHomeDir,
+		c:          c,
+		gitHome:    gitHomeDir,
+		pushLock:   concurrentPushLock,
+		deleteLock: concurrentDeleteLock,
 	}
 
 	closer := make(chan interface{}, 1)
@@ -88,8 +86,8 @@ func Serve(reg *cookoo.Registry, router *cookoo.Router, serverCircuit *Circuit,
 type server struct {
 	c          cookoo.Context
 	gitHome    string
-	hookTpl    *template.Template
-	createLock sync.Mutex
+	pushLock   RepositoryLock
+	deleteLock RepositoryLock
 }
 
 // listen handles accepting and managing connections. However, since closer
@@ -225,7 +223,17 @@ func (s *server) answer(channel ssh.Channel, requests <-chan *ssh.Request, sshCo
 				}
 
 				repoName := parts[1]
-				if err := buildingRepos.Lock(repoName, time.Duration(0)); err != nil {
+				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
+				}
+				if err := s.pushLock.Lock(repoName, time.Duration(0)); err != nil {
 					log.Errf(s.c, multiplePush)
 					// The error must be in git format
 					if err := gitPktLine(channel, fmt.Sprintf("ERR %v\n", multiplePush)); err != nil {
@@ -244,7 +252,16 @@ func (s *server) answer(channel ssh.Channel, requests <-chan *ssh.Request, sshCo
 				cxt.Put("repository", parts[1])
 				sshGitReceive := cxt.Get("route.sshd.sshGitReceive", "sshGitReceive").(string)
 				err := router.HandleRequest(sshGitReceive, cxt, true)
-				buildingRepos.Unlock(repoName, time.Duration(0))
+				if err := s.pushLock.Unlock(repoName, time.Duration(0)); err != nil {
+					log.Errf(s.c, "unable to unlock repository 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
+				}
+				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
+				}
 				var xs uint32
 				if err != nil {
 					log.Errf(s.c, "Failed git receive: %v", err)