[
https://issues.apache.org/jira/browse/BEAM-8941?focusedWorklogId=374496&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-374496
]
ASF GitHub Bot logged work on BEAM-8941:
----------------------------------------
Author: ASF GitHub Bot
Created on: 20/Jan/20 10:17
Start Date: 20/Jan/20 10:17
Worklog Time Spent: 10m
Work Description: pawelpasterz commented on pull request #10543:
[BEAM-8941] Implement simple DSL for load tests
URL: https://github.com/apache/beam/pull/10543#discussion_r368467539
##########
File path: .test-infra/jenkins/LoadTestConfig.groovy
##########
@@ -0,0 +1,621 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import CommonTestProperties.Runner
+import CommonTestProperties.SDK
+import LoadTestConfig.SerializableOption
+import groovy.json.JsonBuilder
+import org.codehaus.groovy.runtime.InvokerHelper
+
+import java.util.function.Predicate
+
+import static java.util.Objects.nonNull
+import static java.util.Objects.requireNonNull
+
+/**
+ * This class contains simple DSL for load tests configuration. Configuration
as Map<String, Serializable>
+ * [{@link LoadTestConfig#config config} -- returns configuration map]
+ * [{@link LoadTestConfig#templateConfig templateConfig} -- return
LoadTestConfig reusable object]
+ * [{@link LoadTestConfig#fromTemplate fromTemplate} -- returns configuration
from given template].<br><br>
+ *
+ * Example:
+ * <blockquote><pre>
+ * LoadTestConfig template = templateConfig {
+ * title 'Load test'
+ * test 'Beam Load test'
+ * dataflow()
+ * pipelineOptions {
+ * python()
+ * jobName 'Any job name'
+ * publishToBigQuery true
+ * //other fields
+ * }
+ * specificParameters([
+ * fanout: 4
+ * ])
+ * }
+ * Map<String, Serializable> configMap = fromTemplate(template) {
+ * //fields can be changed or/and added
+ * flink()
+ * pipelineOptions {
+ * parallelism 5
+ * inputOptions {
+ * numRecords 20000
+ * keySize 1000
+ * valueSize 10
+ * }
+ * }
+ * }
+ * </pre></blockquote>
+ */
+class LoadTestConfig implements SerializableOption<Map<String, Serializable>> {
+
+ private String _title
+ private String _test
+ private Runner _runner
+ private PipelineOptions _pipelineOptions
+
+ private LoadTestConfig() {}
+
+ void title(final String title) { _title = title }
+ void test(final String test) { _test = test }
+
+ //runners
+ void dataflow() { setRunnerAndUpdatePipelineOptions(Runner.DATAFLOW)}
+ void spark() { setRunnerAndUpdatePipelineOptions(Runner.SPARK) }
+ void flink() { setRunnerAndUpdatePipelineOptions(Runner.FLINK) }
+ void direct() { setRunnerAndUpdatePipelineOptions(Runner.DIRECT) }
+ void portable() { setRunnerAndUpdatePipelineOptions(Runner.PORTABLE) }
+
+ private void setRunnerAndUpdatePipelineOptions(final Runner runner) {
+ _runner = runner
+ final def pipeline = _pipelineOptions ?: new PipelineOptions()
+ pipeline.i_runner = runner
+ _pipelineOptions = pipeline
+ }
+
+ void pipelineOptions(final Closure cl = {}) {
+ final def options = _pipelineOptions ?: new PipelineOptions()
+ delegateAndInvoke(options, cl)
+ _pipelineOptions = options
+ }
+
+ /**
+ * Returns load test config object which can be reusable.</br>
+ * All possible fields that can be set:
+ * <blockquote><pre>
+ * templateConfig {
+ * title [String]
+ * test [String]
+ * [dataflow(), spark(), flink(), direct(), portable()] -- runner
+ * pipelineOptions {
+ * [python(), python37(), java()] -- sdk
+ * jobName [String]
+ * project [String]
+ * publishToBigQuery [boolean]
+ * metricsDataset (python) [String]
+ * metricsTable (python) [String]
+ * bigQueryDataset (java) [String]
+ * bigQueryTable (java) [String]
+ * numWorkers [int]
+ * parallelism [int]
+ * tempLocation [String]
+ * autoscalingAlgorithm [String]
+ * jobEndpoint [String]
+ * environmentType [String]
+ * environmentConfig [String]
+ * inputOptions/coInputOptions (for python) {
+ * numRecords [int]
+ * keySize [int]
+ * valueSize [int]
+ * numHotKeys [int]
+ * hotKeyFraction [int]
+ * }
+ * sourceOptions/coSourceOptions (for java) {
+ * numRecords [int]
+ * keySizeBytes [int]
+ * valueSizeBytes [int]
+ * numHotKeys [int]
+ * hotKeyFraction [int]
+ * splitPointFrequencyRecords [int]
+ * }
+ * stepOptions {
+ * outputRecordsPerInputRecord [int]
+ * preservesInputKeyDistribution [boolean]
+ * }
+ * specificParameters [Map<String, Object>]
+ * }
+ * }
+ * </pre></blockquote>
+ * @param cl Closure with fields setting
+ * @return LoadTestConfig object
+ */
+ static LoadTestConfig templateConfig(final Closure cl = {}) {
+ final def config = new LoadTestConfig()
+ delegateAndInvoke(config, cl)
+ return config
+ }
+
+ /**
+ * Returns configuration map from given template. Any field can be changed
or/and added. Validation is performed
+ * before final map is returned (ex. Flink runner requires
<b>environmentConfig</b> to be set). In case of
+ * validation failure exception is thrown.<br>
+ * Example result:
+ *<blockquote><pre>
+ * [
+ * title : 'any given title',
+ * test : 'any given test',
+ * runner : CommonTestProperties.Runner.DATAFLOW,
+ * pipelineOptions: [
+ * job_name : 'any given job name',
+ * publish_to_big_query: true,
+ * project : 'apache-beam-testing',
+ * metrics_dataset : 'given_dataset_name',
+ * metrics_table : 'given_table_name',
+ * input_options : '\'{"num_records": 200000000,"key_size":
1,"value_size":9}\'',
+ * iterations : 1,
+ * fanout : 1,
+ * parallelism : 5,
+ * job_endpoint : 'localhost:1234',
+ * environment_config : 'given_environment_config',
+ * environment_type : 'given_environment_type'
+ * ]
+ * ]
+ * </blockquote></pre>
+ * @param templateConfig LoadTestConfig instance
+ * @param cl Closure with fields setting
+ * @return configuration map
+ * @see LoadTestConfig
+ * @see LoadTestConfig#templateConfig
+ */
+ static Map<String, Serializable> fromTemplate(final LoadTestConfig
templateConfig, final Closure cl = {}) {
+ final def newConfig = of(templateConfig)
+ delegateAndInvoke(newConfig, cl)
+ final def properties = newConfig.propertiesMap
+ verifyProperties(properties)
+ return ConfigHelper.convertProperties(properties)
+ }
+
+ /**
+ * Returns configuration map (see {@link LoadTestConfig#fromTemplate})
directly from given settings
+ * @param cl Closure with settings
+ * @return configuration map
+ */
+ static Map<String, Serializable> config(final Closure cl = {}) {
+ final def config = new LoadTestConfig()
+ delegateAndInvoke(config, cl)
+ final def properties = config.propertiesMap
+ verifyProperties(properties)
+ return ConfigHelper.convertProperties(config.propertiesMap)
+ }
+
+ private static void delegateAndInvoke(final delegate, final Closure cl =
{}) {
+ final def code = cl.rehydrate(delegate, this, this)
+ code.resolveStrategy = Closure.DELEGATE_ONLY
+ code()
+ }
+
+ private static LoadTestConfig of(final LoadTestConfig oldConfig) {
+ final def newConfig = new LoadTestConfig()
+
+ //primitive values
+ InvokerHelper.setProperties(newConfig, oldConfig.propertiesMap)
+
+ //non-primitive values
+ newConfig._pipelineOptions = oldConfig._pipelineOptions ?
PipelineOptions.of(oldConfig._pipelineOptions) : null
+
+ return newConfig
+ }
+
+ @Override
+ Map<String, Serializable> toPrimitiveValues() {
+ final def map = propertiesMap
+ verifyProperties(map)
+ return ConfigHelper.convertProperties(map)
+ }
+
+ LinkedHashMap<String, Object> getPropertiesMap() {
+ return [
+ _title: _title,
+ _test: _test,
+ _runner: _runner,
+ _pipelineOptions: _pipelineOptions
+ ]
+ }
+
+ private static void verifyProperties(final LinkedHashMap<String, Object>
map) {
+ for (entry in map.entrySet()) {
+ requireNonNull(entry.value, "Missing ${entry.key.substring(1)} in
configuration")
+ }
+ }
+
+ private static class PipelineOptions implements
SerializableOption<Map<String, Serializable>> {
+ private Map<String, Object> _specificParameters = new HashMap<>()
+ private boolean _streaming = false
+ private SourceOptions _coSourceOptions
+ private InputOptions _coInputOptions
+ private StepOptions _stepOptions
+
+ //required
+ private String _jobName
+ private String _project
+ private String _publishToBigQuery
+
+ //java required
+ private String _bigQueryDataset
+ private String _bigQueryTable
+ private SourceOptions _sourceOptions
+
+ //python required
+ private String _metricsDataset
+ private String _metricsTable
+ private InputOptions _inputOptions
+
+ //internal usage
+ private SDK i_sdk
+ private Runner i_runner
+ private static final i_required = ["_jobName", "_project",
"_publishToBigQuery"]
+ private static final i_dataflowRequired = ["_numWorkers",
"_tempLocation", "_autoscalingAlgorithm"]
+ private static final i_flinkRequired = ["_jobEndpoint",
"_environmentType", "_environmentConfig", "_parallelism"]
+ private static final i_javaRequired = ["_bigQueryDataset",
"_bigQueryTable", "_sourceOptions"]
+ private static final i_pythonRequired = ["_metricsDataset",
"_metricsTable", "_inputOptions"]
+
+ //dataflow required
+ private def _numWorkers
+ private String _tempLocation
+ private String _autoscalingAlgorithm
+
+ //flink required
+ private String _jobEndpoint
+ private String _environmentType
+ private String _environmentConfig
+ private def _parallelism
+
+ void jobName(final String name) { _jobName = name }
+ void project(final String project) { _project = project }
+ void tempLocation(final String location) { _tempLocation = location }
+ void publishToBigQuery(final boolean publish) { _publishToBigQuery =
publish }
+ void metricsDataset(final String dataset) { _metricsDataset = dataset }
+ void metricsTable(final String table) { _metricsTable = table }
+ void inputOptions(final InputOptions options) { _inputOptions =
options }
+ void numWorkers(final int workers) { _numWorkers = workers }
+ void autoscalingAlgorithm(final String algorithm) {
_autoscalingAlgorithm = algorithm }
+ void jobEndpoint(final String endpoint) { _jobEndpoint = endpoint }
+ void environmentType(final String type) { _environmentType = type }
+ void environmentConfig(final String config) { _environmentConfig =
config }
+ void parallelism(final int parallelism) { _parallelism = parallelism }
+ void bigQueryDataset(final String dataset) { _bigQueryDataset =
dataset }
+ void bigQueryTable(final String table) { _bigQueryTable = table }
+ void streaming(final boolean isStreaming) { _streaming = isStreaming }
+ void sourceOptions(final Closure cl = {}) { _sourceOptions =
makeSourceOptions {cl} }
+ void coSourceOptions(final Closure cl = {}) { _coSourceOptions =
makeSourceOptions(cl) }
+ void inputOptions(final Closure cl = {}) { _inputOptions =
makeInputOptions(cl) }
+ void coInputOptions(final Closure cl = {}) { _coInputOptions =
makeInputOptions(cl) }
+ void stepOptions(final Closure cl = {}) { _stepOptions =
makeStepOptions(cl) }
+ void specificParameters(final Map<String, Object> map) {
_specificParameters.putAll(map) }
+
+ //sdk -- snake_case vs camelCase
+ void python() { i_sdk = SDK.PYTHON }
+ void python37() { i_sdk = SDK.PYTHON_37 }
+ void java() { i_sdk = SDK.JAVA }
+
+
+ private InputOptions makeInputOptions(final Closure cl = {}) {
+ return makeOptions(cl, _inputOptions ?:
InputOptions.withSDK(i_sdk))
+ }
+
+ private SourceOptions makeSourceOptions(final Closure cl = {}) {
+ return makeOptions(cl, _sourceOptions ?:
SourceOptions.withSDK(i_sdk))
+ }
+
+ private StepOptions makeStepOptions(final Closure cl = {}) {
+ return makeOptions(cl, _stepOptions ?: StepOptions.withSDK(i_sdk))
+ }
+
+ private <T> T makeOptions(final Closure cl = {}, final T options) {
+ final def code = cl.rehydrate(options, this, this)
+ code.resolveStrategy = Closure.DELEGATE_ONLY
+ code()
+ return options
+ }
+
+ @Override
+ Map<String, Serializable> toPrimitiveValues() {
+ final def map = propertiesMap
+ verifyPipelineProperties(map)
+ return ConfigHelper.convertProperties(map, i_sdk)
+ }
+
+ private void verifyPipelineProperties(final Map<String, Object> map) {
+ verifyRequired(map)
+ switch (i_runner) {
+ case Runner.DATAFLOW:
+ verifyDataflowProperties(map)
+ break
+ case Runner.FLINK:
+ verifyFlinkProperties(map)
+ break
+ default:
+ break
+ }
+ }
+
+ private void verifyRequired(final Map<String, Object> map) {
+ verifyCommonRequired(map)
+ switch (i_sdk) {
+ case SDK.PYTHON:
+ case SDK.PYTHON_37:
+ verifyPythonRequired(map)
+ break
+ case SDK.JAVA:
+ verifyJavaRequired(map)
+ break
+ default:
+ break
+ }
+ }
+
+ private static void verifyCommonRequired(final Map<String, Object>
map) {
+ verify(map, "") { i_required.contains(it.key) }
+ }
+
+ private static void verifyPythonRequired(final Map<String, Object>
map) {
+ verify(map, "for Python SDK") { i_pythonRequired.contains(it.key) }
+ }
+
+ private static void verifyJavaRequired(final Map<String, Object> map) {
+ verify(map, "for Java SDK") { i_javaRequired.contains(it.key) }
+ }
+
+ private static void verifyDataflowProperties(final Map<String, Object>
map) {
+ verify(map, "for Dataflow runner") {
i_dataflowRequired.contains(it.key) }
+ }
+
+ private static void verifyFlinkProperties(final Map<String, Object>
map) {
+ verify(map, "for Flink runner") { i_flinkRequired.contains(it.key)
}
+ }
+
+ private static void verify(final Map<String, Object> map, final String
message, final Predicate<Map.Entry<String, Object>> predicate) {
+ map.entrySet()
+ .stream()
+ .filter(predicate)
+ .forEach{ requireNonNull(it.value, "${it.key.substring(1)}
is required " + message) }
+ }
+
+ static PipelineOptions of(final PipelineOptions options) {
+ final def newOptions = new PipelineOptions()
+
+ //primitive values
+ InvokerHelper.setProperties(newOptions, options.propertiesMap)
+
+ //non-primitive
+ newOptions._inputOptions = options._inputOptions ?
InputOptions.of(options._inputOptions) : null
+ newOptions._coInputOptions = options._coInputOptions ?
InputOptions.of(options._coInputOptions) : null
+ newOptions._sourceOptions = options._sourceOptions ?
SourceOptions.of(options._sourceOptions) : null
+ newOptions._coSourceOptions = options._coSourceOptions ?
SourceOptions.of(options._coSourceOptions) : null
+ newOptions._stepOptions = options._stepOptions ?
StepOptions.of(options._stepOptions) : null
+ newOptions._specificParameters = new
HashMap<>(options._specificParameters)
+
+ return newOptions
+ }
+
+ Map<String, Object> getPropertiesMap() {
+ return [
+ i_sdk: i_sdk,
+ i_runner: i_runner,
+ _jobName: _jobName,
+ _project: _project,
+ _tempLocation: _tempLocation,
+ _publishToBigQuery: _publishToBigQuery,
+ _metricsDataset: _metricsDataset,
+ _metricsTable: _metricsTable,
+ _numWorkers: _numWorkers,
+ _autoscalingAlgorithm: _autoscalingAlgorithm,
+ _inputOptions: _inputOptions,
+ _coInputOptions: _coInputOptions,
+ _jobEndpoint: _jobEndpoint,
+ _environmentType: _environmentType,
+ _environmentConfig: _environmentConfig,
+ _parallelism: _parallelism,
+ _bigQueryDataset: _bigQueryDataset,
+ _bigQueryTable: _bigQueryTable,
+ _streaming: _streaming,
+ _sourceOptions: _sourceOptions,
+ _coSourceOptions: _coSourceOptions,
+ _stepOptions: _stepOptions
+ ].putAll(_specificParameters.entrySet())
+ }
+
+ private static class InputOptions implements
SerializableOption<String> {
+ private def _numRecords
+ private def _keySize
+ private def _valueSize
+ private def _numHotKeys
+ private def _hotKeyFraction
+
+ //internal usage
+ private SDK i_sdk
+
+ private InputOptions() {}
+
+ static withSDK(final SDK sdk) {
+ final def input = new InputOptions()
+ input.i_sdk = sdk
+ return input
+ }
+
+ void numRecords(final int num) { _numRecords = num }
+ void keySize(final int size) { _keySize = size }
+ void valueSize(final int size) { _valueSize = size }
+ void numHotsKeys(final int num) { _numHotKeys = num }
+ void hotKeyFraction(final int fraction) { _hotKeyFraction =
fraction }
+
+ @Override
+ String toPrimitiveValues() {
+ return '\'' + new
JsonBuilder(ConfigHelper.convertProperties(propertiesMap, i_sdk)).toString() +
'\''
+ }
+
+ static InputOptions of(final InputOptions oldOptions) {
+ final def newOptions = new InputOptions()
+ InvokerHelper.setProperties(newOptions,
oldOptions.propertiesMap)
+ return newOptions
+ }
+
+ LinkedHashMap<String, Object> getPropertiesMap() {
+ return [
+ i_sdk: i_sdk,
+ _numRecords: _numRecords,
+ _keySize: _keySize,
+ _valueSize: _valueSize,
+ _numHotKeys: _numHotKeys,
+ _hotKeyFraction: _hotKeyFraction
+ ] as LinkedHashMap<String, Object>
+ }
+ }
+
+ private static class SourceOptions implements
SerializableOption<String> {
+ private def _numRecords
+ private def _keySizeBytes
+ private def _valueSizeBytes
+ private def _numHotKeys
+ private def _hotKeyFraction
+ private def _splitPointFrequencyRecords
+
+ //internal usage
+ private SDK i_sdk
+
+ private SourceOptions() {}
+
+ static withSDK(final SDK sdk) {
+ final def input = new SourceOptions()
+ input.i_sdk = sdk
+ return input
+ }
+
+ void numRecords(final int num) { _numRecords = num }
+ void keySizeBytes(final int size) { _keySizeBytes = size }
+ void valueSizeBytes(final int size) { _valueSizeBytes = size }
+ void numHotsKeys(final int num) { _numHotKeys = num }
+ void hotKeyFraction(final int fraction) { _hotKeyFraction =
fraction }
+ void splitPointFrequencyRecords(final int splitPoint) {
_splitPointFrequencyRecords = splitPoint }
+
+ @Override
+ String toPrimitiveValues() {
+ return '\'' + new
JsonBuilder(ConfigHelper.convertProperties(propertiesMap, i_sdk)).toString() +
'\''
Review comment:
Thanks, will change it!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 374496)
Time Spent: 3h 10m (was: 3h)
> Create a common place for Load Tests configuration
> --------------------------------------------------
>
> Key: BEAM-8941
> URL: https://issues.apache.org/jira/browse/BEAM-8941
> Project: Beam
> Issue Type: Sub-task
> Components: testing
> Reporter: Kamil Wasilewski
> Assignee: Pawel Pasterz
> Priority: Minor
> Fix For: Not applicable
>
> Time Spent: 3h 10m
> Remaining Estimate: 0h
>
> The Apache Beam community maintains different versions of each Load Test. For
> example, right now, there are two versions of all Python Load Tests: the
> first one runs on Dataflow runner, and the second one runs on Flink. With the
> lack of a common place where configuration for the tests can be stored, the
> configuration is duplicated many times with minimal differences.
> The goal is to create a common place for the configuration, so that it could
> be passed to different files with tests (.test-infra/jenkins/*.groovy) and
> filtered according to needs.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
