support send large file to workload

cluster/calcium/sendlarge.go

@@ -0,0 +1,74 @@
+package calcium
+
+import (
+	"context"
+	"github.com/pkg/errors"
+	"github.com/projecteru2/core/engine"
+	"github.com/projecteru2/core/log"
+	"github.com/projecteru2/core/types"
+	"github.com/projecteru2/core/utils"
+	"github.com/sirupsen/logrus"
+	"io"
+	"sync"
+)
+
+// SendLargeFile send large files by stream to workload
+func (c *Calcium) SendLargeFile(ctx context.Context, opts chan *types.SendLargeFileOptions, resp chan *types.SendMessage) error {
+	logger := log.WithField("Calcium", "SendLargeFile").WithField("opts", opts)
+	utils.SentryGo(func() {
+		defer close(resp)
+		wg := &sync.WaitGroup{}
+
+		writerMap := make(map[string]map[string]*io.PipeWriter)
+		for data := range opts {
+			logrus.Debugf("[SendLargeFile] receive some msg, len: %d", len(data.Data))
+			if err := data.Validate(); err != nil {
+				continue
+			}
+
+			dst := data.FileMetadataOptions.Dst
+			for _, id := range data.FileMetadataOptions.Ids {
+				if _, ok := writerMap[id]; !ok {
+					logrus.Debugf("[SendLargeFile] gen writerMap[%s]", id)
+					writerMap[id] = make(map[string]*io.PipeWriter)
+				}
+				if _, ok := writerMap[id][dst]; !ok {
+					wg.Add(1)
+					pr, pw := io.Pipe()
+					writerMap[id][dst] = pw
+					utils.SentryGo(func(id string) func() {
+						return func() {
+							defer wg.Done()
+							logrus.Debugf("[SendLargeFile] gen goroutine for id:%s", id)
+							if err := c.withWorkloadLocked(ctx, id, func(ctx context.Context, workload *types.Workload) error {
+								err := c.doSendChunkFileToWorkload(ctx, workload.Engine, workload.ID, dst, pr, data.Uid, data.Gid, data.Mode)
+								resp <- &types.SendMessage{ID: id, Path: dst, Error: logger.Err(ctx, err)}
+								return nil
+							}); err != nil {
+								resp <- &types.SendMessage{ID: id, Error: logger.Err(ctx, err)}
+							}
+						}
+					}(id))
+				}
+				logrus.Debugf("[SendLargeFile] send somethine id: %s; dst: %s", id, dst)
+				writerMap[id][dst].Write(data.Data)
+			}
+		}
+		logrus.Debugf("[SendLargeFile] send to the end")
+		for _, m := range writerMap {
+			for _, w := range m{
+				err := w.Close()
+				if err != nil {
+					logrus.Debugf("[SendLargeFile] close writer error: %v", err)
+				}
+			}
+		}
+		wg.Wait()
+	})
+	return nil
+}
+
+func (c *Calcium) doSendChunkFileToWorkload(ctx context.Context, engine engine.API, ID, name string, content io.Reader, uid, gid int, mode int64) error {
+	log.Infof(ctx, "[doSendChunkFileToWorkload] Send file to %s:%s", ID, name)
+	return errors.WithStack(engine.VirtualizationCopyChunkTo(ctx, ID, name, content, uid, gid, mode))
+}

cluster/cluster.go

@@ -73,6 +73,7 @@ type Cluster interface {
 	// file methods
 	Copy(ctx context.Context, opts *types.CopyOptions) (chan *types.CopyMessage, error)
 	Send(ctx context.Context, opts *types.SendOptions) (chan *types.SendMessage, error)
+	SendLargeFile(ctx context.Context, opts chan *types.SendLargeFileOptions, resp chan *types.SendMessage) error
 	// image methods
 	BuildImage(ctx context.Context, opts *types.BuildOptions) (chan *types.BuildImageMessage, error)
 	CacheImage(ctx context.Context, opts *types.ImageOptions) (chan *types.CacheImageMessage, error)

engine/docker/container.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/sirupsen/logrus"
 	"io"
 	"io/ioutil"
 	"math"
@@ -331,6 +332,64 @@ func (e *Engine) VirtualizationCopyTo(ctx context.Context, ID, target string, co
 	})
 }
 
