diff --git a/internal/s3/client.go b/internal/s3/client.go index 0c05d57..c01d727 100644 --- a/internal/s3/client.go +++ b/internal/s3/client.go @@ -70,6 +70,26 @@ func (c *Client) Put(bucket, key string, payload goja.ArrayBuffer) PutResponse { return PutResponse{Success: true} } +func (c *Client) Multipart(bucket, key string, objPartSize int, payload goja.ArrayBuffer, concurrency int) 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) + result, err := uploader.Upload(context.TODO(), &s3.PutObjectInput{ + Bucket: aws.String(bucket), + Key: aws.String(key), + Body: payload.Bytes(), + PartSize: objPartSize, + Concurrency: currency + }) + if err != nil { + stats.Report(c.vu, objPutFails, 1) + return PutResponse{Success: false, Error: err.Error()} + } +} + func (c *Client) Delete(bucket, key string) DeleteResponse { start := time.Now() diff --git a/scenarios/s3_multipart.js b/scenarios/s3_multipart.js new file mode 100644 index 0000000..c04c3b3 --- /dev/null +++ b/scenarios/s3_multipart.js @@ -0,0 +1,101 @@ +import datagen from 'k6/x/frostfs/datagen'; +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'; + +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"; + +// 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 registry_enabled = !!__ENV.REGISTRY_FILE; +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 || ""); + +const scenarios = {}; + +const write_vu_count = parseInt(__ENV.WRITERS || '0'); +const write_multipart_vu_count = parseInt(__ENV.WRITERS_MULTIPART || '0'); +if (write_vu_count > 0) { + scenarios.write = { + 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(`Pregenerated read object size: ${read_size}`); + console.log(`Pregenerated total objects: ${obj_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') + +export function obj_write_multipart() { + 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 {payload, hash} = generator.genPayload(registry_enabled); + const resp = s3_client.multipart(bucket, key, write_multipart_part_size, payload, write_multipart_vu_count); + if (!resp.success) { + log.withFields({bucket: bucket, key: key}).error(resp.error); + return; + } + + if (obj_registry) { + obj_registry.addObject("", "", bucket, key, hash); + } +} +