Repository: algorand/indexer
Pull request #1503 (2023-03-10 20:15:57)

e2e-tests: Add filtered indexer e2e tests
## Summary

Adds an e2e test which uses a conduit writer pipeline w/ a filter processor.

Uses docker-compose w/ a conduit writer, an e2e reader w/ queries, and a pgsql db.

Followup steps: working on separating these tests into conduit and indexer.

## Test Plan

`make e2e-filter-test-nightly`



Patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index 08a5034..be564aa 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -139,6 +139,7 @@ commands:
             ./update.sh -i -c stable -n -d ./ -p /usr/local/go/bin
             export GOPATH=/usr/local/go/
       - run: make e2e
+      - run: make e2e-filter-test
       - run: make e2e-conduit
 
   run_e2e_tests_nightly:
@@ -150,6 +151,7 @@ commands:
             ./update.sh -i -c nightly -n -d ./ -p /usr/local/go/bin
             export GOPATH=/usr/local/go/
       - run: make e2e-nightly
+      - run: make e2e-filter-test-nightly
       - run: make e2e-conduit
 
   run_tests:
diff --git a/Makefile b/Makefile
index 3b51fc1..f8ec294 100644
--- a/Makefile
+++ b/Makefile
@@ -73,6 +73,12 @@ fmt:
 e2e: cmd/algorand-indexer/algorand-indexer
 	cd e2e_tests/docker/indexer/ && docker-compose build --build-arg GO_IMAGE=${GO_IMAGE} && docker-compose up --exit-code-from e2e
 
+e2e-filter-test: cmd/algorand-indexer/algorand-indexer conduit
+	cd e2e_tests/docker/indexer-filtered/ && docker-compose build --build-arg GO_IMAGE=${GO_IMAGE} && docker-compose up --exit-code-from e2e-read
+
+e2e-filter-test-nightly: cmd/algorand-indexer/algorand-indexer conduit
+	cd e2e_tests/docker/indexer-filtered/ && docker-compose build --build-arg GO_IMAGE=${GO_IMAGE} --build-arg CHANNEL=nightly && docker-compose up --exit-code-from e2e-read
+
 e2e-nightly: cmd/algorand-indexer/algorand-indexer
 	cd e2e_tests/docker/indexer/ && docker-compose build --build-arg GO_IMAGE=${GO_IMAGE} --build-arg CHANNEL=nightly && docker-compose up --exit-code-from e2e
 
