cmd/xk6-registry-exporter/root.go

@@ -78,7 +78,7 @@ func rootCmdRun(cmd *cobra.Command, args []string) error {
 	}
 
 	objRegistry := registry.NewObjRegistry(cmd.Context(), args[0])
-	objSelector := registry.NewObjSelector(objRegistry, 0, false, &registry.ObjFilter{
+	objSelector := registry.NewObjSelector(objRegistry, 0, registry.SelectorAwaiting, &registry.ObjFilter{
 		Status: status,
 		Age:    age,
 	})

internal/registry/obj_selector.go

@@ -20,7 +20,7 @@ type ObjSelector struct {
 	boltDB    *bbolt.DB
 	filter    *ObjFilter
 	cacheSize int
-	looped    bool
+	kind      SelectorKind
 }
 
 // objectSelectCache is the default maximum size of a batch to select from DB.
@@ -28,7 +28,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, looped bool, filter *ObjFilter) *ObjSelector {
+func NewObjSelector(registry *ObjRegistry, selectionSize int, kind SelectorKind, filter *ObjFilter) *ObjSelector {
 	if selectionSize <= 0 {
 		selectionSize = objectSelectCache
 	}
@@ -41,7 +41,7 @@ func NewObjSelector(registry *ObjRegistry, selectionSize int, looped bool, filte
 		filter:    filter,
 		objChan:   make(chan *ObjectInfo, selectionSize*2),
 		cacheSize: selectionSize,
-		looped:    looped,
+		kind:      kind,
 	}
 
 	go objSelector.selectLoop()
@@ -162,7 +162,11 @@ func (o *ObjSelector) selectLoop() {
 			}
 		}
 
-		if !o.looped && 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):
@@ -171,7 +175,7 @@ func (o *ObjSelector) selectLoop() {
 			}
 		}
 
-		if o.looped && len(cache) != o.cacheSize {
+		if o.kind == SelectorLooped && len(cache) != o.cacheSize {
 			lastID = 0
 		}
 

internal/registry/registry.go

@@ -74,15 +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, false, filter)
+	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, true, filter)
+	return r.getSelectorInternal(dbFilePath, name, cacheSize, SelectorLooped, filter)
 }
 
-func (r *Registry) getSelectorInternal(dbFilePath string, name string, cacheSize int, looped bool, filter map[string]string) *ObjSelector {
+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)
@@ -94,7 +114,7 @@ func (r *Registry) getSelectorInternal(dbFilePath string, name string, cacheSize
 	selector := r.root.selectors[name]
 	if selector == nil {
 		registry := r.open(dbFilePath)
-		selector = NewObjSelector(registry, cacheSize, looped, 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))

scenarios/grpc.js

@@ -1,5 +1,6 @@
 import {sleep} from 'k6';
 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';
@@ -12,13 +13,13 @@ import {uuidv4} from './libs/k6-utils-1.4.0.js';
 
 parseEnv();
 
-const obj_list = new SharedArray('obj_list', function() {
+const obj_list = new SharedArray(
-  return JSON.parse(open(__ENV.PREGEN_JSON)).objects;
+    'obj_list',
-});
+    function() { return JSON.parse(open(__ENV.PREGEN_JSON)).objects; });
 
-const container_list = new SharedArray('container_list', function() {
+const container_list = new SharedArray(
-  return JSON.parse(open(__ENV.PREGEN_JSON)).containers;
+    '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';
@@ -30,8 +31,8 @@ const grpc_endpoint =
 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' :
+    __ENV.PREPARE_LOCALLY ? __ENV.PREPARE_LOCALLY.toLowerCase() === 'true'
-                            false);
+                          : false);
 const log = logging.new().withField('endpoint', grpc_endpoint);
 
 const registry_enabled = !!__ENV.REGISTRY_FILE;
@@ -44,25 +45,14 @@ 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,
-      });
-}
-
 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',
+        status : 'created',
-        age: read_age,
+        age : read_age,
       })
 }
 
@@ -73,22 +63,39 @@ 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',
+    executor : 'constant-vus',
-    vus: write_vu_count,
+    vus : write_vu_count,
-    duration: `${duration}s`,
+    duration : `${duration}s`,
-    exec: 'obj_write',
+    exec : 'obj_write',
-    gracefulStop: '5s',
+    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',
+    executor : 'constant-vus',
-    vus: read_vu_count,
+    vus : read_vu_count,
-    duration: `${duration}s`,
+    duration : `${duration}s`,
-    exec: 'obj_read',
+    exec : 'obj_read',
-    gracefulStop: '5s',
+    gracefulStop : '5s',
   };
 }
 
@@ -100,17 +107,17 @@ if (delete_vu_count > 0) {
   }
 
   scenarios.delete = {
-    executor: 'constant-vus',
+    executor : 'constant-vus',
-    vus: delete_vu_count,
+    vus : delete_vu_count,
-    duration: `${duration}s`,
+    duration : `${duration}s`,
-    exec: 'obj_delete',
+    exec : 'obj_delete',
-    gracefulStop: '5s',
+    gracefulStop : '5s',
   };
 }
 
 export const options = {
   scenarios,
-  setupTimeout: '5s',
+  setupTimeout : '5s',
 };
 
 export function setup() {
@@ -140,8 +147,8 @@ export function teardown(data) {
 
 export function handleSummary(data) {
   return {
-    'stdout': textSummary(data, {indent: ' ', enableColors: false}),
+    'stdout' : textSummary(data, {indent : ' ', enableColors : false}),
-    [summary_json]: JSON.stringify(data),
+    [summary_json] : JSON.stringify(data),
   };
 }
 
@@ -150,7 +157,7 @@ export function obj_write() {
     sleep(__ENV.SLEEP_WRITE);
   }
 
-  const headers = {unique_header: uuidv4()};
+  const headers = {unique_header : uuidv4()};
   const container =
       container_list[Math.floor(Math.random() * container_list.length)];
 
@@ -179,7 +186,7 @@ export function obj_read() {
     }
     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);
+      log.withFields({cid : obj.c_id, oid : obj.o_id}).error(resp.error);
     }
     return
   }
