Skip to content

Commit 1889e8e

Browse files
tlrxtlrx
committed
Add BlobContainer.writeBlobAtomic() (#30902)
This commit adds a new writeBlobAtomic() method to the BlobContainer interface that can be implemented by repository implementations which support atomic writes operations. When the BlobContainer implementation does not provide a specific implementation of writeBlobAtomic(), then the writeBlob() method is used. Related to #30680
1 parent 2051947 commit 1889e8e

File tree

13 files changed

+222
-137
lines changed

13 files changed

+222
-137
lines changed

‎buildSrc/src/main/resources/checkstyle_suppressions.xml‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -524,8 +524,6 @@
524524
<suppress files="server[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]cluster[/\\]settings[/\\]ClusterSettingsIT.java" checks="LineLength" />
525525
<suppress files="server[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]cluster[/\\]shards[/\\]ClusterSearchShardsIT.java" checks="LineLength" />
526526
<suppress files="server[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]cluster[/\\]structure[/\\]RoutingIteratorTests.java" checks="LineLength" />
527-
<suppress files="server[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]common[/\\]blobstore[/\\]FsBlobStoreContainerTests.java" checks="LineLength" />
528-
<suppress files="server[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]common[/\\]blobstore[/\\]FsBlobStoreTests.java" checks="LineLength" />
529527
<suppress files="server[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]common[/\\]breaker[/\\]MemoryCircuitBreakerTests.java" checks="LineLength" />
530528
<suppress files="server[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]common[/\\]geo[/\\]ShapeBuilderTests.java" checks="LineLength" />
531529
<suppress files="server[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]common[/\\]hash[/\\]MessageDigestsTests.java" checks="LineLength" />

‎server/src/main/java/org/elasticsearch/common/blobstore/BlobContainer.java‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,29 @@ public interface BlobContainer {
7474
*/
7575
void writeBlob(String blobName, InputStream inputStream, long blobSize) throws IOException;
7676

77+
/**
78+
* Reads blob content from the input stream and writes it to the container in a new blob with the given name,
79+
* using an atomic write operation if the implementation supports it. When the BlobContainer implementation
80+
* does not provide a specific implementation of writeBlobAtomic(String, InputStream, long), then
81+
* the {@link #writeBlob(String, InputStream, long)} method is used.
82+
*
83+
* This method assumes the container does not already contain a blob of the same blobName. If a blob by the
84+
* same name already exists, the operation will fail and an {@link IOException} will be thrown.
85+
*
86+
* @param blobName
87+
* The name of the blob to write the contents of the input stream to.
88+
* @param inputStream
89+
* The input stream from which to retrieve the bytes to write to the blob.
90+
* @param blobSize
91+
* The size of the blob to be written, in bytes. It is implementation dependent whether
92+
* this value is used in writing the blob to the repository.
93+
* @throws FileAlreadyExistsException if a blob by the same name already exists
94+
* @throws IOException if the input stream could not be read, or the target blob could not be written to.
95+
*/
96+
default void writeBlobAtomic(final String blobName, final InputStream inputStream, final long blobSize) throws IOException {
97+
writeBlob(blobName, inputStream, blobSize);
98+
}
99+
77100
/**
78101
* Deletes a blob with giving name, if the blob exists. If the blob does not exist,
79102
* this method throws a NoSuchFileException.

‎server/src/main/java/org/elasticsearch/common/blobstore/fs/FsBlobContainer.java‎

Lines changed: 46 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,11 +19,12 @@
1919

2020
package org.elasticsearch.common.blobstore.fs;
2121

22-
import org.elasticsearch.core.internal.io.IOUtils;
22+
import org.elasticsearch.common.UUIDs;
2323
import org.elasticsearch.common.blobstore.BlobMetaData;
2424
import org.elasticsearch.common.blobstore.BlobPath;
2525
import org.elasticsearch.common.blobstore.support.AbstractBlobContainer;
2626
import org.elasticsearch.common.blobstore.support.PlainBlobMetaData;
27+
import org.elasticsearch.core.internal.io.IOUtils;
2728
import org.elasticsearch.core.internal.io.Streams;
2829

2930
import java.io.BufferedInputStream;
@@ -56,8 +57,9 @@
5657
*/
5758
public class FsBlobContainer extends AbstractBlobContainer {
5859

59-
protected final FsBlobStore blobStore;
60+
private static final String TEMP_FILE_PREFIX = "pending-";
6061

62+
protected final FsBlobStore blobStore;
6163
protected final Path path;
6264

6365
public FsBlobContainer(FsBlobStore blobStore, BlobPath blobPath, Path path) {
@@ -131,6 +133,48 @@ public void writeBlob(String blobName, InputStream inputStream, long blobSize) t
131133
IOUtils.fsync(path, true);
132134
}
133135

136+
@Override
137+
public void writeBlobAtomic(final String blobName, final InputStream inputStream, final long blobSize) throws IOException {
138+
final String tempBlob = tempBlobName(blobName);
139+
final Path tempBlobPath = path.resolve(tempBlob);
140+
try {
141+
try (OutputStream outputStream = Files.newOutputStream(tempBlobPath, StandardOpenOption.CREATE_NEW)) {
142+
Streams.copy(inputStream, outputStream);
143+
}
144+
IOUtils.fsync(tempBlobPath, false);
145+
146+
final Path blobPath = path.resolve(blobName);
147+
// If the target file exists then Files.move() behaviour is implementation specific
148+
// the existing file might be replaced or this method fails by throwing an IOException.
149+
if (Files.exists(blobPath)) {
150+
throw new FileAlreadyExistsException("blob [" + blobPath + "] already exists, cannot overwrite");
151+
}
152+
Files.move(tempBlobPath, blobPath, StandardCopyOption.ATOMIC_MOVE);
153+
} catch (IOException ex) {
154+
try {
155+
deleteBlobIgnoringIfNotExists(tempBlob);
156+
} catch (IOException e) {
157+
ex.addSuppressed(e);
158+
}
159+
throw ex;
160+
} finally {
161+
IOUtils.fsync(path, true);
162+
}
163+
}
164+
165+
public static String tempBlobName(final String blobName) {
166+
return "pending-" + blobName + "-" + UUIDs.randomBase64UUID();
167+
}
168+
169+
/**
170+
* Returns true if the blob is a leftover temporary blob.
171+
*
172+
* The temporary blobs might be left after failed atomic write operation.
173+
*/
174+
public static boolean isTempBlobName(final String blobName) {
175+
return blobName.startsWith(TEMP_FILE_PREFIX);
176+
}
177+
134178
@Override
135179
public void move(String source, String target) throws IOException {
136180
Path sourcePath = path.resolve(source);

‎server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java‎

Lines changed: 4 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,7 @@
5050
import org.elasticsearch.common.blobstore.BlobMetaData;
5151
import org.elasticsearch.common.blobstore.BlobPath;
5252
import org.elasticsearch.common.blobstore.BlobStore;
53+
import org.elasticsearch.common.blobstore.fs.FsBlobContainer;
5354
import org.elasticsearch.common.bytes.BytesArray;
5455
import org.elasticsearch.common.bytes.BytesReference;
5556
import org.elasticsearch.common.collect.Tuple;
@@ -555,10 +556,8 @@ public String startVerification() {
555556
String blobName = "master.dat";
556557
BytesArray bytes = new BytesArray(testBytes);
557558
try (InputStream stream = bytes.streamInput()) {
558-
testContainer.writeBlob(blobName + "-temp", stream, bytes.length());
559+
testContainer.writeBlobAtomic(blobName, stream, bytes.length());
559560
}
560-
// Make sure that move is supported
561-
testContainer.move(blobName + "-temp", blobName);
562561
return seed;
563562
}
564563
} catch (IOException exp) {
@@ -774,18 +773,8 @@ private long listBlobsToGetLatestIndexId() throws IOException {
774773
}
775774

776775
private void writeAtomic(final String blobName, final BytesReference bytesRef) throws IOException {
777-
final String tempBlobName = "pending-" + blobName + "-" + UUIDs.randomBase64UUID();
778776
try (InputStream stream = bytesRef.streamInput()) {
779-
snapshotsBlobContainer.writeBlob(tempBlobName, stream, bytesRef.length());
780-
snapshotsBlobContainer.move(tempBlobName, blobName);
781-
} catch (IOException ex) {
782-
// temporary blob creation or move failed - try cleaning up
783-
try {
784-
snapshotsBlobContainer.deleteBlobIgnoringIfNotExists(tempBlobName);
785-
} catch (IOException e) {
786-
ex.addSuppressed(e);
787-
}
788-
throw ex;
777+
snapshotsBlobContainer.writeBlobAtomic(blobName, stream, bytesRef.length());
789778
}
790779
}
791780

@@ -955,7 +944,7 @@ protected void finalize(final List<SnapshotFiles> snapshots,
955944
// Delete temporary index files first, as we might otherwise fail in the next step creating the new index file if an earlier
956945
// attempt to write an index file with this generation failed mid-way after creating the temporary file.
957946
for (final String blobName : blobs.keySet()) {
958-
if (indexShardSnapshotsFormat.isTempBlobName(blobName)) {
947+
if (FsBlobContainer.isTempBlobName(blobName)) {
959948
try {
960949
blobContainer.deleteBlobIgnoringIfNotExists(blobName);
961950
} catch (IOException e) {

‎server/src/main/java/org/elasticsearch/repositories/blobstore/ChecksumBlobStoreFormat.java‎

Lines changed: 20 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import org.apache.lucene.index.IndexFormatTooNewException;
2424
import org.apache.lucene.index.IndexFormatTooOldException;
2525
import org.apache.lucene.store.OutputStreamIndexOutput;
26+
import org.elasticsearch.common.CheckedConsumer;
2627
import org.elasticsearch.common.CheckedFunction;
2728
import org.elasticsearch.common.blobstore.BlobContainer;
2829
import org.elasticsearch.common.bytes.BytesArray;
@@ -52,8 +53,6 @@
5253
*/
5354
public class ChecksumBlobStoreFormat<T extends ToXContent> extends BlobStoreFormat<T> {
5455

55-
private static final String TEMP_FILE_PREFIX = "pending-";
56-
5756
private static final XContentType DEFAULT_X_CONTENT_TYPE = XContentType.SMILE;
5857

5958
// The format version
@@ -120,7 +119,7 @@ public T readBlob(BlobContainer blobContainer, String blobName) throws IOExcepti
120119
}
121120

122121
/**
123-
* Writes blob in atomic manner with resolving the blob name using {@link #blobName} and {@link #tempBlobName} methods.
122+
* Writes blob in atomic manner with resolving the blob name using {@link #blobName} method.
124123
* <p>
125124
* The blob will be compressed and checksum will be written if required.
126125
*
@@ -131,20 +130,12 @@ public T readBlob(BlobContainer blobContainer, String blobName) throws IOExcepti
131130
* @param name blob name
132131
*/
133132
public void writeAtomic(T obj, BlobContainer blobContainer, String name) throws IOException {
134-
String blobName = blobName(name);
135-
String tempBlobName = tempBlobName(name);
136-
writeBlob(obj, blobContainer, tempBlobName);
137-
try {
138-
blobContainer.move(tempBlobName, blobName);
139-
} catch (IOException ex) {
140-
// Move failed - try cleaning up
141-
try {
142-
blobContainer.deleteBlob(tempBlobName);
143-
} catch (Exception e) {
144-
ex.addSuppressed(e);
133+
final String blobName = blobName(name);
134+
writeTo(obj, blobName, bytesArray -> {
135+
try (InputStream stream = bytesArray.streamInput()) {
136+
blobContainer.writeBlobAtomic(blobName, stream, bytesArray.length());
145137
}
146-
throw ex;
147-
}
138+
});
148139
}
149140

150141
/**
@@ -157,51 +148,35 @@ public void writeAtomic(T obj, BlobContainer blobContainer, String name) throws
157148
* @param name blob name
158149
*/
159150
public void write(T obj, BlobContainer blobContainer, String name) throws IOException {
160-
String blobName = blobName(name);
161-
writeBlob(obj, blobContainer, blobName);
151+
final String blobName = blobName(name);
152+
writeTo(obj, blobName, bytesArray -> {
153+
try (InputStream stream = bytesArray.streamInput()) {
154+
blobContainer.writeBlob(blobName, stream, bytesArray.length());
155+
}
156+
});
162157
}
163158

164-
/**
165-
* Writes blob in atomic manner without resolving the blobName using using {@link #blobName} method.
166-
* <p>
167-
* The blob will be compressed and checksum will be written if required.
168-
*
169-
* @param obj object to be serialized
170-
* @param blobContainer blob container
171-
* @param blobName blob name
172-
*/
173-
protected void writeBlob(T obj, BlobContainer blobContainer, String blobName) throws IOException {
174-
BytesReference bytes = write(obj);
175-
try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream()) {
159+
private void writeTo(final T obj, final String blobName, final CheckedConsumer<BytesArray, IOException> consumer) throws IOException {
160+
final BytesReference bytes = write(obj);
161+
try (ByteArrayOutputStream outputStream = new ByteArrayOutputStream()) {
176162
final String resourceDesc = "ChecksumBlobStoreFormat.writeBlob(blob=\"" + blobName + "\")";
177-
try (OutputStreamIndexOutput indexOutput = new OutputStreamIndexOutput(resourceDesc, blobName, byteArrayOutputStream, BUFFER_SIZE)) {
163+
try (OutputStreamIndexOutput indexOutput = new OutputStreamIndexOutput(resourceDesc, blobName, outputStream, BUFFER_SIZE)) {
178164
CodecUtil.writeHeader(indexOutput, codec, VERSION);
179165
try (OutputStream indexOutputOutputStream = new IndexOutputOutputStream(indexOutput) {
180166
@Override
181167
public void close() throws IOException {
182168
// this is important since some of the XContentBuilders write bytes on close.
183169
// in order to write the footer we need to prevent closing the actual index input.
184-
} }) {
170+
}
171+
}) {
185172
bytes.writeTo(indexOutputOutputStream);
186173
}
187174
CodecUtil.writeFooter(indexOutput);
188175
}
189-
BytesArray bytesArray = new BytesArray(byteArrayOutputStream.toByteArray());
190-
try (InputStream stream = bytesArray.streamInput()) {
191-
blobContainer.writeBlob(blobName, stream, bytesArray.length());
192-
}
176+
consumer.accept(new BytesArray(outputStream.toByteArray()));
193177
}
194178
}
195179

196-
/**
197-
* Returns true if the blob is a leftover temporary blob.
198-
*
199-
* The temporary blobs might be left after failed atomic write operation.
200-
*/
201-
public boolean isTempBlobName(String blobName) {
202-
return blobName.startsWith(ChecksumBlobStoreFormat.TEMP_FILE_PREFIX);
203-
}
204-
205180
protected BytesReference write(T obj) throws IOException {
206181
try (BytesStreamOutput bytesStreamOutput = new BytesStreamOutput()) {
207182
if (compress) {
@@ -222,10 +197,4 @@ protected void write(T obj, StreamOutput streamOutput) throws IOException {
222197
builder.endObject();
223198
}
224199
}
225-
226-
227-
protected String tempBlobName(String name) {
228-
return TEMP_FILE_PREFIX + String.format(Locale.ROOT, blobNameFormat, name);
229-
}
230-
231200
}

‎server/src/test/java/org/elasticsearch/common/blobstore/fs/FsBlobContainerTests.java‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
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.common.blobstore.fs;
20+
21+
import org.elasticsearch.test.ESTestCase;
22+
23+
import static org.hamcrest.Matchers.containsString;
24+
import static org.hamcrest.Matchers.is;
25+
import static org.hamcrest.Matchers.startsWith;
26+
27+
public class FsBlobContainerTests extends ESTestCase {
28+
29+
public void testTempBlobName() {
30+
final String blobName = randomAlphaOfLengthBetween(1, 20);
31+
final String tempBlobName = FsBlobContainer.tempBlobName(blobName);
32+
assertThat(tempBlobName, startsWith("pending-"));
33+
assertThat(tempBlobName, containsString(blobName));
34+
}
35+
36+
public void testIsTempBlobName() {
37+
final String tempBlobName = FsBlobContainer.tempBlobName(randomAlphaOfLengthBetween(1, 20));
38+
assertThat(FsBlobContainer.isTempBlobName(tempBlobName), is(true));
39+
}
40+
}

‎server/src/test/java/org/elasticsearch/common/blobstore/FsBlobStoreContainerTests.java‎ renamed to ‎server/src/test/java/org/elasticsearch/common/blobstore/fs/FsBlobStoreContainerTests.java‎

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -16,23 +16,27 @@
1616
* specific language governing permissions and limitations
1717
* under the License.
1818
*/
19-
package org.elasticsearch.common.blobstore;
19+
package org.elasticsearch.common.blobstore.fs;
2020

2121
import org.apache.lucene.util.LuceneTestCase;
22-
import org.elasticsearch.common.blobstore.fs.FsBlobStore;
22+
import org.elasticsearch.common.blobstore.BlobStore;
2323
import org.elasticsearch.common.settings.Settings;
2424
import org.elasticsearch.common.unit.ByteSizeUnit;
2525
import org.elasticsearch.common.unit.ByteSizeValue;
2626
import org.elasticsearch.repositories.ESBlobStoreContainerTestCase;
2727

2828
import java.io.IOException;
29-
import java.nio.file.Path;
3029

3130
@LuceneTestCase.SuppressFileSystems("ExtrasFS")
3231
public class FsBlobStoreContainerTests extends ESBlobStoreContainerTestCase {
32+
3333
protected BlobStore newBlobStore() throws IOException {
34-
Path tempDir = createTempDir();
35-
Settings settings = randomBoolean() ? Settings.EMPTY : Settings.builder().put("buffer_size", new ByteSizeValue(randomIntBetween(1, 100), ByteSizeUnit.KB)).build();
36-
return new FsBlobStore(settings, tempDir);
34+
final Settings settings;
35+
if (randomBoolean()) {
36+
settings = Settings.builder().put("buffer_size", new ByteSizeValue(randomIntBetween(1, 100), ByteSizeUnit.KB)).build();
37+
} else {
38+
settings = Settings.EMPTY;
39+
}
40+
return new FsBlobStore(settings, createTempDir());
3741
}
3842
}

‎server/src/test/java/org/elasticsearch/common/blobstore/FsBlobStoreTests.java‎ renamed to ‎server/src/test/java/org/elasticsearch/common/blobstore/fs/FsBlobStoreTests.java‎

Lines changed: 12 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,12 @@
1616
* specific language governing permissions and limitations
1717
* under the License.
1818
*/
19-
package org.elasticsearch.common.blobstore;
19+
package org.elasticsearch.common.blobstore.fs;
2020

2121
import org.apache.lucene.util.LuceneTestCase;
22-
import org.elasticsearch.common.blobstore.fs.FsBlobStore;
22+
import org.elasticsearch.common.blobstore.BlobContainer;
23+
import org.elasticsearch.common.blobstore.BlobPath;
24+
import org.elasticsearch.common.blobstore.BlobStore;
2325
import org.elasticsearch.common.bytes.BytesArray;
2426
import org.elasticsearch.common.settings.Settings;
2527
import org.elasticsearch.common.unit.ByteSizeUnit;
@@ -32,10 +34,15 @@
3234

3335
@LuceneTestCase.SuppressFileSystems("ExtrasFS")
3436
public class FsBlobStoreTests extends ESBlobStoreTestCase {
37+
3538
protected BlobStore newBlobStore() throws IOException {
36-
Path tempDir = createTempDir();
37-
Settings settings = randomBoolean() ? Settings.EMPTY : Settings.builder().put("buffer_size", new ByteSizeValue(randomIntBetween(1, 100), ByteSizeUnit.KB)).build();
38-
return new FsBlobStore(settings, tempDir);
39+
final Settings settings;
40+
if (randomBoolean()) {
41+
settings = Settings.builder().put("buffer_size", new ByteSizeValue(randomIntBetween(1, 100), ByteSizeUnit.KB)).build();
42+
} else {
43+
settings = Settings.EMPTY;
44+
}
45+
return new FsBlobStore(settings, createTempDir());
3946
}
4047

4148
public void testReadOnly() throws Exception {

0 commit comments

Comments
 (0)