added to OSN new handler for fetch a block of channel part 4

integration/e2e/e2e_test.go

@@ -24,10 +24,14 @@ import (
 
 	docker "github.com/fsouza/go-dockerclient"
 	"github.com/hyperledger/fabric-lib-go/healthz"
+	cb "github.com/hyperledger/fabric-protos-go-apiv2/common"
+	ab "github.com/hyperledger/fabric-protos-go-apiv2/orderer"
 	"github.com/hyperledger/fabric-protos-go-apiv2/orderer/etcdraft"
 	"github.com/hyperledger/fabric/integration/nwo"
 	"github.com/hyperledger/fabric/integration/nwo/commands"
 	"github.com/hyperledger/fabric/integration/nwo/fabricconfig"
+	"github.com/hyperledger/fabric/integration/ordererclient"
+	"github.com/hyperledger/fabric/protoutil"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	"github.com/onsi/gomega/gbytes"
@@ -163,6 +167,9 @@ var _ = Describe("EndToEnd", func() {
 			By("setting up the channel")
 			nwo.JoinOrdererJoinPeersAppChannel(network, "testchannel", orderer, ordererRunner)
 
+			// A special method call for testing metrics
+			deliveryBlock(network, network.PeersWithChannel("testchannel")[0], orderer, "testchannel")
+
 			By("listing channels with osnadmin")
 			cl = nwo.List(network, orderer)
 			nwo.ChannelListMatcher(cl, []string{"testchannel"})
@@ -933,3 +940,35 @@ func hashFile(file string) string {
 func chaincodeContainerNameFilter(n *nwo.Network, chaincode nwo.Chaincode) string {
 	return fmt.Sprintf("^/%s-.*-%s-%s$", n.NetworkID, chaincode.Label, hashFile(chaincode.PackageFile))
 }
+
+func deliveryBlock(network *nwo.Network, peer *nwo.Peer, orderer *nwo.Orderer, channelID string) {
+	By("getting the signer for admin on orderer " + orderer.Name)
+	signer := network.OrdererUserSigner(orderer, "Admin")
+
+	By("starting delivery on orderer " + orderer.ID())
+	deliverEnvelope, err := protoutil.CreateSignedEnvelope(
+		cb.HeaderType_DELIVER_SEEK_INFO,
+		channelID,
+		signer,
+		&ab.SeekInfo{
+			Behavior: ab.SeekInfo_BLOCK_UNTIL_READY,
+			Start: &ab.SeekPosition{
+				Type: &ab.SeekPosition_Specified{
+					Specified: &ab.SeekSpecified{Number: 0},
+				},
+			},
+			Stop: &ab.SeekPosition{
+				Type: &ab.SeekPosition_Specified{
+					Specified: &ab.SeekSpecified{Number: 0},
+				},
+			},
+		},
+		0,
+		0,
+	)
+	Expect(err).NotTo(HaveOccurred())
+
+	blk, err := ordererclient.Deliver(network, orderer, deliverEnvelope)
+	Expect(err).NotTo(HaveOccurred())
+	Expect(blk).ToNot(BeNil())
+}

integration/nwo/channel_participation.go

@@ -21,6 +21,7 @@ import (
 	"github.com/onsi/gomega/gbytes"
 	"github.com/onsi/gomega/gstruct"
 	"github.com/onsi/gomega/types"
+	"github.com/pkg/errors"
 	ginkgomon "github.com/tedsuo/ifrit/ginkgomon_v2"
 	"google.golang.org/protobuf/proto"
 )
@@ -127,7 +128,7 @@ func doBody(client *http.Client, req *http.Request, expectedStatus int) []byte {
 	Expect(err).NotTo(HaveOccurred())
 	resp.Body.Close()
 
-	Expect(resp.StatusCode).To(Equal(expectedStatus), string(bodyBytes))
+	Expect(expectedStatus).To(Equal(resp.StatusCode), string(bodyBytes))
 
 	return bodyBytes
 }
@@ -171,6 +172,20 @@ func getBody(client *http.Client, url string) func() string {
 	}
 }
 
+func getBodyBinary(client *http.Client, url string) func() ([]byte, error) {
+	return func() ([]byte, error) {
+		resp, err := client.Get(url)
+		Expect(err).NotTo(HaveOccurred())
+		bodyBytes, err := io.ReadAll(resp.Body)
+		Expect(err).NotTo(HaveOccurred())
+		resp.Body.Close()
+		if resp.StatusCode != http.StatusOK {
+			return nil, errors.New(string(bodyBytes))
+		}
+		return bodyBytes, nil
+	}
+}
+
 type ChannelInfo struct {
 	Name              string `json:"name"`
 	URL               string `json:"url"`
@@ -195,6 +210,30 @@ func ListOne(n *Network, o *Orderer, channel string) ChannelInfo {
 	return *c
 }
 
+func Fetch(n *Network, o *Orderer, channel string, blockID string) (*common.Block, error) {
+	return FetchTimeShift(n, o, channel, blockID, 0)
+}
+
+func FetchTimeShift(n *Network, o *Orderer, channel string, blockID string, timeShift time.Duration) (*common.Block, error) {
+	authClient, _ := OrdererOperationalClientsTimeShift(n, o, timeShift)
+
+	protocol := "http"
+	if n.TLSEnabled {
+		protocol = "https"
+	}
+	fetchURL := fmt.Sprintf("%s://127.0.0.1:%d/participation/v1/channels/%s/blocks/%s", protocol, n.OrdererPort(o, AdminPort), channel, blockID)
+
+	body, err := getBodyBinary(authClient, fetchURL)()
+	if err != nil {
+		return nil, err
+	}
+
+	b := &common.Block{}
+	err = proto.Unmarshal(body, b)
+
+	return b, err
+}
+
 func Remove(n *Network, o *Orderer, channel string) {
 	authClient, _ := OrdererOperationalClients(n, o)
 

integration/nwo/configblock.go

@@ -33,32 +33,28 @@ func GetConfigBlock(n *Network, peer *Peer, orderer *Orderer, channel string) *c
 	defer os.RemoveAll(tempDir)
 
 	// fetch the config block
-	output := filepath.Join(tempDir, "config_block.pb")
-	FetchConfigBlock(n, peer, orderer, channel, output)
+	configBlock := FetchConfigBlock(n, orderer, channel)
 
-	// unmarshal the config block bytes
-	configBlock := UnmarshalBlockFromFile(output)
 	return configBlock
 }
 
 // FetchConfigBlock fetches latest config block.
-func FetchConfigBlock(n *Network, peer *Peer, orderer *Orderer, channel string, output string) {
+func FetchConfigBlock(n *Network, orderer *Orderer, channel string) *common.Block {
+	var (
+		err error
+		b   *common.Block
+	)
 	fetch := func() int {
-		sess, err := n.OrdererAdminSession(orderer, peer, commands.ChannelFetch{
-			ChannelID:  channel,
-			Block:      "config",
-			Orderer:    n.OrdererAddress(orderer, ListenPort),
-			OutputFile: output,
-			ClientAuth: n.ClientAuthRequired,
-		})
-		Expect(err).NotTo(HaveOccurred())
-		code := sess.Wait(n.EventuallyTimeout).ExitCode()
-		if code == 0 {
-			Expect(sess.Err).To(gbytes.Say("Received block: "))
+		b, err = Fetch(n, orderer, channel, "config")
+		if err != nil || b == nil {
+			return 1
 		}
-		return code
+
+		return 0
 	}
 	Eventually(fetch, n.EventuallyTimeout).Should(Equal(0))
+
+	return b
 }
 
 // GetConfig retrieves the last config of the given channel.
@@ -179,15 +175,12 @@ func CurrentConfigBlockNumber(n *Network, peer *Peer, orderer *Orderer, channel
 	defer os.RemoveAll(tempDir)
 
 	// fetch the config block
-	output := filepath.Join(tempDir, "config_block.pb")
 	if orderer == nil {
+		output := filepath.Join(tempDir, "config_block.pb")
 		return CurrentConfigBlockNumberFromPeer(n, peer, channel, output)
 	}
 
-	FetchConfigBlock(n, peer, orderer, channel, output)
-
-	// unmarshal the config block bytes
-	configBlock := UnmarshalBlockFromFile(output)
+	configBlock := FetchConfigBlock(n, orderer, channel)
 
 	return configBlock.Header.Number
 }

integration/nwo/network.go

@@ -1150,15 +1150,27 @@ func (n *Network) JoinChannel(name string, o *Orderer, peers ...*Peer) {
 	tempFile.Close()
 	defer os.Remove(tempFile.Name())
 
-	sess, err := n.PeerAdminSession(peers[0], commands.ChannelFetch{
-		Block:      "0",
-		ChannelID:  name,
-		Orderer:    n.OrdererAddress(o, ListenPort),
-		OutputFile: tempFile.Name(),
-		ClientAuth: n.ClientAuthRequired,
-	})
-	Expect(err).NotTo(HaveOccurred())
-	Eventually(sess, n.EventuallyTimeout).Should(gexec.Exit(0))
+	Eventually(func() string {
+		block, err := Fetch(n, o, name, "0")
+		if err != nil {
+			return fmt.Sprintf("error is %s", err.Error())
+		}
+
+		if block == nil {
+			return "proto: Marshal called with nil"
+		}
+
+		b, err := proto.Marshal(block)
+		if err != nil {
+			return err.Error()
+		}
+
+		if err = os.WriteFile(tempFile.Name(), b, 0o644); err != nil {
+			return err.Error()
+		}
+
+		return ""
+	}, n.EventuallyTimeout, time.Second).Should(BeEmpty())
 
 	for _, p := range peers {
 		sess, err := n.PeerAdminSession(p, commands.ChannelJoin{

integration/ordererclient/orderer_client.go

@@ -8,6 +8,7 @@ package ordererclient
 
 import (
 	"context"
+	"reflect"
 
 	"github.com/hyperledger/fabric-protos-go-apiv2/common"
 	"github.com/hyperledger/fabric-protos-go-apiv2/orderer"
@@ -56,10 +57,17 @@ func Deliver(n *nwo.Network, o *nwo.Orderer, env *common.Envelope) (*common.Bloc
 		return nil, err
 	}
 
-	blk := resp.GetBlock()
-	if blk == nil {
-		return nil, errors.Errorf("block not found")
-	}
+	switch t := resp.Type.(type) {
+	case *orderer.DeliverResponse_Block:
+		blk := resp.GetBlock()
+		if blk == nil {
+			return nil, errors.Errorf("block not found")
+		}
 
-	return blk, nil
+		return blk, nil
+	case *orderer.DeliverResponse_Status:
+		return nil, errors.Errorf("faulty node, received status: %s", common.Status_name[int32(t.Status)])
+	default:
+		return nil, errors.Errorf("response is of type %v, but expected a block", reflect.TypeOf(resp.Type))
+	}
 }

integration/pvtdata/pvtdata_test.go

@@ -269,21 +269,22 @@ var _ = Describe("PrivateData", func() {
 			Eventually(p.Ready(), network.EventuallyTimeout).Should(BeClosed())
 
 			By("joining peer1.org2 to the channel with its Admin2 user")
+			block, err := nwo.Fetch(network, orderer, channelID, "0")
+			Expect(err).NotTo(HaveOccurred())
+			Expect(block).NotTo(BeNil())
+
 			tempFile, err := os.CreateTemp("", "genesis-block")
 			Expect(err).NotTo(HaveOccurred())
 			tempFile.Close()
 			defer os.Remove(tempFile.Name())
 
-			sess, err := network.PeerUserSession(org2Peer1, "Admin2", commands.ChannelFetch{
-				Block:      "0",
-				ChannelID:  channelID,
-				Orderer:    network.OrdererAddress(orderer, nwo.ListenPort),
-				OutputFile: tempFile.Name(),
-			})
+			b, err := proto.Marshal(block)
+			Expect(err).NotTo(HaveOccurred())
+
+			err = os.WriteFile(tempFile.Name(), b, 0o644)
 			Expect(err).NotTo(HaveOccurred())
-			Eventually(sess, network.EventuallyTimeout).Should(gexec.Exit(0))
 
-			sess, err = network.PeerUserSession(org2Peer1, "Admin2", commands.ChannelJoin{
+			sess, err := network.PeerUserSession(org2Peer1, "Admin2", commands.ChannelJoin{
 				BlockPath: tempFile.Name(),
 			})
 			Expect(err).NotTo(HaveOccurred())
@@ -950,18 +951,11 @@ func addPeer(n *nwo.Network, orderer *nwo.Orderer, peer *nwo.Peer) ifrit.Process
 
 	n.JoinChannel(channelID, orderer, peer)
 	ledgerHeight := nwo.GetLedgerHeight(n, n.Peers[0], channelID)
-	sess, err := n.PeerAdminSession(
-		peer,
-		commands.ChannelFetch{
-			Block:      "newest",
-			ChannelID:  channelID,
-			Orderer:    n.OrdererAddress(orderer, nwo.ListenPort),
-			OutputFile: filepath.Join(n.RootDir, "newest_block.pb"),
-		},
-	)
+
+	b, err := nwo.Fetch(n, orderer, channelID, "newest")
 	Expect(err).NotTo(HaveOccurred())
-	Eventually(sess, n.EventuallyTimeout).Should(gexec.Exit(0))
-	Expect(sess.Err).To(gbytes.Say(fmt.Sprintf("Received block: %d", ledgerHeight-1)))
+	Expect(b).NotTo(BeNil())
+	Expect(b.GetHeader().GetNumber()).To(Equal(uint64(ledgerHeight) - 1))
 
 	n.Peers = append(n.Peers, peer)
 	nwo.WaitUntilEqualLedgerHeight(n, channelID, nwo.GetLedgerHeight(n, n.Peers[0], channelID), n.Peers...)

integration/pvtdatapurge/data_purge_test.go

@@ -27,7 +27,6 @@ import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	"github.com/onsi/gomega/gbytes"
-	"github.com/onsi/gomega/gexec"
 	"github.com/tedsuo/ifrit"
 	ginkgomon "github.com/tedsuo/ifrit/ginkgomon_v2"
 	"google.golang.org/protobuf/proto"
@@ -476,18 +475,10 @@ func startNewPeer(network *nwo.Network, orderer *nwo.Orderer, peer *nwo.Peer, le
 	startPeer(network, processes, runners, peer)
 
 	network.JoinChannel(channelID, orderer, peer)
-	sess, err := network.PeerAdminSession(
-		peer,
-		commands.ChannelFetch{
-			Block:      "newest",
-			ChannelID:  channelID,
-			Orderer:    network.OrdererAddress(orderer, nwo.ListenPort),
-			OutputFile: filepath.Join(network.RootDir, "newest_block.pb"),
-		},
-	)
+
+	b, err := nwo.Fetch(network, orderer, channelID, "newest")
 	Expect(err).NotTo(HaveOccurred())
-	Eventually(sess, network.EventuallyTimeout).Should(gexec.Exit(0))
-	Expect(sess.Err).To(gbytes.Say(fmt.Sprintf("Received block: %d", ledgerHeight-1)))
+	Expect(b.GetHeader().GetNumber()).To(Equal(uint64(ledgerHeight) - 1))
 
 	network.Peers = append(network.Peers, peer)
 	nwo.WaitUntilEqualLedgerHeight(network, channelID, ledgerHeight-1, network.Peers...)
@@ -499,18 +490,10 @@ func addPeer(n *nwo.Network, orderer *nwo.Orderer, peer *nwo.Peer) ifrit.Process
 
 	n.JoinChannel(channelID, orderer, peer)
 	ledgerHeight := nwo.GetLedgerHeight(n, n.Peers[0], channelID)
-	sess, err := n.PeerAdminSession(
-		peer,
-		commands.ChannelFetch{
-			Block:      "newest",
-			ChannelID:  channelID,
-			Orderer:    n.OrdererAddress(orderer, nwo.ListenPort),
-			OutputFile: filepath.Join(n.RootDir, "newest_block.pb"),
-		},
-	)
+
+	b, err := nwo.Fetch(n, orderer, channelID, "newest")
 	Expect(err).NotTo(HaveOccurred())
-	Eventually(sess, n.EventuallyTimeout).Should(gexec.Exit(0))
-	Expect(sess.Err).To(gbytes.Say(fmt.Sprintf("Received block: %d", ledgerHeight-1)))
+	Expect(b.GetHeader().GetNumber()).To(Equal(uint64(ledgerHeight) - 1))
 
 	n.Peers = append(n.Peers, peer)
 	nwo.WaitUntilEqualLedgerHeight(n, channelID, nwo.GetLedgerHeight(n, n.Peers[0], channelID), n.Peers...)