Skip to content

Commit 2f637d4

Browse files
original-brownbearoriginal-brownbear
authored
Add Ability to List Child Containers to BlobContainer (#42653)
* Add Ability to List Child Containers to BlobContainer * This is a prerequisite of #42189
1 parent 18e74cb commit 2f637d4

File tree

18 files changed

+400
-26
lines changed

18 files changed

+400
-26
lines changed

‎modules/repository-url/src/main/java/org/elasticsearch/common/blobstore/url/URLBlobContainer.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
package org.elasticsearch.common.blobstore.url;
2121

2222
import org.elasticsearch.common.SuppressForbidden;
23+
import org.elasticsearch.common.blobstore.BlobContainer;
2324
import org.elasticsearch.common.blobstore.BlobMetaData;
2425
import org.elasticsearch.common.blobstore.BlobPath;
2526
import org.elasticsearch.common.blobstore.support.AbstractBlobContainer;
@@ -74,6 +75,11 @@ public Map<String, BlobMetaData> listBlobs() throws IOException {
7475
throw new UnsupportedOperationException("URL repository doesn't support this operation");
7576
}
7677

78+
@Override
79+
public Map<String, BlobContainer> children() throws IOException {
80+
throw new UnsupportedOperationException("URL repository doesn't support this operation");
81+
}
82+
7783
/**
7884
* This operation is not supported by URLBlobContainer
7985
*/

‎plugins/repository-azure/src/main/java/org/elasticsearch/repositories/azure/AzureBlobContainer.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import org.elasticsearch.action.support.GroupedActionListener;
2828
import org.elasticsearch.action.support.PlainActionFuture;
2929
import org.elasticsearch.common.Nullable;
30+
import org.elasticsearch.common.blobstore.BlobContainer;
3031
import org.elasticsearch.common.blobstore.BlobMetaData;
3132
import org.elasticsearch.common.blobstore.BlobPath;
3233
import org.elasticsearch.common.blobstore.support.AbstractBlobContainer;
@@ -169,6 +170,16 @@ public Map<String, BlobMetaData> listBlobs() throws IOException {
169170
return listBlobsByPrefix(null);
170171
}
171172

173+
@Override
174+
public Map<String, BlobContainer> children() throws IOException {
175+
final BlobPath path = path();
176+
try {
177+
return blobStore.children(path);
178+
} catch (URISyntaxException | StorageException e) {
179+
throw new IOException("Failed to list children in path [" + path.buildAsString() + "].", e);
180+
}
181+
}
182+
172183
protected String buildKey(String blobName) {
173184
return keyPath + (blobName == null ? "" : blobName);
174185
}

‎plugins/repository-azure/src/main/java/org/elasticsearch/repositories/azure/AzureBlobStore.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,10 @@
3434
import java.io.InputStream;
3535
import java.net.URISyntaxException;
3636
import java.nio.file.FileAlreadyExistsException;
37+
import java.util.Collections;
3738
import java.util.Map;
39+
import java.util.function.Function;
40+
import java.util.stream.Collectors;
3841

3942
import static java.util.Collections.emptyMap;
4043

@@ -101,6 +104,11 @@ public Map<String, BlobMetaData> listBlobsByPrefix(String keyPath, String prefix
101104
return service.listBlobsByPrefix(clientName, container, keyPath, prefix);
102105
}
103106

107+
public Map<String, BlobContainer> children(BlobPath path) throws URISyntaxException, StorageException {
108+
return Collections.unmodifiableMap(service.children(clientName, container, path).stream().collect(
109+
Collectors.toMap(Function.identity(), name -> new AzureBlobContainer(path.add(name), this))));
110+
}
111+
104112
public void writeBlob(String blobName, InputStream inputStream, long blobSize, boolean failIfAlreadyExists)
105113
throws URISyntaxException, StorageException, FileAlreadyExistsException {
106114
service.writeBlob(this.clientName, container, blobName, inputStream, blobSize, failIfAlreadyExists);

‎plugins/repository-azure/src/main/java/org/elasticsearch/repositories/azure/AzureStorageService.java‎

Lines changed: 34 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -29,15 +29,18 @@
2929
import com.microsoft.azure.storage.blob.BlobInputStream;
3030
import com.microsoft.azure.storage.blob.BlobListingDetails;
3131
import com.microsoft.azure.storage.blob.BlobProperties;
32+
import com.microsoft.azure.storage.blob.CloudBlob;
3233
import com.microsoft.azure.storage.blob.CloudBlobClient;
3334
import com.microsoft.azure.storage.blob.CloudBlobContainer;
35+
import com.microsoft.azure.storage.blob.CloudBlobDirectory;
3436
import com.microsoft.azure.storage.blob.CloudBlockBlob;
3537
import com.microsoft.azure.storage.blob.DeleteSnapshotsOption;
3638
import com.microsoft.azure.storage.blob.ListBlobItem;
3739
import org.apache.logging.log4j.LogManager;
3840
import org.apache.logging.log4j.Logger;
3941
import org.apache.logging.log4j.message.ParameterizedMessage;
4042
import org.elasticsearch.common.blobstore.BlobMetaData;
43+
import org.elasticsearch.common.blobstore.BlobPath;
4144
import org.elasticsearch.common.blobstore.support.PlainBlobMetaData;
4245
import org.elasticsearch.common.collect.Tuple;
4346
import org.elasticsearch.common.settings.Settings;
@@ -54,7 +57,9 @@
5457
import java.security.InvalidKeyException;
5558
import java.util.EnumSet;
5659
import java.util.HashMap;
60+
import java.util.HashSet;
5761
import java.util.Map;
62+
import java.util.Set;
5863
import java.util.function.Supplier;
5964

6065
import static java.util.Collections.emptyMap;
@@ -214,15 +219,40 @@ public Map<String, BlobMetaData> listBlobsByPrefix(String account, String contai
214219
// uri.getPath is of the form /container/keyPath.* and we want to strip off the /container/
215220
// this requires 1 + container.length() + 1, with each 1 corresponding to one of the /
216221
final String blobPath = uri.getPath().substring(1 + container.length() + 1);
217-
final BlobProperties properties = ((CloudBlockBlob) blobItem).getProperties();
218-
final String name = blobPath.substring(keyPath.length());
219-
logger.trace(() -> new ParameterizedMessage("blob url [{}], name [{}], size [{}]", uri, name, properties.getLength()));
220-
blobsBuilder.put(name, new PlainBlobMetaData(name, properties.getLength()));
222+
if (blobItem instanceof CloudBlob) {
223+
final BlobProperties properties = ((CloudBlob) blobItem).getProperties();
224+
final String name = blobPath.substring(keyPath.length());
225+
logger.trace(() -> new ParameterizedMessage("blob url [{}], name [{}], size [{}]", uri, name, properties.getLength()));
226+
blobsBuilder.put(name, new PlainBlobMetaData(name, properties.getLength()));
227+
}
221228
}
222229
});
223230
return Map.copyOf(blobsBuilder);
224231
}
225232

