[
https://issues.apache.org/jira/browse/BEAM-8941?focusedWorklogId=369902&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-369902
]
ASF GitHub Bot logged work on BEAM-8941:
----------------------------------------
Author: ASF GitHub Bot
Created on: 10/Jan/20 15:56
Start Date: 10/Jan/20 15:56
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_r365295025
##########
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
+ private def _fanout
+
+ //internal usage
+ private SDK i_sdk
+ private Runner i_runner
+ private static final i_dataflowRequired = ["_numWorkers",
"_tempLocation", "_autoscalingAlgorithm"]
+ private static final i_flinkRequired = ["_jobEndpoint",
"_environmentType", "_environmentConfig", "_parallelism"]
+
+ //dataflow only
+ private def _numWorkers
+ private String _tempLocation
+ private String _autoscalingAlgorithm
+
+ //flink only
+ 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 iterations(final int itNumber) { _iterations = itNumber }
+ void fanout(final int fanout) { _fanout = fanout }
+ 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 }
Review comment:
There is also one more important pipeline option which is consumed by every
test (both in Java and Python): `streaming`. Let's set its default value to
False, so we don't have to manually specify it in all batch tests.
----------------------------------------------------------------
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: 369902)
Time Spent: 0.5h (was: 20m)
> 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: 0.5h
> 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)
