core: object chunks; chunk manifest

* move remote metadata from chunk manifest to target-level (`ups` structure)
  - remove metadata storage from chunk manifest
* manifest persistence: add LoadPartial()
  - as in: StorePartial() vs StoreCompleted()
* fix xattr storage asymmetry: _partial_ manifests use chunk #1, and vice versa
* add level 1 unit tests for core Ufest functionality
* ongoing: error handling; validation and cleanup; other refactoring
* tests:
  - rewrite level-1 unit test
  - enable multipart test #2: s3cmd with xxhash end-to-end validation
----
* part ten, prev. commit: e2c6a0d

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

Author: alex-aizman

ais/target.go

@@ -73,7 +73,7 @@ type (
 		res      *res.Res
 		txns     txns
 		regstate regstate
-		ups      uploads
+		ups      ups
 	}
 )
 

ais/test/scripted_cli_test.go

@@ -148,7 +148,6 @@ func TestCopySyncRemaisUsingScript(t *testing.T) {
 // 1. MD5 checksum
 // 2. "apc.S3Scheme+apc.BckProviderSeparator+bck.Name"
 func TestMPU_1_UsingScript(t *testing.T) {
-	t.Skipf("skipping %s - requires xxhsum installed", t.Name())
 	tempdir := t.TempDir()
 	bck := cmn.Bck{Name: trand.String(10), Provider: apc.AIS}
 
@@ -174,7 +173,6 @@ func TestMPU_1_UsingScript(t *testing.T) {
 }
 
 func TestMPU_2_UsingScript(t *testing.T) {
-	t.Skipf("skipping %s - requires xxhsum installed", t.Name())
 	bck := cmn.Bck{Name: trand.String(10), Provider: apc.AIS}
 	cmd := exec.Command("./scripts/multipart-smoke.sh", "--bucket", bck.Cname(""))
 

ais/test/scripts/multipart-smoke.sh

@@ -86,9 +86,9 @@ elif [[ "$bucket" == s3://* ]]; then
   bucket_name="${bucket#s3://}"
   s3_bucket="$bucket"
 
-  # Verify S3 bucket exists (s3cmd will fail if it doesn't)
-  if ! s3cmd ls "$bucket" >/dev/null 2>&1; then
-    echo "Error: S3 bucket $bucket does not exist or is not accessible"
+  # Verify bucket exists
+  if ! ais show bucket "$bucket" -c >/dev/null 2>&1; then
+    echo "Error: $bucket does not exist"
     exit 1
   fi
 
@@ -128,10 +128,18 @@ h1=$(xxhsum "$src" | awk '{print $1}')
 echo "   Source hash: $h1"
 
 echo "3. uploading via s3cmd to $s3_bucket..."
-s3cmd put "$src" "$s3_bucket/${filename}" || exit $?
+s3cmd put "$src" "$s3_bucket/${filename}" \
+  --no-ssl \
+  --host=localhost:8080/s3 \
+  --host-bucket="localhost:8080/s3/%(bucket)" \
+|| exit $?
 
 echo "4. downloading via s3cmd from $s3_bucket..."
-s3cmd get "$s3_bucket/${filename}" "$dst" --force || exit $?
+s3cmd get "$s3_bucket/${filename}" "$dst" --force \
+  --no-ssl \
+  --host=localhost:8080/s3 \
+  --host-bucket="localhost:8080/s3/%(bucket)" \
+|| exit $?
 
 echo "5. computing xxhash of downloaded file..."
 h2=$(xxhsum "$dst" | awk '{print $1}')

ais/tgtmpt.go

@@ -28,66 +28,77 @@ const (
 
 // see also: https://docs.aws.amazon.com/AmazonS3/latest/userguide/qfacts.html
 const (
-	maxPartsPerUpload = 10000
+	maxPartsPerUpload = 9999 // 64K limited (compressed) unless (e.g.) `mkfs.xfs -m attr=128k /dev/sdX`
 )
 
 type (
-	uploads struct {
+	up struct {
+		u   *core.Ufest
+		rmd map[string]string
+	}
+	ups struct {
 		t *target
-		m map[string]*core.Ufest // by upload ID
+		m map[string]up
 		sync.RWMutex
 	}
 )
 
-func (ups *uploads) init(id string, lom *core.LOM, metadata map[string]string) error {
+func (ups *ups) init(id string, lom *core.LOM, rmd map[string]string) error {
 	manifest := core.NewUfest(id, lom, false /*must-exist*/)
-	if err := manifest.SetMeta(metadata); err != nil {
-		return err
-	}
 	ups.Lock()
 	if ups.m == nil {
-		ups.m = make(map[string]*core.Ufest, iniCapUploads)
+		ups.m = make(map[string]up, iniCapUploads)
 	}
-	ups._add(id, manifest)
+	ups._add(id, manifest, rmd)
 	ups.Unlock()
 	return nil
 }
 
-func (ups *uploads) _add(id string, manifest *core.Ufest) {
+func (ups *ups) _add(id string, manifest *core.Ufest, rmd map[string]string) {
 	debug.Assert(manifest.Lom != nil)
 	_, ok := ups.m[id]
 	debug.Assert(!ok, "duplicated upload ID: ", id, " ", manifest.Lom.Cname())
-	ups.m[id] = manifest
+	ups.m[id] = up{manifest, rmd}
+}
+
+func (ups *ups) get(id string) (manifest *core.Ufest) {
+	ups.RLock()
+	manifest = ups.m[id].u
+	ups.RUnlock()
+	return
 }
 
-func (ups *uploads) get(id string) (manifest *core.Ufest) {
+func (ups *ups) getWithMeta(id string) (manifest *core.Ufest, rmd map[string]string) {
 	ups.RLock()
-	manifest = ups.m[id]
+	manifest = ups.m[id].u
+	rmd = ups.m[id].rmd
 	ups.RUnlock()
 	return
 }
 
 // NOTE: must be called with ups unlocked
-func (ups *uploads) fromFS(id string, lom *core.LOM, add bool) (manifest *core.Ufest, err error) {
+func (ups *ups) fromFS(id string, lom *core.LOM, add bool) (manifest *core.Ufest, err error) {
 	manifest = core.NewUfest(id, lom, true /*must-exist*/)
 	if err = manifest.Load(lom); err == nil && add {
 		ups.Lock()
-		ups._add(id, manifest)
+		ups._add(id, manifest, nil /*remote metadata <= LOM custom*/)
 		ups.Unlock()
 	}
 	return
 }
 
-func (ups *uploads) abort(id string, lom *core.LOM) (ecode int, err error) {
+func (ups *ups) abort(id string, lom *core.LOM) (ecode int, err error) {
+	var manifest *core.Ufest
 	ups.Lock()
-	manifest, ok := ups.m[id]
+	up, ok := ups.m[id]
 	if !ok {
 		ups.Unlock()
 		manifest, err = ups.fromFS(id, lom, false /*add*/)
 		if err != nil {
 			return 0, err
 		}
 	} else {
+		manifest = up.u
 		delete(ups.m, id)
 		ups.Unlock()
 	}
@@ -100,17 +111,17 @@ func (ups *uploads) abort(id string, lom *core.LOM) (ecode int, err error) {
 	return 0, nil
 }
 
-func (ups *uploads) toSlice() (all []*core.Ufest) {
+func (ups *ups) toSlice() (all []*core.Ufest) {
 	ups.RLock()
 	all = make([]*core.Ufest, 0, len(ups.m))
-	for _, manifest := range ups.m {
-		all = append(all, manifest)
+	for _, up := range ups.m {
+		all = append(all, up.u)
 	}
 	ups.RUnlock()
 	return
 }
 
-func (ups *uploads) del(id string) {
+func (ups *ups) del(id string) {
 	ups.Lock()
 	delete(ups.m, id)
 	ups.Unlock()
@@ -120,7 +131,7 @@ func (ups *uploads) del(id string) {
 // backend operations - encapsulate IsRemoteS3/IsRemoteOCI pattern
 //
 
-func (ups *uploads) start(w http.ResponseWriter, r *http.Request, lom *core.LOM, q url.Values) (uploadID string, metadata map[string]string, err error) {
+func (ups *ups) start(w http.ResponseWriter, r *http.Request, lom *core.LOM, q url.Values) (uploadID string, metadata map[string]string, err error) {
 	var (
 		ecode int
 		bck   = lom.Bck()
@@ -141,7 +152,7 @@ func (ups *uploads) start(w http.ResponseWriter, r *http.Request, lom *core.LOM,
 	return
 }
 
-func (ups *uploads) putPartRemote(lom *core.LOM, reader io.ReadCloser, r *http.Request, q url.Values, uploadID string, expectedSize int64, partNum int32) (etag string, ecode int, err error) {
+func (ups *ups) putPartRemote(lom *core.LOM, reader io.ReadCloser, r *http.Request, q url.Values, uploadID string, expectedSize int64, partNum int32) (etag string, ecode int, err error) {
 	bck := lom.Bck()
 	if bck.IsRemoteS3() {
 		etag, ecode, err = backend.PutMptPartAWS(lom, reader, r, q, uploadID, expectedSize, partNum)
@@ -152,7 +163,7 @@ func (ups *uploads) putPartRemote(lom *core.LOM, reader io.ReadCloser, r *http.R
 	return
 }
 
-func (ups *uploads) completeRemote(w http.ResponseWriter, r *http.Request, lom *core.LOM, q url.Values, uploadID string, body []byte, partList *s3.CompleteMptUpload) (etag string, err error) {
+func (ups *ups) completeRemote(w http.ResponseWriter, r *http.Request, lom *core.LOM, q url.Values, uploadID string, body []byte, partList *s3.CompleteMptUpload) (etag string, err error) {
 	var (
 		ecode    int
 		provider string
@@ -178,7 +189,7 @@ func (ups *uploads) completeRemote(w http.ResponseWriter, r *http.Request, lom *
 	return
 }
 
-func (ups *uploads) abortRemote(w http.ResponseWriter, r *http.Request, lom *core.LOM, q url.Values, uploadID string) (err error) {
+func (ups *ups) abortRemote(w http.ResponseWriter, r *http.Request, lom *core.LOM, q url.Values, uploadID string) (err error) {
 	var (
 		ecode int
 		bck   = lom.Bck()
@@ -194,7 +205,7 @@ func (ups *uploads) abortRemote(w http.ResponseWriter, r *http.Request, lom *cor
 	return
 }
 
-func (*uploads) encodeRemoteMetadata(lom *core.LOM, metadata map[string]string) (md map[string]string) {
+func (*ups) encodeRemoteMetadata(lom *core.LOM, metadata map[string]string) (md map[string]string) {
 	bck := lom.Bck()
 	if bck.IsRemoteS3() {
 		md = cmn.BackendHelpers.Amazon.EncodeMetadata(metadata)
@@ -209,7 +220,7 @@ func (*uploads) encodeRemoteMetadata(lom *core.LOM, metadata map[string]string)
 // misc
 //
 
-func (*uploads) parsePartNum(s string) (int32, error) {
+func (*ups) parsePartNum(s string) (int32, error) {
 	partNum, err := strconv.ParseInt(s, 10, 32)
 	if err != nil {
 		err = fmt.Errorf("invalid part number %q (must be in 1-%d range): %v", s, maxPartsPerUpload, err)

ais/tgts3mpt.go

@@ -324,7 +324,7 @@ func (t *target) completeMpt(w http.ResponseWriter, r *http.Request, items []str
 	}
 
 	// TODO: compare with listMptUploads (that does fromFS)
-	manifest := t.ups.get(uploadID)
+	manifest, metadata := t.ups.getWithMeta(uploadID)
 	if manifest == nil {
 		s3.WriteMptErr(w, r, s3.NewErrNoSuchUpload(uploadID, nil), http.StatusNotFound, lom, uploadID)
 		return
@@ -378,7 +378,6 @@ func (t *target) completeMpt(w http.ResponseWriter, r *http.Request, items []str
 	// compute "whole" checksum // TODO -- FIXME: sha256 may take precedence when implied by partSHA (see above)
 	var (
 		wholeCksum *cos.CksumHash
-		metadata   = manifest.GetMeta()
 		remote     = bck.IsRemoteS3() || bck.IsRemoteOCI()
 	)
 	if remote && lom.CksumConf().Type != cos.ChecksumNone {
@@ -395,7 +394,7 @@ func (t *target) completeMpt(w http.ResponseWriter, r *http.Request, items []str
 		lom.SetCksum(&wholeCksum.Cksum)
 	}
 
-	// compute ETag if need be
+	// compute multipart-compliant ETag if need be
 	var etag string
 	if !remote {
 		var err error