From b9f3be6a99855c9b0e99da8e9ffd6a85e9fd3993 Mon Sep 17 00:00:00 2001 From: Finn Carroll Date: Mon, 17 Jun 2024 18:28:34 +0000 Subject: [PATCH] Add perms for remote snapshot cache eviction on scripted query Signed-off-by: Finn Carroll --- .../store/remote/utils/TransferManager.java | 68 +++++++++---------- 1 file changed, 34 insertions(+), 34 deletions(-) diff --git a/server/src/main/java/org/opensearch/index/store/remote/utils/TransferManager.java b/server/src/main/java/org/opensearch/index/store/remote/utils/TransferManager.java index 98cad7bfadb09..db290785a181d 100644 --- a/server/src/main/java/org/opensearch/index/store/remote/utils/TransferManager.java +++ b/server/src/main/java/org/opensearch/index/store/remote/utils/TransferManager.java @@ -56,14 +56,20 @@ public IndexInput fetchBlob(BlobFetchRequest blobFetchRequest) throws IOExceptio final Path key = blobFetchRequest.getFilePath(); - final CachedIndexInput cacheEntry = fileCache.compute(key, (path, cachedIndexInput) -> { - if (cachedIndexInput == null || cachedIndexInput.isClosed()) { - // Doesn't exist or is closed, either way create a new one - return new DelayedCreationCachedIndexInput(fileCache, blobContainer, blobFetchRequest); - } else { - // already in the cache and ready to be used (open) - return cachedIndexInput; - } + // We need to do a privileged action here in order to fetch from remote + // and write/evict from local file cache in case this is invoked as a side + // effect of a plugin (such as a scripted search) that doesn't have the + // necessary permissions. + final CachedIndexInput cacheEntry = AccessController.doPrivileged((PrivilegedAction) () -> { + return fileCache.compute(key, (path, cachedIndexInput) -> { + if (cachedIndexInput == null || cachedIndexInput.isClosed()) { + // Doesn't exist or is closed, either way create a new one + return new DelayedCreationCachedIndexInput(fileCache, blobContainer, blobFetchRequest); + } else { + // already in the cache and ready to be used (open) + return cachedIndexInput; + } + }); }); // Cache entry was either retrieved from the cache or newly added, either @@ -78,36 +84,30 @@ public IndexInput fetchBlob(BlobFetchRequest blobFetchRequest) throws IOExceptio @SuppressWarnings("removal") private static FileCachedIndexInput createIndexInput(FileCache fileCache, BlobContainer blobContainer, BlobFetchRequest request) { - // We need to do a privileged action here in order to fetch from remote - // and write to the local file cache in case this is invoked as a side - // effect of a plugin (such as a scripted search) that doesn't have the - // necessary permissions. - return AccessController.doPrivileged((PrivilegedAction) () -> { - try { - if (Files.exists(request.getFilePath()) == false) { - try ( - OutputStream fileOutputStream = Files.newOutputStream(request.getFilePath()); - OutputStream localFileOutputStream = new BufferedOutputStream(fileOutputStream) - ) { - for (BlobFetchRequest.BlobPart blobPart : request.blobParts()) { - try ( - InputStream snapshotFileInputStream = blobContainer.readBlob( - blobPart.getBlobName(), - blobPart.getPosition(), - blobPart.getLength() - ); - ) { - snapshotFileInputStream.transferTo(localFileOutputStream); - } + try { + if (Files.exists(request.getFilePath()) == false) { + try ( + OutputStream fileOutputStream = Files.newOutputStream(request.getFilePath()); + OutputStream localFileOutputStream = new BufferedOutputStream(fileOutputStream) + ) { + for (BlobFetchRequest.BlobPart blobPart : request.blobParts()) { + try ( + InputStream snapshotFileInputStream = blobContainer.readBlob( + blobPart.getBlobName(), + blobPart.getPosition(), + blobPart.getLength() + ); + ) { + snapshotFileInputStream.transferTo(localFileOutputStream); } } } - final IndexInput luceneIndexInput = request.getDirectory().openInput(request.getFileName(), IOContext.READ); - return new FileCachedIndexInput(fileCache, request.getFilePath(), luceneIndexInput); - } catch (IOException e) { - throw new UncheckedIOException(e); } - }); + final IndexInput luceneIndexInput = request.getDirectory().openInput(request.getFileName(), IOContext.READ); + return new FileCachedIndexInput(fileCache, request.getFilePath(), luceneIndexInput); + } catch (IOException e) { + throw new UncheckedIOException(e); + } } /**