add start migration

api/grpc/server/server.go

@@ -501,17 +501,17 @@ func (s *apiServer) Stats(ctx context.Context, r *types.StatsRequest) (*types.St
 	return t, nil
 }
 
-func (s *apiServer)Migration(ctx context.Context, r *types.MigrationRequest) (*types.MigrationResponse, error) {
-	e:=&supervisor.MigrationTask{}
+func (s *apiServer) Migration(ctx context.Context, r *types.MigrationRequest) (*types.MigrationResponse, error) {
+	e := &supervisor.MigrationTask{}
 	e.WithContext(ctx)
-	e.Id=r.Id
-	e.TargetMachine= struct {
+	e.Id = r.Id
+	e.TargetMachine = struct {
 		Host string
 		Port uint32
-	}{Host:r.Targetmachine.Ip , Port:r.Targetmachine.Port }
+	}{Host: r.Targetmachine.Ip, Port: r.Targetmachine.Port}
 	s.sv.SendTask(e)
-	if err:=<-e.ErrorCh();err!=nil {
-		return nil,err
+	if err := <-e.ErrorCh(); err != nil {
+		return nil, err
 	}
-	return &types.MigrationResponse{},nil
-}
+	return &types.MigrationResponse{}, nil
+}

containerd-shim/main.go

@@ -9,8 +9,8 @@ import (
 	"runtime"
 	"syscall"
 
-	"github.com/containerd/containerd/osutils"
 	"github.com/containerd/console"
+	"github.com/containerd/containerd/osutils"
 )
 
 func writeMessage(f *os.File, level string, err error) {

containerd-shim/process.go

@@ -15,9 +15,9 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/containerd/console"
 	"github.com/containerd/containerd/osutils"
 	"github.com/containerd/containerd/specs"
-	"github.com/containerd/console"
 	runc "github.com/containerd/go-runc"
 )
 

containerd-shim/process_linux.go

@@ -9,8 +9,8 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/containerd/containerd/osutils"
 	"github.com/containerd/console"
+	"github.com/containerd/containerd/osutils"
 	runc "github.com/containerd/go-runc"
 	"github.com/tonistiigi/fifo"
 	"golang.org/x/net/context"

ctr/checkpoint_linux.go

@@ -5,8 +5,8 @@ import (
 	"os"
 	"text/tabwriter"
 
-	"github.com/urfave/cli"
 	"github.com/containerd/containerd/api/grpc/types"
+	"github.com/urfave/cli"
 	netcontext "golang.org/x/net/context"
 )
 

ctr/events.go

@@ -6,9 +6,9 @@ import (
 	"text/tabwriter"
 	"time"
 
-	"github.com/urfave/cli"
 	"github.com/containerd/containerd/api/grpc/types"
 	"github.com/golang/protobuf/ptypes"
+	"github.com/urfave/cli"
 	netcontext "golang.org/x/net/context"
 )
 

ctr/migration.go

@@ -1,46 +1,46 @@
 package main
 
 import (
-	"github.com/urfave/cli"
 	"fmt"
-	"os"
-	netcontext "golang.org/x/net/context"
 	"github.com/containerd/containerd/api/grpc/types"
+	"github.com/urfave/cli"
+	netcontext "golang.org/x/net/context"
+	"os"
 )
 
-var migrationCommand=cli.Command{
-	Name:"migration",
-	Usage:"migration containers",
-	ArgsUsage:"<container-id> <ip> <port> || mysql 192.168.18.2 9001",
-	Flags:[]cli.Flag{
+var migrationCommand = cli.Command{
+	Name:      "migration",
+	Usage:     "migration containers",
+	ArgsUsage: "<container-id> <ip> <port> || mysql 192.168.18.2 9001",
+	Flags: []cli.Flag{
 		cli.StringFlag{
-			Name:"host,H",
-			Usage:"host ip address",
+			Name:  "host,H",
+			Usage: "host ip address",
 		},
 		cli.UintFlag{
-			Name:"port,p",
-			Usage:"host port",
+			Name:  "port,p",
+			Usage: "host port",
 		},
 	},
 	Action: func(context *cli.Context) error {
-		if err:=checkArgs(context,1);err!=nil {
+		if err := checkArgs(context, 1); err != nil {
 			return err
 		}
 
-		id:=context.Args().First()
-		ip:=context.String("host")
-		port:=context.Uint("port")
-		fmt.Printf("id:%v, ip %v, port:%v\n",id,ip,port)
-		c:=getClient(context)
-		machine:=&types.TargetMachine{
-			Ip:ip,
-			Port:uint32(port),
+		id := context.Args().First()
+		ip := context.String("host")
+		port := context.Uint("port")
+		fmt.Printf("id:%v, ip %v, port:%v\n", id, ip, port)
+		c := getClient(context)
+		machine := &types.TargetMachine{
+			Ip:   ip,
+			Port: uint32(port),
 		}
-		_,err:=c.Migration(netcontext.Background(),&types.MigrationRequest{
-			Id:id,
-			Targetmachine:machine,
+		_, err := c.Migration(netcontext.Background(), &types.MigrationRequest{
+			Id:            id,
+			Targetmachine: machine,
 		})
-		if err!=nil {
+		if err != nil {
 			fmt.Println(err)
 			return err
 		}
@@ -51,12 +51,12 @@ var migrationCommand=cli.Command{
 
 func checkArgs(context *cli.Context, expected int) error {
 	var err error
-	cmdName:=context.Command.Name
-	fmt.Printf("nums is %v\n",context.NArg())
-	if context.NArg()!=expected {
+	cmdName := context.Command.Name
+	fmt.Printf("nums is %v\n", context.NArg())
+	if context.NArg() != expected {
 		err = fmt.Errorf("%s: %q requires exactly %d argument(s)", os.Args[0], cmdName, expected)
 	}
-	if err!=nil {
+	if err != nil {
 		fmt.Printf("Incorrect Usage.\n\n")
 		return err
 	}

supervisor/migration.go

@@ -1,21 +1,21 @@
 package supervisor
 
 import (
-	"time"
-	"github.com/sirupsen/logrus"
 	"errors"
 	"fmt"
 	"github.com/containerd/containerd/runtime"
+	"github.com/containerd/containerd/supervisor/migration"
+	"github.com/sirupsen/logrus"
 	"net"
 	"strings"
-	"github.com/containerd/containerd/supervisor/migration"
+	"time"
+
 )
 
 type MigrationTask struct {
 	baseTask
 	TargetMachine
 	Id string
-
 }
 
 type TargetMachine struct {
@@ -40,90 +40,105 @@ type TargetMachine struct {
 //	CheckpointDir string
 //}
 
-
 func (s *Supervisor) StartMigration(t *MigrationTask) error {
-	startTime:=time.Now()
+	startTime := time.Now()
 	fmt.Println("begin Migration")
-	logrus.Printf("startMigration %v\n",startTime)
-	c,err:=t.checkContainers(s)
-	if err!=nil {
+	logrus.Printf("startMigration %v\n", startTime)
+	c, err := t.checkContainers(s)
+	if err != nil {
 		fmt.Println(err)
 		return err
 	}
 
-	if err=t.checkTargetMachine(s);err!=nil {
+	if err = t.checkTargetMachine(s); err != nil {
 		return err
 	}
 
-	if err=t.startMigration(c,s);err!=nil {
+	if err = t.startMigration(c, s); err != nil {
 		return err
 	}
 	//logrus.Printf("container %v\n",i.container.ID())
 	fmt.Println("migration Finish")
 	return nil
 }
 
-func (t *MigrationTask)checkContainers(s *Supervisor) (*containerInfo,error) {
-	i,ok:=s.containers[t.Id]
+func (t *MigrationTask) checkContainers(s *Supervisor) (*containerInfo, error) {
+	i, ok := s.containers[t.Id]
 	if !ok {
-		return nil,MigrationWriteErr(fmt.Sprintf("Container %v Not Exist\n",t.Id))
+		return nil, MigrationWriteErr(fmt.Sprintf("Container %v Not Exist\n", t.Id))
 	}
-	if i.container.State()!=runtime.Running {
-		return nil,MigrationWriteErr("Container not running")
+	if i.container.State() != runtime.Running {
+		return nil, MigrationWriteErr("Container not running")
 	}
-	return i,nil
+	return i, nil
 }
 
 func (t *MigrationTask) checkTargetMachine(s *Supervisor) error {
-	ip:=t.Host
-	addrs,err:=net.InterfaceAddrs()
-	if err!=nil {
+	ip := t.Host
+	addrs, err := net.InterfaceAddrs()
+	if err != nil {
 		return MigrationWriteErr(err.Error())
 	}
-	for _,addr:=range addrs {
+	for _, addr := range addrs {
 
-		ips:=strings.SplitN(addr.String(),"/",2)
-		fmt.Printf("network:%v,string:%v,splite:%v\n",addr.Network(),addr.String(),ips[0])
-		if ips[0]==ip {
+		ips := strings.SplitN(addr.String(), "/", 2)
+		fmt.Printf("network:%v,string:%v,splite:%v\n", addr.Network(), addr.String(), ips[0])
+		if ips[0] == ip {
 			return MigrationWriteErr("Cannot Migration Localhost Machine")
 		}
 	}
 	return nil
 }
 
 func MigrationWriteErr(w string) error {
-	return errors.New(fmt.Sprintf("miration failed:%v",w))
+	return errors.New(fmt.Sprintf("Miration Failed:%v", w))
 }
 
-func (t *MigrationTask) startCopyImage(c *containerInfo) error  {
-	image,err:=migration.NewImage(c.container)
-	if err!=nil {
+func (t *MigrationTask) startCopyImage(c *containerInfo) error {
+	image, err := migration.NewImage(c.container)
+	if err != nil {
 		return err
 	}
 	image.Path()
 	return nil
 }
 
-func (t *MigrationTask) startMigration(c *containerInfo,s *Supervisor) error {
-	l,err:=migration.NewLocalMigration(c.container)
+func (t *MigrationTask) startMigration(c *containerInfo, s *Supervisor) error {
+	var (
+		e chan error
+		err error
+	)
+
+	l, err := migration.NewLocalMigration(c.container)
+	if err != nil {
+		return MigrationWriteErr(err.Error())
+	}
+	r,err:=migration.NewRemoteMigration(t.Host,t.Id,t.Port)
 	if err!=nil {
-		return err
+		return MigrationWriteErr(err.Error())
 	}
-	if err=l.DoCheckpoint();err!=nil {
+
+	go r.PreLoadImage(e,l.Imagedir)
+
+	if err = l.DoCheckpoint(); err != nil {
 		return err
 	}
-	if err=l.DoneCheckpoint();err!=nil {
+	if err = l.DoneCheckpoint(); err != nil {
 		return err
 	}
 
+	if err=<-e;err!=nil {
+		return MigrationWriteErr(err.Error())
+	}
 	//r,_:=migration.NewRemoteMigration(t,l)
-
-	logrus.Println("begin restore")
+	if err=r.DoRestore();err!=nil {
+		return MigrationWriteErr(err.Error())
+	}
+	logrus.Println("done restore")
 	//go r.Dorestore(s)
 	return nil
 }
 
-
 //
 ////创建本地dump的目录
 //func newLocalMigration(c *containerInfo) (*localMigration, error) {
@@ -192,4 +207,4 @@ func (t *MigrationTask) startMigration(c *containerInfo,s *Supervisor) error {
 //	//fmt.Println(re.Container.Status())
 //	fmt.Println("after restore")
 //	return nil
-//}
+//}