Merge pull request #744 from teawater/buffer

buffer: Handle the control operations (ContainerCreate) asynchronously

Author: teawater

daemon/buffer/buffer.go

@@ -0,0 +1,98 @@
+package buffer
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/golang/glog"
+	"github.com/hyperhq/hyperd/daemon/pod"
+	apitypes "github.com/hyperhq/hyperd/types"
+)
+
+type Buffer struct {
+	goroutinesLimit uint64
+	goroutinesLock  sync.Mutex
+	ch              chan *pod.ContainerBuffer
+}
+
+const (
+	DefaultBufferChannelSize = 1024
+)
+
+func NewBuffer(cfg *apitypes.HyperConfig) *Buffer {
+	if cfg.BufferGoroutinesMax == 0 {
+		return nil
+	}
+	if cfg.BufferChannelSize == 0 {
+		cfg.BufferChannelSize = DefaultBufferChannelSize
+	}
+
+	daemon := &Buffer{
+		goroutinesLimit: cfg.BufferGoroutinesMax,
+		ch:              make(chan *pod.ContainerBuffer, cfg.BufferChannelSize),
+	}
+
+	return daemon
+}
+
+func (b *Buffer) CreateContainerInPod(p *pod.XPod, c *apitypes.UserContainer) (string, error) {
+	if !p.IsAlive() {
+		err := fmt.Errorf("pod is not running")
+		p.Log(pod.ERROR, "%v", err)
+		return "", err
+	}
+	if err := p.ReserveContainerName(c); err != nil {
+		return "", err
+	}
+
+	cb, err := p.AddContainerBuffer(c)
+	if err != nil {
+		return "", err
+	}
+
+	b.goroutinesLock.Lock()
+	defer b.goroutinesLock.Unlock()
+	if b.goroutinesLimit != 0 {
+		b.goroutinesLimit--
+		go b.containerHandler(cb)
+		glog.V(3).Infof("Put %+v to containerHandler, current limit %v", c, b.goroutinesLimit)
+	} else {
+		select {
+		case b.ch <- cb:
+			glog.V(3).Infof("Put %+v to channel", c)
+		default:
+			err := fmt.Errorf("%+v dropped because channel is full", c)
+			glog.Errorf("%s", err)
+			p.RemoveContainerBufferAll(cb)
+			return "", err
+		}
+	}
+
+	return cb.Id, nil
+}
+
+func (b *Buffer) containerHandler(cb *pod.ContainerBuffer) {
+loop:
+	for {
+		glog.V(3).Infof("Buffer begin to handle %+v", cb)
+		id, err := cb.P.DoContainerCreate(cb.Spec, cb.Id)
+		if err == nil {
+			cb.P.RemoveContainerBuffer(cb)
+			glog.V(3).Infof("Buffer handle %+v done, new id is %s", cb, id)
+		} else {
+			cb.P.RemoveContainerBufferAll(cb)
+			glog.Errorf("Buffer handle %+v failed %v", cb, err)
+		}
+
+		b.goroutinesLock.Lock()
+		select {
+		case cb = <-b.ch:
+			b.goroutinesLock.Unlock()
+		default:
+			defer b.goroutinesLock.Unlock()
+			b.goroutinesLimit++
+			break loop
+		}
+	}
+	glog.V(3).Infof("Channel is empty, current limit %v", b.goroutinesLimit)
+}

daemon/daemon.go

@@ -7,6 +7,7 @@ import (
 	"path"
 	"strings"
 
+	"github.com/hyperhq/hyperd/daemon/buffer"
 	"github.com/hyperhq/hyperd/daemon/daemondb"
 	"github.com/hyperhq/hyperd/daemon/pod"
 	"github.com/hyperhq/hyperd/networking/portmapping"
@@ -41,6 +42,8 @@ type Daemon struct {
 	Storage    Storage
 	Hypervisor string
 	DefaultLog *pod.GlobalLogConfig
+
+	buffer *buffer.Buffer
 }
 
 func (daemon *Daemon) Restore() error {
@@ -117,6 +120,7 @@ func NewDaemon(cfg *apitypes.HyperConfig) (*Daemon, error) {
 		db:      db,
 		PodList: pod.NewPodList(),
 		Host:    cfg.Host,
+		buffer:  buffer.NewBuffer(cfg),
 	}
 
 	daemon.Daemon, err = docker.NewDaemon(dockerCfg, registryCfg)

daemon/list.go

@@ -55,6 +55,7 @@ func (daemon *Daemon) ListContainers(podId, vmId string) ([]*apitypes.ContainerL
 				result = append(result, status)
 			}
 		}
+		result = p.AppendContainerBufferStatus(result)
 	}
 	return result, nil
 }
