exp/lighthorizon/build/index-batch: carry over map/reduce updates to latest docker layout on feature branch (#4543)

Author: sreuland

.github/workflows/horizon.yml

@@ -134,14 +134,27 @@ jobs:
         # Any range should do for basic testing, this range was chosen pretty early in history so that it only takes a few mins to run
         run: |
           chmod 755 ./exp/lighthorizon/build/build.sh
-          ./exp/lighthorizon/build/build.sh ledgerexporter stellar latest false
-          docker run -e ARCHIVE_TARGET=file:///ledgerexport-test\
+          mkdir $PWD/ledgerexport
+          mkdir $PWD/index
+
+          ./exp/lighthorizon/build/build.sh all stellar latest false
+          docker run -e ARCHIVE_TARGET=file:///ledgerexport\
                      -e START=5\
-                     -e END=50\
+                     -e END=150\
                      -e NETWORK_PASSPHRASE="Test SDF Network ; September 2015"\
                      -e CAPTIVE_CORE_CONFIG="/captive-core-testnet.cfg"\
                      -e HISTORY_ARCHIVE_URLS="https://history.stellar.org/prd/core-testnet/core_testnet_001,https://history.stellar.org/prd/core-testnet/core_testnet_002"\
+                     -v $PWD/ledgerexport:/ledgerexport\
                      stellar/lighthorizon-ledgerexporter
+          
+          # run map job
+          docker run -e NETWORK_PASSPHRASE='testnet' -e JOB_INDEX_ENV=AWS_BATCH_JOB_ARRAY_INDEX -e AWS_BATCH_JOB_ARRAY_INDEX=0 -e BATCH_SIZE=64 -e FIRST_CHECKPOINT=64 \
+                     -e WORKER_COUNT=1 -e RUN_MODE=map -v $PWD/ledgerexport:/ledgermeta -e TXMETA_SOURCE=file:///ledgermeta -v $PWD/index:/index -e INDEX_TARGET=file:///index stellar/lighthorizon-index-batch
+
+          # run reduce job
+          docker run -e NETWORK_PASSPHRASE='testnet' -e JOB_INDEX_ENV=AWS_BATCH_JOB_ARRAY_INDEX -e AWS_BATCH_JOB_ARRAY_INDEX=0 -e MAP_JOB_COUNT=1 -e REDUCE_JOB_COUNT=1 \
+                     -e WORKER_COUNT=1 -e RUN_MODE=reduce -v $PWD/index:/index -e INDEX_SOURCE_ROOT=file:///index -e INDEX_TARGET=file:///index stellar/lighthorizon-index-batch
+                     
       # Push images
       - if: github.ref == 'refs/heads/master' || github.ref == 'refs/heads/lighthorizon'
         name: Login to DockerHub

exp/lighthorizon/build/index-batch/README.md

@@ -0,0 +1,7 @@
+# `stellar/horizon-indexer`
+
+This docker image contains the ledger/checkpoint indexing executables. It allows running multiple instances of `map`/`reduce` on a single machine or running it in [AWS Batch](https://aws.amazon.com/batch/).
+
+## Env variables
+
+See the [package documentation](../../index/cmd/batch/doc.go) for more details

exp/lighthorizon/build/index-batch/start

@@ -5,7 +5,6 @@ set -e
 
 export TRACY_NO_INVARIANT_CHECK=1
 NETWORK_PASSPHRASE="${NETWORK_PASSPHRASE:=Public Global Stellar Network ; September 2015}"
-
 if [ "$RUN_MODE" == "reduce" ]; then
     echo "Running Reduce, REDUCE JOBS: $REDUCE_JOB_COUNT MAP JOBS: $MAP_JOB_COUNT TARGET INDEX: $INDEX_TARGET"
     /reduce 

exp/lighthorizon/build/k8s/lighthorizon_batch_map_job.yml

@@ -0,0 +1,43 @@
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: 'batch-map-job'
+spec:
+  completions: 52
+  parallelism: 10
+  completionMode: Indexed
+  template:
+    spec:
+      restartPolicy: Never
+      containers:
+      - name: 'worker'
+        image: 'stellar/lighthorizon-index-batch'
+        imagePullPolicy: Always
+        envFrom:
+        - secretRef:
+            name: <reference to secret name here if needed for source/target>
+        env:
+        - name: RUN_MODE
+          value: "map" 
+        - name: BATCH_SIZE
+          value: "10048"   
+        - name: FIRST_CHECKPOINT
+          value: "41426080"
+        - name: WORKER_COUNT
+          value: "8"   
+        - name: TXMETA_SOURCE
+          value: "<url of txmeta source>"
+        - name: JOB_INDEX_ENV
+          value: "JOB_COMPLETION_INDEX"   
+        - name: NETWORK_PASSPHRASE
+          value: "pubnet"
+        - name: INDEX_TARGET
+          value: "url of target index"
+        resources:
+          limits:
+            cpu:     4
+            memory:  5Gi
+          requests:
+            cpu:     500m
+            memory:  500Mi
+        

exp/lighthorizon/build/k8s/lighthorizon_batch_reduce_job copy.yml

@@ -0,0 +1,42 @@
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: 'batch-reduce-job'
+spec:
+  completions: 52
+  parallelism: 10
+  completionMode: Indexed
+  template:
+    spec:
+      restartPolicy: Never
+      containers:
+      - name: 'worker'
+        image: 'stellar/lighthorizon-index-batch'
+        imagePullPolicy: Always
+        envFrom:
+        - secretRef:
+            name: <reference to secret name here if needed for source/target>
+        env:
+        - name: RUN_MODE
+          value: "reduce" 
+        - name: MAP_JOB_COUNT
+          value: 52
+        - name: REDUCE_JOB_COUNT
+          value: 52   
+        - name: WORKER_COUNT
+          value: 8   
+        - name: INDEX_SOURCE_ROOT
+          value: "<url of index location>"
+        - name: JOB_INDEX_ENV
+          value: JOB_COMPLETION_INDEX   
+        - name: INDEX_TARGET
+          value: "<url of index location>"
+        resources:
+          limits:
+            cpu:     4
+            memory:  5Gi
+          requests:
+            cpu:     500m
+            memory:  500Mi
+        
+        

exp/lighthorizon/index/backend/file.go

@@ -20,11 +20,19 @@ type FileBackend struct {
 	parallel uint32
 }
 
+// NewFileBackend connects to indices stored at `dir`, creating the directory if one doesn't
+// exist, and uses `parallel` to control how many workers to use when flushing to disk.
 func NewFileBackend(dir string, parallel uint32) (*FileBackend, error) {
 	if parallel <= 0 {
 		parallel = 1
 	}
 
+	err := os.MkdirAll(dir, fs.ModeDir|0755)
+	if err != nil {
+		log.Errorf("Unable to mkdir %s, %v", dir, err)
+		return nil, err
+	}
+
 	return &FileBackend{
 		dir:      dir,
 		parallel: parallel,

exp/lighthorizon/index/cmd/batch/map/main.go

@@ -24,7 +24,7 @@ type BatchConfig struct {
 
 const (
 	batchSizeEnv         = "BATCH_SIZE"
-	jobIndexEnv          = "AWS_BATCH_JOB_ARRAY_INDEX"
+	jobIndexEnvName      = "JOB_INDEX_ENV"
 	firstCheckpointEnv   = "FIRST_CHECKPOINT"
 	txmetaSourceUrlEnv   = "TXMETA_SOURCE"
 	indexTargetUrlEnv    = "INDEX_TARGET"
@@ -39,6 +39,10 @@ func NewBatchConfig() (*BatchConfig, error) {
 		return nil, errors.New("required parameter: " + indexTargetUrlEnv)
 	}
 
+	jobIndexEnv := os.Getenv(jobIndexEnvName)
+	if jobIndexEnv == "" {
+		return nil, errors.New("env variable can't be empty " + jobIndexEnvName)
+	}
 	jobIndex, err := strconv.ParseUint(os.Getenv(jobIndexEnv), 10, 32)
 	if err != nil {
 		return nil, errors.Wrap(err, "invalid parameter "+jobIndexEnv)

exp/lighthorizon/index/cmd/batch/reduce/main.go

@@ -37,11 +37,16 @@ func ReduceConfigFromEnvironment() (*ReduceConfig, error) {
 		mapJobsEnv         = "MAP_JOB_COUNT"
 		reduceJobsEnv      = "REDUCE_JOB_COUNT"
 		workerCountEnv     = "WORKER_COUNT"
-		jobIndexEnv        = "AWS_BATCH_JOB_ARRAY_INDEX"
+		jobIndexEnvName    = "JOB_INDEX_ENV"
 		indexRootSourceEnv = "INDEX_SOURCE_ROOT"
 		indexTargetEnv     = "INDEX_TARGET"
 	)
 
+	jobIndexEnv := os.Getenv(jobIndexEnvName)
+	if jobIndexEnv == "" {
+		return nil, errors.New("env variable can't be empty " + jobIndexEnvName)
+	}
+
 	jobIndex, err := strconv.ParseUint(strings.TrimSpace(os.Getenv(jobIndexEnv)), 10, 32)
 	if err != nil {
 		return nil, errors.Wrap(err, "invalid parameter "+jobIndexEnv)

exp/lighthorizon/index/cmd/map.sh

@@ -68,7 +68,7 @@ for (( i=0; i < $BATCH_COUNT; i++ ))
 do
     echo -n "Creating map job $i... "
 
-    NETWORK_PASSPHRASE='testnet' MODULES='accounts_unbacked,transactions' \
+    NETWORK_PASSPHRASE='testnet' JOB_INDEX_ENV='AWS_BATCH_JOB_ARRAY_INDEX' MODULES='accounts_unbacked,transactions' \
     AWS_BATCH_JOB_ARRAY_INDEX=$i BATCH_SIZE=$BATCH_SIZE FIRST_CHECKPOINT=$FIRST \
     TXMETA_SOURCE=file://$1 INDEX_TARGET=file://$2 WORKER_COUNT=1 \
         ./map &

exp/lighthorizon/index/cmd/reduce.sh

@@ -47,7 +47,7 @@ for (( i=0; i < $REDUCE_JOB_COUNT; i++ ))
 do
     echo -n "Creating reduce job $i... "
 
-    AWS_BATCH_JOB_ARRAY_INDEX=$i MAP_JOB_COUNT=$MAP_JOB_COUNT \
+    AWS_BATCH_JOB_ARRAY_INDEX=$i JOB_INDEX_ENV="AWS_BATCH_JOB_ARRAY_INDEX" MAP_JOB_COUNT=$MAP_JOB_COUNT \
     REDUCE_JOB_COUNT=$REDUCE_JOB_COUNT WORKER_COUNT=4 \
     INDEX_SOURCE_ROOT=file://$1 INDEX_TARGET=file://$2 \
         timeout -k 30s 10s ./reduce &