[#100] preset_s3: Add a flag for percent of versioned buckets

Add flag "--buckets_versioned". Default is 0 (no versioned buckets)

Signed-off-by: Nikita Zinkevich <n.zinkevich@yadro.com>

.forgejo/workflows/dco.yml

@@ -0,0 +1,21 @@
+name: DCO action
+on: [pull_request]
+
+jobs:
+  dco:
+    name: DCO
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+        with:
+          fetch-depth: 0
+
+      - name: Setup Go
+        uses: actions/setup-go@v3
+        with:
+          go-version: '1.22'
+
+      - name: Run commit format checker
+        uses: https://git.frostfs.info/TrueCloudLab/dco-go@v2
+        with:
+          from: 'origin/${{ github.event.pull_request.base.ref }}'

.forgejo/workflows/tests.yml

@@ -0,0 +1,56 @@
+name: Tests and linters
+on: [pull_request]
+
+jobs:
+  lint:
+    name: Lint
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+
+      - name: Set up Go
+        uses: actions/setup-go@v3
+        with:
+          go-version: '1.23'
+          cache: true
+
+      - name: Install linters
+        run: make lint-install
+
+      - name: Run linters
+        run: make lint
+
+  tests:
+    name: Tests
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        go_versions: [ '1.22', '1.23' ]
+      fail-fast: false
+    steps:
+      - uses: actions/checkout@v3
+
+      - name: Set up Go
+        uses: actions/setup-go@v3
+        with:
+          go-version: '${{ matrix.go_versions }}'
+          cache: true
+
+      - name: Run tests
+        run: make test
+
+  tests-race:
+    name: Tests with -race
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+
+      - name: Set up Go
+        uses: actions/setup-go@v3
+        with:
+          go-version: '1.22'
+          cache: true
+
+      - name: Run tests
+        run: go test ./... -count=1 -race
+

.github/workflows/dco.yml

@@ -1,21 +0,0 @@
-name: DCO check
-
-on:
-  pull_request:
-    branches:
-      - master
-
-jobs:
-  commits_check_job:
-    runs-on: ubuntu-latest
-    name: Commits Check
-    steps:
-      - name: Get PR Commits
-        id: 'get-pr-commits'
-        uses: tim-actions/get-pr-commits@master
-        with:
-          token: ${{ secrets.GITHUB_TOKEN }}
-      - name: DCO Check
-        uses: tim-actions/dco@master
-        with:
-          commits: ${{ steps.get-pr-commits.outputs.commits }}

.github/workflows/go.yml

@@ -1,34 +0,0 @@
-name: Tests
-
-on:
-  pull_request:
-    branches:
-      - master
-    types: [opened, synchronize]
-    paths-ignore:
-      - '**/*.md'
-  workflow_dispatch:
-
-jobs:
-  lint:
-    name: Lint
-    runs-on: ubuntu-20.04
-    steps:
-      - name: Check out code
-        uses: actions/checkout@v2
-
-      - name: golangci-lint
-        uses: golangci/golangci-lint-action@v2
-        with:
-          version: latest
-          args: --timeout=2m
-
-  tests:
-    name: Tests
-    runs-on: ubuntu-20.04
-    strategy:
-      matrix:
-        go_versions: [ '1.17', '1.18', '1.19' ]
-      fail-fast: false
-    steps:
-    - uses: actions/checkout@v3

.gitignore

@@ -1,3 +1,6 @@
 k6
 *.bolt
 presets
+bin
+# Preset script artifacts.
+__pycache__

CONTRIBUTING.md

@@ -3,8 +3,8 @@
 First, thank you for contributing! We love and encourage pull requests from
 everyone. Please follow the guidelines:
 
-- Check the open [issues](https://github.com/TrueCloudLab/xk6-frostfs/issues) and
-  [pull requests](https://github.com/TrueCloudLab/xk6-frostfs/pulls) for existing
+- Check the open [issues](https://git.frostfs.info/TrueCloudLab/xk6-frostfs/issues) and
+  [pull requests](https://git.frostfs.info/TrueCloudLab/xk6-frostfs/pulls) for existing
   discussions.
 
 - Open an issue first, to discuss a new feature or enhancement.
@@ -27,19 +27,20 @@ Start by forking the `xk6-frostfs` repository, make changes in a branch and then
 send a pull request. We encourage pull requests to discuss code changes. Here
 are the steps in details:
 
-### Set up your GitHub Repository
-Fork [xk6-frostfs upstream](https://github.com/TrueCloudLab/xk6-frostfs/fork) source
+### Set up your repository
+
+Fork [xk6-frostfs upstream](https://git.frostfs.info/TrueCloudLab/xk6-frostfs/fork) source
 repository to your own personal repository. Copy the URL of your fork (you will
 need it for the `git clone` command below).
 
 ```sh
-$ git clone https://github.com/TrueCloudLab/xk6-frostfs
+$ git clone https://git.frostfs.info/TrueCloudLab/xk6-frostfs
 ```
 
 ### Set up git remote as ``upstream``
 ```sh
 $ cd xk6-frostfs
-$ git remote add upstream https://github.com/TrueCloudLab/xk6-frostfs
+$ git remote add upstream https://git.frostfs.info/TrueCloudLab/xk6-frostfs
 $ git fetch upstream
 $ git merge upstream/master
 ...
@@ -89,7 +90,7 @@ $ git push origin feature/123-something_awesome
 ```
 
 ### Create a Pull Request
-Pull requests can be created via GitHub. Refer to [this
+Pull requests can be created via git.frostfs.info. Refer to [this
 document](https://help.github.com/articles/creating-a-pull-request/) for
 detailed steps on how to create a pull request. After a Pull Request gets peer
 reviewed and approved, it will be merged.

Makefile

@@ -0,0 +1,114 @@
+#!/usr/bin/make -f
+
+# Common variables
+REPO ?= $(shell go list -m)
+VERSION ?= $(shell git describe --tags --dirty --match "v*" --always --abbrev=8 2>/dev/null || cat VERSION 2>/dev/null || echo "develop")
+GO_VERSION ?= 1.22
+LINT_VERSION ?= 1.60.3
+TRUECLOUDLAB_LINT_VERSION ?= 0.0.7
+BINDIR = bin
+
+OUTPUT_LINT_DIR ?= $(abspath $(BINDIR))/linters
+LINT_DIR = $(OUTPUT_LINT_DIR)/golangci-lint-$(LINT_VERSION)-v$(TRUECLOUDLAB_LINT_VERSION)
+TMP_DIR := .cache
+
+# Binaries to build
+CMDS = $(addprefix frostfs-, $(notdir $(wildcard cmd/*)))
+BINS = $(addprefix $(BINDIR)/, $(CMDS))
+
+.PHONY: all $(BINS) $(BINDIR) dep docker/ test cover format lint docker/lint pre-commit unpre-commit version clean
+
+# Make all binaries
+all: $(BINS)
+
+$(BINS): $(BINDIR) dep
+	@echo "⇒ Build $@"
+	CGO_ENABLED=0 \
+	go build -v -trimpath \
+	-ldflags "-X $(REPO)/internal/version.Version=$(VERSION)" \
+	-o $@ ./cmd/$(subst frostfs-,,$(notdir $@))
+
+$(BINDIR):
+	@echo "⇒ Ensure dir: $@"
+	@mkdir -p $@
+
+# Pull go dependencies
+dep:
+	@printf "⇒ Download requirements: "
+	@CGO_ENABLED=0 \
+	go mod download && echo OK
+	@printf "⇒ Tidy requirements: "
+	@CGO_ENABLED=0 \
+	go mod tidy -v && echo OK
+
+# Run `make %` in Golang container, for more information run `make help.docker/%`
+docker/%:
+	$(if $(filter $*,all $(BINS)), \
+		@echo "=> Running 'make $*' in clean Docker environment" && \
+		docker run --rm -t \
+		  -v `pwd`:/src \
+		  -w /src \
+		  -u `stat -c "%u:%g" .` \
+		  --env HOME=/src \
+		  golang:$(GO_VERSION) make $*,\
+	  	@echo "supported docker targets: all $(BINS) lint")
+
+# Run tests
+test:
+	@go test ./... -cover
+
+# Run tests with race detection and produce coverage output
+cover:
+	@go test -v -race ./... -coverprofile=coverage.txt -covermode=atomic
+	@go tool cover -html=coverage.txt -o coverage.html
+
+# Reformat code
+format:
+	@echo "⇒ Processing gofmt check"
+	@gofmt -s -w ./
+
+# Run linters
+lint:
+	@if [ ! -d "$(LINT_DIR)" ]; then \
+		make lint-install; \
+	fi
+	$(LINT_DIR)/golangci-lint run --timeout=5m
+
+# Install linters
+lint-install:
+	@rm -rf $(OUTPUT_LINT_DIR)
+	@mkdir -p $(OUTPUT_LINT_DIR)
+	@mkdir -p $(TMP_DIR)
+	@rm -rf $(TMP_DIR)/linters
+	@git -c advice.detachedHead=false clone --branch v$(TRUECLOUDLAB_LINT_VERSION) https://git.frostfs.info/TrueCloudLab/linters.git $(TMP_DIR)/linters
+	@@make -C $(TMP_DIR)/linters lib CGO_ENABLED=1 OUT_DIR=$(OUTPUT_LINT_DIR)
+	@rm -rf $(TMP_DIR)/linters
+	@rmdir $(TMP_DIR) 2>/dev/null || true
+	@CGO_ENABLED=1 GOBIN=$(LINT_DIR) go install -trimpath github.com/golangci/golangci-lint/cmd/golangci-lint@v$(LINT_VERSION)
+
+# Run linters in Docker
+docker/lint:
+	docker run --rm -it \
+	-v `pwd`:/src \
+	-u `stat -c "%u:%g" .` \
+	--env HOME=/src \
+	golangci/golangci-lint:v$(LINT_VERSION) bash -c 'cd /src/ && make lint'
+
+# Activate pre-commit hooks
+pre-commit:
+	pre-commit install -t pre-commit -t commit-msg
+
+# Deactivate pre-commit hooks
+unpre-commit:
+	pre-commit uninstall -t pre-commit -t commit-msg
+
+# Show current version
+version:
+	@echo $(VERSION)
+
+# Clean up files
+clean:
+	rm -rf .cache
+	rm -rf $(BINDIR)
+
+include help.mk

README.md

@@ -1,5 +1,5 @@
 <p align="center">
-<img src="./.github/logo.svg" width="500px" alt="FrostFS logo">
+<img src="./.forgejo/logo.svg" width="500px" alt="FrostFS logo">
 </p>
 <p align="center">
   <a href="https://go.k6.io/k6">k6</a> extension to test and benchmark FrostFS related protocols.
@@ -48,10 +48,11 @@ Create native client with `connect` method. Arguments:
 - dial timeout in seconds (0 for the default value)
 - stream timeout in seconds (0 for the default value)
 - generate object header on the client side (for big object - split locally too)
+- max size for generated object header on the client side (for big object - the size that the object is splitted into)
 
 ```js
 import native from 'k6/x/frostfs/native';
-const frostfs_cli = native.connect("s01.frostfs.devenv:8080", "", 0, 0, false)
+const frostfs_cli = native.connect("s01.frostfs.devenv:8080", "", 0, 0, false, 0)
 ```
 
 ### Methods
@@ -74,12 +75,13 @@ const frostfs_cli = native.connect("s01.frostfs.devenv:8080", "", 0, 0, false)
 
 Create a local client with `connect` method. Arguments:
 - local path to frostfs storage node configuration file
+- local path to frostfs storage node configuration directory
 - hex encoded private key (empty value produces random key)
 - whether to use the debug logger (warning: very verbose)
 
 ```js
 import local from 'k6/x/frostfs/local';
-const local_client = local.connect("/path/to/config.yaml", "", false)
+const local_client = local.connect("/path/to/config.yaml", "/path/to/config/dir", "", false)
 ```
 
 ### Methods
@@ -99,13 +101,13 @@ Credentials are taken from default AWS configuration files and ENVs.
 
 ```js
 import s3 from 'k6/x/frostfs/s3';
-const s3_cli = s3.connect("http://s3.frostfs.devenv:8080")
+const s3_cli = s3.connect("https://s3.frostfs.devenv:8080")
 ```
 
 You can also provide additional options:
 ```js
 import s3 from 'k6/x/frostfs/s3';
-const s3_cli = s3.connect("http://s3.frostfs.devenv:8080", {'no_verify_ssl': 'true', 'timeout': '60s'})
+const s3_cli = s3.connect("https://s3.frostfs.devenv:8080", {'no_verify_ssl': 'true', 'timeout': '60s'})
 ```
 
 * `no_verify_ss` - Bool. If `true` - skip verifying the s3 certificate chain and host name (useful if s3 uses self-signed certificates)
@@ -123,6 +125,7 @@ const s3_cli = s3.connect("http://s3.frostfs.devenv:8080", {'no_verify_ssl': 'tr
 
 Create local s3 client with `connect` method. Arguments:
 - local path to frostfs storage node configuration file
+- local path to frostfs storage node configuration directory
 - parameter map with the following options:
   * `hex_key`: private key to use as a hexadecimal string. A random one is created if none is provided.
   * `node_position`: position of this node in the node array if loading multiple nodes independently (default: 0).
@@ -135,7 +138,7 @@ Create local s3 client with `connect` method. Arguments:
 import local from 'k6/x/frostfs/local';
 const params = {'node_position': 1, 'node_count': 3}
 const bucketMapping = {'mytestbucket': 'GBQDDUM1hdodXmiRHV57EUkFWJzuntsG8BG15wFSwam6'}
-const local_client = local.connect("/path/to/config.yaml", params, bucketMapping)
+const local_client = local.connect("/path/to/config.yaml", "/path/to/config/dir", params, bucketMapping)
 ```
 
 ### Methods
@@ -148,6 +151,60 @@ const local_client = local.connect("/path/to/config.yaml", params, bucketMapping
 
 See native protocol and s3 test suite examples in [examples](./examples) dir.
 
+# Command line utils
+
+To build all command line utils just run:
+
+```shell
+$ make
+```
+
+All binaries will be in `bin` directory.
+
+## Export registry db
+
+You can export registry bolt db to json file, that can be used as pregen for scenarios (see [docs](./scenarios/run_scenarios.md)).
+To do this use `frostfs-xk6-registry-exporter`, available flags can be seen in help:
+
+```shell
+$ ./bin/frostfs-xk6-registry-exporter -h
+Registry exporter for xk6
+
+Usage:
+registry-exporter [flags]
+
+Examples:
+registry-exporter registry.bolt
+registry-exporter --status created --out out.json registry.bolt
+
+Flags:
+--age int         Object age
+--format string   Output format (default "json")
+-h, --help            help for registry-exporter
+--out string      Path to output file (default "dumped-registry.json")
+--status string   Object status (default "created")
+-v, --version         version for registry-exporter
+```
+
+## Import pregen into registry db
+
+You can import pregenerated json files into registry bolt db. Use `frostfs-xk6-registry import`. Usage examples are in help:
+
+```shell
+$ ./bin/frostfs-xk6-registry import -h
+Import objects into registry from pregenerated files
+
+Usage:
+  xk6-registry import [flags]
+
+Examples:
+xk6-registry import registry.bolt preset.json
+xk6-registry import registry.bolt preset.json another_preset.json
+
+Flags:
+  -h, --help   help for import
+```
+
 # License
 
 - [GNU General Public License v3.0](LICENSE)

cmd/xk6-registry-exporter/main.go

@@ -0,0 +1,18 @@
+package main
+
+import (
+	"context"
+	"os"
+	"os/signal"
+	"syscall"
+)
+
+func main() {
+	ctx, _ := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP)
+
+	if cmd, err := rootCmd.ExecuteContextC(ctx); err != nil {
+		cmd.PrintErrln("Error:", err.Error())
+		cmd.PrintErrf("Run '%v --help' for usage.\n", cmd.CommandPath())
+		os.Exit(1)
+	}
+}

cmd/xk6-registry-exporter/root.go

@@ -0,0 +1,89 @@
+package main
+
+import (
+	"fmt"
+	"runtime"
+
+	"git.frostfs.info/TrueCloudLab/xk6-frostfs/internal/registry"
+	"git.frostfs.info/TrueCloudLab/xk6-frostfs/internal/version"
+	"github.com/spf13/cobra"
+)
+
+var rootCmd = &cobra.Command{
+	Use:     "registry-exporter",
+	Version: version.Version,
+	Short:   "Registry exporter",
+	Long:    "Registry exporter for xk6",
+	Example: `registry-exporter registry.bolt
+registry-exporter --status created --out out.json registry.bolt`,
+	SilenceErrors: true,
+	SilenceUsage:  true,
+	RunE:          rootCmdRun,
+}
+
+const (
+	outFlag    = "out"
+	formatFlag = "format"
+	statusFlag = "status"
+	ageFlag    = "age"
+)
+
+const (
+	defaultOutPath = "dumped-registry.json"
+
+	jsonFormat = "json"
+
+	createdStatus = "created"
+)
+
+func init() {
+	rootCmd.Flags().String(outFlag, defaultOutPath, "Path to output file")
+	rootCmd.Flags().String(formatFlag, jsonFormat, "Output format")
+	rootCmd.Flags().String(statusFlag, createdStatus, "Object status")
+	rootCmd.Flags().Int(ageFlag, 0, "Object age")
+
+	cobra.AddTemplateFunc("runtimeVersion", runtime.Version)
+	rootCmd.SetVersionTemplate(`FrostFS xk6 Registry Exporter
+{{printf "Version: %s" .Version }}
+GoVersion: {{ runtimeVersion }}
+`)
+}
+
+func rootCmdRun(cmd *cobra.Command, args []string) error {
+	if len(args) != 1 {
+		return fmt.Errorf("expected exacly one non-flag argumet: path to the registry, got: %s", args)
+	}
+
+	format, err := cmd.Flags().GetString(formatFlag)
+	if err != nil {
+		return fmt.Errorf("get '%s' flag: %w", formatFlag, err)
+	}
+	if format != jsonFormat {
+		return fmt.Errorf("unknown format '%s', only '%s' is supported", format, jsonFormat)
+	}
+
+	out, err := cmd.Flags().GetString(outFlag)
+	if err != nil {
+		return fmt.Errorf("get '%s' flag: %w", outFlag, err)
+	}
+
+	status, err := cmd.Flags().GetString(statusFlag)
+	if err != nil {
+		return fmt.Errorf("get '%s' flag: %w", statusFlag, err)
+	}
+
+	age, err := cmd.Flags().GetInt(ageFlag)
+	if err != nil {
+		return fmt.Errorf("get '%s' flag: %w", ageFlag, err)
+	}
+
+	objRegistry := registry.NewObjRegistry(cmd.Context(), args[0])
+	objSelector := registry.NewObjSelector(objRegistry, 0, registry.SelectorAwaiting, &registry.ObjFilter{
+		Status: status,
+		Age:    age,
+	})
+	objExporter := registry.NewObjExporter(objSelector)
+
+	cmd.Println("Writing result file:", out)
+	return objExporter.ExportJSONPreGen(out)
+}

cmd/xk6-registry/importer/import.go

@@ -0,0 +1,55 @@
+package importer
+
+import (
+	"encoding/json"
+	"os"
+
+	"git.frostfs.info/TrueCloudLab/xk6-frostfs/internal/registry"
+)
+
+type PreGenObj struct {
+	Bucket    string `json:"bucket"`
+	Object    string `json:"object"`
+	Container string `json:"container"`
+}
+
+type PreGenerateInfo struct {
+	Buckets    []string    `json:"buckets"`
+	Containers []string    `json:"containers"`
+	Objects    []PreGenObj `json:"objects"`
+	ObjSize    string      `json:"obj_size"`
+}
+
+// ImportJSONPreGen writes objects from pregenerated JSON file
+// to the registry.
+// Note that ImportJSONPreGen does not check if object already
+// exists in the registry so in case of re-entry the registry
+// will have two entities representing the same object.
+func ImportJSONPreGen(o *registry.ObjRegistry, filename string) error {
+	f, err := os.ReadFile(filename)
+	if err != nil {
+		return err
+	}
+
+	var pregenInfo PreGenerateInfo
+	err = json.Unmarshal(f, &pregenInfo)
+	if err != nil {
+		return err
+	}
+
+	// AddObject uses DB.Batch to combine concurrent Batch calls
+	// into a single Bolt transaction. DB.Batch is limited by
+	// DB.MaxBatchDelay which may affect perfomance.
+	for _, obj := range pregenInfo.Objects {
+		if obj.Bucket != "" {
+			err = o.AddObject("", "", obj.Bucket, obj.Object, "")
+		} else {
+			err = o.AddObject(obj.Container, obj.Object, "", "", "")
+		}
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

cmd/xk6-registry/importer/root.go

@@ -0,0 +1,27 @@
+package importer
+
+import (
+	"git.frostfs.info/TrueCloudLab/xk6-frostfs/internal/registry"
+	"github.com/spf13/cobra"
+)
+
+// Cmd represents the import command.
+var Cmd = &cobra.Command{
+	Use:   "import",
+	Short: "Import objects into registry",
+	Long:  "Import objects into registry from pregenerated files",
+	Example: `xk6-registry import registry.bolt preset.json
+xk6-registry import registry.bolt preset.json another_preset.json`,
+	RunE: runCmd,
+	Args: cobra.MinimumNArgs(2),
+}
+
+func runCmd(cmd *cobra.Command, args []string) error {
+	objRegistry := registry.NewObjRegistry(cmd.Context(), args[0])
+	for i := 1; i < len(args); i++ {
+		if err := ImportJSONPreGen(objRegistry, args[i]); err != nil {
+			return err
+		}
+	}
+	return nil
+}

cmd/xk6-registry/main.go

@@ -0,0 +1,18 @@
+package main
+
+import (
+	"context"
+	"os"
+	"os/signal"
+	"syscall"
+)
+
+func main() {
+	ctx, _ := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP)
+
+	if cmd, err := rootCmd.ExecuteContextC(ctx); err != nil {
+		cmd.PrintErrln("Error:", err.Error())
+		cmd.PrintErrf("Run '%v --help' for usage.\n", cmd.CommandPath())
+		os.Exit(1)
+	}
+}

cmd/xk6-registry/root.go

@@ -0,0 +1,33 @@
+package main
+
+import (
+	"runtime"
+
+	"git.frostfs.info/TrueCloudLab/xk6-frostfs/cmd/xk6-registry/importer"
+	"git.frostfs.info/TrueCloudLab/xk6-frostfs/internal/version"
+	"github.com/spf13/cobra"
+)
+
+var rootCmd = &cobra.Command{
+	Use:     "xk6-registry",
+	Version: version.Version,
+	Short:   "Command Line Tool to work with Registry",
+	Long: `Registry provides tools to work with object registry for xk6.
+It contains command for importing objects in registry from preset`,
+	SilenceErrors: true,
+	SilenceUsage:  true,
+	Run:           rootCmdRun,
+}
+
+func init() {
+	cobra.AddTemplateFunc("runtimeVersion", runtime.Version)
+	rootCmd.SetVersionTemplate(`FrostFS xk6-registry
+{{printf "Version: %s" .Version }}
+GoVersion: {{ runtimeVersion }}
+`)
+	rootCmd.AddCommand(importer.Cmd)
+}
+
+func rootCmdRun(cmd *cobra.Command, _ []string) {
+	_ = cmd.Usage()
+}

examples/local.js

@@ -2,7 +2,7 @@ import local from 'k6/x/frostfs/local';
 import { uuidv4 } from '../scenarios/libs/k6-utils-1.4.0.js';
 
 const payload = open('../go.sum', 'b');
-const local_cli = local.connect("/path/to/config.yaml", "", false)
+const local_cli = local.connect("/path/to/config.yaml", "/path/to/config/dir", "", false)
 
 export const options = {
     stages: [

examples/native.js

@@ -3,11 +3,11 @@ import { fail } from "k6";
 import { uuidv4 } from '../scenarios/libs/k6-utils-1.4.0.js';
 
 const payload = open('../go.sum', 'b');
-const frostfs_cli = native.connect("s01.frostfs.devenv:8080", "1dd37fba80fec4e6a6f13fd708d8dcb3b29def768017052f6c930fa1c5d90bbb", 0, 0, false)
+const frostfs_cli = native.connect("s01.frostfs.devenv:8080", "1dd37fba80fec4e6a6f13fd708d8dcb3b29def768017052f6c930fa1c5d90bbb", 0, 0, false, 0)
 
 export const options = {
     stages: [
-        {duration: '30s', target: 10},
+        { duration: '30s', target: 10 },
     ],
 };
 
@@ -24,7 +24,7 @@ export function setup() {
         fail(res.error)
     }
     console.info("created container", res.container_id)
-    return {container_id: res.container_id}
+    return { container_id: res.container_id }
 }
 
 export default function (data) {

examples/native_onsite.js

@@ -3,7 +3,7 @@ import { uuidv4 } from '../scenarios/libs/k6-utils-1.4.0.js';
 
 const payload = open('../go.sum', 'b');
 const container = "AjSxSNNXbJUDPqqKYm1VbFVDGCakbpUNH8aGjPmGAH3B"
-const frostfs_cli = native.connect("s01.frostfs.devenv:8080", "", 0, 0, false)
+const frostfs_cli = native.connect("s01.frostfs.devenv:8080", "", 0, 0, false, 0)
 const frostfs_obj = frostfs_cli.onsite(container, payload)
 
 export const options = {
@@ -14,11 +14,11 @@ export const options = {
 
 export default function () {
     let headers = {
-       'unique_header': uuidv4()
+        'unique_header': uuidv4()
     }
     let resp = frostfs_obj.put(headers)
     if (resp.success) {
-       frostfs_cli.get(container, resp.object_id)
+        frostfs_cli.get(container, resp.object_id)
     } else {
         console.log(resp.error)
     }

examples/s3local.js

@@ -3,7 +3,7 @@ import { uuidv4 } from '../scenarios/libs/k6-utils-1.4.0.js';
 
 const bucket = "testbucket"
 const payload = open('../go.sum', 'b');
-const s3local_cli = s3local.connect("path/to/storage/config.yml", {}, {
+const s3local_cli = s3local.connect("path/to/storage/config.yml", "path/to/storage/config/dir", {}, {
     'testbucket': 'GBQDDUM1hdodXmiRHV57EUkFWJzuntsG8BG15wFSwam6',
 });
 

go.mod

@@ -1,126 +1,132 @@
 module git.frostfs.info/TrueCloudLab/xk6-frostfs
 
-go 1.19
+go 1.22
 
 require (
-	git.frostfs.info/TrueCloudLab/frostfs-node v0.22.2-0.20230704155826-b520a3049e6f
-	git.frostfs.info/TrueCloudLab/frostfs-s3-gw v0.27.0-rc.2
-	git.frostfs.info/TrueCloudLab/frostfs-sdk-go v0.0.0-20230705125206-769f6eec0565
+	git.frostfs.info/TrueCloudLab/frostfs-node v0.38.3-0.20240502170333-ec2873caa7c6
+	git.frostfs.info/TrueCloudLab/frostfs-s3-gw v0.29.0-rc.1.0.20240422122918-034396d554ec
+	git.frostfs.info/TrueCloudLab/frostfs-sdk-go v0.0.0-20240502080121-12ddefe07877
 	git.frostfs.info/TrueCloudLab/tzhash v1.8.0
-	github.com/aws/aws-sdk-go-v2 v1.18.1
-	github.com/aws/aws-sdk-go-v2/config v1.18.27
-	github.com/aws/aws-sdk-go-v2/service/s3 v1.36.0
+	github.com/aws/aws-sdk-go-v2 v1.19.0
+	github.com/aws/aws-sdk-go-v2/config v1.18.28
+	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.11.72
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.37.0
 	github.com/dop251/goja v0.0.0-20230626124041-ba8a63e79201
 	github.com/go-loremipsum/loremipsum v1.1.3
-	github.com/google/uuid v1.3.0
+	github.com/google/uuid v1.6.0
 	github.com/joho/godotenv v1.5.1
-	github.com/nspcc-dev/neo-go v0.101.2
-	github.com/panjf2000/ants/v2 v2.8.0
+	github.com/nspcc-dev/neo-go v0.105.1
+	github.com/panjf2000/ants/v2 v2.9.0
 	github.com/sirupsen/logrus v1.9.3
+	github.com/spf13/cobra v1.8.0
 	github.com/stretchr/testify v1.8.4
-	go.etcd.io/bbolt v1.3.7
-	go.k6.io/k6 v0.45.0
-	go.uber.org/zap v1.24.0
-	golang.org/x/sys v0.10.0
+	go.etcd.io/bbolt v1.3.8
+	go.k6.io/k6 v0.45.1
+	go.uber.org/zap v1.26.0
+	golang.org/x/sys v0.18.0
 )
 
 require (
-	git.frostfs.info/TrueCloudLab/frostfs-api-go/v2 v2.15.1-0.20230602142716-68021b910acb // indirect
+	git.frostfs.info/TrueCloudLab/frostfs-api-go/v2 v2.16.1-0.20240427200446-67c6f305b21f // indirect
+	git.frostfs.info/TrueCloudLab/frostfs-contract v0.19.3-0.20240409115729-6eb492025bdd // indirect
 	git.frostfs.info/TrueCloudLab/frostfs-crypto v0.6.0 // indirect
-	git.frostfs.info/TrueCloudLab/frostfs-observability v0.0.0-20230531082742-c97d21411eb6 // indirect
+	git.frostfs.info/TrueCloudLab/frostfs-observability v0.0.0-20231101111734-b3ad3335ff65 // indirect
 	git.frostfs.info/TrueCloudLab/hrw v1.2.1 // indirect
+	git.frostfs.info/TrueCloudLab/policy-engine v0.0.0-20240416071728-04a79f57ef1f // indirect
 	git.frostfs.info/TrueCloudLab/rfc6979 v0.4.0 // indirect
+	git.frostfs.info/TrueCloudLab/zapjournald v0.0.0-20240124114243-cb2e66427d02 // indirect
 	github.com/antlr4-go/antlr/v4 v4.13.0 // indirect
 	github.com/aws/aws-sdk-go v1.44.296 // indirect
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.4.10 // indirect
-	github.com/aws/aws-sdk-go-v2/credentials v1.13.26 // indirect
-	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.13.4 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.34 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.28 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/ini v1.3.35 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/v4a v1.0.26 // indirect
+	github.com/aws/aws-sdk-go-v2/credentials v1.13.27 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.13.5 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.35 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.29 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.3.36 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/v4a v1.0.27 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.9.11 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.1.29 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.28 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.14.3 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sso v1.12.12 // indirect
-	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.14.12 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sts v1.19.2 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.1.30 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.29 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.14.4 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.12.13 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.14.13 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.19.3 // indirect
 	github.com/aws/smithy-go v1.13.5 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bluele/gcache v0.0.2 // indirect
 	github.com/cenkalti/backoff/v4 v4.2.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
-	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0 // indirect
 	github.com/dlclark/regexp2 v1.10.0 // indirect
 	github.com/fatih/color v1.15.0 // indirect
-	github.com/fsnotify/fsnotify v1.6.0 // indirect
-	github.com/go-logr/logr v1.2.4 // indirect
+	github.com/fsnotify/fsnotify v1.7.0 // indirect
+	github.com/go-chi/chi/v5 v5.0.8 // indirect
+	github.com/go-logr/logr v1.4.1 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/go-pkgz/expirable-cache/v3 v3.0.0 // indirect
 	github.com/go-sourcemap/sourcemap v2.1.4-0.20211119122758-180fcef48034+incompatible // indirect
-	github.com/golang/protobuf v1.5.3 // indirect
-	github.com/google/pprof v0.0.0-20230705174524-200ffdc848b8 // indirect
-	github.com/gorilla/mux v1.8.0 // indirect
-	github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0 // indirect
-	github.com/hashicorp/golang-lru v0.6.0 // indirect
-	github.com/hashicorp/golang-lru/v2 v2.0.4 // indirect
+	github.com/golang/snappy v0.0.4 // indirect
+	github.com/google/pprof v0.0.0-20230817174616-7a8ec2ada47b // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.19.0 // indirect
+	github.com/hashicorp/golang-lru v1.0.2 // indirect
+	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
+	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
-	github.com/klauspost/compress v1.16.7 // indirect
+	github.com/klauspost/compress v1.17.4 // indirect
 	github.com/magiconair/properties v1.8.7 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.19 // indirect
-	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
-	github.com/minio/highwayhash v1.0.2 // indirect
 	github.com/minio/sio v0.3.1 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/mr-tron/base58 v1.2.0 // indirect
 	github.com/mstoykov/atlas v0.0.0-20220811071828-388f114305dd // indirect
-	github.com/nats-io/jwt/v2 v2.4.1 // indirect
-	github.com/nats-io/nats.go v1.27.1 // indirect
-	github.com/nats-io/nkeys v0.4.4 // indirect
+	github.com/nats-io/nats.go v1.32.0 // indirect
+	github.com/nats-io/nkeys v0.4.7 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
+	github.com/nspcc-dev/go-ordered-json v0.0.0-20240112074137-296698a162ae // indirect
 	github.com/nspcc-dev/rfc6979 v0.2.0 // indirect
-	github.com/onsi/ginkgo v1.16.5 // indirect
 	github.com/onsi/gomega v1.20.2 // indirect
-	github.com/pelletier/go-toml/v2 v2.0.8 // indirect
-	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/prometheus/client_golang v1.16.0 // indirect
-	github.com/prometheus/client_model v0.4.0 // indirect
-	github.com/prometheus/common v0.44.0 // indirect
-	github.com/prometheus/procfs v0.11.0 // indirect
+	github.com/pelletier/go-toml/v2 v2.1.1 // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
+	github.com/prometheus/client_golang v1.18.0 // indirect
+	github.com/prometheus/client_model v0.5.0 // indirect
+	github.com/prometheus/common v0.46.0 // indirect
+	github.com/prometheus/procfs v0.12.0 // indirect
+	github.com/sagikazarmark/locafero v0.4.0 // indirect
+	github.com/sagikazarmark/slog-shim v0.1.0 // indirect
 	github.com/serenize/snaker v0.0.0-20201027110005-a7ad2135616e // indirect
-	github.com/spf13/afero v1.9.5 // indirect
-	github.com/spf13/cast v1.5.1 // indirect
-	github.com/spf13/jwalterweatherman v1.1.0 // indirect
+	github.com/sourcegraph/conc v0.3.0 // indirect
+	github.com/spf13/afero v1.11.0 // indirect
+	github.com/spf13/cast v1.6.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
-	github.com/spf13/viper v1.16.0 // indirect
-	github.com/subosito/gotenv v1.4.2 // indirect
+	github.com/spf13/viper v1.18.2 // indirect
+	github.com/ssgreg/journald v1.0.0 // indirect
+	github.com/subosito/gotenv v1.6.0 // indirect
+	github.com/syndtr/goleveldb v1.0.1-0.20220721030215-126854af5e6d // indirect
 	github.com/twmb/murmur3 v1.1.8 // indirect
-	go.opentelemetry.io/otel v1.16.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/internal/retry v1.16.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.16.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.16.0 // indirect
-	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.16.0 // indirect
-	go.opentelemetry.io/otel/metric v1.16.0 // indirect
-	go.opentelemetry.io/otel/sdk v1.16.0 // indirect
-	go.opentelemetry.io/otel/trace v1.16.0 // indirect
-	go.opentelemetry.io/proto/otlp v0.20.0 // indirect
-	go.uber.org/atomic v1.11.0 // indirect
+	go.opentelemetry.io/otel v1.22.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.22.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.22.0 // indirect
+	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.22.0 // indirect
+	go.opentelemetry.io/otel/metric v1.22.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.22.0 // indirect
+	go.opentelemetry.io/otel/trace v1.22.0 // indirect
+	go.opentelemetry.io/proto/otlp v1.1.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	golang.org/x/crypto v0.11.0 // indirect
-	golang.org/x/exp v0.0.0-20230626212559-97b1e661b5df // indirect
-	golang.org/x/net v0.12.0 // indirect
-	golang.org/x/sync v0.3.0 // indirect
-	golang.org/x/text v0.11.0 // indirect
-	golang.org/x/time v0.3.0 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20230629202037-9506855d4529 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20230629202037-9506855d4529 // indirect
-	google.golang.org/grpc v1.56.1 // indirect
-	google.golang.org/protobuf v1.31.0 // indirect
+	golang.org/x/crypto v0.21.0 // indirect
+	golang.org/x/exp v0.0.0-20240119083558-1b970713d09a // indirect
+	golang.org/x/net v0.23.0 // indirect
+	golang.org/x/sync v0.6.0 // indirect
+	golang.org/x/text v0.14.0 // indirect
+	golang.org/x/time v0.5.0 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20240227224415-6ceb2ff114de // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240227224415-6ceb2ff114de // indirect
+	google.golang.org/grpc v1.63.2 // indirect
+	google.golang.org/protobuf v1.33.0 // indirect
 	gopkg.in/guregu/null.v3 v3.5.0 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect

help.mk

@@ -0,0 +1,22 @@
+.PHONY: help
+
+# Show this help prompt
+help:
+	@echo '  Usage:'
+	@echo ''
+	@echo '    make <target>'
+	@echo ''
+	@echo '  Targets:'
+	@echo ''
+	@awk '/^#/{ comment = substr($$0,3) } comment && /^[a-zA-Z][a-zA-Z0-9.%_/-]+ ?:/{ print "   ", $$1, comment }' $(MAKEFILE_LIST) | column -t -s ':' | grep -v 'IGNORE' | sort | uniq
+
+# Show help for docker/% IGNORE
+help.docker/%:
+	$(eval TARGETS:=$(notdir all lint) ${BINS})
+	@echo '  Usage:'
+	@echo ''
+	@echo '    make docker/% -- Run `make %` in Golang container'
+	@echo ''
+	@echo '  Supported docker targets:'
+	@echo ''
+	@$(foreach bin, $(TARGETS), echo '   ' $(bin);)

internal/datagen/datagen.go

@@ -38,7 +38,7 @@ func (d *Datagen) Exports() modules.Exports {
 	return modules.Exports{Default: d}
 }
 
-func (d *Datagen) Generator(size int, typ string) *Generator {
-	g := NewGenerator(d.vu, size, strings.ToLower(typ))
+func (d *Datagen) Generator(size int, typ string, streaming bool) *Generator {
+	g := NewGenerator(d.vu, size, strings.ToLower(typ), streaming)
 	return &g
 }

internal/datagen/generator.go

@@ -2,12 +2,10 @@ package datagen
 
 import (
 	"bytes"
-	"crypto/sha256"
-	"encoding/hex"
 	"math/rand"
+	"sync/atomic"
 	"time"
 
-	"github.com/dop251/goja"
 	"github.com/go-loremipsum/loremipsum"
 	"go.k6.io/k6/js/modules"
 )
@@ -28,11 +26,9 @@ type (
 		buf    []byte
 		typ    string
 		offset int
-	}
 
-	GenPayloadResponse struct {
-		Payload goja.ArrayBuffer
-		Hash    string
+		streaming bool
+		seed      *atomic.Int64
 	}
 )
 
@@ -45,7 +41,7 @@ var payloadTypes = []string{
 	"",
 }
 
-func NewGenerator(vu modules.VU, size int, typ string) Generator {
+func NewGenerator(vu modules.VU, size int, typ string, streaming bool) Generator {
 	if size <= 0 {
 		panic("size should be positive")
 	}
@@ -60,17 +56,20 @@ func NewGenerator(vu modules.VU, size int, typ string) Generator {
 	if !found {
 		vu.InitEnv().Logger.Info("Unknown payload type '%s', random will be used.", typ)
 	}
-
-	r := rand.New(rand.NewSource(time.Now().UnixNano()))
-	buf := make([]byte, size+TailSize)
 	g := Generator{
 		vu:   vu,
 		size: size,
-		rand: r,
-		buf:  buf,
 		typ:  typ,
 	}
-	g.fillBuffer()
+
+	if streaming {
+		g.streaming = true
+		g.seed = new(atomic.Int64)
+	} else {
+		g.rand = rand.New(rand.NewSource(time.Now().UnixNano()))
+		g.buf = make([]byte, size+TailSize)
+		g.fillBuffer()
+	}
 	return g
 }
 
@@ -85,21 +84,17 @@ func (g *Generator) fillBuffer() {
 		}
 		g.buf = b.Bytes()
 	default:
-		rand.Read(g.buf) // Per docs, err is always nil here
+		g.rand.Read(g.buf) // Per docs, err is always nil here
 	}
 }
 
-func (g *Generator) GenPayload(calcHash bool) GenPayloadResponse {
-	data := g.nextSlice()
-
-	dataHash := ""
-	if calcHash {
-		hashBytes := sha256.Sum256(data)
-		dataHash = hex.EncodeToString(hashBytes[:])
+func (g *Generator) GenPayload() Payload {
+	if g.streaming {
+		return NewStreamPayload(g.size, g.seed.Add(1), g.typ)
 	}
 
-	payload := g.vu.Runtime().NewArrayBuffer(data)
-	return GenPayloadResponse{Payload: payload, Hash: dataHash}
+	data := g.nextSlice()
+	return NewFixedPayload(data)
 }
 
 func (g *Generator) nextSlice() []byte {

internal/datagen/generator_test.go

@@ -16,25 +16,25 @@ func TestGenerator(t *testing.T) {
 
 	t.Run("fails on negative size", func(t *testing.T) {
 		require.Panics(t, func() {
-			_ = NewGenerator(vu, -1, "")
+			_ = NewGenerator(vu, -1, "", false)
 		})
 	})
 
 	t.Run("fails on zero size", func(t *testing.T) {
 		require.Panics(t, func() {
-			_ = NewGenerator(vu, 0, "")
+			_ = NewGenerator(vu, 0, "", false)
 		})
 	})
 
 	t.Run("creates slice of specified size", func(t *testing.T) {
 		size := 10
-		g := NewGenerator(vu, size, "")
+		g := NewGenerator(vu, size, "", false)
 		slice := g.nextSlice()
 		require.Len(t, slice, size)
 	})
 
 	t.Run("creates a different slice on each call", func(t *testing.T) {
-		g := NewGenerator(vu, 1000, "")
+		g := NewGenerator(vu, 1000, "", false)
 		slice1 := g.nextSlice()
 		slice2 := g.nextSlice()
 		// Each slice should be unique (assuming that 1000 random bytes will never coincide
@@ -43,7 +43,7 @@ func TestGenerator(t *testing.T) {
 	})
 
 	t.Run("keeps generating slices after consuming entire tail", func(t *testing.T) {
-		g := NewGenerator(vu, 1000, "")
+		g := NewGenerator(vu, 1000, "", false)
 		initialSlice := g.nextSlice()
 		for i := 0; i < TailSize; i++ {
 			g.nextSlice()

internal/datagen/payload.go

@@ -0,0 +1,121 @@
+package datagen
+
+import (
+	"bufio"
+	"bytes"
+	"crypto/sha256"
+	"encoding/hex"
+	"hash"
+	"io"
+	"math/rand"
+
+	"github.com/go-loremipsum/loremipsum"
+)
+
+// Payload represents arbitrary data to be packed into S3 or native object.
+// Implementations could be thread-unsafe.
+type Payload interface {
+	// Reader returns io.Reader instance to read the payload.
+	// Must not be called twice.
+	Reader() io.Reader
+	// Bytes is a helper which reads all data from Reader() into slice.
+	// The sole purpose of this method is to simplify HTTP scenario,
+	// where all payload needs to be read and wrapped.
+	Bytes() []byte
+	// Size returns payload size, which is equal to the total amount of data
+	// that could be read from the Reader().
+	Size() int
+	// Hash returns payload sha256 hash. Must be called after all data is read from the reader.
+	Hash() string
+}
+
+type bytesPayload struct {
+	data []byte
+}
+
+func (p *bytesPayload) Reader() io.Reader {
+	return bytes.NewReader(p.data)
+}
+
+func (p *bytesPayload) Size() int {
+	return len(p.data)
+}
+
+func (p *bytesPayload) Hash() string {
+	h := sha256.Sum256(p.data[:])
+	return hex.EncodeToString(h[:])
+}
+
+func (p *bytesPayload) Bytes() []byte {
+	return p.data
+}
+
+func NewFixedPayload(data []byte) Payload {
+	return &bytesPayload{data: data}
+}
+
+type randomPayload struct {
+	r    io.Reader
+	s    hash.Hash
+	h    string
+	size int
+}
+
+func NewStreamPayload(size int, seed int64, typ string) Payload {
+	var rr io.Reader
+	switch typ {
+	case "text":
+		rr = &textReader{li: loremipsum.NewWithSeed(seed)}
+	default:
+		rr = rand.New(rand.NewSource(seed))
+	}
+
+	lr := io.LimitReader(rr, int64(size))
+	// We need some buffering to write complete blocks in the TeeReader.
+	// Streaming payload read is expected to be used for big objects, thus 4k seems like a good choice.
+	br := bufio.NewReaderSize(lr, 4096)
+	s := sha256.New()
+	tr := io.TeeReader(br, s)
+	return &randomPayload{
+		r:    tr,
+		s:    s,
+		size: size,
+	}
+}
+
+func (p *randomPayload) Reader() io.Reader {
+	return p.r
+}
+
+func (p *randomPayload) Size() int {
+	return p.size
+}
+
+func (p *randomPayload) Hash() string {
+	if p.h == "" {
+		p.h = hex.EncodeToString(p.s.Sum(nil))
+		// Prevent possible misuse.
+		p.r = nil
+		p.s = nil
+	}
+	return p.h
+}
+
+func (p *randomPayload) Bytes() []byte {
+	data, err := io.ReadAll(p.r)
+	if err != nil {
+		// We use only 2 readers, either `bytes.Reader` or `rand.Reader`.
+		// None of them returns errors, thus encountering an error is a fatal error.
+		panic(err)
+	}
+	return data
+}
+
+type textReader struct {
+	li *loremipsum.LoremIpsum
+}
+
+func (r *textReader) Read(p []byte) (n int, err error) {
+	paragraph := r.li.Paragraph()
+	return copy(p, paragraph), nil
+}

internal/datagen/payload_test.go

@@ -0,0 +1,40 @@
+package datagen
+
+import (
+	"crypto/rand"
+	"crypto/sha256"
+	"encoding/hex"
+	"io"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestFixedPayload(t *testing.T) {
+	const size = 123
+	data := make([]byte, size)
+	_, err := rand.Read(data)
+	require.NoError(t, err)
+
+	p := NewFixedPayload(data)
+	require.Equal(t, size, p.Size())
+
+	actual, err := io.ReadAll(p.Reader())
+	require.NoError(t, err)
+	require.Equal(t, data, actual)
+
+	h := sha256.Sum256(data)
+	require.Equal(t, hex.EncodeToString(h[:]), p.Hash())
+}
+
+func TestStreamingPayload(t *testing.T) {
+	const size = 123
+
+	p := NewStreamPayload(size, 0, "")
+	require.Equal(t, size, p.Size())
+
+	actual, err := io.ReadAll(p.Reader())
+	require.NoError(t, err)
+	require.Equal(t, size, len(actual))
+	require.Equal(t, sha256.Size*2, len(p.Hash()))
+}

internal/local/client.go

@@ -5,14 +5,15 @@ import (
 
 	cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
 	oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
+	"git.frostfs.info/TrueCloudLab/xk6-frostfs/internal/datagen"
 	"git.frostfs.info/TrueCloudLab/xk6-frostfs/internal/local/rawclient"
-	"github.com/dop251/goja"
 	"go.k6.io/k6/js/modules"
 )
 
 type Client struct {
 	vu modules.VU
 	rc *rawclient.RawClient
+	l  Limiter
 }
 
 type (
@@ -25,13 +26,21 @@ type (
 		Success  bool
 		ObjectID string
 		Error    string
+		Abort    bool
 	}
 
 	GetResponse    SuccessOrErrorResponse
 	DeleteResponse SuccessOrErrorResponse
 )
 
-func (c *Client) Put(containerID string, headers map[string]string, payload goja.ArrayBuffer) PutResponse {
+func (c *Client) Put(containerID string, headers map[string]string, payload datagen.Payload) PutResponse {
+	if c.l.IsFull() {
+		return PutResponse{
+			Success: false,
+			Error:   "engine size limit reached",
+			Abort:   true,
+		}
+	}
 	id, err := c.rc.Put(c.vu.Context(), mustParseContainerID(containerID), nil, headers, payload.Bytes())
 	if err != nil {
 		return PutResponse{Error: err.Error()}

internal/local/limiter.go

@@ -0,0 +1,105 @@
+package local
+
+import (
+	"sync/atomic"
+	"time"
+
+	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/engine"
+	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard/mode"
+	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/metrics"
+)
+
+var (
+	_ Limiter = &noopLimiter{}
+	_ Limiter = &sizeLimiter{}
+)
+
+type Limiter interface {
+	engine.MetricRegister
+	IsFull() bool
+}
+
+func NewLimiter(maxSizeGB int64) Limiter {
+	if maxSizeGB < 0 {
+		panic("max size is negative")
+	}
+	if maxSizeGB == 0 {
+		return &noopLimiter{}
+	}
+	return &sizeLimiter{
+		maxSize:     maxSizeGB * 1024 * 1024 * 1024,
+		currentSize: &atomic.Int64{},
+	}
+}
+
+type sizeLimiter struct {
+	maxSize     int64
+	currentSize *atomic.Int64
+}
+
+func (*sizeLimiter) AddMethodDuration(method string, d time.Duration)                {}
+func (*sizeLimiter) AddToContainerSize(cnrID string, size int64)                     {}
+func (*sizeLimiter) AddToObjectCounter(shardID string, objectType string, delta int) {}
+func (*sizeLimiter) ClearErrorCounter(shardID string)                                {}
+func (*sizeLimiter) DeleteShardMetrics(shardID string)                               {}
+func (*sizeLimiter) GC() metrics.GCMetrics                                           { return &noopGCMetrics{} }
+func (*sizeLimiter) IncErrorCounter(shardID string)                                  {}
+func (*sizeLimiter) SetMode(shardID string, mode mode.Mode)                          {}
+func (*sizeLimiter) SetObjectCounter(shardID string, objectType string, v uint64)    {}
+func (*sizeLimiter) WriteCache() metrics.WriteCacheMetrics                           { return &noopWriteCacheMetrics{} }
+func (*sizeLimiter) DeleteContainerSize(cnrID string)                                {}
+func (*sizeLimiter) DeleteContainerCount(cnrID string)                               {}
+func (*sizeLimiter) SetContainerObjectCounter(_, _, _ string, _ uint64)              {}
+func (*sizeLimiter) IncContainerObjectCounter(_, _, _ string)                        {}
+func (*sizeLimiter) SubContainerObjectCounter(_, _, _ string, _ uint64)              {}
+func (*sizeLimiter) IncRefillObjectsCount(_, _ string, _ int, _ bool)                {}
+func (*sizeLimiter) SetRefillPercent(_, _ string, _ uint32)                          {}
+func (*sizeLimiter) SetRefillStatus(_, _, _ string)                                  {}
+
+func (sl *sizeLimiter) AddToPayloadCounter(shardID string, size int64) {
+	sl.currentSize.Add(size)
+}
+
+func (sl *sizeLimiter) IsFull() bool {
+	cur := sl.currentSize.Load()
+	return cur > sl.maxSize
+}
+
+type noopLimiter struct{}
+
+func (*noopLimiter) AddMethodDuration(method string, d time.Duration)                {}
+func (*noopLimiter) AddToContainerSize(cnrID string, size int64)                     {}
+func (*noopLimiter) AddToObjectCounter(shardID string, objectType string, delta int) {}
+func (*noopLimiter) AddToPayloadCounter(shardID string, size int64)                  {}
+func (*noopLimiter) ClearErrorCounter(shardID string)                                {}
+func (*noopLimiter) DeleteShardMetrics(shardID string)                               {}
+func (*noopLimiter) GC() metrics.GCMetrics                                           { return &noopGCMetrics{} }
+func (*noopLimiter) IncErrorCounter(shardID string)                                  {}
+func (*noopLimiter) SetMode(shardID string, mode mode.Mode)                          {}
+func (*noopLimiter) SetObjectCounter(shardID string, objectType string, v uint64)    {}
+func (*noopLimiter) WriteCache() metrics.WriteCacheMetrics                           { return &noopWriteCacheMetrics{} }
+func (*noopLimiter) IsFull() bool                                                    { return false }
+func (*noopLimiter) DeleteContainerSize(cnrID string)                                {}
+func (*noopLimiter) DeleteContainerCount(cnrID string)                               {}
+func (*noopLimiter) SetContainerObjectCounter(_, _, _ string, _ uint64)              {}
+func (*noopLimiter) IncContainerObjectCounter(_, _, _ string)                        {}
+func (*noopLimiter) SubContainerObjectCounter(_, _, _ string, _ uint64)              {}
+func (*noopLimiter) IncRefillObjectsCount(_, _ string, _ int, _ bool)                {}
+func (*noopLimiter) SetRefillPercent(_, _ string, _ uint32)                          {}
+func (*noopLimiter) SetRefillStatus(_, _, _ string)                                  {}
+
+type noopGCMetrics struct{}
+
+func (*noopGCMetrics) AddDeletedCount(shardID string, deleted uint64, failed uint64)          {}
+func (*noopGCMetrics) AddExpiredObjectCollectionDuration(string, time.Duration, bool, string) {}
+func (*noopGCMetrics) AddInhumedObjectCount(shardID string, count uint64, objectType string)  {}
+func (*noopGCMetrics) AddRunDuration(shardID string, d time.Duration, success bool)           {}
+
+type noopWriteCacheMetrics struct{}
+
+func (*noopWriteCacheMetrics) AddMethodDuration(_, _, _, _ string, _ bool, _ time.Duration) {}
+func (*noopWriteCacheMetrics) Close(_, _ string)                                            {}
+func (*noopWriteCacheMetrics) IncOperationCounter(_, _, _, _ string, _ metrics.NullBool)    {}
+func (*noopWriteCacheMetrics) SetActualCount(_, _, _ string, count uint64)                  {}
+func (*noopWriteCacheMetrics) SetEstimateSize(_, _, _ string, _ uint64)                     {}
+func (*noopWriteCacheMetrics) SetMode(shardID string, mode string)                          {}

internal/local/local.go

@@ -40,15 +40,18 @@ type RootModule struct {
 	mu sync.Mutex
 	// configFile is the name of the configuration file used during one test.
 	configFile string
+	// configDir is the name of the configuration directory used during one test.
+	configDir string
 	// ng is the engine instance used during one test, corresponding to the configFile. Each VU
 	// gets the same engine instance.
 	ng *engine.StorageEngine
+	l  Limiter
 }
 
 // Local represents an instance of the module for every VU.
 type Local struct {
 	vu            modules.VU
-	ResolveEngine func(context.Context, string, bool) (*engine.StorageEngine, error)
+	ResolveEngine func(context.Context, string, string, bool, int64) (*engine.StorageEngine, Limiter, error)
 }
 
 // Ensure the interfaces are implemented correctly.
@@ -56,9 +59,9 @@ var (
 	_ modules.Module   = &RootModule{}
 	_ modules.Instance = &Local{}
 
-	objPutTotal, objPutFails, objPutDuration          *metrics.Metric
-	objGetTotal, objGetFails, objGetDuration          *metrics.Metric
-	objDeleteTotal, objDeleteFails, objDeleteDuration *metrics.Metric
+	objPutSuccess, objPutFails, objPutDuration, objPutData *metrics.Metric
+	objGetSuccess, objGetFails, objGetDuration, objGetData *metrics.Metric
+	objDeleteSuccess, objDeleteFails, objDeleteDuration    *metrics.Metric
 )
 
 func init() {
@@ -71,7 +74,7 @@ func (r *RootModule) NewModuleInstance(vu modules.VU) modules.Instance {
 	return NewLocalModuleInstance(vu, r.GetOrCreateEngine)
 }
 
-func NewLocalModuleInstance(vu modules.VU, resolveEngine func(context.Context, string, bool) (*engine.StorageEngine, error)) *Local {
+func NewLocalModuleInstance(vu modules.VU, resolveEngine func(context.Context, string, string, bool, int64) (*engine.StorageEngine, Limiter, error)) *Local {
 	return &Local{
 		vu:            vu,
 		ResolveEngine: resolveEngine,
@@ -100,45 +103,53 @@ func checkResourceLimits() error {
 	return nil
 }
 
-// GetOrCreateEngine returns the current engine instance for the given configuration file,
+// GetOrCreateEngine returns the current engine instance for the given configuration file or directory,
 // creating a new one if none exists. Note that the identity of configuration files is their
 // file name for the purposes of test runs.
-func (r *RootModule) GetOrCreateEngine(ctx context.Context, configFile string, debug bool) (*engine.StorageEngine, error) {
+func (r *RootModule) GetOrCreateEngine(ctx context.Context, configFile string, configDir string, debug bool, maxSizeGB int64) (*engine.StorageEngine, Limiter, error) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
-	if len(configFile) == 0 {
-		return nil, errors.New("configFile cannot be empty")
+	if len(configFile) == 0 && len(configDir) == 0 {
+		return nil, nil, errors.New("provide configFile or configDir")
 	}
 
+	if r.l == nil {
+		r.l = NewLimiter(maxSizeGB)
+	}
 	// Create and initialize engine for the given configFile if it doesn't exist already
 	if r.ng == nil {
 		r.configFile = configFile
-		appCfg := config.New(configFile, "", "")
-		ngOpts, shardOpts, err := storageEngineOptionsFromConfig(appCfg, debug)
+		r.configDir = configDir
+		appCfg := config.New(configFile, configDir, "")
+		ngOpts, shardOpts, err := storageEngineOptionsFromConfig(ctx, appCfg, debug, r.l)
 		if err != nil {
-			return nil, fmt.Errorf("creating engine options from config: %v", err)
+			return nil, nil, fmt.Errorf("creating engine options from config: %v", err)
 		}
 		if err := checkResourceLimits(); err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 		r.ng = engine.New(ngOpts...)
 		for i, opts := range shardOpts {
-			if _, err := r.ng.AddShard(opts...); err != nil {
-				return nil, fmt.Errorf("adding shard %d: %v", i, err)
+			if _, err := r.ng.AddShard(ctx, opts...); err != nil {
+				return nil, nil, fmt.Errorf("adding shard %d: %v", i, err)
 			}
 		}
-		if err := r.ng.Open(); err != nil {
-			return nil, fmt.Errorf("opening engine: %v", err)
+		if err := r.ng.Open(ctx); err != nil {
+			return nil, nil, fmt.Errorf("opening engine: %v", err)
 		}
 		if err := r.ng.Init(ctx); err != nil {
-			return nil, fmt.Errorf("initializing engine: %v", err)
+			return nil, nil, fmt.Errorf("initializing engine: %v", err)
 		}
 	} else if configFile != r.configFile {
-		return nil, fmt.Errorf("GetOrCreateEngine called with mismatching configFile after engine was initialized: got %q, want %q", configFile, r.configFile)
+		return nil, nil, fmt.Errorf("GetOrCreateEngine called with mismatching configFile after engine was "+
+			"initialized: got %q, want %q", configFile, r.configFile)
+	} else if configDir != r.configDir {
+		return nil, nil, fmt.Errorf("GetOrCreateEngine called with mismatching configDir after engine was "+
+			"initialized: got %q, want %q", configDir, r.configDir)
 	}
 
-	return r.ng, nil
+	return r.ng, r.l, nil
 }
 
 // Exports implements the modules.Instance interface and returns the exports
@@ -149,10 +160,10 @@ func (s *Local) Exports() modules.Exports {
 
 func (s *Local) VU() modules.VU { return s.vu }
 
-func (s *Local) Connect(configFile, hexKey string, debug bool) (*Client, error) {
-	ng, err := s.ResolveEngine(s.VU().Context(), configFile, debug)
+func (s *Local) Connect(configFile, configDir, hexKey string, debug bool, maxSizeGB int64) (*Client, error) {
+	ng, l, err := s.ResolveEngine(s.VU().Context(), configFile, configDir, debug, maxSizeGB)
 	if err != nil {
-		return nil, fmt.Errorf("connecting to engine for config %q: %v", configFile, err)
+		return nil, fmt.Errorf("connecting to engine for config - file %q dir %q: %v", configFile, configDir, err)
 	}
 
 	key, err := ParseOrCreateKey(hexKey)
@@ -161,18 +172,19 @@ func (s *Local) Connect(configFile, hexKey string, debug bool) (*Client, error)
 	}
 
 	// Register metrics.
-	registry := metrics.NewRegistry()
-	objPutTotal, _ = registry.NewMetric("local_obj_put_total", metrics.Counter)
-	objPutFails, _ = registry.NewMetric("local_obj_put_fails", metrics.Counter)
-	objPutDuration, _ = registry.NewMetric("local_obj_put_duration", metrics.Trend, metrics.Time)
+	objPutSuccess, _ = stats.Registry.NewMetric("local_obj_put_success", metrics.Counter)
+	objPutFails, _ = stats.Registry.NewMetric("local_obj_put_fails", metrics.Counter)
+	objPutDuration, _ = stats.Registry.NewMetric("local_obj_put_duration", metrics.Trend, metrics.Time)
+	objPutData, _ = stats.Registry.NewMetric("local_obj_put_bytes", metrics.Counter, metrics.Data)
 
-	objGetTotal, _ = registry.NewMetric("local_obj_get_total", metrics.Counter)
-	objGetFails, _ = registry.NewMetric("local_obj_get_fails", metrics.Counter)
-	objGetDuration, _ = registry.NewMetric("local_obj_get_duration", metrics.Trend, metrics.Time)
+	objGetSuccess, _ = stats.Registry.NewMetric("local_obj_get_success", metrics.Counter)
+	objGetFails, _ = stats.Registry.NewMetric("local_obj_get_fails", metrics.Counter)
+	objGetDuration, _ = stats.Registry.NewMetric("local_obj_get_duration", metrics.Trend, metrics.Time)
+	objGetData, _ = stats.Registry.NewMetric("local_obj_get_bytes", metrics.Counter, metrics.Data)
 
-	objDeleteTotal, _ = registry.NewMetric("local_obj_delete_total", metrics.Counter)
-	objDeleteFails, _ = registry.NewMetric("local_obj_delete_fails", metrics.Counter)
-	objDeleteDuration, _ = registry.NewMetric("local_obj_delete_duration", metrics.Trend, metrics.Time)
+	objDeleteSuccess, _ = stats.Registry.NewMetric("local_obj_delete_success", metrics.Counter)
+	objDeleteFails, _ = stats.Registry.NewMetric("local_obj_delete_fails", metrics.Counter)
+	objDeleteDuration, _ = stats.Registry.NewMetric("local_obj_delete_duration", metrics.Trend, metrics.Time)
 
 	// Create raw client backed by local storage engine.
 	rc := rawclient.New(ng,
@@ -181,30 +193,32 @@ func (s *Local) Connect(configFile, hexKey string, debug bool) (*Client, error)
 			if err != nil {
 				stats.Report(s.vu, objPutFails, 1)
 			} else {
-				stats.Report(s.vu, objPutTotal, 1)
+				stats.Report(s.vu, objPutSuccess, 1)
 				stats.ReportDataSent(s.vu, float64(sz))
 				stats.Report(s.vu, objPutDuration, metrics.D(dt))
+				stats.Report(s.vu, objPutData, float64(sz))
 			}
 		}),
 		rawclient.WithGetHandler(func(sz uint64, err error, dt time.Duration) {
 			if err != nil {
 				stats.Report(s.vu, objGetFails, 1)
 			} else {
-				stats.Report(s.vu, objGetTotal, 1)
+				stats.Report(s.vu, objGetSuccess, 1)
 				stats.Report(s.vu, objGetDuration, metrics.D(dt))
 				stats.ReportDataReceived(s.vu, float64(sz))
+				stats.Report(s.vu, objGetData, float64(sz))
 			}
 		}),
 		rawclient.WithDeleteHandler(func(err error, dt time.Duration) {
 			if err != nil {
 				stats.Report(s.vu, objDeleteFails, 1)
 			} else {
-				stats.Report(s.vu, objDeleteTotal, 1)
+				stats.Report(s.vu, objDeleteSuccess, 1)
 				stats.Report(s.vu, objDeleteDuration, metrics.D(dt))
 			}
 		}),
 	)
-	return &Client{vu: s.vu, rc: rc}, nil
+	return &Client{vu: s.vu, rc: rc, l: l}, nil
 }
 
 type epochState struct{}
@@ -217,7 +231,7 @@ func (epochState) CurrentEpoch() uint64 { return 0 }
 // preloaded the storage (if any), by using the same configuration file.
 //
 // Note that the configuration file only needs to contain the storage-specific sections.
-func storageEngineOptionsFromConfig(c *config.Config, debug bool) ([]engine.Option, [][]shard.Option, error) {
+func storageEngineOptionsFromConfig(ctx context.Context, c *config.Config, debug bool, l Limiter) ([]engine.Option, [][]shard.Option, error) {
 	log := zap.L()
 	if debug {
 		var err error
@@ -231,11 +245,12 @@ func storageEngineOptionsFromConfig(c *config.Config, debug bool) ([]engine.Opti
 		engine.WithErrorThreshold(engineconfig.ShardErrorThreshold(c)),
 		engine.WithShardPoolSize(engineconfig.ShardPoolSize(c)),
 		engine.WithLogger(&logger.Logger{Logger: log}),
+		engine.WithMetrics(l),
 	}
 
 	var shOpts [][]shard.Option
 
-	engineconfig.IterateShards(c, false, func(sc *shardconfig.Config) error {
+	err := engineconfig.IterateShards(c, false, func(sc *shardconfig.Config) error {
 		opts := []shard.Option{
 			shard.WithRefillMetabase(sc.RefillMetabase()),
 			shard.WithMode(sc.Mode()),
@@ -251,6 +266,7 @@ func storageEngineOptionsFromConfig(c *config.Config, debug bool) ([]engine.Opti
 					cfg := blobovniczaconfig.From((*config.Config)(scfg))
 					ss := blobstor.SubStorage{
 						Storage: blobovniczatree.NewBlobovniczaTree(
+							ctx,
 							blobovniczatree.WithRootPath(scfg.Path()),
 							blobovniczatree.WithPermissions(scfg.Perm()),
 							blobovniczatree.WithBlobovniczaSize(cfg.Size()),
@@ -295,15 +311,17 @@ func storageEngineOptionsFromConfig(c *config.Config, debug bool) ([]engine.Opti
 			opts = append(opts,
 				shard.WithWriteCache(true),
 				shard.WithWriteCacheOptions(
-					writecache.WithPath(wc.Path()),
-					writecache.WithMaxBatchSize(wc.BoltDB().MaxBatchSize()),
-					writecache.WithMaxBatchDelay(wc.BoltDB().MaxBatchDelay()),
-					writecache.WithMaxObjectSize(wc.MaxObjectSize()),
-					writecache.WithSmallObjectSize(wc.SmallObjectSize()),
-					writecache.WithFlushWorkersCount(wc.WorkersNumber()),
-					writecache.WithMaxCacheSize(wc.SizeLimit()),
-					writecache.WithNoSync(wc.NoSync()),
-					writecache.WithLogger(&logger.Logger{Logger: log}),
+					[]writecache.Option{
+						writecache.WithPath(wc.Path()),
+						writecache.WithMaxBatchSize(wc.BoltDB().MaxBatchSize()),
+						writecache.WithMaxBatchDelay(wc.BoltDB().MaxBatchDelay()),
+						writecache.WithMaxObjectSize(wc.MaxObjectSize()),
+						writecache.WithSmallObjectSize(wc.SmallObjectSize()),
+						writecache.WithFlushWorkersCount(wc.WorkerCount()),
+						writecache.WithMaxCacheSize(wc.SizeLimit()),
+						writecache.WithNoSync(wc.NoSync()),
+						writecache.WithLogger(&logger.Logger{Logger: log}),
+					},
 				),
 			)
 		}
@@ -357,7 +375,9 @@ func storageEngineOptionsFromConfig(c *config.Config, debug bool) ([]engine.Opti
 
 		return nil
 	})
-
+	if err != nil {
+		return nil, nil, fmt.Errorf("iterate shards: %w", err)
+	}
 	return ngOpts, shOpts, nil
 }
 

internal/local/rawclient/rawclient.go

@@ -52,7 +52,7 @@ func (c *RawClient) Put(ctx context.Context, containerID cid.ID, ownerID *user.I
 
 	obj := object.New()
 	obj.SetContainerID(containerID)
-	obj.SetOwnerID(ownerID)
+	obj.SetOwnerID(*ownerID)
 	obj.SetAttributes(attrs...)
 	obj.SetPayload(payload)
 	obj.SetPayloadSize(uint64(sz))

internal/native/cache.go

@@ -0,0 +1,58 @@
+package native
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
+	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
+)
+
+const networkCacheTTL = time.Minute
+
+var networkInfoCache = &networkInfoCacheT{}
+
+type networkInfoCacheT struct {
+	guard   sync.RWMutex
+	current *netmap.NetworkInfo
+	fetchTS time.Time
+}
+
+func (c *networkInfoCacheT) getOrFetch(ctx context.Context, cli *client.Client) (*netmap.NetworkInfo, error) {
+	if v := c.get(); v != nil {
+		return v, nil
+	}
+	return c.fetch(ctx, cli)
+}
+
+func (c *networkInfoCacheT) get() *netmap.NetworkInfo {
+	c.guard.RLock()
+	defer c.guard.RUnlock()
+
+	if c.current == nil || time.Since(c.fetchTS) > networkCacheTTL {
+		return nil
+	}
+
+	return c.current
+}
+
+func (c *networkInfoCacheT) fetch(ctx context.Context, cli *client.Client) (*netmap.NetworkInfo, error) {
+	c.guard.Lock()
+	defer c.guard.Unlock()
+
+	if time.Since(c.fetchTS) <= networkCacheTTL {
+		return c.current, nil
+	}
+
+	res, err := cli.NetworkInfo(ctx, client.PrmNetworkInfo{})
+	if err != nil {
+		return nil, err
+	}
+
+	v := res.Info()
+	c.current = &v
+	c.fetchTS = time.Now()
+
+	return c.current, nil
+}

internal/native/client.go

@@ -1,7 +1,6 @@
 package native
 
 import (
-	"bytes"
 	"context"
 	"crypto/ecdsa"
 	"crypto/sha256"
@@ -23,8 +22,8 @@ import (
 	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/user"
 	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/version"
 	"git.frostfs.info/TrueCloudLab/tzhash/tz"
+	"git.frostfs.info/TrueCloudLab/xk6-frostfs/internal/datagen"
 	"git.frostfs.info/TrueCloudLab/xk6-frostfs/internal/stats"
-	"github.com/dop251/goja"
 	"go.k6.io/k6/js/modules"
 	"go.k6.io/k6/metrics"
 )
@@ -36,6 +35,7 @@ type (
 		tok            session.Object
 		cli            *client.Client
 		prepareLocally bool
+		maxObjSize     uint64
 	}
 
 	PutResponse struct {
@@ -72,12 +72,13 @@ type (
 		hdr            object.Object
 		payload        []byte
 		prepareLocally bool
+		maxObjSize     uint64
 	}
 )
 
 const defaultBufferSize = 64 * 1024
 
-func (c *Client) Put(containerID string, headers map[string]string, payload goja.ArrayBuffer, chunkSize int) PutResponse {
+func (c *Client) Put(containerID string, headers map[string]string, payload datagen.Payload, chunkSize int) PutResponse {
 	cliContainerID := parseContainerID(containerID)
 
 	tok := c.tok
@@ -101,10 +102,10 @@ func (c *Client) Put(containerID string, headers map[string]string, payload goja
 
 	var o object.Object
 	o.SetContainerID(cliContainerID)
-	o.SetOwnerID(&owner)
+	o.SetOwnerID(owner)
 	o.SetAttributes(attrs...)
 
-	resp, err := put(c.vu, c.cli, c.prepareLocally, &tok, &o, payload.Bytes(), chunkSize)
+	resp, err := put(c.vu, c.cli, c.prepareLocally, &tok, &o, payload, chunkSize, c.maxObjSize)
 	if err != nil {
 		return PutResponse{Success: false, Error: err.Error()}
 	}
@@ -128,9 +129,9 @@ func (c *Client) Delete(containerID string, objectID string) DeleteResponse {
 	start := time.Now()
 
 	var prm client.PrmObjectDelete
-	prm.ByID(cliObjectID)
-	prm.FromContainer(cliContainerID)
-	prm.WithinSession(tok)
+	prm.ObjectID = &cliObjectID
+	prm.ContainerID = &cliContainerID
+	prm.Session = &tok
 
 	_, err = c.cli.ObjectDelete(c.vu.Context(), prm)
 	if err != nil {
@@ -138,7 +139,7 @@ func (c *Client) Delete(containerID string, objectID string) DeleteResponse {
 		return DeleteResponse{Success: false, Error: err.Error()}
 	}
 
-	stats.Report(c.vu, objDeleteTotal, 1)
+	stats.Report(c.vu, objDeleteSuccess, 1)
 	stats.Report(c.vu, objDeleteDuration, metrics.D(time.Since(start)))
 	return DeleteResponse{Success: true}
 }
@@ -159,11 +160,11 @@ func (c *Client) Get(containerID, objectID string) GetResponse {
 	start := time.Now()
 
 	var prm client.PrmObjectGet
-	prm.ByID(cliObjectID)
-	prm.FromContainer(cliContainerID)
-	prm.WithinSession(tok)
+	prm.ObjectID = &cliObjectID
+	prm.ContainerID = &cliContainerID
+	prm.Session = &tok
 
-	var objSize = 0
+	objSize := 0
 	err = get(c.cli, prm, c.vu.Context(), func(data []byte) {
 		objSize += len(data)
 	})
@@ -172,9 +173,10 @@ func (c *Client) Get(containerID, objectID string) GetResponse {
 		return GetResponse{Success: false, Error: err.Error()}
 	}
 
-	stats.Report(c.vu, objGetTotal, 1)
+	stats.Report(c.vu, objGetSuccess, 1)
 	stats.Report(c.vu, objGetDuration, metrics.D(time.Since(start)))
 	stats.ReportDataReceived(c.vu, float64(objSize))
+	stats.Report(c.vu, objGetData, float64(objSize))
 	return GetResponse{Success: true}
 }
 
@@ -184,7 +186,7 @@ func get(
 	ctx context.Context,
 	onDataChunk func(chunk []byte),
 ) error {
-	var buf = make([]byte, defaultBufferSize)
+	buf := make([]byte, defaultBufferSize)
 
 	objectReader, err := cli.ObjectGetInit(ctx, prm)
 	if err != nil {
@@ -227,9 +229,9 @@ func (c *Client) VerifyHash(containerID, objectID, expectedHash string) VerifyHa
 	}
 
 	var prm client.PrmObjectGet
-	prm.ByID(cliObjectID)
-	prm.FromContainer(cliContainerID)
-	prm.WithinSession(tok)
+	prm.ObjectID = &cliObjectID
+	prm.ContainerID = &cliContainerID
+	prm.Session = &tok
 
 	hasher := sha256.New()
 	err = get(c.cli, prm, c.vu.Context(), func(data []byte) {
@@ -240,7 +242,7 @@ func (c *Client) VerifyHash(containerID, objectID, expectedHash string) VerifyHa
 	}
 	actualHash := hex.EncodeToString(hasher.Sum(nil))
 	if actualHash != expectedHash {
-		return VerifyHashResponse{Success: true, Error: "hash mismatch"}
+		return VerifyHashResponse{Success: false, Error: "hash mismatch"}
 	}
 
 	return VerifyHashResponse{Success: true}
@@ -309,10 +311,9 @@ func (c *Client) PutContainer(params map[string]string) PutContainerResponse {
 	}
 
 	start := time.Now()
-	var prm client.PrmContainerPut
-	prm.SetContainer(cnr)
-
-	res, err := c.cli.ContainerPut(c.vu.Context(), prm)
+	res, err := c.cli.ContainerPut(c.vu.Context(), client.PrmContainerPut{
+		Container: &cnr,
+	})
 	if err != nil {
 		return c.putCnrErrorResponse(err)
 	}
@@ -328,7 +329,7 @@ func (c *Client) PutContainer(params map[string]string) PutContainerResponse {
 	return PutContainerResponse{Success: true, ContainerID: res.ID().EncodeToString()}
 }
 
-func (c *Client) Onsite(containerID string, payload goja.ArrayBuffer) PreparedObject {
+func (c *Client) Onsite(containerID string, payload datagen.Payload) PreparedObject {
 	maxObjectSize, epoch, hhDisabled, err := parseNetworkInfo(c.vu.Context(), c.cli)
 	if err != nil {
 		panic(err)
@@ -355,7 +356,7 @@ func (c *Client) Onsite(containerID string, payload goja.ArrayBuffer) PreparedOb
 	obj.SetVersion(&apiVersion)
 	obj.SetType(object.TypeRegular)
 	obj.SetContainerID(cliContainerID)
-	obj.SetOwnerID(&owner)
+	obj.SetOwnerID(owner)
 	obj.SetPayloadSize(uint64(ln))
 	obj.SetCreationEpoch(epoch)
 
@@ -374,6 +375,7 @@ func (c *Client) Onsite(containerID string, payload goja.ArrayBuffer) PreparedOb
 		hdr:            *obj,
 		payload:        data,
 		prepareLocally: c.prepareLocally,
+		maxObjSize:     c.maxObjSize,
 	}
 }
 
@@ -399,7 +401,7 @@ func (p PreparedObject) Put(headers map[string]string) PutResponse {
 		return PutResponse{Success: false, Error: err.Error()}
 	}
 
-	_, err = put(p.vu, p.cli, p.prepareLocally, nil, &obj, p.payload, 0)
+	_, err = put(p.vu, p.cli, p.prepareLocally, nil, &obj, datagen.NewFixedPayload(p.payload), 0, p.maxObjSize)
 	if err != nil {
 		return PutResponse{Success: false, Error: err.Error()}
 	}
@@ -414,33 +416,37 @@ func (s epochSource) CurrentEpoch() uint64 {
 }
 
 func put(vu modules.VU, cli *client.Client, prepareLocally bool, tok *session.Object,
-	hdr *object.Object, payload []byte, chunkSize int) (*client.ResObjectPut, error) {
+	hdr *object.Object, payload datagen.Payload, chunkSize int, maxObjSize uint64,
+) (*client.ResObjectPut, error) {
 	bufSize := defaultBufferSize
 	if chunkSize > 0 {
 		bufSize = chunkSize
 	}
 	buf := make([]byte, bufSize)
-	rdr := bytes.NewReader(payload)
-	sz := rdr.Size()
+	rdr := payload.Reader()
+	sz := payload.Size()
 
 	// starting upload
 	start := time.Now()
 
 	var prm client.PrmObjectPutInit
 	if tok != nil {
-		prm.WithinSession(*tok)
+		prm.Session = tok
 	}
 	if chunkSize > 0 {
-		prm.SetGRPCPayloadChunkLen(chunkSize)
+		prm.MaxChunkLength = chunkSize
 	}
 	if prepareLocally {
-		res, err := cli.NetworkInfo(vu.Context(), client.PrmNetworkInfo{})
+		ni, err := networkInfoCache.getOrFetch(vu.Context(), cli)
 		if err != nil {
 			return nil, err
 		}
-		prm.WithObjectMaxSize(res.Info().MaxObjectSize())
-		prm.WithEpochSource(epochSource(res.Info().CurrentEpoch()))
-		prm.WithoutHomomorphicHash(true)
+		prm.MaxSize = ni.MaxObjectSize()
+		prm.EpochSource = epochSource(ni.CurrentEpoch())
+		prm.WithoutHomomorphHash = true
+		if maxObjSize > 0 {
+			prm.MaxSize = maxObjSize
+		}
 	}
 
 	objectWriter, err := cli.ObjectPutInit(vu.Context(), prm)
@@ -469,9 +475,10 @@ func put(vu modules.VU, cli *client.Client, prepareLocally bool, tok *session.Ob
 		return nil, err
 	}
 
-	stats.Report(vu, objPutTotal, 1)
+	stats.Report(vu, objPutSuccess, 1)
 	stats.ReportDataSent(vu, float64(sz))
 	stats.Report(vu, objPutDuration, metrics.D(time.Since(start)))
+	stats.Report(vu, objPutData, float64(sz))
 
 	return resp, nil
 }
@@ -499,10 +506,9 @@ func (x *waitParams) setDefaults() {
 
 func (c *Client) waitForContainerPresence(ctx context.Context, cnrID cid.ID, wp *waitParams) error {
 	return waitFor(ctx, wp, func(ctx context.Context) bool {
-		var prm client.PrmContainerGet
-		prm.SetContainer(cnrID)
-
-		_, err := c.cli.ContainerGet(ctx, prm)
+		_, err := c.cli.ContainerGet(ctx, client.PrmContainerGet{
+			ContainerID: &cnrID,
+		})
 		return err == nil
 	})
 }

internal/native/native.go

@@ -8,6 +8,7 @@ import (
 	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
 	frostfsecdsa "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/crypto/ecdsa"
 	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/session"
+	"git.frostfs.info/TrueCloudLab/xk6-frostfs/internal/stats"
 	"github.com/google/uuid"
 	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
 	"go.k6.io/k6/js/modules"
@@ -28,10 +29,10 @@ var (
 	_ modules.Instance = &Native{}
 	_ modules.Module   = &RootModule{}
 
-	objPutTotal, objPutFails, objPutDuration          *metrics.Metric
-	objGetTotal, objGetFails, objGetDuration          *metrics.Metric
-	objDeleteTotal, objDeleteFails, objDeleteDuration *metrics.Metric
-	cnrPutTotal, cnrPutFails, cnrPutDuration          *metrics.Metric
+	objPutSuccess, objPutFails, objPutDuration, objPutData *metrics.Metric
+	objGetSuccess, objGetFails, objGetDuration, objGetData *metrics.Metric
+	objDeleteSuccess, objDeleteFails, objDeleteDuration    *metrics.Metric
+	cnrPutTotal, cnrPutFails, cnrPutDuration               *metrics.Metric
 )
 
 func init() {
@@ -51,13 +52,17 @@ func (n *Native) Exports() modules.Exports {
 	return modules.Exports{Default: n}
 }
 
-func (n *Native) Connect(endpoint, hexPrivateKey string, dialTimeout, streamTimeout int, prepareLocally bool) (*Client, error) {
+func (n *Native) Connect(endpoint, hexPrivateKey string, dialTimeout, streamTimeout int, prepareLocally bool, maxObjSize int) (*Client, error) {
 	var (
 		cli client.Client
 		pk  *keys.PrivateKey
 		err error
 	)
 
+	if maxObjSize < 0 {
+		return nil, fmt.Errorf("max object size value must be positive")
+	}
+
 	pk, err = keys.NewPrivateKey()
 	if len(hexPrivateKey) != 0 {
 		pk, err = keys.NewPrivateKeyFromHex(hexPrivateKey)
@@ -67,19 +72,18 @@ func (n *Native) Connect(endpoint, hexPrivateKey string, dialTimeout, streamTime
 	}
 
 	var prmInit client.PrmInit
-	prmInit.ResolveFrostFSFailures()
-	prmInit.SetDefaultPrivateKey(pk.PrivateKey)
+	prmInit.Key = pk.PrivateKey
 	cli.Init(prmInit)
 
 	var prmDial client.PrmDial
-	prmDial.SetServerURI(endpoint)
+	prmDial.Endpoint = endpoint
 
 	if dialTimeout > 0 {
-		prmDial.SetTimeout(time.Duration(dialTimeout) * time.Second)
+		prmDial.DialTimeout = time.Duration(dialTimeout) * time.Second
 	}
 
 	if streamTimeout > 0 {
-		prmDial.SetStreamTimeout(time.Duration(streamTimeout) * time.Second)
+		prmDial.StreamTimeout = time.Duration(streamTimeout) * time.Second
 	}
 
 	err = cli.Dial(n.vu.Context(), prmDial)
@@ -89,9 +93,9 @@ func (n *Native) Connect(endpoint, hexPrivateKey string, dialTimeout, streamTime
 
 	// generate session token
 	exp := uint64(math.MaxUint64)
-	var prmSessionCreate client.PrmSessionCreate
-	prmSessionCreate.SetExp(exp)
-	sessionResp, err := cli.SessionCreate(n.vu.Context(), prmSessionCreate)
+	sessionResp, err := cli.SessionCreate(n.vu.Context(), client.PrmSessionCreate{
+		Expiration: exp,
+	})
 	if err != nil {
 		return nil, fmt.Errorf("dial endpoint: %s %w", endpoint, err)
 	}
@@ -114,23 +118,35 @@ func (n *Native) Connect(endpoint, hexPrivateKey string, dialTimeout, streamTime
 	tok.SetAuthKey(&key)
 	tok.SetExp(exp)
 
+	if prepareLocally && maxObjSize > 0 {
+		res, err := cli.NetworkInfo(n.vu.Context(), client.PrmNetworkInfo{})
+		if err != nil {
+			return nil, err
+		}
+		if uint64(maxObjSize) > res.Info().MaxObjectSize() {
+			return nil, fmt.Errorf("max object size must be not greater than %d bytes", res.Info().MaxObjectSize())
+		}
+	}
+
 	// register metrics
-	registry := metrics.NewRegistry()
-	objPutTotal, _ = registry.NewMetric("frostfs_obj_put_total", metrics.Counter)
-	objPutFails, _ = registry.NewMetric("frostfs_obj_put_fails", metrics.Counter)
-	objPutDuration, _ = registry.NewMetric("frostfs_obj_put_duration", metrics.Trend, metrics.Time)
 
-	objGetTotal, _ = registry.NewMetric("frostfs_obj_get_total", metrics.Counter)
-	objGetFails, _ = registry.NewMetric("frostfs_obj_get_fails", metrics.Counter)
-	objGetDuration, _ = registry.NewMetric("frostfs_obj_get_duration", metrics.Trend, metrics.Time)
+	objPutSuccess, _ = stats.Registry.NewMetric("frostfs_obj_put_success", metrics.Counter)
+	objPutFails, _ = stats.Registry.NewMetric("frostfs_obj_put_fails", metrics.Counter)
+	objPutDuration, _ = stats.Registry.NewMetric("frostfs_obj_put_duration", metrics.Trend, metrics.Time)
+	objPutData, _ = stats.Registry.NewMetric("frostfs_obj_put_bytes", metrics.Counter, metrics.Data)
 
-	objDeleteTotal, _ = registry.NewMetric("frostfs_obj_delete_total", metrics.Counter)
-	objDeleteFails, _ = registry.NewMetric("frostfs_obj_delete_fails", metrics.Counter)
-	objDeleteDuration, _ = registry.NewMetric("frostfs_obj_delete_duration", metrics.Trend, metrics.Time)
+	objGetSuccess, _ = stats.Registry.NewMetric("frostfs_obj_get_success", metrics.Counter)
+	objGetFails, _ = stats.Registry.NewMetric("frostfs_obj_get_fails", metrics.Counter)
+	objGetDuration, _ = stats.Registry.NewMetric("frostfs_obj_get_duration", metrics.Trend, metrics.Time)
+	objGetData, _ = stats.Registry.NewMetric("frostfs_obj_get_bytes", metrics.Counter, metrics.Data)
 
-	cnrPutTotal, _ = registry.NewMetric("frostfs_cnr_put_total", metrics.Counter)
-	cnrPutFails, _ = registry.NewMetric("frostfs_cnr_put_fails", metrics.Counter)
-	cnrPutDuration, _ = registry.NewMetric("frostfs_cnr_put_duration", metrics.Trend, metrics.Time)
+	objDeleteSuccess, _ = stats.Registry.NewMetric("frostfs_obj_delete_success", metrics.Counter)
+	objDeleteFails, _ = stats.Registry.NewMetric("frostfs_obj_delete_fails", metrics.Counter)
+	objDeleteDuration, _ = stats.Registry.NewMetric("frostfs_obj_delete_duration", metrics.Trend, metrics.Time)
+
+	cnrPutTotal, _ = stats.Registry.NewMetric("frostfs_cnr_put_total", metrics.Counter)
+	cnrPutFails, _ = stats.Registry.NewMetric("frostfs_cnr_put_fails", metrics.Counter)
+	cnrPutDuration, _ = stats.Registry.NewMetric("frostfs_cnr_put_duration", metrics.Trend, metrics.Time)
 
 	return &Client{
 		vu:             n.vu,
@@ -138,5 +154,6 @@ func (n *Native) Connect(endpoint, hexPrivateKey string, dialTimeout, streamTime
 		tok:            tok,
 		cli:            &cli,
 		prepareLocally: prepareLocally,
+		maxObjSize:     uint64(maxObjSize),
 	}, nil
 }

internal/registry/obj_exporter.go

@@ -0,0 +1,124 @@
+package registry
+
+import (
+	"fmt"
+	"os"
+)
+
+type ObjExporter struct {
+	selector *ObjSelector
+}
+
+type PreGenerateInfo struct {
+	Buckets    []string  `json:"buckets"`
+	Containers []string  `json:"containers"`
+	Objects    []ObjInfo `json:"objects"`
+	ObjSize    string    `json:"obj_size"`
+}
+
+type ObjInfo struct {
+	Bucket string `json:"bucket"`
+	Object string `json:"object"`
+	CID    string `json:"cid"`
+	OID    string `json:"oid"`
+}
+
+func NewObjExporter(selector *ObjSelector) *ObjExporter {
+	return &ObjExporter{selector: selector}
+}
+
+func (o *ObjExporter) ExportJSONPreGen(fileName string) error {
+	f, err := os.Create(fileName)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	// there can be a lot of object, so manually form json
+	if _, err = f.WriteString(`{"objects":[`); err != nil {
+		return err
+	}
+
+	bucketMap := make(map[string]struct{})
+	containerMap := make(map[string]struct{})
+
+	count, err := o.selector.Count()
+	if err != nil {
+		return err
+	}
+
+	var comma string
+	for i := 0; i < count; i++ {
+		info := o.selector.NextObject()
+		if info == nil {
+			break
+		}
+
+		if err = writeObjectInfo(comma, info, f); err != nil {
+			return err
+		}
+
+		if i == 0 {
+			comma = ","
+		}
+
+		if info.S3Bucket != "" {
+			bucketMap[info.S3Bucket] = struct{}{}
+		}
+		if info.CID != "" {
+			containerMap[info.CID] = struct{}{}
+		}
+	}
+
+	if _, err = f.WriteString(`]`); err != nil {
+		return err
+	}
+
+	if len(bucketMap) > 0 {
+		if err = writeContainerInfo("buckets", bucketMap, f); err != nil {
+			return err
+		}
+	}
+	if len(containerMap) > 0 {
+		if err = writeContainerInfo("containers", containerMap, f); err != nil {
+			return err
+		}
+	}
+
+	if _, err = f.WriteString(`}`); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func writeObjectInfo(comma string, info *ObjectInfo, f *os.File) (err error) {
+	var res string
+	if info.S3Bucket != "" || info.S3Key != "" {
+		res = fmt.Sprintf(`%s{"bucket":"%s","object":"%s"}`, comma, info.S3Bucket, info.S3Key)
+	} else {
+		res = fmt.Sprintf(`%s{"cid":"%s","oid":"%s"}`, comma, info.CID, info.OID)
+	}
+	_, err = f.WriteString(res)
+	return err
+}
+
+func writeContainerInfo(attrName string, bucketMap map[string]struct{}, f *os.File) (err error) {
+	if _, err = f.WriteString(fmt.Sprintf(`,"%s":[`, attrName)); err != nil {
+		return err
+	}
+
+	i := 0
+	comma := ""
+	for bucket := range bucketMap {
+		if _, err = f.WriteString(fmt.Sprintf(`%s"%s"`, comma, bucket)); err != nil {
+			return err
+		}
+		if i == 0 {
+			comma = ","
+		}
+		i++
+	}
+	_, err = f.WriteString(`]`)
+	return err
+}

internal/registry/obj_exporter_test.go

@@ -0,0 +1,156 @@
+package registry
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"math/rand"
+	"os"
+	"path/filepath"
+	"slices"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+type expectedResult struct {
+	mode     string
+	objects  []ObjectInfo
+	dir      string
+	dbName   string
+	jsonName string
+}
+
+func TestObjectExporter(t *testing.T) {
+	names := []string{"s3", "grpc"}
+	for _, name := range names {
+		t.Run(name, runExportTest)
+		t.Run(name+"-changed", runExportChangedTest)
+		t.Run(name+"-empty", runExportEmptyTest)
+	}
+}
+
+func runExportTest(t *testing.T) {
+	expected := getExpectedResult(t)
+	objReg := getFilledRegistry(t, expected)
+	objExp := NewObjExporter(NewObjSelector(objReg, 0, SelectorOneshot, &ObjFilter{Status: statusCreated}))
+
+	require.NoError(t, objExp.ExportJSONPreGen(expected.jsonName))
+	require.NoError(t, checkExported(expected.objects, expected.jsonName))
+}
+
+func runExportChangedTest(t *testing.T) {
+	expected := getExpectedResult(t)
+	objReg := getFilledRegistry(t, expected)
+
+	newStatus := randString(10)
+	num := randPositiveInt(1, len(expected.objects))
+	changedObjects := make([]ObjectInfo, num)
+	require.Equal(t, num, copy(changedObjects[:], expected.objects[:]))
+
+	sel := NewObjSelector(objReg, 0, SelectorOneshot, &ObjFilter{Status: statusCreated})
+	for i := range changedObjects {
+		changedObjects[i].Status = newStatus
+		require.NoError(t, objReg.SetObjectStatus(sel.NextObject().Id, statusCreated, newStatus))
+	}
+
+	objExp := NewObjExporter(NewObjSelector(objReg, 0, SelectorOneshot, &ObjFilter{Status: newStatus}))
+	require.NoError(t, objExp.ExportJSONPreGen(expected.jsonName))
+	require.NoError(t, checkExported(changedObjects, expected.jsonName))
+}
+
+func runExportEmptyTest(t *testing.T) {
+	expected := getExpectedResult(t)
+	expected.objects = make([]ObjectInfo, 0)
+	objReg := getFilledRegistry(t, expected)
+	objExp := NewObjExporter(NewObjSelector(objReg, 0, SelectorOneshot, &ObjFilter{Status: statusCreated}))
+
+	require.NoError(t, objExp.ExportJSONPreGen(expected.jsonName))
+	require.NoError(t, checkExported(expected.objects, expected.jsonName))
+}
+
+func getExpectedResult(t *testing.T) expectedResult {
+	num := randPositiveInt(2, 100)
+	mode := getMode(t.Name())
+	require.NotEqual(t, "", mode, "test mode should contain either \"s3\" or\"grpc\"")
+	dir := t.TempDir()
+	res := expectedResult{
+		mode:     mode,
+		objects:  generateObjectInfo(num, t.Name()),
+		dir:      dir,
+		dbName:   filepath.Join(dir, "registry-"+mode+".db"),
+		jsonName: filepath.Join(dir, "registry-"+mode+".json"),
+	}
+	return res
+}
+
+func randPositiveInt(min, max int) int {
+	return rand.Intn(max-min) + min
+}
+
+func getMode(name string) (res string) {
+	if strings.Contains(name, "s3") {
+		res = filepath.Base(name)
+	}
+	if strings.Contains(name, "grpc") {
+		res = filepath.Base(name)
+	}
+	return res
+}
+
+func generateObjectInfo(num int, mode string) []ObjectInfo {
+	res := make([]ObjectInfo, num)
+	for i := range res {
+		res[i] = randomObjectInfo()
+		if !strings.Contains(mode, "s3") {
+			res[i].S3Bucket = ""
+			res[i].S3Key = ""
+		}
+		if !strings.Contains(mode, "grpc") {
+			res[i].CID = ""
+			res[i].OID = ""
+		}
+	}
+	return res
+}
+
+func getFilledRegistry(t *testing.T, expected expectedResult) *ObjRegistry {
+	objReg := NewObjRegistry(context.Background(), expected.dbName)
+	for i := range expected.objects {
+		require.NoError(t, objReg.AddObject(expected.objects[i].CID, expected.objects[i].OID, expected.objects[i].S3Bucket, expected.objects[i].S3Key, expected.objects[i].PayloadHash))
+	}
+	return objReg
+}
+
+func checkExported(expected []ObjectInfo, fileName string) error {
+	file, err := os.ReadFile(fileName)
+	if err != nil {
+		return err
+	}
+
+	if !json.Valid(file) {
+		return fmt.Errorf("exported json file %s is invalid", fileName)
+	}
+
+	var actual PreGenerateInfo
+	if json.Unmarshal(file, &actual) != nil {
+		return err
+	}
+
+	if len(expected) != len(actual.Objects) {
+		return fmt.Errorf("expected len(): %v, got len(): %v", len(expected), len(actual.Objects))
+	}
+
+	for i := range expected {
+		if !slices.ContainsFunc(actual.Objects, func(oi ObjInfo) bool {
+			compareS3 := oi.Bucket == expected[i].S3Bucket && oi.Object == expected[i].S3Key
+			comparegRPC := oi.CID == expected[i].CID && oi.OID == expected[i].OID
+			return compareS3 && comparegRPC
+		}) {
+			return fmt.Errorf("object %v not found in exported json file %s", expected[i], fileName)
+		}
+	}
+
+	return nil
+}

internal/registry/obj_info_test.go

@@ -101,7 +101,7 @@ func randomObjectInfo() ObjectInfo {
 func randString(n int) string {
 	var sb strings.Builder
 	for i := 0; i < n; i++ {
-		sb.WriteRune('a' + rune(rand.Int())%('z'-'a'+1))
+		sb.WriteRune('a' + rune(rand.Int31())%('z'-'a'+1))
 	}
 	return sb.String()
 }

internal/registry/obj_registry.go

@@ -44,7 +44,7 @@ func NewObjRegistry(ctx context.Context, dbFilePath string) *ObjRegistry {
 }
 
 func (o *ObjRegistry) AddObject(cid, oid, s3Bucket, s3Key, payloadHash string) error {
-	return o.boltDB.Update(func(tx *bbolt.Tx) error {
+	return o.boltDB.Batch(func(tx *bbolt.Tx) error {
 		b, err := tx.CreateBucketIfNotExists([]byte(statusCreated))
 		if err != nil {
 			return err
@@ -75,7 +75,7 @@ func (o *ObjRegistry) AddObject(cid, oid, s3Bucket, s3Key, payloadHash string) e
 }
 
 func (o *ObjRegistry) SetObjectStatus(id uint64, oldStatus, newStatus string) error {
-	return o.boltDB.Update(func(tx *bbolt.Tx) error {
+	return o.boltDB.Batch(func(tx *bbolt.Tx) error {
 		oldB := tx.Bucket([]byte(oldStatus))
 		if oldB == nil {
 			return fmt.Errorf("bucket doesn't exist: '%s'", oldStatus)
@@ -110,7 +110,7 @@ func (o *ObjRegistry) SetObjectStatus(id uint64, oldStatus, newStatus string) er
 }
 
 func (o *ObjRegistry) DeleteObject(id uint64) error {
-	return o.boltDB.Update(func(tx *bbolt.Tx) error {
+	return o.boltDB.Batch(func(tx *bbolt.Tx) error {
 		return tx.ForEach(func(_ []byte, b *bbolt.Bucket) error {
 			return b.Delete(encodeId(id))
 		})

internal/registry/obj_selector.go

@@ -3,12 +3,15 @@ package registry
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/nspcc-dev/neo-go/pkg/io"
 	"go.etcd.io/bbolt"
 )
 
+const nextObjectTimeout = 10 * time.Second
+
 type ObjFilter struct {
 	Status string
 	Age    int
@@ -20,6 +23,9 @@ type ObjSelector struct {
 	boltDB    *bbolt.DB
 	filter    *ObjFilter
 	cacheSize int
+	kind      SelectorKind
+	// Sync synchronizes VU used for deletion.
+	Sync sync.WaitGroup
 }
 
 // objectSelectCache is the default maximum size of a batch to select from DB.
@@ -27,7 +33,7 @@ const objectSelectCache = 1000
 
 // NewObjSelector creates a new instance of object selector that can iterate over
 // objects in the specified registry.
-func NewObjSelector(registry *ObjRegistry, selectionSize int, filter *ObjFilter) *ObjSelector {
+func NewObjSelector(registry *ObjRegistry, selectionSize int, kind SelectorKind, filter *ObjFilter) *ObjSelector {
 	if selectionSize <= 0 {
 		selectionSize = objectSelectCache
 	}
@@ -40,6 +46,7 @@ func NewObjSelector(registry *ObjRegistry, selectionSize int, filter *ObjFilter)
 		filter:    filter,
 		objChan:   make(chan *ObjectInfo, selectionSize*2),
 		cacheSize: selectionSize,
+		kind:      kind,
 	}
 
 	go objSelector.selectLoop()
@@ -55,12 +62,21 @@ func NewObjSelector(registry *ObjRegistry, selectionSize int, filter *ObjFilter)
 //   - underlying registry context is done, nil objects will be returned on the
 //     currently blocked and every further NextObject calls.
 func (o *ObjSelector) NextObject() *ObjectInfo {
-	return <-o.objChan
+	if o.kind == SelectorOneshot {
+		return <-o.objChan
+	}
+
+	select {
+	case <-time.After(nextObjectTimeout):
+		return nil
+	case obj := <-o.objChan:
+		return obj
+	}
 }
 
 // Count returns total number of objects that match filter of the selector.
 func (o *ObjSelector) Count() (int, error) {
-	var count = 0
+	count := 0
 	err := o.boltDB.View(func(tx *bbolt.Tx) error {
 		b := tx.Bucket([]byte(o.filter.Status))
 		if b == nil {
@@ -160,15 +176,23 @@ func (o *ObjSelector) selectLoop() {
 			}
 		}
 
-		if len(cache) != o.cacheSize {
+		if o.kind == SelectorOneshot && len(cache) != o.cacheSize {
+			return
+		}
+
+		if o.kind != SelectorLooped && len(cache) != o.cacheSize {
 			// no more objects, wait a little; the logic could be improved.
 			select {
-			case <-time.After(time.Second * time.Duration(o.filter.Age/2)):
+			case <-time.After(time.Second):
 			case <-o.ctx.Done():
 				return
 			}
 		}
 
+		if o.kind == SelectorLooped && len(cache) != o.cacheSize {
+			lastID = 0
+		}
+
 		// clean handled objects
 		cache = cache[:0]
 	}

internal/registry/registry.go

@@ -74,7 +74,35 @@ func (r *Registry) open(dbFilePath string) *ObjRegistry {
 	return registry
 }
 
+// SelectorKind represents selector behaviour when no items are available.
+type SelectorKind byte
+
+const (
+	// SelectorAwaiting waits for a new item to arrive.
+	// This selector visits each item exactly once and can be used when items
+	// to select are being pushed into registry concurrently.
+	SelectorAwaiting = iota
+	// SelectorLooped rewinds cursor to the start after all items have been read.
+	// It can encounter duplicates and should be used mostly for read scenarious.
+	SelectorLooped
+	// SelectorOneshot visits each item exactly once and exits immediately afterwards.
+	// It may be used to artificially abort the test after all items were processed.
+	SelectorOneshot
+)
+
 func (r *Registry) GetSelector(dbFilePath string, name string, cacheSize int, filter map[string]string) *ObjSelector {
+	return r.getSelectorInternal(dbFilePath, name, cacheSize, SelectorAwaiting, filter)
+}
+
+func (r *Registry) GetLoopedSelector(dbFilePath string, name string, cacheSize int, filter map[string]string) *ObjSelector {
+	return r.getSelectorInternal(dbFilePath, name, cacheSize, SelectorLooped, filter)
+}
+
+func (r *Registry) GetOneshotSelector(dbFilePath string, name string, cacheSize int, filter map[string]string) *ObjSelector {
+	return r.getSelectorInternal(dbFilePath, name, cacheSize, SelectorOneshot, filter)
+}
+
+func (r *Registry) getSelectorInternal(dbFilePath string, name string, cacheSize int, kind SelectorKind, filter map[string]string) *ObjSelector {
 	objFilter, err := parseFilter(filter)
 	if err != nil {
 		panic(err)
@@ -86,7 +114,7 @@ func (r *Registry) GetSelector(dbFilePath string, name string, cacheSize int, fi
 	selector := r.root.selectors[name]
 	if selector == nil {
 		registry := r.open(dbFilePath)
-		selector = NewObjSelector(registry, cacheSize, objFilter)
+		selector = NewObjSelector(registry, cacheSize, kind, objFilter)
 		r.root.selectors[name] = selector
 	} else if !reflect.DeepEqual(selector.filter, objFilter) {
 		panic(fmt.Sprintf("selector %s already has been created with a different filter", name))
@@ -94,6 +122,10 @@ func (r *Registry) GetSelector(dbFilePath string, name string, cacheSize int, fi
 	return selector
 }
 
+func (r *Registry) GetExporter(selector *ObjSelector) *ObjExporter {
+	return NewObjExporter(selector)
+}
+
 func parseFilter(filter map[string]string) (*ObjFilter, error) {
 	objFilter := ObjFilter{}
 	objFilter.Status = filter["status"]

internal/s3/client.go

@@ -1,18 +1,19 @@
 package s3
 
 import (
-	"bytes"
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
+	"fmt"
 	"strconv"
 	"time"
 
+	"git.frostfs.info/TrueCloudLab/xk6-frostfs/internal/datagen"
 	"git.frostfs.info/TrueCloudLab/xk6-frostfs/internal/stats"
 	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/feature/s3/manager"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/aws/aws-sdk-go-v2/service/s3/types"
-	"github.com/dop251/goja"
 	"go.k6.io/k6/js/modules"
 	"go.k6.io/k6/metrics"
 )
@@ -49,9 +50,9 @@ type (
 	}
 )
 
-func (c *Client) Put(bucket, key string, payload goja.ArrayBuffer) PutResponse {
-	rdr := bytes.NewReader(payload.Bytes())
-	sz := rdr.Size()
+func (c *Client) Put(bucket, key string, payload datagen.Payload) PutResponse {
+	rdr := payload.Reader()
+	sz := payload.Size()
 
 	start := time.Now()
 	_, err := c.cli.PutObject(c.vu.Context(), &s3.PutObjectInput{
@@ -64,9 +65,44 @@ func (c *Client) Put(bucket, key string, payload goja.ArrayBuffer) PutResponse {
 		return PutResponse{Success: false, Error: err.Error()}
 	}
 
-	stats.Report(c.vu, objPutTotal, 1)
+	stats.Report(c.vu, objPutSuccess, 1)
 	stats.ReportDataSent(c.vu, float64(sz))
 	stats.Report(c.vu, objPutDuration, metrics.D(time.Since(start)))
+	stats.Report(c.vu, objPutData, float64(sz))
+	return PutResponse{Success: true}
+}
+
+const multipartUploadMinPartSize = 5 * 1024 * 1024 // 5MB
+
+func (c *Client) Multipart(bucket, key string, objPartSize, concurrency int, payload datagen.Payload) PutResponse {
+	if objPartSize < multipartUploadMinPartSize {
+		stats.Report(c.vu, objPutFails, 1)
+		return PutResponse{Success: false, Error: fmt.Sprintf("part size '%d' must be greater than '%d'(5 MB)", objPartSize, multipartUploadMinPartSize)}
+	}
+
+	start := time.Now()
+	uploader := manager.NewUploader(c.cli, func(u *manager.Uploader) {
+		u.PartSize = int64(objPartSize)
+		u.Concurrency = concurrency
+	})
+
+	payloadReader := payload.Reader()
+	sz := payload.Size()
+
+	_, err := uploader.Upload(c.vu.Context(), &s3.PutObjectInput{
+		Bucket: aws.String(bucket),
+		Key:    aws.String(key),
+		Body:   payloadReader,
+	})
+	if err != nil {
+		stats.Report(c.vu, objPutFails, 1)
+		return PutResponse{Success: false, Error: err.Error()}
+	}
+
+	stats.Report(c.vu, objPutSuccess, 1)
+	stats.ReportDataSent(c.vu, float64(sz))
+	stats.Report(c.vu, objPutDuration, metrics.D(time.Since(start)))
+	stats.Report(c.vu, objPutData, float64(sz))
 	return PutResponse{Success: true}
 }
 
@@ -82,7 +118,7 @@ func (c *Client) Delete(bucket, key string) DeleteResponse {
 		return DeleteResponse{Success: false, Error: err.Error()}
 	}
 
-	stats.Report(c.vu, objDeleteTotal, 1)
+	stats.Report(c.vu, objDeleteSuccess, 1)
 	stats.Report(c.vu, objDeleteDuration, metrics.D(time.Since(start)))
 	return DeleteResponse{Success: true}
 }
@@ -90,7 +126,7 @@ func (c *Client) Delete(bucket, key string) DeleteResponse {
 func (c *Client) Get(bucket, key string) GetResponse {
 	start := time.Now()
 
-	var objSize = 0
+	objSize := 0
 	err := get(c.cli, bucket, key, func(chunk []byte) {
 		objSize += len(chunk)
 	})
@@ -99,12 +135,77 @@ func (c *Client) Get(bucket, key string) GetResponse {
 		return GetResponse{Success: false, Error: err.Error()}
 	}
 
-	stats.Report(c.vu, objGetTotal, 1)
+	stats.Report(c.vu, objGetSuccess, 1)
 	stats.Report(c.vu, objGetDuration, metrics.D(time.Since(start)))
 	stats.ReportDataReceived(c.vu, float64(objSize))
+	stats.Report(c.vu, objGetData, float64(objSize))
 	return GetResponse{Success: true}
 }
 
+// DeleteObjectVersion deletes object version with specified versionID.
+// If version argument is empty, deletes all versions and delete-markers of specified object.
+func (c *Client) DeleteObjectVersion(bucket, key, version string) DeleteResponse {
+	var toDelete []types.ObjectIdentifier
+
+	if version != "" {
+		toDelete = append(toDelete, types.ObjectIdentifier{
+			Key:       aws.String(key),
+			VersionId: aws.String(version),
+		})
+	} else {
+		versions, err := c.cli.ListObjectVersions(c.vu.Context(), &s3.ListObjectVersionsInput{
+			Bucket: aws.String(bucket),
+			Prefix: aws.String(key),
+		})
+		if err != nil {
+			stats.Report(c.vu, objDeleteFails, 1)
+			return DeleteResponse{Success: false, Error: err.Error()}
+		}
+		toDelete = filterObjectVersions(versions, key)
+	}
+	if len(toDelete) == 0 {
+		return c.Delete(bucket, key)
+	} else {
+		_, err := c.cli.DeleteObjects(c.vu.Context(), &s3.DeleteObjectsInput{
+			Bucket: aws.String(bucket),
+			Delete: &types.Delete{
+				Objects: toDelete,
+				Quiet:   true,
+			},
+		})
+		if err != nil {
+			stats.Report(c.vu, objDeleteFails, 1)
+			return DeleteResponse{Success: false, Error: err.Error()}
+		}
+	}
+
+	return DeleteResponse{Success: true}
+}
+
+func filterObjectVersions(versions *s3.ListObjectVersionsOutput, key string) []types.ObjectIdentifier {
+	var result []types.ObjectIdentifier
+
+	for _, v := range versions.Versions {
+		if *v.Key == key {
+			result = append(result, types.ObjectIdentifier{
+				Key:       v.Key,
+				VersionId: v.VersionId,
+			})
+		}
+	}
+
+	for _, marker := range versions.DeleteMarkers {
+		if *marker.Key == key {
+			result = append(result, types.ObjectIdentifier{
+				Key:       marker.Key,
+				VersionId: marker.VersionId,
+			})
+		}
+	}
+
+	return result
+}
+
 func get(
 	c *s3.Client,
 	bucket string,
@@ -143,7 +244,7 @@ func (c *Client) VerifyHash(bucket, key, expectedHash string) VerifyHashResponse
 	}
 	actualHash := hex.EncodeToString(hasher.Sum(nil))
 	if actualHash != expectedHash {
-		return VerifyHashResponse{Success: true, Error: "hash mismatch"}
+		return VerifyHashResponse{Success: false, Error: "hash mismatch"}
 	}
 
 	return VerifyHashResponse{Success: true}
@@ -178,7 +279,27 @@ func (c *Client) CreateBucket(bucket string, params map[string]string) CreateBuc
 		return CreateBucketResponse{Success: false, Error: err.Error()}
 	}
 
-	stats.Report(c.vu, createBucketTotal, 1)
+	var versioning bool
+	if strVersioned, ok := params["versioning"]; ok {
+		if versioning, err = strconv.ParseBool(strVersioned); err != nil {
+			stats.Report(c.vu, createBucketFails, 1)
+			return CreateBucketResponse{Success: false, Error: err.Error()}
+		}
+	}
+	if versioning {
+		_, err = c.cli.PutBucketVersioning(context.TODO(), &s3.PutBucketVersioningInput{
+			Bucket: aws.String(bucket),
+			VersioningConfiguration: &types.VersioningConfiguration{
+				Status: types.BucketVersioningStatusEnabled,
+			},
+		})
+		if err != nil {
+			stats.Report(c.vu, createBucketFails, 1)
+			return CreateBucketResponse{Success: false, Error: err.Error()}
+		}
+	}
+
+	stats.Report(c.vu, createBucketSuccess, 1)
 	stats.Report(c.vu, createBucketDuration, metrics.D(time.Since(start)))
 	return CreateBucketResponse{Success: true}
 }

internal/s3/s3.go

@@ -7,6 +7,7 @@ import (
 	"strconv"
 	"time"
 
+	"git.frostfs.info/TrueCloudLab/xk6-frostfs/internal/stats"
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
@@ -28,10 +29,10 @@ var (
 	_ modules.Instance = &S3{}
 	_ modules.Module   = &RootModule{}
 
-	objPutTotal, objPutFails, objPutDuration                   *metrics.Metric
-	objGetTotal, objGetFails, objGetDuration                   *metrics.Metric
-	objDeleteTotal, objDeleteFails, objDeleteDuration          *metrics.Metric
-	createBucketTotal, createBucketFails, createBucketDuration *metrics.Metric
+	objPutSuccess, objPutFails, objPutDuration, objPutData       *metrics.Metric
+	objGetSuccess, objGetFails, objGetDuration, objGetData       *metrics.Metric
+	objDeleteSuccess, objDeleteFails, objDeleteDuration          *metrics.Metric
+	createBucketSuccess, createBucketFails, createBucketDuration *metrics.Metric
 )
 
 func init() {
@@ -94,22 +95,23 @@ func (s *S3) Connect(endpoint string, params map[string]string) (*Client, error)
 	})
 
 	// register metrics
-	registry := metrics.NewRegistry()
-	objPutTotal, _ = registry.NewMetric("aws_obj_put_total", metrics.Counter)
-	objPutFails, _ = registry.NewMetric("aws_obj_put_fails", metrics.Counter)
-	objPutDuration, _ = registry.NewMetric("aws_obj_put_duration", metrics.Trend, metrics.Time)
+	objPutSuccess, _ = stats.Registry.NewMetric("aws_obj_put_success", metrics.Counter)
+	objPutFails, _ = stats.Registry.NewMetric("aws_obj_put_fails", metrics.Counter)
+	objPutDuration, _ = stats.Registry.NewMetric("aws_obj_put_duration", metrics.Trend, metrics.Time)
+	objPutData, _ = stats.Registry.NewMetric("aws_obj_put_bytes", metrics.Counter, metrics.Data)
 
-	objGetTotal, _ = registry.NewMetric("aws_obj_get_total", metrics.Counter)
-	objGetFails, _ = registry.NewMetric("aws_obj_get_fails", metrics.Counter)
-	objGetDuration, _ = registry.NewMetric("aws_obj_get_duration", metrics.Trend, metrics.Time)
+	objGetSuccess, _ = stats.Registry.NewMetric("aws_obj_get_success", metrics.Counter)
+	objGetFails, _ = stats.Registry.NewMetric("aws_obj_get_fails", metrics.Counter)
+	objGetDuration, _ = stats.Registry.NewMetric("aws_obj_get_duration", metrics.Trend, metrics.Time)
+	objGetData, _ = stats.Registry.NewMetric("aws_obj_get_bytes", metrics.Counter, metrics.Data)
 
-	objDeleteTotal, _ = registry.NewMetric("aws_obj_delete_total", metrics.Counter)
-	objDeleteFails, _ = registry.NewMetric("aws_obj_delete_fails", metrics.Counter)
-	objDeleteDuration, _ = registry.NewMetric("aws_obj_delete_duration", metrics.Trend, metrics.Time)
+	objDeleteSuccess, _ = stats.Registry.NewMetric("aws_obj_delete_success", metrics.Counter)
+	objDeleteFails, _ = stats.Registry.NewMetric("aws_obj_delete_fails", metrics.Counter)
+	objDeleteDuration, _ = stats.Registry.NewMetric("aws_obj_delete_duration", metrics.Trend, metrics.Time)
 
-	createBucketTotal, _ = registry.NewMetric("aws_create_bucket_total", metrics.Counter)
-	createBucketFails, _ = registry.NewMetric("aws_create_bucket_fails", metrics.Counter)
-	createBucketDuration, _ = registry.NewMetric("aws_create_bucket_duration", metrics.Trend, metrics.Time)
+	createBucketSuccess, _ = stats.Registry.NewMetric("aws_create_bucket_success", metrics.Counter)
+	createBucketFails, _ = stats.Registry.NewMetric("aws_create_bucket_fails", metrics.Counter)
+	createBucketDuration, _ = stats.Registry.NewMetric("aws_create_bucket_duration", metrics.Trend, metrics.Time)
 
 	return &Client{
 		vu:  s.vu,

internal/s3local/client.go

@@ -1,14 +1,14 @@
 package s3local
 
 import (
-	"bytes"
 	"time"
 
 	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/data"
 	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/layer"
 	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/user"
+	"git.frostfs.info/TrueCloudLab/xk6-frostfs/internal/datagen"
+	"git.frostfs.info/TrueCloudLab/xk6-frostfs/internal/local"
 	"git.frostfs.info/TrueCloudLab/xk6-frostfs/internal/stats"
-	"github.com/dop251/goja"
 	"go.k6.io/k6/js/modules"
 	"go.k6.io/k6/metrics"
 )
@@ -18,11 +18,13 @@ type Client struct {
 	l        layer.Client
 	ownerID  *user.ID
 	resolver layer.BucketResolver
+	limiter  local.Limiter
 }
 
 type (
 	SuccessOrErrorResponse struct {
 		Success bool
+		Abort   bool
 		Error   string
 	}
 
@@ -32,7 +34,14 @@ type (
 	GetResponse          SuccessOrErrorResponse
 )
 
-func (c *Client) Put(bucket, key string, payload goja.ArrayBuffer) PutResponse {
+func (c *Client) Put(bucket, key string, payload datagen.Payload) PutResponse {
+	if c.limiter.IsFull() {
+		return PutResponse{
+			Success: false,
+			Abort:   true,
+			Error:   "engine size limit reached",
+		}
+	}
 	cid, err := c.resolver.Resolve(c.vu.Context(), bucket)
 	if err != nil {
 		stats.Report(c.vu, objPutFails, 1)
@@ -48,8 +57,8 @@ func (c *Client) Put(bucket, key string, payload goja.ArrayBuffer) PutResponse {
 		},
 		Header: map[string]string{},
 		Object: key,
-		Size:   int64(len(payload.Bytes())),
-		Reader: bytes.NewReader(payload.Bytes()),
+		Size:   uint64(payload.Size()),
+		Reader: payload.Reader(),
 	}
 
 	start := time.Now()
@@ -59,8 +68,9 @@ func (c *Client) Put(bucket, key string, payload goja.ArrayBuffer) PutResponse {
 	}
 
 	stats.Report(c.vu, objPutDuration, metrics.D(time.Since(start)))
-	stats.Report(c.vu, objPutTotal, 1)
+	stats.Report(c.vu, objPutSuccess, 1)
 	stats.ReportDataSent(c.vu, float64(prm.Size))
+	stats.Report(c.vu, objPutData, float64(prm.Size))
 
 	return PutResponse{Success: true}
 }
@@ -98,16 +108,22 @@ func (c *Client) Get(bucket, key string) GetResponse {
 			Start: 0,
 			End:   uint64(extInfo.ObjectInfo.Size),
 		},
-		Writer: wr,
 	}
-	if err := c.l.GetObject(c.vu.Context(), getPrm); err != nil {
+	objPayload, err := c.l.GetObject(c.vu.Context(), getPrm)
+	if err != nil {
+		stats.Report(c.vu, objGetFails, 1)
+		return GetResponse{Error: err.Error()}
+	}
+	err = objPayload.StreamTo(wr)
+	if err != nil {
 		stats.Report(c.vu, objGetFails, 1)
 		return GetResponse{Error: err.Error()}
 	}
 
 	stats.Report(c.vu, objGetDuration, metrics.D(time.Since(start)))
-	stats.Report(c.vu, objGetTotal, 1)
+	stats.Report(c.vu, objGetSuccess, 1)
 	stats.ReportDataReceived(c.vu, wr.total)
+	stats.Report(c.vu, objGetData, wr.total)
 
 	return GetResponse{Success: true}
 }

internal/s3local/frostfs.go

@@ -13,7 +13,6 @@ import (
 	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/eacl"
 	oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
 	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/session"
-	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/user"
 	"git.frostfs.info/TrueCloudLab/xk6-frostfs/internal/local/rawclient"
 )
 
@@ -30,15 +29,15 @@ func unimplementedMessage(fname string) string {
 		"something other than filling a cluster (i.e. PUT or GET).", fname)
 }
 
-func (*frostfs) CreateContainer(context.Context, layer.PrmContainerCreate) (cid.ID, error) {
+func (*frostfs) CreateContainer(context.Context, layer.PrmContainerCreate) (*layer.ContainerCreateResult, error) {
 	panic(unimplementedMessage("CreateContainer"))
 }
 
-func (*frostfs) Container(context.Context, cid.ID) (*container.Container, error) {
+func (*frostfs) Container(ctx context.Context, prmContainer layer.PrmContainer) (*container.Container, error) {
 	panic(unimplementedMessage("Container"))
 }
 
-func (*frostfs) UserContainers(context.Context, user.ID) ([]cid.ID, error) {
+func (*frostfs) UserContainers(ctx context.Context, containers layer.PrmUserContainers) ([]cid.ID, error) {
 	panic(unimplementedMessage("UserContainers"))
 }
 
@@ -46,7 +45,7 @@ func (*frostfs) SetContainerEACL(context.Context, eacl.Table, *session.Container
 	panic(unimplementedMessage("SetContainerEACL"))
 }
 
-func (*frostfs) ContainerEACL(context.Context, cid.ID) (*eacl.Table, error) {
+func (*frostfs) ContainerEACL(ctx context.Context, containerEACL layer.PrmContainerEACL) (*eacl.Table, error) {
 	panic(unimplementedMessage("ContainerEACL"))
 }
 
@@ -78,7 +77,7 @@ func (f *frostfs) CreateObject(ctx context.Context, prm layer.PrmObjectCreate) (
 	for _, attr := range prm.Attributes {
 		hdrs[attr[0]] = attr[1]
 	}
-	return f.Put(ctx, prm.Container, &prm.Creator, hdrs, payload)
+	return f.Put(ctx, prm.Container, nil, hdrs, payload)
 }
 
 func (f *frostfs) DeleteObject(context.Context, layer.PrmObjectDelete) error {
@@ -88,3 +87,7 @@ func (f *frostfs) DeleteObject(context.Context, layer.PrmObjectDelete) error {
 func (f *frostfs) TimeToEpoch(ctx context.Context, now time.Time, future time.Time) (uint64, uint64, error) {
 	panic(unimplementedMessage("TimeToEpoch"))
 }
+
+func (f *frostfs) SearchObjects(ctx context.Context, search layer.PrmObjectSearch) ([]oid.ID, error) {
+	panic(unimplementedMessage("SearchObjects"))
+}

internal/s3local/local.go

@@ -32,10 +32,10 @@ var (
 	_ modules.Module   = &RootModule{}
 	_ modules.Instance = &Local{}
 
-	internalObjPutTotal, internalObjPutFails, internalObjPutDuration *metrics.Metric
-	internalObjGetTotal, internalObjGetFails, internalObjGetDuration *metrics.Metric
-	objPutTotal, objPutFails, objPutDuration                         *metrics.Metric
-	objGetTotal, objGetFails, objGetDuration                         *metrics.Metric
+	internalObjPutSuccess, internalObjPutFails, internalObjPutDuration, internalObjPutData *metrics.Metric
+	internalObjGetSuccess, internalObjGetFails, internalObjGetDuration, internalObjGetData *metrics.Metric
+	objPutSuccess, objPutFails, objPutDuration, objPutData                                 *metrics.Metric
+	objGetSuccess, objGetFails, objGetDuration, objGetData                                 *metrics.Metric
 )
 
 func init() {
@@ -56,7 +56,7 @@ func (s *Local) Exports() modules.Exports {
 	return modules.Exports{Default: s}
 }
 
-func (s *Local) Connect(configFile string, params map[string]string, bucketMapping map[string]string) (*Client, error) {
+func (s *Local) Connect(configFile string, configDir string, params map[string]string, bucketMapping map[string]string, maxSizeGB int64) (*Client, error) {
 	// Parse configuration flags.
 	fs := flag.NewFlagSet("s3local", flag.ContinueOnError)
 
@@ -88,28 +88,30 @@ func (s *Local) Connect(configFile string, params map[string]string, bucketMappi
 	}
 
 	// Register metrics.
-	registry := metrics.NewRegistry()
+	internalObjPutSuccess, _ = stats.Registry.NewMetric("s3local_internal_obj_put_success", metrics.Counter)
+	internalObjPutFails, _ = stats.Registry.NewMetric("s3local_internal_obj_put_fails", metrics.Counter)
+	internalObjPutDuration, _ = stats.Registry.NewMetric("s3local_internal_obj_put_duration", metrics.Trend, metrics.Time)
+	internalObjPutData, _ = stats.Registry.NewMetric("s3local_internal_obj_put_bytes", metrics.Counter, metrics.Data)
 
-	internalObjPutTotal, _ = registry.NewMetric("s3local_internal_obj_put_total", metrics.Counter)
-	internalObjPutFails, _ = registry.NewMetric("s3local_internal_obj_put_fails", metrics.Counter)
-	internalObjPutDuration, _ = registry.NewMetric("s3local_internal_obj_put_duration", metrics.Trend, metrics.Time)
+	internalObjGetSuccess, _ = stats.Registry.NewMetric("s3local_internal_obj_get_success", metrics.Counter)
+	internalObjGetFails, _ = stats.Registry.NewMetric("s3local_internal_obj_get_fails", metrics.Counter)
+	internalObjGetDuration, _ = stats.Registry.NewMetric("s3local_internal_obj_get_duration", metrics.Trend, metrics.Time)
+	internalObjGetData, _ = stats.Registry.NewMetric("s3local_internal_obj_get_bytes", metrics.Counter, metrics.Data)
 
-	internalObjGetTotal, _ = registry.NewMetric("s3local_internal_obj_get_total", metrics.Counter)
-	internalObjGetFails, _ = registry.NewMetric("s3local_internal_obj_get_fails", metrics.Counter)
-	internalObjGetDuration, _ = registry.NewMetric("s3local_internal_obj_get_duration", metrics.Trend, metrics.Time)
+	objPutSuccess, _ = stats.Registry.NewMetric("s3local_obj_put_success", metrics.Counter)
+	objPutFails, _ = stats.Registry.NewMetric("s3local_obj_put_fails", metrics.Counter)
+	objPutDuration, _ = stats.Registry.NewMetric("s3local_obj_put_duration", metrics.Trend, metrics.Time)
+	objPutData, _ = stats.Registry.NewMetric("s3local_obj_put_bytes", metrics.Counter, metrics.Data)
 
-	objPutTotal, _ = registry.NewMetric("s3local_obj_put_total", metrics.Counter)
-	objPutFails, _ = registry.NewMetric("s3local_obj_put_fails", metrics.Counter)
-	objPutDuration, _ = registry.NewMetric("s3local_obj_put_duration", metrics.Trend, metrics.Time)
-
-	objGetTotal, _ = registry.NewMetric("s3local_obj_get_total", metrics.Counter)
-	objGetFails, _ = registry.NewMetric("s3local_obj_get_fails", metrics.Counter)
-	objGetDuration, _ = registry.NewMetric("s3local_obj_get_duration", metrics.Trend, metrics.Time)
+	objGetSuccess, _ = stats.Registry.NewMetric("s3local_obj_get_success", metrics.Counter)
+	objGetFails, _ = stats.Registry.NewMetric("s3local_obj_get_fails", metrics.Counter)
+	objGetDuration, _ = stats.Registry.NewMetric("s3local_obj_get_duration", metrics.Trend, metrics.Time)
+	objGetData, _ = stats.Registry.NewMetric("s3local_obj_get_bytes", metrics.Counter, metrics.Data)
 
 	// Create S3 layer backed by local storage engine and tree service.
-	ng, err := s.l.ResolveEngine(s.l.VU().Context(), configFile, *debugLogger)
+	ng, limiter, err := s.l.ResolveEngine(s.l.VU().Context(), configFile, configDir, *debugLogger, maxSizeGB)
 	if err != nil {
-		return nil, fmt.Errorf("connecting to engine for config %q: %v", configFile, err)
+		return nil, fmt.Errorf("connecting to engine for config - file %q dir %q: %v", configFile, configDir, err)
 	}
 
 	treeSvc := tree.NewTree(treeServiceEngineWrapper{
@@ -124,16 +126,18 @@ func (s *Local) Connect(configFile string, params map[string]string, bucketMappi
 			if err != nil {
 				stats.Report(s.l.VU(), internalObjPutFails, 1)
 			} else {
-				stats.Report(s.l.VU(), internalObjPutTotal, 1)
+				stats.Report(s.l.VU(), internalObjPutSuccess, 1)
 				stats.Report(s.l.VU(), internalObjPutDuration, metrics.D(dt))
+				stats.Report(s.l.VU(), internalObjPutData, float64(sz))
 			}
 		}),
 		rawclient.WithGetHandler(func(sz uint64, err error, dt time.Duration) {
 			if err != nil {
 				stats.Report(s.l.VU(), internalObjGetFails, 1)
 			} else {
-				stats.Report(s.l.VU(), internalObjGetTotal, 1)
+				stats.Report(s.l.VU(), internalObjGetSuccess, 1)
 				stats.Report(s.l.VU(), internalObjGetDuration, metrics.D(dt))
+				stats.Report(s.l.VU(), internalObjGetData, float64(sz))
 			}
 		}),
 	)
@@ -144,20 +148,24 @@ func (s *Local) Connect(configFile string, params map[string]string, bucketMappi
 	}
 
 	cfg := &layer.Config{
-		Caches:      layer.DefaultCachesConfigs(zap.L()),
+		Cache:       layer.NewCache(layer.DefaultCachesConfigs(zap.L())),
 		AnonKey:     layer.AnonymousKey{Key: key},
 		Resolver:    resolver,
 		TreeService: treeSvc,
 	}
 
 	l := layer.NewLayer(zap.L(), &frostfs{rc}, cfg)
-	l.Initialize(s.l.VU().Context(), nopEventListener{})
+	err = l.Initialize(s.l.VU().Context(), nopEventListener{})
+	if err != nil {
+		return nil, fmt.Errorf("initialize: %w", err)
+	}
 
 	return &Client{
 		vu:       s.l.VU(),
 		l:        l,
 		ownerID:  rc.OwnerID(),
 		resolver: resolver,
+		limiter:  limiter,
 	}, nil
 }
 

internal/s3local/treeService.go

@@ -115,7 +115,7 @@ func (s treeServiceEngineWrapper) GetSubTree(ctx context.Context, bktInfo *data.
 			return fmt.Errorf("getting children: %v", err)
 		}
 		for _, child := range children {
-			if err := traverse(child, curDepth+1); err != nil {
+			if err := traverse(child.ID, curDepth+1); err != nil {
 				return err
 			}
 		}
@@ -191,6 +191,12 @@ func (s treeServiceEngineWrapper) RemoveNode(ctx context.Context, bktInfo *data.
 	return err
 }
 
+func (s treeServiceEngineWrapper) GetSubTreeStream(
+	ctx context.Context, bktInfo *data.BucketInfo, treeID string, rootID uint64, depth uint32,
+) (tree.SubTreeStream, error) {
+	panic(unimplementedMessage("TreeService.GetSubTreeStream"))
+}
+
 func mapToKV(m map[string]string) []pilorama.KeyValue {
 	var kvs []pilorama.KeyValue
 	for k, v := range m {

internal/stats/stats.go

@@ -1,16 +1,54 @@
 package stats
 
 import (
+	"strings"
 	"time"
 
 	"go.k6.io/k6/js/modules"
 	"go.k6.io/k6/metrics"
 )
 
+// RootModule is the global module object type. It is instantiated once per test
+// run and will be used to create k6/x/frostfs/stats module instances for each VU.
+type RootModule struct {
+	Instance string
+}
+
+var (
+	tagSet *metrics.TagSet
+
+	Registry *metrics.Registry
+)
+
+func init() {
+	Registry = metrics.NewRegistry()
+	tagSet = Registry.RootTagSet()
+	modules.Register("k6/x/frostfs/stats", &RootModule{})
+}
+
+// SetTags sets additional tags to custom metrics.
+// Format: "key1:value1;key2:value2".
+// Panics if input has invalid format.
+func (m *RootModule) SetTags(labels string) {
+	kv := make(map[string]string)
+	pairs := strings.Split(labels, ";")
+	for _, pair := range pairs {
+		items := strings.Split(pair, ":")
+		if len(items) != 2 {
+			panic("invalid labels format")
+		}
+		kv[strings.TrimSpace(items[0])] = strings.TrimSpace(items[1])
+	}
+	for k, v := range kv {
+		tagSet = tagSet.With(k, v)
+	}
+}
+
 func Report(vu modules.VU, metric *metrics.Metric, value float64) {
 	metrics.PushIfNotDone(vu.Context(), vu.State().Samples, metrics.Sample{
 		TimeSeries: metrics.TimeSeries{
 			Metric: metric,
+			Tags:   tagSet,
 		},
 		Time:  time.Now(),
 		Value: value,
@@ -22,9 +60,11 @@ func ReportDataReceived(vu modules.VU, value float64) {
 		metrics.Sample{
 			TimeSeries: metrics.TimeSeries{
 				Metric: &metrics.Metric{},
+				Tags:   tagSet,
 			},
 			Value: value,
-			Time:  time.Now()},
+			Time:  time.Now(),
+		},
 	)
 }
 
@@ -34,8 +74,10 @@ func ReportDataSent(vu modules.VU, value float64) {
 		metrics.Sample{
 			TimeSeries: metrics.TimeSeries{
 				Metric: &metrics.Metric{},
+				Tags:   tagSet,
 			},
 			Value: value,
-			Time:  time.Now()},
+			Time:  time.Now(),
+		},
 	)
 }

internal/version/version.go

@@ -0,0 +1,6 @@
+package version
+
+var (
+	// Version is the xk6 command-line utils version.
+	Version = "dev"
+)

scenarios/grpc.js

@@ -1,177 +1,222 @@
-import datagen from 'k6/x/frostfs/datagen';
-import native from 'k6/x/frostfs/native';
-import logging from 'k6/x/frostfs/logging';
-import registry from 'k6/x/frostfs/registry';
-import { SharedArray } from 'k6/data';
 import { sleep } from 'k6';
-import { textSummary } from './libs/k6-summary-0.0.2.js';
+import { SharedArray } from 'k6/data';
+import exec from 'k6/execution';
+import logging from 'k6/x/frostfs/logging';
+import native from 'k6/x/frostfs/native';
+import registry from 'k6/x/frostfs/registry';
+import stats from 'k6/x/frostfs/stats';
+
+import { newGenerator } from './libs/datagen.js';
 import { parseEnv } from './libs/env-parser.js';
+import { textSummary } from './libs/k6-summary-0.0.2.js';
 import { uuidv4 } from './libs/k6-utils-1.4.0.js';
 
 parseEnv();
 
-const obj_list = new SharedArray('obj_list', function () {
-    return JSON.parse(open(__ENV.PREGEN_JSON)).objects;
-});
+const obj_list = new SharedArray(
+  'obj_list',
+  function () { return JSON.parse(open(__ENV.PREGEN_JSON)).objects; });
 
-const container_list = new SharedArray('container_list', function () {
-    return JSON.parse(open(__ENV.PREGEN_JSON)).containers;
-});
+const container_list = new SharedArray(
+  'container_list',
+  function () { return JSON.parse(open(__ENV.PREGEN_JSON)).containers; });
 
 const read_size = JSON.parse(open(__ENV.PREGEN_JSON)).obj_size;
-const summary_json = __ENV.SUMMARY_JSON || "/tmp/summary.json";
+const summary_json = __ENV.SUMMARY_JSON || '/tmp/summary.json';
 
 // Select random gRPC endpoint for current VU
 const grpc_endpoints = __ENV.GRPC_ENDPOINTS.split(',');
-const grpc_endpoint = grpc_endpoints[Math.floor(Math.random() * grpc_endpoints.length)];
-const grpc_client = native.connect(grpc_endpoint, '',
-    __ENV.DIAL_TIMEOUT ? parseInt(__ENV.DIAL_TIMEOUT) : 5,
-    __ENV.STREAM_TIMEOUT ? parseInt(__ENV.STREAM_TIMEOUT) : 60,
-    __ENV.PREPARE_LOCALLY ? __ENV.PREPARE_LOCALLY.toLowerCase() === "true" : false);
-const log = logging.new().withField("endpoint", grpc_endpoint);
+const grpc_endpoint =
+  grpc_endpoints[Math.floor(Math.random() * grpc_endpoints.length)];
+const grpc_client = native.connect(
+  grpc_endpoint, '', __ENV.DIAL_TIMEOUT ? parseInt(__ENV.DIAL_TIMEOUT) : 5,
+  __ENV.STREAM_TIMEOUT ? parseInt(__ENV.STREAM_TIMEOUT) : 60,
+  __ENV.PREPARE_LOCALLY ? __ENV.PREPARE_LOCALLY.toLowerCase() === 'true' : false,
+  1024 * parseInt(__ENV.MAX_OBJECT_SIZE || '0'));
+const log = logging.new().withField('endpoint', grpc_endpoint);
 
 const registry_enabled = !!__ENV.REGISTRY_FILE;
-const obj_registry = registry_enabled ? registry.open(__ENV.REGISTRY_FILE) : undefined;
+const obj_registry =
+  registry_enabled ? registry.open(__ENV.REGISTRY_FILE) : undefined;
 
 const duration = __ENV.DURATION;
 
-const delete_age = __ENV.DELETE_AGE ? parseInt(__ENV.DELETE_AGE) : undefined;
-let obj_to_delete_selector = undefined;
-if (registry_enabled && delete_age) {
-    obj_to_delete_selector = registry.getSelector(
-        __ENV.REGISTRY_FILE,
-        "obj_to_delete",
-        __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0,
-        {
-            status: "created",
-            age: delete_age,
-        }
-    );
+if (!!__ENV.METRIC_TAGS) {
+  stats.setTags(__ENV.METRIC_TAGS)
 }
 
-
-const generator = datagen.generator(1024 * parseInt(__ENV.WRITE_OBJ_SIZE), __ENV.PAYLOAD_TYPE || "");
+const read_age = __ENV.READ_AGE ? parseInt(__ENV.READ_AGE) : 10;
+let obj_to_read_selector = undefined;
+if (registry_enabled) {
+  obj_to_read_selector = registry.getLoopedSelector(
+    __ENV.REGISTRY_FILE, 'obj_to_read',
+    __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0, {
+    status: 'created',
+    age: read_age,
+  })
+}
 
 const scenarios = {};
 
 const write_vu_count = parseInt(__ENV.WRITERS || '0');
 const write_grpc_chunk_size = 1024 * parseInt(__ENV.GRPC_CHUNK_SIZE || '0')
+const generator = newGenerator(write_vu_count > 0);
 if (write_vu_count > 0) {
-    scenarios.write = {
-        executor: 'constant-vus',
-        vus: write_vu_count,
-        duration: `${duration}s`,
-        exec: 'obj_write', 
-        gracefulStop: '5s',
-    };
+  scenarios.write = {
+    executor: 'constant-vus',
+    vus: write_vu_count,
+    duration: `${duration}s`,
+    exec: 'obj_write',
+    gracefulStop: '5s',
+  };
+}
+
+const delete_age = __ENV.DELETE_AGE ? parseInt(__ENV.DELETE_AGE) : undefined;
+let obj_to_delete_selector = undefined;
+let obj_to_delete_exit_on_null = undefined;
+if (registry_enabled && delete_age) {
+  obj_to_delete_exit_on_null = write_vu_count == 0;
+
+  let constructor = obj_to_delete_exit_on_null ? registry.getOneshotSelector
+    : registry.getSelector;
+
+  obj_to_delete_selector =
+    constructor(__ENV.REGISTRY_FILE, 'obj_to_delete',
+      __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0, {
+      status: 'created',
+      age: delete_age,
+    });
 }
 
 const read_vu_count = parseInt(__ENV.READERS || '0');
 if (read_vu_count > 0) {
-    scenarios.read = {
-        executor: 'constant-vus',
-        vus: read_vu_count,
-        duration: `${duration}s`,
-        exec: 'obj_read', 
-        gracefulStop: '5s',
-    };
+  scenarios.read = {
+    executor: 'constant-vus',
+    vus: read_vu_count,
+    duration: `${duration}s`,
+    exec: 'obj_read',
+    gracefulStop: '5s',
+  };
 }
 
 const delete_vu_count = parseInt(__ENV.DELETERS || '0');
 if (delete_vu_count > 0) {
-    if (!obj_to_delete_selector) {
-        throw new Error('Positive DELETE worker number without a proper object selector');
-    }
+  if (!obj_to_delete_selector) {
+    throw new Error(
+      'Positive DELETE worker number without a proper object selector');
+  }
 
-    scenarios.delete = {
-        executor: 'constant-vus',
-        vus: delete_vu_count,
-        duration: `${duration}s`,
-        exec: 'obj_delete', 
-        gracefulStop: '5s',
-    };
+  scenarios.delete = {
+    executor: 'constant-vus',
+    vus: delete_vu_count,
+    duration: `${duration}s`,
+    exec: 'obj_delete',
+    gracefulStop: '5s',
+  };
 }
 
 export const options = {
-    scenarios,
-    setupTimeout: '5s',
+  scenarios,
+  setupTimeout: '5s',
 };
 
 export function setup() {
-    const total_vu_count = write_vu_count + read_vu_count + delete_vu_count;
+  const total_vu_count = write_vu_count + read_vu_count + delete_vu_count;
 
-    console.log(`Pregenerated containers:       ${container_list.length}`);
-    console.log(`Pregenerated read object size: ${read_size}`);
-    console.log(`Pregenerated total objects:    ${obj_list.length}`);
-    console.log(`Reading VUs:                   ${read_vu_count}`);
-    console.log(`Writing VUs:                   ${write_vu_count}`);
-    console.log(`Deleting VUs:                  ${delete_vu_count}`);
-    console.log(`Total VUs:                     ${total_vu_count}`);
+  console.log(`Pregenerated containers:       ${container_list.length}`);
+  console.log(`Pregenerated read object size: ${read_size}`);
+  console.log(`Pregenerated total objects:    ${obj_list.length}`);
+  console.log(`Reading VUs:                   ${read_vu_count}`);
+  console.log(`Writing VUs:                   ${write_vu_count}`);
+  console.log(`Deleting VUs:                  ${delete_vu_count}`);
+  console.log(`Total VUs:                     ${total_vu_count}`);
+
+  const start_timestamp = Date.now()
+  console.log(
+    `Load started at:               ${Date(start_timestamp).toString()}`)
 }
 
 export function teardown(data) {
-    if (obj_registry) {
-        obj_registry.close();
-    }
+  if (obj_registry) {
+    obj_registry.close();
+  }
+  const end_timestamp = Date.now()
+  console.log(
+    `Load finished at:              ${Date(end_timestamp).toString()}`)
 }
 
 export function handleSummary(data) {
-    return {
-        'stdout': textSummary(data, { indent: ' ', enableColors: false }),
-        [summary_json]: JSON.stringify(data),
-    };
+  return {
+    'stdout': textSummary(data, { indent: ' ', enableColors: false }),
+    [summary_json]: JSON.stringify(data),
+  };
 }
 
 export function obj_write() {
-    if (__ENV.SLEEP_WRITE) {
-        sleep(__ENV.SLEEP_WRITE);
-    }
+  if (__ENV.SLEEP_WRITE) {
+    sleep(__ENV.SLEEP_WRITE);
+  }
 
-    const headers = {
-        unique_header: uuidv4()
-    };
-    const container = container_list[Math.floor(Math.random() * container_list.length)];
+  const headers = { unique_header: uuidv4() };
+  const container =
+    container_list[Math.floor(Math.random() * container_list.length)];
 
-    const { payload, hash } = generator.genPayload(registry_enabled);
-    const resp = grpc_client.put(container, headers, payload, write_grpc_chunk_size);
-    if (!resp.success) {
-        log.withField("cid", container).error(resp.error);
-        return;
-    }
+  const payload = generator.genPayload();
+  const resp =
+    grpc_client.put(container, headers, payload, write_grpc_chunk_size);
+  if (!resp.success) {
+    log.withField('cid', container).error(resp.error);
+    return;
+  }
 
-    if (obj_registry) {
-        obj_registry.addObject(container, resp.object_id, "", "", hash);
-    }
+  if (obj_registry) {
+    obj_registry.addObject(container, resp.object_id, '', '', payload.hash());
+  }
 }
 
 export function obj_read() {
-    if (__ENV.SLEEP_READ) {
-        sleep(__ENV.SLEEP_READ);
-    }
+  if (__ENV.SLEEP_READ) {
+    sleep(__ENV.SLEEP_READ);
+  }
 
-    const obj = obj_list[Math.floor(Math.random() * obj_list.length)];
-    const resp = grpc_client.get(obj.container, obj.object)
-    if (!resp.success) {
-        log.withFields({cid: obj.container, oid: obj.object}).error(resp.error);
+  if (obj_to_read_selector) {
+    const obj = obj_to_read_selector.nextObject();
+    if (!obj) {
+      return;
     }
+    const resp = grpc_client.get(obj.c_id, obj.o_id)
+    if (!resp.success) {
+      log.withFields({ cid: obj.c_id, oid: obj.o_id }).error(resp.error);
+    }
+    return
+  }
+
+  const obj = obj_list[Math.floor(Math.random() * obj_list.length)];
+  const resp = grpc_client.get(obj.container, obj.object)
+  if (!resp.success) {
+    log.withFields({ cid: obj.container, oid: obj.object }).error(resp.error);
+  }
 }
 
 export function obj_delete() {
-    if (__ENV.SLEEP_DELETE) {
-        sleep(__ENV.SLEEP_DELETE);
-    }
+  if (__ENV.SLEEP_DELETE) {
+    sleep(__ENV.SLEEP_DELETE);
+  }
 
-    const obj = obj_to_delete_selector.nextObject();
-    if (!obj) {
-        return;
+  const obj = obj_to_delete_selector.nextObject();
+  if (!obj) {
+    if (obj_to_delete_exit_on_null) {
+      exec.test.abort("No more objects to select");
     }
+    return;
+  }
 
-    const resp = grpc_client.delete(obj.c_id, obj.o_id);
-    if (!resp.success) {
-        // Log errors except (2052 - object already deleted)
-        log.withFields({cid: obj.c_id, oid: obj.o_id}).error(resp.error);
-        return;
-    }
+  const resp = grpc_client.delete(obj.c_id, obj.o_id);
+  if (!resp.success) {
+    // Log errors except (2052 - object already deleted)
+    log.withFields({ cid: obj.c_id, oid: obj.o_id }).error(resp.error);
+    return;
+  }
 
-    obj_registry.deleteObject(obj.id);
+  obj_registry.deleteObject(obj.id);
 }

scenarios/grpc_car.js

@@ -1,56 +1,70 @@
-import datagen from 'k6/x/frostfs/datagen';
-import native from 'k6/x/frostfs/native';
-import logging from 'k6/x/frostfs/logging';
-import registry from 'k6/x/frostfs/registry';
-import { SharedArray } from 'k6/data';
 import { sleep } from 'k6';
-import { textSummary } from './libs/k6-summary-0.0.2.js';
+import { SharedArray } from 'k6/data';
+import logging from 'k6/x/frostfs/logging';
+import native from 'k6/x/frostfs/native';
+import registry from 'k6/x/frostfs/registry';
+import stats from 'k6/x/frostfs/stats';
+
+import { newGenerator } from './libs/datagen.js';
 import { parseEnv } from './libs/env-parser.js';
+import { textSummary } from './libs/k6-summary-0.0.2.js';
 import { uuidv4 } from './libs/k6-utils-1.4.0.js';
 
 parseEnv();
 
 const obj_list = new SharedArray('obj_list', function () {
-    return JSON.parse(open(__ENV.PREGEN_JSON)).objects;
+  return JSON.parse(open(__ENV.PREGEN_JSON)).objects;
 });
 
 const container_list = new SharedArray('container_list', function () {
-    return JSON.parse(open(__ENV.PREGEN_JSON)).containers;
+  return JSON.parse(open(__ENV.PREGEN_JSON)).containers;
 });
 
 const read_size = JSON.parse(open(__ENV.PREGEN_JSON)).obj_size;
-const summary_json = __ENV.SUMMARY_JSON || "/tmp/summary.json";
+const summary_json = __ENV.SUMMARY_JSON || '/tmp/summary.json';
 
 // Select random gRPC endpoint for current VU
 const grpc_endpoints = __ENV.GRPC_ENDPOINTS.split(',');
-const grpc_endpoint = grpc_endpoints[Math.floor(Math.random() * grpc_endpoints.length)];
-const grpc_client = native.connect(grpc_endpoint, '',
-    __ENV.DIAL_TIMEOUT ? parseInt(__ENV.DIAL_TIMEOUT) : 5,
-    __ENV.STREAM_TIMEOUT ? parseInt(__ENV.STREAM_TIMEOUT) : 60,
-    __ENV.PREPARE_LOCALLY ? __ENV.PREPARE_LOCALLY.toLowerCase() === "true" : false);
-const log = logging.new().withField("endpoint", grpc_endpoint);
+const grpc_endpoint =
+  grpc_endpoints[Math.floor(Math.random() * grpc_endpoints.length)];
+const grpc_client = native.connect(
+  grpc_endpoint, '', __ENV.DIAL_TIMEOUT ? parseInt(__ENV.DIAL_TIMEOUT) : 5,
+  __ENV.STREAM_TIMEOUT ? parseInt(__ENV.STREAM_TIMEOUT) : 60,
+  __ENV.PREPARE_LOCALLY ? __ENV.PREPARE_LOCALLY.toLowerCase() === 'true' : false,
+  1024 * parseInt(__ENV.MAX_OBJECT_SIZE || '0'));
+const log = logging.new().withField('endpoint', grpc_endpoint);
 
 const registry_enabled = !!__ENV.REGISTRY_FILE;
-const obj_registry = registry_enabled ? registry.open(__ENV.REGISTRY_FILE) : undefined;
+const obj_registry =
+  registry_enabled ? registry.open(__ENV.REGISTRY_FILE) : undefined;
 
 const duration = __ENV.DURATION;
 
+if (!!__ENV.METRIC_TAGS) {
+  stats.setTags(__ENV.METRIC_TAGS)
+}
+
 const delete_age = __ENV.DELETE_AGE ? parseInt(__ENV.DELETE_AGE) : undefined;
 let obj_to_delete_selector = undefined;
 if (registry_enabled && delete_age) {
-    obj_to_delete_selector = registry.getSelector(
-        __ENV.REGISTRY_FILE,
-        "obj_to_delete",
-        __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0,
-        {
-            status: "created",
-            age: delete_age,
-        }
-    );
+  obj_to_delete_selector = registry.getSelector(
+    __ENV.REGISTRY_FILE, 'obj_to_delete',
+    __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0, {
+    status: 'created',
+    age: delete_age,
+  });
 }
 
-
-const generator = datagen.generator(1024 * parseInt(__ENV.WRITE_OBJ_SIZE));
+const read_age = __ENV.READ_AGE ? parseInt(__ENV.READ_AGE) : 10;
+let obj_to_read_selector = undefined;
+if (registry_enabled) {
+  obj_to_read_selector = registry.getLoopedSelector(
+    __ENV.REGISTRY_FILE, 'obj_to_read',
+    __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0, {
+    status: 'created',
+    age: read_age,
+  })
+}
 
 const scenarios = {};
 
@@ -59,144 +73,166 @@ const pre_alloc_write_vus = parseInt(__ENV.PRE_ALLOC_WRITERS || '0');
 const max_write_vus = parseInt(__ENV.MAX_WRITERS || pre_alloc_write_vus);
 const write_rate = parseInt(__ENV.WRITE_RATE || '0');
 const write_grpc_chunk_size = 1024 * parseInt(__ENV.GRPC_CHUNK_SIZE || '0')
+const generator = newGenerator(write_rate > 0);
 if (write_rate > 0) {
-    scenarios.write = {
-        executor: 'constant-arrival-rate',
-        duration: `${duration}s`,
-        preAllocatedVUs: pre_alloc_write_vus,
-        maxVUs: max_write_vus,
-        rate: write_rate,
-        timeUnit: time_unit,
-        exec: 'obj_write', 
-        gracefulStop: '5s',
-    };
+  scenarios.write = {
+    executor: 'constant-arrival-rate',
+    duration: `${duration}s`,
+    preAllocatedVUs: pre_alloc_write_vus,
+    maxVUs: max_write_vus,
+    rate: write_rate,
+    timeUnit: time_unit,
+    exec: 'obj_write',
+    gracefulStop: '5s',
+  };
 }
 
 const pre_alloc_read_vus = parseInt(__ENV.PRE_ALLOC_READERS || '0');
 const max_read_vus = parseInt(__ENV.MAX_READERS || pre_alloc_read_vus);
 const read_rate = parseInt(__ENV.READ_RATE || '0');
 if (read_rate > 0) {
-    scenarios.read = {
-        executor: 'constant-arrival-rate',
-        duration: `${duration}s`,
-        preAllocatedVUs: pre_alloc_write_vus,
-        maxVUs: max_read_vus,
-        rate: read_rate,
-        timeUnit: time_unit,
-        exec: 'obj_read', 
-        gracefulStop: '5s',
-    };
+  scenarios.read = {
+    executor: 'constant-arrival-rate',
+    duration: `${duration}s`,
+    preAllocatedVUs: pre_alloc_write_vus,
+    maxVUs: max_read_vus,
+    rate: read_rate,
+    timeUnit: time_unit,
+    exec: 'obj_read',
+    gracefulStop: '5s',
+  };
 }
 
 const pre_alloc_delete_vus = parseInt(__ENV.PRE_ALLOC_DELETERS || '0');
 const max_delete_vus = parseInt(__ENV.MAX_DELETERS || pre_alloc_write_vus);
 const delete_rate = parseInt(__ENV.DELETE_RATE || '0');
 if (delete_rate > 0) {
-    if (!obj_to_delete_selector) {
-        throw new Error('Positive DELETE worker number without a proper object selector');
-    }
+  if (!obj_to_delete_selector) {
+    throw new Error(
+      'Positive DELETE worker number without a proper object selector');
+  }
 
-    scenarios.delete = {
-        executor: 'constant-arrival-rate',
-        duration: `${duration}s`,
-        preAllocatedVUs: pre_alloc_delete_vus,
-        maxVUs: max_delete_vus,
-        rate: delete_rate,
-        timeUnit: time_unit,
-        exec: 'obj_delete', 
-        gracefulStop: '5s',
-    };
+  scenarios.delete = {
+    executor: 'constant-arrival-rate',
+    duration: `${duration}s`,
+    preAllocatedVUs: pre_alloc_delete_vus,
+    maxVUs: max_delete_vus,
+    rate: delete_rate,
+    timeUnit: time_unit,
+    exec: 'obj_delete',
+    gracefulStop: '5s',
+  };
 }
 
 export const options = {
-    scenarios,
-    setupTimeout: '5s',
+  scenarios,
+  setupTimeout: '5s',
 };
 
 export function setup() {
-    const total_pre_allocated_vu_count = pre_alloc_write_vus + pre_alloc_read_vus + pre_alloc_delete_vus;
-    const total_max_vu_count = max_read_vus + max_write_vus + max_delete_vus
+  const total_pre_allocated_vu_count =
+    pre_alloc_write_vus + pre_alloc_read_vus + pre_alloc_delete_vus;
+  const total_max_vu_count = max_read_vus + max_write_vus + max_delete_vus
 
-    console.log(`Pregenerated containers:       ${container_list.length}`);
-    console.log(`Pregenerated read object size: ${read_size}`);
-    console.log(`Pregenerated total objects:    ${obj_list.length}`);
-    console.log(`Pre allocated reading VUs:     ${pre_alloc_read_vus}`);
-    console.log(`Pre allocated writing VUs:     ${pre_alloc_write_vus}`);
-    console.log(`Pre allocated deleting VUs:    ${pre_alloc_delete_vus}`);
-    console.log(`Total pre allocated VUs:       ${total_pre_allocated_vu_count}`);
-    console.log(`Max reading VUs:               ${max_read_vus}`);
-    console.log(`Max writing VUs:               ${max_write_vus}`);
-    console.log(`Max deleting VUs:              ${max_delete_vus}`);
-    console.log(`Total max VUs:                 ${total_max_vu_count}`);
-    console.log(`Time unit:                     ${time_unit}`);
-    console.log(`Read rate:                     ${read_rate}`);
-    console.log(`Writing rate:                  ${write_rate}`);
-    console.log(`Delete rate:                   ${delete_rate}`);
+  console.log(`Pregenerated containers:       ${container_list.length}`);
+  console.log(`Pregenerated read object size: ${read_size}`);
+  console.log(`Pregenerated total objects:    ${obj_list.length}`);
+  console.log(`Pre allocated reading VUs:     ${pre_alloc_read_vus}`);
+  console.log(`Pre allocated writing VUs:     ${pre_alloc_write_vus}`);
+  console.log(`Pre allocated deleting VUs:    ${pre_alloc_delete_vus}`);
+  console.log(`Total pre allocated VUs:       ${total_pre_allocated_vu_count}`);
+  console.log(`Max reading VUs:               ${max_read_vus}`);
+  console.log(`Max writing VUs:               ${max_write_vus}`);
+  console.log(`Max deleting VUs:              ${max_delete_vus}`);
+  console.log(`Total max VUs:                 ${total_max_vu_count}`);
+  console.log(`Time unit:                     ${time_unit}`);
+  console.log(`Read rate:                     ${read_rate}`);
+  console.log(`Writing rate:                  ${write_rate}`);
+  console.log(`Delete rate:                   ${delete_rate}`);
+
+  const start_timestamp = Date.now()
+  console.log(
+    `Load started at:               ${Date(start_timestamp).toString()}`)
 }
 
 export function teardown(data) {
-    if (obj_registry) {
-        obj_registry.close();
-    }
+  if (obj_registry) {
+    obj_registry.close();
+  }
+  const end_timestamp = Date.now()
+  console.log(
+    `Load finished at:              ${Date(end_timestamp).toString()}`)
 }
 
 export function handleSummary(data) {
-    return {
-        'stdout': textSummary(data, { indent: ' ', enableColors: false }),
-        [summary_json]: JSON.stringify(data),
-    };
+  return {
+    'stdout': textSummary(data, { indent: ' ', enableColors: false }),
+    [summary_json]: JSON.stringify(data),
+  };
 }
 
 export function obj_write() {
-    if (__ENV.SLEEP_WRITE) {
-        sleep(__ENV.SLEEP_WRITE);
-    }
+  if (__ENV.SLEEP_WRITE) {
+    sleep(__ENV.SLEEP_WRITE);
+  }
 
-    const headers = {
-        unique_header: uuidv4()
-    };
-    const container = container_list[Math.floor(Math.random() * container_list.length)];
+  const headers = { unique_header: uuidv4() };
+  const container =
+    container_list[Math.floor(Math.random() * container_list.length)];
 
-    const { payload, hash } = generator.genPayload(registry_enabled);
-    const resp = grpc_client.put(container, headers, payload, write_grpc_chunk_size);
-    if (!resp.success) {
-        log.withField("cid", container).error(resp.error);
-        return;
-    }
+  const payload = generator.genPayload();
+  const resp =
+    grpc_client.put(container, headers, payload, write_grpc_chunk_size);
+  if (!resp.success) {
+    log.withField('cid', container).error(resp.error);
+    return;
+  }
 
-    if (obj_registry) {
-        obj_registry.addObject(container, resp.object_id, "", "", hash);
-    }
+  if (obj_registry) {
+    obj_registry.addObject(container, resp.object_id, '', '', payload.hash());
+  }
 }
 
 export function obj_read() {
-    if (__ENV.SLEEP_READ) {
-        sleep(__ENV.SLEEP_READ);
-    }
+  if (__ENV.SLEEP_READ) {
+    sleep(__ENV.SLEEP_READ);
+  }
 
-    const obj = obj_list[Math.floor(Math.random() * obj_list.length)];
-    const resp = grpc_client.get(obj.container, obj.object)
-    if (!resp.success) {
-        log.withFields({cid: obj.container, oid: obj.object}).error(resp.error);
+  if (obj_to_read_selector) {
+    const obj = obj_to_read_selector.nextObject();
+    if (!obj) {
+      return;
     }
+    const resp = grpc_client.get(obj.c_id, obj.o_id)
+    if (!resp.success) {
+      log.withFields({ cid: obj.c_id, oid: obj.o_id }).error(resp.error);
+    }
+    return
+  }
+
+  const obj = obj_list[Math.floor(Math.random() * obj_list.length)];
+  const resp = grpc_client.get(obj.container, obj.object)
+  if (!resp.success) {
+    log.withFields({ cid: obj.container, oid: obj.object }).error(resp.error);
+  }
 }
 
 export function obj_delete() {
-    if (__ENV.SLEEP_DELETE) {
-        sleep(__ENV.SLEEP_DELETE);
-    }
+  if (__ENV.SLEEP_DELETE) {
+    sleep(__ENV.SLEEP_DELETE);
+  }
 
-    const obj = obj_to_delete_selector.nextObject();
-    if (!obj) {
-        return;
-    }
+  const obj = obj_to_delete_selector.nextObject();
+  if (!obj) {
+    return;
+  }
 
-    const resp = grpc_client.delete(obj.c_id, obj.o_id);
-    if (!resp.success) {
-        // Log errors except (2052 - object already deleted)
-        log.withFields({cid: obj.c_id, oid: obj.o_id}).error(resp.error);
-        return;
-    }
+  const resp = grpc_client.delete(obj.c_id, obj.o_id);
+  if (!resp.success) {
+    // Log errors except (2052 - object already deleted)
+    log.withFields({ cid: obj.c_id, oid: obj.o_id }).error(resp.error);
+    return;
+  }
 
-    obj_registry.deleteObject(obj.id);
+  obj_registry.deleteObject(obj.id);
 }

scenarios/http.js

@@ -1,123 +1,143 @@
-import datagen from 'k6/x/frostfs/datagen';
+import {sleep} from 'k6';
+import {SharedArray} from 'k6/data';
+import http from 'k6/http';
 import logging from 'k6/x/frostfs/logging';
 import registry from 'k6/x/frostfs/registry';
-import http from 'k6/http';
-import { SharedArray } from 'k6/data';
-import { sleep } from 'k6';
-import { textSummary } from './libs/k6-summary-0.0.2.js';
-import { parseEnv } from './libs/env-parser.js';
-import { uuidv4 } from './libs/k6-utils-1.4.0.js';
+import stats from 'k6/x/frostfs/stats';
+
+import {newGenerator} from './libs/datagen.js';
+import {parseEnv} from './libs/env-parser.js';
+import {textSummary} from './libs/k6-summary-0.0.2.js';
+import {uuidv4} from './libs/k6-utils-1.4.0.js';
 
 parseEnv();
 
-const obj_list = new SharedArray('obj_list', function () {
-    return JSON.parse(open(__ENV.PREGEN_JSON)).objects;
+const obj_list = new SharedArray('obj_list', function() {
+  return JSON.parse(open(__ENV.PREGEN_JSON)).objects;
 });
 
-const container_list = new SharedArray('container_list', function () {
-    return JSON.parse(open(__ENV.PREGEN_JSON)).containers;
+const container_list = new SharedArray('container_list', function() {
+  return JSON.parse(open(__ENV.PREGEN_JSON)).containers;
 });
 
 const read_size = JSON.parse(open(__ENV.PREGEN_JSON)).obj_size;
-const summary_json = __ENV.SUMMARY_JSON || "/tmp/summary.json";
+const summary_json = __ENV.SUMMARY_JSON || '/tmp/summary.json';
 
 // Select random HTTP endpoint for current VU
 const http_endpoints = __ENV.HTTP_ENDPOINTS.split(',');
-const http_endpoint = http_endpoints[Math.floor(Math.random() * http_endpoints.length)];
-const log = logging.new().withField("endpoint", http_endpoint);
+const http_endpoint =
+    http_endpoints[Math.floor(Math.random() * http_endpoints.length)];
+const log = logging.new().withField('endpoint', http_endpoint);
 
 const registry_enabled = !!__ENV.REGISTRY_FILE;
-const obj_registry = registry_enabled ? registry.open(__ENV.REGISTRY_FILE) : undefined;
+const obj_registry =
+    registry_enabled ? registry.open(__ENV.REGISTRY_FILE) : undefined;
 
 const duration = __ENV.DURATION;
 
-const generator = datagen.generator(1024 * parseInt(__ENV.WRITE_OBJ_SIZE), __ENV.PAYLOAD_TYPE || "");
+if (!!__ENV.METRIC_TAGS) {
+  stats.setTags(__ENV.METRIC_TAGS)
+}
 
 const scenarios = {};
 
 const write_vu_count = parseInt(__ENV.WRITERS || '0');
+const generator = newGenerator(write_vu_count > 0);
 if (write_vu_count > 0) {
-    scenarios.write = {
-        executor: 'constant-vus',
-        vus: write_vu_count,
-        duration: `${duration}s`,
-        exec: 'obj_write', 
-        gracefulStop: '5s',
-    }
+  scenarios.write = {
+    executor: 'constant-vus',
+    vus: write_vu_count,
+    duration: `${duration}s`,
+    exec: 'obj_write',
+    gracefulStop: '5s',
+  }
 }
 
 const read_vu_count = parseInt(__ENV.READERS || '0');
 if (read_vu_count > 0) {
-    scenarios.read = {
-        executor: 'constant-vus',
-        vus: read_vu_count,
-        duration: `${duration}s`,
-        exec: 'obj_read', 
-        gracefulStop: '5s',
-    }
+  scenarios.read = {
+    executor: 'constant-vus',
+    vus: read_vu_count,
+    duration: `${duration}s`,
+    exec: 'obj_read',
+    gracefulStop: '5s',
+  }
 }
 
 export const options = {
-    scenarios,
-    setupTimeout: '5s',
+  scenarios,
+  setupTimeout: '5s',
 };
 
 export function setup() {
-    const total_vu_count = write_vu_count + read_vu_count;
+  const total_vu_count = write_vu_count + read_vu_count;
 
-    console.log(`Pregenerated containers:       ${container_list.length}`);
-    console.log(`Pregenerated read object size: ${read_size}`);
-    console.log(`Pregenerated total objects:    ${obj_list.length}`);
-    console.log(`Reading VUs:                   ${read_vu_count}`);
-    console.log(`Writing VUs:                   ${write_vu_count}`);
-    console.log(`Total VUs:                     ${total_vu_count}`);
+  console.log(`Pregenerated containers:       ${container_list.length}`);
+  console.log(`Pregenerated read object size: ${read_size}`);
+  console.log(`Pregenerated total objects:    ${obj_list.length}`);
+  console.log(`Reading VUs:                   ${read_vu_count}`);
+  console.log(`Writing VUs:                   ${write_vu_count}`);
+  console.log(`Total VUs:                     ${total_vu_count}`);
+
+  const start_timestamp = Date.now()
+  console.log(
+      `Load started at:               ${Date(start_timestamp).toString()}`)
 }
 
 export function teardown(data) {
-    if (obj_registry) {
-        obj_registry.close();
-    }
+  if (obj_registry) {
+    obj_registry.close();
+  }
+  const end_timestamp = Date.now()
+  console.log(
+      `Load finished at:              ${Date(end_timestamp).toString()}`)
 }
 
 export function handleSummary(data) {
-    return {
-        'stdout': textSummary(data, { indent: ' ', enableColors: false }),
-        [summary_json]: JSON.stringify(data),
-    };
+  return {
+    'stdout': textSummary(data, {indent: ' ', enableColors: false}),
+    [summary_json]: JSON.stringify(data),
+  };
 }
 
 export function obj_write() {
-    if (__ENV.SLEEP_WRITE) {
-        sleep(__ENV.SLEEP_WRITE);
-    }
+  if (__ENV.SLEEP_WRITE) {
+    sleep(__ENV.SLEEP_WRITE);
+  }
 
-    const container = container_list[Math.floor(Math.random() * container_list.length)];
+  const container =
+      container_list[Math.floor(Math.random() * container_list.length)];
 
-    const { payload, hash } = generator.genPayload(registry_enabled);
-    const data = {
-        field: uuidv4(),
-        file: http.file(payload, "random.data"),
-    };
+  const payload = generator.genPayload();
+  const data = {
+    field: uuidv4(),
+    // Because we use `file` wrapping and it is not straightforward to use
+    // streams here,
+    // `-e STREAMING=1` has no effect for this scenario.
+    file: http.file(payload.bytes(), 'random.data'),
+  };
 
-    const resp = http.post(`http://${http_endpoint}/upload/${container}`, data);
-    if (resp.status != 200) {
-        log.withFields({status: resp.status, cid: container}).error(resp.error);
-        return;
-    }
-    const object_id = JSON.parse(resp.body).object_id;
-    if (obj_registry) {
-        obj_registry.addObject(container, object_id, "", "", hash);
-    }
+  const resp = http.post(`http://${http_endpoint}/upload/${container}`, data);
+  if (resp.status != 200) {
+    log.withFields({status: resp.status, cid: container}).error(resp.error);
+    return;
+  }
+  const object_id = JSON.parse(resp.body).object_id;
+  if (obj_registry) {
+    obj_registry.addObject(container, object_id, '', '', payload.hash());
+  }
 }
 
 export function obj_read() {
-    if (__ENV.SLEEP_READ) {
-        sleep(__ENV.SLEEP_READ);
-    }
+  if (__ENV.SLEEP_READ) {
+    sleep(__ENV.SLEEP_READ);
+  }
 
-    const obj = obj_list[Math.floor(Math.random() * obj_list.length)];
-    const resp = http.get(`http://${http_endpoint}/get/${obj.container}/${obj.object}`);
-    if (resp.status != 200) {
-        log.withFields({status: resp.status, cid: obj.container, oid: obj.object}).error(resp.error);
-    }
+  const obj = obj_list[Math.floor(Math.random() * obj_list.length)];
+  const resp =
+      http.get(`http://${http_endpoint}/get/${obj.container}/${obj.object}`);
+  if (resp.status != 200) {
+    log.withFields({status: resp.status, cid: obj.container, oid: obj.object})
+        .error(resp.error);
+  }
 }

scenarios/libs/datagen.js

@@ -0,0 +1,8 @@
+import datagen from 'k6/x/frostfs/datagen';
+
+export function newGenerator(condition) {
+	if (condition) {
+		return datagen.generator(1024 * parseInt(__ENV.WRITE_OBJ_SIZE), __ENV.PAYLOAD_TYPE || "", !!__ENV.STREAMING);
+	}
+	return undefined;
+}

scenarios/libs/keygen.js

@@ -0,0 +1,34 @@
+import { uuidv4 } from './k6-utils-1.4.0.js';
+
+export function generateS3Key() {
+  let width = parseInt(__ENV.DIR_WIDTH || '0');
+  let height = parseInt(__ENV.DIR_HEIGHT || '0');
+
+  let key = ''
+  if (width > 0 && height > 0) {
+    for (let index = 0; index < height; index++) {
+      const w = Math.floor(Math.random() * width) + 1;
+      key = key + 'dir' + w + '/';
+    }
+  }
+
+  key += objName();
+  return key;
+}
+
+const asciiLetters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890"
+
+function objName() {
+  if (__ENV.OBJ_NAME) {
+    return __ENV.OBJ_NAME;
+  }
+  const length = parseInt(__ENV.OBJ_NAME_LENGTH || '0');
+  if (length > 0) {
+    let name = "";
+    for (let i = 0; i < length; i++) {
+      name += asciiLetters.charAt(Math.floor(Math.random() * asciiLetters.length));
+    }
+    return name;
+  }
+  return uuidv4();
+}

scenarios/local.js

@@ -1,158 +1,177 @@
-import datagen from 'k6/x/frostfs/datagen';
+import {SharedArray} from 'k6/data';
+import exec from 'k6/execution';
 import local from 'k6/x/frostfs/local';
 import logging from 'k6/x/frostfs/logging';
 import registry from 'k6/x/frostfs/registry';
-import { SharedArray } from 'k6/data';
-import { textSummary } from './libs/k6-summary-0.0.2.js';
-import { parseEnv } from './libs/env-parser.js';
-import { uuidv4 } from './libs/k6-utils-1.4.0.js';
+import stats from 'k6/x/frostfs/stats';
+
+import {newGenerator} from './libs/datagen.js';
+import {parseEnv} from './libs/env-parser.js';
+import {textSummary} from './libs/k6-summary-0.0.2.js';
+import {uuidv4} from './libs/k6-utils-1.4.0.js';
 
 parseEnv();
 
-const obj_list = new SharedArray('obj_list', function () {
-    return JSON.parse(open(__ENV.PREGEN_JSON)).objects;
+const obj_list = new SharedArray('obj_list', function() {
+  return JSON.parse(open(__ENV.PREGEN_JSON)).objects;
 });
 
-const container_list = new SharedArray('container_list', function () {
-    return JSON.parse(open(__ENV.PREGEN_JSON)).containers;
+const container_list = new SharedArray('container_list', function() {
+  return JSON.parse(open(__ENV.PREGEN_JSON)).containers;
 });
 
 const read_size = JSON.parse(open(__ENV.PREGEN_JSON)).obj_size;
-const summary_json = __ENV.SUMMARY_JSON || "/tmp/summary.json";
+const summary_json = __ENV.SUMMARY_JSON || '/tmp/summary.json';
 
 const config_file = __ENV.CONFIG_FILE;
+const config_dir = __ENV.CONFIG_DIR;
 const debug_logger = (__ENV.DEBUG_LOGGER || 'false') == 'true';
-const local_client = local.connect(config_file, '', debug_logger);
-const log = logging.new().withField("config", config_file);
+const max_total_size_gb =
+    __ENV.MAX_TOTAL_SIZE_GB ? parseInt(__ENV.MAX_TOTAL_SIZE_GB) : 0;
+const local_client =
+    local.connect(config_file, config_dir, '', debug_logger, max_total_size_gb);
+const log = logging.new().withFields(
+    {'config_file': config_file, 'config_dir': config_dir});
 
 const registry_enabled = !!__ENV.REGISTRY_FILE;
-const obj_registry = registry_enabled ? registry.open(__ENV.REGISTRY_FILE) : undefined;
+const obj_registry =
+    registry_enabled ? registry.open(__ENV.REGISTRY_FILE) : undefined;
 
 const duration = __ENV.DURATION;
 
+if (!!__ENV.METRIC_TAGS) {
+  stats.setTags(__ENV.METRIC_TAGS)
+}
+
 const delete_age = __ENV.DELETE_AGE ? parseInt(__ENV.DELETE_AGE) : undefined;
 let obj_to_delete_selector = undefined;
 if (registry_enabled && delete_age) {
-    obj_to_delete_selector = registry.getSelector(
-        __ENV.REGISTRY_FILE,
-        "obj_to_delete",
-        __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0,
-        {
-            status: "created",
-            age: delete_age,
-        }
-    );
+  obj_to_delete_selector = registry.getSelector(
+      __ENV.REGISTRY_FILE, 'obj_to_delete',
+      __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0, {
+        status: 'created',
+        age: delete_age,
+      });
 }
 
-const generator = datagen.generator(1024 * parseInt(__ENV.WRITE_OBJ_SIZE));
-
 const scenarios = {};
 
 const write_vu_count = parseInt(__ENV.WRITERS || '0');
+const generator = newGenerator(write_vu_count > 0);
 if (write_vu_count > 0) {
-    scenarios.write = {
-        executor: 'constant-vus',
-        vus: write_vu_count,
-        duration: `${duration}s`,
-        exec: 'obj_write', 
-        gracefulStop: '5s',
-    };
+  scenarios.write = {
+    executor: 'constant-vus',
+    vus: write_vu_count,
+    duration: `${duration}s`,
+    exec: 'obj_write',
+    gracefulStop: '5s',
+  };
 }
 
 const read_vu_count = parseInt(__ENV.READERS || '0');
 if (read_vu_count > 0) {
-    scenarios.read = {
-        executor: 'constant-vus',
-        vus: read_vu_count,
-        duration: `${duration}s`,
-        exec: 'obj_read', 
-        gracefulStop: '5s',
-    };
+  scenarios.read = {
+    executor: 'constant-vus',
+    vus: read_vu_count,
+    duration: `${duration}s`,
+    exec: 'obj_read',
+    gracefulStop: '5s',
+  };
 }
 
 const delete_vu_count = parseInt(__ENV.DELETERS || '0');
 if (delete_vu_count > 0) {
-    if (!obj_to_delete_selector) {
-        throw new Error('Positive DELETE worker number without a proper object selector');
-    }
+  if (!obj_to_delete_selector) {
+    throw new Error(
+        'Positive DELETE worker number without a proper object selector');
+  }
 
-    scenarios.delete = {
-        executor: 'constant-vus',
-        vus: delete_vu_count,
-        duration: `${duration}s`,
-        exec: 'obj_delete', 
-        gracefulStop: '5s',
-    };
+  scenarios.delete = {
+    executor: 'constant-vus',
+    vus: delete_vu_count,
+    duration: `${duration}s`,
+    exec: 'obj_delete',
+    gracefulStop: '5s',
+  };
 }
 
 export const options = {
-    scenarios,
-    setupTimeout: '5s',
+  scenarios,
+  setupTimeout: '5s',
 };
 
 export function setup() {
-    const total_vu_count = write_vu_count + read_vu_count + delete_vu_count;
+  const total_vu_count = write_vu_count + read_vu_count + delete_vu_count;
 
-    console.log(`Pregenerated containers:       ${container_list.length}`);
-    console.log(`Pregenerated read object size: ${read_size}`);
-    console.log(`Pregenerated total objects:    ${obj_list.length}`);
-    console.log(`Reading VUs:                   ${read_vu_count}`);
-    console.log(`Writing VUs:                   ${write_vu_count}`);
-    console.log(`Deleting VUs:                  ${delete_vu_count}`);
-    console.log(`Total VUs:                     ${total_vu_count}`);
+  console.log(`Pregenerated containers:       ${container_list.length}`);
+  console.log(`Pregenerated read object size: ${read_size}`);
+  console.log(`Pregenerated total objects:    ${obj_list.length}`);
+  console.log(`Reading VUs:                   ${read_vu_count}`);
+  console.log(`Writing VUs:                   ${write_vu_count}`);
+  console.log(`Deleting VUs:                  ${delete_vu_count}`);
+  console.log(`Total VUs:                     ${total_vu_count}`);
+
+  const start_timestamp = Date.now()
+  console.log(
+      `Load started at:               ${Date(start_timestamp).toString()}`)
 }
 
 export function teardown(data) {
-    if (obj_registry) {
-        obj_registry.close();
-    }
+  if (obj_registry) {
+    obj_registry.close();
+  }
+  const end_timestamp = Date.now()
+  console.log(
+      `Load finished at:              ${Date(end_timestamp).toString()}`)
 }
 
 export function handleSummary(data) {
-    return {
-        'stdout': textSummary(data, { indent: ' ', enableColors: false }),
-        [summary_json]: JSON.stringify(data),
-    };
+  return {
+    'stdout': textSummary(data, {indent: ' ', enableColors: false}),
+    [summary_json]: JSON.stringify(data),
+  };
 }
 
 export function obj_write() {
-    const headers = {
-        unique_header: uuidv4()
-    };
-    const container = container_list[Math.floor(Math.random() * container_list.length)];
+  const headers = {unique_header: uuidv4()};
+  const container =
+      container_list[Math.floor(Math.random() * container_list.length)];
 
-    const { payload, hash } = generator.genPayload(registry_enabled);
-    const resp = local_client.put(container, headers, payload);
-    if (!resp.success) {
-        log.withField("cid", container).error(resp.error);
-        return;
+  const payload = generator.genPayload();
+  const resp = local_client.put(container, headers, payload);
+  if (!resp.success) {
+    if (resp.abort) {
+      exec.test.abort(resp.error);
     }
+    log.withField('cid', container).error(resp.error);
+    return;
+  }
 
-    if (obj_registry) {
-        obj_registry.addObject(container, resp.object_id, "", "", hash);
-    }
+  if (obj_registry) {
+    obj_registry.addObject(container, resp.object_id, '', '', payload.hash());
+  }
 }
 
 export function obj_read() {
-    const obj = obj_list[Math.floor(Math.random() * obj_list.length)];
-    const resp = local_client.get(obj.container, obj.object)
-    if (!resp.success) {
-        log.withFields({cid: obj.container, oid: obj.object}).error(resp.error);
-    }
+  const obj = obj_list[Math.floor(Math.random() * obj_list.length)];
+  const resp = local_client.get(obj.container, obj.object)
+  if (!resp.success) {
+    log.withFields({cid: obj.container, oid: obj.object}).error(resp.error);
+  }
 }
 
 export function obj_delete() {
-    const obj = obj_to_delete_selector.nextObject();
-    if (!obj) {
-        return;
-    }
+  const obj = obj_to_delete_selector.nextObject();
+  if (!obj) {
+    return;
+  }
 
-    const resp = local_client.delete(obj.c_id, obj.o_id);
-    if (!resp.success) {
-        // Log errors except (2052 - object already deleted)
-        log.withFields({cid: obj.c_id, oid: obj.o_id}).error(resp.error);
-        return;
-    }
+  const resp = local_client.delete(obj.c_id, obj.o_id);
+  if (!resp.success) {
+    // Log errors except (2052 - object already deleted)
+    log.withFields({cid: obj.c_id, oid: obj.o_id}).error(resp.error);
+    return;
+  }
 
-    obj_registry.deleteObject(obj.id);
+  obj_registry.deleteObject(obj.id);
 }

scenarios/preset/helpers/aws_cli.py

@@ -1,46 +1,54 @@
 import uuid
 
-from helpers.cmd import execute_cmd
+from helpers.cmd import execute_cmd, log
 
 
-def create_bucket(endpoint, versioning, location):
+def create_bucket(endpoint, versioning, location, acl, no_verify_ssl):
+    configuration = ""
     if location:
-        location = f"--create-bucket-configuration 'LocationConstraint={location}'"
+        configuration = f"--create-bucket-configuration 'LocationConstraint={location}'"
+    if acl:
+        acl = f"--acl {acl}"
+
     bucket_name = str(uuid.uuid4())
+    no_verify_ssl_str = "--no-verify-ssl" if no_verify_ssl else ""
+    cmd_line = f"aws {no_verify_ssl_str} s3api create-bucket --bucket {bucket_name} " \
+               f"--endpoint {endpoint} {configuration} {acl} "
+    cmd_line_ver = f"aws {no_verify_ssl_str} s3api put-bucket-versioning --bucket {bucket_name} " \
+                   f"--versioning-configuration Status=Enabled --endpoint {endpoint}"
 
-    cmd_line = f"aws --no-verify-ssl s3api create-bucket --bucket {bucket_name} " \
-               f"--endpoint http://{endpoint} {location}"
-    cmd_line_ver = f"aws --no-verify-ssl s3api put-bucket-versioning --bucket {bucket_name} " \
-                   f"--versioning-configuration Status=Enabled --endpoint http://{endpoint} "
+    output, success = execute_cmd(cmd_line)
 
-    out, success = execute_cmd(cmd_line)
-
-    if not success and "succeeded and you already own it" not in out:
-        print(f" > Bucket {bucket_name} has not been created:\n{out}")
+    if not success and "succeeded and you already own it" not in output:
+        log(f"{cmd_line}\n"
+            f"Bucket {bucket_name} has not been created:\n"
+            f"Error: {output}", endpoint)
         return False
-    
-    print(f"cmd: {cmd_line}")
 
-    if versioning == "True":
-        out, success = execute_cmd(cmd_line_ver)
+    if versioning:
+        output, success = execute_cmd(cmd_line_ver)
         if not success:
-            print(f" > Bucket versioning has not been applied for bucket {bucket_name}:\n{out}")
+            log(f"{cmd_line_ver}\n"
+                f"Bucket versioning has not been applied for bucket {bucket_name}\n"
+                f"Error: {output}", endpoint)
         else:
-            print(f" > Bucket versioning has been applied.")
+            log(f"Bucket versioning has been applied for bucket {bucket_name}", endpoint)
     
-    print(f"Created bucket: {bucket_name} via endpoint {endpoint}")
+    log(f"Created bucket: {bucket_name} ({location})", endpoint)
     return bucket_name
 
 
-def upload_object(bucket, payload_filepath, endpoint):
+def upload_object(bucket, payload_filepath, endpoint, no_verify_ssl):
     object_name = str(uuid.uuid4())
-
-    cmd_line = f"aws --no-verify-ssl s3api put-object --bucket {bucket} --key {object_name} " \
-               f"--body {payload_filepath} --endpoint http://{endpoint}"
-    out, success = execute_cmd(cmd_line)
+    no_verify_ssl_str = "--no-verify-ssl" if no_verify_ssl else ""
+    cmd_line = f"aws {no_verify_ssl_str} s3api put-object --bucket {bucket} --key {object_name} " \
+               f"--body {payload_filepath} --endpoint {endpoint}"
+    output, success = execute_cmd(cmd_line)
 
     if not success:
-        print(f" > Object {object_name} has not been uploaded.")
+        log(f"{cmd_line}\n"
+            f"Object {object_name} has not been uploaded\n"
+            f"Error: {output}", endpoint)
         return False
     
     return bucket, endpoint, object_name

scenarios/preset/helpers/cmd.py

@@ -1,9 +1,12 @@
 import os
 import shlex
 import sys
-
+from datetime import datetime
 from subprocess import check_output, CalledProcessError, STDOUT
 
+def log(message, endpoint):
+    time = datetime.utcnow()
+    print(f"{time} at {endpoint}: {message}")
 
 def execute_cmd(cmd_line):
     cmd_args = shlex.split(cmd_line)

scenarios/preset/helpers/frostfs_cli.py

@@ -1,47 +1,119 @@
 import re
+from helpers.cmd import execute_cmd, log
 
-from helpers.cmd import execute_cmd
+def create_container(endpoint, policy, container_creation_retry, wallet_path, config, acl, local=False, retry=0):
+    if retry > int(container_creation_retry):
+        raise ValueError(f"unable to create container: too many unsuccessful attempts")
 
-
-def create_container(endpoint, policy, wallet_file, wallet_config):
-    cmd_line = f"frostfs-cli --rpc-endpoint {endpoint} container create --wallet {wallet_file} --config {wallet_config} " \
-               f" --policy '{policy}' --basic-acl public-read-write --await"
+    if wallet_path:
+        wallet_file = f"--wallet {wallet_path}"
+    if config:
+        wallet_config = f"--config {config}"
+    if acl:
+        acl_param = f"--basic-acl {acl}"
+    cmd_line = f"frostfs-cli --rpc-endpoint {endpoint} container create {wallet_file} {wallet_config} " \
+               f" --policy '{policy}' {acl_param} --await"
 
     output, success = execute_cmd(cmd_line)
 
     if not success:
-        print(f" > Container has not been created:\n{output}")
+        log(f"{cmd_line}\n"
+            f"Container has not been created\n"
+            f"{output}", endpoint)
         return False
     
     try:
         fst_str = output.split('\n')[0]
     except Exception:
-        print(f"Got empty output: {output}")
+        log(f"{cmd_line}\n"
+            f"Incorrect output\n"
+            f"Output: {output or '<empty>'}", endpoint)
         return False
     splitted = fst_str.split(": ")
     if len(splitted) != 2:
-        raise ValueError(f"no CID was parsed from command output: \t{fst_str}")
+        raise ValueError(f"no CID was parsed from command output:\t{fst_str}")
+    cid = splitted[1]
 
-    print(f"Created container: {splitted[1]} via endpoint {endpoint}")
+    log(f"Created container: {cid} ({policy})", endpoint)
 
-    return splitted[1]
+    if not local:
+        return cid
+    
+    cmd_line = f"frostfs-cli netmap nodeinfo --rpc-endpoint {endpoint} {wallet_file} {wallet_config}"
+    output, success = execute_cmd(cmd_line)
+
+    if not success:
+        log(f"{cmd_line}\n"
+            f"Failed to get nodeinfo\n"
+            f"{output}", endpoint)
+        return False
+    
+    try:
+        fst_str = output.split('\n')[0]
+    except Exception:
+        log(f"{cmd_line}\n"
+            f"Incorrect output\n"
+            f"Output: {output or '<empty>'}", endpoint)
+        return False
+    splitted = fst_str.split(": ")
+    if len(splitted) != 2 or len(splitted[1]) == 0:
+        raise ValueError(f"no node key was parsed from command output:\t{fst_str}")
+    
+    node_key = splitted[1]
+
+    cmd_line = f"frostfs-cli container nodes --rpc-endpoint {endpoint} {wallet_file} {wallet_config} --cid {cid}"
+    output, success = execute_cmd(cmd_line)
+
+    if not success:
+        log(f"{cmd_line}\n"
+            f"Failed to get container nodes\n"
+            f"{output}", endpoint)
+        return False
+
+    for output_str in output.split('\n'):
+        output_str = output_str.lstrip().rstrip()
+        if not output_str.startswith("Node "):
+            continue
+        splitted = output_str.split(": ")
+        if len(splitted) != 2 or len(splitted[1]) == 0:
+            continue
+        try:
+            k = splitted[1].split(" ")[0]
+        except Exception:
+            log(f"{cmd_line}\n"
+                f"Incorrect output\n"
+                f"Output: {output or '<empty>'}", endpoint)
+            continue
+        if k == node_key:
+            return cid
+    
+    log(f"Created container {cid} is not stored on {endpoint}, creating another one...", endpoint)
+    return create_container(endpoint, policy, container_creation_retry, wallet_path, config, acl, local, retry + 1)
 
 
 def upload_object(container, payload_filepath, endpoint, wallet_file, wallet_config):
     object_name = ""
-    cmd_line = f"frostfs-cli --rpc-endpoint {endpoint} object put --file {payload_filepath} --wallet {wallet_file} --config {wallet_config} " \
+    if wallet_file:
+        wallet_file = "--wallet " + wallet_file
+    if wallet_config:
+        wallet_config = "--config " + wallet_config
+    cmd_line = f"frostfs-cli --rpc-endpoint {endpoint} object put --file {payload_filepath} {wallet_file} {wallet_config} " \
                f"--cid {container} --no-progress"
     output, success = execute_cmd(cmd_line)
 
     if not success:
-        print(f" > Object {object_name} has not been uploaded:\n{output}")
+        log(f"{cmd_line}\n"
+            f"Object {object_name} has not been uploaded\n"
+            f"Error: {output}", endpoint)
         return False
 
     try:
         # taking second string from command output
         snd_str = output.split('\n')[1]
     except Exception:
-        print(f"Got empty input: {output}")
+        log(f"{cmd_line}\n"
+            f"Incorrect output\n"
+            f"Output: {output or '<empty>'}", endpoint)
         return False
     splitted = snd_str.split(": ")
     if len(splitted) != 2:
@@ -50,32 +122,42 @@ def upload_object(container, payload_filepath, endpoint, wallet_file, wallet_con
 
 
 def get_object(cid, oid, endpoint, out_filepath, wallet_file, wallet_config):
-    cmd_line = f"frostfs-cli object get -r {endpoint} --cid {cid} --oid {oid} --wallet {wallet_file} --config {wallet_config} " \
+    if wallet_file:
+        wallet_file = "--wallet " + wallet_file
+    if wallet_config:
+        wallet_config = "--config " + wallet_config
+    cmd_line = f"frostfs-cli object get -r {endpoint} --cid {cid} --oid {oid} {wallet_file} {wallet_config} " \
                f"--file {out_filepath}"
 
     output, success = execute_cmd(cmd_line)
 
     if not success:
-        print(f" > Failed to get object {output} from container {cid} \r\n"
-              f" > Error: {output}")
+        log(f"{cmd_line}\n"
+            f"Failed to get object {oid} from container {cid}\n"
+            f"Error: {output}", endpoint)
         return False
 
     return True
 
 
 def search_object_by_id(cid, oid, endpoint, wallet_file, wallet_config, ttl=2):
-    cmd_line = f"frostfs-cli object search --ttl {ttl} -r {endpoint} --cid {cid} --oid {oid} --wallet {wallet_file} --config {wallet_config} "
+    if wallet_file:
+        wallet_file = "--wallet " + wallet_file
+    if wallet_config:
+        wallet_config = "--config " + wallet_config
+    cmd_line = f"frostfs-cli object search --ttl {ttl} -r {endpoint} --cid {cid} --oid {oid} {wallet_file} {wallet_config} "
 
     output, success = execute_cmd(cmd_line)
 
     if not success:
-        print(f" > Failed to search object {oid} for container {cid} \r\n"
-              f" > Error: {output}")
+        log(f"{cmd_line}\n"
+            f"Failed to search object {oid} for container {cid}\n"
+            f"Error: {output}", endpoint)
         return False
 
     re_rst = re.search(r'Found (\d+) objects', output)
 
     if not re_rst:
-        raise Exception("Failed to parce search results")
+        raise Exception("Failed to parse search results")
 
     return re_rst.group(1)

scenarios/preset/preset_grpc.py

@@ -3,39 +3,41 @@
 import argparse
 from itertools import cycle
 import json
-import random
 import sys
 import tempfile
 import time
 
 from argparse import Namespace
 from concurrent.futures import ProcessPoolExecutor
-
 from helpers.cmd import random_payload
 from helpers.frostfs_cli import create_container, upload_object
 
 ERROR_WRONG_CONTAINERS_COUNT = 1
 ERROR_WRONG_OBJECTS_COUNT = 2
 MAX_WORKERS = 50
+DEFAULT_POLICY = "REP 2 IN X CBF 2 SELECT 2 FROM * AS X"
 
 parser = argparse.ArgumentParser()
 parser.add_argument('--size', help='Upload objects size in kb')
 parser.add_argument('--containers', help='Number of containers to create')
+parser.add_argument('--retry', default=20, help='Maximum number of retries to create a container')
 parser.add_argument('--out', help='JSON file with output')
 parser.add_argument('--preload_obj', help='Number of pre-loaded objects')
 parser.add_argument('--wallet', help='Wallet file path')
 parser.add_argument('--config', help='Wallet config file path')
 parser.add_argument(
     "--policy",
-    help="Container placement policy",
-    default="REP 2 IN X CBF 2 SELECT 2 FROM * AS X"
+    help=f"Container placement policy. Default is {DEFAULT_POLICY}",
+    action="append"
 )
 parser.add_argument('--endpoint', help='Nodes addresses separated by comma.')
 parser.add_argument('--update', help='Save existed containers')
-parser.add_argument('--ignore-errors', help='Ignore preset errors')
+parser.add_argument('--ignore-errors', help='Ignore preset errors', action='store_true')
 parser.add_argument('--workers', help='Count of workers in preset. Max = 50, Default = 50', default=50)
-parser.add_argument('--sleep', help='Time to sleep between container creation and object PUT (in seconds), '
+parser.add_argument('--sleep', help='Time to sleep between containers creation and objects upload (in seconds), '
                                     'Default = 8', default=8)
+parser.add_argument('--local', help='Create containers that store data on provided endpoints. Warning: additional empty containers may be created.', action='store_true')
+parser.add_argument('--acl', help='Container ACL. Default is public-read-write.', default='public-read-write')
 
 args: Namespace = parser.parse_args()
 print(args)
@@ -46,13 +48,16 @@ def main():
     objects_list = []
 
     endpoints = args.endpoint.split(',')
+    if not args.policy:
+        args.policy = [DEFAULT_POLICY]
 
+    container_creation_retry = args.retry
     wallet = args.wallet
     wallet_config = args.config
     workers = int(args.workers)
     objects_per_container = int(args.preload_obj)
     
-    ignore_errors = True if args.ignore_errors else False
+    ignore_errors = args.ignore_errors
     if args.update:
         # Open file
         with open(args.out) as f:
@@ -63,9 +68,9 @@ def main():
         containers_count = int(args.containers)
         print(f"Create containers: {containers_count}")
         with ProcessPoolExecutor(max_workers=min(MAX_WORKERS, workers)) as executor:
-            containers_runs = [executor.submit(create_container, endpoint, args.policy, wallet, wallet_config)
-                               for _, endpoint in 
-                               zip(range(containers_count), cycle(endpoints))]
+            containers_runs = [executor.submit(create_container, endpoint, policy, container_creation_retry, wallet, wallet_config, args.acl, args.local)
+                               for _, endpoint, policy in 
+                               zip(range(containers_count), cycle(endpoints), cycle(args.policy))]
 
         for run in containers_runs:
             container_id = run.result()
@@ -75,7 +80,7 @@ def main():
         print("Create containers: Completed")
 
     print(f" > Containers: {containers}")
-    if containers_count == 0 or len(containers) != containers_count:
+    if containers_count > 0 and len(containers) != containers_count:
         print(f"Containers mismatch in preset: expected {containers_count}, created {len(containers)}")
         if not ignore_errors:
             sys.exit(ERROR_WRONG_CONTAINERS_COUNT)
@@ -98,7 +103,7 @@ def main():
                         
     for run in objects_runs:
         result = run.result()
-        if run.result:
+        if result:
             container_id = result[0]
             endpoint = result[1]
             object_id = result[2]

scenarios/preset/preset_s3.py

@@ -11,6 +11,12 @@ from concurrent.futures import ProcessPoolExecutor
 from helpers.cmd import random_payload
 from helpers.aws_cli import create_bucket, upload_object
 
+ERROR_WRONG_CONTAINERS_COUNT = 1
+ERROR_WRONG_OBJECTS_COUNT = 2
+ERROR_WRONG_PERCENTAGE = 3
+MAX_WORKERS = 50
+DEFAULT_LOCATION = ""
+
 parser = argparse.ArgumentParser()
 
 parser.add_argument('--size', help='Upload objects size in kb.')
@@ -20,26 +26,28 @@ parser.add_argument('--preload_obj', help='Number of pre-loaded objects.')
 parser.add_argument('--endpoint', help='S3 Gateways addresses separated by comma.')
 parser.add_argument('--update', help='True/False, False by default. Save existed buckets from target file (--out). '
                                      'New buckets will not be created.')
-parser.add_argument('--location', help='AWS location. Will be empty, if has not be declared.', default="")
-parser.add_argument('--versioning', help='True/False, False by default.')
-parser.add_argument('--ignore-errors', help='Ignore preset errors')
+parser.add_argument('--location', help=f'AWS location constraint. Default is "{DEFAULT_LOCATION}"', action="append")
+parser.add_argument('--versioning', help='True/False, False by default. Alias of --buckets_versioned=100')
+parser.add_argument('--buckets_versioned', help='Percent of versioned buckets. Default is 0', default=0)
+parser.add_argument('--ignore-errors', help='Ignore preset errors', action='store_true')
+parser.add_argument('--no-verify-ssl', help='Ignore SSL verifications', action='store_true')
 parser.add_argument('--workers', help='Count of workers in preset. Max = 50, Default = 50', default=50)
-parser.add_argument('--sleep', help='Time to sleep between container creation and object PUT (in seconds), '
+parser.add_argument('--sleep', help='Time to sleep between buckets creation and objects upload (in seconds), '
                                     'Default = 8', default=8)
+parser.add_argument('--acl', help='Bucket ACL. Default is private. Expected values are: private, public-read or public-read-write.', default="private")
 
 args = parser.parse_args()
 print(args)
 
-ERROR_WRONG_CONTAINERS_COUNT = 1
-ERROR_WRONG_OBJECTS_COUNT = 2
-MAX_WORKERS = 50
-
 def main():
     buckets = []
     objects_list = []
-    ignore_errors = True if args.ignore_errors else False
+    ignore_errors = args.ignore_errors
+    no_verify_ssl = args.no_verify_ssl
 
     endpoints = args.endpoint.split(',')
+    if not args.location:
+        args.location = [DEFAULT_LOCATION]
 
     workers = int(args.workers)
     objects_per_bucket = int(args.preload_obj)
@@ -56,9 +64,18 @@ def main():
         print(f"Create buckets: {buckets_count}")
 
         with ProcessPoolExecutor(max_workers=min(MAX_WORKERS, workers)) as executor:
-            buckets_runs = [executor.submit(create_bucket, endpoint, args.versioning, args.location) 
-                            for _, endpoint in 
-                            zip(range(buckets_count), cycle(endpoints))]
+            if not 0 <= int(args.buckets_versioned) <= 100:
+                print(f"Percent of versioned buckets must be between 0 and 100: got {args.buckets_versioned}")
+                if not ignore_errors:
+                    sys.exit(ERROR_WRONG_PERCENTAGE)
+            if args.versioning == "True":
+                versioning_per_bucket = [True] * buckets_count
+            else:
+                num_versioned_buckets = int((int(args.buckets_versioned) / 100) * buckets_count)
+                versioning_per_bucket = [True] * num_versioned_buckets + [False] * (buckets_count -  num_versioned_buckets)
+            buckets_runs = [executor.submit(create_bucket, endpoint, versioning_per_bucket[i], location, args.acl, no_verify_ssl)
+                            for i, endpoint, location in
+                            zip(range(buckets_count), cycle(endpoints), cycle(args.location))]
 
         for run in buckets_runs:
             bucket_name = run.result()
@@ -68,7 +85,7 @@ def main():
         print("Create buckets: Completed")
 
     print(f" > Buckets: {buckets}")
-    if buckets_count == 0 or len(buckets) != buckets_count:
+    if buckets_count > 0 and len(buckets) != buckets_count:
         print(f"Buckets mismatch in preset: expected {buckets_count}, created {len(buckets)}")
         if not ignore_errors:
             sys.exit(ERROR_WRONG_CONTAINERS_COUNT)
@@ -85,13 +102,13 @@ def main():
     total_objects = objects_per_bucket * buckets_count
 
     with ProcessPoolExecutor(max_workers=min(MAX_WORKERS, workers)) as executor:
-        objects_runs = [executor.submit(upload_object, bucket, payload_file.name, endpoint)
+        objects_runs = [executor.submit(upload_object, bucket, payload_file.name, endpoint, no_verify_ssl)
                         for _, bucket, endpoint in 
                         zip(range(total_objects), cycle(buckets), cycle(endpoints))]
                         
     for run in objects_runs:
         result = run.result()
-        if run.result:
+        if result:
             bucket = result[0]
             endpoint = result[1]
             object_id = result[2]

scenarios/preset/resolve_containers_in_preset.py

@@ -2,7 +2,8 @@
 
 import argparse
 import json
-import requests
+import http.client
+import ssl
 
 parser = argparse.ArgumentParser()
 parser.add_argument('--endpoint', help='Endpoint of the S3 gateway')
@@ -16,10 +17,13 @@ def main():
 
     preset = json.loads(preset_text)
 
+    conn = http.client.HTTPSConnection(args.endpoint, context = ssl._create_unverified_context())
     containers = []
     for bucket in preset.get('buckets'):
-        resp = requests.head(f'{args.endpoint}/{bucket}', verify=False)
-        containers.append(resp.headers['X-Container-Id'])
+        conn.request("HEAD", f'/{bucket}')
+        response = conn.getresponse()
+        containers.append(response.getheader('X-Container-Id'))
+        response.read()
 
     preset['containers'] = containers
     with open(args.preset_file, 'w+') as f:

scenarios/run_scenarios.md

@@ -19,6 +19,8 @@ Scenarios `grpc.js`, `local.js`, `http.js` and `s3.js` support the following opt
   * `SLEEP_READ` - time interval (in seconds) between reading VU iterations.
   * `SELECTION_SIZE` - size of batch to select for deletion (default: 1000).
   * `PAYLOAD_TYPE` - type of an object payload ("random" or "text", default: "random").
+  * `STREAMING` - if set, the payload is generated on the fly and is not read into memory fully.
+  * `METRIC_TAGS` - custom metrics tags (format `tag1:value1;tag2:value2`).
 
 Additionally, the profiling extension can be enabled to generate CPU and memory profiles which can be inspected with `go tool pprof file.prof`:
 ```shell
@@ -69,13 +71,15 @@ $ ./scenarios/preset/preset_grpc.py --size 1024 --containers 1 --out grpc.json -
 2. Execute scenario with options:
 
 ```shell
-$ ./k6 run -e DURATION=60 -e WRITE_OBJ_SIZE=8192 -e READERS=20 -e WRITERS=20 -e DELETERS=30 -e DELETE_AGE=10 -e REGISTRY_FILE=registry.bolt -e CONFIG_FILE=/path/to/config.yaml -e PREGEN_JSON=./grpc.json scenarios/local.js
+$ ./k6 run -e DURATION=60 -e WRITE_OBJ_SIZE=8192 -e READERS=20 -e WRITERS=20 -e DELETERS=30 -e DELETE_AGE=10 -e REGISTRY_FILE=registry.bolt -e CONFIG_FILE=/path/to/config.yaml -e CONFIG_DIR=/path/to/dir/ -e PREGEN_JSON=./grpc.json scenarios/local.js
 ```
 
 Options (in addition to the common options):
   * `CONFIG_FILE` - path to the local configuration file used for the storage node. Only the storage configuration section is used.
+  * `CONFIG_DIR` - path to the folder with local configuration files used for the storage node.
   * `DELETERS` - number of VUs performing delete operations (using deleters requires that options `DELETE_AGE` and `REGISTRY_FILE` are specified as well).
   * `DELETE_AGE` - age of object in seconds before which it can not be deleted. This parameter can be used to control how many objects we have in the system under load.
+  * `MAX_TOTAL_SIZE_GB` - if specified, max payload size in GB of the storage engine. If the storage engine is already full, no new objects will be saved.
 
 ## HTTP
 
@@ -121,7 +125,7 @@ The tests will use all pre-created buckets for PUT operations and all pre-create
 $ ./scenarios/preset/preset_s3.py --size 1024 --buckets 1 --out s3_1024kb.json --endpoint host1:8084 --preload_obj 500 --location load-1-4
 ```
   * '--location' - specify the name of container policy (from policy.json file). It's important to run 'aws configure' each time when the policy file has been changed to pick up the latest policies.
-
+  * '--buckets_versioned' - specify the percentage of versioned buckets from the total number of created buckets. Default is 0
 3. Execute scenario with options:
 
 ```shell
@@ -134,6 +138,33 @@ Options (in addition to the common options):
   * `DELETE_AGE` - age of object in seconds before which it can not be deleted. This parameter can be used to control how many objects we have in the system under load.
   * `SLEEP_DELETE` - time interval (in seconds) between deleting VU iterations.
   * `OBJ_NAME` - if specified, this name will be used for all write operations instead of random generation.
+  * `OBJ_NAME_LENGTH` - if specified, then name of the object will be generated with the specified length of ASCII characters.
+  * `DIR_HEIGHT`, `DIR_WIDTH` - if both specified, object name will consist of `DIR_HEIGHT` directories, each of which can have `DIR_WIDTH` subdirectories, for example for `DIR_HEIGHT = 3, DIR_WIDTH = 100`, object names will be `/dir{1...100}/dir{1...100}/dir{1...100}/{uuid || OBJ_NAME}`
+
+## S3 Multipart
+
+Perform multipart upload operation, break up large objects, so they can be transferred in multiple parts, in parallel
+
+```shell
+$ ./k6 run -e DURATION=600 \
+-e WRITERS=400 -e WRITERS_MULTIPART=10 \
+-e WRITE_OBJ_SIZE=524288 -e WRITE_OBJ_PART_SIZE=10240  \
+-e S3_ENDPOINTS=10.78.70.142:8084,10.78.70.143:8084,10.78.70.144:8084,10.78.70.145:8084 \
+-e PREGEN_JSON=/home/service/s3_4kb.json \
+scenarios/s3_multipart.js
+```
+
+Options:
+* `DURATION` - duration of scenario in seconds.
+* `REGISTRY_FILE` - if set, all produced objects will be stored in database for subsequent verification. Database file name will be set to the value of `REGISTRY_FILE`.
+* `PREGEN_JSON` - path to json file with pre-generated containers.
+* `SLEEP_WRITE` - time interval (in seconds) between writing VU iterations.
+* `PAYLOAD_TYPE` - type of an object payload ("random" or "text", default: "random").
+* `S3_ENDPOINTS` - - endpoints of S3 gateways in format `host:port`. To specify multiple endpoints separate them by comma.
+* `WRITERS` - number of VUs performing upload payload operation
+* `WRITERS_MULTIPART` - number of goroutines that will upload parts in parallel
+* `WRITE_OBJ_SIZE` - object size in kb for write(PUT) operations.
+* `WRITE_OBJ_PART_SIZE` - part size in kb for multipart upload operations (must be greater or equal 5mb).
 
 ## S3 Local
 
@@ -150,13 +181,50 @@ After this, the `pregen.json` file will contain a `containers` list field the sa
 
 3. Execute the scenario with the desired options. For example:
 ```shell
-$ ./k6 run -e DURATION=60 -e WRITE_OBJ_SIZE=8192 -e READERS=20 -e WRITERS=20 -e CONFIG_FILE=/path/to/node/config.yml -e PREGEN_JSON=pregen.json scenarios/s3local.js
+$ ./k6 run -e DURATION=60 -e WRITE_OBJ_SIZE=8192 -e READERS=20 -e WRITERS=20 -e CONFIG_FILE=/path/to/node/config.yml -e CONFIG_DIR=/path/to/dir/ -e PREGEN_JSON=pregen.json scenarios/s3local.js
 ```
 
 Note that the `s3local` scenario currently does not support deleters.
 
 Options (in addition to the common options):
   * `OBJ_NAME` - if specified, this name will be used for all write operations instead of random generation.
+  * `MAX_TOTAL_SIZE_GB` - if specified, max payload size in GB of the storage engine. If the storage engine is already full, no new objects will be saved.
+
+## Export metrics
+
+To export metrics to Prometheus (also Grafana and Victoria Metrics support Prometheus format), you need to run `k6` with an option `-o experimental-prometheus-rw` and
+an environment variable `K6_PROMETHEUS_RW_SERVER_URL` whose value corresponds to the URL for the remote write endpoint.
+To specify percentiles for trend metrics, use an environment variable `K6_PROMETHEUS_RW_TREND_STATS`.
+See [k6 docs](https://k6.io/docs/results-output/real-time/prometheus-remote-write/) for a list of all possible options.
+To distinct metrics from different loaders, use an option `METRIC_TAGS`. These tags does not apply to builtin `k6` metrics.
+
+Example:
+```bash
+K6_PROMETHEUS_RW_SERVER_URL=http://host:8428/api/v1/write \
+K6_PROMETHEUS_RW_TREND_STATS="p(95),p(99),min,max" \
+./k6 run ... -o experimental-prometheus-rw -e METRIC_TAGS="instance:server1;run:run1" scenario.js
+```
+
+## Grafana annotations
+
+There is no option to export Grafana annotaions, but it can be easily done with `curl` and Grafana's annotations API.
+Example:
+```shell
+curl --request POST \
+  --url https://user:password@grafana.host/api/annotations \
+  --header 'Content-Type: application/json' \
+  --data '{
+    "dashboardUID": "YsVWNpMIk",
+    "time": 1706533045014,
+    "timeEnd": 1706533085100,
+    "tags": [
+        "tag1",
+        "tag2"
+    ],
+    "text": "Test annotation"
+}'
+```
+See [Grafana docs](https://grafana.com/docs/grafana/latest/developers/http_api/annotations/) for details.
 
 ## Verify
 

scenarios/s3.js

@@ -1,170 +1,226 @@
-import datagen from 'k6/x/frostfs/datagen';
+import {sleep} from 'k6';
+import {SharedArray} from 'k6/data';
+import exec from 'k6/execution';
 import logging from 'k6/x/frostfs/logging';
 import registry from 'k6/x/frostfs/registry';
 import s3 from 'k6/x/frostfs/s3';
-import { SharedArray } from 'k6/data';
-import { sleep } from 'k6';
-import { textSummary } from './libs/k6-summary-0.0.2.js';
-import { parseEnv } from './libs/env-parser.js';
-import { uuidv4 } from './libs/k6-utils-1.4.0.js';
+import stats from 'k6/x/frostfs/stats';
+
+import {generateS3Key} from './libs/keygen.js';
+import {parseEnv} from './libs/env-parser.js';
+import {textSummary} from './libs/k6-summary-0.0.2.js';
+import {newGenerator} from './libs/datagen.js';
 
 parseEnv();
 
-const obj_list = new SharedArray('obj_list', function () {
-    return JSON.parse(open(__ENV.PREGEN_JSON)).objects;
-});
+const obj_list = new SharedArray(
+    'obj_list',
+    function() { return JSON.parse(open(__ENV.PREGEN_JSON)).objects; });
 
-const bucket_list = new SharedArray('bucket_list', function () {
-    return JSON.parse(open(__ENV.PREGEN_JSON)).buckets;
-});
+const bucket_list = new SharedArray(
+    'bucket_list',
+    function() { return JSON.parse(open(__ENV.PREGEN_JSON)).buckets; });
 
 const read_size = JSON.parse(open(__ENV.PREGEN_JSON)).obj_size;
-const summary_json = __ENV.SUMMARY_JSON || "/tmp/summary.json";
+const summary_json = __ENV.SUMMARY_JSON || '/tmp/summary.json';
 
+const no_verify_ssl = __ENV.NO_VERIFY_SSL || 'true';
+const connection_args = {
+  no_verify_ssl : no_verify_ssl
+}
 // Select random S3 endpoint for current VU
 const s3_endpoints = __ENV.S3_ENDPOINTS.split(',');
-const s3_endpoint = s3_endpoints[Math.floor(Math.random() * s3_endpoints.length)];
-const s3_client = s3.connect(`http://${s3_endpoint}`);
-const log = logging.new().withField("endpoint", s3_endpoint);
+const s3_endpoint =
+    s3_endpoints[Math.floor(Math.random() * s3_endpoints.length)];
+const s3_client = s3.connect(s3_endpoint, connection_args);
+const log = logging.new().withField('endpoint', s3_endpoint);
 
 const registry_enabled = !!__ENV.REGISTRY_FILE;
-const obj_registry = registry_enabled ? registry.open(__ENV.REGISTRY_FILE) : undefined;
+const obj_registry =
+    registry_enabled ? registry.open(__ENV.REGISTRY_FILE) : undefined;
 
 const duration = __ENV.DURATION;
 
-const delete_age = __ENV.DELETE_AGE ? parseInt(__ENV.DELETE_AGE) : undefined;
-let obj_to_delete_selector = undefined;
-if (registry_enabled && delete_age) {
-    obj_to_delete_selector = registry.getSelector(
-        __ENV.REGISTRY_FILE,
-        "obj_to_delete",
-        __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0,
-        {
-            status: "created",
-            age: delete_age,
-        }
-    );
+if (!!__ENV.METRIC_TAGS) {
+  stats.setTags(__ENV.METRIC_TAGS)
 }
 
-const generator = datagen.generator(1024 * parseInt(__ENV.WRITE_OBJ_SIZE), __ENV.PAYLOAD_TYPE || "");
+const read_age = __ENV.READ_AGE ? parseInt(__ENV.READ_AGE) : 10;
+let obj_to_read_selector = undefined;
+if (registry_enabled) {
+  obj_to_read_selector = registry.getLoopedSelector(
+      __ENV.REGISTRY_FILE, 'obj_to_read',
+      __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0, {
+        status : 'created',
+        age : read_age,
+      })
+}
 
 const scenarios = {};
 
 const write_vu_count = parseInt(__ENV.WRITERS || '0');
+const generator = newGenerator(write_vu_count > 0);
 if (write_vu_count > 0) {
-    scenarios.write = {
-        executor: 'constant-vus',
-        vus: write_vu_count,
-        duration: `${duration}s`,
-        exec: 'obj_write', 
-        gracefulStop: '5s',
-    };
+  scenarios.write = {
+    executor : 'constant-vus',
+    vus : write_vu_count,
+    duration : `${duration}s`,
+    exec : 'obj_write',
+    gracefulStop : '5s',
+  };
+}
+
+const delete_age = __ENV.DELETE_AGE ? parseInt(__ENV.DELETE_AGE) : undefined;
+let obj_to_delete_selector = undefined;
+let obj_to_delete_exit_on_null = undefined;
+if (registry_enabled && delete_age) {
+  obj_to_delete_exit_on_null = write_vu_count == 0;
+
+  let constructor = obj_to_delete_exit_on_null ? registry.getOneshotSelector
+                                               : registry.getSelector;
+
+  obj_to_delete_selector =
+      constructor(__ENV.REGISTRY_FILE, 'obj_to_delete',
+                  __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0, {
+                    status : 'created',
+                    age : delete_age,
+                  });
 }
 
 const read_vu_count = parseInt(__ENV.READERS || '0');
 if (read_vu_count > 0) {
-    scenarios.read = {
-        executor: 'constant-vus',
-        vus: read_vu_count,
-        duration: `${duration}s`,
-        exec: 'obj_read', 
-        gracefulStop: '5s',
-    };
+  scenarios.read = {
+    executor : 'constant-vus',
+    vus : read_vu_count,
+    duration : `${duration}s`,
+    exec : 'obj_read',
+    gracefulStop : '5s',
+  };
 }
 
 const delete_vu_count = parseInt(__ENV.DELETERS || '0');
 if (delete_vu_count > 0) {
-    if (!obj_to_delete_selector) {
-        throw 'Positive DELETE worker number without a proper object selector';
-    }
+  if (!obj_to_delete_selector) {
+    throw 'Positive DELETE worker number without a proper object selector';
+  }
 
-    scenarios.delete = {
-        executor: 'constant-vus',
-        vus: delete_vu_count,
-        duration: `${duration}s`,
-        exec: 'obj_delete', 
-        gracefulStop: '5s',
-    };
+  scenarios.delete = {
+    executor : 'constant-vus',
+    vus : delete_vu_count,
+    duration : `${duration}s`,
+    exec : 'obj_delete',
+    gracefulStop : '5s',
+  };
 }
 
 export const options = {
-    scenarios,
-    setupTimeout: '5s',
+  scenarios,
+  setupTimeout : '5s',
 };
 
 export function setup() {
-    const total_vu_count = write_vu_count + read_vu_count + delete_vu_count;
+  const total_vu_count = write_vu_count + read_vu_count + delete_vu_count;
 
-    console.log(`Pregenerated buckets:          ${bucket_list.length}`);
-    console.log(`Pregenerated read object size: ${read_size}`);
-    console.log(`Pregenerated total objects:    ${obj_list.length}`);
-    console.log(`Reading VUs:                   ${read_vu_count}`);
-    console.log(`Writing VUs:                   ${write_vu_count}`);
-    console.log(`Deleting VUs:                  ${delete_vu_count}`);
-    console.log(`Total VUs:                     ${total_vu_count}`);
+  console.log(`Pregenerated buckets:          ${bucket_list.length}`);
+  console.log(`Pregenerated read object size: ${read_size}`);
+  console.log(`Pregenerated total objects:    ${obj_list.length}`);
+  console.log(`Reading VUs:                   ${read_vu_count}`);
+  console.log(`Writing VUs:                   ${write_vu_count}`);
+  console.log(`Deleting VUs:                  ${delete_vu_count}`);
+  console.log(`Total VUs:                     ${total_vu_count}`);
+
+  const start_timestamp = Date.now()
+  console.log(
+      `Load started at:               ${Date(start_timestamp).toString()}`)
+
+  if (delete_vu_count > 0){
+    obj_to_delete_selector.sync.add(delete_vu_count)
+  }
 }
 
 export function teardown(data) {
-    if (obj_registry) {
-        obj_registry.close();
-    }
+  if (obj_registry) {
+    obj_registry.close();
+  }
+  const end_timestamp = Date.now()
+  console.log(
+      `Load finished at:              ${Date(end_timestamp).toString()}`)
 }
 
 export function handleSummary(data) {
-    return {
-        'stdout': textSummary(data, { indent: ' ', enableColors: false }),
-        [summary_json]: JSON.stringify(data),
-    };
-  }
+  return {
+    'stdout' : textSummary(data, {indent : ' ', enableColors : false}),
+    [summary_json] : JSON.stringify(data),
+  };
+}
 
 export function obj_write() {
-    if (__ENV.SLEEP_WRITE) {
-        sleep(__ENV.SLEEP_WRITE);
-    }
+  if (__ENV.SLEEP_WRITE) {
+    sleep(__ENV.SLEEP_WRITE);
+  }
 
-    const key = __ENV.OBJ_NAME || uuidv4();
-    const bucket = bucket_list[Math.floor(Math.random() * bucket_list.length)];
+  const key = generateS3Key();
+  const bucket = bucket_list[Math.floor(Math.random() * bucket_list.length)];
 
-    const { payload, hash } = generator.genPayload(registry_enabled);
-    const resp = s3_client.put(bucket, key, payload);
-    if (!resp.success) {
-        log.withFields({bucket: bucket, key: key}).error(resp.error);
-        return;
-    }
+  const payload = generator.genPayload();
+  const resp = s3_client.put(bucket, key, payload);
+  if (!resp.success) {
+    log.withFields({bucket : bucket, key : key}).error(resp.error);
+    return;
+  }
 
-    if (obj_registry) {
-        obj_registry.addObject("", "", bucket, key, hash);
-    }
+  if (obj_registry) {
+    obj_registry.addObject('', '', bucket, key, payload.hash());
+  }
 }
 
 export function obj_read() {
-    if (__ENV.SLEEP_READ) {
-        sleep(__ENV.SLEEP_READ);
+  if (__ENV.SLEEP_READ) {
+    sleep(__ENV.SLEEP_READ);
+  }
+
+  if (obj_to_read_selector) {
+    const obj = obj_to_read_selector.nextObject();
+    if (!obj) {
+      return;
     }
-
-    const obj = obj_list[Math.floor(Math.random() * obj_list.length)];
-
-    const resp = s3_client.get(obj.bucket, obj.object);
+    const resp = s3_client.get(obj.s3_bucket, obj.s3_key)
     if (!resp.success) {
-        log.withFields({bucket: obj.bucket, key: obj.object}).error(resp.error);
-    } 
+      log.withFields({bucket : obj.s3_bucket, key : obj.s3_key})
+          .error(resp.error);
+    }
+    return
+  }
+
+  const obj = obj_list[Math.floor(Math.random() * obj_list.length)];
+
+  const resp = s3_client.get(obj.bucket, obj.object);
+  if (!resp.success) {
+    log.withFields({bucket : obj.bucket, key : obj.object}).error(resp.error);
+  }
 }
 
 export function obj_delete() {
-    if (__ENV.SLEEP_DELETE) {
-        sleep(__ENV.SLEEP_DELETE);
-    }
+  if (__ENV.SLEEP_DELETE) {
+    sleep(__ENV.SLEEP_DELETE);
+  }
 
-    const obj = obj_to_delete_selector.nextObject();
-    if (!obj) {
-        return;
+  const obj = obj_to_delete_selector.nextObject();
+  if (!obj) {
+    if (obj_to_delete_exit_on_null) {
+      obj_to_delete_selector.sync.done()
+      obj_to_delete_selector.sync.wait()
+      exec.test.abort("No more objects to select");
     }
+    return;
+  }
 
-    const resp = s3_client.delete(obj.s3_bucket, obj.s3_key);
-    if (!resp.success) {
-        log.withFields({bucket: obj.s3_bucket, key: obj.s3_key, op: "DELETE"}).error(resp.error);
-        return;
-    }
+  const resp = s3_client.delete(obj.s3_bucket, obj.s3_key);
+  if (!resp.success) {
+    log.withFields({bucket : obj.s3_bucket, key : obj.s3_key, op : 'DELETE'})
+        .error(resp.error);
+    return;
+  }
 
-    obj_registry.deleteObject(obj.id);
+  obj_registry.deleteObject(obj.id);
 }

scenarios/s3_car.js

@@ -1,52 +1,70 @@
-import datagen from 'k6/x/frostfs/datagen';
+import {sleep} from 'k6';
+import {SharedArray} from 'k6/data';
 import logging from 'k6/x/frostfs/logging';
 import registry from 'k6/x/frostfs/registry';
 import s3 from 'k6/x/frostfs/s3';
-import { SharedArray } from 'k6/data';
-import { sleep } from 'k6';
-import { textSummary } from './libs/k6-summary-0.0.2.js';
-import { parseEnv } from './libs/env-parser.js';
-import { uuidv4 } from './libs/k6-utils-1.4.0.js';
+import stats from 'k6/x/frostfs/stats';
+
+import {generateS3Key} from './libs/keygen.js';
+import {newGenerator} from './libs/datagen.js';
+import {parseEnv} from './libs/env-parser.js';
+import {textSummary} from './libs/k6-summary-0.0.2.js';
 
 parseEnv();
 
-const obj_list = new SharedArray('obj_list', function () {
-    return JSON.parse(open(__ENV.PREGEN_JSON)).objects;
+const obj_list = new SharedArray('obj_list', function() {
+  return JSON.parse(open(__ENV.PREGEN_JSON)).objects;
 });
 
-const bucket_list = new SharedArray('bucket_list', function () {
-    return JSON.parse(open(__ENV.PREGEN_JSON)).buckets;
+const bucket_list = new SharedArray('bucket_list', function() {
+  return JSON.parse(open(__ENV.PREGEN_JSON)).buckets;
 });
 
 const read_size = JSON.parse(open(__ENV.PREGEN_JSON)).obj_size;
-const summary_json = __ENV.SUMMARY_JSON || "/tmp/summary.json";
+const summary_json = __ENV.SUMMARY_JSON || '/tmp/summary.json';
 
 // Select random S3 endpoint for current VU
 const s3_endpoints = __ENV.S3_ENDPOINTS.split(',');
-const s3_endpoint = s3_endpoints[Math.floor(Math.random() * s3_endpoints.length)];
-const s3_client = s3.connect(`http://${s3_endpoint}`);
-const log = logging.new().withField("endpoint", s3_endpoint);
+const s3_endpoint =
+    s3_endpoints[Math.floor(Math.random() * s3_endpoints.length)];
+const no_verify_ssl = __ENV.NO_VERIFY_SSL || 'true';
+const connection_args = {
+  no_verify_ssl: no_verify_ssl
+};
+const s3_client = s3.connect(s3_endpoint, connection_args);
+const log = logging.new().withField('endpoint', s3_endpoint);
 
 const registry_enabled = !!__ENV.REGISTRY_FILE;
-const obj_registry = registry_enabled ? registry.open(__ENV.REGISTRY_FILE) : undefined;
+const obj_registry =
+    registry_enabled ? registry.open(__ENV.REGISTRY_FILE) : undefined;
 
 const duration = __ENV.DURATION;
 
+if (!!__ENV.METRIC_TAGS) {
+  stats.setTags(__ENV.METRIC_TAGS)
+}
+
 const delete_age = __ENV.DELETE_AGE ? parseInt(__ENV.DELETE_AGE) : undefined;
 let obj_to_delete_selector = undefined;
 if (registry_enabled && delete_age) {
-    obj_to_delete_selector = registry.getSelector(
-        __ENV.REGISTRY_FILE,
-        "obj_to_delete",
-        __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0,
-        {
-            status: "created",
-            age: delete_age,
-        }
-    );
+  obj_to_delete_selector = registry.getSelector(
+      __ENV.REGISTRY_FILE, 'obj_to_delete',
+      __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0, {
+        status: 'created',
+        age: delete_age,
+      });
 }
 
-const generator = datagen.generator(1024 * parseInt(__ENV.WRITE_OBJ_SIZE));
+const read_age = __ENV.READ_AGE ? parseInt(__ENV.READ_AGE) : 10;
+let obj_to_read_selector = undefined;
+if (registry_enabled) {
+  obj_to_read_selector = registry.getLoopedSelector(
+      __ENV.REGISTRY_FILE, 'obj_to_read',
+      __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0, {
+        status: 'created',
+        age: read_age,
+      })
+}
 
 const scenarios = {};
 
@@ -54,17 +72,18 @@ const time_unit = __ENV.TIME_UNIT || '1s';
 const pre_alloc_write_vus = parseInt(__ENV.PRE_ALLOC_WRITERS || '0');
 const max_write_vus = parseInt(__ENV.MAX_WRITERS || pre_alloc_write_vus);
 const write_rate = parseInt(__ENV.WRITE_RATE || '0');
+const generator = newGenerator(write_rate > 0);
 if (write_rate > 0) {
-    scenarios.write = {
-        executor: 'constant-arrival-rate',
-        duration: `${duration}s`,
-        preAllocatedVUs: pre_alloc_write_vus,
-        maxVUs: max_write_vus,
-        rate: write_rate,
-        timeUnit: time_unit,
-        exec: 'obj_write', 
-        gracefulStop: '5s',
-    };
+  scenarios.write = {
+    executor: 'constant-arrival-rate',
+    duration: `${duration}s`,
+    preAllocatedVUs: pre_alloc_write_vus,
+    maxVUs: max_write_vus,
+    rate: write_rate,
+    timeUnit: time_unit,
+    exec: 'obj_write',
+    gracefulStop: '5s',
+  };
 }
 
 
@@ -72,16 +91,16 @@ const pre_alloc_read_vus = parseInt(__ENV.PRE_ALLOC_READERS || '0');
 const max_read_vus = parseInt(__ENV.MAX_READERS || pre_alloc_read_vus);
 const read_rate = parseInt(__ENV.READ_RATE || '0');
 if (read_rate > 0) {
-    scenarios.read = {
-        executor: 'constant-arrival-rate',
-        duration: `${duration}s`,
-        preAllocatedVUs: pre_alloc_write_vus,
-        maxVUs: max_read_vus,
-        rate: read_rate,
-        timeUnit: time_unit,
-        exec: 'obj_read', 
-        gracefulStop: '5s',
-    };
+  scenarios.read = {
+    executor: 'constant-arrival-rate',
+    duration: `${duration}s`,
+    preAllocatedVUs: pre_alloc_write_vus,
+    maxVUs: max_read_vus,
+    rate: read_rate,
+    timeUnit: time_unit,
+    exec: 'obj_read',
+    gracefulStop: '5s',
+  };
 }
 
 
@@ -89,109 +108,132 @@ const pre_alloc_delete_vus = parseInt(__ENV.PRE_ALLOC_DELETERS || '0');
 const max_delete_vus = parseInt(__ENV.MAX_DELETERS || pre_alloc_write_vus);
 const delete_rate = parseInt(__ENV.DELETE_RATE || '0');
 if (delete_rate > 0) {
-    if (!obj_to_delete_selector) {
-        throw new Error('Positive DELETE worker number without a proper object selector');
-    }
+  if (!obj_to_delete_selector) {
+    throw new Error(
+        'Positive DELETE worker number without a proper object selector');
+  }
 
-    scenarios.delete = {
-        executor: 'constant-arrival-rate',
-        duration: `${duration}s`,
-        preAllocatedVUs: pre_alloc_delete_vus,
-        maxVUs: max_delete_vus,
-        rate: delete_rate,
-        timeUnit: time_unit,
-        exec: 'obj_delete', 
-        gracefulStop: '5s',
-    };
+  scenarios.delete = {
+    executor: 'constant-arrival-rate',
+    duration: `${duration}s`,
+    preAllocatedVUs: pre_alloc_delete_vus,
+    maxVUs: max_delete_vus,
+    rate: delete_rate,
+    timeUnit: time_unit,
+    exec: 'obj_delete',
+    gracefulStop: '5s',
+  };
 }
 
 export const options = {
-    scenarios,
-    setupTimeout: '5s',
+  scenarios,
+  setupTimeout: '5s',
 };
 
 export function setup() {
-    const total_pre_allocated_vu_count = pre_alloc_write_vus + pre_alloc_read_vus + pre_alloc_delete_vus;
-    const total_max_vu_count = max_read_vus + max_write_vus + max_delete_vus
+  const total_pre_allocated_vu_count =
+      pre_alloc_write_vus + pre_alloc_read_vus + pre_alloc_delete_vus;
+  const total_max_vu_count = max_read_vus + max_write_vus + max_delete_vus
 
-    console.log(`Pregenerated buckets:          ${bucket_list.length}`);
-    console.log(`Pregenerated read object size: ${read_size}`);
-    console.log(`Pregenerated total objects:    ${obj_list.length}`);
-    console.log(`Pre allocated reading VUs:     ${pre_alloc_read_vus}`);
-    console.log(`Pre allocated writing VUs:     ${pre_alloc_write_vus}`);
-    console.log(`Pre allocated deleting VUs:    ${pre_alloc_delete_vus}`);
-    console.log(`Total pre allocated VUs:       ${total_pre_allocated_vu_count}`);
-    console.log(`Max reading VUs:               ${max_read_vus}`);
-    console.log(`Max writing VUs:               ${max_write_vus}`);
-    console.log(`Max deleting VUs:              ${max_delete_vus}`);
-    console.log(`Total max VUs:                 ${total_max_vu_count}`);
-    console.log(`Time unit:                     ${time_unit}`);
-    console.log(`Read rate:                     ${read_rate}`);
-    console.log(`Writing rate:                  ${write_rate}`);
-    console.log(`Delete rate:                   ${delete_rate}`);
+  console.log(`Pregenerated buckets:          ${bucket_list.length}`);
+  console.log(`Pregenerated read object size: ${read_size}`);
+  console.log(`Pregenerated total objects:    ${obj_list.length}`);
+  console.log(`Pre allocated reading VUs:     ${pre_alloc_read_vus}`);
+  console.log(`Pre allocated writing VUs:     ${pre_alloc_write_vus}`);
+  console.log(`Pre allocated deleting VUs:    ${pre_alloc_delete_vus}`);
+  console.log(`Total pre allocated VUs:       ${total_pre_allocated_vu_count}`);
+  console.log(`Max reading VUs:               ${max_read_vus}`);
+  console.log(`Max writing VUs:               ${max_write_vus}`);
+  console.log(`Max deleting VUs:              ${max_delete_vus}`);
+  console.log(`Total max VUs:                 ${total_max_vu_count}`);
+  console.log(`Time unit:                     ${time_unit}`);
+  console.log(`Read rate:                     ${read_rate}`);
+  console.log(`Writing rate:                  ${write_rate}`);
+  console.log(`Delete rate:                   ${delete_rate}`);
+
+  const start_timestamp = Date.now()
+  console.log(
+      `Load started at:               ${Date(start_timestamp).toString()}`)
 }
 
 export function teardown(data) {
-    if (obj_registry) {
-        obj_registry.close();
-    }
+  if (obj_registry) {
+    obj_registry.close();
+  }
+  const end_timestamp = Date.now()
+  console.log(
+      `Load finished at:              ${Date(end_timestamp).toString()}`)
 }
 
 export function handleSummary(data) {
-    return {
-        'stdout': textSummary(data, { indent: ' ', enableColors: false }),
-        [summary_json]: JSON.stringify(data),
-    };
-  }
+  return {
+    'stdout': textSummary(data, {indent: ' ', enableColors: false}),
+    [summary_json]: JSON.stringify(data),
+  };
+}
 
 export function obj_write() {
-    if (__ENV.SLEEP_WRITE) {
-        sleep(__ENV.SLEEP_WRITE);
-    }
+  if (__ENV.SLEEP_WRITE) {
+    sleep(__ENV.SLEEP_WRITE);
+  }
 
-    const key = __ENV.OBJ_NAME || uuidv4();
-    const bucket = bucket_list[Math.floor(Math.random() * bucket_list.length)];
+  const key = generateS3Key();
+  const bucket = bucket_list[Math.floor(Math.random() * bucket_list.length)];
 
-    const { payload, hash } = generator.genPayload(registry_enabled);
-    const resp = s3_client.put(bucket, key, payload);
-    if (!resp.success) {
-        log.withFields({bucket: bucket, key: key}).error(resp.error);
-        return;
-    }
+  const payload = generator.genPayload();
+  const resp = s3_client.put(bucket, key, payload);
+  if (!resp.success) {
+    log.withFields({bucket: bucket, key: key}).error(resp.error);
+    return;
+  }
 
-    if (obj_registry) {
-        obj_registry.addObject("", "", bucket, key, hash);
-    }
+  if (obj_registry) {
+    obj_registry.addObject('', '', bucket, key, payload.hash());
+  }
 }
 
 export function obj_read() {
-    if (__ENV.SLEEP_READ) {
-        sleep(__ENV.SLEEP_READ);
+  if (__ENV.SLEEP_READ) {
+    sleep(__ENV.SLEEP_READ);
+  }
+
+  if (obj_to_read_selector) {
+    const obj = obj_to_read_selector.nextObject();
+    if (!obj) {
+      return;
     }
-
-    const obj = obj_list[Math.floor(Math.random() * obj_list.length)];
-
-    const resp = s3_client.get(obj.bucket, obj.object);
+    const resp = s3_client.get(obj.s3_bucket, obj.s3_key)
     if (!resp.success) {
-        log.withFields({bucket: obj.bucket, key: obj.object}).error(resp.error);
-    } 
+      log.withFields({bucket: obj.s3_bucket, key: obj.s3_key})
+          .error(resp.error);
+    }
+    return
+  }
+
+  const obj = obj_list[Math.floor(Math.random() * obj_list.length)];
+
+  const resp = s3_client.get(obj.bucket, obj.object);
+  if (!resp.success) {
+    log.withFields({bucket: obj.bucket, key: obj.object}).error(resp.error);
+  }
 }
 
 export function obj_delete() {
-    if (__ENV.SLEEP_DELETE) {
-        sleep(__ENV.SLEEP_DELETE);
-    }
+  if (__ENV.SLEEP_DELETE) {
+    sleep(__ENV.SLEEP_DELETE);
+  }
 
-    const obj = obj_to_delete_selector.nextObject();
-    if (!obj) {
-        return;
-    }
+  const obj = obj_to_delete_selector.nextObject();
+  if (!obj) {
+    return;
+  }
 
-    const resp = s3_client.delete(obj.s3_bucket, obj.s3_key);
-    if (!resp.success) {
-        log.withFields({bucket: obj.s3_bucket, key: obj.s3_key, op: "DELETE"}).error(resp.error);
-        return;
-    }
+  const resp = s3_client.delete(obj.s3_bucket, obj.s3_key);
+  if (!resp.success) {
+    log.withFields({bucket: obj.s3_bucket, key: obj.s3_key, op: 'DELETE'})
+        .error(resp.error);
+    return;
+  }
 
-    obj_registry.deleteObject(obj.id);
+  obj_registry.deleteObject(obj.id);
 }

scenarios/s3_dar.js

@@ -0,0 +1,233 @@
+import {sleep} from 'k6';
+import {SharedArray} from 'k6/data';
+import exec from 'k6/execution';
+import logging from 'k6/x/frostfs/logging';
+import registry from 'k6/x/frostfs/registry';
+import s3 from 'k6/x/frostfs/s3';
+import stats from 'k6/x/frostfs/stats';
+
+import {generateS3Key} from './libs/keygen.js';
+import {newGenerator} from './libs/datagen.js';
+import {parseEnv} from './libs/env-parser.js';
+import {textSummary} from './libs/k6-summary-0.0.2.js';
+
+parseEnv();
+
+const obj_list = new SharedArray(
+    'obj_list',
+    function() { return JSON.parse(open(__ENV.PREGEN_JSON)).objects; });
+
+const bucket_list = new SharedArray(
+    'bucket_list',
+    function() { return JSON.parse(open(__ENV.PREGEN_JSON)).buckets; });
+
+const read_size = JSON.parse(open(__ENV.PREGEN_JSON)).obj_size;
+const summary_json = __ENV.SUMMARY_JSON || '/tmp/summary.json';
+
+const no_verify_ssl = __ENV.NO_VERIFY_SSL || 'true';
+
+const connection_args = {
+    no_verify_ssl : no_verify_ssl
+}
+// Select random S3 endpoint for current VU
+const s3_endpoints = __ENV.S3_ENDPOINTS.split(',');
+const s3_endpoint =
+    s3_endpoints[Math.floor(Math.random() * s3_endpoints.length)];
+const s3_client = s3.connect(s3_endpoint, connection_args);
+const log = logging.new().withField('endpoint', s3_endpoint);
+
+const registry_enabled = !!__ENV.REGISTRY_FILE;
+const obj_registry =
+    registry_enabled ? registry.open(__ENV.REGISTRY_FILE) : undefined;
+
+const duration = __ENV.DURATION;
+
+if (!!__ENV.METRIC_TAGS) {
+    stats.setTags(__ENV.METRIC_TAGS)
+}
+
+const read_age = __ENV.READ_AGE ? parseInt(__ENV.READ_AGE) : 10;
+let obj_to_read_selector = undefined;
+if (registry_enabled) {
+    obj_to_read_selector = registry.getSelector(
+        __ENV.REGISTRY_FILE, 'obj_to_read',
+        __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0, {
+            status : 'created',
+            age : read_age,
+        })
+}
+
+const scenarios = {};
+
+const write_vu_count = parseInt(__ENV.WRITERS || '0');
+const generator = newGenerator(write_vu_count > 0);
+if (write_vu_count > 0) {
+    scenarios.write = {
+        executor : 'constant-vus',
+        vus : write_vu_count,
+        duration : `${duration}s`,
+        exec : 'obj_write',
+        gracefulStop : '5s',
+    };
+}
+
+const read_vu_count = parseInt(__ENV.READERS || '0');
+if (read_vu_count > 0) {
+    scenarios.read = {
+        executor : 'constant-vus',
+        vus : read_vu_count,
+        duration : `${duration}s`,
+        exec : 'obj_read',
+        gracefulStop : '5s',
+    };
+}
+
+const delete_age = __ENV.DELETE_AGE ? parseInt(__ENV.DELETE_AGE) : undefined;
+let obj_to_delete_selector = undefined;
+let obj_to_delete_exit_on_null = undefined;
+
+if (registry_enabled ) {
+    obj_to_delete_exit_on_null = (write_vu_count == 0) && (read_vu_count == 0)
+
+    let constructor = obj_to_delete_exit_on_null ? registry.getOneshotSelector
+        : registry.getSelector;
+
+    obj_to_delete_selector =
+        constructor(__ENV.REGISTRY_FILE, 'obj_to_delete',
+            __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0, {
+                status : 'read',
+                age : delete_age,
+            });
+}
+
+
+
+const delete_vu_count = parseInt(__ENV.DELETERS || '0');
+if (delete_vu_count > 0) {
+    if (!obj_to_delete_selector) {
+        throw 'Positive DELETE worker number without a proper object selector';
+    }
+
+    scenarios.delete = {
+        executor : 'constant-vus',
+        vus : delete_vu_count,
+        duration : `${duration}s`,
+        exec : 'obj_delete',
+        gracefulStop : '5s',
+    };
+}
+
+export const options = {
+    scenarios,
+    setupTimeout : '5s',
+};
+
+export function setup() {
+    const total_vu_count = write_vu_count + read_vu_count + delete_vu_count;
+
+    console.log(`Pregenerated buckets:          ${bucket_list.length}`);
+    console.log(`Pregenerated read object size: ${read_size}`);
+    console.log(`Pregenerated total objects:    ${obj_list.length}`);
+    console.log(`Reading VUs:                   ${read_vu_count}`);
+    console.log(`Writing VUs:                   ${write_vu_count}`);
+    console.log(`Deleting VUs:                  ${delete_vu_count}`);
+    console.log(`Total VUs:                     ${total_vu_count}`);
+
+    const start_timestamp = Date.now()
+    console.log(
+        `Load started at:               ${Date(start_timestamp).toString()}`)
+}
+
+export function teardown(data) {
+    if (obj_registry) {
+        obj_registry.close();
+    }
+    const end_timestamp = Date.now()
+    console.log(
+        `Load finished at:              ${Date(end_timestamp).toString()}`)
+}
+
+export function handleSummary(data) {
+    return {
+        'stdout' : textSummary(data, {indent : ' ', enableColors : false}),
+        [summary_json] : JSON.stringify(data),
+    };
+}
+
+export function obj_write() {
+    if (__ENV.SLEEP_WRITE) {
+        sleep(__ENV.SLEEP_WRITE);
+    }
+
+    const key = generateS3Key();
+    const bucket = bucket_list[Math.floor(Math.random() * bucket_list.length)];
+
+    const payload = generator.genPayload();
+    const resp = s3_client.put(bucket, key, payload);
+    if (!resp.success) {
+        log.withFields({bucket : bucket, key : key}).error(resp.error);
+        return;
+    }
+
+    if (obj_registry) {
+        obj_registry.addObject('', '', bucket, key, payload.hash());
+    }
+}
+
+export function obj_read() {
+    if (__ENV.SLEEP_READ) {
+        sleep(__ENV.SLEEP_READ);
+    }
+
+    if (obj_to_read_selector) {
+        const obj = obj_to_read_selector.nextObject();
+        if (!obj  ) {
+            return;
+        }
+        const resp = s3_client.get(obj.s3_bucket, obj.s3_key)
+        if (!resp.success) {
+            log.withFields({bucket : obj.s3_bucket, key : obj.s3_key, status: obj.status, op: `READ`})
+                .error(resp.error);
+        } else {
+            obj_registry.setObjectStatus(obj.id, obj.status, 'read');
+        }
+
+        return
+    }
+
+    const obj = obj_list[Math.floor(Math.random() * obj_list.length)];
+
+    const resp = s3_client.get(obj.bucket, obj.object);
+    if (!resp.success) {
+        log.withFields({bucket : obj.bucket, key : obj.object}).error(resp.error);
+    } else {
+        obj_registry.setObjectStatus(obj.id, obj.status, 'read');
+    }
+}
+
+export function obj_delete() {
+    if (__ENV.SLEEP_DELETE) {
+        sleep(__ENV.SLEEP_DELETE);
+    }
+
+    const obj = obj_to_delete_selector.nextObject();
+    delete_object(obj)
+}
+
+export function delete_object(obj) {
+    if (!obj) {
+        if (obj_to_delete_exit_on_null) {
+            exec.test.abort("No more objects to select");
+        }
+        return;
+    }
+
+    const resp = s3_client.delete(obj.s3_bucket, obj.s3_key);
+    if (!resp.success) {
+        log.withFields({bucket : obj.s3_bucket, key : obj.s3_key, op : 'DELETE'})
+            .error(resp.error);
+        return;
+    }
+
+    obj_registry.deleteObject(obj.id);
+}

scenarios/s3_multipart.js

@@ -0,0 +1,119 @@
+import {sleep} from 'k6';
+import {SharedArray} from 'k6/data';
+import logging from 'k6/x/frostfs/logging';
+import registry from 'k6/x/frostfs/registry';
+import s3 from 'k6/x/frostfs/s3';
+import stats from 'k6/x/frostfs/stats';
+
+import {generateS3Key} from './libs/keygen.js';
+import {newGenerator} from './libs/datagen.js';
+import {parseEnv} from './libs/env-parser.js';
+import {textSummary} from './libs/k6-summary-0.0.2.js';
+
+parseEnv();
+
+const bucket_list = new SharedArray('bucket_list', function() {
+  return JSON.parse(open(__ENV.PREGEN_JSON)).buckets;
+});
+
+const summary_json = __ENV.SUMMARY_JSON || '/tmp/summary.json';
+
+// Select random S3 endpoint for current VU
+const s3_endpoints = __ENV.S3_ENDPOINTS.split(',');
+const s3_endpoint =
+    s3_endpoints[Math.floor(Math.random() * s3_endpoints.length)];
+const no_verify_ssl = __ENV.NO_VERIFY_SSL || 'true';
+const connection_args = {
+  no_verify_ssl: no_verify_ssl
+};
+const s3_client = s3.connect(s3_endpoint, connection_args);
+const log = logging.new().withField('endpoint', s3_endpoint);
+
+const registry_enabled = !!__ENV.REGISTRY_FILE;
+const obj_registry =
+    registry_enabled ? registry.open(__ENV.REGISTRY_FILE) : undefined;
+
+const duration = __ENV.DURATION;
+
+if (!!__ENV.METRIC_TAGS) {
+  stats.setTags(__ENV.METRIC_TAGS)
+}
+
+const scenarios = {};
+
+const write_vu_count = parseInt(__ENV.WRITERS || '0');
+if (write_vu_count < 1) {
+  throw 'number of VUs (env WRITERS) performing write operations should be greater than 0';
+}
+
+const write_multipart_vu_count = parseInt(__ENV.WRITERS_MULTIPART || '0');
+if (write_multipart_vu_count < 1) {
+  throw 'number of parts (env WRITERS_MULTIPART) to upload in parallel should be greater than 0';
+}
+
+const generator =
+    newGenerator(write_vu_count > 0 || write_multipart_vu_count > 0);
+if (write_vu_count > 0) {
+  scenarios.write_multipart = {
+    executor: 'constant-vus',
+    vus: write_vu_count,
+    duration: `${duration}s`,
+    exec: 'obj_write_multipart',
+    gracefulStop: '5s',
+  };
+}
+
+export const options = {
+  scenarios,
+  setupTimeout: '5s',
+};
+
+export function setup() {
+  const total_vu_count = write_vu_count * write_multipart_vu_count;
+
+  console.log(`Pregenerated buckets:          ${bucket_list.length}`);
+  console.log(`Writing VUs:                   ${write_vu_count}`);
+  console.log(`Writing multipart VUs:         ${write_multipart_vu_count}`);
+  console.log(`Total VUs:                     ${total_vu_count}`);
+}
+
+export function teardown(data) {
+  if (obj_registry) {
+    obj_registry.close();
+  }
+}
+
+export function handleSummary(data) {
+  return {
+    'stdout': textSummary(data, {indent: ' ', enableColors: false}),
+    [summary_json]: JSON.stringify(data),
+  };
+}
+
+const write_multipart_part_size =
+    1024 * parseInt(__ENV.WRITE_OBJ_PART_SIZE || '0')
+if (write_multipart_part_size < 5 * 1024 * 1024) {
+  throw 'part size (env WRITE_OBJ_PART_SIZE * 1024) must be greater than (5 MB)';
+}
+
+export function obj_write_multipart() {
+  if (__ENV.SLEEP_WRITE) {
+    sleep(__ENV.SLEEP_WRITE);
+  }
+
+  const key = generateS3Key();
+  const bucket = bucket_list[Math.floor(Math.random() * bucket_list.length)];
+
+  const payload = generator.genPayload();
+  const resp = s3_client.multipart(
+      bucket, key, write_multipart_part_size, write_multipart_vu_count,
+      payload);
+  if (!resp.success) {
+    log.withFields({bucket: bucket, key: key}).error(resp.error);
+    return;
+  }
+
+  if (obj_registry) {
+    obj_registry.addObject('', '', bucket, key, payload.hash());
+  }
+}

scenarios/s3local.js

@@ -1,127 +1,173 @@
-import datagen from 'k6/x/frostfs/datagen';
+import {SharedArray} from 'k6/data';
+import exec from 'k6/execution';
 import logging from 'k6/x/frostfs/logging';
 import registry from 'k6/x/frostfs/registry';
 import s3local from 'k6/x/frostfs/s3local';
-import { SharedArray } from 'k6/data';
-import { textSummary } from './libs/k6-summary-0.0.2.js';
-import { parseEnv } from './libs/env-parser.js';
-import { uuidv4 } from './libs/k6-utils-1.4.0.js';
+import stats from 'k6/x/frostfs/stats';
+
+import {generateS3Key} from './libs/keygen.js';
+import {newGenerator} from './libs/datagen.js';
+import {parseEnv} from './libs/env-parser.js';
+import {textSummary} from './libs/k6-summary-0.0.2.js';
+import {uuidv4} from './libs/k6-utils-1.4.0.js';
 
 parseEnv();
 
-const obj_list = new SharedArray('obj_list', function () {
-    return JSON.parse(open(__ENV.PREGEN_JSON)).objects;
+const obj_list = new SharedArray('obj_list', function() {
+  return JSON.parse(open(__ENV.PREGEN_JSON)).objects;
 });
 
-const container_list = new SharedArray('container_list', function () {
-    return JSON.parse(open(__ENV.PREGEN_JSON)).containers;
+const container_list = new SharedArray('container_list', function() {
+  return JSON.parse(open(__ENV.PREGEN_JSON)).containers;
 });
 
-const bucket_list = new SharedArray('bucket_list', function () {
-    return JSON.parse(open(__ENV.PREGEN_JSON)).buckets;
+const bucket_list = new SharedArray('bucket_list', function() {
+  return JSON.parse(open(__ENV.PREGEN_JSON)).buckets;
 });
 
 function bucket_mapping() {
-    if (container_list.length != bucket_list.length) {
-        throw 'The number of containers and buckets in the preset file must be the same.';
-    }
-    let mapping = {};
-    for (let i = 0; i < container_list.length; ++i) {
-        mapping[bucket_list[i]] = container_list[i];
-    }
-    return mapping;
+  if (container_list.length != bucket_list.length) {
+    throw 'The number of containers and buckets in the preset file must be the same.';
+  }
+  let mapping = {};
+  for (let i = 0; i < container_list.length; ++i) {
+    mapping[bucket_list[i]] = container_list[i];
+  }
+  return mapping;
 }
 
 const read_size = JSON.parse(open(__ENV.PREGEN_JSON)).obj_size;
-const summary_json = __ENV.SUMMARY_JSON || "/tmp/summary.json";
+const summary_json = __ENV.SUMMARY_JSON || '/tmp/summary.json';
 
 const config_file = __ENV.CONFIG_FILE;
-const s3_client = s3local.connect(config_file, {
-    'debug_logger': __ENV.DEBUG_LOGGER || 'false',
-}, bucket_mapping());
-const log = logging.new().withField("config", config_file);
+const config_dir = __ENV.CONFIG_DIR;
+const max_total_size_gb =
+    __ENV.MAX_TOTAL_SIZE_GB ? parseInt(__ENV.MAX_TOTAL_SIZE_GB) : 0;
+const s3_client = s3local.connect(
+    config_file, config_dir, {
+      'debug_logger': __ENV.DEBUG_LOGGER || 'false',
+    },
+    bucket_mapping(), max_total_size_gb);
+const log = logging.new().withFields(
+    {'config_file': config_file, 'config_dir': config_dir});
+
+if (!!__ENV.METRIC_TAGS) {
+  stats.setTags(__ENV.METRIC_TAGS)
+}
 
 const registry_enabled = !!__ENV.REGISTRY_FILE;
-const obj_registry = registry_enabled ? registry.open(__ENV.REGISTRY_FILE) : undefined;
+const obj_registry =
+    registry_enabled ? registry.open(__ENV.REGISTRY_FILE) : undefined;
+
+let obj_to_read_selector = undefined;
+if (registry_enabled) {
+  obj_to_read_selector = registry.getLoopedSelector(
+      __ENV.REGISTRY_FILE, 'obj_to_read',
+      __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0, {
+        status: 'created',
+      })
+}
 
 const duration = __ENV.DURATION;
 
-const generator = datagen.generator(1024 * parseInt(__ENV.WRITE_OBJ_SIZE));
-
 const scenarios = {};
 
 const write_vu_count = parseInt(__ENV.WRITERS || '0');
+const generator = newGenerator(write_vu_count > 0);
 if (write_vu_count > 0) {
-    scenarios.write = {
-        executor: 'constant-vus',
-        vus: write_vu_count,
-        duration: `${duration}s`,
-        exec: 'obj_write', 
-        gracefulStop: '5s',
-    };
+  scenarios.write = {
+    executor: 'constant-vus',
+    vus: write_vu_count,
+    duration: `${duration}s`,
+    exec: 'obj_write',
+    gracefulStop: '5s',
+  };
 }
 
 const read_vu_count = parseInt(__ENV.READERS || '0');
 if (read_vu_count > 0) {
-    scenarios.read = {
-        executor: 'constant-vus',
-        vus: read_vu_count,
-        duration: `${duration}s`,
-        exec: 'obj_read', 
-        gracefulStop: '5s',
-    };
+  scenarios.read = {
+    executor: 'constant-vus',
+    vus: read_vu_count,
+    duration: `${duration}s`,
+    exec: 'obj_read',
+    gracefulStop: '5s',
+  };
 }
 
 export const options = {
-    scenarios,
-    setupTimeout: '5s',
+  scenarios,
+  setupTimeout: '5s',
 };
 
 export function setup() {
-    const total_vu_count = write_vu_count + read_vu_count;
+  const total_vu_count = write_vu_count + read_vu_count;
 
-    console.log(`Pregenerated buckets:          ${bucket_list.length}`);
-    console.log(`Pregenerated read object size: ${read_size}`);
-    console.log(`Pregenerated total objects:    ${obj_list.length}`);
-    console.log(`Reading VUs:                   ${read_vu_count}`);
-    console.log(`Writing VUs:                   ${write_vu_count}`);
-    console.log(`Total VUs:                     ${total_vu_count}`);
+  console.log(`Pregenerated buckets:          ${bucket_list.length}`);
+  console.log(`Pregenerated read object size: ${read_size}`);
+  console.log(`Pregenerated total objects:    ${obj_list.length}`);
+  console.log(`Reading VUs:                   ${read_vu_count}`);
+  console.log(`Writing VUs:                   ${write_vu_count}`);
+  console.log(`Total VUs:                     ${total_vu_count}`);
+
+  const start_timestamp = Date.now()
+  console.log(
+      `Load started at:               ${Date(start_timestamp).toString()}`)
 }
 
 export function teardown(data) {
-    if (obj_registry) {
-        obj_registry.close();
-    }
+  if (obj_registry) {
+    obj_registry.close();
+  }
+  const end_timestamp = Date.now()
+  console.log(
+      `Load finished at:              ${Date(end_timestamp).toString()}`)
 }
 
 export function handleSummary(data) {
-    return {
-        'stdout': textSummary(data, { indent: ' ', enableColors: false }),
-        [summary_json]: JSON.stringify(data),
-    };
-  }
+  return {
+    'stdout': textSummary(data, {indent: ' ', enableColors: false}),
+    [summary_json]: JSON.stringify(data),
+  };
+}
 
 export function obj_write() {
-    const key = __ENV.OBJ_NAME || uuidv4();
-    const bucket = bucket_list[Math.floor(Math.random() * bucket_list.length)];
+  const key = generateS3Key();
+  const bucket = bucket_list[Math.floor(Math.random() * bucket_list.length)];
 
-    const { payload, hash } = generator.genPayload(registry_enabled);
-    const resp = s3_client.put(bucket, key, payload);
-    if (!resp.success) {
-        log.withFields({bucket: bucket, key: key}).error(resp.error);
-        return;
+  const payload = generator.genPayload();
+  const resp = s3_client.put(bucket, key, payload);
+  if (!resp.success) {
+    if (resp.abort) {
+      exec.test.abort(resp.error);
     }
+    log.withFields({bucket: bucket, key: key}).error(resp.error);
+    return;
+  }
 
-    if (obj_registry) {
-        obj_registry.addObject("", "", bucket, key, hash);
-    }
+  if (obj_registry) {
+    obj_registry.addObject('', '', bucket, key, payload.hash());
+  }
 }
 
 export function obj_read() {
-    const obj = obj_list[Math.floor(Math.random() * obj_list.length)];
-
-    const resp = s3_client.get(obj.bucket, obj.object);
+  if (obj_to_read_selector) {
+    const obj = obj_to_read_selector.nextObject();
+    if (!obj) {
+      return;
+    }
+    const resp = s3_client.get(obj.s3_bucket, obj.s3_key)
     if (!resp.success) {
-        log.withFields({bucket: obj.bucket, key: obj.object}).error(resp.error);
-    } 
+      log.withFields({bucket: obj.s3_bucket, key: obj.s3_key})
+          .error(resp.error);
+    }
+    return
+  }
+
+  const obj = obj_list[Math.floor(Math.random() * obj_list.length)];
+
+  const resp = s3_client.get(obj.bucket, obj.object);
+  if (!resp.success) {
+    log.withFields({bucket: obj.bucket, key: obj.object}).error(resp.error);
+  }
 }

scenarios/verify.js

@@ -1,19 +1,21 @@
+import { sleep } from 'k6';
+import { Counter } from 'k6/metrics';
+import logging from 'k6/x/frostfs/logging';
 import native from 'k6/x/frostfs/native';
 import registry from 'k6/x/frostfs/registry';
 import s3 from 'k6/x/frostfs/s3';
-import logging from 'k6/x/frostfs/logging';
-import { sleep } from 'k6';
-import { Counter } from 'k6/metrics';
-import { textSummary } from './libs/k6-summary-0.0.2.js';
+import stats from 'k6/x/frostfs/stats';
+
 import { parseEnv } from './libs/env-parser.js';
+import { textSummary } from './libs/k6-summary-0.0.2.js';
 
 parseEnv();
 
 const obj_registry = registry.open(__ENV.REGISTRY_FILE);
 
 // Time limit (in seconds) for the run
-const time_limit = __ENV.TIME_LIMIT || "60";
-const summary_json = __ENV.SUMMARY_JSON || "/tmp/summary.json";
+const time_limit = __ENV.TIME_LIMIT || '60';
+const summary_json = __ENV.SUMMARY_JSON || '/tmp/summary.json';
 
 // Number of objects in each status. These counters are cumulative in a
 // sense that they reflect total number of objects in the registry, not just
@@ -21,138 +23,147 @@ const summary_json = __ENV.SUMMARY_JSON || "/tmp/summary.json";
 // This allows to run this scenario multiple times and collect overall
 // statistics in the final run.
 const obj_counters = {
-    verified: new Counter('verified_obj'),
-    skipped: new Counter('skipped_obj'),
-    invalid: new Counter('invalid_obj'),
+  verified: new Counter('verified_obj'),
+  skipped: new Counter('skipped_obj'),
+  invalid: new Counter('invalid_obj'),
 };
 
 let log = logging.new();
 
+if (!!__ENV.METRIC_TAGS) {
+  stats.setTags(__ENV.METRIC_TAGS)
+}
+
 // Connect to random gRPC endpoint
 let grpc_client = undefined;
 if (__ENV.GRPC_ENDPOINTS) {
-    const grpcEndpoints = __ENV.GRPC_ENDPOINTS.split(',');
-    const grpcEndpoint = grpcEndpoints[Math.floor(Math.random() * grpcEndpoints.length)];
-    log = log.withField("endpoint", grpcEndpoint);
-    grpc_client = native.connect(grpcEndpoint, '',
-        __ENV.DIAL_TIMEOUT ? parseInt(__ENV.DIAL_TIMEOUT) : 0,
-        __ENV.STREAM_TIMEOUT ? parseInt(__ENV.STREAM_TIMEOUT) : 0,
-        __ENV.PREPARE_LOCALLY ? __ENV.PREPARE_LOCALLY.toLowerCase() === "true" : false, '');
+  const grpcEndpoints = __ENV.GRPC_ENDPOINTS.split(',');
+  const grpcEndpoint =
+    grpcEndpoints[Math.floor(Math.random() * grpcEndpoints.length)];
+  log = log.withField('endpoint', grpcEndpoint);
+  grpc_client = native.connect(
+    grpcEndpoint, '', __ENV.DIAL_TIMEOUT ? parseInt(__ENV.DIAL_TIMEOUT) : 0,
+    __ENV.STREAM_TIMEOUT ? parseInt(__ENV.STREAM_TIMEOUT) : 0,
+    __ENV.PREPARE_LOCALLY ? __ENV.PREPARE_LOCALLY.toLowerCase() === 'true' : false,
+    1024 * parseInt(__ENV.MAX_OBJECT_SIZE || '0'));
 }
 
 // Connect to random S3 endpoint
 let s3_client = undefined;
 if (__ENV.S3_ENDPOINTS) {
-    const s3_endpoints = __ENV.S3_ENDPOINTS.split(',');
-    const s3_endpoint = s3_endpoints[Math.floor(Math.random() * s3_endpoints.length)];
-    log = log.withField("endpoint", s3_endpoint);
-    s3_client = s3.connect(`http://${s3_endpoint}`);
+  const no_verify_ssl = __ENV.NO_VERIFY_SSL || 'true';
+  const connection_args = { no_verify_ssl: no_verify_ssl };
+  const s3_endpoints = __ENV.S3_ENDPOINTS.split(',');
+  const s3_endpoint =
+    s3_endpoints[Math.floor(Math.random() * s3_endpoints.length)];
+  log = log.withField('endpoint', s3_endpoint);
+  s3_client = s3.connect(s3_endpoint, connection_args);
 }
 
-// We will attempt to verify every object in "created" status. The scenario will execute
-// as many iterations as there are objects. Each object will have 3 retries to be verified
+// We will attempt to verify every object in "created" status. The scenario will
+// execute as many iterations as there are objects. Each object will have 3
+// retries to be verified
 const obj_to_verify_selector = registry.getSelector(
-    __ENV.REGISTRY_FILE,
-    "obj_to_verify",
-    __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0,
-    {
-        status: "created",
-    }
-);
+  __ENV.REGISTRY_FILE, 'obj_to_verify',
+  __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0, {
+  status: 'created',
+});
 const obj_to_verify_count = obj_to_verify_selector.count();
-// Execute at least one iteration (executor shared-iterations can't run 0 iterations)
+// Execute at least one iteration (executor shared-iterations can't run 0
+// iterations)
 const iterations = Math.max(1, obj_to_verify_count);
-// Executor shared-iterations requires number of iterations to be larger than number of VUs
+// Executor shared-iterations requires number of iterations to be larger than
+// number of VUs
 const vus = Math.min(__ENV.CLIENTS, iterations);
 
 const scenarios = {
-    verify: {
-        executor: 'shared-iterations',
-        vus,
-        iterations,
-        maxDuration: `${time_limit}s`,
-        exec: 'obj_verify',
-        gracefulStop: '5s',
-    }
+  verify: {
+    executor: 'shared-iterations',
+    vus,
+    iterations,
+    maxDuration: `${time_limit}s`,
+    exec: 'obj_verify',
+    gracefulStop: '5s',
+  }
 };
 
 export const options = {
-    scenarios,
-    setupTimeout: '5s',
+  scenarios,
+  setupTimeout: '5s',
 };
 
 export function setup() {
-    // Populate counters with initial values
-    for (const [status, counter] of Object.entries(obj_counters)) {
-        const obj_selector = registry.getSelector(
-            __ENV.REGISTRY_FILE,
-            status,
-            __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0,
-            { status });
-        counter.add(obj_selector.count());
-    }
+  // Populate counters with initial values
+  for (const [status, counter] of Object.entries(obj_counters)) {
+    const obj_selector = registry.getSelector(
+      __ENV.REGISTRY_FILE, status,
+      __ENV.SELECTION_SIZE ? parseInt(__ENV.SELECTION_SIZE) : 0, { status });
+    counter.add(obj_selector.count());
+  }
 }
 
 export function handleSummary(data) {
-    return {
-        'stdout': textSummary(data, { indent: ' ', enableColors: false }),
-        [summary_json]: JSON.stringify(data),
-    };
+  return {
+    'stdout': textSummary(data, { indent: ' ', enableColors: false }),
+    [summary_json]: JSON.stringify(data),
+  };
 }
 
 export function obj_verify() {
-    if (obj_to_verify_count == 0) {
-        log.info("Nothing to verify");
-        return;
-    }
+  if (obj_to_verify_count == 0) {
+    log.info('Nothing to verify');
+    return;
+  }
 
-    if (__ENV.SLEEP) {
-        sleep(__ENV.SLEEP);
-    }
+  if (__ENV.SLEEP) {
+    sleep(__ENV.SLEEP);
+  }
 
-    const obj = obj_to_verify_selector.nextObject();
-    if (!obj) {
-        log.info("All objects have been verified");
-        return;
-    }
+  const obj = obj_to_verify_selector.nextObject();
+  if (!obj) {
+    log.info('All objects have been verified');
+    return;
+  }
 
-    const obj_status = verify_object_with_retries(obj, 3);
-    obj_counters[obj_status].add(1);
-    obj_registry.setObjectStatus(obj.id, obj.status, obj_status);
+  const obj_status = verify_object_with_retries(obj, 3);
+  obj_counters[obj_status].add(1);
+  obj_registry.setObjectStatus(obj.id, obj.status, obj_status);
 }
 
 function verify_object_with_retries(obj, attempts) {
-    for (let i = 0; i < attempts; i++) {
-        let result;
-        // Different name is required.
-        // ReferenceError: Cannot access a variable before initialization.
-        let lg = log;
-        if (obj.c_id && obj.o_id) {
-            lg = lg.withFields({cid: obj.c_id, oid: obj.o_id});
-            result = grpc_client.verifyHash(obj.c_id, obj.o_id, obj.payload_hash);
-        } else if (obj.s3_bucket && obj.s3_key) {
-            lg = lg.withFields({bucket: obj.s3_bucket, key: obj.s3_key});
-            result = s3_client.verifyHash(obj.s3_bucket, obj.s3_key, obj.payload_hash);
-        } else {
-            lg.withFields({
-                cid: obj.c_id,
-                oid: obj.o_id,
-                bucket: obj.s3_bucket,
-                key: obj.s3_key
-            }).warn(`Object cannot be verified with supported protocols`);
-            return "skipped";
-        }
-
-        if (result.success) {
-            return "verified";
-        } else if (result.error == "hash mismatch") {
-            return "invalid";
-        }
-
-        // Unless we explicitly saw that there was a hash mismatch, then we will retry after a delay
-        lg.error(`Verify error: ${result.error}. Object will be re-tried`);
-        sleep(__ENV.SLEEP);
+  for (let i = 0; i < attempts; i++) {
+    let result;
+    // Different name is required.
+    // ReferenceError: Cannot access a variable before initialization.
+    let lg = log;
+    if (obj.c_id && obj.o_id) {
+      lg = lg.withFields({ cid: obj.c_id, oid: obj.o_id });
+      result = grpc_client.verifyHash(obj.c_id, obj.o_id, obj.payload_hash);
+    } else if (obj.s3_bucket && obj.s3_key) {
+      lg = lg.withFields({ bucket: obj.s3_bucket, key: obj.s3_key });
+      result =
+        s3_client.verifyHash(obj.s3_bucket, obj.s3_key, obj.payload_hash);
+    } else {
+      lg.withFields({
+        cid: obj.c_id,
+        oid: obj.o_id,
+        bucket: obj.s3_bucket,
+        key: obj.s3_key
+      }).warn(`Object cannot be verified with supported protocols`);
+      return 'skipped';
     }
 
-    return "invalid";
+    if (result.success) {
+      return 'verified';
+    } else if (result.error == 'hash mismatch') {
+      return 'invalid';
+    }
+
+    // Unless we explicitly saw that there was a hash mismatch, then we will
+    // retry after a delay
+    lg.error(`Verify error: ${result.error}. Object will be re-tried`);
+    sleep(__ENV.SLEEP);
+  }
+
+  return 'invalid';
 }