lasdf1234 commented on code in PR #13563: URL: https://github.com/apache/gravitino/pull/13563#discussion_r4227168913
########## core/src/main/java/org/apache/gravitino/connector/job/JobResourceUtils.java: ########## @@ -0,0 +1,158 @@ +/* + * 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.gravitino.connector.job; + +import java.io.File; +import java.net.URI; +import java.net.URISyntaxException; +import java.util.List; +import java.util.stream.Collectors; +import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.annotation.DeveloperApi; +import org.apache.gravitino.job.JobTemplate; +import org.apache.gravitino.job.ShellJobTemplate; +import org.apache.gravitino.job.SparkJobTemplate; +import org.apache.gravitino.utils.FileFetcher; + +/** + * Utilities to fetch the resources of a runtime job template, that is its executable, scripts, + * jars, files and archives, into a local directory. Job executors that run jobs on the Gravitino + * server use {@link #localizeJobTemplate} to get a job template whose resources are local files + * before launching a job. + */ +@DeveloperApi +public final class JobResourceUtils { + + /** The connect and read timeout in milliseconds used to fetch a remote file. */ + public static final int DEFAULT_FETCH_TIMEOUT_IN_MS = 30 * 1000; + + private JobResourceUtils() {} + + /** + * Localizes a runtime job template: fetches all its resources into a directory, and returns a + * copy of the job template whose resources point to the fetched local files. The other fields of + * the job template are kept as they are. + * + * @param jobTemplate the runtime job template, whose resources are the URIs to fetch + * @param dir the directory to fetch the resources into + * @return the localized job template, whose resources are the paths of the fetched local files + * @throws RuntimeException if a resource cannot be fetched + */ + public static JobTemplate localizeJobTemplate(JobTemplate jobTemplate, File dir) { + if (jobTemplate instanceof ShellJobTemplate) { + ShellJobTemplate shellJobTemplate = (ShellJobTemplate) jobTemplate; + // A command name is looked up in the environment the job runs in, not fetched. + String executable = + isCommandName(shellJobTemplate.executable()) + ? shellJobTemplate.executable() + : fetchFile(shellJobTemplate.executable(), dir, DEFAULT_FETCH_TIMEOUT_IN_MS); Review Comment: **Medium:** Bare `run.sh` as executable does not run a fetched script of the same name. `isCommandName("run.sh")` is true, so the executable is left as a PATH lookup while `scripts` are still fetched into the staging dir. A template such as: ```json {"executable": "run.sh", "scripts": ["https://repo.example.com/jobs/run.sh"]} ``` downloads `run.sh` into the staging directory, but `ShellProcessBuilder` then execs whatever `run.sh` is on the Gravitino server's `PATH`, not `./run.sh` in the staging directory. This is a silent behavior change from earlier versions (relative file under the server cwd). Docs mention the new rule, but the footgun remains: same basename in `executable` + `scripts` looks like it should run the fetched file and does not. Consider rejecting a command-name executable that equals a fetched script basename, or documenting a hard error / requiring `./run.sh` / an absolute URI for file executables at template validate time. ########## core/src/main/java/org/apache/gravitino/connector/job/JobResourceUtils.java: ########## @@ -0,0 +1,158 @@ +/* + * 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.gravitino.connector.job; + +import java.io.File; +import java.net.URI; +import java.net.URISyntaxException; +import java.util.List; +import java.util.stream.Collectors; +import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.annotation.DeveloperApi; +import org.apache.gravitino.job.JobTemplate; +import org.apache.gravitino.job.ShellJobTemplate; +import org.apache.gravitino.job.SparkJobTemplate; +import org.apache.gravitino.utils.FileFetcher; + +/** + * Utilities to fetch the resources of a runtime job template, that is its executable, scripts, + * jars, files and archives, into a local directory. Job executors that run jobs on the Gravitino + * server use {@link #localizeJobTemplate} to get a job template whose resources are local files + * before launching a job. + */ +@DeveloperApi +public final class JobResourceUtils { + + /** The connect and read timeout in milliseconds used to fetch a remote file. */ + public static final int DEFAULT_FETCH_TIMEOUT_IN_MS = 30 * 1000; + + private JobResourceUtils() {} + + /** + * Localizes a runtime job template: fetches all its resources into a directory, and returns a + * copy of the job template whose resources point to the fetched local files. The other fields of + * the job template are kept as they are. + * + * @param jobTemplate the runtime job template, whose resources are the URIs to fetch + * @param dir the directory to fetch the resources into + * @return the localized job template, whose resources are the paths of the fetched local files + * @throws RuntimeException if a resource cannot be fetched + */ + public static JobTemplate localizeJobTemplate(JobTemplate jobTemplate, File dir) { + if (jobTemplate instanceof ShellJobTemplate) { + ShellJobTemplate shellJobTemplate = (ShellJobTemplate) jobTemplate; + // A command name is looked up in the environment the job runs in, not fetched. + String executable = + isCommandName(shellJobTemplate.executable()) + ? shellJobTemplate.executable() + : fetchFile(shellJobTemplate.executable(), dir, DEFAULT_FETCH_TIMEOUT_IN_MS); + return ShellJobTemplate.builder() + .withName(shellJobTemplate.name()) + .withComment(shellJobTemplate.comment()) + .withExecutable(executable) + .withArguments(shellJobTemplate.arguments()) + .withEnvironments(shellJobTemplate.environments()) + .withCustomFields(shellJobTemplate.customFields()) + .withScripts(fetchFiles(shellJobTemplate.scripts(), dir, DEFAULT_FETCH_TIMEOUT_IN_MS)) + .build(); + } + + if (jobTemplate instanceof SparkJobTemplate) { + SparkJobTemplate sparkJobTemplate = (SparkJobTemplate) jobTemplate; + return SparkJobTemplate.builder() + .withName(sparkJobTemplate.name()) + .withComment(sparkJobTemplate.comment()) + .withExecutable( + fetchFile(sparkJobTemplate.executable(), dir, DEFAULT_FETCH_TIMEOUT_IN_MS)) + .withArguments(sparkJobTemplate.arguments()) + .withEnvironments(sparkJobTemplate.environments()) + .withCustomFields(sparkJobTemplate.customFields()) + .withClassName(sparkJobTemplate.className()) + .withJars(fetchFiles(sparkJobTemplate.jars(), dir, DEFAULT_FETCH_TIMEOUT_IN_MS)) + .withFiles(fetchFiles(sparkJobTemplate.files(), dir, DEFAULT_FETCH_TIMEOUT_IN_MS)) + .withArchives(fetchFiles(sparkJobTemplate.archives(), dir, DEFAULT_FETCH_TIMEOUT_IN_MS)) + .withConfigs(sparkJobTemplate.configs()) + .build(); + } + + throw new IllegalArgumentException("Unsupported job type: " + jobTemplate.jobType()); + } + + /** + * Whether the executable of a shell job template is a command name, such as {@code python}: a + * name with no scheme and no path separator, which is looked up in the environment the job runs + * in, for example on the {@code PATH} of the process that launches it, instead of a file to + * fetch. + * + * @param executable the executable of a shell job template + * @return true if the executable is a command name + */ + public static boolean isCommandName(String executable) { + if (StringUtils.isBlank(executable) + || executable.equals(".") + || executable.equals("..") + || executable.indexOf('/') >= 0 + || executable.indexOf(File.separatorChar) >= 0) { + return false; + } + try { + return new URI(executable).getScheme() == null; + } catch (URISyntaxException e) { + return false; + } + } + + /** + * Fetches the files of the given URIs into a directory. + * + * @param uris the URIs of the files + * @param dir the directory to fetch the resources into + * @param timeoutInMs the connect and read timeout in milliseconds for a remote file + * @return the paths of the fetched local files, in the order of the URIs + * @throws RuntimeException if a file cannot be fetched + */ + public static List<String> fetchFiles(List<String> uris, File dir, int timeoutInMs) { + return uris.stream().map(uri -> fetchFile(uri, dir, timeoutInMs)).collect(Collectors.toList()); + } + + /** + * Fetches the file of the given URI into a directory, keeping its file name. + * + * @param uri the URI of the file, a URI without scheme is a local path + * @param dir the directory to fetch the file into + * @param timeoutInMs the connect and read timeout in milliseconds for a remote file + * @return the path of the fetched local file + * @throws RuntimeException if the file cannot be fetched + */ + public static String fetchFile(String uri, File dir, int timeoutInMs) { + try { + URI fileUri = new URI(uri); + File destFile = new File(dir, new File(fileUri.getPath()).getName()); + return FileFetcher.get() + .fetchFileFromUri( + uri, + destFile, + timeoutInMs, + null /* hadoopConf: job file URIs never use the hdfs scheme */); + } catch (Exception e) { + throw new RuntimeException(String.format("Failed to fetch file from URI %s", uri), e); + } Review Comment: **Medium:** Empty basename makes `destFile` equal to the staging directory. For URIs like `http://host/` or `http://host`, `fileUri.getPath()` is `/` or empty, so `new File(path).getName()` is `""` and `new File(dir, "")` is `dir` itself. Fetch then targets the staging directory rather than a file under it (confusing failure, or worse depending on the scheme). Reject a blank file name (or require a non-empty final path segment) before calling `FileFetcher`. ########## core/src/main/java/org/apache/gravitino/connector/job/JobResourceUtils.java: ########## @@ -0,0 +1,158 @@ +/* + * 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.gravitino.connector.job; + +import java.io.File; +import java.net.URI; +import java.net.URISyntaxException; +import java.util.List; +import java.util.stream.Collectors; +import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.annotation.DeveloperApi; +import org.apache.gravitino.job.JobTemplate; +import org.apache.gravitino.job.ShellJobTemplate; +import org.apache.gravitino.job.SparkJobTemplate; +import org.apache.gravitino.utils.FileFetcher; + +/** + * Utilities to fetch the resources of a runtime job template, that is its executable, scripts, + * jars, files and archives, into a local directory. Job executors that run jobs on the Gravitino + * server use {@link #localizeJobTemplate} to get a job template whose resources are local files + * before launching a job. + */ +@DeveloperApi +public final class JobResourceUtils { + + /** The connect and read timeout in milliseconds used to fetch a remote file. */ + public static final int DEFAULT_FETCH_TIMEOUT_IN_MS = 30 * 1000; + + private JobResourceUtils() {} + + /** + * Localizes a runtime job template: fetches all its resources into a directory, and returns a + * copy of the job template whose resources point to the fetched local files. The other fields of + * the job template are kept as they are. + * + * @param jobTemplate the runtime job template, whose resources are the URIs to fetch + * @param dir the directory to fetch the resources into + * @return the localized job template, whose resources are the paths of the fetched local files + * @throws RuntimeException if a resource cannot be fetched + */ + public static JobTemplate localizeJobTemplate(JobTemplate jobTemplate, File dir) { + if (jobTemplate instanceof ShellJobTemplate) { + ShellJobTemplate shellJobTemplate = (ShellJobTemplate) jobTemplate; + // A command name is looked up in the environment the job runs in, not fetched. + String executable = + isCommandName(shellJobTemplate.executable()) + ? shellJobTemplate.executable() + : fetchFile(shellJobTemplate.executable(), dir, DEFAULT_FETCH_TIMEOUT_IN_MS); + return ShellJobTemplate.builder() + .withName(shellJobTemplate.name()) + .withComment(shellJobTemplate.comment()) + .withExecutable(executable) + .withArguments(shellJobTemplate.arguments()) + .withEnvironments(shellJobTemplate.environments()) + .withCustomFields(shellJobTemplate.customFields()) + .withScripts(fetchFiles(shellJobTemplate.scripts(), dir, DEFAULT_FETCH_TIMEOUT_IN_MS)) + .build(); + } + + if (jobTemplate instanceof SparkJobTemplate) { + SparkJobTemplate sparkJobTemplate = (SparkJobTemplate) jobTemplate; + return SparkJobTemplate.builder() + .withName(sparkJobTemplate.name()) + .withComment(sparkJobTemplate.comment()) + .withExecutable( + fetchFile(sparkJobTemplate.executable(), dir, DEFAULT_FETCH_TIMEOUT_IN_MS)) + .withArguments(sparkJobTemplate.arguments()) + .withEnvironments(sparkJobTemplate.environments()) + .withCustomFields(sparkJobTemplate.customFields()) + .withClassName(sparkJobTemplate.className()) + .withJars(fetchFiles(sparkJobTemplate.jars(), dir, DEFAULT_FETCH_TIMEOUT_IN_MS)) + .withFiles(fetchFiles(sparkJobTemplate.files(), dir, DEFAULT_FETCH_TIMEOUT_IN_MS)) + .withArchives(fetchFiles(sparkJobTemplate.archives(), dir, DEFAULT_FETCH_TIMEOUT_IN_MS)) + .withConfigs(sparkJobTemplate.configs()) + .build(); + } + + throw new IllegalArgumentException("Unsupported job type: " + jobTemplate.jobType()); + } + + /** + * Whether the executable of a shell job template is a command name, such as {@code python}: a + * name with no scheme and no path separator, which is looked up in the environment the job runs + * in, for example on the {@code PATH} of the process that launches it, instead of a file to + * fetch. + * + * @param executable the executable of a shell job template + * @return true if the executable is a command name + */ + public static boolean isCommandName(String executable) { + if (StringUtils.isBlank(executable) + || executable.equals(".") + || executable.equals("..") + || executable.indexOf('/') >= 0 + || executable.indexOf(File.separatorChar) >= 0) { + return false; + } + try { + return new URI(executable).getScheme() == null; + } catch (URISyntaxException e) { + return false; + } + } + + /** + * Fetches the files of the given URIs into a directory. + * + * @param uris the URIs of the files + * @param dir the directory to fetch the resources into + * @param timeoutInMs the connect and read timeout in milliseconds for a remote file + * @return the paths of the fetched local files, in the order of the URIs + * @throws RuntimeException if a file cannot be fetched + */ + public static List<String> fetchFiles(List<String> uris, File dir, int timeoutInMs) { + return uris.stream().map(uri -> fetchFile(uri, dir, timeoutInMs)).collect(Collectors.toList()); + } + + /** + * Fetches the file of the given URI into a directory, keeping its file name. + * + * @param uri the URI of the file, a URI without scheme is a local path + * @param dir the directory to fetch the file into + * @param timeoutInMs the connect and read timeout in milliseconds for a remote file + * @return the path of the fetched local file + * @throws RuntimeException if the file cannot be fetched + */ + public static String fetchFile(String uri, File dir, int timeoutInMs) { + try { + URI fileUri = new URI(uri); + File destFile = new File(dir, new File(fileUri.getPath()).getName()); Review Comment: **High:** Destination path is derived only from the URI basename. ```java File destFile = new File(dir, new File(fileUri.getPath()).getName()); ``` Two different resources that share a file name (e.g. `https://a.example/run.sh` and `https://b.example/run.sh`, or a jar and a file both named `app.jar`) silently overwrite each other via `FileFetcher`'s `REPLACE_EXISTING` symlink/move. The job then runs with the wrong file and no error. Please disambiguate destinations (hash of the URI, counter suffix, or reject duplicates) before writing into `dir`. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
