INDY-1205: Use RocksDB as a key-value storage

build-scripts/ubuntu-1604/Dockerfile

@@ -1,5 +1,13 @@
 FROM ubuntu:16.04
 
+RUN apt-get update -y && apt-get install -y \
+        apt-transport-https \
+        ca-certificates
+
+RUN apt-key adv --keyserver keyserver.ubuntu.com --recv-keys 68DB5E88 && \
+    echo "deb https://repo.sovrin.org/test/deb xenial rocksdb" >> /etc/apt/sources.list && \
+    apt-get update
+
 RUN apt-get update -y && apt-get install -y \
     # common stuff
         git \
@@ -9,12 +17,20 @@ RUN apt-get update -y && apt-get install -y \
         python3-pip \
         python-setuptools \
         python3-venv \
-    # fmp
+    # fpm
         ruby \
         ruby-dev \
         rubygems \
         gcc \
-        make
+        make \
+    # rocksdb python wrapper
+        libbz2-dev \
+        zlib1g-dev \
+        liblz4-dev \
+        libsnappy-dev \
+        rocksdb=5.8.8
+
+RUN pip3 install -U setuptools
 
 # install fpm
 RUN gem install --no-ri --no-rdoc fpm

build-scripts/ubuntu-1604/build-3rd-parties.sh

