Merge pull request #2884 from 0chain/fix/separate-validator-for-condu…

…ctor-testing

Fix blobber conductor tests

code/go/0chain.net/conductor/conductor/executor.go

@@ -323,6 +323,7 @@ func (r *Runner) WaitAdd(wadd config.WaitAdd, tm time.Duration) (err error) {
 		// start nodes that haven't been started yet
 		allNodes := append(wadd.Sharders, wadd.Miners...)
 		allNodes = append(allNodes, wadd.Blobbers...)
+		allNodes = append(allNodes, wadd.Validators...)
 		allNodes = append(allNodes, wadd.Authorizers...)
 
 		for _, name := range allNodes {

code/go/0chain.net/conductor/conductor/main.go

@@ -100,6 +100,16 @@ func main() {
 	var success bool
 	// not always error means failure
 
+	// Clean contents of sdxproxy logs directory but keep the directory itself
+	err = os.RemoveAll(conf.Logs + "/sdkproxy")
+	if err != nil {
+		log.Printf("Error while cleaning sdxproxy logs directory: %v", err)
+	}
+	err = os.MkdirAll(conf.Logs+"/sdkproxy", 0777)
+	if err != nil {
+		log.Printf("Error while creating sdxproxy logs directory: %v", err)
+	}
+
 	err, success = r.Run()
 	if err != nil {
 		log.Print("[ERR] ", err)
@@ -228,7 +238,7 @@ func (r *Runner) isWaiting() (tm *time.Timer, ok bool) {
 		fmt.Printf("wait for view change %v\n", r.waitViewChange)
 		return tm, true
 	case !r.waitAdd.IsZero():
-		log.Printf("wait for adding sharders (%+v), miners (%+v), blobbers (%+v) and authorizers (%+v)", r.waitAdd.Sharders, r.waitAdd.Miners, r.waitAdd.Blobbers, r.waitAdd.Authorizers)
+		log.Printf("wait for adding sharders (%+v), miners (%+v), blobbers (%+v), validators (%+v) and authorizers (%+v)", r.waitAdd.Sharders, r.waitAdd.Miners, r.waitAdd.Blobbers, r.waitAdd.Validators, r.waitAdd.Authorizers)
 		return tm, true
 	case !r.waitSharderKeep.IsZero():
 		log.Println("wait for sharder keep")
@@ -555,6 +565,38 @@ func (r *Runner) acceptAddBlobber(addb *conductrpc.AddBlobberEvent) (
 	return
 }
 
+func (r *Runner) acceptAddValidator(addv *conductrpc.AddValidatorEvent) (
+	err error) {
+
+	if addv.Sender != r.monitor {
+		return // not the monitor node
+	}
+	var (
+		sender, sok = r.conf.Nodes.NodeByName(addv.Sender)
+		added, aok  = r.conf.Nodes.NodeByName(addv.Validator)
+	)
+	if !sok {
+		return fmt.Errorf("unexpected add_validator sender: %q", addv.Sender)
+	}
+	if !aok {
+		return fmt.Errorf("unexpected validator %q added by add_validator of %q",
+			addv.Validator, sender.Name)
+	}
+
+	if r.verbose {
+		log.Print(" [INF] add_validator ", added.Name)
+	}
+
+	if r.waitAdd.IsZero() {
+		return // doesn't wait for a node
+	}
+
+	if r.waitAdd.TakeValidator(added.Name) {
+		log.Print("[OK] add_validator ", added.Name)
+	}
+	return
+}
+
 func (r *Runner) acceptAddAuthorizer(addb *conductrpc.AddAuthorizerEvent) (
 	err error) {
 	if addb.Sender != r.monitor {
@@ -869,6 +911,8 @@ func (r *Runner) proceedWaiting() (err error) {
 			err = r.acceptAddSharder(adds)
 		case addb := <-r.server.OnAddBlobber():
 			err = r.acceptAddBlobber(addb)
+		case addv := <-r.server.OnAddValidator():
+			err = r.acceptAddValidator(addv)
 		case adda := <-r.server.OnAddAuthorizer():
 			err = r.acceptAddAuthorizer(adda)
 		case sk := <-r.server.OnSharderKeep():

code/go/0chain.net/conductor/conductrpc/server.go

@@ -61,6 +61,12 @@ type AddBlobberEvent struct {
 	Blobber NodeName // the added blobber
 }
 
+// AddValidatorEvent in miner SC.
+type AddValidatorEvent struct {
+	Sender  NodeName // event emitter
+	Validator NodeName // the added validator
+}
+
 // AddAuthorizerEvent in miner SC.
 type AddAuthorizerEvent struct {
 	Sender     NodeName // event emitter
@@ -114,6 +120,8 @@ type Server struct {
 	onAddSharder chan *AddSharderEvent
 	// onAddBlobber occurs where blobber added in storage SC
 	onAddBlobber chan *AddBlobberEvent
+	// onAddValidator occurs where blobber added in storage SC
+	onAddValidator chan *AddValidatorEvent
 	// onAddAuthorizer occurs where authorizer added in storage SC
 	onAddAuthorizer chan *AddAuthorizerEvent
 	// onSharderKeep occurs where miner SC proceed sharder_keep function
@@ -301,6 +309,10 @@ func (s *Server) OnAddBlobber() chan *AddBlobberEvent {
 	return s.onAddBlobber
 }
 
+func (s *Server) OnAddValidator() chan *AddValidatorEvent {
+	return s.onAddValidator
+}
+
 func (s *Server) OnAddAuthorizer() chan *AddAuthorizerEvent {
 	return s.onAddAuthorizer
 }

code/go/0chain.net/conductor/config/waiters.go

@@ -100,6 +100,7 @@ type WaitAdd struct {
 	Miners      []NodeName `json:"miners" yaml:"miners" mapstructure:"miners"`
 	Sharders    []NodeName `json:"sharders" yaml:"sharders" mapstructure:"sharders"`
 	Blobbers    []NodeName `json:"blobbers" yaml:"blobbers" mapstructure:"blobbers"`
+	Validators  []NodeName `json:"validators" yaml:"validators" mapstructure:"validators"`
 	Authorizers []NodeName `json:"authorizers" yaml:"authorizers" mapstructure:"authorizers"`
 	Start       bool       `json:"start" yaml:"start" mapstructure:"start"`
 }
@@ -115,6 +116,8 @@ func (wa *WaitAdd) Take(name NodeName) (ok bool) {
 		return wa.TakeSharder(name)
 	} else if strings.Contains(string(name), "blobber") {
 		return wa.TakeBlobber(name)
+	} else if strings.Contains(string(name), "validator") {
+		return wa.TakeValidator(name)	
 	} else if strings.Contains(string(name), "authorizer") {
 		return wa.TakeAuthorizer(name)
 	}
@@ -152,6 +155,16 @@ func (wa *WaitAdd) TakeBlobber(name NodeName) (ok bool) {
 	return
 }
 
+func (wa *WaitAdd) TakeValidator(name NodeName) (ok bool) {
+	for i, validatorName := range wa.Validators {
+		if validatorName == name {
+			wa.Validators = append(wa.Validators[:i], wa.Validators[i+1:]...)
+			return true
+		}
+	}
+	return
+}
+
 func (wa *WaitAdd) TakeAuthorizer(name NodeName) (ok bool) {
 	for i, authorizerName := range wa.Authorizers {
 		if authorizerName == name {

code/go/0chain.net/conductor/sdkproxy/main.go

@@ -9,14 +9,17 @@ import (
 	"fmt"
 	"io"
 	"log"
+	"math/rand"
 	"mime/multipart"
 	"net/http"
 	"os"
 	"os/exec"
 	"os/signal"
+	"path/filepath"
 	"strings"
 	"sync"
 	"syscall"
+	"time"
 )
 
 const (
@@ -394,23 +397,46 @@ func waitSigInt() {
 	log.Printf("got signal %s, exiting...", <-c)
 }
 
-func execute(r, address string, codes chan int) {
+func execute(r, address string, codes chan int, logsDir string) {
 	var (
-		cmd  = exec.Command("sh", "-x", r)
+		ctx, cancel = context.WithTimeout(context.Background(), 2 * time.Minute)
+		cmd  = exec.CommandContext(ctx, "sh", "-x", r)
 		err  error
 		code int
 	)
 
+	defer cancel()
+
 	log.Print("execute: ", r)
 	defer func() { log.Printf("executed (%s) with %d exit code", r, code) }()
 
-	cmd.Stdout = os.Stdout
-	cmd.Stderr = os.Stderr
+	executableName := filepath.Base(r)
+	logFileBaseName := fmt.Sprintf("%v-%v.log", executableName, time.Now().Local().Format("2006-01-02_15-04-05"))
+
+	foutPath := filepath.Join(logsDir, fmt.Sprintf("sdkproxy/stdout-%v.log", logFileBaseName))
+	fout, err := os.Create(foutPath)
+	if err != nil {
+		log.Printf("creating temp file: %v", err)
+		return
+	}
+	defer fout.Close()
+
+	ferrPath := filepath.Join(logsDir, fmt.Sprintf("sdkproxy/stderr-%v.log", logFileBaseName))
+	ferr, err := os.Create(ferrPath)
+	if err != nil {
+		log.Printf("creating temp file: %v", err)
+		return
+	}
+	defer ferr.Close()
+
+	cmd.Stdout = fout
+	cmd.Stderr = ferr
 	cmd.Env = append(os.Environ(), "HTTP_PROXY=http://"+address)
 
 	err = cmd.Run()
 	if err != nil {
 		if ee, ok := err.(*exec.ExitError); ok {
+
 			code = ee.ExitCode()
 		}
 		log.Printf("executing %s: %v", r, err)
@@ -423,16 +449,18 @@ func main() {
 
 	// address
 	var (
+		logsDir string = ""				 // logs directory
 		markers string = ""              // markers arriving order
 		filter  string = ""              // filter multipart forms fields
-		addr    string = "0.0.0.0:15211" // bind
+		addr    string = fmt.Sprintf("0.0.0.0:%v", rand.Intn(65535-10000) + 10000) // bind
 
 		back = context.Background() //
 
 		s   http.Server // server instance
 		run Run         // run parallel with HTTP_PROXY
 	)
 
+	flag.StringVar(&logsDir, "l", logsDir, "logs directory")
 	flag.StringVar(&markers, "m", markers, "markers arriving order")
 	flag.StringVar(&filter, "f", filter, "filter multipart form fields")
 	flag.StringVar(&addr, "a", addr, "bind proxy address")
@@ -469,20 +497,16 @@ func main() {
 
 	// start the proxy
 	go func() { log.Fatal(s.ListenAndServe()) }()
-	defer func() {
-		if err := s.Shutdown(back); err != nil {
-			log.Printf("shutdown error: %\ns", err)
-		}
-	}()
 
 	if len(run) == 0 {
 		waitSigInt()
+		cleanup(back, &s)
 		return
 	}
 
 	var codes = make(chan int, len(run))
 	for _, r := range run {
-		go execute(r, addr, codes)
+		go execute(r, addr, codes, logsDir)
 	}
 
 	var code int
@@ -493,9 +517,17 @@ func main() {
 		}
 	}
 
+	cleanup(back, &s)
 	os.Exit(code)
 }
 
+
+func cleanup(ctx context.Context, s *http.Server) {
+	log.Println("shutdown server")
+	if err := s.Shutdown(ctx); err != nil {
+		log.Printf("shutdown error: %\ns", err)
+	}
+}
 // ========================================================================== //
 //                                    note                                    //
 // ========================================================================== //

docker.local/bin/conductor/cleanup.sh

@@ -0,0 +1,19 @@
+#!/bin/bash
+
+cd 0chain;
+rm -rf docker.local/miner*/;
+rm -rf docker.local/sharder*/;
+./docker.local/bin/init.setup.sh;
+
+cd ../blobber;
+rm -rf docker.local/blobber*/;
+rm -rf docker.local/validator*/;
+./docker.local/bin/blobber.init.setup.sh;
+
+cd ../0dns;
+rm -rf docker.local/0dns/;
+./docker.local/bin/init.sh;
+
+cd ../0box;
+rm -rf docker.local/0box/;
+./docker.local/bin/init.sh;

docker.local/bin/conductor/download_no_rm.sh

@@ -3,12 +3,11 @@
 set -e
 
 # add to read pools
-./zboxcli/zbox --wallet testing.json rp-lock \
- --allocation "$(cat ~/.zcn/allocation.txt)" --tokens 2.0
+./zboxcli/zbox --wallet testing.json rp-lock --tokens 2.0
 
 # auth user
 # ./zboxcli/zbox --wallet testing-auth.json rp-lock \
 #     --allocation "$(cat ~/.zcn/allocation.txt)" --tokens 2.0
 
-go run 0chain/code/go/0chain.net/conductor/sdkproxy/main.go -f read_marker \
+go run 0chain/code/go/0chain.net/conductor/sdkproxy/main.go -f read_marker -l "0chain/conductor/logs" \
     -run 0chain/docker.local/bin/conductor/proxied/download_b.sh

docker.local/bin/conductor/init_allocation_3_blobbers_no_read_tokens.sh

@@ -0,0 +1,53 @@
+#!/bin/sh
+
+set -e
+
+data_shards=$1
+parity_shards=$2
+
+rm -rf ~/.zcn/testing.json
+rm -rf ~/.zcn/allocation.txt
+
+printf '{"client_id":"1746b06bb09f55ee01b33b5e2e055d6cc7a900cb57c0a3a5eaabb8a0e7745802","client_key":"7b630ba670dac2f22d43c2399b70eff378689a53ee03ea20957bb7e73df016200fea410ba5102558b0c39617e5afd2c1843b161a1dedec15e1ab40543a78a518","keys":[{"public_key":"7b630ba670dac2f22d43c2399b70eff378689a53ee03ea20957bb7e73df016200fea410ba5102558b0c39617e5afd2c1843b161a1dedec15e1ab40543a78a518","private_key":"c06b6f6945ba02d5a3be86b8779deca63bb636ce7e46804a479c50e53c864915"}],"mnemonics":"cactus panther essence ability copper fox wise actual need cousin boat uncover ride diamond group jacket anchor current float rely tragic omit child payment","version":"1.0","date_created":"2021-08-04 18:53:56.949069945 +0100 BST m=+0.018986002"}' > ~/.zcn/testing.json
+
+
+for run in $(seq 1 28)
+do
+  ./zwalletcli/zwallet --wallet testing.json faucet \
+      --methodName pour --input "{Pay day}"
+done
+
+./zwalletcli/zwallet --wallet testing.json getbalance
+
+BLOBBER1=f65af5d64000c7cd2883f4910eb69086f9d6e6635c744e62afcfab58b938ee25
+BLOBBER2=7a90e6790bcd3d78422d7a230390edc102870fe58c15472073922024985b1c7d
+BLOBBER3=2f051ca6447d8712a020213672bece683dbd0d23a81fdf93ff273043a0764d18
+
+# stake blobbers
+./zboxcli/zbox --wallet testing.json sp-lock \
+    --blobber_id $BLOBBER1 --tokens 2
+./zboxcli/zbox --wallet testing.json sp-lock \
+    --blobber_id $BLOBBER2 --tokens 2
+./zboxcli/zbox --wallet testing.json sp-lock \
+    --blobber_id $BLOBBER3 --tokens 2
+
+
+VALIDATOR1=41313b795d2c057b6277801e9ed277b444770c2af75f5209afd00bd07c72cc0b
+VALIDATOR2=ab549edb7cea822dab0b460f65dcde85f698c1e97d730e3ffc6b0f8b576b65bd
+VALIDATOR3=86cf791f03f01e3e4d318b1ca009a51c91dd43f7cf3c87a32f531b609cc5044b
+
+./zboxcli/zbox --wallet testing.json sp-lock \
+    --validator_id $VALIDATOR1 --tokens 2
+./zboxcli/zbox --wallet testing.json sp-lock \
+    --validator_id $VALIDATOR2 --tokens 2
+./zboxcli/zbox --wallet testing.json sp-lock \
+    --validator_id $VALIDATOR3 --tokens 2
+
+# for test logs
+./zboxcli/zbox --wallet testing.json ls-blobbers
+
+# create allocation
+./zboxcli/zbox --wallet testing.json newallocation \
+    --read_price 0.001-10 --write_price 0.01-10 --size 104857600 --lock 2 \
+    --data $data_shards --parity $parity_shards
+

docker.local/bin/conductor/init_allocation_only.sh

@@ -10,6 +10,7 @@ rm -rf ~/.zcn/allocation.txt
 
 printf '{"client_id":"1746b06bb09f55ee01b33b5e2e055d6cc7a900cb57c0a3a5eaabb8a0e7745802","client_key":"7b630ba670dac2f22d43c2399b70eff378689a53ee03ea20957bb7e73df016200fea410ba5102558b0c39617e5afd2c1843b161a1dedec15e1ab40543a78a518","keys":[{"public_key":"7b630ba670dac2f22d43c2399b70eff378689a53ee03ea20957bb7e73df016200fea410ba5102558b0c39617e5afd2c1843b161a1dedec15e1ab40543a78a518","private_key":"c06b6f6945ba02d5a3be86b8779deca63bb636ce7e46804a479c50e53c864915"}],"mnemonics":"cactus panther essence ability copper fox wise actual need cousin boat uncover ride diamond group jacket anchor current float rely tragic omit child payment","version":"1.0","date_created":"2021-08-04 18:53:56.949069945 +0100 BST m=+0.018986002"}' > ~/.zcn/testing.json
 
+
 for run in $(seq 1 28)
 do
   ./zwalletcli/zwallet --wallet testing.json faucet \

docker.local/bin/conductor/markers_out_of_order.sh

@@ -31,7 +31,7 @@ head -c 32430848 < /dev/urandom > b2.bin
     --remotepath=/remote/b1.bin
 
 # then update both files, download and delete then
-go run 0chain/code/go/0chain.net/conductor/sdkproxy/main.go -m 'dm-rm-wm' \
+go run 0chain/code/go/0chain.net/conductor/sdkproxy/main.go -m 'dm-rm-wm' -l "0chain/conductor/logs" \
     -run 0chain/docker.local/bin/conductor/proxied/update_b.sh            \
     -run 0chain/docker.local/bin/conductor/proxied/download_b.sh          \
     -run 0chain/docker.local/bin/conductor/proxied/delete_b.sh