[FAB-4866] Add orderer msg bytes trace

For hard to debug problems, actually having access to the bytes of a
message can be the only definitive way to diagnose.

This CR adds a debug configuration section, and allows for setting a
directory to log all Broadcast messages, as well as a directory to log
all Deliver messages.

Although there is no support for dynamically changing debugging
parameters at this time, the code deliberately retrieves the debug
parameters from the debug struct at every instance to allow for dynamic
control of the debugging.

Change-Id: Ib046f263dc95b374b5883af66397a9d29049ffef
Signed-off-by: Jason Yellick <jyellick@us.ibm.com>

Author: Jason Yellick

orderer/common/localconfig/config.go

@@ -65,6 +65,7 @@ type TopLevel struct {
 	FileLedger FileLedger
 	RAMLedger  RAMLedger
 	Kafka      Kafka
+	Debug      Debug
 }
 
 // General contains config which should be common among all orderer types.
@@ -162,6 +163,12 @@ type Consumer struct {
 	RetryBackoff time.Duration
 }
 
+// Debug contains configuration for the orderer's debug parameters
+type Debug struct {
+	BroadcastTraceDir string
+	DeliverTraceDir   string
+}
+
 var defaults = TopLevel{
 	General: General{
 		LedgerType:     "file",
@@ -216,6 +223,10 @@ var defaults = TopLevel{
 			Enabled: false,
 		},
 	},
+	Debug: Debug{
+		BroadcastTraceDir: "",
+		DeliverTraceDir:   "",
+	},
 }
 
 // Load parses the orderer.yaml file and environment, producing a struct suitable for config use

orderer/common/server/main.go

