aisloader: add support for percentage-based archive workload (read, write)

* aisloader work-order now includes `archpath`
* PUT: create shards at specified percentage:
      `arch.pct=30` means 30% of all PUTs create shards, 70% - plain objects
      new aisloader operation: `opPutShard`
* GET:
      read archived files from existing shards
* aisloader's list-objects:
  - use `apc.LsArchDir` flag
  - filter shard objects using 'entry-is-archive'
* with refactoring, minor fixes, and micro-optimizations
-------------
* target's list-objects:
  - contained files should not be listed with 'entry-is-archive' (fix)
-------------
* add misc helpers:
  * archive.SplitAtExtension() common helper
  * encoding/decoding with zero-byte separator
-------------
* TODO:
  - get-batch from archives
  - multipart upload shards

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

Author: alex-aizman

bench/tools/aisloader/client.go

@@ -377,26 +377,28 @@ func newTraceCtx(proxyURL string) *traceCtx {
 	return tctx
 }
 
-func newGetRequest(proxyURL string, bck cmn.Bck, objName string, offset, length int64, latest bool) (*http.Request, error) {
-	var (
-		hdr   http.Header
-		query = url.Values{}
-	)
-	query = bck.AddToQuery(query)
+func newGetRequest(proxyURL string, wo *workOrder, p *params) (*http.Request, error) {
+	query := p.bck.AddToQuery(nil)
 	if etlName != "" {
 		query.Add(apc.QparamETLName, etlName)
 	}
-	if latest {
+	if p.latest {
 		query.Add(apc.QparamLatestVer, "true")
 	}
-	if length > 0 {
-		rng := cmn.MakeRangeHdr(offset, length)
+	if wo.archpath != "" {
+		query.Add(apc.QparamArchpath, wo.archpath)
+	}
+
+	var hdr http.Header
+	if p.readLen > 0 {
+		rng := cmn.MakeRangeHdr(p.readOff, p.readLen)
 		hdr = http.Header{cos.HdrRange: []string{rng}}
 	}
+
 	reqArgs := cmn.HreqArgs{
 		Method: http.MethodGet,
 		Base:   proxyURL,
-		Path:   apc.URLPathObjects.Join(bck.Name, objName),
+		Path:   apc.URLPathObjects.Join(p.bck.Name, wo.objName),
 		Query:  query,
 		Header: hdr,
 	}
@@ -431,8 +433,8 @@ func s3getDiscard(bck cmn.Bck, objName string) (int64, error) {
 }
 
 // getDiscard sends a GET request and discards returned data.
-func getDiscard(proxyURL string, bck cmn.Bck, objName string, offset, length int64, validate, latest bool) (int64, error) {
-	req, err := newGetRequest(proxyURL, bck, objName, offset, length, latest)
+func getDiscard(proxyURL string, wo *workOrder, p *params) (int64, error) {
+	req, err := newGetRequest(proxyURL, wo, p)
 	if err != nil {
 		return 0, err
 	}
@@ -446,30 +448,33 @@ func getDiscard(proxyURL string, bck cmn.Bck, objName string, offset, length int
 	}
 
 	var hdrCksumValue, hdrCksumType string
-	if validate {
+	if p.verifyHash {
 		hdrCksumValue = resp.Header.Get(apc.HdrObjCksumVal)
 		hdrCksumType = resp.Header.Get(apc.HdrObjCksumType)
 	}
-	src := "GET " + bck.Cname(objName)
-	n, cksumValue, err := readDiscard(resp, src, hdrCksumType)
+	n, cksumValue, err := readDiscard(resp, hdrCksumType)
 
 	resp.Body.Close()
 	if err != nil {
-		return 0, err
+		tag := "GET " + p.bck.Cname(wo.objName)
+		if wo.archpath != "" {
+			tag += "/" + wo.archpath
+		}
+		return 0, fmt.Errorf("%s: %v", tag, err)
 	}
-	if validate && hdrCksumValue != cksumValue {
+	if p.verifyHash && hdrCksumValue != cksumValue {
 		return 0, cmn.NewErrInvalidCksum(hdrCksumValue, cksumValue)
 	}
 	return n, err
 }
 
 // Same as above, but with HTTP trace.
-func getTraceDiscard(proxyURL string, bck cmn.Bck, objName string, latencies *httpLatencies, offset, length int64, validate, latest bool) (int64, error) {
+func getTraceDiscard(proxyURL string, wo *workOrder, p *params) (int64, error) {
 	var (
 		hdrCksumValue string
 		hdrCksumType  string
 	)
-	req, err := newGetRequest(proxyURL, bck, objName, offset, length, latest)
+	req, err := newGetRequest(proxyURL, wo, p)
 	if err != nil {
 		return 0, err
 	}
@@ -486,21 +491,24 @@ func getTraceDiscard(proxyURL string, bck cmn.Bck, objName string, latencies *ht
 	defer resp.Body.Close()
 
 	tctx.tr.tsHTTPEnd = time.Now()
-	if validate {
+	if p.verifyHash {
 		hdrCksumValue = resp.Header.Get(apc.HdrObjCksumVal)
 		hdrCksumType = resp.Header.Get(apc.HdrObjCksumType)
 	}
 
-	src := "GET " + bck.Cname(objName)
-	n, cksumValue, err := readDiscard(resp, src, hdrCksumType)
+	n, cksumValue, err := readDiscard(resp, hdrCksumType)
 	if err != nil {
-		return 0, err
+		tag := "GET " + p.bck.Cname(wo.objName)
+		if wo.archpath != "" {
+			tag += "/" + wo.archpath
+		}
+		return 0, fmt.Errorf("%s: %v", tag, err)
 	}
-	if validate && hdrCksumValue != cksumValue {
+	if p.verifyHash && hdrCksumValue != cksumValue {
 		err = cmn.NewErrInvalidCksum(hdrCksumValue, cksumValue)
 	}
 
-	tctx.tr.set(latencies)
+	tctx.tr.set(wo.latencies)
 	return n, err
 }
 
@@ -557,7 +565,7 @@ func listObjCallback(ctx *api.LsoCounter) {
 }
 
 // listObjectNames returns a slice of object names of all objects that match the prefix in a bucket.
-func listObjectNames(p *params) ([]string, error) {
+func listObjectNames(p *params) (names []string, fcnt int /*num archived files*/, _ error) {
 	var (
 		bp       = p.bp
 		bck      = p.bck
@@ -572,16 +580,37 @@ func listObjectNames(p *params) ([]string, error) {
 		msg.Flags |= apc.LsNoDirs // aisloader's default (to override, use --list-dirs)
 	}
 	args := api.ListArgs{Callback: listObjCallback, CallAfter: longListTime}
+	if p.archParams.pct > 0 {
+		msg.Flags |= apc.LsArchDir
+	}
+
 	lst, err := api.ListObjects(bp, bck, msg, args)
 	if err != nil {
-		return nil, err
-	}
+		return nil, 0, err
+	}
+
+	names = make([]string, 0, len(lst.Entries))
+	for _, en := range lst.Entries {
+		if en.Flags&apc.EntryInArch != 0 {
+			debug.Assert(msg.Flags&apc.LsArchDir != 0)
+			objName, archPath := archive.SplitAtExtension(en.Name)
+			if archPath != "" {
+				names = append(names, encodeArchName(objName, archPath)) // with `archSep` delimiter
+				fcnt++
+				continue
+			}
+		}
+		// skip archive/shard objects themselves
+		if en.Flags&apc.EntryIsArchive != 0 {
+			debug.Assert(msg.Flags&apc.LsArchDir != 0)
+			continue
+		}
 
-	objs := make([]string, 0, len(lst.Entries))
-	for _, obj := range lst.Entries {
-		objs = append(objs, obj.Name)
+		// plain object or shard itself
+		names = append(names, en.Name)
 	}
-	return objs, nil
+
+	return names, fcnt, nil
 }
 
 func initS3Svc() error {
@@ -668,13 +697,13 @@ func s3ListObjects() ([]string, error) {
 	return names, nil
 }
 
-func readDiscard(r *http.Response, tag, cksumType string) (int64, string, error) {
+func readDiscard(r *http.Response, cksumType string) (int64, string, error) {
 	if r.StatusCode >= http.StatusBadRequest {
 		bytes, err := cos.ReadAll(r.Body)
 		if err == nil {
-			return 0, "", fmt.Errorf("bad status %d from %s, response: %s", r.StatusCode, tag, string(bytes))
+			return 0, "", fmt.Errorf("bad status %d, response: %s", r.StatusCode, string(bytes))
 		}
-		return 0, "", fmt.Errorf("bad status %d from %s: %v", r.StatusCode, tag, err)
+		return 0, "", fmt.Errorf("bad status %d: %v", r.StatusCode, err)
 	}
 
 	n, cksum, err := cos.CopyAndChecksum(io.Discard, r.Body, nil, cksumType)

bench/tools/aisloader/params.go

@@ -90,7 +90,7 @@ type (
 		minSz int64
 		maxSz int64
 
-		use bool // whether to enable sharding across all supported workloads (PUT, GET, etc.)
+		pct int // broadly: percentage of archive workload (affects both reading and writing when positive)
 	}
 
 	// Object naming strategy and distribution
@@ -236,7 +236,7 @@ func addCmdLine(f *flag.FlagSet, p *params) {
 	f.IntVar(&p.archParams.numFiles, "arch.num-files", 0, "number of archived files per shard (PUT only; default gets computed from sizes)")
 	f.StringVar(&p.archParams.minSzStr, "arch.minsize", "", "minimum file size (with or without multiplicative suffix K, MB, GiB, etc.)")
 	f.StringVar(&p.archParams.maxSzStr, "arch.maxsize", "", "maximum file size (with or without multiplicative suffix K, MB, GiB, etc.)")
-	f.BoolVar(&p.archParams.use, "arch.use", false, "enable sharding across all workloads: PUT, GET, chunks, batches, etc.")
+	f.IntVar(&p.archParams.pct, "arch.pct", 0, "when writing: percentage of shards vs plain objects; when reading: include archived files in the GET pool (ratio is ultimately determined by dataset)")
 
 	// ============ Naming ============
 	f.StringVar(&p.subDir, "subdir", "", "For GET requests, '-subdir' is a prefix that may or may not be an actual _virtual directory_;\n"+
@@ -349,12 +349,20 @@ func initParams(p *params) (err error) {
 	if p.putSizeUpperBound, err = _parseSize(p.putSizeUpperBoundStr, "totalputsize", 0); err != nil {
 		return err
 	}
+
 	if p.minSize, err = _parseSize(p.minSizeStr, "minsize", cos.MiB); err != nil {
 		return err
 	}
 	if p.maxSize, err = _parseSize(p.maxSizeStr, "maxsize", cos.GiB); err != nil {
 		return err
 	}
+	if p.minSizeStr == "" && p.maxSizeStr != "" {
+		p.minSize = min(p.minSize, p.maxSize)
+	}
+	if p.maxSizeStr == "" && p.minSizeStr != "" {
+		p.maxSize = max(p.minSize, p.maxSize)
+	}
+
 	if p.readOff, err = _parseSize(p.readOffStr, "readoff", 0); err != nil {
 		return err
 	}
@@ -659,7 +667,21 @@ func (p *params) validate() error {
 	}
 
 	// validate archive
-	if p.archParams.use {
+	if p.archParams.pct != 0 {
+		if p.archParams.pct < 0 || p.archParams.pct > 100 {
+			return fmt.Errorf("invalid option: arch.pct %d (must be 0-100)", p.archParams.pct)
+		}
+		if s3Endpoint != "" {
+			return errors.New("archive operations require AIStore cluster; not supported with '-s3endpoint' (direct S3 access)")
+		}
+
+		// TODO:
+		// - currently, we create each part independently with fresh readers
+		// - multipart upload of shards requires a special streaming chunk writer
+		if p.multipartPct > 0 {
+			return errors.New("multipart uploads and archive operations are mutually exclusive (not supported yet)")
+		}
+
 		mime := cos.NonZero(p.archParams.format, archive.ExtTar)
 		arch := &readers.Arch{
 			Mime:    mime,

bench/tools/aisloader/print.go

@@ -369,6 +369,7 @@ type (
 		Cleanup       bool    `json:"cleanup"`
 	}
 	ppArch struct {
+		Pct      int    `json:"% workload"`
 		Format   string `json:"format"`
 		Prefix   string `json:"prefix,omitempty"`
 		NumFiles int    `json:"files per shard,omitempty"`
@@ -379,7 +380,7 @@ type (
 
 func printRunParams(p *params) {
 	var arch *ppArch
-	if p.archParams.use {
+	if p.archParams.pct != 0 {
 		// temp readers.Arch to run Init() and get computed values
 		// aisloader's default format: TAR
 		mime := cos.NonZero(p.archParams.format, archive.ExtTar)
@@ -395,6 +396,7 @@ func printRunParams(p *params) {
 
 		numf := cos.Ternary(tmpArch.Num == readers.DynamicNumFiles, 0, p.archParams.numFiles)
 		arch = &ppArch{
+			Pct:      p.archParams.pct,
 			Format:   tmpArch.Mime,
 			Prefix:   tmpArch.Prefix,
 			NumFiles: numf,

bench/tools/aisloader/run.go

@@ -247,7 +247,7 @@ func Start(version, buildtime string) (err error) {
 
 		// note that "'-skiplist' option is ignored when '-filelist' is specified"
 	case !runParams.skipList:
-		names, err := listObjects()
+		names, fcnt, err := listObjects()
 		if err != nil {
 			return err
 		}
@@ -269,7 +269,15 @@ func Start(version, buildtime string) (err error) {
 		// initialize global name-getter and announce
 		objnameGetter = ng
 		objnameGetter.Init(names, rnd)
-		fmt.Printf("Found %s existing object%s\n\n", cos.FormatBigInt(l), cos.Plural(l))
+		switch fcnt {
+		case 0:
+			fmt.Printf("Found %s existing object%s\n\n", cos.FormatBigInt(l), cos.Plural(l))
+		case l:
+			fmt.Printf("Found %s archived file%s\n\n", cos.FormatBigInt(l), cos.Plural(l))
+		default:
+			fmt.Printf("Found %s plain object%s and %s archived file%s\n\n",
+				cos.FormatBigInt(l-fcnt), cos.Plural(l-fcnt), cos.FormatBigInt(fcnt), cos.Plural(fcnt))
+		}
 
 	default:
 		objnameGetter = &namegetter.Random{}
@@ -735,16 +743,16 @@ func objNamesFromFile() (names []string, err error) {
 	return
 }
 
-func listObjects() (names []string, err error) {
+func listObjects() (names []string, fcnt int, err error) {
 	switch {
 	case runParams.fileList != "":
 		names, err = objNamesFromFile()
 	case isDirectS3():
 		names, err = s3ListObjects()
 	default:
-		names, err = listObjectNames(runParams)
+		names, fcnt, err = listObjectNames(runParams)
 	}
-	return
+	return names, fcnt, err
 }
 
 func newNameGetter(names []string) (ng namegetter.Basic, isPermBased bool) {