core: object chunks; chunk manifest

* manifest metadata: deterministic ordering when serializing
* enforce uint16 max
* reset 'completed' on error
* part size, prev. commit: 05174b6

Signed-off-by: Alex Aizman <alex.aizman@gmail.com>

Author: alex-aizman

ais/tgtmpt.go

@@ -39,16 +39,18 @@ type (
 	}
 )
 
-func (ups *uploads) init(id string, lom *core.LOM, metadata map[string]string) {
+func (ups *uploads) init(id string, lom *core.LOM, metadata map[string]string) error {
 	manifest := core.NewUfest(id, lom)
-	manifest.Metadata = metadata
-
+	if err := manifest.SetMeta(metadata); err != nil {
+		return err
+	}
 	ups.Lock()
 	if ups.m == nil {
 		ups.m = make(map[string]*core.Ufest, iniCapUploads)
 	}
 	ups._add(id, manifest)
 	ups.Unlock()
+	return nil
 }
 
 func (ups *uploads) _add(id string, manifest *core.Ufest) {

ais/tgts3mpt.go

@@ -101,7 +101,10 @@ func (t *target) startMpt(w http.ResponseWriter, r *http.Request, items []string
 	if err != nil {
 		return
 	}
-	t.ups.init(uploadID, lom, metadata)
+	if err := t.ups.init(uploadID, lom, metadata); err != nil {
+		s3.WriteErr(w, r, err, 0)
+		return
+	}
 	result := &s3.InitiateMptUploadResult{Bucket: bck.Name, Key: objName, UploadID: uploadID}
 
 	sgl := t.gmm.NewSGL(0)
@@ -264,16 +267,14 @@ func (t *target) putMptPart(w http.ResponseWriter, r *http.Request, items []stri
 	chunk := &core.Uchunk{
 		MD5:  md5,
 		Path: chunkPath,
-		Siz:  size,
-		Num:  uint16(partNum),
 	}
 	if checkPartSHA {
 		chunk.Cksum = &cksumSHA.Cksum
 	}
 
 	// - see NOTE above in re "active uploads in memory"
 	// - TODO: this is the place to call Ufest.Store(partial manifest)
-	if err := manifest.Add(chunk); err != nil {
+	if err := manifest.Add(chunk, size, int64(partNum)); err != nil {
 		s3.WriteMptErr(w, r, err, 0, lom, uploadID)
 		return
 	}
@@ -400,7 +401,7 @@ func (t *target) completeMpt(w http.ResponseWriter, r *http.Request, items []str
 		nparts[i] = c
 	}
 
-	metadata := manifest.Metadata
+	metadata := manifest.GetMeta()
 	manifest.Unlock()
 
 	// 2. Create final work file

core/lchunk_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/NVIDIA/aistore/api/apc"
 	"github.com/NVIDIA/aistore/cmn"
 	"github.com/NVIDIA/aistore/cmn/cos"
+	"github.com/NVIDIA/aistore/cmn/debug"
 	"github.com/NVIDIA/aistore/core"
 	"github.com/NVIDIA/aistore/core/meta"
 	"github.com/NVIDIA/aistore/core/mock"
@@ -26,6 +27,13 @@ const (
 	xattrChunk = "user.ais.chunk"
 )
 
+var manifestMeta = map[string]string{
+	"content-type":  "application/octet-stream",
+	"storage-class": "STANDARD",
+	"cache-control": "no-cache",
+	"user-agent":    "ais-client/1.0",
+}
+
 var _ = Describe("Chunk Manifest Xattrs", func() {
 	const (
 		tmpDir     = "/tmp/chunk_xattr_test"
@@ -73,15 +81,17 @@ var _ = Describe("Chunk Manifest Xattrs", func() {
 			manifest.Num = numChunks
 			manifest.Chunks = make([]core.Uchunk, numChunks)
 
+			err := manifest.SetMeta(manifestMeta)
+			debug.AssertNoErr(err)
+
 			for i := range numChunks {
-				// Create a proper checksum instead of random string
 				cksumVal := trand.String(16)
 				manifest.Chunks[i] = core.Uchunk{
 					Siz:   chunkSizes[i],
 					Num:   i + 1,
 					Path:  trand.String(7),
-					Cksum: cos.NewCksum(cos.ChecksumCesXxh, cksumVal), // Use proper checksum type
-					MD5:   trand.String(32),                           // S3 legacy MD5
+					Cksum: cos.NewCksum(cos.ChecksumCesXxh, cksumVal),
+					MD5:   trand.String(32),
 				}
 			}
 			return manifest
@@ -148,6 +158,12 @@ var _ = Describe("Chunk Manifest Xattrs", func() {
 				err = loadedManifest.Load(lom)
 				Expect(err).NotTo(HaveOccurred())
 
+				loadedMeta := loadedManifest.GetMeta()
+				Expect(loadedMeta).To(HaveLen(len(manifestMeta)))
+				for k, v := range manifestMeta {
+					Expect(loadedMeta[k]).To(Equal(v), "metadata key %q mismatch", k)
+				}
+
 				// Verify manifest contents including new fields
 				Expect(loadedManifest.ID).To(Equal("test-session-001"))
 				Expect(loadedManifest.Created.Unix()).To(Equal(manifest.Created.Unix()))
@@ -222,6 +238,12 @@ var _ = Describe("Chunk Manifest Xattrs", func() {
 				err = loadedManifest.Load(lom)
 				Expect(err).NotTo(HaveOccurred())
 
+				loadedMeta := loadedManifest.GetMeta()
+				Expect(loadedMeta).To(HaveLen(len(manifestMeta)))
+				for k, v := range manifestMeta {
+					Expect(loadedMeta[k]).To(Equal(v), "metadata key %q mismatch", k)
+				}
+
 				Expect(loadedManifest.ID).To(Equal("many-chunks-session"))
 				Expect(loadedManifest.Num).To(Equal(numChunks))
 				Expect(loadedManifest.Chunks).To(HaveLen(int(numChunks)))

core/lchunk.go renamed to core/ufest.go

@@ -10,6 +10,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math"
 	"os"
 	"sort"
 	"sync"
@@ -35,7 +36,26 @@ const (
 
 	iniChunksCap = 16
 
-	completed = 1 // Ufest.Flags
+	maxChunkSize = 5 * cos.GiB
+	maxMetaKeys  = 1000
+)
+
+const (
+	completed uint16 = 1 << 0 // Ufest.Flags
+)
+
+// on-disk xattr
+const (
+	xattrChunk = "user.ais.chunk"
+
+	xattrChunkDflt = memsys.DefaultBufSize
+	xattrChunkMax  = memsys.DefaultBuf2Size // NOTE: 64K hard limit
+)
+
+const (
+	utag     = "chunk-manifest"
+	itag     = "invalid " + utag
+	tooShort = "failed to unpack: too short"
 )
 
 type (
@@ -47,13 +67,14 @@ type (
 		Flags uint16     // bit flags (future use)
 		MD5   string     // S3/legacy
 	}
-	Ufest struct {
-		ID       string    // upload/manifest ID
-		Created  time.Time // creation time
-		Num      uint16    // number of chunks (so far)
-		Flags    uint16    // bit flags { completed, ...}
-		Chunks   []Uchunk
-		Metadata map[string]string // remote object metadata
+	mdpair struct{ k, v string }
+	Ufest  struct {
+		ID      string    // upload/manifest ID
+		Created time.Time // creation time
+		Num     uint16    // number of chunks (so far)
+		Flags   uint16    // bit flags { completed, ...}
+		Chunks  []Uchunk
+		mdmap   map[string]string // remote object metadata (gets sorted when packing)
 
 		// runtime state
 		Lom    *LOM
@@ -63,20 +84,6 @@ type (
 	}
 )
 
-// on-disk xattr
-const (
-	xattrChunk = "user.ais.chunk"
-
-	xattrChunkDflt = memsys.DefaultBufSize
-	xattrChunkMax  = memsys.DefaultBuf2Size // maximum 64K
-)
-
-const (
-	utag     = "chunk-manifest"
-	itag     = "invalid " + utag
-	tooShort = "failed to unpack: too short"
-)
-
 func NewUfest(id string, lom *LOM) *Ufest {
 	startTime := time.Now()
 	if id == "" {
@@ -93,10 +100,32 @@ func NewUfest(id string, lom *LOM) *Ufest {
 
 func (u *Ufest) Completed() bool { return u.Flags&completed == completed }
 
+// NOTE:
+// - GetMeta() returns a reference that callers must not mutate
+// - alternatively, clone it and call SetMeta
+// - on-disk order is deterministic (sorted at pack time)
+func (u *Ufest) SetMeta(md map[string]string) error {
+	if l := len(md); l > maxMetaKeys {
+		return fmt.Errorf("%s: number of metadata entries %d exceeds %d limit", utag, l, maxMetaKeys)
+	}
+	u.mdmap = md
+	return nil
+}
+func (u *Ufest) GetMeta() map[string]string { return u.mdmap }
+
 func (u *Ufest) Lock()   { u.mu.Lock() }
 func (u *Ufest) Unlock() { u.mu.Unlock() }
 
-func (u *Ufest) Add(c *Uchunk) error {
+func (u *Ufest) Add(c *Uchunk, size, num int64) error {
+	if size > maxChunkSize {
+		return fmt.Errorf("%s [add] chunk size %d exceeds %d limit", utag, size, maxChunkSize)
+	}
+	c.Siz = size
+	if num > math.MaxUint16 || len(u.Chunks) >= math.MaxUint16 {
+		return fmt.Errorf("%s [add] chunk number (%d, %d) exceeds %d limit", utag, num, len(u.Chunks), math.MaxUint16)
+	}
+	c.Num = uint16(num)
+
 	u.mu.Lock()
 	defer u.mu.Unlock()
 
@@ -115,7 +144,7 @@ func (u *Ufest) Add(c *Uchunk) error {
 	dup := &u.Chunks[idx]
 	if idx < l && dup.Num == c.Num {
 		if err := cos.RemoveFile(dup.Path); err != nil {
-			return fmt.Errorf("failed to replace chunk [%d, %s]: %v", c.Num, dup.Path, err)
+			return fmt.Errorf("%s [add] failed to replace chunk [%d, %s]: %v", utag, c.Num, dup.Path, err)
 		}
 		u.Size += c.Siz - dup.Siz
 		*dup = *c
@@ -164,7 +193,7 @@ func (u *Ufest) Abort(lom *LOM) error {
 func (u *Ufest) ChunkName(num int) (string, error) {
 	lom := u.Lom
 	if lom == nil {
-		return "", errors.New("nil lom")
+		return "", errors.New(utag + ": nil lom")
 	}
 	if num <= 0 {
 		return "", fmt.Errorf("%s: invalid chunk number (%d)", _utag(lom.Cname()), num)
@@ -293,7 +322,9 @@ func (u *Ufest) Store(lom *LOM) error {
 	}
 
 	// write
-	if err := lom.setXchunk(sgl.Bytes()); err != nil {
+	b := sgl.Bytes()
+	if err := lom.setXchunk(b); err != nil {
+		u.Flags &^= completed
 		return err
 	}
 	lom.md.lid.setlmfl(lmflChunk) // TODO -- FIXME: persist
@@ -340,12 +371,21 @@ func (u *Ufest) pack(w io.Writer) {
 	binary.BigEndian.PutUint16(b16[:], u.Flags)
 	w.Write(b16[:])
 
-	// metadata map
-	binary.BigEndian.PutUint16(b16[:], uint16(len(u.Metadata)))
+	// metadata
+	l := len(u.mdmap)
+	binary.BigEndian.PutUint16(b16[:], uint16(l))
 	w.Write(b16[:])
-	for k, v := range u.Metadata {
-		_packStr(w, k)
-		_packStr(w, v)
+	if l > 0 {
+		mdslice := make([]mdpair, 0, l)
+		for k, v := range u.mdmap {
+			mdslice = append(mdslice, mdpair{k, v})
+		}
+		sort.Slice(mdslice, func(i, j int) bool { return mdslice[i].k < mdslice[j].k }) // deterministic
+		for _, kv := range mdslice {
+			_packStr(w, kv.k)
+			_packStr(w, kv.v)
+		}
+		clear(mdslice)
 	}
 
 	// chunks
@@ -435,7 +475,7 @@ func (u *Ufest) unpack(data []byte) (err error) {
 	offset += cos.SizeofI16
 
 	if metaCount > 0 {
-		u.Metadata = make(map[string]string, metaCount)
+		u.mdmap = make(map[string]string, metaCount)
 		for range metaCount {
 			var k, v string
 			if k, offset, err = _unpackStr(data, offset); err != nil {
@@ -444,7 +484,7 @@ func (u *Ufest) unpack(data []byte) (err error) {
 			if v, offset, err = _unpackStr(data, offset); err != nil {
 				return err
 			}
-			u.Metadata[k] = v
+			u.mdmap[k] = v
 		}
 	}