testgen-pulsar.py

#!/usr/bin/env python

from __future__ import print_function
import json
import sys

test_list = []

localWorker = False
tarball = 'package/target/openmessaging-benchmark-0.0.1-SNAPSHOT-bin.tar.gz'
build = False

for repeat in range(1):
    for producerWorkers in [1]:
        numWorkers = 0 if localWorker else producerWorkers*2
        for testDurationMinutes in [4]:
            for messageSize in [100, 10000]:
                messageSize = int(messageSize)
                eps = []
                if messageSize == 100:
                    eps = [1e2, 1e3, 5e3, 1e4, 5e4, 1e5, 2e5, 3e5, 5e5, 8e5, -1]
                elif messageSize == 10000:
                    eps = [1e3, 3e3, 6e3, 9e3, 12e3, 15e3, 20e3, 25e3, 30e3, 35e3, 40e3]
                else:
                    print("Wrong message size")
                    continue

                for producerRateEventsPerSec in eps:
                    for topics in [1]:
                        for partitionsPerTopic in [1, 16]:
                            for producersPerWorker in [1]:
                                producersPerTopic = int(producersPerWorker * producerWorkers)
                                for consumerBacklogSizeGB in [0]:
                                    for subscriptionsPerTopic in [1]:
                                        for consumerPerSubscription in [partitionsPerTopic]:
                                            for ackQuorum in [2]:   # 2 causes OOM in Bookie at max rate
                                                driver = {
                                                    'name': 'Pulsar',
                                                    'driverClass': 'io.openmessaging.benchmark.driver.pulsar.PulsarBenchmarkDriver',
                                                    'enableTiering': True,
                                                    'client': {
                                                               'ioThreads': 8,
                                                               'connectionsPerBroker': 8,
                                                               'clusterName': 'local',
                                                               'namespacePrefix': 'benchmark/ns',
                                                               'topicType': 'persistent',
                                                               'persistence': {'ensembleSize': 3,
                                                                               'writeQuorum': 3,
                                                                               'ackQuorum': ackQuorum,
                                                                               'deduplicationEnabled': True},
                                                               'tlsAllowInsecureConnection': False,
                                                               'tlsEnableHostnameVerification': False,
                                                               'tlsTrustCertsFilePath': None,
                                                               'authentication': {'plugin': None, 'data': None}},
                                                    'producer': {'batchingEnabled': True,
                                                                 'batchingMaxPublishDelayMs': 1,
                                                                 'blockIfQueueFull': True,
                                                                 'pendingQueueSize': 10000},
                                                }
                                                workload = {
                                                    'messageSize': messageSize,
                                                    'topics':  topics,
                                                    'partitionsPerTopic': partitionsPerTopic,
                                                    'subscriptionsPerTopic': subscriptionsPerTopic,
                                                    'consumerPerSubscription': consumerPerSubscription,
                                                    'producersPerTopic': producersPerTopic,
                                                    'producerRate': producerRateEventsPerSec,
                                                    'consumerBacklogSizeGB': consumerBacklogSizeGB,
                                                    'testDurationMinutes': testDurationMinutes,
                                                    'keyDistributor': 'RANDOM_NANO',
                                                }
                                                t = dict(
                                                    test='openmessaging-benchmark',
                                                    max_test_attempts=1,
                                                    result_filename='data/results/json/%(test)s_%(test_uuid)s.json',
                                                    driver=driver,
                                                    workload=workload,
                                                    numWorkers=numWorkers,
                                                    localWorker=localWorker,
                                                    tarball=tarball,
                                                    build=build,
                                                    undeploy=True,
                                                )
                                                test_list += [t]
                                                build = False


print(json.dumps(test_list, sort_keys=True, indent=4, ensure_ascii=False))
print('Number of tests generated: %d' % len(test_list), file=sys.stderr)