[
https://issues.apache.org/jira/browse/BEAM-8941?focusedWorklogId=370736&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-370736
]
ASF GitHub Bot logged work on BEAM-8941:
----------------------------------------
Author: ASF GitHub Bot
Created on: 13/Jan/20 12:18
Start Date: 13/Jan/20 12:18
Worklog Time Spent: 10m
Work Description: kamilwu commented on pull request #10543: [BEAM-8941]
Implement simple DSL for load tests
URL: https://github.com/apache/beam/pull/10543#discussion_r365774295
##########
File path: .test-infra/jenkins/LoadTestConfig.groovy
##########
@@ -0,0 +1,436 @@
+/*
+ * 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 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
+ * }
+ * }
+ * 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 [String]
+ * metricsTable [String]
+ * iterations [int]
+ * fanout [int]
+ * numWorkers [int]
+ * parallelism [int]
+ * tempLocation [String]
+ * autoscalingAlgorithm [String]
+ * jobEndpoint [String]
+ * environmentType [String]
+ * environmentConfig [String]
+ * inputOptions {
+ * numRecords [int]
+ * keySize [int]
+ * valueSize [int]
+ * numHotKeys [int]
+ * hotKeyFraction [int]
+ * }
+ * }
+ * }
+ * </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 String _jobName
+ private String _project
+ private String _publishToBigQuery
+ private String _metricsDataset
+ private String _metricsTable
+ private InputOptions _inputOptions
+ private def _iterations
Review comment:
I'm afraid I don't have such a list.
I wonder whether we should explicitly specify all possible parameters. Every
time we add a new parameter to one of the tests, we must also remember about
this place.
What about having a fixed list of parameters (publish_to_big_query,
metrics_dataset, metrics_table, input_options, runner's parameters, etc.) that
every test require and accepting all the other parameters without validation?
----------------------------------------------------------------
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: 370736)
Time Spent: 1h (was: 50m)
> 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: 1h
> 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)