@@ -118,6 +119,7 @@ func (daemon *Daemon) List(item, podId, vmId string) (map[string][]string, error
 				if status != "" {
 					containerJsonResponse = append(containerJsonResponse, status)
 				}
+				containerJsonResponse = p.AppendContainerBufferStatusString(containerJsonResponse)
 			}
 		}
 	}

daemon/pod/buffer.go

@@ -0,0 +1,133 @@
+package pod
+
+import (
+	"github.com/docker/docker/pkg/stringid"
+	apitypes "github.com/hyperhq/hyperd/types"
+	"strings"
+)
+
+type ContainerBuffer struct {
+	Id   string
+	P    *XPod
+	Spec *apitypes.UserContainer
+}
+
+func (cb *ContainerBuffer) info() *apitypes.Container {
+	cinfo := &apitypes.Container{
+		Name:            "/" + cb.Spec.Name,
+		ContainerID:     cb.Id,
+		Image:           cb.Spec.Image,
+		Commands:        cb.Spec.Command,
+		WorkingDir:      cb.Spec.Workdir,
+		Labels:          cb.Spec.Labels,
+		Ports:           make([]*apitypes.ContainerPort, 0, len(cb.Spec.Ports)),
+		VolumeMounts:    make([]*apitypes.VolumeMount, 0, len(cb.Spec.Volumes)),
+		Env:             make([]*apitypes.EnvironmentVar, 0, len(cb.Spec.Envs)),
+		Tty:             cb.Spec.Tty,
+		ImagePullPolicy: "",
+	}
+	for _, port := range cb.Spec.Ports {
+		cinfo.Ports = append(cinfo.Ports, &apitypes.ContainerPort{
+			HostPort:      port.HostPort,
+			ContainerPort: port.ContainerPort,
+			Protocol:      port.Protocol,
+		})
+	}
+	for _, vol := range cb.Spec.Volumes {
+		cinfo.VolumeMounts = append(cinfo.VolumeMounts, &apitypes.VolumeMount{
+			Name:      vol.Volume,
+			MountPath: vol.Path,
+			ReadOnly:  vol.ReadOnly,
+		})
+	}
+	for _, env := range cb.Spec.Envs {
+		cinfo.Env = append(cinfo.Env, &apitypes.EnvironmentVar{
+			Env:   env.Env,
+			Value: env.Value,
+		})
+	}
+	return cinfo
+}
+
+func (cb *ContainerBuffer) infoStatus() *apitypes.ContainerStatus {
+	s := &apitypes.ContainerStatus{
+		Name:        cb.Spec.Name,
+		ContainerID: cb.Id,
+		Waiting:     &apitypes.WaitingStatus{Reason: "Pending"},
+		Running:     &apitypes.RunningStatus{StartedAt: ""},
+		Terminated:  &apitypes.TermStatus{},
+		Phase:       "pending",
+	}
+	return s
+}
+
+func (p *XPod) AddContainerBuffer(c *apitypes.UserContainer) (*ContainerBuffer, error) {
+	cid := stringid.GenerateNonCryptoID()
+
+	cb := &ContainerBuffer{
+		Id:   cid,
+		P:    p,
+		Spec: c,
+	}
+	p.statusLock.Lock()
+	p.containerBuffers[cid] = cb
+	p.statusLock.Unlock()
+
+	err := p.factory.registry.ReserveContainer(cid, c.Name, p.Id())
+	if err != nil {
+		p.RemoveContainerBuffer(cb)
+		return nil, err
+	}
+
+	return cb, nil
+}
+
+func (p *XPod) RemoveContainerBuffer(cb *ContainerBuffer) {
+	p.statusLock.Lock()
+	defer p.statusLock.Unlock()
+	delete(p.containerBuffers, cb.Id)
+}
+
+func (p *XPod) RemoveContainerBufferAll(cb *ContainerBuffer) {
+	p.factory.registry.ReleaseContainer(cb.Id, cb.Spec.Name)
+
+	p.RemoveContainerBuffer(cb)
+}
+
+func (p *XPod) AppendContainerBufferStatus(result []*apitypes.ContainerListResult) []*apitypes.ContainerListResult {
+	p.statusLock.RLock()
+	defer p.statusLock.RUnlock()
+	for id, cb := range p.containerBuffers {
+		result = append(result, &apitypes.ContainerListResult{
+			ContainerID:   id,
+			ContainerName: cb.Spec.Name,
+			PodID:         p.Id(),
+			Status:        "pending",
+		})
+	}
+	return result
+}
+
+func (p *XPod) AppendContainerBufferStatusString(result []string) []string {
+	p.statusLock.RLock()
+	defer p.statusLock.RUnlock()
+	for id, cb := range p.containerBuffers {
+		result = append(result, strings.Join([]string{id, cb.Spec.Name, p.Id(), "pending"}, ":"))
+	}
+	return result
+}
+
+func (p *XPod) ContainerBufferInfo(cid string) *apitypes.ContainerInfo {
+	p.statusLock.RLock()
+	defer p.statusLock.RUnlock()
+	if cb, ok := p.containerBuffers[cid]; ok {
+		//Not set CreatedAt
+		ci := &apitypes.ContainerInfo{
+			PodID:     p.Id(),
+			Container: cb.info(),
+			Status:    cb.infoStatus(),
+		}
+		return ci
+	}
+	return nil
+}

