feat(deisctl) : add new feautes and update core-os updatectl to updateservicectl

add timeout for cmdInstall during updateservice function
add lock instance during update
move core-os updatectl package to updateservicectl package

Author: smothiki

cmd/cmd.go

@@ -15,9 +15,9 @@ func List(c client.Client) error {
 }
 
 func PullImage(service string) error {
-	dockercli, _, _ := utils.GetNewClient()
+	//dockercli, _, _ := utils.GetNewClient()
 	fmt.Println("pulling image :" + strings.Replace(strings.Split(service, ".")[0], "-", "/", 1) + ":latest")
-	err := utils.PullImage(dockercli, strings.Replace(strings.Split(service, ".")[0], "-", "/", 1)+":latest")
+	err := utils.PullImage(strings.Replace(strings.Split(service, ".")[0], "-", "/", 1) + ":latest")
 	if err != nil {
 		return err
 	}

constant/constants.go

@@ -1,9 +1,13 @@
 package constant
 
+import "time"
+
 const (
-	UnitsDir     = "/var/lib/deis/units/"
-	HooksDir     = "/var/lib/deis/hooks/"
-	Version      = "/etc/deis-version"
-	MachineId    = "/etc/machine-id"
-	UpdatekeyDir = "/deis/update/"
+	UnitsDir        = "/var/lib/deis/units/"
+	HooksDir        = "/var/lib/deis/hooks/"
+	Version         = "/etc/deis-version"
+	MachineId       = "/etc/machine-id"
+	UpdatekeyDir    = "/deis/update/"
+	InitialInterval = time.Second * 10
+	MaxInterval     = time.Minute * 7
 )

lock/client.go

@@ -0,0 +1,7 @@
+package lock
+
+type LockClient interface {
+	Init() error
+	Get() (*Semaphore, error)
+	Set(*Semaphore) error
+}

lock/etcd.go

@@ -0,0 +1,78 @@
+package lock
+
+import (
+	"encoding/json"
+
+	etcdError "github.com/coreos/locksmith/third_party/github.com/coreos/etcd/error"
+	"github.com/coreos/locksmith/third_party/github.com/coreos/go-etcd/etcd"
+)
+
+const (
+	keyPrefix       = "coreos.com/updateengine/rebootlock"
+	holdersPrefix   = keyPrefix + "/holders"
+	SemaphorePrefix = keyPrefix + "/semaphore"
+)
+
+// EtcdLockClient is a wrapper around the go-etcd client that provides
+// simple primitives to operate on the internal semaphore and holders
+// structs through etcd.
+type EtcdLockClient struct {
+	client *etcd.Client
+}
+
+func NewEtcdLockClient(machines []string) (client *EtcdLockClient, err error) {
+	ec := etcd.NewClient(machines)
+	client = &EtcdLockClient{ec}
+	err = client.Init()
+
+	return client, err
+}
+
+// Init sets an initial copy of the semaphore if it doesn't exist yet.
+func (c *EtcdLockClient) Init() (err error) {
+	sem := newSemaphore()
+	b, err := json.Marshal(sem)
+	if err != nil {
+		return err
+	}
+
+	_, err = c.client.Create(SemaphorePrefix, string(b), 0)
+	if err != nil {
+		eerr, ok := err.(*etcd.EtcdError)
+		if ok && eerr.ErrorCode == etcdError.EcodeNodeExist {
+			return nil
+		}
+	}
+
+	return err
+}
+
+// Get fetches the Semaphore from etcd.
+func (c *EtcdLockClient) Get() (sem *Semaphore, err error) {
+	resp, err := c.client.Get(SemaphorePrefix, false, false)
+	if err != nil {
+		return nil, err
+	}
+
+	sem = &Semaphore{}
+	err = json.Unmarshal([]byte(resp.Node.Value), sem)
+	if err != nil {
+		return nil, err
+	}
+
+	sem.Index = resp.Node.ModifiedIndex
+
+	return sem, nil
+}
+
+// Set sets a Semaphore in etcd.
+func (c *EtcdLockClient) Set(sem *Semaphore) (err error) {
+	b, err := json.Marshal(sem)
+	if err != nil {
+		return err
+	}
+
+	_, err = c.client.CompareAndSwap(SemaphorePrefix, string(b), 0, "", sem.Index)
+
+	return err
+}

lock/lock.go

@@ -0,0 +1,62 @@
+package lock
+
+type Lock struct {
+	id     string
+	client LockClient
+}
+
+func New(id string, client LockClient) (lock *Lock) {
+	return &Lock{id, client}
+}
+
+func (l *Lock) store(f func(*Semaphore) error) (err error) {
+	sem, err := l.client.Get()
+	if err != nil {
+		return err
+	}
+
+	if err := f(sem); err != nil {
+		return err
+	}
+
+	err = l.client.Set(sem)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (l *Lock) Get() (sem *Semaphore, err error) {
+	sem, err = l.client.Get()
+	if err != nil {
+		return nil, err
+	}
+
+	return sem, nil
+}
+
+func (l *Lock) SetMax(max int) (sem *Semaphore, oldMax int, err error) {
+	var (
+		semRet *Semaphore
+		old    int
+	)
+
+	return semRet, old, l.store(func(sem *Semaphore) error {
+		old = sem.Max
+		semRet = sem
+		return sem.SetMax(max)
+	})
+}
+
+func (l *Lock) Lock() (err error) {
+	return l.store(func(sem *Semaphore) error {
+		return sem.Lock(l.id)
+	})
+}
+
+func (l *Lock) Unlock() error {
+	return l.store(func(sem *Semaphore) error {
+		return sem.Unlock(l.id)
+	})
+}

lock/semaphore.go

@@ -0,0 +1,92 @@
+package lock
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sort"
+)
+
+var (
+	ErrExist    = errors.New("holder exists")
+	ErrNotExist = errors.New("holder does not exist")
+)
+
+type Semaphore struct {
+	Index     uint64   `json:"-"`
+	Semaphore int      `json:"semaphore"`
+	Max       int      `json:"max"`
+	Holders   []string `json:"holders"`
+}
+
+func (s *Semaphore) SetMax(max int) error {
+	diff := s.Max - max
+
+	s.Semaphore = s.Semaphore - diff
+	s.Max = s.Max - diff
+
+	return nil
+}
+
+func (s *Semaphore) String() string {
+	b, _ := json.Marshal(s)
+	return string(b)
+}
+
+func (s *Semaphore) addHolder(h string) error {
+	loc := sort.SearchStrings(s.Holders, h)
+	switch {
+	case loc == len(s.Holders):
+		s.Holders = append(s.Holders, h)
+	case s.Holders[loc] == h:
+		return ErrExist
+	default:
+		s.Holders = append(s.Holders[:loc], append([]string{h}, s.Holders[loc:]...)...)
+	}
+
+	return nil
+}
+
+func (s *Semaphore) removeHolder(h string) error {
+	loc := sort.SearchStrings(s.Holders, h)
+	if loc < len(s.Holders) && s.Holders[loc] == h {
+		s.Holders = append(s.Holders[:loc], s.Holders[loc+1:]...)
+	} else {
+		return ErrNotExist
+	}
+
+	return nil
+}
+
+func (s *Semaphore) Lock(h string) error {
+	if s.Semaphore <= 0 {
+		return fmt.Errorf("semaphore is at %v", s.Semaphore)
+	}
+
+	if err := s.addHolder(h); err != nil {
+		return err
+	}
+
+	s.Semaphore = s.Semaphore - 1
+
+	return nil
+}
+
+func (s *Semaphore) Unlock(h string) error {
+	if err := s.removeHolder(h); err != nil {
+		return err
+	}
+
+	s.Semaphore = s.Semaphore + 1
+
+	return nil
+}
+
+func newSemaphore() (sem *Semaphore) {
+	return &Semaphore{0, 1, 1, nil}
+}
+
+type holder struct {
+	ID        string `json:"-"`
+	StartTime int64  `json:"startTime"`
+}

updatectl/instance.go

@@ -15,10 +15,11 @@ import (
 
 	"code.google.com/p/go-uuid/uuid"
 	"github.com/coreos/go-omaha/omaha"
-	update "github.com/coreos/updatectl/client/update/v1"
+	update "github.com/coreos/updateservicectl/client/update/v1"
 	"github.com/deis/deisctl/client"
 	"github.com/deis/deisctl/cmd"
 	"github.com/deis/deisctl/constant"
+	"github.com/deis/deisctl/lock"
 	"github.com/deis/deisctl/utils"
 )
 
@@ -71,6 +72,14 @@ func init() {
 	cmdInstanceDeis.Flags.StringVar(&instanceFlags.version, "version", utils.GetVersion(), "Version to report.")
 }
 
+func expBackoff(interval time.Duration) time.Duration {
+	interval = interval * 2
+	if interval > constant.MaxInterval {
+		interval = constant.MaxInterval
+	}
+	return interval
+}
+
 //+ downloadDir + "deis.tar.gz"
 
 type serverConfig struct {
@@ -86,6 +95,7 @@ type Client struct {
 	config         *serverConfig
 	errorRate      int
 	pingsRemaining int
+	lock           *lock.Lock
 }
 
 func (c *Client) Log(format string, v ...interface{}) {
@@ -102,6 +112,19 @@ func (c *Client) getCodebaseUrl(uc *omaha.UpdateCheck) string {
 	return uc.Urls.Urls[0].CodeBase + uc.Manifest.Packages.Packages[0].Name
 }
 
+func (c *Client) RequestLock() {
+	elc, err := lock.NewEtcdLockClient(nil)
+	if err != nil {
+		fmt.Fprintln(os.Stderr, "Error initializing etcd client:", err)
+	}
+	var mID string
+	mID = utils.GetMachineID("/")
+	if mID == "" {
+		fmt.Fprintln(os.Stderr, "Cannot read machine-id")
+	}
+	c.lock = lock.New(mID, elc)
+}
+
 func (c *Client) updateservice() (err error) {
 	fmt.Println("starting systemd units")
 	// files, _ := utils.ListFiles(constant.UnitsDir + "*.service")
@@ -111,7 +134,6 @@ func (c *Client) updateservice() (err error) {
 	Services := utils.GetServices()
 	if localServices.Len() == 0 {
 		fmt.Println("no local services")
-		return
 	}
 	for _, service := range localServices {
 		if strings.HasSuffix(service, "-data.service") {
@@ -122,8 +144,10 @@ func (c *Client) updateservice() (err error) {
 		if err != nil {
 			return err
 		}
-		time.Sleep(1 * time.Second)
-		err = cmd.Install(deis, []string{localService})
+
+		err = utils.Timeout("Install unit"+service, 300*time.Second, func() {
+			cmd.Install(deis, []string{localService})
+		})
 		if err != nil {
 			return err
 		}
@@ -284,6 +308,7 @@ func (c *Client) SetVersion(resp *omaha.Response) {
 
 // Sleep between n and m seconds
 func (c *Client) Loop(n, m int) {
+	interval := constant.InitialInterval
 	for {
 		randSleep(n, m)
 		resp, err := c.MakeRequest("3", "2", true, false)
@@ -312,7 +337,27 @@ func (c *Client) Loop(n, m int) {
 				continue
 			}
 			c.MakeRequest("14", "1", false, false)
+			c.RequestLock()
+			for {
+				err = c.lock.Lock()
+				if err != nil && err != lock.ErrExist {
+					interval = expBackoff(interval)
+					fmt.Printf("Retrying in %v. Error locking: %v\n", interval, err)
+					time.Sleep(interval)
+					continue
+				} else {
+					break
+				}
+			}
 			c.SetVersion(resp)
+			err = c.lock.Unlock()
+			if err == lock.ErrNotExist {
+				fmt.Println("no lock found")
+			} else if err == nil {
+				fmt.Println("Unlocked existing lock for this machine")
+			} else {
+				fmt.Fprintln(os.Stderr, "Error unlocking:", err)
+			}
 		}
 	}
 }

updatectl/update.go

@@ -3,8 +3,8 @@ package updatectl
 import (
 	"flag"
 	"fmt"
-	"github.com/coreos/updatectl/auth"
-	"github.com/coreos/updatectl/client/update/v1"
+	"github.com/coreos/updateservicectl/auth"
+	"github.com/coreos/updateservicectl/client/update/v1"
 	"log"
 	"net/http"
 	"os"