233+
public Set<String> children(String account, String container, BlobPath path) throws URISyntaxException, StorageException {
234+
final var blobsBuilder = new HashSet<String>();
235+
final Tuple<CloudBlobClient, Supplier<OperationContext>> client = client(account);
236+
final CloudBlobContainer blobContainer = client.v1().getContainerReference(container);
237+
final String keyPath = path.buildAsString();
238+
final EnumSet<BlobListingDetails> enumBlobListingDetails = EnumSet.of(BlobListingDetails.METADATA);
239+
240+
SocketAccess.doPrivilegedVoidException(() -> {
241+
for (ListBlobItem blobItem : blobContainer.listBlobs(keyPath, false, enumBlobListingDetails, null, client.v2().get())) {
242+
if (blobItem instanceof CloudBlobDirectory) {
243+
final URI uri = blobItem.getUri();
244+
logger.trace(() -> new ParameterizedMessage("blob url [{}]", uri));
245+
// uri.getPath is of the form /container/keyPath.* and we want to strip off the /container/
246+
// this requires 1 + container.length() + 1, with each 1 corresponding to one of the /.
247+
// Lastly, we add the length of keyPath to the offset to strip this container's path.
248+
final String uriPath = uri.getPath();
249+
blobsBuilder.add(uriPath.substring(1 + container.length() + 1 + keyPath.length(), uriPath.length() - 1));
250+
}
251+
}
252+
});
253+
return Set.copyOf(blobsBuilder);
254+
}
255+
226256
public void writeBlob(String account, String container, String blobName, InputStream inputStream, long blobSize,
227257
boolean failIfAlreadyExists)
228258
throws URISyntaxException, StorageException, FileAlreadyExistsException {

‎plugins/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobContainer.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919

2020
package org.elasticsearch.repositories.gcs;
2121

22+
import org.elasticsearch.common.blobstore.BlobContainer;
2223
import org.elasticsearch.common.blobstore.BlobMetaData;
2324
import org.elasticsearch.common.blobstore.BlobPath;
2425
import org.elasticsearch.common.blobstore.BlobStoreException;
@@ -55,6 +56,11 @@ public Map<String, BlobMetaData> listBlobs() throws IOException {
5556
return blobStore.listBlobs(path);
5657
}
5758

59+
@Override
60+
public Map<String, BlobContainer> children() throws IOException {
61+
return blobStore.listChildren(path());
62+
}
63+
5864
@Override
5965
public Map<String, BlobMetaData> listBlobsByPrefix(String prefix) throws IOException {
6066
return blobStore.listBlobsByPrefix(path, prefix);

‎plugins/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobStore.java‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -142,6 +142,23 @@ Map<String, BlobMetaData> listBlobsByPrefix(String path, String prefix) throws I
142142
return mapBuilder.immutableMap();
143143
}
144144

145+
Map<String, BlobContainer> listChildren(BlobPath path) throws IOException {
146+
final String pathStr = path.buildAsString();
147+
final MapBuilder<String, BlobContainer> mapBuilder = MapBuilder.newMapBuilder();
148+
SocketAccess.doPrivilegedVoidIOException
149+
(() -> client().get(bucketName).list(BlobListOption.currentDirectory(), BlobListOption.prefix(pathStr)).iterateAll().forEach(
150+
blob -> {
151+
if (blob.isDirectory()) {
152+
assert blob.getName().startsWith(pathStr);
153+
final String suffixName = blob.getName().substring(pathStr.length());
154+
if (suffixName.isEmpty() == false) {
155+
mapBuilder.put(suffixName, new GoogleCloudStorageBlobContainer(path.add(suffixName), this));
156+
}
157+
}
158+
}));
159+
return mapBuilder.immutableMap();
160+
}
161+
145162
/**
146163
* Returns true if the blob exists in the specific bucket
147164
*

‎plugins/repository-hdfs/src/main/java/org/elasticsearch/repositories/hdfs/HdfsBlobContainer.java‎

Lines changed: 20 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import org.apache.hadoop.fs.Options.CreateOpts;
2626
import org.apache.hadoop.fs.Path;
2727
import org.elasticsearch.common.Nullable;
28+
import org.elasticsearch.common.blobstore.BlobContainer;
2829
import org.elasticsearch.common.blobstore.BlobMetaData;
2930
import org.elasticsearch.common.blobstore.BlobPath;
3031
import org.elasticsearch.common.blobstore.fs.FsBlobContainer;
@@ -137,11 +138,13 @@ public void writeBlobAtomic(String blobName, InputStream inputStream, long blobS
137138

138139
@Override
139140
public Map<String, BlobMetaData> listBlobsByPrefix(@Nullable final String prefix) throws IOException {
140-
FileStatus[] files = store.execute(fileContext -> (fileContext.util().listStatus(path,
141-
path -> prefix == null || path.getName().startsWith(prefix))));
142-
Map<String, BlobMetaData> map = new LinkedHashMap<String, BlobMetaData>();
141+
FileStatus[] files = store.execute(fileContext -> fileContext.util().listStatus(path,
142+
path -> prefix == null || path.getName().startsWith(prefix)));
143+
Map<String, BlobMetaData> map = new LinkedHashMap<>();
143144
for (FileStatus file : files) {
144-
map.put(file.getPath().getName(), new PlainBlobMetaData(file.getPath().getName(), file.getLen()));
145+
if (file.isFile()) {
146+
map.put(file.getPath().getName(), new PlainBlobMetaData(file.getPath().getName(), file.getLen()));
147+
}
145148
}
146149
return Collections.unmodifiableMap(map);
147150
}
@@ -151,6 +154,19 @@ public Map<String, BlobMetaData> listBlobs() throws IOException {
151154
return listBlobsByPrefix(null);
152155
}
153156

157+
@Override
158+
public Map<String, BlobContainer> children() throws IOException {
159+
FileStatus[] files = store.execute(fileContext -> fileContext.util().listStatus(path));
160+
Map<String, BlobContainer> map = new LinkedHashMap<>();
161+
for (FileStatus file : files) {
162+
if (file.isDirectory()) {
163+
final String name = file.getPath().getName();
164+
map.put(name, new HdfsBlobContainer(path().add(name), store, new Path(path, name), bufferSize, securityContext));
165+
}
166+
}
167+
return Collections.unmodifiableMap(map);
168+
}
169+
154170
/**
155171
* Exists to wrap underlying InputStream methods that might make socket connections in
156172
* doPrivileged blocks. This is due to the way that hdfs client libraries might open

‎plugins/repository-hdfs/src/test/java/org/elasticsearch/repositories/hdfs/HdfsRepositoryTests.java‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
/*
2+
* Licensed to Elasticsearch under one or more contributor
3+
* license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright
5+
* ownership. Elasticsearch licenses this file to you under
6+
* the Apache License, Version 2.0 (the "License"); you may
7+
* not use this file except in compliance with the License.
8+
* You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
package org.elasticsearch.repositories.hdfs;
20+
21+
import com.carrotsearch.randomizedtesting.annotations.ThreadLeakFilters;
22+
import org.elasticsearch.action.support.master.AcknowledgedResponse;
23+
import org.elasticsearch.bootstrap.JavaVersion;
24+
import org.elasticsearch.common.settings.MockSecureSettings;
25+
import org.elasticsearch.common.settings.SecureSettings;
26+
import org.elasticsearch.common.settings.Settings;
27+
import org.elasticsearch.plugins.Plugin;
28+
import org.elasticsearch.repositories.AbstractThirdPartyRepositoryTestCase;
29+
30+
import java.util.Collection;
31+
32+
import static org.hamcrest.Matchers.equalTo;
33+
34+
@ThreadLeakFilters(filters = HdfsClientThreadLeakFilter.class)
35+
public class HdfsRepositoryTests extends AbstractThirdPartyRepositoryTestCase {
36+
37+
@Override
38+
protected Collection<Class<? extends Plugin>> getPlugins() {
39+
return pluginList(HdfsPlugin.class);
40+
}
41+
42+
@Override
43+
protected SecureSettings credentials() {
44+
return new MockSecureSettings();
45+
}
46+
47+
@Override
48+
protected void createRepository(String repoName) {
49+
assumeFalse("https://github.com/elastic/elasticsearch/issues/31498", JavaVersion.current().equals(JavaVersion.parse("11")));
50+
AcknowledgedResponse putRepositoryResponse = client().admin().cluster().preparePutRepository(repoName)
51+
.setType("hdfs")
52+
.setSettings(Settings.builder()
53+
.put("uri", "hdfs:///")
54+
.put("conf.fs.AbstractFileSystem.hdfs.impl", TestingFs.class.getName())
55+
.put("path", "foo")
56+
.put("chunk_size", randomIntBetween(100, 1000) + "k")
57+
.put("compress", randomBoolean())
58+
).get();
59+
assertThat(putRepositoryResponse.isAcknowledged(), equalTo(true));
60+
}
61+
}

‎plugins/repository-s3/build.gradle‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -164,7 +164,7 @@ if (useFixture) {
164164
def minioAddress = {
165165
int minioPort = postProcessFixture.ext."test.fixtures.minio-fixture.tcp.9000"
166166
assert minioPort > 0
167-
return 'http://127.0.0.1:' + minioPort
167+
'http://127.0.0.1:' + minioPort
168168
}
169169

170170
File minioAddressFile = new File(project.buildDir, 'generated-resources/s3Fixture.address')
@@ -173,15 +173,13 @@ if (useFixture) {
173173
// and pass its name instead.
174174
task writeMinioAddress {
175175
dependsOn tasks.bundlePlugin, tasks.postProcessFixture
176-
outputs.file(minioAddressFile)
177176
doLast {
178177
file(minioAddressFile).text = "${ -> minioAddress.call() }"
179178
}
180179
}
181180

182181
thirdPartyTest {
183182
dependsOn writeMinioAddress
184-
inputs.file(minioAddressFile)
185183
systemProperty 'test.s3.endpoint', minioAddressFile.name
186184
}
187185

0 commit comments

Comments
 (0)