opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/persistence/LocalSnapshotStore.java

   1 /*
   2  * Copyright (c) 2017 Brocade Communications Systems, Inc. and others.  All rights reserved.
   3  *
   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
   7  */
   8 package org.opendaylight.controller.cluster.persistence;
   9
  10 import static com.google.common.base.Preconditions.checkArgument;
  11
  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;
  25 import java.io.File;
  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.net.URLDecoder;
  32 import java.net.URLEncoder;
  33 import java.nio.charset.StandardCharsets;
  34 import java.nio.file.Files;
  35 import java.nio.file.StandardCopyOption;
  36 import java.util.ArrayDeque;
  37 import java.util.Arrays;
  38 import java.util.Collection;
  39 import java.util.Collections;
  40 import java.util.Deque;
  41 import java.util.List;
  42 import java.util.Optional;
  43 import java.util.concurrent.Callable;
  44 import java.util.stream.Collector;
  45 import java.util.stream.Collectors;
  46 import java.util.stream.Stream;
  47 import org.eclipse.jdt.annotation.Nullable;
  48 import org.opendaylight.controller.cluster.io.InputOutputStreamFactory;
  49 import org.slf4j.Logger;
  50 import org.slf4j.LoggerFactory;
  51 import scala.concurrent.ExecutionContext;
  52 import scala.concurrent.Future;
  53
  54 /**
  55  * Akka SnapshotStore implementation backed by the local file system. This class was patterned after akka's
  56  * LocalSnapshotStore class and exists because akka's version serializes to a byte[] before persisting
  57  * to the file which will fail if the data reaches or exceeds Integer.MAX_VALUE in size. This class avoids that issue
  58  * by serializing the data directly to the file.
  59  *
  60  * @author Thomas Pantelis
  61  */
  62 public class LocalSnapshotStore extends SnapshotStore {
  63     private static final Logger LOG = LoggerFactory.getLogger(LocalSnapshotStore.class);
  64     private static final int PERSISTENCE_ID_START_INDEX = "snapshot-".length();
  65
  66     private final InputOutputStreamFactory streamFactory;
  67     private final ExecutionContext executionContext;
  68     private final int maxLoadAttempts;
  69     private final File snapshotDir;
  70
  71     public LocalSnapshotStore(final Config config) {
  72         executionContext = context().system().dispatchers().lookup(config.getString("stream-dispatcher"));
  73         snapshotDir = new File(config.getString("dir"));
  74
  75         final int localMaxLoadAttempts = config.getInt("max-load-attempts");
  76         maxLoadAttempts = localMaxLoadAttempts > 0 ? localMaxLoadAttempts : 1;
  77
  78         if (config.getBoolean("use-lz4-compression")) {
  79             final String size = config.getString("lz4-blocksize");
  80             streamFactory = InputOutputStreamFactory.lz4(size);
  81             LOG.debug("Using LZ4 Input/Output Stream, blocksize: {}", size);
  82         } else {
  83             streamFactory = InputOutputStreamFactory.simple();
  84             LOG.debug("Using plain Input/Output Stream");
  85         }
  86
  87         LOG.debug("LocalSnapshotStore ctor: snapshotDir: {}, maxLoadAttempts: {}", snapshotDir, maxLoadAttempts);
  88     }
  89
  90     @Override
  91     public void preStart() throws Exception {
  92         if (!snapshotDir.isDirectory()) {
  93             // Try to create the directory, on failure double check if someone else beat us to it.
  94             if (!snapshotDir.mkdirs() && !snapshotDir.isDirectory()) {
  95                 throw new IOException("Failed to create snapshot directory " + snapshotDir.getCanonicalPath());
  96             }
  97         }
  98
  99         super.preStart();
 100     }
 101
 102     @Override
 103     public Future<Optional<SelectedSnapshot>> doLoadAsync(final String persistenceId,
 104                                                           final SnapshotSelectionCriteria criteria) {
 105         LOG.debug("In doLoadAsync - persistenceId: {}, criteria: {}", persistenceId, criteria);
 106
 107         // Select the youngest 'maxLoadAttempts' snapshots that match the criteria. This may help in situations where
 108         // saving of a snapshot could not be completed because of a JVM crash. Hence, an attempt to load that snapshot
 109         // will fail but loading an older snapshot may succeed.
 110
 111         Deque<SnapshotMetadata> metadatas = getSnapshotMetadatas(persistenceId, criteria).stream()
 112                 .sorted(LocalSnapshotStore::compare).collect(reverse()).stream().limit(maxLoadAttempts)
 113                     .collect(Collectors.toCollection(ArrayDeque::new));
 114
 115         if (metadatas.isEmpty()) {
 116             return Futures.successful(Optional.empty());
 117         }
 118
 119         LOG.debug("doLoadAsync - found: {}", metadatas);
 120
 121         return Futures.future(() -> doLoad(metadatas), executionContext);
 122     }
 123
 124     private Optional<SelectedSnapshot> doLoad(final Deque<SnapshotMetadata> metadatas) throws IOException {
 125         SnapshotMetadata metadata = metadatas.removeFirst();
 126         File file = toSnapshotFile(metadata);
 127
 128         LOG.debug("doLoad {}", file);
 129
 130         try {
 131             Object data = deserialize(file);
 132
 133             LOG.debug("deserialized data: {}", data);
 134
 135             return Optional.of(new SelectedSnapshot(metadata, data));
 136         } catch (IOException e) {
 137             LOG.error("Error loading snapshot file {}, remaining attempts: {}", file, metadatas.size(), e);
 138
 139             if (metadatas.isEmpty()) {
 140                 throw e;
 141             }
 142
 143             return doLoad(metadatas);
 144         }
 145     }
 146
 147     private Object deserialize(final File file) throws IOException {
 148         return JavaSerializer.currentSystem().withValue((ExtendedActorSystem) context().system(),
 149             (Callable<Object>) () -> {
 150                 try (ObjectInputStream in = new ObjectInputStream(streamFactory.createInputStream(file))) {
 151                     return in.readObject();
 152                 } catch (ClassNotFoundException e) {
 153                     throw new IOException("Error loading snapshot file " + file, e);
 154                 } catch (IOException e) {
 155                     LOG.debug("Error loading snapshot file {}", file, e);
 156                     return tryDeserializeAkkaSnapshot(file);
 157                 }
 158             });
 159     }
 160
 161     private Object tryDeserializeAkkaSnapshot(final File file) throws IOException {
 162         LOG.debug("tryDeserializeAkkaSnapshot {}", file);
 163
 164         // The snapshot was probably previously stored via akka's LocalSnapshotStore which wraps the data
 165         // in a Snapshot instance and uses the SnapshotSerializer to serialize it to a byte[]. So we'll use
 166         // the SnapshotSerializer to try to de-serialize it.
 167
 168         SnapshotSerializer snapshotSerializer = new SnapshotSerializer((ExtendedActorSystem) context().system());
 169
 170         try (InputStream in = new BufferedInputStream(new FileInputStream(file))) {
 171             return ((Snapshot)snapshotSerializer.fromBinary(ByteStreams.toByteArray(in))).data();
 172         }
 173     }
 174
 175     @Override
 176     public Future<Void> doSaveAsync(final SnapshotMetadata metadata, final Object snapshot) {
 177         LOG.debug("In doSaveAsync - metadata: {}, snapshot: {}", metadata, snapshot);
 178
 179         return Futures.future(() -> doSave(metadata, snapshot), executionContext);
 180     }
 181
 182     private Void doSave(final SnapshotMetadata metadata, final Object snapshot) throws IOException {
 183         final File actual = toSnapshotFile(metadata);
 184         final File temp = File.createTempFile(actual.getName(), null, snapshotDir);
 185
 186         LOG.debug("Saving to temp file: {}", temp);
 187
 188         try (ObjectOutputStream out = new ObjectOutputStream(streamFactory.createOutputStream(temp))) {
 189             out.writeObject(snapshot);
 190         } catch (IOException e) {
 191             LOG.error("Error saving snapshot file {}. Deleting file..", temp, e);
 192             if (!temp.delete()) {
 193                 LOG.error("Failed to successfully delete file {}", temp);
 194             }
 195             throw e;
 196         }
 197
 198         LOG.debug("Renaming to: {}", actual);
 199         try {
 200             Files.move(temp.toPath(), actual.toPath(), StandardCopyOption.ATOMIC_MOVE);
 201         } catch (IOException e) {
 202             LOG.warn("Failed to move {} to {}. Deleting {}..", temp, actual, temp, e);
 203             if (!temp.delete()) {
 204                 LOG.error("Failed to successfully delete file {}", temp);
 205             }
 206             throw e;
 207         }
 208
 209         return null;
 210     }
 211
 212     @Override
 213     public Future<Void> doDeleteAsync(final SnapshotMetadata metadata) {
 214         LOG.debug("In doDeleteAsync - metadata: {}", metadata);
 215
 216         // Multiple snapshot files here mean that there were multiple snapshots for this seqNr - we delete all of them.
 217         // Usually snapshot-stores would keep one snapshot per sequenceNr however here in the file-based one we
 218         // timestamp snapshots and allow multiple to be kept around (for the same seqNr) if desired.
 219
 220         return Futures.future(() -> doDelete(metadata), executionContext);
 221     }
 222
 223     @Override
 224     public Future<Void> doDeleteAsync(final String persistenceId, final SnapshotSelectionCriteria criteria) {
 225         LOG.debug("In doDeleteAsync - persistenceId: {}, criteria: {}", persistenceId, criteria);
 226
 227         return Futures.future(() -> doDelete(persistenceId, criteria), executionContext);
 228     }
 229
 230     private Void doDelete(final String persistenceId, final SnapshotSelectionCriteria criteria) {
 231         final List<File> files = getSnapshotMetadatas(persistenceId, criteria).stream()
 232                 .flatMap(md -> Stream.of(toSnapshotFile(md))).collect(Collectors.toList());
 233
 234         LOG.debug("Deleting files: {}", files);
 235
 236         files.forEach(file -> {
 237             try {
 238                 Files.delete(file.toPath());
 239             } catch (IOException | SecurityException e) {
 240                 LOG.error("Unable to delete snapshot file: {}, persistenceId: {} ", file, persistenceId);
 241             }
 242         });
 243         return null;
 244     }
 245
 246     private Void doDelete(final SnapshotMetadata metadata) {
 247         final Collection<File> files = getSnapshotFiles(metadata);
 248
 249         LOG.debug("Deleting files: {}", files);
 250
 251         files.forEach(file -> {
 252             try {
 253                 Files.delete(file.toPath());
 254             } catch (IOException | SecurityException e) {
 255                 LOG.error("Unable to delete snapshot file: {}", file);
 256             }
 257         });
 258         return null;
 259     }
 260
 261     private Collection<File> getSnapshotFiles(final String persistenceId) {
 262         String encodedPersistenceId = encode(persistenceId);
 263
 264         File[] files = snapshotDir.listFiles((dir, name) -> {
 265             int persistenceIdEndIndex = name.lastIndexOf('-', name.lastIndexOf('-') - 1);
 266             return PERSISTENCE_ID_START_INDEX + encodedPersistenceId.length() == persistenceIdEndIndex
 267                     && name.startsWith(encodedPersistenceId, PERSISTENCE_ID_START_INDEX) && !name.endsWith(".tmp");
 268         });
 269
 270         if (files == null) {
 271             return Collections.emptyList();
 272         }
 273
 274         if (LOG.isDebugEnabled()) {
 275             LOG.debug("getSnapshotFiles for persistenceId: {}, found files: {}", encodedPersistenceId,
 276                     Arrays.toString(files));
 277         }
 278
 279         return Arrays.asList(files);
 280     }
 281
 282     private Collection<File> getSnapshotFiles(final SnapshotMetadata metadata) {
 283         return getSnapshotFiles(metadata.persistenceId()).stream().filter(file -> {
 284             SnapshotMetadata possible = extractMetadata(file);
 285             return possible != null && possible.sequenceNr() == metadata.sequenceNr()
 286                     && (metadata.timestamp() == 0L || possible.timestamp() == metadata.timestamp());
 287         }).collect(Collectors.toList());
 288     }
 289
 290     private Collection<SnapshotMetadata> getSnapshotMetadatas(final String persistenceId,
 291             final SnapshotSelectionCriteria criteria) {
 292         return getSnapshotFiles(persistenceId).stream().flatMap(file -> toStream(extractMetadata(file)))
 293                 .filter(criteria::matches).collect(Collectors.toList());
 294     }
 295
 296     private static Stream<SnapshotMetadata> toStream(final @Nullable SnapshotMetadata md) {
 297         return md != null ? Stream.of(md) : Stream.empty();
 298     }
 299
 300     private static @Nullable SnapshotMetadata extractMetadata(final File file) {
 301         String name = file.getName();
 302         int sequenceNumberEndIndex = name.lastIndexOf('-');
 303         int persistenceIdEndIndex = name.lastIndexOf('-', sequenceNumberEndIndex - 1);
 304         if (PERSISTENCE_ID_START_INDEX >= persistenceIdEndIndex) {
 305             return null;
 306         }
 307
 308         try {
 309             // Since the persistenceId is url encoded in the filename, we need
 310             // to decode relevant filename's part to obtain persistenceId back
 311             String persistenceId = decode(name.substring(PERSISTENCE_ID_START_INDEX, persistenceIdEndIndex));
 312             long sequenceNumber = Long.parseLong(name.substring(persistenceIdEndIndex + 1, sequenceNumberEndIndex));
 313             long timestamp = Long.parseLong(name.substring(sequenceNumberEndIndex + 1));
 314             return new SnapshotMetadata(persistenceId, sequenceNumber, timestamp);
 315         } catch (NumberFormatException e) {
 316             return null;
 317         }
 318     }
 319
 320     private File toSnapshotFile(final SnapshotMetadata metadata) {
 321         return new File(snapshotDir, String.format("snapshot-%s-%d-%d", encode(metadata.persistenceId()),
 322             metadata.sequenceNr(), metadata.timestamp()));
 323     }
 324
 325     private static <T> Collector<T, ?, List<T>> reverse() {
 326         return Collectors.collectingAndThen(Collectors.toList(), list -> {
 327             Collections.reverse(list);
 328             return list;
 329         });
 330     }
 331
 332     private static String encode(final String str) {
 333         return URLEncoder.encode(str, StandardCharsets.UTF_8);
 334     }
 335
 336     private static String decode(final String str) {
 337         return URLDecoder.decode(str, StandardCharsets.UTF_8);
 338     }
 339
 340     @VisibleForTesting
 341     static int compare(final SnapshotMetadata m1, final SnapshotMetadata m2) {
 342         checkArgument(m1.persistenceId().equals(m2.persistenceId()),
 343                 "Persistence id does not match. id1: %s, id2: %s", m1.persistenceId(), m2.persistenceId());
 344         final int cmp = Long.compare(m1.timestamp(), m2.timestamp());
 345         return cmp != 0 ? cmp : Long.compare(m1.sequenceNr(), m2.sequenceNr());
 346     }
 347 }