@@ -187,7 +194,7 @@ export function obj_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);
+    log.withFields({cid : obj.container, oid : obj.object}).error(resp.error);
   }
 }
 
@@ -198,13 +205,16 @@ export function obj_delete() {
 
   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);
+    log.withFields({cid : obj.c_id, oid : obj.o_id}).error(resp.error);
     return;
   }
 

scenarios/s3.js

@@ -1,5 +1,6 @@
 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';
@@ -12,20 +13,20 @@ import {uuidv4} from './libs/k6-utils-1.4.0.js';
 
 parseEnv();
 
-const obj_list = new SharedArray('obj_list', function() {
+const obj_list = new SharedArray(
-  return JSON.parse(open(__ENV.PREGEN_JSON)).objects;
+    'obj_list',
-});
+    function() { return JSON.parse(open(__ENV.PREGEN_JSON)).objects; });
 
-const bucket_list = new SharedArray('bucket_list', function() {
+const bucket_list = new SharedArray(
-  return JSON.parse(open(__ENV.PREGEN_JSON)).buckets;
+    '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
+  no_verify_ssl : no_verify_ssl
 }
 // Select random S3 endpoint for current VU
 const s3_endpoints = __ENV.S3_ENDPOINTS.split(',');
@@ -44,25 +45,14 @@ 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,
-      });
-}
-
 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',
+        status : 'created',
-        age: read_age,
+        age : read_age,
       })
 }
 
@@ -72,22 +62,39 @@ 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',
+    executor : 'constant-vus',
-    vus: write_vu_count,
+    vus : write_vu_count,
-    duration: `${duration}s`,
+    duration : `${duration}s`,
-    exec: 'obj_write',
+    exec : 'obj_write',
-    gracefulStop: '5s',
+    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',
+    executor : 'constant-vus',
-    vus: read_vu_count,
+    vus : read_vu_count,
-    duration: `${duration}s`,
+    duration : `${duration}s`,
-    exec: 'obj_read',
+    exec : 'obj_read',
-    gracefulStop: '5s',
+    gracefulStop : '5s',
   };
 }
 
@@ -98,17 +105,17 @@ if (delete_vu_count > 0) {
   }
 
   scenarios.delete = {
-    executor: 'constant-vus',
+    executor : 'constant-vus',
-    vus: delete_vu_count,
+    vus : delete_vu_count,
-    duration: `${duration}s`,
+    duration : `${duration}s`,
-    exec: 'obj_delete',
+    exec : 'obj_delete',
-    gracefulStop: '5s',
+    gracefulStop : '5s',
   };
 }
 
 export const options = {
   scenarios,
-  setupTimeout: '5s',
+  setupTimeout : '5s',
 };
 
 export function setup() {
@@ -138,8 +145,8 @@ export function teardown(data) {
 
 export function handleSummary(data) {
   return {
-    'stdout': textSummary(data, {indent: ' ', enableColors: false}),
+    'stdout' : textSummary(data, {indent : ' ', enableColors : false}),
-    [summary_json]: JSON.stringify(data),
+    [summary_json] : JSON.stringify(data),
   };
 }
 
@@ -154,7 +161,7 @@ export function obj_write() {
   const payload = generator.genPayload();
   const resp = s3_client.put(bucket, key, payload);
   if (!resp.success) {
-    log.withFields({bucket: bucket, key: key}).error(resp.error);
+    log.withFields({bucket : bucket, key : key}).error(resp.error);
     return;
   }
 
@@ -175,7 +182,7 @@ export function obj_read() {
     }
     const resp = s3_client.get(obj.s3_bucket, obj.s3_key)
     if (!resp.success) {
-      log.withFields({bucket: obj.s3_bucket, key: obj.s3_key})
+      log.withFields({bucket : obj.s3_bucket, key : obj.s3_key})
           .error(resp.error);
     }
     return
@@ -185,7 +192,7 @@ export function obj_read() {
 
   const resp = s3_client.get(obj.bucket, obj.object);
   if (!resp.success) {
-    log.withFields({bucket: obj.bucket, key: obj.object}).error(resp.error);
+    log.withFields({bucket : obj.bucket, key : obj.object}).error(resp.error);
   }
 }
 
@@ -196,12 +203,15 @@ export function obj_delete() {
 
   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 = s3_client.delete(obj.s3_bucket, obj.s3_key);
   if (!resp.success) {
-    log.withFields({bucket: obj.s3_bucket, key: obj.s3_key, op: 'DELETE'})
+    log.withFields({bucket : obj.s3_bucket, key : obj.s3_key, op : 'DELETE'})
         .error(resp.error);
     return;
   }