daemon/pod/pod.go

@@ -78,6 +78,8 @@ type XPod struct {
 	// on it too.
 	stoppedChan chan bool
 	initCond    *sync.Cond
+
+	containerBuffers map[string]*ContainerBuffer
 }
 
 // The Log infrastructure, to add pod name as prefix of the log message.
@@ -291,6 +293,12 @@ func (p *XPod) ContainerInfo(cid string) (*apitypes.ContainerInfo, error) {
 		}
 		return ci, nil
 	}
+
+	ci := p.ContainerBufferInfo(cid)
+	if ci != nil {
+		return ci, nil
+	}
+
 	err := fmt.Errorf("container %s does not existing", cid)
 	p.Log(ERROR, err)
 	return nil, err
@@ -343,7 +351,7 @@ func (p *XPod) updatePodInfo() error {
 	defer p.statusLock.RUnlock()
 
 	var (
-		containers      = make([]*apitypes.Container, 0, len(p.containers))
+		containers      = make([]*apitypes.Container, 0, len(p.containers)+len(p.containerBuffers))
 		volumes         = make([]*apitypes.PodVolume, 0, len(p.volumes))
 		containerStatus = make([]*apitypes.ContainerStatus, 0, len(p.containers))
 	)
@@ -365,6 +373,12 @@ func (p *XPod) updatePodInfo() error {
 			succeeeded = "Failed"
 		}
 	}
+	for _, cb := range p.containerBuffers {
+		ci := cb.info()
+		cs := cb.infoStatus()
+		containers = append(containers, ci)
+		containerStatus = append(containerStatus, cs)
+	}
 	p.info.Spec.Containers = containers
 	p.info.Status.ContainerStatus = containerStatus
 

daemon/pod/podlist.go

@@ -71,6 +71,20 @@ func (pl *PodList) ReserveContainer(id, name, pod string) error {
 	return nil
 }
 
+func (pl *PodList) ChangeContainerId(oldId, newId, pod string) error {
+	pl.mu.Lock()
+	defer pl.mu.Unlock()
+	if _, ok := pl.pods[pod]; !ok {
+		return fmt.Errorf("pod %s not exist for Changing container %s", pod, oldId)
+	}
+	if pn, ok := pl.containers[newId]; ok && pn != pod {
+		return fmt.Errorf("the container id %s has already taken by pod %s", newId, pn)
+	}
+	delete(pl.containers, oldId)
+	pl.containers[newId] = pod
+	return nil
+}
+
 func (pl *PodList) ReservePod(p *XPod) error {
 	pl.mu.Lock()
 	defer pl.mu.Unlock()