[FAB-3017] Ledger backup restore

 - Added the genesis block into ledger provider which may help
   in future for addition/restoreing of peer by merely copying
   ledger provider folder (a few MBs of data)
 - Added a test for backup and restore of ledger

The draft documentation for the instructions on backup and restore
can be found at "https://jira.hyperledger.org/browse/FAB-3017#comment-22720"

Change-Id: I6da2bdece4bf7f084dbd4d1da0a89706747b89da
Signed-off-by: manish <manish.sethi@gmail.com>

core/ledger/kvledger/kv_ledger_provider.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 
+	"github.com/golang/protobuf/proto"
 	"github.com/hyperledger/fabric/common/ledger/blkstorage"
 	"github.com/hyperledger/fabric/common/ledger/blkstorage/fsblkstorage"
 	"github.com/hyperledger/fabric/common/ledger/util/leveldbhelper"
@@ -134,7 +135,7 @@ func (provider *Provider) Create(genesisBlock *common.Block) (ledger.PeerLedger,
 		ledger.Close()
 		return nil, err
 	}
-	panicOnErr(provider.idStore.createLedgerID(ledgerID), "Error while marking ledger as created")
+	panicOnErr(provider.idStore.createLedgerID(ledgerID, genesisBlock), "Error while marking ledger as created")
 	return ledger, nil
 }
 