diff --git a/idb/postgres/postgres.go b/idb/postgres/postgres.go
index 6c6ab88..d09f05a 100644
--- a/idb/postgres/postgres.go
+++ b/idb/postgres/postgres.go
@@ -175,6 +175,12 @@ func (db *IndexerDb) init(opts idb.IndexerDbOptions) (chan struct{}, error) {
 
 // AddBlock is part of idb.IndexerDb.
 func (db *IndexerDb) AddBlock(vb *itypes.ValidatedBlock) error {
+	protoVersion := protocol.ConsensusVersion(vb.Block.CurrentProtocol)
+	_, ok := config.Consensus[protoVersion]
+	if !ok {
+		return fmt.Errorf("unknown protocol (%s) detected, this usually means you need to upgrade", protoVersion)
+	}
+
 	block := vb.Block
 	round := block.BlockHeader.Round
 	db.log.Printf("adding block %d", round)


Test patch:
diff --git a/e2e_tests/docker/indexer-filtered/docker-compose.yml b/e2e_tests/docker/indexer-filtered/docker-compose.yml
new file mode 100644
index 0000000..7d74945
--- /dev/null
+++ b/e2e_tests/docker/indexer-filtered/docker-compose.yml
@@ -0,0 +1,39 @@
+version: '3'
+
+services:
+  e2e-write:
+    build:
+      context: ../../../.
+      dockerfile: ./e2e_tests/docker/indexer-filtered/e2e-write/Dockerfile
+      args:
+        - CI_E2E_FILENAME=${CI_E2E_FILENAME}
+    depends_on:
+      e2e-db-filtered:
+        condition: service_healthy
+    environment:
+      CONNECTION_STRING: "host=e2e-db-filtered port=5432 user=algorand password=algorand dbname=indexer_db sslmode=disable"
+      # Add --keep-alive to this variable to pause the container on an error
+      # EXTRA: "${EXTRA}"
+
+  e2e-read:
+    build:
+      context: ../../../.
+      dockerfile: ./e2e_tests/docker/indexer-filtered/e2e-read/Dockerfile
+    depends_on:
+      - e2e-write
+    environment:
+      CONNECTION_STRING: "host=e2e-db-filtered port=5432 user=algorand password=algorand dbname=indexer_db sslmode=disable"
+      # Add --keep-alive to this variable to pause the container on an error
+      EXTRA: "${EXTRA}"
+
+  e2e-db-filtered:
+    image: "postgres:13-alpine"
+    ports:
+      - 45432:5432
+    environment:
+      POSTGRES_USER: algorand
+      POSTGRES_PASSWORD: algorand
+      POSTGRES_DB: indexer_db
+    healthcheck:
+      test: [ "CMD-SHELL", "pg_isready -q " ]
+
diff --git a/e2e_tests/docker/indexer-filtered/e2e-read/Dockerfile b/e2e_tests/docker/indexer-filtered/e2e-read/Dockerfile
new file mode 100644
index 0000000..4bbdff8
--- /dev/null
+++ b/e2e_tests/docker/indexer-filtered/e2e-read/Dockerfile
@@ -0,0 +1,24 @@
+ARG GO_IMAGE=golang:1.14.7
+FROM $GO_IMAGE
+
+RUN echo "Go image: $GO_IMAGE"
+
+# Misc dependencies
+ENV HOME /opt
+ENV DEBIAN_FRONTEND noninteractive
+RUN apt-get update && apt-get install -y apt-utils curl git git-core bsdmainutils python3 python3-pip make bash libtool libffi-dev wget
+
+# Setup files for test
+RUN mkdir -p /opt/go/indexer
+COPY . /opt/go/indexer
+WORKDIR /opt/go/indexer
+RUN rm -f /opt/go/indexer/cmd/algorand-indexer/algorand-indexer
+RUN make
+ENV PATH="${HOME}/go/bin/:${PATH}"
+WORKDIR ./e2e_tests
+RUN pip3 install ./
+
+ENV INDEXER_DATA="${HOME}/indexer/"
+WORKDIR /opt/go/indexer
+# Run test script
+ENTRYPOINT ["/bin/bash", "-c", "sleep 5 && e2elive $EXTRA --connection-string \"$CONNECTION_STRING\" --indexer-bin /opt/go/indexer/cmd/algorand-indexer/algorand-indexer --indexer-port 9890 --read-only True --verbose"]
diff --git a/e2e_tests/docker/indexer-filtered/e2e-write/Dockerfile b/e2e_tests/docker/indexer-filtered/e2e-write/Dockerfile
new file mode 100644
index 0000000..1ef46c0
--- /dev/null
+++ b/e2e_tests/docker/indexer-filtered/e2e-write/Dockerfile
@@ -0,0 +1,42 @@
+ARG GO_IMAGE=golang:1.14.7
+FROM $GO_IMAGE
+ARG CHANNEL=stable
+ARG CI_E2E_FILENAME
+
+RUN echo "Go image: $GO_IMAGE"
+
+# Misc dependencies
+ENV HOME /opt
+ENV DEBIAN_FRONTEND noninteractive
+RUN apt-get update && apt-get install -y apt-utils curl git git-core bsdmainutils python3 python3-pip make bash libtool libboost-math-dev libffi-dev wget gettext-base
+
+# Setup files for test
+RUN mkdir -p /opt/go/indexer
+RUN mkdir -p /opt/conduit-dir
+RUN mkdir -p /opt/algod-dir
+COPY . /opt/go/indexer
+WORKDIR /opt/go/indexer
+RUN rm -f $HOME/go/bin/algod
+RUN rm -f /opt/go/indexer/cmd/conduit/conduit
+WORKDIR /opt/go/node
+RUN wget https://raw.githubusercontent.com/algorand/go-algorand/rel/stable/cmd/updater/update.sh && chmod 744 update.sh
+RUN ./update.sh -i -c $CHANNEL -n -d ./ -p /go/bin/
+WORKDIR /opt/go/indexer
+RUN make
+ENV PATH="${HOME}/go/bin/:${PATH}"
+WORKDIR ./e2e_tests
+RUN pip3 install ./
+
+ENV INDEXER_DATA="${HOME}/indexer/"
+WORKDIR /opt/go/indexer
+
+# Retrieve the network data
+RUN get-test-data --s3-source-net ${CI_E2E_FILENAME} --algod-dir /opt/algod-dir
+
+# Run test script
+CMD ["/bin/bash", "-c", \
+"goal network start -r /opt/algod-dir/net/ && \
+export ALGOD_ADDR=http://$(cat /opt/algod-dir/net/Node/algod.net) && \
+export ALGOD_TOKEN=$(cat /opt/algod-dir/net/Node/algod.token) && \
+cat ./e2e_tests/docker/indexer-filtered/e2e-write/conduit.yml | envsubst > /opt/conduit-dir/conduit.yml && \
+/opt/go/indexer/cmd/conduit/conduit -d /opt/conduit-dir"]
diff --git a/e2e_tests/docker/indexer-filtered/e2e-write/conduit.yml b/e2e_tests/docker/indexer-filtered/e2e-write/conduit.yml
new file mode 100644
index 0000000..b0deadb
--- /dev/null
+++ b/e2e_tests/docker/indexer-filtered/e2e-write/conduit.yml
@@ -0,0 +1,19 @@
+log-level: info
+importer:
+  name: "algod"
+  config:
+    netaddr: "$ALGOD_ADDR"
+    token: "$ALGOD_TOKEN"
+    mode: "follower"
+processors:
+  - name: "filter_processor"
+    config:
+      filters:
+        - any:
+          - tag: "txn.type"
+            expression-type: "equal"
+            expression: "pay"
+exporter:
+  name: "postgresql"
+  config:
+    connection-string: $CONNECTION_STRING
\ No newline at end of file
diff --git a/e2e_tests/pyproject.toml b/e2e_tests/pyproject.toml
index 37f9ea1..a11cac1 100644
--- a/e2e_tests/pyproject.toml
+++ b/e2e_tests/pyproject.toml
@@ -21,3 +21,4 @@ dependencies = [
 e2elive = "e2e_indexer.e2elive:main"
 e2econduit = "e2e_conduit.e2econduit:main"
 validate-accounting = "e2e_indexer.validate_accounting:main"
+get-test-data = "e2e_common.get_test_data:main"
diff --git a/e2e_tests/src/e2e_common/get_test_data.py b/e2e_tests/src/e2e_common/get_test_data.py
new file mode 100644
index 0000000..8aec942
--- /dev/null
+++ b/e2e_tests/src/e2e_common/get_test_data.py
@@ -0,0 +1,74 @@
+import argparse
+import json
+import logging
+import os
+import sys
+
+import boto3
+from botocore.config import Config
+from botocore import UNSIGNED
+
+from e2e_common.util import (
+    xrun,
+    atexitrun,
+    firstFromS3Prefix,
+    hassuffix,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument(
+        "--s3-source-net",
+        help="AWS S3 key suffix to test network tarball containing Primary and other nodes. Must be a tar bz2 file.",
+    )
+    ap.add_argument(
+        "--algod-dir",
+        help="Directory to run algod network in.",
+    )
+    ap.add_argument("--verbose", default=False, action="store_true")
+    args = ap.parse_args()
+    if not args.s3_source_net:
+        raise Exception("Must provide --s3-source-net")
+    if not args.algod_dir:
+        raise Exception("Must provide --algod-dir")
+
+    tarname = f"{args.s3_source_net}.tar.bz2"
+     # fetch test data from S3
+    bucket = "algorand-testdata"
+    s3 = boto3.client("s3", config=Config(signature_version=UNSIGNED))
+    prefix = "indexer/e2e4"
+    if "/" in tarname:
+        cmhash_tarnme = tarname.split("/")
+        cmhash = cmhash_tarnme[0]
+        tarname =cmhash_tarnme[1]
+        prefix+="/"+cmhash
+        tarpath = os.path.join(args.algod_dir, tarname)
+    else:
+        tarpath = os.path.join(args.algod_dir, tarname)
+    success = firstFromS3Prefix(s3, bucket, prefix, tarname, outpath=tarpath)
+    if not success:
+        raise Exception(f"failed to locate tarname={tarname} from AWS S3 path {bucket}/{prefix}")
+    sourcenet = tarpath
+    tempnet = os.path.join(args.algod_dir, "net")
+    xrun(["tar", "-C", args.algod_dir, "-x", "-f", sourcenet])
+
+    # 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))
+
+    return 0
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/e2e_tests/src/e2e_indexer/e2elive.py b/e2e_tests/src/e2e_indexer/e2elive.py
index 5fe0a1b..4d0a843 100644
--- a/e2e_tests/src/e2e_indexer/e2elive.py
+++ b/e2e_tests/src/e2e_indexer/e2elive.py
@@ -59,6 +59,12 @@ def main():
         "--s3-source-net",
         help="AWS S3 key suffix to test network tarball containing Primary and other nodes. Must be a tar bz2 file.",
     )
+    ap.add_argument(
+        "--read-only",
+        default=False,
+        type=bool,
+        help="Run the indexer daemon in read-only mode.",
+    )
     ap.add_argument("--verbose", default=False, action="store_true")
     args = ap.parse_args()
     if args.verbose:
@@ -66,23 +72,96 @@ def main():
     else:
         logging.basicConfig(level=logging.INFO)
     indexer_bin = find_binary(args.indexer_bin)
-    sourcenet = args.source_net
-    source_is_tar = False
-    if not sourcenet:
-        e2edata = os.getenv("E2EDATA")
-        sourcenet = e2edata and os.path.join(e2edata, "net")
-    if sourcenet and hassuffix(sourcenet, ".tar", ".tar.gz", ".tar.bz2", ".tar.xz"):
-        source_is_tar = True
     tempdir = tempfile.mkdtemp()
     if not args.keep_temps:
         atexit.register(shutil.rmtree, tempdir, onerror=logger.error)
     else:
         logger.info("leaving temp dir %r", tempdir)
+    # Sane default for lastblock--used for read_only runs
+    lastblock = 93
+    algoddir = None
     if args.keep_alive:
         # register after keep_temps so that the temp files are still there.
         atexitrun(["sleep", "infinity"])
+    psqlstring = ensure_test_db(args.connection_string, args.keep_temps)
+    aiport = args.indexer_port or random.randint(4000, 30000)
+    indexerdir = os.path.join(tempdir, "indexer_data_dir")
+    cmd = [
+        indexer_bin,
+        "daemon",
+        "--data-dir",
+        indexerdir,
+        "-P",
+        psqlstring,
+        "--dev-mode",
+        "--server",
+        ":{}".format(aiport),
+    ]
+    if args.read_only:
+        cmd.append("--no-algod")
+    else:
+        tempnet, lastblock = setup_algod(tempdir, args.source_net, args.s3_source_net)
+        algoddir = os.path.join(tempnet, "Node")
+        cmd.append("--algod")
+        cmd.append(algoddir)
+    logger.debug("%s", " ".join(map(repr, cmd)))
+    indexerdp = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+    indexerout = subslurp(indexerdp.stdout)
+    indexerout.start()
+    atexit.register(indexerdp.kill)
+    time.sleep(0.2)
+
+    indexerurl = "http://localhost:{}/".format(aiport)
+    healthurl = indexerurl + "health"
+    for attempt in range(20):
+        (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, healthJson
+            )
+        )
+        sys.stderr.write(indexerout.dump())
+        return 1
+    try:
+        logger.info("reached expected round={}".format(lastblock))
+        if not args.read_only:
+            xrun(
+                [
+                    "validate-accounting",
+                    "--verbose",
+                    "--algod",
+                    algoddir,
+                    "--indexer",
+                    indexerurl,
+                ],
+                timeout=20,
+            )
+        xrun(
+            ["go", "run", "cmd/e2equeries/main.go", "-pg", psqlstring, "-q"], timeout=15
+        )
+    except Exception:
+        sys.stderr.write(indexerout.dump())
+        raise
+    dt = time.time() - start
+    sys.stdout.write("indexer e2etest OK ({:.1f}s)\n".format(dt))
+
+    return 0
+
+
+def setup_algod(tempdir, sourcenet, s3_source_net):
+    source_is_tar = False
+    if not sourcenet:
+        e2edata = os.getenv("E2EDATA")
+        sourcenet = e2edata and os.path.join(e2edata, "net")
+    if sourcenet and hassuffix(sourcenet, ".tar", ".tar.gz", ".tar.bz2", ".tar.xz"):
+        source_is_tar = True
     if not (source_is_tar or (sourcenet and os.path.isdir(sourcenet))):
-        tarname = args.s3_source_net
+        tarname = s3_source_net
         if not tarname:
             raise Exception(
                 "Must provide either local or s3 network to run test against"
@@ -154,70 +233,7 @@ def main():
         raise
 
     atexitrun(["goal", "network", "stop", "-r", tempnet])
-
-    psqlstring = ensure_test_db(args.connection_string, args.keep_temps)
-    algoddir = os.path.join(tempnet, "Node")
-    aiport = args.indexer_port or random.randint(4000, 30000)
-    indexerdir = os.path.join(tempdir, "indexer_data_dir")
-    cmd = [
-        indexer_bin,
-        "daemon",
-        "--data-dir",
-        indexerdir,
-        "-P",
-        psqlstring,
-        "--dev-mode",
-        "--algod",
-        algoddir,
-        "--server",
-        ":{}".format(aiport),
-    ]
-    logger.debug("%s", " ".join(map(repr, cmd)))
-    indexerdp = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
-    indexerout = subslurp(indexerdp.stdout)
-    indexerout.start()
-    atexit.register(indexerdp.kill)
-    time.sleep(0.2)
-
-    indexerurl = "http://localhost:{}/".format(aiport)
-    healthurl = indexerurl + "health"
-    for attempt in range(20):
-        (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, healthJson
-            )
-        )
-        sys.stderr.write(indexerout.dump())
-        return 1
-    try:
-        logger.info("reached expected round={}".format(lastblock))
-        xrun(
-            [
-                "validate-accounting",
-                "--verbose",
-                "--algod",
-                algoddir,
-                "--indexer",
-                indexerurl,
-            ],
-            timeout=20,
-        )
-        xrun(
-            ["go", "run", "cmd/e2equeries/main.go", "-pg", psqlstring, "-q"], timeout=15
-        )
-    except Exception:
-        sys.stderr.write(indexerout.dump())
-        raise
-    dt = time.time() - start
-    sys.stdout.write("indexer e2etest OK ({:.1f}s)\n".format(dt))
-
-    return 0
+    return tempnet, lastblock
 
 
 def tryhealthurl(healthurl, verbose=False, waitforround=200):
diff --git a/idb/postgres/postgres_test.go b/idb/postgres/postgres_test.go
index f1f2176..4b91fec 100644
--- a/idb/postgres/postgres_test.go
+++ b/idb/postgres/postgres_test.go
@@ -6,8 +6,12 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	sdk "github.com/algorand/go-algorand-sdk/v2/types"
 
 	"github.com/algorand/indexer/idb"
+	"github.com/algorand/indexer/types"
 )
 
 func Test_txnFilterOptimization(t *testing.T) {
@@ -54,3 +58,19 @@ func Test_txnFilterOptimization(t *testing.T) {
 		})
 	}
 }
+
+func Test_UnknownProtocol(t *testing.T) {
+	db := IndexerDb{}
+	protocol := "zzzzzzz"
+	err := db.AddBlock(&types.ValidatedBlock{
+		Block: sdk.Block{
+			BlockHeader: sdk.BlockHeader{
+				UpgradeState: sdk.UpgradeState{
+					CurrentProtocol: protocol,
+				},
+			},
+		},
+	})
+	require.ErrorContains(t, err, protocol)
+	require.ErrorContains(t, err, "you need to upgrade")
+}