@@ -17,8 +17,13 @@ function build_from_pypi {
     PREREM_TMP=prerm-${PACKAGE_NAME}
     cp postinst ${POSTINST_TMP}
     cp prerm ${PREREM_TMP}
-    sed -i 's/{package_name}/python3-'${PACKAGE_NAME}'/' ${POSTINST_TMP}
-    sed -i 's/{package_name}/python3-'${PACKAGE_NAME}'/' ${PREREM_TMP}
+    if [[ ${PACKAGE_NAME} =~ ^python-* ]]; then
+        PACKAGE_NAME_TMP="${PACKAGE_NAME/python-/}"
+    else
+        PACKAGE_NAME_TMP=$PACKAGE_NAME
+    fi
+    sed -i 's/{package_name}/python3-'${PACKAGE_NAME_TMP}'/' ${POSTINST_TMP}
+    sed -i 's/{package_name}/python3-'${PACKAGE_NAME_TMP}'/' ${PREREM_TMP}
 
     fpm --input-type "python" \
         --output-type "deb" \
@@ -50,3 +55,5 @@ build_from_pypi pyzmq 16.0.2
 build_from_pypi intervaltree 2.1.0
 build_from_pypi portalocker 0.5.7
 build_from_pypi sortedcontainers 1.5.7
+build_from_pypi setuptools 38.5.2
+build_from_pypi python-rocksdb 0.6.9

ci/ubuntu.dockerfile

@@ -5,10 +5,19 @@ ARG uid=1000
 ARG user=indy
 ARG venv=venv
 
+RUN echo "deb https://repo.sovrin.org/test/deb xenial rocksdb" >> /etc/apt/sources.list && \
+    apt-get update
+
 RUN apt-get update -y && apt-get install -y \
     python3-nacl \
     libindy-crypto=0.2.0 \
-    libindy=1.3.1~403
+    libindy=1.3.1~403 \
+# rocksdb python wrapper
+    libbz2-dev \
+    zlib1g-dev \
+    liblz4-dev \
+    libsnappy-dev \
+    rocksdb=5.8.8
 
 RUN indy_ci_add_user $uid $user $venv
 

ledger/ledger.py

@@ -10,7 +10,7 @@
 from ledger.tree_hasher import TreeHasher
 from ledger.util import F, ConsistencyVerificationFailed
 from storage.kv_store import KeyValueStorage
-from storage.kv_store_leveldb_int_keys import KeyValueStorageLeveldbIntKeys
+from storage.kv_store_rocksdb_int_keys import KeyValueStorageRocksdbIntKeys
 
 
 class Ledger(ImmutableStore):
@@ -19,7 +19,7 @@ def _defaultStore(dataDir,
                       logName,
                       ensureDurability,
                       open=True) -> KeyValueStorage:
-        return KeyValueStorageLeveldbIntKeys(dataDir, logName, open)
+        return KeyValueStorageRocksdbIntKeys(dataDir, logName, open)
 
     def __init__(self,
                  tree: MerkleTree,
@@ -222,8 +222,11 @@ def reset(self):
     # TODO: rename getAllTxn to get_txn_slice with required parameters frm to
     # add get_txn_all without args.
     def getAllTxn(self, frm: int = None, to: int = None):
-        yield from ((int(seq_no), self.txn_serializer.deserialize(txn))
-                    for seq_no, txn in self._transactionLog.iterator(start=frm, end=to))
+        for seq_no, txn in self._transactionLog.iterator(start=frm, end=to):
+            if to is None or int(seq_no) <= to:
+                yield (int(seq_no), self.txn_serializer.deserialize(txn))
+            else:
+                break
 
     @staticmethod
     def hashToStr(h):

plenum/bls/bls_store.py

@@ -1,5 +1,5 @@
 from common.serializers.serialization import multi_sig_store_serializer
-from plenum.persistence.storage import initKeyValueStorage
+from storage.helper import initKeyValueStorage
 from crypto.bls.bls_multi_signature import MultiSignature
 from typing import Optional
 

plenum/common/constants.py

@@ -141,6 +141,7 @@ class StorageType(IntEnum):
 class KeyValueStorageType(IntEnum):
     Leveldb = 1
     Memory = 2
+    Rocksdb = 3
 
 
 @unique
@@ -165,6 +166,7 @@ class LedgerState(IntEnum):
 HS_FILE = "file"
 HS_MEMORY = "memory"
 HS_LEVELDB = 'leveldb'
+HS_ROCKSDB = 'rocksdb'
 
 PLUGIN_BASE_DIR_PATH = "PluginBaseDirPath"
 POOL_LEDGER_ID = 0

plenum/common/stack_manager.py

@@ -1,11 +1,10 @@
 from abc import abstractmethod, ABCMeta
 from collections import OrderedDict
-import os
 
 from ledger.genesis_txn.genesis_txn_initiator_from_file import GenesisTxnInitiatorFromFile
 from plenum.common.keygen_utils import initRemoteKeys
 from plenum.common.tools import lazy_field
-from plenum.persistence.leveldb_hash_store import LevelDbHashStore
+from storage.helper import initHashStore
 from stp_core.types import HA
 from stp_core.network.exceptions import RemoteNotFound
 from stp_core.common.log import getlogger
@@ -43,8 +42,7 @@ def ledgerFile(self) -> str:
 
     @lazy_field
     def hashStore(self):
-        return LevelDbHashStore(dataDir=self.ledgerLocation,
-                                fileNamePrefix='pool')
+        return initHashStore(self.ledgerLocation, 'pool', self.config)
 
     # noinspection PyTypeChecker
     @lazy_field

plenum/config.py

@@ -5,7 +5,8 @@
 
 import logging
 
-from plenum.common.constants import ClientBootStrategy, HS_FILE, KeyValueStorageType
+from plenum.common.constants import ClientBootStrategy, HS_FILE, HS_LEVELDB, \
+    HS_ROCKSDB, HS_MEMORY, KeyValueStorageType
 from plenum.common.types import PLUGIN_TYPE_STATS_CONSUMER
 
 # Each entry in registry is (stack name, ((host, port), verkey, pubkey))
@@ -59,7 +60,7 @@
 clientBootStrategy = ClientBootStrategy.PoolTxn
 
 hashStore = {
-    "type": HS_FILE
+    "type": HS_LEVELDB
 }
 
 primaryStorage = None

plenum/persistence/client_txn_log.py

@@ -4,7 +4,7 @@
 from plenum.common.has_file_storage import HasFileStorage
 from plenum.common.txn_util import getTxnOrderedFields
 from plenum.common.util import updateFieldsWithSeqNo
-from storage.kv_store_leveldb import KeyValueStorageLeveldb
+from storage.kv_store_rocksdb import KeyValueStorageRocksdb
 
 
 class ClientTxnLog(HasFileStorage):
@@ -17,9 +17,7 @@ def __init__(self, dataLocation):
         self.clientDataLocation = self.dataLocation
         if not os.path.exists(self.clientDataLocation):
             os.makedirs(self.clientDataLocation)
-        # self.transactionLog = TextFileStore(self.clientDataLocation,
-        #                                     "transactions")
-        self.transactionLog = KeyValueStorageLeveldb(
+        self.transactionLog = KeyValueStorageRocksdb(
             self.clientDataLocation, "transactions")
         self.serializer = ledger_txn_serializer
 

plenum/persistence/leveldb_hash_store.py → plenum/persistence/db_hash_store.py

@@ -1,13 +1,18 @@
+import storage.helper
+
 from ledger.hash_stores.hash_store import HashStore
-from storage.kv_store_leveldb import KeyValueStorageLeveldb
+from plenum.common.constants import KeyValueStorageType, HS_LEVELDB, HS_ROCKSDB
 from stp_core.common.log import getlogger
 
 logger = getlogger()
 
 
-class LevelDbHashStore(HashStore):
-    def __init__(self, dataDir, fileNamePrefix=""):
+class DbHashStore(HashStore):
+    def __init__(self, dataDir, fileNamePrefix="", db_type=HS_ROCKSDB):
         self.dataDir = dataDir
+        assert db_type == HS_ROCKSDB or db_type == HS_LEVELDB
+        self.db_type = KeyValueStorageType.Leveldb if db_type == HS_LEVELDB \
+            else KeyValueStorageType.Rocksdb
         self.nodesDb = None
         self.leavesDb = None
         self._leafCount = 0
@@ -76,9 +81,10 @@ def closed(self):
                (self.nodesDb.closed and self.leavesDb.closed)
 
     def open(self):
-        self.nodesDb = KeyValueStorageLeveldb(self.dataDir, self.nodes_db_name)
-        self.leavesDb = KeyValueStorageLeveldb(
-            self.dataDir, self.leaves_db_name)
+        self.nodesDb = storage.helper.initKeyValueStorage(
+            self.db_type, self.dataDir, self.nodes_db_name)
+        self.leavesDb = storage.helper.initKeyValueStorage(
+            self.db_type, self.dataDir, self.leaves_db_name)
         self._leafCount = self.leavesDb.size
 
     def close(self):

plenum/persistence/storage.py

@@ -1,11 +1,8 @@
 from abc import abstractmethod, ABC
 
-from plenum.common.constants import StorageType, KeyValueStorageType
-from plenum.common.exceptions import DataDirectoryNotFound, KeyValueStorageConfigNotFound
+from plenum.common.constants import StorageType
+from plenum.common.exceptions import DataDirectoryNotFound
 from plenum.common.messages.node_messages import Reply
-from storage.kv_in_memory import KeyValueStorageInMemory
-from storage.kv_store import KeyValueStorage
-from storage.kv_store_leveldb import KeyValueStorageLeveldb
 from storage.text_file_store import TextFileStore
 
 
@@ -27,16 +24,6 @@ async def get(self, identifier: str, reqId: int, **kwargs):
         pass
 
 
-def initKeyValueStorage(keyValueType, dataLocation,
-                        keyValueStorageName) -> KeyValueStorage:
-    if keyValueType == KeyValueStorageType.Leveldb:
-        return KeyValueStorageLeveldb(dataLocation, keyValueStorageName)
-    elif keyValueType == KeyValueStorageType.Memory:
-        return KeyValueStorageInMemory()
-    else:
-        raise KeyValueStorageConfigNotFound
-
-
 def initStorage(storageType, name, dataDir=None, config=None):
     if storageType == StorageType.File:
         if dataDir is None:

plenum/server/node.py

@@ -11,17 +11,15 @@
 from intervaltree import IntervalTree
 from ledger.compact_merkle_tree import CompactMerkleTree
 from ledger.genesis_txn.genesis_txn_initiator_from_file import GenesisTxnInitiatorFromFile
-from ledger.hash_stores.file_hash_store import FileHashStore
 from ledger.hash_stores.hash_store import HashStore
-from ledger.hash_stores.memory_hash_store import MemoryHashStore
 from ledger.util import F
 from plenum.bls.bls_bft_factory import create_default_bls_bft_factory
 from plenum.bls.bls_crypto_factory import create_default_bls_crypto_factory
 from plenum.client.wallet import Wallet
 from plenum.common.config_util import getConfig
 from plenum.common.constants import POOL_LEDGER_ID, DOMAIN_LEDGER_ID, \
     CLIENT_BLACKLISTER_SUFFIX, CONFIG_LEDGER_ID, \
-    NODE_BLACKLISTER_SUFFIX, NODE_PRIMARY_STORAGE_SUFFIX, HS_FILE, HS_LEVELDB, \
+    NODE_BLACKLISTER_SUFFIX, NODE_PRIMARY_STORAGE_SUFFIX, \
     TXN_TYPE, LEDGER_STATUS, \
     CLIENT_STACK_SUFFIX, PRIMARY_SELECTION_PREFIX, VIEW_CHANGE_PREFIX, \
     OP_FIELD_NAME, CATCH_UP_PREFIX, NYM, \
@@ -58,9 +56,8 @@
 from plenum.common.util import friendlyEx, getMaxFailures, pop_keys, \
     compare_3PC_keys, get_utc_epoch
 from plenum.common.verifier import DidVerifier
-from plenum.persistence.leveldb_hash_store import LevelDbHashStore
 from plenum.persistence.req_id_to_txn import ReqIdrToTxn
-from plenum.persistence.storage import Storage, initStorage, initKeyValueStorage
+from plenum.persistence.storage import Storage, initStorage
 from plenum.server.blacklister import Blacklister
 from plenum.server.blacklister import SimpleBlacklister
 from plenum.server.client_authn import ClientAuthNr, SimpleAuthNr, CoreAuthNr
@@ -91,6 +88,7 @@
 from plenum.common.config_helper import PNodeConfigHelper
 from state.pruning_state import PruningState
 from state.state import State
+from storage.helper import initKeyValueStorage, initHashStore
 from stp_core.common.log import getlogger
 from stp_core.crypto.signer import Signer
 from stp_core.network.exceptions import RemoteNotFound
@@ -447,9 +445,7 @@ def setup_config_req_handler(self):
         self.register_req_handler(CONFIG_LEDGER_ID, self.configReqHandler)
 
     def getConfigLedger(self):
-        hashStore = LevelDbHashStore(
-            dataDir=self.dataLocation, fileNamePrefix='config')
-        return Ledger(CompactMerkleTree(hashStore=hashStore),
+        return Ledger(CompactMerkleTree(hashStore=self.getHashStore('config')),
                       dataDir=self.dataLocation,
                       fileName=self.config.configTransactionsFile,
                       ensureDurability=self.config.EnsureLedgerDurability)
@@ -717,15 +713,7 @@ def getHashStore(self, name) -> HashStore:
         """
         Create and return a hashStore implementation based on configuration
         """
-        hsConfig = self.config.hashStore['type'].lower()
-        if hsConfig == HS_FILE:
-            return FileHashStore(dataDir=self.dataLocation,
-                                 fileNamePrefix=name)
-        elif hsConfig == HS_LEVELDB:
-            return LevelDbHashStore(dataDir=self.dataLocation,
-                                    fileNamePrefix=name)
-        else:
-            return MemoryHashStore()
+        return initHashStore(self.dataLocation, name, self.config)
 
     def get_new_ledger_manager(self) -> LedgerManager:
         ledger_sync_order = self.ledger_ids
@@ -940,7 +928,7 @@ def onStopping(self):
 
     def closeAllKVStores(self):
         # Clear leveldb lock files
-        logger.debug("{} closing level dbs".format(self), extra={"cli": False})
+        logger.debug("{} closing key-value storages".format(self), extra={"cli": False})
         for ledgerId in self.ledgerManager.ledgerRegistry:
             state = self.getState(ledgerId)
             if state:

plenum/server/pool_manager.py

@@ -14,7 +14,7 @@
 from plenum.common.exceptions import UnsupportedOperation
 from plenum.common.stack_manager import TxnStackManager
 from plenum.common.types import NodeDetail
-from plenum.persistence.storage import initKeyValueStorage
+from storage.helper import initKeyValueStorage
 from plenum.persistence.util import pop_merkle_info
 from plenum.server.pool_req_handler import PoolRequestHandler
 from state.pruning_state import PruningState

plenum/test/plugin/demo_plugin/storage.py

@@ -1,13 +1,15 @@
 from ledger.compact_merkle_tree import CompactMerkleTree
 from plenum.common.ledger import Ledger
-from plenum.persistence.leveldb_hash_store import LevelDbHashStore
-from plenum.persistence.storage import initKeyValueStorage
+from plenum.persistence.db_hash_store import DbHashStore
 from state.pruning_state import PruningState
+from storage.helper import initKeyValueStorage
+from plenum.common.constants import HS_LEVELDB
 
 
 def get_auction_hash_store(data_dir):
-    return LevelDbHashStore(dataDir=data_dir,
-                            fileNamePrefix='auction')
+    return DbHashStore(dataDir=data_dir,
+                       fileNamePrefix='auction',
+                       db_type=HS_LEVELDB)
 
 
 def get_auction_ledger(data_dir, name, hash_store, config):