conduit: Initial conduit pipeline tool.

.circleci/config.yml

@@ -114,6 +114,7 @@ commands:
             sudo apt update
             sudo apt -y install python3 python3-pip python3-setuptools python3-wheel libboost-math-dev libffi-dev
             pip3 install -r misc/requirements.txt
+            pip3 install e2e_tests/
 
       - run:
           name: sync submodules (go-algorand)
@@ -143,6 +144,7 @@ commands:
       - run: make test-generate
       - run: make fakepackage
       - run: make e2e
+      - run: make e2e-conduit
 
   run_indexer_vs_algod:
     steps:

.codecov.yml

@@ -2,6 +2,11 @@ codecov:
   require_ci_to_pass: no
   branch: develop
 
+ignore:
+  - "idb/mocks"
+  - "idb/dummy"
+  - "util/test"
+
 coverage:
   precision: 2
   round: down

.gitignore

@@ -1,5 +1,6 @@
 # Build artifacts
 cmd/algorand-indexer/algorand-indexer
+cmd/conduit/conduit
 api/.3tmp.json
 api/generate
 tmp/
@@ -28,6 +29,9 @@ _*.json
 
 # Python
 __pycache__
+e2e_tests/dist
+e2e_tests/build
+e2e_tests/*egg-info*
 .venv
 
 # jetbrains IDE
@@ -48,3 +52,6 @@ coverage.txt
 
 # asdf
 .tool-versions
+
+# conduit example
+cmd/conduit/data

Makefile

@@ -22,7 +22,12 @@ COVERPKG := $(shell go list ./...  | grep -v '/cmd/' | egrep -v '(testing|test|m
 # Used for e2e test
 export GO_IMAGE = golang:$(shell go version | cut -d ' ' -f 3 | tail -c +3 )
 
-# This is the default target, build the indexer:
+# This is the default target, build everything:
+all: conduit cmd/algorand-indexer/algorand-indexer go-algorand idb/postgres/internal/schema/setup_postgres_sql.go idb/mocks/IndexerDb.go
+
+conduit: go-algorand
+	go generate ./... && cd cmd/conduit && go build
+
 cmd/algorand-indexer/algorand-indexer: idb/postgres/internal/schema/setup_postgres_sql.go go-algorand
 	cd cmd/algorand-indexer && go build -ldflags="${GOLDFLAGS}"
 
@@ -55,7 +60,7 @@ fakepackage: go-algorand
 test: idb/mocks/IndexerDb.go cmd/algorand-indexer/algorand-indexer
 	go test -coverpkg=$(COVERPKG) ./... -coverprofile=coverage.txt -covermode=atomic ${TEST_FLAG}
 
-lint: go-algorand
+lint: conduit
 	golint -set_exit_status ./...
 	go vet ./...
 
@@ -68,7 +73,11 @@ integration: cmd/algorand-indexer/algorand-indexer
 	test/postgres_integration_test.sh
 
 e2e: cmd/algorand-indexer/algorand-indexer
-	cd misc && docker-compose build --build-arg GO_IMAGE=${GO_IMAGE} && docker-compose up --exit-code-from e2e
+	cd e2e_tests/docker/indexer/ && docker-compose build --build-arg GO_IMAGE=${GO_IMAGE} && docker-compose up --exit-code-from e2e
+
+e2e-conduit: conduit
+	cd third_party/go-algorand && make install
+	export PATH=$(PATH):$(shell go env GOPATH)/bin; pip3 install e2e_tests/ && e2econduit --s3-source-net ${CI_E2E_FILENAME} --conduit-bin cmd/conduit/conduit
 
 deploy:
 	mule/deploy.sh
@@ -96,9 +105,10 @@ indexer-v-algod: nightly-setup indexer-v-algod-swagger nightly-teardown
 # fetch and update submodule. it's default to latest rel/nightly branch.
 # to use a different branch, update the branch in .gitmodules for CI build,
 # and for local testing, you may checkout a specific branch in the submodule.
-# after submodule is updated, CI_E2E_FILE in circleci/config.yml should also
-# be updated to use a newer artifact. path copied from s3 bucket, s3://algorand-testdata/indexer/e2e4/
+# after submodule is updated, CI_E2E_FILENAME in .circleci/config.yml should
+# also be updated to use a newer artifact. path copied from s3 bucket,
+# s3://algorand-testdata/indexer/e2e4/
 update-submodule:
 	git submodule update --remote
 
-.PHONY: test e2e integration fmt lint deploy sign test-package package fakepackage cmd/algorand-indexer/algorand-indexer idb/mocks/IndexerDb.go go-algorand indexer-v-algod
+.PHONY: all test e2e integration fmt lint deploy sign test-package package fakepackage cmd/algorand-indexer/algorand-indexer idb/mocks/IndexerDb.go go-algorand indexer-v-algod conduit

README.md

@@ -48,11 +48,11 @@ Indexer is part of the [sandbox](https://github.com/algorand/sandbox) private ne
 - Search and filter accounts, transactions, assets, and asset balances with many different parameters.
 - Pagination of results.
 - Enriched transaction and account data:
-    - Confirmation round (block containing the transaction)
-    - Confirmation time
-    - Signature type
-    - Close amounts
-    - Create/delete rounds.
+  - Confirmation round (block containing the transaction)
+  - Confirmation time
+  - Signature type
+  - Close amounts
+  - Create/delete rounds.
 - Human readable field names instead of the space optimized protocol level names.
 
 # Contributing

api/app_boxes_fixtures_test.go

@@ -11,7 +11,6 @@ import (
 	"github.com/algorand/go-algorand/data/transactions/logic"
 	"github.com/algorand/go-algorand/ledger"
 	"github.com/algorand/go-algorand/rpcs"
-	"github.com/algorand/indexer/processor"
 	"github.com/algorand/indexer/util/test"
 	"github.com/stretchr/testify/require"
 )
@@ -40,7 +39,7 @@ var goalEncodingExamples map[string]string = map[string]string{
 	"abi":         `(uint64,string,bool[]):[399,"pls pass",[true,false]]`,
 }
 
-func setupLiveBoxes(t *testing.T, proc processor.Processor, l *ledger.Ledger) {
+func setupLiveBoxes(t *testing.T, proc func(cert *rpcs.EncodedBlockCert) error, l *ledger.Ledger) {
 	deleted := "DELETED"
 
 	firstAppid := basics.AppIndex(1)
@@ -69,7 +68,7 @@ func setupLiveBoxes(t *testing.T, proc processor.Processor, l *ledger.Ledger) {
 	block, err := test.MakeBlockForTxns(test.MakeGenesisBlock().BlockHeader, &createTxn, &payNewAppTxn, &createTxn2, &payNewAppTxn2, &createTxn3, &payNewAppTxn3)
 	require.NoError(t, err)
 
-	err = proc.Process(&rpcs.EncodedBlockCert{Block: block})
+	err = proc(&rpcs.EncodedBlockCert{Block: block})
 	require.NoError(t, err)
 
 	// block header handoff: round 1 --> round 2
@@ -104,7 +103,7 @@ func setupLiveBoxes(t *testing.T, proc processor.Processor, l *ledger.Ledger) {
 	block, err = test.MakeBlockForTxns(blockHdr, boxTxns...)
 	require.NoError(t, err)
 
-	err = proc.Process(&rpcs.EncodedBlockCert{Block: block})
+	err = proc(&rpcs.EncodedBlockCert{Block: block})
 	require.NoError(t, err)
 
 	// block header handoff: round 2 --> round 3
@@ -137,7 +136,7 @@ func setupLiveBoxes(t *testing.T, proc processor.Processor, l *ledger.Ledger) {
 	block, err = test.MakeBlockForTxns(blockHdr, boxTxns...)
 	require.NoError(t, err)
 
-	err = proc.Process(&rpcs.EncodedBlockCert{Block: block})
+	err = proc(&rpcs.EncodedBlockCert{Block: block})
 	require.NoError(t, err)
 
 	// block header handoff: round 3 --> round 4
@@ -165,7 +164,7 @@ func setupLiveBoxes(t *testing.T, proc processor.Processor, l *ledger.Ledger) {
 	block, err = test.MakeBlockForTxns(blockHdr, boxTxns...)
 	require.NoError(t, err)
 
-	err = proc.Process(&rpcs.EncodedBlockCert{Block: block})
+	err = proc(&rpcs.EncodedBlockCert{Block: block})
 	require.NoError(t, err)
 
 	// block header handoff: round 4 --> round 5
@@ -194,7 +193,7 @@ func setupLiveBoxes(t *testing.T, proc processor.Processor, l *ledger.Ledger) {
 	block, err = test.MakeBlockForTxns(blockHdr, boxTxns...)
 	require.NoError(t, err)
 
-	err = proc.Process(&rpcs.EncodedBlockCert{Block: block})
+	err = proc(&rpcs.EncodedBlockCert{Block: block})
 	require.NoError(t, err)
 
 	// block header handoff: round 5 --> round 6
@@ -223,7 +222,7 @@ func setupLiveBoxes(t *testing.T, proc processor.Processor, l *ledger.Ledger) {
 	block, err = test.MakeBlockForTxns(blockHdr, boxTxns...)
 	require.NoError(t, err)
 
-	err = proc.Process(&rpcs.EncodedBlockCert{Block: block})
+	err = proc(&rpcs.EncodedBlockCert{Block: block})
 	require.NoError(t, err)
 
 	// block header handoff: round 6 --> round 7
@@ -250,7 +249,7 @@ func setupLiveBoxes(t *testing.T, proc processor.Processor, l *ledger.Ledger) {
 	block, err = test.MakeBlockForTxns(blockHdr, boxTxns...)
 	require.NoError(t, err)
 
-	err = proc.Process(&rpcs.EncodedBlockCert{Block: block})
+	err = proc(&rpcs.EncodedBlockCert{Block: block})
 	require.NoError(t, err)
 
 	// block header handoff: round 7 --> round 8
@@ -273,7 +272,7 @@ func setupLiveBoxes(t *testing.T, proc processor.Processor, l *ledger.Ledger) {
 	block, err = test.MakeBlockForTxns(blockHdr, boxTxns...)
 	require.NoError(t, err)
 
-	err = proc.Process(&rpcs.EncodedBlockCert{Block: block})
+	err = proc(&rpcs.EncodedBlockCert{Block: block})
 	require.NoError(t, err)
 
 	// block header handoff: round 8 --> round 9
@@ -287,7 +286,7 @@ func setupLiveBoxes(t *testing.T, proc processor.Processor, l *ledger.Ledger) {
 	block, err = test.MakeBlockForTxns(blockHdr, &deleteTxn)
 	require.NoError(t, err)
 
-	err = proc.Process(&rpcs.EncodedBlockCert{Block: block})
+	err = proc(&rpcs.EncodedBlockCert{Block: block})
 	require.NoError(t, err)
 
 	// ---- SUMMARY ---- //

api/fixtures_test.go

@@ -15,10 +15,10 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/algorand/go-algorand/ledger"
+	"github.com/algorand/go-algorand/rpcs"
 
 	"github.com/algorand/indexer/api/generated/v2"
 	"github.com/algorand/indexer/idb/postgres"
-	"github.com/algorand/indexer/processor"
 	"github.com/algorand/indexer/util/test"
 )
 
@@ -268,7 +268,7 @@ func getProof(path string) (route string, proof prover, err error) {
 }
 
 // WARNING: receiver should not call l.Close()
-func setupIdbAndReturnShutdownFunc(t *testing.T) (db *postgres.IndexerDb, proc processor.Processor, l *ledger.Ledger, shutdown func()) {
+func setupIdbAndReturnShutdownFunc(t *testing.T) (db *postgres.IndexerDb, proc func(cert *rpcs.EncodedBlockCert) error, l *ledger.Ledger, shutdown func()) {
 	db, dbShutdown, proc, l := setupIdb(t, test.MakeGenesis())
 
 	shutdown = func() {

api/handlers.go

@@ -39,7 +39,7 @@ type ServerImplementation struct {
 
 	db idb.IndexerDb
 
-	fetcher error
+	dataError func() error
 
 	timeout time.Duration
 
@@ -142,8 +142,10 @@ func (si *ServerImplementation) MakeHealthCheck(ctx echo.Context) error {
 		errors = append(errors, fmt.Sprintf("database error: %s", health.Error))
 	}
 
-	if si.fetcher != nil && si.fetcher.Error() != "" {
-		errors = append(errors, fmt.Sprintf("fetcher error: %s", si.fetcher.Error()))
+	if si.dataError != nil {
+		if err := si.dataError(); err != nil {
+			errors = append(errors, fmt.Sprintf("data error: %s", err))
+		}
 	}
 
 	return ctx.JSON(http.StatusOK, common.HealthCheckResponse{