btcd,integration,integration/rpctest: integrate debugstream with btcd

btcd: integrated debugstream, now we can send debug events and improve
testing capabilities. The hidden flag --debugstream=<host:port> starts
the debug Stream when btcd is compiled with the debug tag.

integration: added debugstream integration test.

integration/rpctest: added DebugStreamHandler to harness options,
allowing integration tests to easily process debug events. The btcd
executable build to be used by the Harness is now being built with the
debug tag, but the debug stream will be started only if the
DebugStreamHandler is also passed to the Harness.

Author: allocz

btcd.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/btcsuite/btcd/blockchain/indexers"
 	"github.com/btcsuite/btcd/database"
+	"github.com/btcsuite/btcd/debugstream"
 	"github.com/btcsuite/btcd/limits"
 	"github.com/btcsuite/btcd/ossec"
 )
@@ -56,6 +57,24 @@ func btcdMain(serverChan chan<- *server) error {
 		}
 	}()
 
+	// DebugStream is enabled only if btcd is compiled with the debug tag.
+	// Otherwise a nop implementation is used.
+	debugstream.S = debugstream.New()
+	if dsListen := cfg.DebugStream; dsListen != "" {
+		err := debugstream.S.Listen(dsListen)
+		if err != nil {
+			return fmt.Errorf("error starting debug stream: %v",
+				err)
+		}
+		debugstream.S.Broadcast(debugstream.Event{
+			Code: debugstream.DEStart,
+		})
+		defer debugstream.S.Shutdown()
+		defer debugstream.S.Broadcast(debugstream.Event{
+			Code: debugstream.DEShutdown,
+		})
+	}
+
 	// Get a channel that will be closed when a shutdown signal has been
 	// triggered either from an OS signal such as SIGINT (Ctrl+C) or from
 	// another subsystem such as the RPC server.

config.go