@@ -224,7 +225,9 @@ func (provider *Provider) recoverUnderConstructionLedger() {
 		panicOnErr(provider.idStore.unsetUnderConstructionFlag(), "Error while unsetting under construction flag")
 	case 1:
 		logger.Infof("Genesis block was committed. Hence, marking the peer ledger as created")
-		panicOnErr(provider.idStore.createLedgerID(ledgerID), "Error while adding ledgerID [%s] to created list", ledgerID)
+		genesisBlock, err := ledger.GetBlockByNumber(0)
+		panicOnErr(err, "Error while retrieving genesis block from blockchain for ledger [%s]", ledgerID)
+		panicOnErr(provider.idStore.createLedgerID(ledgerID, genesisBlock), "Error while adding ledgerID [%s] to created list", ledgerID)
 	default:
 		panic(fmt.Errorf(
 			"Data inconsistency: under construction flag is set for ledger [%s] while the height of the blockchain is [%d]",
@@ -281,10 +284,13 @@ func (s *idStore) getUnderConstructionFlag() (string, error) {
 	return string(val), nil
 }
 
-func (s *idStore) createLedgerID(ledgerID string) error {
+func (s *idStore) createLedgerID(ledgerID string, gb *common.Block) error {
 	key := s.encodeLedgerKey(ledgerID)
-	val := []byte{}
-	err := error(nil)
+	var val []byte
+	var err error
+	if val, err = proto.Marshal(gb); err != nil {
+		return err
+	}
 	if val, err = s.db.Get(key); err != nil {
 		return err
 	}

core/ledger/kvledger/kv_ledger_provider_test.go

@@ -18,11 +18,19 @@ package kvledger
 
 import (
 	"fmt"
+	"os"
+	"path/filepath"
 	"testing"
 
 	configtxtest "github.com/hyperledger/fabric/common/configtx/test"
+	"github.com/hyperledger/fabric/common/ledger/blkstorage/fsblkstorage"
 	"github.com/hyperledger/fabric/common/ledger/testutil"
 	"github.com/hyperledger/fabric/core/ledger"
+	"github.com/hyperledger/fabric/core/ledger/ledgerconfig"
+	"github.com/hyperledger/fabric/protos/common"
+	"github.com/hyperledger/fabric/protos/ledger/queryresult"
+	putils "github.com/hyperledger/fabric/protos/utils"
+	"github.com/spf13/viper"
 )
 
 func TestLedgerProvider(t *testing.T) {
@@ -111,6 +119,118 @@ func TestMultipleLedgerBasicRW(t *testing.T) {
 	}
 }
 
+func TestLedgerBackup(t *testing.T) {
+	ledgerid := "TestLedger"
+	originalPath := "/tmp/fabric/ledgertests/kvledger1"
+	restorePath := "/tmp/fabric/ledgertests/kvledger2"
+	viper.Set("ledger.history.enableHistoryDatabase", true)
+
+	// create and populate a ledger in the original environment
+	env := createTestEnv(t, originalPath)
+	provider, _ := NewProvider()
+	bg, gb := testutil.NewBlockGenerator(t, ledgerid, false)
+	gbHash := gb.Header.Hash()
+	ledger, _ := provider.Create(gb)
+
+	simulator, _ := ledger.NewTxSimulator()
+	simulator.SetState("ns1", "key1", []byte("value1"))
+	simulator.SetState("ns1", "key2", []byte("value2"))
+	simulator.SetState("ns1", "key3", []byte("value3"))
+	simulator.Done()
+	simRes, _ := simulator.GetTxSimulationResults()
+	block1 := bg.NextBlock([][]byte{simRes})
+	ledger.Commit(block1)
+
+	simulator, _ = ledger.NewTxSimulator()
+	simulator.SetState("ns1", "key1", []byte("value4"))
+	simulator.SetState("ns1", "key2", []byte("value5"))
+	simulator.SetState("ns1", "key3", []byte("value6"))
+	simulator.Done()
+	simRes, _ = simulator.GetTxSimulationResults()
+	block2 := bg.NextBlock([][]byte{simRes})
+	ledger.Commit(block2)
+
+	ledger.Close()
+	provider.Close()
+
+	// Create restore environment
+	env = createTestEnv(t, restorePath)
+
+	// remove the statedb, historydb, and block indexes (they are suppoed to be auto created during opening of an existing ledger)
+	// and rename the originalPath to restorePath
+	testutil.AssertNoError(t, os.RemoveAll(ledgerconfig.GetStateLevelDBPath()), "")
+	testutil.AssertNoError(t, os.RemoveAll(ledgerconfig.GetHistoryLevelDBPath()), "")
+	testutil.AssertNoError(t, os.RemoveAll(filepath.Join(ledgerconfig.GetBlockStorePath(), fsblkstorage.IndexDir)), "")
+	testutil.AssertNoError(t, os.Rename(originalPath, restorePath), "")
+	defer env.cleanup()
+
+	// Instantiate the ledger from restore environment and this should behave exactly as it would have in the original environment
+	provider, _ = NewProvider()
+	defer provider.Close()
+
+	_, err := provider.Create(gb)
+	testutil.AssertEquals(t, err, ErrLedgerIDExists)
+
+	ledger, _ = provider.Open(ledgerid)
+	defer ledger.Close()
+
+	block1Hash := block1.Header.Hash()
+	block2Hash := block2.Header.Hash()
+	bcInfo, _ := ledger.GetBlockchainInfo()
+	testutil.AssertEquals(t, bcInfo, &common.BlockchainInfo{
+		Height: 3, CurrentBlockHash: block2Hash, PreviousBlockHash: block1Hash})
+
+	b0, _ := ledger.GetBlockByHash(gbHash)
+	testutil.AssertEquals(t, b0, gb)
+
+	b1, _ := ledger.GetBlockByHash(block1Hash)
+	testutil.AssertEquals(t, b1, block1)
+
+	b2, _ := ledger.GetBlockByHash(block2Hash)
+	testutil.AssertEquals(t, b2, block2)
+
+	b0, _ = ledger.GetBlockByNumber(0)
+	testutil.AssertEquals(t, b0, gb)
+
+	b1, _ = ledger.GetBlockByNumber(1)
+	testutil.AssertEquals(t, b1, block1)
+
+	b2, _ = ledger.GetBlockByNumber(2)
+	testutil.AssertEquals(t, b2, block2)
+
+	// get the tran id from the 2nd block, then use it to test GetTransactionByID()
+	txEnvBytes2 := block1.Data.Data[0]
+	txEnv2, err := putils.GetEnvelopeFromBlock(txEnvBytes2)
+	testutil.AssertNoError(t, err, "Error upon GetEnvelopeFromBlock")
+	payload2, err := putils.GetPayload(txEnv2)
+	testutil.AssertNoError(t, err, "Error upon GetPayload")
+	chdr, err := putils.UnmarshalChannelHeader(payload2.Header.ChannelHeader)
+	testutil.AssertNoError(t, err, "Error upon GetChannelHeaderFromBytes")
+	txID2 := chdr.TxId
+	processedTran2, err := ledger.GetTransactionByID(txID2)
+	testutil.AssertNoError(t, err, "Error upon GetTransactionByID")
+	// get the tran envelope from the retrieved ProcessedTransaction
+	retrievedTxEnv2 := processedTran2.TransactionEnvelope
+	testutil.AssertEquals(t, retrievedTxEnv2, txEnv2)
+
+	qe, _ := ledger.NewQueryExecutor()
+	value1, _ := qe.GetState("ns1", "key1")
+	testutil.AssertEquals(t, value1, []byte("value4"))
+
+	hqe, err := ledger.NewHistoryQueryExecutor()
+	testutil.AssertNoError(t, err, "")
+	itr, err := hqe.GetHistoryForKey("ns1", "key1")
+	testutil.AssertNoError(t, err, "")
+	defer itr.Close()
+
+	result1, err := itr.Next()
+	testutil.AssertNoError(t, err, "")
+	testutil.AssertEquals(t, result1.(*queryresult.KeyModification).Value, []byte("value1"))
+	result2, err := itr.Next()
+	testutil.AssertNoError(t, err, "")
+	testutil.AssertEquals(t, result2.(*queryresult.KeyModification).Value, []byte("value4"))
+}
+
 func constructTestLedgerID(i int) string {
 	return fmt.Sprintf("ledger_%06d", i)
 }

core/ledger/kvledger/pkg_test.go

@@ -29,7 +29,11 @@ type testEnv struct {
 }
 
 func newTestEnv(t testing.TB) *testEnv {
-	viper.Set("peer.fileSystemPath", "/tmp/fabric/ledgertests/kvledger")
+	return createTestEnv(t, "/tmp/fabric/ledgertests/kvledger")
+}
+
+func createTestEnv(t testing.TB, path string) *testEnv {
+	viper.Set("peer.fileSystemPath", path)
 	env := &testEnv{t}
 	env.cleanup()
 	return env