Follow: wait for IPFS to be running

cmd/ipfs-cluster-follow/commands.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"time"
 
 	"github.com/ipfs/go-cid"
 	ipfscluster "github.com/ipfs/ipfs-cluster"
@@ -228,6 +229,13 @@ func runCmd(c *cli.Context) error {
 	}
 
 	fmt.Printf("Starting the IPFS Cluster follower peer for \"%s\".\nCTRL-C to stop it.\n", clusterName)
+	fmt.Println("Checking if IPFS is online (will wait for 2 minutes)...")
+	ctxIpfs, cancelIpfs := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancelIpfs()
+	err := cmdutils.WaitForIPFS(ctxIpfs)
+	if err != nil {
+		return cli.Exit("timed out waiting for IPFS to be available", 1)
+	}
 
 	setLogLevels(logLevel) // set to "info" by default.
 	// Avoid API logs polluting the screen everytime we

cmdutils/cmdutils.go

@@ -9,11 +9,14 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
 
 	ipfscluster "github.com/ipfs/ipfs-cluster"
+	ipfshttp "github.com/ipfs/ipfs-cluster/ipfsconn/ipfshttp"
 	host "github.com/libp2p/go-libp2p-host"
 	dht "github.com/libp2p/go-libp2p-kad-dht"
 	ma "github.com/multiformats/go-multiaddr"
+	"github.com/pkg/errors"
 )
 
 // RandomizePorts replaces TCP and UDP ports with random, but valid port
@@ -127,3 +130,53 @@ Note that this may corrupt the local cluster state.
 func ErrorOut(m string, a ...interface{}) {
 	fmt.Fprintf(os.Stderr, m, a...)
 }
+
+// WaitForIPFS hangs until IPFS API becomes available or the given context is
+// cancelled.  The IPFS API location is determined by the default ipfshttp
+// component configuration and can be overriden using environment variables
+// that affect that configuration.  Note that we have to do this in the blind,
+// since we want to wait for IPFS before we even fetch the IPFS component
+// configuration (because the configuration might be hosted on IPFS itself)
+func WaitForIPFS(ctx context.Context) error {
+	ipfshttpCfg := ipfshttp.Config{}
+	ipfshttpCfg.Default()
+	ipfshttpCfg.ApplyEnvVars()
+	ipfshttpCfg.ConnectSwarmsDelay = 0
+	ipfshttpCfg.Tracing = false
+	ipfscluster.SetFacilityLogLevel("ipfshttp", "critical")
+	defer ipfscluster.SetFacilityLogLevel("ipfshttp", "info")
+	ipfs, err := ipfshttp.NewConnector(&ipfshttpCfg)
+	if err != nil {
+		return errors.Wrap(err, "error creating an ipfshttp instance to wait for IPFS")
+	}
+
+	signalChan := make(chan os.Signal, 20)
+	signal.Notify(
+		signalChan,
+		syscall.SIGINT,
+		syscall.SIGTERM,
+		syscall.SIGHUP,
+	)
+
+	i := 0
+	for {
+		select {
+		case <-signalChan:
+			return errors.New("interrupted")
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+			if i%10 == 0 {
+				fmt.Printf("waiting for IPFS to become available on %s...\n", ipfshttpCfg.NodeAddr)
+			}
+			i++
+			time.Sleep(time.Second)
+			_, err := ipfs.ID(ctx)
+			if err == nil {
+				// sleep an extra second and quit
+				time.Sleep(time.Second)
+				return nil
+			}
+		}
+	}
+}