@@ -120,6 +120,7 @@ type config struct {
 	DataDir              string        `short:"b" long:"datadir" description:"Directory to store data"`
 	DbType               string        `long:"dbtype" description:"Database backend to use for the Block Chain"`
 	DebugLevel           string        `short:"d" long:"debuglevel" description:"Logging level for all subsystems {trace, debug, info, warn, error, critical} -- You may also specify <subsystem>=<level>,<subsystem2>=<level>,... to set the log level for individual subsystems -- Use show to list available subsystems"`
+	DebugStream          string        `long:"debugstream" hidden:"true" description:"TCP listen address of the debug stream. To use this feature btcd must also be compiled with debug tag."`
 	DropAddrIndex        bool          `long:"dropaddrindex" description:"Deletes the address-based transaction index from the database on start up and then exits."`
 	DropCfIndex          bool          `long:"dropcfindex" description:"Deletes the index used for committed filtering (CF) support from the database on start up and then exits."`
 	DropTxIndex          bool          `long:"droptxindex" description:"Deletes the hash-based transaction index from the database on start up and then exits."`

integration/debugstream_test.go

@@ -0,0 +1,46 @@
+//go:build rpctest
+
+package integration
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/btcsuite/btcd/debugstream"
+	"github.com/btcsuite/btcd/integration/rpctest"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDebugStream(t *testing.T) {
+	const (
+		sBegin = iota
+		sNodeStarted
+		sNodeShutdown
+	)
+	var state uint64
+	ctx, cancel := context.WithTimeout(t.Context(), time.Second*10)
+	defer cancel()
+	debHandler := func(e debugstream.Event) {
+		switch e.Code {
+		case debugstream.DEStart:
+			state = sNodeStarted
+
+		case debugstream.DEShutdown:
+			state = sNodeShutdown
+			cancel()
+		}
+	}
+
+	h, err := rpctest.New(rpctest.HarnessOpts{
+		DebugStreamHandler: debHandler,
+	})
+	require.NoError(t, err)
+	h.SetUp(false, 0)
+
+	err = h.TearDown()
+	require.NoError(t, err)
+
+	<-ctx.Done()
+	require.Equal(t, true, sNodeShutdown == state)
+}

integration/rpctest/btcd.go

@@ -53,9 +53,8 @@ func btcdExecutablePath() (string, error) {
 	if runtime.GOOS == "windows" {
 		outputPath += ".exe"
 	}
-	cmd := exec.Command(
-		"go", "build", "-o", outputPath, "github.com/btcsuite/btcd",
-	)
+	cmd := exec.Command("go", "build", "-tags=debug", "-o", outputPath,
+		"github.com/btcsuite/btcd")
 	err = cmd.Run()
 	if err != nil {
 		return "", fmt.Errorf("Failed to build btcd: %v", err)

integration/rpctest/rpc_harness.go

@@ -22,6 +22,7 @@ import (
 	"github.com/btcsuite/btcd/btcutil/v2"
 	"github.com/btcsuite/btcd/chaincfg/v2"
 	"github.com/btcsuite/btcd/chainhash/v2"
+	"github.com/btcsuite/btcd/debugstream"
 	"github.com/btcsuite/btcd/rpcclient"
 	"github.com/btcsuite/btcd/wire/v2"
 )
@@ -126,6 +127,8 @@ type Harness struct {
 	testNodeDir string
 	nodeNum     int
 
+	debugClient *debugstream.Client
+
 	sync.Mutex
 }
 
@@ -142,6 +145,11 @@ type HarnessOpts struct {
 	// CustomExePath can be set to the path of a custom btcd executable,
 	// otherwise we build and use a default executable.
 	CustomExePath string
+
+	// DebugStreamHandler makes the harness to start btcd with the debug
+	// stream enabled, and use the DebugStreamHandler as the handler for
+	// the debug events.
+	DebugStreamHandler func(debugstream.Event)
 }
 
 // New creates and initializes new instance of the rpc test harness.
@@ -201,6 +209,15 @@ func New(opts ...HarnessOpts) (*Harness, error) {
 	miningAddr := fmt.Sprintf("--miningaddr=%s", wallet.coinbaseAddr)
 	o.ExtraArgs = append(o.ExtraArgs, miningAddr)
 
+	var debugClient *debugstream.Client
+	if o.DebugStreamHandler != nil {
+		p := NextAvailablePort()
+		streamAddr := fmt.Sprintf("127.0.0.1:%d", p)
+		debugClient = debugstream.NewClient(streamAddr,
+			o.DebugStreamHandler)
+		o.ExtraArgs = append(o.ExtraArgs, "--debugstream="+streamAddr)
+	}
+
 	config, err := newConfig(testNodeDir, certFile, keyFile, o.ExtraArgs,
 		o.CustomExePath)
 	if err != nil {
@@ -260,6 +277,7 @@ func New(opts ...HarnessOpts) (*Harness, error) {
 		ActiveNet:              o.ActiveNet,
 		nodeNum:                nodeNum,
 		wallet:                 wallet,
+		debugClient:            debugClient,
 	}
 
 	// Track this newly created test instance within the package level
@@ -285,6 +303,12 @@ func (h *Harness) SetUp(createTestChain bool, numMatureOutputs uint32) error {
 	if err := h.connectRPCClient(); err != nil {
 		return fmt.Errorf("error connecting RPC client: %w", err)
 	}
+	if h.debugClient != nil {
+		if err := h.debugClient.Start(); err != nil {
+			return fmt.Errorf("error connecting debug client: %w",
+				err)
+		}
+	}
 
 	h.wallet.Start()
 
@@ -347,6 +371,13 @@ func (h *Harness) tearDown(skipCleanup bool) error {
 		h.BatchClient.WaitForShutdown()
 	}
 
+	if h.debugClient != nil {
+		// Is better to stop the client after stopping btcd, because
+		// this way we can use the debugHandler to test the btcd
+		// shutdown behavior.
+		defer h.debugClient.Stop()
+	}
+
 	// In the case of exit errors we still perform the cleanup, but also
 	// return the exit error to the caller.
 	exitErr := &exec.ExitError{}

log.go

@@ -15,6 +15,7 @@ import (
 	"github.com/btcsuite/btcd/blockchain/indexers"
 	"github.com/btcsuite/btcd/connmgr"
 	"github.com/btcsuite/btcd/database"
+	"github.com/btcsuite/btcd/debugstream"
 	"github.com/btcsuite/btcd/mempool"
 	"github.com/btcsuite/btcd/mining"
 	"github.com/btcsuite/btcd/mining/cpuminer"
@@ -61,6 +62,8 @@ var (
 	bcdbLog = backendLog.Logger("BCDB")
 	btcdLog = backendLog.Logger("BTCD")
 	chanLog = backendLog.Logger("CHAN")
+	debsLog = backendLog.Logger("DEBS")
+	debcLog = backendLog.Logger("DEBC")
 	discLog = backendLog.Logger("DISC")
 	indxLog = backendLog.Logger("INDX")
 	minrLog = backendLog.Logger("MINR")
@@ -79,6 +82,7 @@ func init() {
 	connmgr.UseLogger(cmgrLog)
 	database.UseLogger(bcdbLog)
 	blockchain.UseLogger(chanLog)
+	debugstream.UseLoggers(debsLog, debcLog)
 	indexers.UseLogger(indxLog)
 	mining.UseLogger(minrLog)
 	cpuminer.UseLogger(minrLog)
@@ -97,6 +101,8 @@ var subsystemLoggers = map[string]btclog.Logger{
 	"BCDB":                bcdbLog,
 	"BTCD":                btcdLog,
 	"CHAN":                chanLog,
+	"DEBS":                debsLog,
+	"DEBC":                debcLog,
 	"DISC":                discLog,
 	"INDX":                indxLog,
 	"MINR":                minrLog,