jerryshao commented on code in PR #13563: URL: https://github.com/apache/gravitino/pull/13563#discussion_r4227205095
########## 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: Fixed in 01a9b3b2f: `localizeJobTemplate` now checks the file names of all resources before fetching anything, and rejects the job with an `IllegalArgumentException` when two different resources share a file name (the same URI listed twice is still allowed). I chose rejecting over renaming because scripts usually refer to each other by file name. _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ ########## 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: Fixed in 01a9b3b2f: `fetchFile` rejects a URI with no file name (`http://host`, `http://host/`, opaque `file:relative`) with an `IllegalArgumentException` before calling `FileFetcher`, so the staging directory is never the destination. _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ ########## 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: Fixed in 01a9b3b2f: a job is rejected when a fetched resource has the same file name as a command-name executable, with a message telling to use the file's URI or absolute path as the executable. It is checked at run time, when placeholders are resolved, and documented in `manage-jobs-in-gravitino.md`. _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ -- 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]
