Merge pull request #63 from vulcanize/develop

Improve entrypoint to be more helpful

.gitignore

@@ -8,3 +8,4 @@ temp/*
 pkg/beaconclient/ssz-data/
 *.test
 ipld-eth-beacon-indexer.log
+config/local.ipld-eth-beacon-indexer-config.json

entrypoint.sh

@@ -3,17 +3,21 @@
 sleep 10
 echo "Starting ipld-eth-beacon-indexer"
 
-echo /root/ipld-eth-beacon-indexer capture ${CAPTURE_MODE} --config /root/ipld-eth-beacon-config.json
+echo /root/ipld-eth-beacon-indexer capture ${CAPTURE_MODE} --config /root/ipld-eth-beacon-config.json > /root/ipld-eth-beacon-indexer.output
 
-/root/ipld-eth-beacon-indexer capture ${CAPTURE_MODE} --config /root/ipld-eth-beacon-config.json
+/root/ipld-eth-beacon-indexer capture ${CAPTURE_MODE} --config /root/ipld-eth-beacon-config.json > /root/ipld-eth-beacon-indexer.output
 rv=$?
 
 if [ $rv != 0 ]; then
-  echo "ipld-eth-beacon-indexer startup failed"
-  echo 1 > /root/HEALTH
+  echo "ipld-eth-beacon-indexer failed"
+  echo $rv > /root/HEALTH
+  echo $rv
+  cat /root/ipld-eth-beacon-indexer.output
 else
-  echo "ipld-eth-beacon-indexer startup succeeded"
-  echo 0 > /root/HEALTH
+  echo "ipld-eth-beacon-indexer succeeded"
+  echo $rv > /root/HEALTH
+  echo $rv
+  cat /root/ipld-eth-beacon-indexer.output
 fi
 
 tail -f /dev/null

pkg/beaconclient/capturehistoric.go

@@ -100,6 +100,8 @@ func handleBatchProcess(ctx context.Context, maxWorkers int, bp BatchProcessing,
 	// Start workers
 	for w := 1; w <= maxWorkers; w++ {
 		log.WithFields(log.Fields{"maxWorkers": maxWorkers}).Debug("Starting batch  processing workers")
+
+		// Pass in function to increment metric! KnownGapProcessing or HistoricProcessing
 		go processSlotRangeWorker(ctx, workCh, errCh, db, serverEndpoint, metrics, checkDb)
 	}
 
@@ -125,6 +127,7 @@ func handleBatchProcess(ctx context.Context, maxWorkers int, bp BatchProcessing,
 				} else if slots.startSlot == slots.endSlot {
 					log.WithField("slot", slots.startSlot).Debug("Added new slot to workCh")
 					workCh <- slots.startSlot
+					processedCh <- slots
 				} else {
 					for i := slots.startSlot; i <= slots.endSlot; i++ {
 						workCh <- i

pkg/beaconclient/databasewrite.go

@@ -545,7 +545,6 @@ func IsSlotInDb(ctx context.Context, db sql.Database, slot string, blockRoot str
 	var (
 		isInBeaconState       bool
 		isInSignedBeaconBlock bool
-		err                   error
 	)
 	errG, _ := errgroup.WithContext(context.Background())
 	errG.Go(func() error {
@@ -553,6 +552,7 @@ func IsSlotInDb(ctx context.Context, db sql.Database, slot string, blockRoot str
 		case <-ctx.Done():
 			return nil
 		default:
+			var err error
 			isInBeaconState, err = checkSlotAndRoot(db, CheckBeaconStateStmt, slot, stateRoot)
 			if err != nil {
 				loghelper.LogError(err).Error("Unable to check if the slot and stateroot exist in eth_beacon.state")
@@ -565,6 +565,7 @@ func IsSlotInDb(ctx context.Context, db sql.Database, slot string, blockRoot str
 		case <-ctx.Done():
 			return nil
 		default:
+			var err error
 			isInSignedBeaconBlock, err = checkSlotAndRoot(db, CheckSignedBeaconBlockStmt, slot, blockRoot)
 			if err != nil {
 				loghelper.LogError(err).Error("Unable to check if the slot and block_root exist in eth_beacon.signed_block")

pkg/beaconclient/processhistoric.go

@@ -237,8 +237,11 @@ func removeRowPostProcess(ctx context.Context, db sql.Database, processCh <-chan
 		case slots := <-processCh:
 			// Make sure the start and end slot exist in the slots table.
 			go func() {
-				finishedProcess := false
-				for !finishedProcess {
+				log.WithFields(log.Fields{
+					"startSlot": slots.startSlot,
+					"endSlot":   slots.endSlot,
+				}).Debug("Starting to check to see if the following slots have been processed")
+				for {
 					isStartProcess, err := isSlotProcessed(db, checkProcessedStmt, strconv.Itoa(slots.startSlot))
 					if err != nil {
 						errCh <- err
@@ -248,8 +251,9 @@ func removeRowPostProcess(ctx context.Context, db sql.Database, processCh <-chan
 						errCh <- err
 					}
 					if isStartProcess && isEndProcess {
-						finishedProcess = true
+						break
 					}
+					time.Sleep(1000 * time.Millisecond)
 				}
 
 				_, err := db.Exec(context.Background(), removeStmt, strconv.Itoa(slots.startSlot), strconv.Itoa(slots.endSlot))