daemon: Remove bp from indexer daemon (#1480)

* Remove bp from indexer daemon

Author: Eric-Warehime

cmd/algorand-indexer/daemon.go

@@ -21,7 +21,6 @@ import (
 	"github.com/algorand/indexer/conduit/pipeline"
 	_ "github.com/algorand/indexer/conduit/plugins/exporters/postgresql"
 	_ "github.com/algorand/indexer/conduit/plugins/importers/algod"
-	_ "github.com/algorand/indexer/conduit/plugins/processors/blockprocessor"
 	"github.com/algorand/indexer/config"
 	"github.com/algorand/indexer/fetcher"
 	"github.com/algorand/indexer/idb"
@@ -380,18 +379,7 @@ func makeConduitConfig(dCfg *daemonConfig, nextRound uint64) pipeline.Config {
 			Config: map[string]interface{}{
 				"netaddr": dCfg.algodAddr,
 				"token":   dCfg.algodToken,
-			},
-		},
-		Processors: []pipeline.NameConfigPair{
-			{
-				Name: "block_evaluator",
-				Config: map[string]interface{}{
-					"catchpoint":     dCfg.catchpoint,
-					"data-dir":       dCfg.indexerDataDir,
-					"algod-data-dir": dCfg.algodDataDir,
-					"algod-token":    dCfg.algodToken,
-					"algod-addr":     dCfg.algodAddr,
-				},
+				"mode":    "follower",
 			},
 		},
 		Exporter: pipeline.NameConfigPair{

conduit/plugins/importers/algod/algod_importer.go

@@ -18,6 +18,7 @@ import (
 	"github.com/algorand/indexer/data"
 
 	"github.com/algorand/go-algorand-sdk/v2/client/v2/algod"
+	"github.com/algorand/go-algorand-sdk/v2/client/v2/common"
 	"github.com/algorand/go-algorand-sdk/v2/client/v2/common/models"
 	"github.com/algorand/go-algorand-sdk/v2/encoding/json"
 	"github.com/algorand/go-algorand-sdk/v2/encoding/msgpack"
@@ -155,6 +156,20 @@ func (algodImp *algodImporter) Close() error {
 	return nil
 }
 
+func (algodImp *algodImporter) getDelta(rnd uint64) (sdk.LedgerStateDelta, error) {
+	var delta sdk.LedgerStateDelta
+	params := struct {
+		Format string `url:"format,omitempty"`
+	}{Format: "msgp"}
+	bytes, err := (*common.Client)(algodImp.aclient).GetRaw(algodImp.ctx, fmt.Sprintf("/v2/deltas/%d", rnd), params, nil)
+	if err != nil {
+		return delta, err
+	}
+
+	err = msgpack.Decode(bytes, &delta)
+	return delta, err
+}
+
 func (algodImp *algodImporter) GetBlock(rnd uint64) (data.BlockData, error) {
 	var blockbytes []byte
 	var err error
@@ -195,7 +210,7 @@ func (algodImp *algodImporter) GetBlock(rnd uint64) (data.BlockData, error) {
 			// else converted over
 			// Round 0 has no delta associated with it
 			if rnd != 0 {
-				delta, err := algodImp.aclient.GetLedgerStateDelta(rnd).Do(algodImp.ctx)
+				delta, err := algodImp.getDelta(rnd)
 				if err != nil {
 					algodImp.logger.Errorf(
 						"r=%d error getting delta %d", r, rnd)

e2e_tests/src/e2e_indexer/e2elive.py

@@ -124,6 +124,18 @@ def main():
         os.path.join(tempdir, "net", "Primary", "*", "*.block.sqlite")
     )
     lastblock = countblocks(blockfiles[0])
+    # Reset the secondary node, and enable follow mode.
+    # This is what conduit will connect to for data access.
+    for root, dirs, files in os.walk(os.path.join(tempnet, 'Node', 'tbd-v1')):
+        for f in files:
+            if ".sqlite" in f:
+                os.remove(os.path.join(root, f))
+    cf = {}
+    with open(os.path.join(tempnet, "Node", "config.json"), "r") as config_file:
+        cf = json.load(config_file)
+        cf['EnableFollowMode'] = True
+    with open(os.path.join(tempnet, "Node", "config.json"), "w") as config_file:
+        config_file.write(json.dumps(cf))
     try:
         xrun(["goal", "network", "start", "-r", tempnet])
     except Exception:
@@ -146,7 +158,7 @@ def main():
     atexitrun(["goal", "network", "stop", "-r", tempnet])
 
     psqlstring = ensure_test_db(args.connection_string, args.keep_temps)
-    algoddir = os.path.join(tempnet, "Primary")
+    algoddir = os.path.join(tempnet, "Node")
     aiport = args.indexer_port or random.randint(4000, 30000)
     indexerdir = os.path.join(tempdir, "indexer_data_dir")
     cmd = [
@@ -172,15 +184,15 @@ def main():
     indexerurl = "http://localhost:{}/".format(aiport)
     healthurl = indexerurl + "health"
     for attempt in range(20):
-        (ok, json) = tryhealthurl(healthurl, args.verbose, waitforround=lastblock)
+        (ok, healthJson) = tryhealthurl(healthurl, args.verbose, waitforround=lastblock)
         if ok:
             logger.debug("health round={} OK".format(lastblock))
             break
         time.sleep(0.5)
     if not ok:
         logger.error(
             "could not get indexer health, or did not reach round={}\n{}".format(
-                lastblock, json
+                lastblock, healthJson
             )
         )
         sys.stderr.write(indexerout.dump())

go.mod

@@ -7,7 +7,7 @@ replace github.com/algorand/go-algorand => ./third_party/go-algorand
 require (
 	github.com/algorand/avm-abi v0.2.0
 	github.com/algorand/go-algorand v0.0.0-20220211161928-53b157beb10f
-	github.com/algorand/go-algorand-sdk/v2 v2.0.0-20230216185416-22ef3b66c617
+	github.com/algorand/go-algorand-sdk/v2 v2.0.0-20230228201805-5b8c99b1412c
 	github.com/algorand/go-codec/codec v1.1.8
 	github.com/algorand/go-deadlock v0.2.2
 	github.com/algorand/oapi-codegen v1.12.0-algorand.0

go.sum

@@ -92,6 +92,10 @@ github.com/algorand/go-algorand-sdk/v2 v2.0.0-20230209164439-140b997ed9ee h1:OGE
 github.com/algorand/go-algorand-sdk/v2 v2.0.0-20230209164439-140b997ed9ee/go.mod h1:Nt3EHpP8AznLs0/EFfhr0/xsVf5ucnvjNeRygGgbUzM=
 github.com/algorand/go-algorand-sdk/v2 v2.0.0-20230216185416-22ef3b66c617 h1:IOqgNBpxSk23ixJzysAYqRYz04rAE2+d1DJI1sD+z8M=
 github.com/algorand/go-algorand-sdk/v2 v2.0.0-20230216185416-22ef3b66c617/go.mod h1:Nt3EHpP8AznLs0/EFfhr0/xsVf5ucnvjNeRygGgbUzM=
+github.com/algorand/go-algorand-sdk/v2 v2.0.0-20230228201805-5b8c99b1412c h1:KAX6gb3+DLCTBcVhjDtuhcdbCeKnwIYKdj5Dv2JA/nI=
+github.com/algorand/go-algorand-sdk/v2 v2.0.0-20230228201805-5b8c99b1412c/go.mod h1:Nt3EHpP8AznLs0/EFfhr0/xsVf5ucnvjNeRygGgbUzM=
+github.com/algorand/go-algorand-sdk/v2 v2.0.0 h1:hfi+YrzcaH20mrjdIQYOPFKPlJ/ULq9KLhLG9I6LJEY=
+github.com/algorand/go-algorand-sdk/v2 v2.0.0/go.mod h1:Nt3EHpP8AznLs0/EFfhr0/xsVf5ucnvjNeRygGgbUzM=
 github.com/algorand/go-codec v1.1.8 h1:XDSreeeZY8gMst6Edz4RBkl08/DGMJOeHYkoXL2B7wI=
 github.com/algorand/go-codec v1.1.8/go.mod h1:XhzVs6VVyWMLu6cApb9/192gBjGRVGm5cX5j203Heg4=
 github.com/algorand/go-codec/codec v1.1.8 h1:lsFuhcOH2LiEhpBH3BVUUkdevVmwCRyvb7FCAAPeY6U=

util/test/mock_algod.go

@@ -9,9 +9,9 @@ import (
 
 	"github.com/algorand/go-algorand-sdk/v2/client/v2/algod"
 	"github.com/algorand/go-algorand-sdk/v2/client/v2/common/models"
+	"github.com/algorand/go-algorand-sdk/v2/encoding/json"
 	"github.com/algorand/go-algorand-sdk/v2/encoding/msgpack"
 	"github.com/algorand/go-algorand-sdk/v2/types"
-	"github.com/algorand/go-algorand/protocol"
 )
 
 // AlgodHandler is used to handle http requests to a mock algod server
@@ -67,7 +67,7 @@ func MakeGenesisResponder(genesis types.Genesis) func(reqPath string, w http.Res
 	return func(reqPath string, w http.ResponseWriter) bool {
 		if strings.Contains(reqPath, "/genesis") {
 			w.WriteHeader(http.StatusOK)
-			blockbytes := protocol.EncodeJSON(&genesis)
+			blockbytes := json.Encode(&genesis)
 			_, _ = w.Write(blockbytes)
 			return true
 		}
@@ -86,7 +86,7 @@ func BlockAfterResponder(reqPath string, w http.ResponseWriter) bool {
 	if strings.Contains(reqPath, "/wait-for-block-after") {
 		w.WriteHeader(http.StatusOK)
 		nStatus := models.NodeStatus{}
-		_, _ = w.Write(protocol.EncodeJSON(nStatus))
+		_, _ = w.Write(json.Encode(nStatus))
 		return true
 	}
 	return false
@@ -97,7 +97,7 @@ func LedgerStateDeltaResponder(reqPath string, w http.ResponseWriter) bool {
 	if strings.Contains(reqPath, "v2/deltas/") {
 		w.WriteHeader(http.StatusOK)
 		delta := types.LedgerStateDelta{}
-		_, _ = w.Write(protocol.EncodeJSON(delta))
+		_, _ = w.Write(msgpack.Encode(delta))
 		return true
 	}
 	return false