mattcasters commented on code in PR #8788: URL: https://github.com/apache/hop/pull/8788#discussion_r4210135242
########## plugins/transforms/watchfiles/src/main/java/org/apache/hop/pipeline/transforms/watchfiles/VfsFileScanner.java: ########## @@ -0,0 +1,259 @@ +/* + * 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.hop.pipeline.transforms.watchfiles; + +import java.io.IOException; +import java.net.URI; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayDeque; +import java.util.Deque; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.function.BooleanSupplier; +import java.util.regex.Pattern; +import org.apache.commons.vfs2.FileObject; +import org.apache.commons.vfs2.FileType; +import org.apache.commons.vfs2.provider.UriParser; +import org.apache.commons.vfs2.provider.local.LocalFile; +import org.apache.hop.core.exception.HopFileException; +import org.apache.hop.core.variables.IVariables; +import org.apache.hop.core.vfs.HopVfs; + +/** A failed or cancelled traversal never returns a partial snapshot. */ +public class VfsFileScanner { + private final FileObject root; + private final boolean recursive; + private final Pattern include; + private final Pattern exclude; + private final int maximumEntries; + private final BooleanSupplier stopped; + + public VfsFileScanner( + FileObject root, + boolean recursive, + String include, + String exclude, + int maximumEntries, + BooleanSupplier stopped) { + this.root = root; + this.recursive = recursive; + this.include = include == null || include.isEmpty() ? null : Pattern.compile(include); + this.exclude = exclude == null || exclude.isEmpty() ? null : Pattern.compile(exclude); + this.maximumEntries = maximumEntries; + this.stopped = stopped; + } + + public Map<String, FileState> snapshot() throws IOException { + root.refresh(); + if (!root.exists() || !root.isFolder() || !root.isReadable()) { + throw new IOException( + "Watch root is missing, unreadable or not a directory: " + HopVfs.getFriendlyURI(root)); + } + Map<String, FileState> files = new HashMap<>(); + Deque<String> directories = new ArrayDeque<>(); + directories.add("."); + Set<String> visited = new HashSet<>(); + while (!directories.isEmpty()) { + try (FileObject directory = root.resolveFile(directories.removeFirst())) { + scan(directory, files, visited, directories); + } + } + return files; + } + + private void scan( + FileObject directory, + Map<String, FileState> files, + Set<String> visited, + Deque<String> directories) + throws IOException { + checkStopped(); + if (!visited.add(directory.getName().getURI())) { + return; + } + if (visited.size() > maximumEntries) { + throw new WatchLimitException("Directory entry limit exceeded. Increase maximum entries."); + } + directory.refresh(); + if (!directory.exists() + || !directory.isFolder() + || !directory.isReadable() + || isLink(directory)) { + throw new IOException("Directory changed or became unreadable during reconciliation."); + } + FileObject[] children = directory.getChildren(); + if (children == null) { + throw new IOException("VFS provider returned no directory listing."); + } + // Close every handle, including children after the one that failed. + Throwable traversalFailure = null; + try { + for (FileObject child : children) { + checkStopped(); + child.refresh(); + if (isLink(child)) { + continue; + } + FileType type = child.getType(); + if (type == FileType.FOLDER && recursive) { + if (!child.isReadable()) { + throw new IOException("Unreadable child directory."); + } + if (visited.size() + directories.size() >= maximumEntries) { + throw new WatchLimitException( + "Directory entry limit exceeded. Increase maximum entries."); + } + directories.add(root.getName().getRelativeName(child.getName())); + } else if (type == FileType.FILE + && matches(UriParser.decode(child.getName().getBaseName()))) { + FileState file = readState(child); + if (files.size() >= maximumEntries) { + throw new WatchLimitException("File entry limit exceeded. Increase maximum entries."); + } + files.put(file.getUri(), file); + } else if (type == FileType.IMAGINARY) { + // Concurrent removal: restart on the next iteration rather than infer mass deletion. + throw new IOException("Directory changed during listing; retry reconciliation."); + } + } + } catch (IOException | RuntimeException e) { + traversalFailure = e; + throw e; + } finally { + IOException failure = null; + for (FileObject child : children) { + try { + child.close(); + } catch (IOException e) { + if (traversalFailure != null) { + traversalFailure.addSuppressed(e); + } else if (failure == null) { + failure = e; + } else { + failure.addSuppressed(e); + } + } + } + if (failure != null) { + throw failure; + } + } + } + + public FileState readRelative(String relative) throws IOException { + checkStopped(); + try (FileObject file = root.resolveFile(relative)) { + file.refresh(); + if (!file.exists() + || file.getType() != FileType.FILE + || isLink(file) + || !matches(UriParser.decode(file.getName().getBaseName()))) { + return null; + } + return readState(file); + } + } + + public String relativeUri(String uri) throws IOException { + try (FileObject resolved = root.resolveFile(uri)) { + return root.getName().getRelativeName(resolved.getName()); + } + } + + public String uri(String relative) throws IOException { + try (FileObject file = root.resolveFile(relative)) { + return file.getName().getURI(); + } + } + + public boolean matches(String name) { + return (include == null || include.matcher(name).matches()) + && (exclude == null || !exclude.matcher(name).matches()); + } + + private FileState readState(FileObject file) throws IOException { + long size = file.getContent().getSize(); + long modified = file.getContent().getLastModifiedTime(); + if (size < 0 || modified < 0) { + throw new IOException("VFS provider must supply a non-negative size and modification time."); + } + String filename = HopVfs.getFilename(file); + if (file instanceof LocalFile && filename.contains("%")) { + // A literal percent in a raw local filename is otherwise interpreted as a VFS escape. + filename = file.getName().getURI(); + } + return new FileState( + file.getName().getURI(), + filename, + UriParser.decode(file.getName().getBaseName()), + file.getName().getParent().getURI(), + file.getName().getScheme(), + size, + modified); + } + + private boolean isLink(FileObject file) throws IOException { + return file.isSymbolicLink() + || (file instanceof LocalFile && Files.isSymbolicLink(localPath(file))); + } + + public static Path localPath(FileObject file) { + // Commons VFS may retain literal Unicode in an otherwise escaped file URI. Unix NIO expects + // its raw path to contain UTF-8 percent escapes; Windows happens to accept the Unicode form. + // VFS can leave these filename characters unescaped; they are illegal in a URI path. + URI uri = + URI.create( Review Comment: **[bug]** `localPath` is on the listing hot path (`isLink` calls it for every local file and directory). Commons VFS 2.10 only percent-encodes `#`, space, and `%` in `getURI()`. This method then encodes `[]{}|^`` and still calls `URI.create`. A legal Unix name containing `?`, `"`, `<`, `>`, `\`, or a newline is left raw. `URI.create` throws `IllegalArgumentException` for most of those, and `Path.of` throws for `?` (`URI has a query component`) and for `#` if it were unescaped. `tick` only catches `IOException`, and `processRow` turns that runtime exception into a transform failure, so one such file stops the watcher. The same `URI.create` pattern is repeated in `resolveFile` for explicit `file:` paths. **Suggestion:** Percent-encode every character that is illegal in a URI path (at least `?`, `"`, `<`, `>`, `\`, and controls) before `URI.create`, or build the `Path` from the decoded VFS path instead of round-tripping the URI. Add a regression with a name such as `report?.csv` and `a<b.txt`. ########## docs/hop-user-manual/modules/ROOT/nav.adoc: ########## @@ -337,6 +337,7 @@ under the License. *** xref:pipeline/transforms/webservices.adoc[Web services lookup] *** xref:pipeline/transforms/workflow-executor.adoc[Workflow Executor] *** xref:pipeline/transforms/workflow-logging.adoc[Workflow logging] +*** xref:pipeline/transforms/watchfiles.adoc[Watch Files] Review Comment: **[nit]** The transform list is alphabetical by title. Watch Files is inserted after Workflow logging and before Write to log, so it sits past Web services lookup and Workflow Executor. **Suggestion:** Move this xref to just before Web services lookup. ########## plugins/transforms/watchfiles/src/main/java/org/apache/hop/pipeline/transforms/watchfiles/LocalFileWatcher.java: ########## @@ -0,0 +1,164 @@ +/* + * 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.hop.pipeline.transforms.watchfiles; + +import static java.nio.file.StandardWatchEventKinds.ENTRY_CREATE; +import static java.nio.file.StandardWatchEventKinds.ENTRY_DELETE; +import static java.nio.file.StandardWatchEventKinds.ENTRY_MODIFY; +import static java.nio.file.StandardWatchEventKinds.OVERFLOW; + +import java.io.IOException; +import java.nio.file.ClosedWatchServiceException; +import java.nio.file.FileVisitResult; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.SimpleFileVisitor; +import java.nio.file.WatchEvent; +import java.nio.file.WatchKey; +import java.nio.file.WatchService; +import java.nio.file.attribute.BasicFileAttributes; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +/** No worker threads. One WatchService for the whole tree; closing it wakes the Hop runner. */ +public class LocalFileWatcher implements FileWatcher { + private final Path root; + private final boolean recursive; + private final int capacity; + private final int maximumDirectories; + private final WatchService service; + private final Map<WatchKey, Path> keys = new HashMap<>(); + private final Set<Path> registered = new HashSet<>(); + private volatile boolean closed; + + public LocalFileWatcher(Path root, boolean recursive, int capacity, int maximumDirectories) + throws IOException { + this.root = root; + this.recursive = recursive; + this.capacity = capacity; + this.maximumDirectories = maximumDirectories; + service = root.getFileSystem().newWatchService(); + try { + reconcileDirectories(); + } catch (IOException | RuntimeException e) { + service.close(); + throw e; + } + } + + @Override + public void reconcileDirectories() throws IOException { + if (closed) { + return; + } + if (!recursive) { + register(root); + return; + } + // No FOLLOW_LINKS: recursive symlinks and junctions cannot create loops. + Files.walkFileTree( + root, + new SimpleFileVisitor<>() { + @Override + public FileVisitResult preVisitDirectory(Path directory, BasicFileAttributes attributes) + throws IOException { + register(directory); + return FileVisitResult.CONTINUE; + } + }); + } + + private void register(Path directory) throws IOException { + if (registered.contains(directory)) { + return; + } + if (registered.size() >= maximumDirectories) { + throw new WatchLimitException("Watch directory limit exceeded. Increase maximum entries."); + } + WatchKey key = directory.register(service, ENTRY_CREATE, ENTRY_MODIFY, ENTRY_DELETE); + keys.put(key, directory); + registered.add(directory); + } + + @Override + public WatchBatch poll(long timeoutMillis) throws IOException, InterruptedException { + Set<Path> paths = new LinkedHashSet<>(); + boolean reconcile = false; + long overflows = 0; + try { + WatchKey key = service.poll(timeoutMillis, TimeUnit.MILLISECONDS); + // Limit keys per iteration so storms cannot starve stability/checkpoints/stop. + int remaining = capacity; + while (key != null && remaining-- > 0 && !closed) { + Path directory = keys.get(key); + for (WatchEvent<?> event : key.pollEvents()) { + if (event.kind() == OVERFLOW) { + overflows++; + reconcile = true; + continue; + } + if (directory == null || !(event.context() instanceof Path relative)) { + reconcile = true; + continue; + } + Path path = directory.resolve(relative).normalize(); + if (Files.isDirectory(path, java.nio.file.LinkOption.NOFOLLOW_LINKS)) { + // Creation/rename of a populated directory needs a scan, not just registration. + if (recursive) { + reconcile = true; + } + } else if (paths.size() < capacity) { + paths.add(path); + } else { + reconcile = true; + } + if (event.kind() == ENTRY_DELETE && registered.contains(path)) { + reconcile = true; + } + } + if (!key.reset()) { + Path removed = keys.remove(key); + registered.remove(removed); + reconcile = true; + } + key = service.poll(); + } + // A key already removed from the service must be processed on the next scan. + if (key != null) { + key.pollEvents(); + key.reset(); Review Comment: **[suggestion]** When the hint cap stops the drain, this already-dequeued key is discarded with `pollEvents()` and `reset()` without using the return value. The in-loop path removes `keys` and `registered` when `reset()` is false; this path does not. If the directory is gone, it stays in `registered`, so `register` will not watch it again after it is recreated, and it still counts toward the directory limit. Reconciliation can still see the files, but native delivery for that directory stays off for the rest of the run. **Suggestion:** Use the same `reset()` handling as the loop: on false, drop the key and its path from `keys` and `registered` before setting `reconcile`. -- 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]