@@ -72,7 +72,7 @@ func Main() {
 func Start(cmd string, conf *config.TopLevel) {
 	signer := localmsp.NewSigner()
 	manager := initializeMultichannelRegistrar(conf, signer)
-	server := NewServer(manager, signer)
+	server := NewServer(manager, signer, &conf.Debug)
 
 	switch cmd {
 	case start.FullCommand(): // "start" command

orderer/common/server/server.go

@@ -7,14 +7,21 @@ SPDX-License-Identifier: Apache-2.0
 package server
 
 import (
+	"fmt"
+	"io/ioutil"
+	"os"
+	"runtime/debug"
+	"time"
+
 	"github.com/hyperledger/fabric/common/crypto"
 	"github.com/hyperledger/fabric/orderer/common/broadcast"
 	"github.com/hyperledger/fabric/orderer/common/deliver"
+	localconfig "github.com/hyperledger/fabric/orderer/common/localconfig"
 	"github.com/hyperledger/fabric/orderer/common/multichannel"
 	cb "github.com/hyperledger/fabric/protos/common"
 	ab "github.com/hyperledger/fabric/protos/orderer"
 
-	"runtime/debug"
+	"github.com/golang/protobuf/proto"
 )
 
 type broadcastSupport struct {
@@ -34,19 +41,73 @@ func (bs deliverSupport) GetChain(chainID string) (deliver.Support, bool) {
 }
 
 type server struct {
-	bh broadcast.Handler
-	dh deliver.Handler
+	bh    broadcast.Handler
+	dh    deliver.Handler
+	debug *localconfig.Debug
 }
 
 // NewServer creates an ab.AtomicBroadcastServer based on the broadcast target and ledger Reader
-func NewServer(r *multichannel.Registrar, signer crypto.LocalSigner) ab.AtomicBroadcastServer {
+func NewServer(r *multichannel.Registrar, signer crypto.LocalSigner, debug *localconfig.Debug) ab.AtomicBroadcastServer {
 	s := &server{
-		dh: deliver.NewHandlerImpl(deliverSupport{Registrar: r}),
-		bh: broadcast.NewHandlerImpl(broadcastSupport{Registrar: r}),
+		dh:    deliver.NewHandlerImpl(deliverSupport{Registrar: r}),
+		bh:    broadcast.NewHandlerImpl(broadcastSupport{Registrar: r}),
+		debug: debug,
 	}
 	return s
 }
 
+type msgTracer struct {
+	function string
+	debug    *localconfig.Debug
+}
+
+func (mt *msgTracer) trace(traceDir string, msg *cb.Envelope, err error) {
+	if err != nil {
+		return
+	}
+
+	now := time.Now().UnixNano()
+	path := fmt.Sprintf("%s%c%d_%p.%s", traceDir, os.PathSeparator, now, msg, mt.function)
+	logger.Debugf("Writing %s request trace to %s", mt.function, path)
+	go func() {
+		pb, err := proto.Marshal(msg)
+		if err != nil {
+			logger.Debugf("Error marshaling trace msg for %s: %s", path, err)
+			return
+		}
+		err = ioutil.WriteFile(path, pb, 0660)
+		if err != nil {
+			logger.Debugf("Error writing trace msg for %s: %s", path, err)
+		}
+	}()
+}
+
+type broadcastMsgTracer struct {
+	ab.AtomicBroadcast_BroadcastServer
+	msgTracer
+}
+
+func (bmt *broadcastMsgTracer) Recv() (*cb.Envelope, error) {
+	msg, err := bmt.AtomicBroadcast_BroadcastServer.Recv()
+	if traceDir := bmt.debug.BroadcastTraceDir; traceDir != "" {
+		bmt.trace(bmt.debug.BroadcastTraceDir, msg, err)
+	}
+	return msg, err
+}
+
+type deliverMsgTracer struct {
+	ab.AtomicBroadcast_DeliverServer
+	msgTracer
+}
+
+func (dmt *deliverMsgTracer) Recv() (*cb.Envelope, error) {
+	msg, err := dmt.AtomicBroadcast_DeliverServer.Recv()
+	if traceDir := dmt.debug.DeliverTraceDir; traceDir != "" {
+		dmt.trace(traceDir, msg, err)
+	}
+	return msg, err
+}
+
 // Broadcast receives a stream of messages from a client for ordering
 func (s *server) Broadcast(srv ab.AtomicBroadcast_BroadcastServer) error {
 	logger.Debugf("Starting new Broadcast handler")
@@ -56,7 +117,13 @@ func (s *server) Broadcast(srv ab.AtomicBroadcast_BroadcastServer) error {
 		}
 		logger.Debugf("Closing Broadcast stream")
 	}()
-	return s.bh.Handle(srv)
+	return s.bh.Handle(&broadcastMsgTracer{
+		AtomicBroadcast_BroadcastServer: srv,
+		msgTracer: msgTracer{
+			debug:    s.debug,
+			function: "Broadcast",
+		},
+	})
 }
 
 // Deliver sends a stream of blocks to a client after ordering
@@ -68,5 +135,11 @@ func (s *server) Deliver(srv ab.AtomicBroadcast_DeliverServer) error {
 		}
 		logger.Debugf("Closing Deliver stream")
 	}()
-	return s.dh.Handle(srv)
+	return s.dh.Handle(&deliverMsgTracer{
+		AtomicBroadcast_DeliverServer: srv,
+		msgTracer: msgTracer{
+			debug:    s.debug,
+			function: "Deliver",
+		},
+	})
 }

orderer/common/server/server_test.go

@@ -7,9 +7,26 @@ SPDX-License-Identifier: Apache-2.0
 package server
 
 import (
+	"io/ioutil"
+	"os"
+	"path/filepath"
 	"testing"
+	"time"
+
+	localconfig "github.com/hyperledger/fabric/orderer/common/localconfig"
+	cb "github.com/hyperledger/fabric/protos/common"
+	ab "github.com/hyperledger/fabric/protos/orderer"
+	"github.com/hyperledger/fabric/protos/utils"
+
+	logging "github.com/op/go-logging"
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/grpc"
 )
 
+func init() {
+	logging.SetLevel(logging.DEBUG, "")
+}
+
 func TestBroadcastNoPanic(t *testing.T) {
 	// Defer recovers from the panic
 	_ = (&server{}).Broadcast(nil)
@@ -19,3 +36,102 @@ func TestDeliverNoPanic(t *testing.T) {
 	// Defer recovers from the panic
 	_ = (&server{}).Deliver(nil)
 }
+
+type recvr interface {
+	Recv() (*cb.Envelope, error)
+}
+
+type mockSrv struct {
+	grpc.ServerStream
+	msg *cb.Envelope
+	err error
+}
+
+type mockBroadcastSrv mockSrv
+
+func (mbs *mockBroadcastSrv) Recv() (*cb.Envelope, error) {
+	return mbs.msg, mbs.err
+}
+
+func (mb *mockBroadcastSrv) Send(br *ab.BroadcastResponse) error {
+	panic("Unimplimented")
+}
+
+type mockDeliverSrv mockSrv
+
+func (mds *mockDeliverSrv) Recv() (*cb.Envelope, error) {
+	return mds.msg, mds.err
+}
+
+func (mds *mockDeliverSrv) Send(br *ab.DeliverResponse) error {
+	panic("Unimplimented")
+}
+
+func testMsgTrace(handler func(dir string, msg *cb.Envelope) recvr, t *testing.T) {
+	dir, err := ioutil.TempDir("", "TestMsgTrace")
+	if err != nil {
+		t.Fatalf("Could not create temp dir")
+	}
+	defer os.RemoveAll(dir)
+
+	msg := &cb.Envelope{Payload: []byte("somedata")}
+
+	r := handler(dir, msg)
+
+	rMsg, err := r.Recv()
+	assert.Equal(t, msg, rMsg)
+	assert.Nil(t, err)
+
+	var fileData []byte
+	for i := 0; i < 100; i++ {
+		// Writing the trace file is deliberately non-blocking, wait up to a second, checking every 10 ms to see if the file now exists.
+		time.Sleep(10 * time.Millisecond)
+		filepath.Walk(dir, func(path string, info os.FileInfo, err error) error {
+			assert.Nil(t, err)
+			if path == dir {
+				return nil
+			}
+			assert.Nil(t, fileData, "Should only be one file")
+			fileData, err = ioutil.ReadFile(path)
+			assert.Nil(t, err)
+			return nil
+		})
+		if fileData != nil {
+			break
+		}
+	}
+
+	assert.Equal(t, utils.MarshalOrPanic(msg), fileData)
+}
+
+func TestBroadcastMsgTrace(t *testing.T) {
+	testMsgTrace(func(dir string, msg *cb.Envelope) recvr {
+		return &broadcastMsgTracer{
+			AtomicBroadcast_BroadcastServer: &mockBroadcastSrv{
+				msg: msg,
+			},
+			msgTracer: msgTracer{
+				debug: &localconfig.Debug{
+					BroadcastTraceDir: dir,
+				},
+				function: "Broadcast",
+			},
+		}
+	}, t)
+}
+
+func TestDeliverMsgTrace(t *testing.T) {
+	testMsgTrace(func(dir string, msg *cb.Envelope) recvr {
+		return &deliverMsgTracer{
+			AtomicBroadcast_DeliverServer: &mockDeliverSrv{
+				msg: msg,
+			},
+			msgTracer: msgTracer{
+				debug: &localconfig.Debug{
+					DeliverTraceDir: dir,
+				},
+				function: "Deliver",
+			},
+		}
+	}, t)
+}

protos/peer/proposal.pb.go

@@ -221,6 +221,23 @@ Kafka:
         # following "File" key and specify the file name from which to load the
         # value of RootCAs.
         #File: path/to/RootCAs
-
     # Kafka version of the Kafka cluster brokers (defaults to 0.9.0.1)
     Version:
+        #
+
+################################################################################
+#
+#   Debug Configuration
+#
+#   - This controls the debugging options for the orderer
+#
+################################################################################
+Debug:
+
+    # BroadcasTraceDir when set will cause each request to the Broadcast service
+    # for this orderer to be written to a file in this directory
+    BroadcastTraceDir:
+
+    # DeliverTraceDir when set will cause each request to the Deliver service
+    # for this orderer to be written to a file in this directory
+    DeliverTraceDir: