Motivation

The most of open-source distributed systems provide `cluster snapshots` functionality, but the Apache Ignite doesn't have such one. Cluster snapshots will allow users to copy their data from an active cluster and load it later on another, such as copying data from a production system into a smaller QA or development system. 

Management

Create snapshot

[public] Java API

public interface IgniteSnapshot {
    /**
     * @return List of all known snapshots.
     */
    public List<String> getSnapshots();

    /**
     * Create a consistent copy of all persistence cache groups from the whole cluster.
     *
     * @param name Snapshot name.
     * @return Future which will be completed when a process ends.
     */
    public IgniteFuture<Void> createSnapshot(String name);
}

[public] JMX MBean

package org.apache.ignite.mxbean;

/**
 * Snapshot features MBean.
 */
@MXBeanDescription("MBean that provides access for snapshot features.")
public interface SnapshotMXBean {
    /**
     * Gets all created snapshots on the cluster.
     *
     * @return List of all known snapshots.
     */
    @MXBeanDescription("List of all known snapshots.")
    public List<String> getSnapshots();

    /**
     * Create the cluster-wide snapshot with given name.
     *
     * @param snpName Snapshot name to created.
     * @see IgniteSnapshot#createSnapshot(String) (String)
     */
    @MXBeanDescription("Create cluster-wide snapshot.")
    public void createSnapshot(@MXBeanParameter(name = "snpName", description = "Snapshot name.") String snpName);
}

[public] Command Line

# Starts cluster snapshot operation.
control.sh --snapshot ERIB_23012020

# Display all known cluster snapshots.
control.sh --snapshot -list

[internal] File Transmission

Internal API which allows to request and receive the required snapshot of cache groups from a remote. Used as a part of IEP-28: Rebalance peer-2-peer to send created local snapshot to the remote (demander) node.

/**
 * @param parts Collection of pairs group and appropriate cache partition to be snapshot.
 * @param rmtNodeId The remote node to connect to.
 * @param partConsumer Received partition handler.
 * @return Future which will be completed when requested snapshot fully received.
 */
public IgniteInternalFuture<Void> createRemoteSnapshot(
    UUID rmtNodeId,
    Map<Integer, Set<Integer>> parts,
    BiConsumer<File, GroupPartitionId> partConsumer);

Restore snapshot (manually)

The snapshot procedure stores all internal files (binary meta, marshaller meta, cache group data files, and cache group configuration) the same directory structure way as the Apache Ignite does with preserving configured consistent node id.

To restore a cluster from snapshot user must manually do the following:

1. Remove data from the checkpoint, wal, binary_meta, marshaller directories.
2. Copy all snapshot data files to the IGNITE_HOME/work  directory with paying attention to consistent node ids.

maxmuzaf@TYE-SNE-0009931 ignite % tree work
work
└── snapshots
    └── backup23012020
        ├── binary_meta
        │   ├── snapshot_IgniteClusterSnapshotSelfTest0
        │   ├── snapshot_IgniteClusterSnapshotSelfTest1
        │   └── snapshot_IgniteClusterSnapshotSelfTest2
        ├── db
        │   ├── snapshot_IgniteClusterSnapshotSelfTest0
        │   │   ├── cache-default
        │   │   │   ├── cache_data.dat
        │   │   │   ├── part-0.bin
        │   │   │   ├── part-2.bin
        │   │   │   ├── part-3.bin
        │   │   │   ├── part-4.bin
        │   │   │   ├── part-5.bin
        │   │   │   └── part-6.bin
        │   │   └── cache-txCache
        │   │       ├── cache_data.dat
        │   │       ├── part-3.bin
        │   │       ├── part-4.bin
        │   │       └── part-6.bin
        │   ├── snapshot_IgniteClusterSnapshotSelfTest1
        │   │   ├── cache-default
        │   │   │   ├── cache_data.dat
        │   │   │   ├── part-1.bin
        │   │   │   ├── part-3.bin
        │   │   │   ├── part-5.bin
        │   │   │   ├── part-6.bin
        │   │   │   └── part-7.bin
        │   │   └── cache-txCache
        │   │       ├── cache_data.dat
        │   │       ├── part-1.bin
        │   │       ├── part-5.bin
        │   │       └── part-7.bin
        │   └── snapshot_IgniteClusterSnapshotSelfTest2
        │       ├── cache-default
        │       │   ├── cache_data.dat
        │       │   ├── part-0.bin
        │       │   ├── part-1.bin
        │       │   ├── part-2.bin
        │       │   ├── part-4.bin
        │       │   └── part-7.bin
        │       └── cache-txCache
        │           ├── cache_data.dat
        │           ├── part-0.bin
        │           └── part-2.bin
        └── marshaller

17 directories, 30 files

Snapshot requirements

1. Users must have the ability to create snapshot of persisted user data (in-memory is out of the scope).
2. Users must have the ability to create a snapshot from the cluster under the load without cluster deactivation.
3. The snapshot process must not block for a long time any of the user transactions (short-time blocks are acceptable).
4. The snapshot process must allow creating a data snapshot on each node and transfer it to any of the remote nodes for internal cluster needs.
5. The created snapshot at the cluster-level must be fully consistent from cluster-wide terms, there should not be any incomplete transactions inside.
6. The snapshot of each node must be consistent – cache partitions, binary meta, etc. must not have unnecessary changes.