+// VirtualizationCopyChunkTo copy chunk to virtualization
+func (e *Engine) VirtualizationCopyChunkTo(ctx context.Context, ID, target string, content io.Reader, uid, gid int, mode int64) error {
+	// todo err 怎么抛出
+	pr, pw := io.Pipe()
+	//bf := new(bytes.Buffer)
+	tw := tar.NewWriter(pw)
+	defer tw.Close()
+	utils.SentryGo(func(writer io.Writer, reader io.Reader) func() {
+		return func() {
+			hdr := &tar.Header{
+				Name: filepath.Base(target),
+				// todo client 传过来
+				Size: 26357760,
+				Mode: mode,
+				Uid:  uid,
+				Gid:  gid,
+			}
+			logrus.Debugf("[VirtualizationCopyChunkTo] write header, target: %s, id: %s", target, ID)
+			if err := tw.WriteHeader(hdr); err != nil {
+				logrus.Debugf("[VirtualizationCopyChunkTo] write header err, err: %v", err)
+				return
+			}
+			logrus.Debugf("[VirtualizationCopyChunkTo] write header done, target: %s, id: %s", target, ID)
+			for {
+				data := make([]byte, 10 * 1024 * 1024)
+				logrus.Debugf("try to read.....")
+				n, err := reader.Read(data)
+				logrus.Debugf("[VirtualizationCopyChunkTo] read something, len: %d", n)
+				if err != nil {
+					if err == io.EOF {
+						logrus.Debugf("[VirtualizationCopyChunkTo] end!!, target: %s, id: %s", target, ID)
+					}
+					logrus.Debugf("[VirtualizationCopyChunkTo] read something error, len: %d", n)
+					err := writer.(*io.PipeWriter).Close()
+					if err != nil {
+						logrus.Debugf("[VirtualizationCopyChunkTo] close pipe writer, err: %v", err)
+					}
+					return
+				}
+				if n < len(data) {
+					data = data[:n]
+				}
+				_, err = tw.Write(data)
+				logrus.Debugf("[VirtualizationCopyChunkTo] write something, len: %d", len(data))
+				if err != nil {
+					logrus.Debugf("[VirtualizationCopyChunkTo] write something err, err: %v", err)
+					err := writer.(*io.PipeWriter).Close()
+					if err != nil {
+						logrus.Debugf("[VirtualizationCopyChunkTo] close pipe writer, err: %v", err)
+					}
+					return
+				}
+			}
+		}
+	}(pw, content))
+	return e.client.CopyToContainer(ctx, ID, filepath.Dir(target), pr, dockertypes.CopyToContainerOptions{AllowOverwriteDirWithFile: true, CopyUIDGID: false})
+}
+
 // VirtualizationStart start virtualization
 func (e *Engine) VirtualizationStart(ctx context.Context, ID string) error {
 	return e.client.ContainerStart(ctx, ID, dockertypes.ContainerStartOptions{})

engine/engine.go

@@ -39,6 +39,7 @@ type API interface {
 	VirtualizationCreate(ctx context.Context, opts *enginetypes.VirtualizationCreateOptions) (*enginetypes.VirtualizationCreated, error)
 	VirtualizationResourceRemap(context.Context, *enginetypes.VirtualizationRemapOptions) (<-chan enginetypes.VirtualizationRemapMessage, error)
 	VirtualizationCopyTo(ctx context.Context, ID, target string, content []byte, uid, gid int, mode int64) error
+	VirtualizationCopyChunkTo(ctx context.Context, ID, target string, content io.Reader, uid, gid int, mode int64) error
 	VirtualizationStart(ctx context.Context, ID string) error
 	VirtualizationStop(ctx context.Context, ID string, gracefulTimeout time.Duration) error
 	VirtualizationRemove(ctx context.Context, ID string, volumes, force bool) error

engine/fake/fake.go

@@ -128,6 +128,11 @@ func (f *Engine) VirtualizationCopyTo(ctx context.Context, ID, target string, co
 	return types.ErrNilEngine
 }
 
+// VirtualizationCopyChunkTo .
+func (f *Engine) VirtualizationCopyChunkTo(ctx context.Context, ID, target string, content io.Reader, uid, gid int, mode int64) error {
+	return types.ErrNilEngine
+}
+
 // VirtualizationStart .
 func (f *Engine) VirtualizationStart(ctx context.Context, ID string) error {
 	return types.ErrNilEngine

engine/mocks/fakeengine/mock.go

@@ -95,6 +95,7 @@ func MakeClient(ctx context.Context, config coretypes.Config, nodename, endpoint
 	close(ch)
 	e.On("VirtualizationResourceRemap", mock.Anything, mock.Anything).Return((<-chan enginetypes.VirtualizationRemapMessage)(ch), nil)
 	e.On("VirtualizationCopyTo", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	e.On("VVirtualizationCopyChunkTo", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	e.On("VirtualizationStart", mock.Anything, mock.Anything).Return(nil)
 	e.On("VirtualizationStop", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	e.On("VirtualizationRemove", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)

engine/types/virtualization.go

@@ -82,3 +82,10 @@ type VirtualizationRemapMessage struct {
 	ID    string
 	Error error
 }
+
+// SendMessage returns from engine
+type SendMessage struct {
+	ID    string
+	Path  string
+	Error error
+}

engine/virt/virt.go

@@ -214,6 +214,11 @@ func (v *Virt) VirtualizationCopyTo(ctx context.Context, ID, dest string, conten
 	return v.client.CopyToGuest(ctx, ID, dest, bytes.NewReader(content), true, true)
 }
 
+// VirtualizationCopyChunkTo copies one.
+func (v *Virt) VirtualizationCopyChunkTo(ctx context.Context, ID, dest string, content io.Reader, uid, gid int, mode int64) error {
+	return v.client.CopyToGuest(ctx, ID, dest, content, true, true)
+}
+
 // VirtualizationStart boots a guest.
 func (v *Virt) VirtualizationStart(ctx context.Context, ID string) (err error) {
 	_, err = v.client.StartGuest(ctx, ID)

rpc/errors.go

@@ -61,6 +61,8 @@ const (
 	Copy codes.Code = 1061
 	// Send .
 	Send codes.Code = 1062
+	// SendLargeFile .
+	SendLargeFile codes.Code = 1063
 
 	// BuildImage .
 	BuildImage codes.Code = 1071