[refactor] load data from level db.

Signed-off-by: Wang Xu <gnawux@gmail.com>

Author: gnawux

daemon/daemon.go

@@ -1,7 +1,6 @@
 package daemon
 
 import (
-"encoding/json"
 "errors"
 "fmt"
 "os"
@@ -46,46 +45,30 @@ func (daemon *Daemon) Restore() error {
 return nil
 }
 
-ch := daemon.db.GetAllPods()
+ch := pod.LoadAllPods(daemon.db)
 if ch == nil {
 estr := "Cannot list pods in leveldb"
 glog.Error(estr)
 return errors.New(estr)
 }
 
 for {
-item, ok := <-ch
+layout, ok := <-ch
 if !ok {
 break
 }
-if item == nil {
+if layout == nil {
 estr := "error during load pods from leveldb"
 glog.Error(estr)
 return errors.New(estr)
 }
 
-podId := string(item.K[4:])
-
-glog.V(1).Infof("reloading pod %s with args %s", podId, string(item.V))
-
-daemon.db.DeletePod(podId)
-
-var podSpec apitypes.UserPod
-err := json.Unmarshal(item.V, &podSpec)
-if err != nil {
-return err
-}
-
-vmId, err := daemon.db.GetP2V(podId)
-if err != nil {
-glog.V(1).Infof("no existing VM for pod %s: %v", podId, err)
-}
-
+glog.V(1).Infof("reloading pod %s: %#v", layout.Id, layout)
 fc := pod.NewPodFactory(daemon.Factory, daemon.PodList, daemon.db, daemon.Storage, daemon.Daemon, daemon.DefaultLog)
 
-p, err := pod.LoadXPod(fc, &podSpec, vmId)
+p, err := pod.LoadXPod(fc, layout)
 if err != nil {
-glog.Warningf("Got a unexpected error when creating(load) pod %s, %v", podId, err)
+glog.Warningf("Got a unexpected error when creating(load) pod %s, %v", layout.Id, err)
 continue
 }
 

daemon/daemondb/daemondb.go

@@ -227,7 +227,11 @@ func (d *DaemonDB) PrefixList2Chan(prefix []byte, keyFilter KeyFilter) chan *KVP
 for iter.Next() {
 glog.V(3).Infof("got key from leveldb %s", string(iter.Key()))
 if keyFilter == nil || keyFilter(iter.Key()) {
-ch <- &KVPair{append([]byte{}, iter.Key()...), append([]byte{}, iter.Value()...)}
+k := make([]byte, 0, len(iter.Key()))
+v := make([]byte, 0, len(iter.Value()))
+copy(k, iter.Key())
+copy(v, iter.Value())
+ch <- &KVPair{k, v}
 }
 }
 iter.Release()

daemon/pod/persist.go

@@ -4,8 +4,9 @@ import (
 "fmt"
 
 "github.com/golang/protobuf/proto"
-"github.com/hyperhq/hyperd/types"
 "github.com/hyperhq/hypercontainer-utils/hlog"
+"github.com/hyperhq/hyperd/daemon/daemondb"
+"github.com/hyperhq/hyperd/types"
 )
 
 /// Layout of Persistent Info of a Pod:
@@ -25,19 +26,105 @@ import (
 
 const (
 LAYOUT_KEY_PREFIX = "PL-"
-LAYOUT_KEY_FMT = "PL-%s"
-SB_KEY_FMT = "SB-%s"
-PS_KEY_FMT = "PS-%s"
-PM_KEY_FMT = "PM-%s"
-CX_KEY_FMT = "CX-%s"
-VX_KEY_FMT = "VX-%s-%s"
-IF_KEY_FMT = "IF-%s-%s"
+LAYOUT_KEY_FMT  = "PL-%s"
+SB_KEY_FMT  = "SB-%s"
+PS_KEY_FMT  = "PS-%s"
+PM_KEY_FMT  = "PM-%s"
+CX_KEY_FMT  = "CX-%s"
+VX_KEY_FMT  = "VX-%s-%s"
+IF_KEY_FMT  = "IF-%s-%s"
 )
 
+func LoadAllPods(db *daemondb.DaemonDB) chan *types.PersistPodLayout {
+kvchan := db.PrefixList2Chan([]byte(LAYOUT_KEY_PREFIX), nil)
+if kvchan == nil {
+return nil
+}
+ch := make(chan *types.PersistPodLayout, 128)
+go func() {
+for {
+kv, ok := <-kvchan
+if !ok {
+hlog.Log(INFO, "layout loading finished")
+close(ch)
+return
+}
+hlog.Log(TRACE, "loading layout of container %s", string(kv.K))
+
+var layout types.PersistPodLayout
+err := proto.Unmarshal(kv.V, &layout)
+if err != nil {
+hlog.Log(ERROR, "failed to decode layout of contaienr %s: %v", string(kv.K), err)
+continue
+}
+ch <- &layout
+}
+}()
+return ch
+}
+
+func LoadXPod(factory *PodFactory, layout *types.PersistPodLayout) (*XPod, error) {
+spec, err := loadGloabalSpec(factory.db, layout.Id)
+if err != nil {
+return nil, err
+}
+
+p, err := newXPod(factory, spec)
+if err != nil {
+hlog.Log(ERROR, "failed to create pod from spec: %v", err)
+//remove spec from daemonDB
+//remove vm from daemonDB
+return nil, err
+}
+err = p.reserveNames(spec.Containers)
+if err != nil {
+return nil, err
+}
+
+for _, ix := range layout.Interfaces {
+if err := p.loadInterface(ix); err != nil {
+return nil, err
+}
+}
+
+for _, vid := range layout.Volumes {
+if err := p.loadVolume(vid); err != nil {
+return nil, err
+}
+}
+
+for _, cid := range layout.Containers {
+if err := p.loadContainer(cid); err != nil {
+return nil, err
+}
+}
+
+err = p.loadSandbox()
+if err != nil {
+//remove vm from daemonDB
+return nil, err
+}
+
+err = p.loadPodMeta()
+if err != nil {
+return nil, err
+}
+
+//resume logging
+if p.status == S_POD_RUNNING {
+for _, c := range p.containers {
+c.startLogging()
+}
+}
+
+// don't need to reserve name again, because this is load
+return p, nil
+}
+
 func (p *XPod) savePod() error {
 var (
 containers = make([]string, 0, len(p.containers))
-volumes = make([]string, 0, len(p.volumes))
+volumes  = make([]string, 0, len(p.volumes))
 interfaces = make([]string, 0, len(p.interfaces))
 )
 
@@ -63,76 +150,171 @@ func (p *XPod) savePod() error {
 }
 }
 
-for inf, i := range  p.interfaces {
+for inf, i := range p.interfaces {
 interfaces = append(interfaces, inf)
 if err := i.saveInterface(); err != nil {
 return err
 }
 }
 
 pl := &types.PersistPodLayout{
-Id: p.Id(),
+Id:  p.Id(),
 Containers: containers,
-Volumes: volumes,
+Volumes:  volumes,
 Interfaces: interfaces,
 }
-return p.saveMessage(fmt.Sprintf(LAYOUT_KEY_FMT, p.Id()), pl, p, "pod layout")
+return saveMessage(p.factory.db, fmt.Sprintf(LAYOUT_KEY_FMT, p.Id()), pl, p, "pod layout")
 }
 
 func (p *XPod) saveGlobalSpec() error {
-return p.saveMessage(fmt.Sprintf(PS_KEY_FMT, p.Id()), p.globalSpec, p, "global spec")
+return saveMessage(p.factory.db, fmt.Sprintf(PS_KEY_FMT, p.Id()), p.globalSpec, p, "global spec")
+}
+
+func loadGloabalSpec(db *daemondb.DaemonDB, id string) (*types.UserPod, error) {
+var spec types.UserPod
+err := loadMessage(db, fmt.Sprintf(LAYOUT_KEY_FMT, id), &spec, nil, fmt.Sprintf("spec for %s", id))
+if err != nil {
+return nil, err
+}
+return &spec, nil
 }
 
 func (p *XPod) savePodMeta() error {
 meta := &types.PersistPodMeta{
-Id: p.Id(),
+Id:  p.Id(),
 Services: p.services,
-Labels: p.labels,
+Labels: p.labels,
+}
+if p.info != nil {
+meta.CreatedAt = p.info.CreatedAt
+}
+return saveMessage(p.factory.db, fmt.Sprintf(PM_KEY_FMT, p.Id()), meta, p, "pod meta")
+}
+
+func (p *XPod) loadPodMeta() error {
+var meta types.PersistPodMeta
+err := loadMessage(p.factory.db, fmt.Sprintf(PM_KEY_FMT, p.Id()), &meta, p, "pod meta")
+if err != nil {
+return err
+}
+p.initPodInfo()
+if meta.CreatedAt > 0 {
+p.info.CreatedAt = meta.CreatedAt
 }
-return p.saveMessage(fmt.Sprintf(PM_KEY_FMT, p.Id()), meta, p, "pod meta")
+p.labels = meta.Labels
+p.services = meta.Services
+return nil
 }
 
 func (c *Container) saveContainer() error {
 cx := &types.PersistContainer{
-Id: c.Id(),
-Pod: c.p.Id(),
-Spec: c.spec,
+Id:  c.Id(),
+Pod:  c.p.Id(),
+Spec:  c.spec,
 Descript: c.descript,
 }
-return c.p.saveMessage(fmt.Sprintf(CX_KEY_FMT, c.Id()), cx, c, "container info")
+return saveMessage(c.p.factory.db, fmt.Sprintf(CX_KEY_FMT, c.Id()), cx, c, "container info")
+}
+
+func (p *XPod) loadContainer(id string) error {
+var cx types.PersistContainer
+err := loadMessage(p.factory.db, fmt.Sprintf(CX_KEY_FMT, id), &cx, p, fmt.Sprintf("container info of %s", id))
+if err != nil {
+return err
+}
+c, err := newContainer(p, cx.Spec, false)
+if err != nil {
+p.Log(ERROR, "failed to reload container %s from spec: %v", id, err)
+return err
+}
+err = p.factory.registry.ReserveContainer(c.Id(), c.SpecName(), p.Id())
+if err != nil {
+p.Log(ERROR, "failed to register name of container %s (%s) during load", c.Id(), c.SpecName(), err)
+return err
+}
+p.containers[c.Id()] = c
+return nil
 }
 
 func (v *Volume) saveVolume() error {
 vx := &types.PersistVolume{
-Name: v.spec.Name,
-Pod: v.p.Id(),
-Spec: v.spec,
+Name:  v.spec.Name,
+Pod:  v.p.Id(),
+Spec:  v.spec,
 Descript: v.descript,
 }
-return v.p.saveMessage(fmt.Sprintf(VX_KEY_FMT, v.p.Id(), v.spec.Name), vx, v, "volume info")
+return saveMessage(v.p.factory.db, fmt.Sprintf(VX_KEY_FMT, v.p.Id(), v.spec.Name), vx, v, "volume info")
+}
+
+func (p *XPod) loadVolume(id string) error {
+var vx types.PersistVolume
+err := loadMessage(p.factory.db, fmt.Sprintf(VX_KEY_FMT, p.Id(), id), &vx, p, fmt.Sprintf("volume info of %s", id))
+if err != nil {
+return err
+}
+v := newVolume(p, vx.Spec)
+v.descript = vx.Descript
+v.status = S_VOLUME_CREATED
+p.volumes[v.spec.Name] = v
+return nil
 }
 
 func (inf *Interface) saveInterface() error {
 ix := &types.PersistInterface{
-Id: inf.descript.Id,
-Pod: inf.p.Id(),
-Spec: inf.spec,
+Id:  inf.descript.Id,
+Pod:  inf.p.Id(),
+Spec:  inf.spec,
 Descript: inf.descript,
 }
-return inf.p.saveMessage(fmt.Sprintf(IF_KEY_FMT, inf.p.Id(), inf.descript.Id), ix, inf, "interface info")
+return saveMessage(inf.p.factory.db, fmt.Sprintf(IF_KEY_FMT, inf.p.Id(), inf.descript.Id), ix, inf, "interface info")
 }
 
-func (p *XPod) saveMessage(key string, message proto.Message, owner hlog.LogOwner, op string) error {
+func (p *XPod) loadInterface(id string) error {
+var ix types.PersistInterface
+err := loadMessage(p.factory.db, fmt.Sprintf(IF_KEY_FMT, p.Id(), id), &ix, p, fmt.Sprintf("inf info of %s", id))
+if err != nil {
+return err
+}
+inf := newInterface(p, ix.Spec)
+inf.descript = ix.Descript
+p.interfaces[inf.descript.Id] = inf
+return nil
+}
+
+func (p *XPod) loadSandbox() error {
+var sb types.SandboxPersistInfo
+err := loadMessage(p.factory.db, fmt.Sprintf(SB_KEY_FMT, p.Id()), &sb, p, "load sandbox info")
+if err != nil {
+return err
+}
+return p.reconnectSandbox(sb.Id, sb.PersistInfo)
+}
+
+func saveMessage(db *daemondb.DaemonDB, key string, message proto.Message, owner hlog.LogOwner, op string) error {
 pm, err := proto.Marshal(message)
 if err != nil {
 hlog.HLog(ERROR, owner, 2, "failed to serialize %s: %v", op, err)
 return err
 }
-err = p.factory.db.Update([]byte(key),pm)
+err = db.Update([]byte(key), pm)
 if err != nil {
 hlog.HLog(ERROR, owner, 2, "failed to write %s to db: %v", op, err)
 return err
 }
 hlog.HLog(DEBUG, owner, 2, "%s serialized to db", op)
 return nil
-}
+}
+
+func loadMessage(db *daemondb.DaemonDB, key string, message proto.Message, owner hlog.LogOwner, op string) error {
+v, err := db.Get([]byte(key))
+if err != nil {
+hlog.HLog(ERROR, owner, 2, "failed to load %s: %v", op, err)
+return err
+}
+err = proto.Unmarshal(v, message)
+if err != nil {
+hlog.HLog(ERROR, owner, 2, "failed to unpack loaded %s: %v", op, err)
+return err
+}
+return nil
+}