Snapshot process

With respect to the cluster-wide snapshot operation, the process of creating a copy of user data can be split into the following high-level steps:

1. Start a cluster-wide snapshot operation using any of the available public API.
2. Each node will receive such an event and start a local snapshot task which must create a consistent copy of available user data on a local node.
3. Collect the completion results of local snapshot tasks from each node and send the results back to the user.

The Distributed Process is used to complete steps [1, 3]. To achieve the step [2] a new SnapshotFutureTask  must be developed.

Cluster snapshot

To achieve cluster-wide snapshot consistency the Partition-Map-Exchange will be reused to block for a while all user transactions.

In a short amount of time while user transactions are blocked the local snapshot task will be started by forcing the checkpoint process. These actions have the following guarantees: all current transactions are finished and all new transactions are blocked, all data from the PageMemory will be flushed on a disk at checkpoint end. This is a short time-window when all cluster data will be eventually fully consistent on the disk.

The cluster-wide snapshot task steps overview in terms of distributed process:

1. The snapshot distributed process start a new snapshot operation by sending an initial discovery message.
2. The distributed process configured action initiates a new local snapshot task on each cluster node.
3. The discovery event from the distributed process pushes a new exchange task to the exchange worker to start PME.
4. While transactions are blocked (see onDoneBeforeTopologyUnlock) each local node forces the checkpoint thread and waits while an appropriate local snapshot task starts.
5. The distributed process collects completion results from each cluster node:
   1. If there are no execution errors and no baseline nodes left the cluster – the snapshot created successfully.
   2. If there are some errors or some of the cluster nodes left – all local snapshot results will be reverted from each node, the snapshot fails.

Local snapshot

The local snapshot task is an operation that executes on each local node and copies all the persistence user files from the Ignite work directory to the target snapshot directory with additional machinery to achieve consistency. This task is closely connected with the node checkpointing process due to, for instance, cache partition files are only eventually consistent on disk during the ongoing checkpoint process and fully consistent only when the checkpoint ends.

The local snapshot operation on each cluster node reflects as – SnapshotFutureTask .

User data to copy to snapshot

The following must be copied to snapshot:

• cache partition files
• cache configuration
• binary meta information
• marshaller meta information

Base copy strategy

Binary meta, marshaller meta, configurations still stored in on-heap, so it is easy to collect and keep this persistent user information consistent under the checkpoint write-lock (no changes allowed).

Another strategy must be used for cache partition files. The checkpoint process will write dirty pages from PageMemory to the cache partition files simultaneously with an another process copy them to the target directory. Each cache partition file is consistent only at checkpoint end. So, to avoid long-time transaction blocks during the cluster-wide snapshot process it should not wait when checkpoint ends on each node. The process of copying cache partition files must do the following:

1. Start copying each partition file to the target directory as he is. These files will have dirty data due to concurrent checkpoint thread writes.
2. Collect all dirty pages related to ongoing checkpoint process and corresponding partition files and apply them to the copied file right after the copy process ends. These dirty pages will be written to special new .delta files. Each file created per each partition. (e.g. part-1.bin  with part-1.bin.delta  – fully consistent cache partition).

There are two possible cases during copy cache partition files simultaneously with the checkpoint thread:

1. Cache partition file already copied, but the checkpoint still not ended – wait while checkpoint ends and start merging cache partition file with its delta.
2. The current checkpoint process ended, but the cache partition file is still copying – the next checkpoint process must read and copy the old version of a page to delta file prior to writing its dirty page.

Snapshot task process

1. A new checkpoint starts (forced by node or a regular one).
2. Under the checkpoint write lock – fix cache partition length for each partition (copy from 0  - to length ).
3. The task creates new on-heap collections with marshaller meta, binary meta to copy.
4. The task starts copying partition files.
5. The checkpoint thread:
   1. If the associated with task checkpoint is not finished - write a dirty page to the original partition file and to delta file.
   2. If the associated with task checkpoint is finished and partition file still copying – read original page from the original partition file and copy it to the delta file prior to the dirty page write.
6. If partition file is copied – start merging copied partition with its delta file.
7. The task ends then all data successfully copied to the target directory and all cache partition files merged with its deltas.

Remote snapshot

Crash recovery

Limitations

1. The cluster snapshot operation will be stopped, partial snapshots will be reverted if any of the baseline nodes left the cluster.

Risks and Assumptions

// Describe project risks, such as API or binary compatibility issues, major protocol changes, etc.

Discussion Links

http://apache-ignite-developers.2346864.n4.nabble.com/DISCUSSION-Hot-cache-backup-td41034.html

Reference Links

1. Apache Geode – Cache and Region Snapshots 
   https://geode.apache.org/docs/guide/16/managing/cache_snapshots/chapter_overview.html
2. Apache Cassandra – Backing up and restoring data
   https://docs.datastax.com/en/cassandra-oss/3.0/cassandra/operations/opsBackupRestore.html

Tickets

// Links or report with relevant JIRA tickets.