Hyperledger Fabric BYFN之配置进阶篇

http://www.javatree.cn/news/b25c5e3b9d4640e4a0da2039981c7a03

1. byfn.sh up down会清理所有容器镜像，生成的配置文件和证书，现实场景应该可以停止或恢复区块链网络，该如何处理？

byfn.sh down会调用networkdDown去销毁整个网络和已生成的配置。byfn.sh restart则不会清理。

# Tear down running network

function networkDown() {

# stop org3 containers also in addition to org1 and org2, in case we were running sample to add org3

docker-compose -f COMPOSE_FILE -f COMPOSE_FILE_COUCH -f

# Don't remove the generated artifacts -- note, the ledgers are always removed

if [ "$MODE" != "restart" ]; then

# Bring down the network, deleting the volumes

#Delete any ledger backups

docker run -v PWD:/tmp/first-network --rm hyperledger/fabric-tools:IMAGETAG rm -Rf /tmp/first-network/ledgers-backup

#Cleanup the chaincode containers

clearContainers

#Cleanup images

removeUnwantedImages

# remove orderer block and other channel configuration transactions and certs

rm -rf channel-artifacts/*.block channel-artifacts/*.tx crypto-config ./org3-artifacts/crypto-config/ channel-artifacts/org3.json

# remove the docker-compose yaml file that was customized to the example

rm -f docker-compose-e2e.yaml

fi

}

# Generate the needed certificates, the genesis block and start the network.

function networkUp() {

checkPrereqs

# generate artifacts if they don't exist

if [ ! -d "crypto-config" ]; then

generateCerts

replacePrivateKey

generateChannelArtifacts

fi

if [ "${IF_COUCHDB}" == "couchdb" ]; then

IMAGE_TAG=IMAGETAG docker-compose -f COMPOSE_FILE -f

else

IMAGE_TAG=IMAGETAG docker-compose -f COMPOSE_FILE up -d 2>&1

fi

if [ $? -ne 0 ]; then

echo "ERROR !!!! Unable to start network"

exit 1

fi

# now run the end to end script

docker exec cli scripts/script.sh CHANNEL_NAME CLI_DELAY LANGUAGE CLI_TIMEOUT

if [ $? -ne 0 ]; then

echo "ERROR !!!! Test failed"

exit 1

fi

}

注意

docker-compose down会停止和删除容器,网络，镜像和映射的卷。

只是停止服务的话的会最好使用docker-compose stop。

docker-compose up则是创建和启动容器服务, 这里用于启动。

我们也可以参考例子fabric-samples/fabcar, 它会重用fabric-samples/basic-network中的start.sh, stop.sh允许停止和重启。

2. byfn.sh还是fabcar两个例子即使重启区块链网络，通道需要创新创建，节点需要重新加入通道，链码也要全要重新安装，如果节点多维护起来就麻烦且费时，重启的时候能让通道，节点，链码，State DB自动恢复？

这自然是有的，但是必须开启orderer， peer等的持久化配置。

以/fabric-samples/basic-networkd的docker-compose.yaml为例, 配置services, 容器路径通常包含production的则是持久化路径的映射配置, 见蓝色部分。

orderer.example.com:

container_name: orderer.example.com

image: hyperledger/fabric-orderer

environment:

- ORDERER_GENERAL_LOGLEVEL=info

- ORDERER_GENERAL_LISTENADDRESS=0.0.0.0

- ORDERER_GENERAL_GENESISMETHOD=file

- ORDERER_GENERAL_GENESISFILE=/etc/hyperledger/configtx/genesis.block

- ORDERER_GENERAL_LOCALMSPID=OrdererMSP

- ORDERER_GENERAL_LOCALMSPDIR=/etc/hyperledger/msp/orderer/msp

working_dir: /opt/gopath/src/github.com/hyperledger/fabric/orderer

command: orderer

ports:

- 7050:7050

volumes:

- ./config/:/etc/hyperledger/configtx

- ./crypto-config/ordererOrganizations/example.com/orderers/orderer.example.com/:/etc/hyperledger/msp/orderer

- ./crypto-config/peerOrganizations/org1.example.com/peers/peer0.org1.example.com/:/etc/hyperledger/msp/peerOrg1

- /mnt/hyperledger/orderer:/var/hyperledger/production/orderer

networks

peer0.org1.example.com:

container_name: peer0.org1.example.com

image: hyperledger/fabric-peer

environment:

- CORE_VM_ENDPOINT=unix:///host/var/run/docker.sock

- CORE_PEER_ID=peer0.org1.example.com

- CORE_LOGGING_PEER=info

- CORE_CHAINCODE_LOGGING_LEVEL=info

- CORE_PEER_LOCALMSPID=Org1MSP

- CORE_PEER_MSPCONFIGPATH=/etc/hyperledger/msp/peer/

- CORE_PEER_ADDRESS=peer0.org1.example.com:7051

# # the following setting starts chaincode containers on the same

# # bridge network as the peers

# # https://docs.docker.com/compose/networking/

- CORE_VM_DOCKER_HOSTCONFIG_NETWORKMODE=${COMPOSE_PROJECT_NAME}_basic

- CORE_LEDGER_STATE_STATEDATABASE=CouchDB

- CORE_LEDGER_STATE_COUCHDBCONFIG_COUCHDBADDRESS=couchdb:5984

# The CORE_LEDGER_STATE_COUCHDBCONFIG_USERNAME and CORE_LEDGER_STATE_COUCHDBCONFIG_PASSWORD

# provide the credentials for ledger to connect to CouchDB. The username and password must

# match the username and password set for the associated CouchDB.

- CORE_LEDGER_STATE_COUCHDBCONFIG_USERNAME=

- CORE_LEDGER_STATE_COUCHDBCONFIG_PASSWORD=

working_dir: /opt/gopath/src/github.com/hyperledger/fabric

command: peer node start

# command: peer node start --peer-chaincodedev=true

ports:

- 7051:7051

- 7053:7053

volumes:

- /var/run/:/host/var/run/

- ./crypto-config/peerOrganizations/org1.example.com/peers/peer0.org1.example.com/msp:/etc/hyperledger/msp/peer

- ./crypto-config/peerOrganizations/org1.example.com/users:/etc/hyperledger/msp/users

- ./config:/etc/hyperledger/configtx

- /mnt/hyperledger/org1/peer0:/var/hyperledger/production

depends_on:

- orderer.example.com

- couchdb

networks:

- basic

couchdb:

container_name: couchdb

image: hyperledger/fabric-couchdb

# Populate the COUCHDB_USER and COUCHDB_PASSWORD to set an admin user and password

# for CouchDB. This will prevent CouchDB from operating in an "Admin Party" mode.

environment:

- COUCHDB_USER=

- COUCHDB_PASSWORD=

ports:

- 5984:5984

networks:

- basic

volumes:

- /mnt/hyperledger/couchdb:/opt/couchdb/data

3. Peer节点默认使用level DB作为state DB, key-value键值对查询较弱， couch DB支付富查询，如何配置？

byfn.sh up -c mychannel -s couchdb 实际使用的配置文件是cker-compose-couch.yaml,

services:

couchdb0:

container_name: couchdb0

image: hyperledger/fabric-couchdb

# Populate the COUCHDB_USER and COUCHDB_PASSWORD to set an admin user and password

# for CouchDB. This will prevent CouchDB from operating in an "Admin Party" mode.

environment:

- COUCHDB_USER=

- COUCHDB_PASSWORD=

# Comment/Uncomment the port mapping if you want to hide/expose the CouchDB service,

# for example map it to utilize Fauxton User Interface in dev environments.

ports:

- "5984:5984"

networks:

- byfn

peer0.org1.example.com:

environment:

- CORE_LEDGER_STATE_STATEDATABASE=CouchDB

- CORE_LEDGER_STATE_COUCHDBCONFIG_COUCHDBADDRESS=couchdb0:5984

# The CORE_LEDGER_STATE_COUCHDBCONFIG_USERNAME and CORE_LEDGER_STATE_COUCHDBCONFIG_PASSWORD

# provide the credentials for ledger to connect to CouchDB. The username and password must

# match the username and password set for the associated CouchDB.

- CORE_LEDGER_STATE_COUCHDBCONFIG_USERNAME=

- CORE_LEDGER_STATE_COUCHDBCONFIG_PASSWORD=

depends_on:

- couchdb0

peer主要设置连接的couchdb地址和账号, couchdb主要是配置对应端口号，couchDB还支持一些字段的索引，在学习链码的时候我们再深入。

4. Peer节点我们配置了两个端口，配置用来做什么？

参看fabric-samples/first-network/base/docker-compose-base.yaml

peer0.org1.example.com:

container_name: peer0.org1.example.com

extends:

file: peer-base.yaml

service: peer-base

environment:

- CORE_PEER_ID=peer0.org1.example.com

- CORE_PEER_ADDRESS=peer0.org1.example.com:7051

- CORE_PEER_GOSSIP_BOOTSTRAP=peer1.org1.example.com:7051

- CORE_PEER_GOSSIP_EXTERNALENDPOINT=peer0.org1.example.com:7051

- CORE_PEER_LOCALMSPID=Org1MSP

volumes:

- /var/run/:/host/var/run/

- ../crypto-config/peerOrganizations/org1.example.com/peers/peer0.org1.example.com/msp:/etc/hyperledger/fabric/msp

- ../crypto-config/peerOrganizations/org1.example.com/peers/peer0.org1.example.com/tls:/etc/hyperledger/fabric/tls

- peer0.org1.example.com:/var/hyperledger/production

ports:

- 7051:7051

- 7053:7053

7051是Peer启动的gRPC, 一般是客户端应用接入。

7053是事件端口(Peer Event)

Fabric 1.1之前, 被定位为Event Hub(节点的消息中心), 当Peer节点记账本副本追加了新的区块的时候，都会通知订阅了这些消息的客户端应用。以fabcar的invoke.js代码为例

console.log(util.format(

'Successfully sent Proposal and received ProposalResponse: Status - %s, message - "%s"',

proposalResponses[0].response.status, proposalResponses[0].response.message));

// build up the request for the orderer to have the transaction committed

var request = {

proposalResponses: proposalResponses,

proposal: proposal

};

// set the transaction listener and set a timeout of 30 sec

// if the transaction did not get committed within the timeout period,

// report a TIMEOUT status

var transaction_id_string = tx_id.getTransactionID(); //Get the transaction ID string to be used by the event processing

var promises = [];

var sendPromise = channel.sendTransaction(request);

promises.push(sendPromise); //we want the send transaction first, so that we know where to check status

// get an eventhub once the fabric client has a user assigned. The user

// is required bacause the event registration must be signed

let event_hub = fabric_client.newEventHub();

event_hub.setPeerAddr('grpc://localhost:7053');

// using resolve the promise so that result status may be processed

// under the then clause rather than having the catch clause process

// the status

let txPromise = new Promise((resolve, reject) => {

let handle = setTimeout(() => {

event_hub.disconnect();

resolve({event_status : 'TIMEOUT'}); //we could use reject(new Error('Trnasaction did not complete within 30 seconds'));

}, 3000);

event_hub.connect();

event_hub.registerTxEvent(transaction_id_string, (tx, code) => {

// this is the callback for transaction event status

// first some clean up of event listener

clearTimeout(handle);

event_hub.unregisterTxEvent(transaction_id_string);

event_hub.disconnect();

// now let the application know what happened

var return_status = {event_status : code, tx_id : transaction_id_string};

if (code !== 'VALID') {

console.error('The transaction was invalid, code = ' + code);

resolve(return_status); // we could use reject(new Error('Problem with the tranaction, event status ::'+code));

} else {

console.log('The transaction has been committed on peer ' + event_hub._ep._endpoint.addr);

resolve(return_status);

}

}, (err) => {

//this is the callback if something goes wrong with the event registration or processing

reject(new Error('There was a problem with the eventhub ::'+err));

});

});

promises.push(txPromise);

return Promise.all(promises);

Fabric 1.1之后peer event做了完全不同的设计，消息的监听不在peer节点了，而是基于channel，这样设计提供了对Peer数据更细粒度的采访控制和提供了接收消息的可靠性。(官方文档是这么扯，我也有点疑惑) 主要提供两种服务，Deliver(通知提交到记账本的整个区块内容) 和DeliverFiltered(过滤一些区块减少消息通知返回的区块大小)

有点晦涩， 找了下Java SDK的代码看下。

/**

* This code test the replay feature of the new peer event services.

* Instead of the default of starting the eventing peer to retrieve the newest block it sets it

* retrieve starting from the start parameter. Also checks with block and filterblock replays.

* Depends on end2end and end2endAndBackagain of have fully run to have the blocks need to work with.

*

* @param client

* @param replayTestChannel

* @param start

* @param stop

* @param useFilteredBlocks

* @throws InvalidArgumentException

*/

private void testPeerServiceEventingReplay(HFClient client, Channel replayTestChannel, final long start, final long stop,

final boolean useFilteredBlocks) throws InvalidArgumentException {

if (testConfig.isRunningAgainstFabric10()) {

return; // not supported for v1.0

}

assertFalse(replayTestChannel.isInitialized()); //not yet initialized

assertFalse(replayTestChannel.isShutdown()); // not yet shutdown.

//Remove all peers just have one ledger peer and one eventing peer.

List<Peer> savedPeers = new ArrayList<>(replayTestChannel.getPeers());

for (Peer peer : savedPeers) {

replayTestChannel.removePeer(peer);

}

assertTrue(savedPeers.size() > 1); //need at least two

Peer eventingPeer = savedPeers.remove(0);

eventingPeer = client.newPeer(eventingPeer.getName(), eventingPeer.getUrl(), eventingPeer.getProperties());

Peer ledgerPeer = savedPeers.remove(0);

ledgerPeer = client.newPeer(ledgerPeer.getName(), ledgerPeer.getUrl(), ledgerPeer.getProperties());

assertTrue(replayTestChannel.getPeers().isEmpty()); // no more peers.

assertTrue(replayTestChannel.getPeers(EnumSet.of(PeerRole.CHAINCODE_QUERY, PeerRole.ENDORSING_PEER)).isEmpty()); // just checking :)

assertTrue(replayTestChannel.getPeers(EnumSet.of(PeerRole.LEDGER_QUERY)).isEmpty()); // just checking

assertNotNull(client.getChannel(replayTestChannel.getName())); // should be known by client.

final PeerOptions eventingPeerOptions = createPeerOptions().setPeerRoles(EnumSet.of(PeerRole.EVENT_SOURCE));

if (useFilteredBlocks) {

eventingPeerOptions.registerEventsForFilteredBlocks();

}

if (-1L == stop) { //the height of the blockchain

replayTestChannel.addPeer(eventingPeer, eventingPeerOptions.startEvents(start)); // Eventing peer start getting blocks from block 0

} else {

replayTestChannel.addPeer(eventingPeer, eventingPeerOptions

.startEvents(start).stopEvents(stop)); // Eventing peer start getting blocks from block 0

}

//add a ledger peer

replayTestChannel.addPeer(ledgerPeer, createPeerOptions().setPeerRoles(EnumSet.of(PeerRole.LEDGER_QUERY)));

CompletableFuture<Long> done = new CompletableFuture<>(); // future to set when done.

// some variable used by the block listener being set up.

final AtomicLong bcount = new AtomicLong(0);

final AtomicLong stopValue = new AtomicLong(stop == -1L ? Long.MAX_VALUE : stop);

final Channel finalChannel = replayTestChannel;

final Map<Long, BlockEvent> blockEvents = Collections.synchronizedMap(new HashMap<>(100));

final String blockListenerHandle = replayTestChannel.registerBlockListener(blockEvent -> { // register a block listener

try {

final long blockNumber = blockEvent.getBlockNumber();

BlockEvent seen = blockEvents.put(blockNumber, blockEvent);

assertNull(format("Block number %d seen twice", blockNumber), seen);

assertTrue(format("Wrong type of block seen block number %d. expected filtered block %b but got %b",

blockNumber, useFilteredBlocks, blockEvent.isFiltered()),

useFilteredBlocks ? blockEvent.isFiltered() : !blockEvent.isFiltered());

final long count = bcount.getAndIncrement(); //count starts with 0 not 1 !

//out("Block count: %d, block number: %d received from peer: %s", count, blockNumber, blockEvent.getPeer().getName());

if (count == 0 && stop == -1L) {

final BlockchainInfo blockchainInfo = finalChannel.queryBlockchainInfo();

long lh = blockchainInfo.getHeight();

stopValue.set(lh - 1L); // blocks 0L 9L are on chain height 10 .. stop on 9

// out("height: %d", lh);

if (bcount.get() + start > stopValue.longValue()) { // test with latest count.

done.complete(bcount.get()); // report back latest count.

}

} else {

if (bcount.longValue() + start > stopValue.longValue()) {

done.complete(count);

}

}

} catch (AssertionError | Exception e) {

e.printStackTrace();

done.completeExceptionally(e);

}

});

try {

replayTestChannel.initialize(); // start it all up.

done.get(30, TimeUnit.SECONDS); // give a timeout here.

Thread.sleep(1000); // sleep a little to see if more blocks trickle in .. they should not

replayTestChannel.unregisterBlockListener(blockListenerHandle);

final long expectNumber = stopValue.longValue() - start + 1L; // Start 2 and stop is 3 expect 2

assertEquals(format("Didn't get number we expected %d but got %d block events. Start: %d, end: %d, height: %d",

expectNumber, blockEvents.size(), start, stop, stopValue.longValue()), expectNumber, blockEvents.size());

for (long i = stopValue.longValue(); i >= start; i--) { //make sure all are there.

final BlockEvent blockEvent = blockEvents.get(i);

assertNotNull(format("Missing block event for block number %d. Start= %d", i, start), blockEvent);

}

//light weight test just see if we get reasonable values for traversing the block. Test just whats common between

// Block and FilteredBlock.

int transactionEventCounts = 0;

int chaincodeEventsCounts = 0;

for (long i = stopValue.longValue(); i >= start; i--) {

final BlockEvent blockEvent = blockEvents.get(i);

// out("blockwalker %b, start: %d, stop: %d, i: %d, block %d", useFilteredBlocks, start, stopValue.longValue(), i, blockEvent.getBlockNumber());

assertEquals(useFilteredBlocks, blockEvent.isFiltered()); // check again

if (useFilteredBlocks) {

assertNull(blockEvent.getBlock()); // should not have raw block event.

assertNotNull(blockEvent.getFilteredBlock()); // should have raw filtered block.

} else {

assertNotNull(blockEvent.getBlock()); // should not have raw block event.

assertNull(blockEvent.getFilteredBlock()); // should have raw filtered block.

}

assertEquals(replayTestChannel.getName(), blockEvent.getChannelId());

for (BlockInfo.EnvelopeInfo envelopeInfo : blockEvent.getEnvelopeInfos()) {

if (envelopeInfo.getType() == TRANSACTION_ENVELOPE) {

BlockInfo.TransactionEnvelopeInfo transactionEnvelopeInfo = (BlockInfo.TransactionEnvelopeInfo) envelopeInfo;

assertTrue(envelopeInfo.isValid()); // only have valid blocks.

assertEquals(envelopeInfo.getValidationCode(), 0);

++transactionEventCounts;

for (BlockInfo.TransactionEnvelopeInfo.TransactionActionInfo ta : transactionEnvelopeInfo.getTransactionActionInfos()) {

// out("\nTA:", ta + "\n\n");

ChaincodeEvent event = ta.getEvent();

if (event != null) {

assertNotNull(event.getChaincodeId());

assertNotNull(event.getEventName());

chaincodeEventsCounts++;

}

}

} else {

assertEquals("Only non transaction block should be block 0.", blockEvent.getBlockNumber(), 0);

}

}

}

assertTrue(transactionEventCounts > 0);

if (expectNumber > 4) { // this should be enough blocks with CC events.

assertTrue(chaincodeEventsCounts > 0);

}

replayTestChannel.shutdown(true); //all done.

} catch (Exception e) {

e.printStackTrace();

fail(e.getMessage());

}

}

再结合初始化通道的代码

boolean everyOther = false;

for (String peerName : sampleOrg.getPeerNames()) {

String peerLocation = sampleOrg.getPeerLocation(peerName);

Properties peerProperties = testConfig.getPeerProperties(peerName);

Peer peer = client.newPeer(peerName, peerLocation, peerProperties);

final PeerOptions peerEventingOptions = // we have two peers on one use block on other use filtered

everyOther ?

createPeerOptions().registerEventsForBlocks().setPeerRoles(EnumSet.of(PeerRole.ENDORSING_PEER, PeerRole.LEDGER_QUERY, PeerRole.CHAINCODE_QUERY, PeerRole.EVENT_SOURCE)) :

createPeerOptions().registerEventsForFilteredBlocks().setPeerRoles(EnumSet.of(PeerRole.ENDORSING_PEER, PeerRole.LEDGER_QUERY, PeerRole.CHAINCODE_QUERY, PeerRole.EVENT_SOURCE));

newChannel.addPeer(peer, IS_FABRIC_V10 ?

createPeerOptions().setPeerRoles(EnumSet.of(PeerRole.ENDORSING_PEER, PeerRole.LEDGER_QUERY, PeerRole.CHAINCODE_QUERY)) : peerEventingOptions);

everyOther = !everyOther;

}

Fabric-SDK在设计的时候，无论是查询，更新，大多操作都是基于org.hyperledger.fabric.sdk.Channel作为入口， 例如：

Collection<ProposalResponse> queryProposals = channel.queryByChaincode(queryByChaincodeRequest, channel.getPeers());

Collection<ProposalResponse> transactionPropResp = channel.sendTransactionProposal(transactionProposalRequest, channel.getPeers());

Event Hub基于单个Peer去监听它的消息，可能单节点会不稳定等，而通道包含多节点，消息通知可能会更稳定一些，这个7053端口应该是专门给1.1前的Event Hub版本用的，1.1之后的具体怎么回调通知， 具体实现还有待查阅源码， 这里是保守的猜测吧。

5. Orderer大多例子都是SOLO配置的，Kafka如何配置？

老实说要用好Kafka里面蛮多参数需要考量，毕竟Orderer集群是共识实现的重点，后面我们专门探讨下kafka实现的共识和配置。

暂时先参考官方文档

https://hyperledger-fabric.readthedocs.io/en/release-1.2/kafka.html?highlight=kafka