2 * Copyright (c) 2017 Brocade Communications Systems, Inc. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
8 package org.opendaylight.controller.cluster.persistence;
10 import static com.google.common.base.Preconditions.checkArgument;
12 import akka.actor.ExtendedActorSystem;
13 import akka.dispatch.Futures;
14 import akka.persistence.SelectedSnapshot;
15 import akka.persistence.SnapshotMetadata;
16 import akka.persistence.SnapshotSelectionCriteria;
17 import akka.persistence.serialization.Snapshot;
18 import akka.persistence.serialization.SnapshotSerializer;
19 import akka.persistence.snapshot.japi.SnapshotStore;
20 import akka.serialization.JavaSerializer;
21 import com.google.common.annotations.VisibleForTesting;
22 import com.google.common.io.ByteStreams;
23 import com.typesafe.config.Config;
24 import java.io.BufferedInputStream;
26 import java.io.FileInputStream;
27 import java.io.IOException;
28 import java.io.InputStream;
29 import java.io.ObjectInputStream;
30 import java.io.ObjectOutputStream;
31 import java.io.UnsupportedEncodingException;
32 import java.net.URLDecoder;
33 import java.net.URLEncoder;
34 import java.nio.charset.StandardCharsets;
35 import java.nio.file.Files;
36 import java.nio.file.StandardCopyOption;
37 import java.util.ArrayDeque;
38 import java.util.Arrays;
39 import java.util.Collection;
40 import java.util.Collections;
41 import java.util.Deque;
42 import java.util.List;
43 import java.util.Optional;
44 import java.util.concurrent.Callable;
45 import java.util.stream.Collector;
46 import java.util.stream.Collectors;
47 import java.util.stream.Stream;
48 import org.eclipse.jdt.annotation.Nullable;
49 import org.opendaylight.controller.cluster.io.InputOutputStreamFactory;
50 import org.slf4j.Logger;
51 import org.slf4j.LoggerFactory;
52 import scala.concurrent.ExecutionContext;
53 import scala.concurrent.Future;
56 * Akka SnapshotStore implementation backed by the local file system. This class was patterned after akka's
57 * LocalSnapshotStore class and exists because akka's version serializes to a byte[] before persisting
58 * to the file which will fail if the data reaches or exceeds Integer.MAX_VALUE in size. This class avoids that issue
59 * by serializing the data directly to the file.
61 * @author Thomas Pantelis
63 public class LocalSnapshotStore extends SnapshotStore {
64 private static final Logger LOG = LoggerFactory.getLogger(LocalSnapshotStore.class);
65 private static final int PERSISTENCE_ID_START_INDEX = "snapshot-".length();
67 private final InputOutputStreamFactory streamFactory;
68 private final ExecutionContext executionContext;
69 private final int maxLoadAttempts;
70 private final File snapshotDir;
72 public LocalSnapshotStore(final Config config) {
73 this.executionContext = context().system().dispatchers().lookup(config.getString("stream-dispatcher"));
74 snapshotDir = new File(config.getString("dir"));
76 final int localMaxLoadAttempts = config.getInt("max-load-attempts");
77 maxLoadAttempts = localMaxLoadAttempts > 0 ? localMaxLoadAttempts : 1;
79 if (config.getBoolean("use-lz4-compression")) {
80 final String size = config.getString("lz4-blocksize");
81 streamFactory = InputOutputStreamFactory.lz4(size);
82 LOG.debug("Using LZ4 Input/Output Stream, blocksize: {}", size);
84 streamFactory = InputOutputStreamFactory.simple();
85 LOG.debug("Using plain Input/Output Stream");
88 LOG.debug("LocalSnapshotStore ctor: snapshotDir: {}, maxLoadAttempts: {}", snapshotDir, maxLoadAttempts);
92 public void preStart() throws Exception {
93 if (!snapshotDir.isDirectory()) {
94 // Try to create the directory, on failure double check if someone else beat us to it.
95 if (!snapshotDir.mkdirs() && !snapshotDir.isDirectory()) {
96 throw new IOException("Failed to create snapshot directory " + snapshotDir.getCanonicalPath());
104 public Future<Optional<SelectedSnapshot>> doLoadAsync(final String persistenceId,
105 final SnapshotSelectionCriteria criteria) {
106 LOG.debug("In doLoadAsync - persistenceId: {}, criteria: {}", persistenceId, criteria);
108 // Select the youngest 'maxLoadAttempts' snapshots that match the criteria. This may help in situations where
109 // saving of a snapshot could not be completed because of a JVM crash. Hence, an attempt to load that snapshot
110 // will fail but loading an older snapshot may succeed.
112 Deque<SnapshotMetadata> metadatas = getSnapshotMetadatas(persistenceId, criteria).stream()
113 .sorted(LocalSnapshotStore::compare).collect(reverse()).stream().limit(maxLoadAttempts)
114 .collect(Collectors.toCollection(ArrayDeque::new));
116 if (metadatas.isEmpty()) {
117 return Futures.successful(Optional.empty());
120 LOG.debug("doLoadAsync - found: {}", metadatas);
122 return Futures.future(() -> doLoad(metadatas), executionContext);
125 private Optional<SelectedSnapshot> doLoad(final Deque<SnapshotMetadata> metadatas) throws IOException {
126 SnapshotMetadata metadata = metadatas.removeFirst();
127 File file = toSnapshotFile(metadata);
129 LOG.debug("doLoad {}", file);
132 Object data = deserialize(file);
134 LOG.debug("deserialized data: {}", data);
136 return Optional.of(new SelectedSnapshot(metadata, data));
137 } catch (IOException e) {
138 LOG.error("Error loading snapshot file {}, remaining attempts: {}", file, metadatas.size(), e);
140 if (metadatas.isEmpty()) {
144 return doLoad(metadatas);
148 private Object deserialize(final File file) throws IOException {
149 return JavaSerializer.currentSystem().withValue((ExtendedActorSystem) context().system(),
150 (Callable<Object>) () -> {
151 try (ObjectInputStream in = new ObjectInputStream(streamFactory.createInputStream(file))) {
152 return in.readObject();
153 } catch (ClassNotFoundException e) {
154 throw new IOException("Error loading snapshot file " + file, e);
155 } catch (IOException e) {
156 LOG.debug("Error loading snapshot file {}", file, e);
157 return tryDeserializeAkkaSnapshot(file);
162 private Object tryDeserializeAkkaSnapshot(final File file) throws IOException {
163 LOG.debug("tryDeserializeAkkaSnapshot {}", file);
165 // The snapshot was probably previously stored via akka's LocalSnapshotStore which wraps the data
166 // in a Snapshot instance and uses the SnapshotSerializer to serialize it to a byte[]. So we'll use
167 // the SnapshotSerializer to try to de-serialize it.
169 SnapshotSerializer snapshotSerializer = new SnapshotSerializer((ExtendedActorSystem) context().system());
171 try (InputStream in = new BufferedInputStream(new FileInputStream(file))) {
172 return ((Snapshot)snapshotSerializer.fromBinary(ByteStreams.toByteArray(in))).data();
177 public Future<Void> doSaveAsync(final SnapshotMetadata metadata, final Object snapshot) {
178 LOG.debug("In doSaveAsync - metadata: {}, snapshot: {}", metadata, snapshot);
180 return Futures.future(() -> doSave(metadata, snapshot), executionContext);
183 private Void doSave(final SnapshotMetadata metadata, final Object snapshot) throws IOException {
184 final File actual = toSnapshotFile(metadata);
185 final File temp = File.createTempFile(actual.getName(), null, snapshotDir);
187 LOG.debug("Saving to temp file: {}", temp);
189 try (ObjectOutputStream out = new ObjectOutputStream(streamFactory.createOutputStream(temp))) {
190 out.writeObject(snapshot);
191 } catch (IOException e) {
192 LOG.error("Error saving snapshot file {}. Deleting file..", temp, e);
193 if (!temp.delete()) {
194 LOG.error("Failed to successfully delete file {}", temp);
199 LOG.debug("Renaming to: {}", actual);
201 Files.move(temp.toPath(), actual.toPath(), StandardCopyOption.ATOMIC_MOVE);
202 } catch (IOException e) {
203 LOG.warn("Failed to move {} to {}. Deleting {}..", temp, actual, temp, e);
204 if (!temp.delete()) {
205 LOG.error("Failed to successfully delete file {}", temp);
214 public Future<Void> doDeleteAsync(final SnapshotMetadata metadata) {
215 LOG.debug("In doDeleteAsync - metadata: {}", metadata);
217 // Multiple snapshot files here mean that there were multiple snapshots for this seqNr - we delete all of them.
218 // Usually snapshot-stores would keep one snapshot per sequenceNr however here in the file-based one we
219 // timestamp snapshots and allow multiple to be kept around (for the same seqNr) if desired.
221 return Futures.future(() -> doDelete(metadata), executionContext);
225 public Future<Void> doDeleteAsync(final String persistenceId, final SnapshotSelectionCriteria criteria) {
226 LOG.debug("In doDeleteAsync - persistenceId: {}, criteria: {}", persistenceId, criteria);
228 return Futures.future(() -> doDelete(persistenceId, criteria), executionContext);
231 private Void doDelete(final String persistenceId, final SnapshotSelectionCriteria criteria) {
232 final List<File> files = getSnapshotMetadatas(persistenceId, criteria).stream()
233 .flatMap(md -> Stream.of(toSnapshotFile(md))).collect(Collectors.toList());
235 LOG.debug("Deleting files: {}", files);
237 files.forEach(file -> {
239 Files.delete(file.toPath());
240 } catch (IOException | SecurityException e) {
241 LOG.error("Unable to delete snapshot file: {}, persistenceId: {} ", file, persistenceId);
247 private Void doDelete(final SnapshotMetadata metadata) {
248 final Collection<File> files = getSnapshotFiles(metadata);
250 LOG.debug("Deleting files: {}", files);
252 files.forEach(file -> {
254 Files.delete(file.toPath());
255 } catch (IOException | SecurityException e) {
256 LOG.error("Unable to delete snapshot file: {}", file);
262 private Collection<File> getSnapshotFiles(final String persistenceId) {
263 String encodedPersistenceId = encode(persistenceId);
265 File[] files = snapshotDir.listFiles((dir, name) -> {
266 int persistenceIdEndIndex = name.lastIndexOf('-', name.lastIndexOf('-') - 1);
267 return PERSISTENCE_ID_START_INDEX + encodedPersistenceId.length() == persistenceIdEndIndex
268 && name.startsWith(encodedPersistenceId, PERSISTENCE_ID_START_INDEX) && !name.endsWith(".tmp");
272 return Collections.emptyList();
275 if (LOG.isDebugEnabled()) {
276 LOG.debug("getSnapshotFiles for persistenceId: {}, found files: {}", encodedPersistenceId,
277 Arrays.toString(files));
280 return Arrays.asList(files);
283 private Collection<File> getSnapshotFiles(final SnapshotMetadata metadata) {
284 return getSnapshotFiles(metadata.persistenceId()).stream().filter(file -> {
285 SnapshotMetadata possible = extractMetadata(file);
286 return possible != null && possible.sequenceNr() == metadata.sequenceNr()
287 && (metadata.timestamp() == 0L || possible.timestamp() == metadata.timestamp());
288 }).collect(Collectors.toList());
291 private Collection<SnapshotMetadata> getSnapshotMetadatas(final String persistenceId,
292 final SnapshotSelectionCriteria criteria) {
293 return getSnapshotFiles(persistenceId).stream().flatMap(file -> toStream(extractMetadata(file)))
294 .filter(criteria::matches).collect(Collectors.toList());
297 private static Stream<SnapshotMetadata> toStream(final @Nullable SnapshotMetadata md) {
298 return md != null ? Stream.of(md) : Stream.empty();
301 private static @Nullable SnapshotMetadata extractMetadata(final File file) {
302 String name = file.getName();
303 int sequenceNumberEndIndex = name.lastIndexOf('-');
304 int persistenceIdEndIndex = name.lastIndexOf('-', sequenceNumberEndIndex - 1);
305 if (PERSISTENCE_ID_START_INDEX >= persistenceIdEndIndex) {
310 // Since the persistenceId is url encoded in the filename, we need
311 // to decode relevant filename's part to obtain persistenceId back
312 String persistenceId = decode(name.substring(PERSISTENCE_ID_START_INDEX, persistenceIdEndIndex));
313 long sequenceNumber = Long.parseLong(name.substring(persistenceIdEndIndex + 1, sequenceNumberEndIndex));
314 long timestamp = Long.parseLong(name.substring(sequenceNumberEndIndex + 1));
315 return new SnapshotMetadata(persistenceId, sequenceNumber, timestamp);
316 } catch (NumberFormatException e) {
321 private File toSnapshotFile(final SnapshotMetadata metadata) {
322 return new File(snapshotDir, String.format("snapshot-%s-%d-%d", encode(metadata.persistenceId()),
323 metadata.sequenceNr(), metadata.timestamp()));
326 private static <T> Collector<T, ?, List<T>> reverse() {
327 return Collectors.collectingAndThen(Collectors.toList(), list -> {
328 Collections.reverse(list);
333 private static String encode(final String str) {
335 return URLEncoder.encode(str, StandardCharsets.UTF_8.name());
336 } catch (UnsupportedEncodingException e) {
338 LOG.warn("Error encoding {}", str, e);
343 private static String decode(final String str) {
345 return URLDecoder.decode(str, StandardCharsets.UTF_8.name());
346 } catch (final UnsupportedEncodingException e) {
348 LOG.warn("Error decoding {}", str, e);
354 static int compare(final SnapshotMetadata m1, final SnapshotMetadata m2) {
355 checkArgument(m1.persistenceId().equals(m2.persistenceId()),
356 "Persistence id does not match. id1: %s, id2: %s", m1.persistenceId(), m2.persistenceId());
357 final int cmp = Long.compare(m1.timestamp(), m2.timestamp());
358 return cmp != 0 ? cmp : Long.compare(m1.sequenceNr(), m2.sequenceNr());