introduce external package

Signed-off-by: Yan Song <[email protected]>

Author: imeoer

go.mod

@@ -46,6 +46,7 @@ require (
 	github.com/sirupsen/logrus v1.9.3
 	github.com/stretchr/testify v1.10.0
 	github.com/urfave/cli/v2 v2.27.5
+	github.com/vmihailenco/msgpack/v5 v5.4.1
 	go.etcd.io/bbolt v1.3.11
 	golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56
 	golang.org/x/net v0.30.0
@@ -61,6 +62,8 @@ require (
 	k8s.io/utils v0.0.0-20240711033017-18e509b52bc8
 )
 
+require github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
+
 require (
 	github.com/AdamKorcz/go-118-fuzz-build v0.0.0-20231105174938-2b5cbb29f3e2 // indirect
 	github.com/Microsoft/go-winio v0.6.2 // indirect

go.sum

@@ -319,6 +319,10 @@ github.com/urfave/cli/v2 v2.27.5 h1:WoHEJLdsXr6dDWoJgMq/CboDmyY/8HMMH1fTECbih+w=
 github.com/urfave/cli/v2 v2.27.5/go.mod h1:3Sevf16NykTbInEnD0yKkjDAeZDS0A6bzhBH5hrMvTQ=
 github.com/vbatts/tar-split v0.11.5 h1:3bHCTIheBm1qFTcgh9oPu+nNBtX+XJIupG/vacinCts=
 github.com/vbatts/tar-split v0.11.5/go.mod h1:yZbwRsSeGjusneWgA781EKej9HF8vme8okylkAeNKLk=
+github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IUPn0Bjt8=
+github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok=
+github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g=
+github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds=
 github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
 github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
 github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=

pkg/converter/convert_unix.go

@@ -360,10 +360,6 @@ func Pack(ctx context.Context, dest io.Writer, opt PackOption) (io.WriteCloser,
 }
 
 func packFromDirectory(ctx context.Context, dest io.Writer, opt PackOption, builderPath, sourceDir string) error {
-	if opt.ExternalBlobWriter == nil {
-		return fmt.Errorf("the 'ExternalBlobWriter' option requires the 'AttributesPath' option be specified")
-	}
-
 	workDir, err := ensureWorkDir(opt.WorkDir)
 	if err != nil {
 		return errors.Wrap(err, "ensure work directory")
@@ -381,12 +377,17 @@ func packFromDirectory(ctx context.Context, dest io.Writer, opt PackOption, buil
 	}
 	defer blobFifo.Close()
 
-	externalBlobPath := filepath.Join(workDir, "external-blob")
-	externalBlobFifo, err := fifo.OpenFifo(ctx, externalBlobPath, syscall.O_CREAT|syscall.O_RDONLY|syscall.O_NONBLOCK, 0640)
-	if err != nil {
-		return errors.Wrapf(err, "create fifo file for external blob")
+	externalBlobPath := ""
+	var externalBlobFifo io.ReadWriteCloser
+	if opt.ExternalBlobWriter != nil {
+		var err error
+		externalBlobPath = filepath.Join(workDir, "external-blob")
+		externalBlobFifo, err = fifo.OpenFifo(ctx, externalBlobPath, syscall.O_CREAT|syscall.O_RDONLY|syscall.O_NONBLOCK, 0640)
+		if err != nil {
+			return errors.Wrapf(err, "create fifo file for external blob")
+		}
+		defer externalBlobFifo.Close()
 	}
-	defer externalBlobFifo.Close()
 
 	go func() {
 		err := tool.Pack(tool.PackOption{
@@ -422,14 +423,17 @@ func packFromDirectory(ctx context.Context, dest io.Writer, opt PackOption, buil
 		}
 		return nil
 	})
-	eg.Go(func() error {
-		buffer := bufPool.Get().(*[]byte)
-		defer bufPool.Put(buffer)
-		if _, err := io.CopyBuffer(opt.ExternalBlobWriter, externalBlobFifo, *buffer); err != nil {
-			return errors.Wrap(err, "pack to nydus external blob")
-		}
-		return nil
-	})
+
+	if opt.ExternalBlobWriter != nil {
+		eg.Go(func() error {
+			buffer := bufPool.Get().(*[]byte)
+			defer bufPool.Put(buffer)
+			if _, err := io.CopyBuffer(opt.ExternalBlobWriter, externalBlobFifo, *buffer); err != nil {
+				return errors.Wrap(err, "pack to nydus external blob")
+			}
+			return nil
+		})
+	}
 
 	return eg.Wait()
 }

pkg/converter/tool/builder.go

@@ -76,7 +76,7 @@ type outputJSON struct {
 	Blobs []string
 }
 
-func buildPackArgs(option PackOption) []string {
+func buildPackArgs(option PackOption) ([]string, error) {
 	if option.FsVersion == "" {
 		option.FsVersion = "6"
 	}
@@ -100,7 +100,11 @@ func buildPackArgs(option PackOption) []string {
 			args,
 			"--blob-inline-meta",
 		)
-		if option.AttributesPath != "" {
+		info, err := os.Stat(option.SourcePath)
+		if err != nil {
+			return nil, err
+		}
+		if info.IsDir() {
 			args = append(
 				args,
 				"--type",
@@ -160,7 +164,7 @@ func buildPackArgs(option PackOption) []string {
 	}
 	args = append(args, option.SourcePath)
 
-	return args
+	return args, nil
 }
 
 func Pack(option PackOption) error {
@@ -175,7 +179,10 @@ func Pack(option PackOption) error {
 		defer cancel()
 	}
 
-	args := buildPackArgs(option)
+	args, err := buildPackArgs(option)
+	if err != nil {
+		return err
+	}
 	logrus.Debugf("\tCommand: %s %s", option.BuilderPath, strings.Join(args, " "))
 
 	cmd := exec.CommandContext(ctx, option.BuilderPath, args...)

pkg/external/backend/backend.go

@@ -0,0 +1,59 @@
+package backend
+
+import "context"
+
+const (
+	DefaultFileChunkSize    = 1024 * 1024 * 1 // 1 MB
+	DefaultThrottleFileSize = 1024 * 1024 * 2 // 2 MB
+)
+
+type Backend struct {
+	Type   string            `json:"type"`
+	Config map[string]string `json:"config"`
+}
+
+type Result struct {
+	Chunks  []Chunk
+	Files   []string
+	Backend Backend
+}
+
+type File struct {
+	RelativePath string
+	Size         int64
+}
+
+// Handler is the interface for backend handler.
+type Handler interface {
+	// Backend returns the backend information.
+	Backend(ctx context.Context) (*Backend, error)
+	// Handle handles the file and returns the object information.
+	Handle(ctx context.Context, file File) ([]Chunk, error)
+}
+
+type Chunk interface {
+	ObjectID() uint32
+	ObjectContent() interface{}
+	ObjectOffset() uint64
+}
+
+// SplitObjectOffsets splits the total size into object offsets
+// with the specified chunk size.
+func SplitObjectOffsets(totalSize, chunkSize int64) []uint64 {
+	objectOffsets := []uint64{}
+	if chunkSize <= 0 {
+		return objectOffsets
+	}
+
+	chunkN := totalSize / chunkSize
+
+	for i := int64(0); i < chunkN; i++ {
+		objectOffsets = append(objectOffsets, uint64(i*chunkSize))
+	}
+
+	if totalSize%chunkSize > 0 {
+		objectOffsets = append(objectOffsets, uint64(chunkN*chunkSize))
+	}
+
+	return objectOffsets
+}

pkg/external/backend/backend_test.go

@@ -0,0 +1,15 @@
+package backend
+
+import (
+	"fmt"
+	"testing"
+	"unsafe"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestLayout(t *testing.T) {
+	require.Equal(t, fmt.Sprintf("%d", 4096), fmt.Sprintf("%d", unsafe.Sizeof(Header{})))
+	require.Equal(t, fmt.Sprintf("%d", 256), fmt.Sprintf("%d", unsafe.Sizeof(ChunkMeta{})))
+	require.Equal(t, fmt.Sprintf("%d", 256), fmt.Sprintf("%d", unsafe.Sizeof(ObjectMeta{})))
+}

pkg/external/backend/layout.go

@@ -0,0 +1,55 @@
+package backend
+
+const MetaMagic uint32 = 0x0AF5_E1E2
+const MetaVersion uint32 = 0x0000_0001
+
+// Layout
+//
+// header: magic | version | chunk_meta_offset | object_meta_offset
+// chunks: chunk_meta | chunk | chunk | ...
+// objects: object_meta | [object_offsets] | object | object | ...
+
+// 4096 bytes
+type Header struct {
+	Magic   uint32
+	Version uint32
+
+	ChunkMetaOffset  uint32
+	ObjectMetaOffset uint32
+
+	Reserved2 [4080]byte
+}
+
+// 256 bytes
+type ChunkMeta struct {
+	EntryCount uint32
+	EntrySize  uint32
+
+	Reserved [248]byte
+}
+
+// 256 bytes
+type ObjectMeta struct {
+	EntryCount uint32
+	// = 0 means indeterminate entry size, and len(object_offsets) > 0.
+	// > 0 means fixed entry size, and len(object_offsets) == 0.
+	EntrySize uint32
+
+	Reserved [248]byte
+}
+
+// 8 bytes
+type ChunkOndisk struct {
+	ObjectIndex  uint32
+	Reserved     [4]byte
+	ObjectOffset uint64
+}
+
+// 4 bytes
+type ObjectOffset uint32
+
+// Size depends on different external backend implementations
+type ObjectOndisk struct {
+	EntrySize   uint32
+	EncodedData []byte
+}

pkg/external/backend/local/local.go

@@ -0,0 +1,68 @@
+package local
+
+import (
+	"context"
+
+	"github.com/containerd/nydus-snapshotter/pkg/external/backend"
+)
+
+type object struct {
+	Path string `msgpack:"p"`
+}
+
+type chunk struct {
+	objectID      uint32
+	objectContent object
+	objectOffset  uint64
+}
+
+func (c *chunk) ObjectID() uint32 {
+	return c.objectID
+}
+
+func (c *chunk) ObjectContent() interface{} {
+	return c.objectContent
+}
+
+func (c *chunk) ObjectOffset() uint64 {
+	return c.objectOffset
+}
+
+type Handler struct {
+	root     string
+	objectID uint32
+}
+
+func NewHandler(root string) *Handler {
+	return &Handler{
+		root:     root,
+		objectID: 0,
+	}
+}
+
+func (handler *Handler) Handle(_ context.Context, file backend.File) ([]backend.Chunk, error) {
+	chunks := []backend.Chunk{}
+	objectOffsets := backend.SplitObjectOffsets(file.Size, backend.DefaultFileChunkSize)
+
+	for _, objectOffset := range objectOffsets {
+		chunks = append(chunks, &chunk{
+			objectID: handler.objectID,
+			objectContent: object{
+				Path: file.RelativePath,
+			},
+			objectOffset: objectOffset,
+		})
+	}
+	handler.objectID++
+
+	return chunks, nil
+}
+
+func (handler *Handler) Backend(_ context.Context) (*backend.Backend, error) {
+	return &backend.Backend{
+		Type: "local",
+		Config: map[string]string{
+			"root": handler.root,
+		},
+	}, nil
+}

pkg/external/backend/local/local_test.go

@@ -0,0 +1,43 @@
+package local
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/vmihailenco/msgpack/v5"
+)
+
+type Object2 struct {
+	Kind   string `msgpack:"k"`
+	Number uint64 `msgpack:"n"`
+	Path   string `msgpack:"p"`
+}
+
+func TestSerializeCompatibility(t *testing.T) {
+	object1 := object{
+		Path: "test1",
+	}
+	object2 := Object2{}
+	buf, err := msgpack.Marshal(&object1)
+	require.NoError(t, err)
+	err = msgpack.Unmarshal(buf, &object2)
+	require.NoError(t, err)
+	require.Equal(t, object1.Path, object2.Path)
+
+	object1 = object{}
+	object2 = Object2{
+		Kind:   "test2",
+		Number: 123,
+		Path:   "test1",
+	}
+	object3 := Object2{}
+	buf, err = msgpack.Marshal(&object2)
+	require.NoError(t, err)
+	err = msgpack.Unmarshal(buf, &object1)
+	require.NoError(t, err)
+	require.Equal(t, object2.Path, object1.Path)
+
+	err = msgpack.Unmarshal(buf, &object3)
+	require.NoError(t, err)
+	require.Equal(t, object2, object3)
+}