Merge pull request #4901 from bhandras/etcd_itest_ci

etcd+itests: use itest generated ports for embedded etcd and add etcd itests to travis CI

.travis.yml

@@ -63,6 +63,11 @@ jobs:
         - bash ./scripts/install_bitcoind.sh
         - make itest-parallel backend=bitcoind
 
+    - name: Bitcoind Integration with etcd (txindex enabled)
+      script:
+        - bash ./scripts/install_bitcoind.sh
+        - make itest-parallel backend=bitcoind etcd=1
+
     - name: Bitcoind Integration (txindex disabled)
       script:
         - bash ./scripts/install_bitcoind.sh

Makefile

@@ -184,7 +184,7 @@ itest: build-itest itest-only
 itest-parallel: build-itest
 	@$(call print, "Running tests")
 	rm -rf lntest/itest/*.log lntest/itest/.logs-*; date
-	EXEC_SUFFIX=$(EXEC_SUFFIX) echo "$$(seq 0 $$(expr $(ITEST_PARALLELISM) - 1))" | xargs -P $(ITEST_PARALLELISM) -n 1 -I {} scripts/itest_part.sh {} $(NUM_ITEST_TRANCHES) $(TEST_FLAGS)
+	EXEC_SUFFIX=$(EXEC_SUFFIX) echo "$$(seq 0 $$(expr $(ITEST_PARALLELISM) - 1))" | xargs -P $(ITEST_PARALLELISM) -n 1 -I {} scripts/itest_part.sh {} $(NUM_ITEST_TRANCHES) $(TEST_FLAGS) $(ITEST_FLAGS)
 	lntest/itest/log_check_errors.sh
 
 unit: btcd

channeldb/kvdb/backend.go

@@ -253,7 +253,7 @@ func GetTestBackend(path, name string) (Backend, func(), error) {
 		}
 		return db, empty, nil
 	} else if TestBackend == EtcdBackendName {
-		return GetEtcdTestBackend(path, name)
+		return GetEtcdTestBackend(path, 0, 0)
 	}
 
 	return nil, nil, fmt.Errorf("unknown backend")

channeldb/kvdb/config.go

@@ -36,7 +36,11 @@ type BoltConfig struct {
 
 // EtcdConfig holds etcd configuration.
 type EtcdConfig struct {
-	Embedded bool `long:"embedded" description:"Use embedded etcd instance instead of the external one."`
+	Embedded bool `long:"embedded" description:"Use embedded etcd instance instead of the external one. Note: use for testing only."`
+
+	EmbeddedClientPort uint16 `long:"embedded_client_port" description:"Client port to use for the embedded instance. Note: use for testing only."`
+
+	EmbeddedPeerPort uint16 `long:"embedded_peer_port" description:"Peer port to use for the embedded instance. Note: use for testing only."`
 
 	Host string `long:"host" description:"Etcd database host."`
 

channeldb/kvdb/etcd/db.go

@@ -304,15 +304,3 @@ func (db *db) Copy(w io.Writer) error {
 func (db *db) Close() error {
 	return db.cli.Close()
 }
-
-// Batch opens a database read/write transaction and executes the function f
-// with the transaction passed as a parameter.  After f exits, if f did not
-// error, the transaction is committed.  Otherwise, if f did error, the
-// transaction is rolled back.  If the rollback fails, the original error
-// returned by f is still returned.  If the commit fails, the commit error is
-// returned.
-//
-// Batch is only useful when there are multiple goroutines calling it.
-func (db *db) Batch(apply func(tx walletdb.ReadWriteTx) error) error {
-	return db.Update(apply, func() {})
-}

channeldb/kvdb/etcd/embed.go

@@ -61,7 +61,9 @@ func getFreePort() int {
 // NewEmbeddedEtcdInstance creates an embedded etcd instance for testing,
 // listening on random open ports. Returns the backend config and a cleanup
 // func that will stop the etcd instance.
-func NewEmbeddedEtcdInstance(path string) (*BackendConfig, func(), error) {
+func NewEmbeddedEtcdInstance(path string, clientPort, peerPort uint16) (
+	*BackendConfig, func(), error) {
+
 	cfg := embed.NewConfig()
 	cfg.Dir = path
 
@@ -69,9 +71,17 @@ func NewEmbeddedEtcdInstance(path string) (*BackendConfig, func(), error) {
 	cfg.MaxTxnOps = 8192
 	cfg.MaxRequestBytes = 16384 * 1024
 
-	// Listen on random free ports.
-	clientURL := fmt.Sprintf("127.0.0.1:%d", getFreePort())
-	peerURL := fmt.Sprintf("127.0.0.1:%d", getFreePort())
+	// Listen on random free ports if no ports were specified.
+	if clientPort == 0 {
+		clientPort = uint16(getFreePort())
+	}
+
+	if peerPort == 0 {
+		peerPort = uint16(getFreePort())
+	}
+
+	clientURL := fmt.Sprintf("127.0.0.1:%d", clientPort)
+	peerURL := fmt.Sprintf("127.0.0.1:%d", peerPort)
 	cfg.LCUrls = []url.URL{{Host: clientURL}}
 	cfg.LPUrls = []url.URL{{Host: peerURL}}
 

channeldb/kvdb/etcd/fixture_test.go

@@ -32,7 +32,7 @@ type EtcdTestFixture struct {
 func NewTestEtcdInstance(t *testing.T, path string) (*BackendConfig, func()) {
 	t.Helper()
 
-	config, cleanup, err := NewEmbeddedEtcdInstance(path)
+	config, cleanup, err := NewEmbeddedEtcdInstance(path, 0, 0)
 	if err != nil {
 		t.Fatalf("error while staring embedded etcd instance: %v", err)
 	}

channeldb/kvdb/interface.go

@@ -44,8 +44,17 @@ func View(db Backend, f func(tx RTx) error, reset func()) error {
 // Batch is identical to the Update call, but it attempts to combine several
 // individual Update transactions into a single write database transaction on
 // an optimistic basis. This only has benefits if multiple goroutines call
-// Batch.
-var Batch = walletdb.Batch
+// Batch. For etcd Batch simply does an Update since combination is more complex
+// in that case due to STM retries.
+func Batch(db Backend, f func(tx RwTx) error) error {
+	if extendedDB, ok := db.(ExtendedBackend); ok {
+		// Since Batch calls handle external state reset, we can safely
+		// pass in an empty reset closure.
+		return extendedDB.Update(f, func() {})
+	}
+
+	return walletdb.Batch(db, f)
+}
 
 // Create initializes and opens a database for the specified type. The
 // arguments are specific to the database type driver. See the documentation

channeldb/kvdb/kvdb_etcd.go

@@ -37,10 +37,14 @@ func GetEtcdBackend(ctx context.Context, prefix string,
 
 // GetEtcdTestBackend creates an embedded etcd backend for testing
 // storig the database at the passed path.
-func GetEtcdTestBackend(path, name string) (Backend, func(), error) {
+func GetEtcdTestBackend(path string, clientPort, peerPort uint16) (
+	Backend, func(), error) {
+
 	empty := func() {}
 
-	config, cleanup, err := etcd.NewEmbeddedEtcdInstance(path)
+	config, cleanup, err := etcd.NewEmbeddedEtcdInstance(
+		path, clientPort, peerPort,
+	)
 	if err != nil {
 		return nil, empty, err
 	}

channeldb/kvdb/kvdb_no_etcd.go

@@ -22,6 +22,8 @@ func GetEtcdBackend(ctx context.Context, prefix string,
 
 // GetTestEtcdBackend  is a stub returning nil, an empty closure and an
 // errEtcdNotAvailable error.
-func GetEtcdTestBackend(path, name string) (Backend, func(), error) {
+func GetEtcdTestBackend(path string, clientPort, peerPort uint16) (
+	Backend, func(), error) {
+
 	return nil, func() {}, errEtcdNotAvailable
 }

contractcourt/chain_watcher.go

@@ -1128,19 +1128,6 @@ func (c *chainWatcher) dispatchContractBreach(spendEvent *chainntnfs.SpendDetail
 		return err
 	}
 
-	// With the event processed, we'll now notify all subscribers of the
-	// event.
-	c.Lock()
-	for _, sub := range c.clientSubscriptions {
-		select {
-		case sub.ContractBreach <- retribution:
-		case <-c.quit:
-			c.Unlock()
-			return fmt.Errorf("quitting")
-		}
-	}
-	c.Unlock()
-
 	// At this point, we've successfully received an ack for the breach
 	// close. We now construct and persist  the close summary, marking the
 	// channel as pending force closed.
@@ -1182,6 +1169,19 @@ func (c *chainWatcher) dispatchContractBreach(spendEvent *chainntnfs.SpendDetail
 	log.Infof("Breached channel=%v marked pending-closed",
 		c.cfg.chanState.FundingOutpoint)
 
+	// With the event processed and channel closed, we'll now notify all
+	// subscribers of the event.
+	c.Lock()
+	for _, sub := range c.clientSubscriptions {
+		select {
+		case sub.ContractBreach <- retribution:
+		case <-c.quit:
+			c.Unlock()
+			return fmt.Errorf("quitting")
+		}
+	}
+	c.Unlock()
+
 	return nil
 }
 

lncfg/db.go

@@ -83,10 +83,15 @@ func (db *DB) GetBackends(ctx context.Context, dbPath string,
 
 	if db.Backend == EtcdBackend {
 		if db.Etcd.Embedded {
-			remoteDB, _, err = kvdb.GetEtcdTestBackend(dbPath, dbName)
+			remoteDB, _, err = kvdb.GetEtcdTestBackend(
+				dbPath, db.Etcd.EmbeddedClientPort,
+				db.Etcd.EmbeddedPeerPort,
+			)
 		} else {
 			// Prefix will separate key/values in the db.
-			remoteDB, err = kvdb.GetEtcdBackend(ctx, networkName, db.Etcd)
+			remoteDB, err = kvdb.GetEtcdBackend(
+				ctx, networkName, db.Etcd,
+			)
 		}
 		if err != nil {
 			return nil, err

lntest/node.go

@@ -298,6 +298,19 @@ func (cfg NodeConfig) genArgs() []string {
 	if cfg.Etcd {
 		args = append(args, "--db.backend=etcd")
 		args = append(args, "--db.etcd.embedded")
+		args = append(
+			args, fmt.Sprintf(
+				"--db.etcd.embedded_client_port=%v",
+				nextAvailablePort(),
+			),
+		)
+		args = append(
+			args, fmt.Sprintf(
+				"--db.etcd.embedded_peer_port=%v",
+				nextAvailablePort(),
+			),
+		)
+		args = append(args, "--db.etcd.embedded")
 	}
 
 	if cfg.FeeURL != "" {

sample-lnd.conf

@@ -995,6 +995,12 @@ litecoin.node=ltcd
 ; Useful for testing.
 ; db.etcd.embedded=false
 
+; If non zero, LND will use this as client port for the embedded etcd instance.
+; db.etcd.embedded_client_port=1234
+
+; If non zero, LND will use this as peer port for the embedded etcd instance.
+; db.etcd.embedded_peer_port=1235
+
 [bolt]
 ; If true, prevents the database from syncing its freelist to disk. 
 ; db.bolt.nofreelistsync=1