[GitHub] [dolphinscheduler] github-code-scanning[bot] commented on a diff in pull request #12843: [Feature][Task Plugin] Add Kubeflow task plugin for MLOps scenario #12166

[GitBox <gi...@apache.org>]

github-code-scanning[bot] commented on code in PR #12843:
URL: https://github.com/apache/dolphinscheduler/pull/12843#discussion_r1018098864


##########
dolphinscheduler-task-plugin/dolphinscheduler-task-kubeflow/src/main/java/org/apache/dolphinscheduler/plugin/kubeflow/KubeflowTask.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.plugin.kubeflow;
+
+import org.apache.dolphinscheduler.common.thread.ThreadUtils;
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
+import org.apache.dolphinscheduler.plugin.task.api.AbstractRemoteTask;
+import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
+import org.apache.dolphinscheduler.plugin.task.api.TaskException;
+import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
+import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import org.apache.dolphinscheduler.plugin.task.api.parser.ParamUtils;
+import org.apache.dolphinscheduler.plugin.task.api.parser.ParameterUtils;
+import org.apache.dolphinscheduler.plugin.task.api.utils.OSUtils;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public class KubeflowTask extends AbstractRemoteTask {
+
+    private final TaskExecutionContext taskExecutionContext;
+    protected KubeflowHelper kubeflowHelper;
+    private KubeflowParameters kubeflowParameters;
+    private Path clusterYAMLPath;
+
+    private Path yamlPath;
+
+    public KubeflowTask(TaskExecutionContext taskExecutionContext) {
+        super(taskExecutionContext);
+        this.taskExecutionContext = taskExecutionContext;
+    }
+
+    public void init() throws TaskException {
+        logger.info("Kubeflow task params {}", taskExecutionContext.getTaskParams());
+        kubeflowParameters = JSONUtils.parseObject(taskExecutionContext.getTaskParams(), KubeflowParameters.class);
+
+        kubeflowParameters.setClusterYAML(taskExecutionContext.getK8sTaskExecutionContext().getConfigYaml());
+        if (!kubeflowParameters.checkParameters()) {
+            throw new TaskException("Kubeflow task params is not valid");
+        }
+
+        writeFiles();
+        kubeflowHelper = new KubeflowHelper(clusterYAMLPath.toString());
+    }
+
+    public void submitApplication() throws TaskException {
+        String command = kubeflowHelper.buildSubmitCommand(yamlPath.toString());
+        logger.info("Kubeflow task submit command: \n{}", command);
+        String message = runCommand(command);
+        logger.info("Kubeflow task submit result: \n{}", message);
+
+        KubeflowHelper.ApplicationIds applicationIds = new KubeflowHelper.ApplicationIds();
+        applicationIds.setAlreadySubmitted(true);
+        setAppIds(JSONUtils.toJsonString(applicationIds));
+    }
+
+    /**
+     * keep checking application status
+     *
+     * @throws TaskException
+     */
+    public void trackApplicationStatus() throws TaskException {

Review Comment:
   ## Missing Override annotation
   
   This method overrides [AbstractRemoteTask.trackApplicationStatus](1); it is advisable to add an Override annotation.
   
   [Show more details](https://github.com/apache/dolphinscheduler/security/code-scanning/2253)



##########
dolphinscheduler-task-plugin/dolphinscheduler-task-kubeflow/src/main/java/org/apache/dolphinscheduler/plugin/kubeflow/KubeflowTask.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.plugin.kubeflow;
+
+import org.apache.dolphinscheduler.common.thread.ThreadUtils;
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
+import org.apache.dolphinscheduler.plugin.task.api.AbstractRemoteTask;
+import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
+import org.apache.dolphinscheduler.plugin.task.api.TaskException;
+import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
+import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import org.apache.dolphinscheduler.plugin.task.api.parser.ParamUtils;
+import org.apache.dolphinscheduler.plugin.task.api.parser.ParameterUtils;
+import org.apache.dolphinscheduler.plugin.task.api.utils.OSUtils;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public class KubeflowTask extends AbstractRemoteTask {
+
+    private final TaskExecutionContext taskExecutionContext;
+    protected KubeflowHelper kubeflowHelper;
+    private KubeflowParameters kubeflowParameters;
+    private Path clusterYAMLPath;
+
+    private Path yamlPath;
+
+    public KubeflowTask(TaskExecutionContext taskExecutionContext) {
+        super(taskExecutionContext);
+        this.taskExecutionContext = taskExecutionContext;
+    }
+
+    public void init() throws TaskException {
+        logger.info("Kubeflow task params {}", taskExecutionContext.getTaskParams());
+        kubeflowParameters = JSONUtils.parseObject(taskExecutionContext.getTaskParams(), KubeflowParameters.class);
+
+        kubeflowParameters.setClusterYAML(taskExecutionContext.getK8sTaskExecutionContext().getConfigYaml());
+        if (!kubeflowParameters.checkParameters()) {
+            throw new TaskException("Kubeflow task params is not valid");
+        }
+
+        writeFiles();
+        kubeflowHelper = new KubeflowHelper(clusterYAMLPath.toString());
+    }
+
+    public void submitApplication() throws TaskException {

Review Comment:
   ## Missing Override annotation
   
   This method overrides [AbstractRemoteTask.submitApplication](1); it is advisable to add an Override annotation.
   
   [Show more details](https://github.com/apache/dolphinscheduler/security/code-scanning/2254)



##########
dolphinscheduler-task-plugin/dolphinscheduler-task-kubeflow/src/main/java/org/apache/dolphinscheduler/plugin/kubeflow/KubeflowTask.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.plugin.kubeflow;
+
+import org.apache.dolphinscheduler.common.thread.ThreadUtils;
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
+import org.apache.dolphinscheduler.plugin.task.api.AbstractRemoteTask;
+import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
+import org.apache.dolphinscheduler.plugin.task.api.TaskException;
+import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
+import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import org.apache.dolphinscheduler.plugin.task.api.parser.ParamUtils;
+import org.apache.dolphinscheduler.plugin.task.api.parser.ParameterUtils;
+import org.apache.dolphinscheduler.plugin.task.api.utils.OSUtils;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public class KubeflowTask extends AbstractRemoteTask {
+
+    private final TaskExecutionContext taskExecutionContext;
+    protected KubeflowHelper kubeflowHelper;
+    private KubeflowParameters kubeflowParameters;
+    private Path clusterYAMLPath;
+
+    private Path yamlPath;
+
+    public KubeflowTask(TaskExecutionContext taskExecutionContext) {
+        super(taskExecutionContext);
+        this.taskExecutionContext = taskExecutionContext;
+    }
+
+    public void init() throws TaskException {

Review Comment:
   ## Missing Override annotation
   
   This method overrides [AbstractTask.init](1); it is advisable to add an Override annotation.
   
   [Show more details](https://github.com/apache/dolphinscheduler/security/code-scanning/2255)



##########
dolphinscheduler-task-plugin/dolphinscheduler-task-kubeflow/src/test/java/org/apache/dolphinscheduler/plugin/kubeflow/KubeflowTaskTest.java:
##########
@@ -0,0 +1,136 @@
+package org.apache.dolphinscheduler.plugin.kubeflow;
+
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
+import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
+import org.apache.dolphinscheduler.plugin.task.api.TaskException;
+import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
+import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContextCacheManager;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.stream.Collectors;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class KubeflowTaskTest {
+
+    public static String clusterConfigName = "clusterConfigYAML.yaml";
+
+    public static String jobConfigName = "jobConfigYAML.yaml";
+
+    public static String readFile(String fileName) throws IOException {
+        String path = KubeflowHelperTest.class.getClassLoader().getResource(fileName).getPath();
+        String content = Files.lines(Paths.get(path), StandardCharsets.UTF_8)
+                .collect(Collectors.joining(System.lineSeparator()));
+
+        return content;
+    }
+
+    @Test
+    public void testInit() throws IOException {
+        KubeflowParameters kubeflowParameters = createKubeflowParameters();
+        KubeflowTask task = createTask(kubeflowParameters);
+        Assertions.assertEquals(kubeflowParameters.getClusterYAML(), task.getParameters().getClusterYAML());
+        Assertions.assertEquals(kubeflowParameters.getYamlContent(), task.getParameters().getYamlContent());
+
+        KubeflowParameters kubeflowParametersError2 = new KubeflowParameters();
+        kubeflowParameters.setYamlContent(readFile(clusterConfigName));
+        Assertions.assertThrows(TaskException.class, () -> {
+            createTask(kubeflowParametersError2);
+        });
+
+    }
+
+    @Test
+    public void TestSubmit() throws IOException {
+        KubeflowParameters kubeflowParameters = createKubeflowParameters();
+        KubeflowTask task = Mockito.spy(createTask(kubeflowParameters));
+        Mockito.when(task.runCommand(Mockito.anyString())).thenReturn("test_result");
+        task.submitApplication();
+        Assertions.assertNotEquals(task.getAppIds(), null);
+        Assertions.assertEquals(task.getExitStatusCode(), TaskConstants.EXIT_CODE_SUCCESS);
+    }
+
+    @Test
+    public void TestTrack() throws IOException {
+        KubeflowParameters kubeflowParameters = createKubeflowParameters();
+
+        TaskExecutionContext taskExecutionContext = createTaskExecutionContext(kubeflowParameters);
+        TestTask task = Mockito.spy(new TestTask(taskExecutionContext));
+        Mockito.when(task.runCommand(Mockito.anyString())).thenReturn("track_result");
+        task.init();
+
+        KubeflowHelper kubeflowHelper = Mockito.mock(KubeflowHelper.class);
+        Mockito.when(kubeflowHelper.buildGetCommand(Mockito.anyString())).thenReturn("");
+        task.setKubeflowHelper(kubeflowHelper);
+
+        Mockito.when(kubeflowHelper.parseGetMessage(Mockito.anyString())).thenReturn("Succeeded");
+        task.trackApplicationStatus();
+        Assertions.assertEquals(task.getExitStatusCode(), TaskConstants.EXIT_CODE_SUCCESS);
+
+        Mockito.when(kubeflowHelper.parseGetMessage(Mockito.anyString())).thenReturn("Failed");
+        task.trackApplicationStatus();
+        Assertions.assertEquals(task.getExitStatusCode(), TaskConstants.EXIT_CODE_FAILURE);
+
+        Mockito.when(kubeflowHelper.parseGetMessage(Mockito.anyString())).thenReturn("", "Succeeded");
+        task.trackApplicationStatus();
+        Assertions.assertEquals(task.getExitStatusCode(), TaskConstants.EXIT_CODE_SUCCESS);
+    }
+
+    @Test
+    public void TestCancel() throws IOException {
+        KubeflowParameters kubeflowParameters = createKubeflowParameters();
+        KubeflowTask task = Mockito.spy(createTask(kubeflowParameters));
+        Mockito.when(task.runCommand(Mockito.anyString())).thenReturn("delete_result");
+        task.cancelApplication();
+        Assertions.assertEquals(task.getExitStatusCode(), TaskConstants.EXIT_CODE_SUCCESS);
+    }
+
+    public KubeflowTask createTask(KubeflowParameters kubeflowParameters) {
+        TaskExecutionContext taskExecutionContext = createTaskExecutionContext(kubeflowParameters);
+        KubeflowTask kubeflowTask = new KubeflowTask(taskExecutionContext);
+        kubeflowTask.init();
+        return kubeflowTask;
+    }
+
+    public TaskExecutionContext createTaskExecutionContext(KubeflowParameters kubeflowParameters) {
+        String parameters = JSONUtils.toJsonString(kubeflowParameters);
+        TaskExecutionContext taskExecutionContext =
+                Mockito.mock(TaskExecutionContext.class, Mockito.RETURNS_DEEP_STUBS);
+        Mockito.when(taskExecutionContext.getTaskParams()).thenReturn(parameters);
+        Mockito.when(taskExecutionContext.getExecutePath()).thenReturn("/tmp/dolphinscheduler/kubeflow");
+        File file = new File("/tmp/dolphinscheduler/kubeflow");
+        if (!file.exists()) {
+            file.mkdirs();
+        }
+        Mockito.when(taskExecutionContext.getK8sTaskExecutionContext().getConfigYaml())
+                .thenReturn(kubeflowParameters.getClusterYAML());
+        TaskExecutionContextCacheManager.cacheTaskExecutionContext(taskExecutionContext);
+
+        return taskExecutionContext;
+    }
+
+    public KubeflowParameters createKubeflowParameters() throws IOException {
+        KubeflowParameters kubeflowParameters = new KubeflowParameters();
+        kubeflowParameters.setClusterYAML(readFile(clusterConfigName));
+        kubeflowParameters.setYamlContent(readFile(jobConfigName));
+        return kubeflowParameters;
+    }
+
+    public class TestTask extends KubeflowTask {

Review Comment:
   ## Inner class could be static
   
   TestTask should be made static, since the enclosing instance is not used.
   
   [Show more details](https://github.com/apache/dolphinscheduler/security/code-scanning/2256)



-- 
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.

To unsubscribe, e-mail: commits-unsubscribe@dolphinscheduler.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org