BoYiZhang
4 years ago
committed by
GitHub
35 changed files with 1053 additions and 295 deletions
@ -0,0 +1,78 @@ |
|||||||
|
/* |
||||||
|
* 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. |
||||||
|
*/ |
||||||
|
|
||||||
|
package org.apache.dolphinscheduler.common.task; |
||||||
|
|
||||||
|
import java.util.Map; |
||||||
|
|
||||||
|
public class TaskParams { |
||||||
|
|
||||||
|
private String rawScript; |
||||||
|
private Map<String, String>[] localParams; |
||||||
|
|
||||||
|
public void setRawScript(String rawScript) { |
||||||
|
this.rawScript = rawScript; |
||||||
|
} |
||||||
|
|
||||||
|
public void setLocalParams(Map<String, String>[] localParams) { |
||||||
|
this.localParams = localParams; |
||||||
|
} |
||||||
|
|
||||||
|
public String getRawScript() { |
||||||
|
return rawScript; |
||||||
|
} |
||||||
|
|
||||||
|
public void setLocalParamValue(String prop, Object value) { |
||||||
|
if (localParams == null || value == null) { |
||||||
|
return; |
||||||
|
} |
||||||
|
for (int i = 0; i < localParams.length; i++) { |
||||||
|
if (localParams[i].get("prop").equals(prop)) { |
||||||
|
localParams[i].put("value", (String)value); |
||||||
|
} |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
public void setLocalParamValue(Map<String, Object> propToValue) { |
||||||
|
if (localParams == null || propToValue == null) { |
||||||
|
return; |
||||||
|
} |
||||||
|
for (int i = 0; i < localParams.length; i++) { |
||||||
|
String prop = localParams[i].get("prop"); |
||||||
|
if (propToValue.containsKey(prop)) { |
||||||
|
localParams[i].put("value",(String)propToValue.get(prop)); |
||||||
|
} |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
public String getLocalParamValue(String prop) { |
||||||
|
if (localParams == null) { |
||||||
|
return null; |
||||||
|
} |
||||||
|
for (int i = 0; i < localParams.length; i++) { |
||||||
|
String tmpProp = localParams[i].get("prop"); |
||||||
|
if (tmpProp.equals(prop)) { |
||||||
|
return localParams[i].get("value"); |
||||||
|
} |
||||||
|
} |
||||||
|
return null; |
||||||
|
} |
||||||
|
|
||||||
|
public Map<String, String>[] getLocalParams() { |
||||||
|
return localParams; |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,124 @@ |
|||||||
|
/* |
||||||
|
* 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. |
||||||
|
*/ |
||||||
|
|
||||||
|
package org.apache.dolphinscheduler.common.utils; |
||||||
|
|
||||||
|
import org.apache.dolphinscheduler.common.model.TaskNode; |
||||||
|
import org.apache.dolphinscheduler.common.task.TaskParams; |
||||||
|
|
||||||
|
import java.text.ParseException; |
||||||
|
import java.util.Map; |
||||||
|
|
||||||
|
public class VarPoolUtils { |
||||||
|
/** |
||||||
|
* getTaskNodeLocalParam |
||||||
|
* @param taskNode taskNode |
||||||
|
* @param prop prop |
||||||
|
* @return localParamForProp |
||||||
|
*/ |
||||||
|
public static Object getTaskNodeLocalParam(TaskNode taskNode, String prop) { |
||||||
|
String taskParamsJson = taskNode.getParams(); |
||||||
|
TaskParams taskParams = JSONUtils.parseObject(taskParamsJson, TaskParams.class); |
||||||
|
if (taskParams == null) { |
||||||
|
return null; |
||||||
|
} |
||||||
|
return taskParams.getLocalParamValue(prop); |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* setTaskNodeLocalParams |
||||||
|
* @param taskNode taskNode |
||||||
|
* @param prop LocalParamName |
||||||
|
* @param value LocalParamValue |
||||||
|
*/ |
||||||
|
public static void setTaskNodeLocalParams(TaskNode taskNode, String prop, Object value) { |
||||||
|
String taskParamsJson = taskNode.getParams(); |
||||||
|
TaskParams taskParams = JSONUtils.parseObject(taskParamsJson, TaskParams.class); |
||||||
|
if (taskParams == null) { |
||||||
|
return; |
||||||
|
} |
||||||
|
taskParams.setLocalParamValue(prop, value); |
||||||
|
taskNode.setParams(JSONUtils.toJsonString(taskParams)); |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* setTaskNodeLocalParams |
||||||
|
* @param taskNode taskNode |
||||||
|
* @param propToValue propToValue |
||||||
|
*/ |
||||||
|
public static void setTaskNodeLocalParams(TaskNode taskNode, Map<String, Object> propToValue) { |
||||||
|
String taskParamsJson = taskNode.getParams(); |
||||||
|
TaskParams taskParams = JSONUtils.parseObject(taskParamsJson, TaskParams.class); |
||||||
|
if (taskParams == null) { |
||||||
|
return; |
||||||
|
} |
||||||
|
taskParams.setLocalParamValue(propToValue); |
||||||
|
taskNode.setParams(JSONUtils.toJsonString(taskParams)); |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* convertVarPoolToMap |
||||||
|
* @param propToValue propToValue |
||||||
|
* @param varPool varPool |
||||||
|
* @throws ParseException ParseException |
||||||
|
*/ |
||||||
|
public static void convertVarPoolToMap(Map<String, Object> propToValue, String varPool) throws ParseException { |
||||||
|
if (varPool == null || propToValue == null) { |
||||||
|
return; |
||||||
|
} |
||||||
|
String[] splits = varPool.split("\\$VarPool\\$"); |
||||||
|
for (String kv : splits) { |
||||||
|
String[] kvs = kv.split(","); |
||||||
|
if (kvs.length == 2) { |
||||||
|
propToValue.put(kvs[0], kvs[1]); |
||||||
|
} else { |
||||||
|
throw new ParseException(kv, 2); |
||||||
|
} |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* convertPythonScriptPlaceholders |
||||||
|
* @param rawScript rawScript |
||||||
|
* @return String |
||||||
|
* @throws StringIndexOutOfBoundsException StringIndexOutOfBoundsException |
||||||
|
*/ |
||||||
|
public static String convertPythonScriptPlaceholders(String rawScript) throws StringIndexOutOfBoundsException { |
||||||
|
int len = "${setShareVar(${".length(); |
||||||
|
int scriptStart = 0; |
||||||
|
while ((scriptStart = rawScript.indexOf("${setShareVar(${", scriptStart)) != -1) { |
||||||
|
int start = -1; |
||||||
|
int end = rawScript.indexOf('}', scriptStart + len); |
||||||
|
String prop = rawScript.substring(scriptStart + len, end); |
||||||
|
|
||||||
|
start = rawScript.indexOf(',', end); |
||||||
|
end = rawScript.indexOf(')', start); |
||||||
|
|
||||||
|
String value = rawScript.substring(start + 1, end); |
||||||
|
|
||||||
|
start = rawScript.indexOf('}', start) + 1; |
||||||
|
end = rawScript.length(); |
||||||
|
|
||||||
|
String replaceScript = String.format("print(\"${{setValue({},{})}}\".format(\"%s\",%s))", prop, value); |
||||||
|
|
||||||
|
rawScript = rawScript.substring(0, scriptStart) + replaceScript + rawScript.substring(start, end); |
||||||
|
|
||||||
|
scriptStart += replaceScript.length(); |
||||||
|
} |
||||||
|
return rawScript; |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,73 @@ |
|||||||
|
/* |
||||||
|
* 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. |
||||||
|
*/ |
||||||
|
|
||||||
|
package org.apache.dolphinscheduler.common.utils; |
||||||
|
|
||||||
|
import org.apache.dolphinscheduler.common.model.TaskNode; |
||||||
|
|
||||||
|
import java.util.concurrent.ConcurrentHashMap; |
||||||
|
|
||||||
|
import org.junit.Assert; |
||||||
|
import org.junit.Test; |
||||||
|
import org.slf4j.Logger; |
||||||
|
import org.slf4j.LoggerFactory; |
||||||
|
|
||||||
|
public class VarPoolUtilsTest { |
||||||
|
|
||||||
|
private static final Logger logger = LoggerFactory.getLogger(VarPoolUtilsTest.class); |
||||||
|
|
||||||
|
@Test |
||||||
|
public void testSetTaskNodeLocalParams() { |
||||||
|
String taskJson = "{\"conditionResult\":\"{\\\"successNode\\\":[\\\"\\\"],\\\"failedNode\\\":[\\\"\\\"]}\"," |
||||||
|
+ "\"conditionsTask\":false,\"depList\":[],\"dependence\":\"{}\",\"forbidden\":false,\"id\":\"tasks-75298\",\"maxRetryTimes\":0,\"name\":\"a1\"," |
||||||
|
+ "\"params\":\"{\\\"rawScript\\\":\\\"print(\\\\\\\"this is python task \\\\\\\",${p0})\\\"," |
||||||
|
+ "\\\"localParams\\\":[{\\\"prop\\\":\\\"p1\\\",\\\"direct\\\":\\\"IN\\\",\\\"type\\\":\\\"VARCHAR\\\",\\\"value\\\":\\\"1\\\"}]," |
||||||
|
+ "\\\"resourceList\\\":[]}\",\"preTasks\":\"[]\",\"retryInterval\":1,\"runFlag\":\"NORMAL\",\"taskInstancePriority\":\"MEDIUM\"," |
||||||
|
+ "\"taskTimeoutParameter\":{\"enable\":false,\"interval\":0},\"timeout\":\"{\\\"enable\\\":false,\\\"strategy\\\":\\\"\\\"}\"," |
||||||
|
+ "\"type\":\"PYTHON\",\"workerGroup\":\"default\"}"; |
||||||
|
TaskNode taskNode = JSONUtils.parseObject(taskJson, TaskNode.class); |
||||||
|
|
||||||
|
VarPoolUtils.setTaskNodeLocalParams(taskNode, "p1", "test1"); |
||||||
|
Assert.assertEquals(VarPoolUtils.getTaskNodeLocalParam(taskNode, "p1"), "test1"); |
||||||
|
|
||||||
|
ConcurrentHashMap<String, Object> propToValue = new ConcurrentHashMap<String, Object>(); |
||||||
|
propToValue.put("p1", "test2"); |
||||||
|
|
||||||
|
VarPoolUtils.setTaskNodeLocalParams(taskNode, propToValue); |
||||||
|
Assert.assertEquals(VarPoolUtils.getTaskNodeLocalParam(taskNode, "p1"), "test2"); |
||||||
|
} |
||||||
|
|
||||||
|
@Test |
||||||
|
public void testConvertVarPoolToMap() throws Exception { |
||||||
|
String varPool = "p1,66$VarPool$p2,69$VarPool$"; |
||||||
|
ConcurrentHashMap<String, Object> propToValue = new ConcurrentHashMap<String, Object>(); |
||||||
|
VarPoolUtils.convertVarPoolToMap(propToValue, varPool); |
||||||
|
Assert.assertEquals((String)propToValue.get("p1"), "66"); |
||||||
|
Assert.assertEquals((String)propToValue.get("p2"), "69"); |
||||||
|
logger.info(propToValue.toString()); |
||||||
|
} |
||||||
|
|
||||||
|
@Test |
||||||
|
public void testConvertPythonScriptPlaceholders() throws Exception { |
||||||
|
String rawScript = "print(${p1});\n${setShareVar(${p1},3)};\n${setShareVar(${p2},4)};"; |
||||||
|
rawScript = VarPoolUtils.convertPythonScriptPlaceholders(rawScript); |
||||||
|
Assert.assertEquals(rawScript, "print(${p1});\n" |
||||||
|
+ "print(\"${{setValue({},{})}}\".format(\"p1\",3));\n" |
||||||
|
+ "print(\"${{setValue({},{})}}\".format(\"p2\",4));"); |
||||||
|
logger.info(rawScript); |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,43 @@ |
|||||||
|
/* |
||||||
|
* 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. |
||||||
|
*/ |
||||||
|
|
||||||
|
package org.apache.dolphinscheduler.server.utils; |
||||||
|
|
||||||
|
import org.apache.dolphinscheduler.remote.utils.Host; |
||||||
|
|
||||||
|
import org.junit.Assert; |
||||||
|
import org.junit.Test; |
||||||
|
|
||||||
|
/** |
||||||
|
* host test |
||||||
|
*/ |
||||||
|
public class HostTest { |
||||||
|
|
||||||
|
@Test |
||||||
|
public void testHostWarmUp() { |
||||||
|
Host host = Host.of(("192.158.2.2:22:100:" + (System.currentTimeMillis() - 60 * 5 * 1000))); |
||||||
|
Assert.assertEquals(50, host.getWeight()); |
||||||
|
host = Host.of(("192.158.2.2:22:100:" + (System.currentTimeMillis() - 60 * 10 * 1000))); |
||||||
|
Assert.assertEquals(100, host.getWeight()); |
||||||
|
} |
||||||
|
|
||||||
|
@Test |
||||||
|
public void testHost() { |
||||||
|
Host host = Host.of("192.158.2.2:22"); |
||||||
|
Assert.assertEquals(22, host.getPort()); |
||||||
|
} |
||||||
|
} |
||||||
Loading…
Reference in new issue