Add support for a helia container

cmd_run.go

@@ -43,6 +43,12 @@ var RunCommand = &cli.Command{
 			EnvVars: []string{"TIROS_RUN_TIMES"},
 			Value:   3,
 		},
+		&cli.BoolFlag{
+			Name:    "lookup-providers",
+			Usage:   "Whether to lookup website providers",
+			EnvVars: []string{"TIROS_RUN_LOOKUP_PROVIDERS"},
+			Value:   true,
+		},
 		&cli.BoolFlag{
 			Name:    "dry-run",
 			Usage:   "Whether to skip DB interactions",
@@ -79,21 +85,21 @@ var RunCommand = &cli.Command{
 			EnvVars: []string{"TIROS_RUN_DATABASE_SSL_MODE"},
 		},
 		&cli.StringFlag{
-			Name:    "kubo-host",
-			Usage:   "port to reach the Kubo Gateway",
-			EnvVars: []string{"TIROS_RUN_KUBO_HOST"},
+			Name:    "ipfs-host",
+			Usage:   "host at which to reach the IPFS Gateway",
+			EnvVars: []string{"TIROS_RUN_IPFS_HOST", "TIROS_RUN_KUBO_HOST" /* <- legacy */},
 			Value:   "localhost",
 		},
 		&cli.IntFlag{
-			Name:    "kubo-api-port",
-			Usage:   "port to reach the Kubo API",
-			EnvVars: []string{"TIROS_RUN_KUBO_API_PORT"},
+			Name:    "ipfs-api-port",
+			Usage:   "port to reach a Kubo-compatible RPC API",
+			EnvVars: []string{"TIROS_RUN_IPFS_API_PORT", "TIROS_RUN_KUBO_API_PORT" /* <- legacy */},
 			Value:   5001,
 		},
 		&cli.IntFlag{
-			Name:    "kubo-gateway-port",
-			Usage:   "port to reach the Kubo Gateway",
-			EnvVars: []string{"TIROS_RUN_KUBO_GATEWAY_PORT"},
+			Name:    "ipfs-gateway-port",
+			Usage:   "port to reach the IPFS Gateway",
+			EnvVars: []string{"TIROS_RUN_IPFS_GATEWAY_PORT", "TIROS_RUN_KUBO_GATEWAY_PORT" /* <- legacy */},
 			Value:   8080,
 		},
 		&cli.IntFlag{
@@ -119,13 +125,19 @@ var RunCommand = &cli.Command{
 			Usage:   "Path to the Udger DB",
 			EnvVars: []string{"TIROS_UDGER_DB_PATH"},
 		},
+		&cli.StringFlag{
+			Name:    "ipfs-implementation",
+			Usage:   "Which implementation are we testing (KUBO, HELIA)",
+			EnvVars: []string{"TIROS_IPFS_IMPLEMENTATION"},
+			Value:   "KUBO",
+		},
 	},
 	Action: RunAction,
 }
 
 type tiros struct {
 	dbClient IDBClient
-	kubo     *shell.Shell
+	ipfs     *shell.Shell
 	dbRun    *models.Run
 	mmClient *maxmind.Client
 	uClient  *udger.Client
@@ -145,8 +157,11 @@ func RunAction(c *cli.Context) error {
 		}
 	}
 
-	// Initialize kubo client
-	kubo := shell.NewShell(fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", c.Int("kubo-api-port")))
+	// Initialize ipfs client
+	var ipfsClient *shell.Shell
+	if c.Int("ipfs-api-port") != 0 {
+		ipfsClient = shell.NewShell(fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", c.Int("ipfs-api-port")))
+	}
 
 	// Initialize maxmind client
 	mmClient, err := maxmind.NewClient()
@@ -163,7 +178,7 @@ func RunAction(c *cli.Context) error {
 	// configure tiros struct
 	t := tiros{
 		dbClient: dbClient,
-		kubo:     kubo,
+		ipfs:     ipfsClient,
 		mmClient: mmClient,
 		uClient:  uClient,
 	}
@@ -182,7 +197,7 @@ func RunAction(c *cli.Context) error {
 	}()
 
 	// shuffle websites, so that we have a different order in which we request the websites.
-	// If we didn't do this a single website would always be requested with a comparatively "cold" kubo node.
+	// If we didn't do this a single website would always be requested with a comparatively "cold" ipfs node.
 	websites := c.StringSlice("websites")
 	rand.Seed(time.Now().UnixNano())
 	rand.Shuffle(len(websites), func(i, j int) {
@@ -199,7 +214,12 @@ func RunAction(c *cli.Context) error {
 	probeResults := make(chan *probeResult)
 
 	go t.measureWebsites(c, websites, probeResults)
-	go t.findAllProviders(c, websites, providerResults)
+
+	if c.Bool("lookup-providers") {
+		go t.findAllProviders(c, websites, providerResults)
+	} else {
+		close(providerResults)
+	}
 
 	for {
 		select {
@@ -242,12 +262,13 @@ func RunAction(c *cli.Context) error {
 }
 
 func (t *tiros) InitRun(c *cli.Context) (*models.Run, error) {
-	version, sha, err := t.kubo.Version()
+	version, sha, err := t.ipfs.Version()
 	if err != nil {
-		return nil, fmt.Errorf("kubo api offline: %w", err)
+		return nil, fmt.Errorf("ipfs api offline: %w", err)
 	}
 
-	dbRun, err := t.dbClient.InsertRun(c, fmt.Sprintf("%s-%s", version, sha))
+	ipfsImpl := c.String("ipfs-implementation")
+	dbRun, err := t.dbClient.InsertRun(c, ipfsImpl, fmt.Sprintf("%s-%s", version, sha))
 	if err != nil {
 		return nil, fmt.Errorf("insert run: %w", err)
 	}

db_client.go

@@ -28,7 +28,7 @@ import (
 var migrations embed.FS
 
 type IDBClient interface {
-	InsertRun(c *cli.Context, version string) (*models.Run, error)
+	InsertRun(c *cli.Context, ipfsImpl string, version string) (*models.Run, error)
 	SaveMeasurement(c *cli.Context, dbRun *models.Run, pr *probeResult) (*models.Measurement, error)
 	SaveProvider(c *cli.Context, dbRun *models.Run, provider *provider) (*models.Provider, error)
 	SealRun(ctx context.Context, dbRun *models.Run) (*models.Run, error)
@@ -135,7 +135,7 @@ func (db *DBClient) SealRun(ctx context.Context, dbRun *models.Run) (*models.Run
 	return dbRun, err
 }
 
-func (db *DBClient) InsertRun(c *cli.Context, version string) (*models.Run, error) {
+func (db *DBClient) InsertRun(c *cli.Context, ipfsImpl string, version string) (*models.Run, error) {
 	log.Infoln("Inserting Run...")
 
 	websites := make([]string, len(c.StringSlice("websites")))
@@ -148,6 +148,7 @@ func (db *DBClient) InsertRun(c *cli.Context, version string) (*models.Run, erro
 		Region:   c.String("region"),
 		Websites: websites,
 		Version:  version,
+		IpfsImpl: ipfsImpl,
 		Times:    int16(c.Int("times")),
 		CPU:      c.Int("cpu"),
 		Memory:   c.Int("memory"),

db_dummy.go

@@ -24,10 +24,11 @@ func (D DBDummyClient) SealRun(ctx context.Context, dbRun *models.Run) (*models.
 	return nil, nil
 }
 
-func (D DBDummyClient) InsertRun(c *cli.Context, version string) (*models.Run, error) {
+func (D DBDummyClient) InsertRun(c *cli.Context, ipfsImpl string, version string) (*models.Run, error) {
 	return &models.Run{
-		ID:     2,
-		Region: "dummy",
+		ID:       2,
+		Region:   "dummy",
+		IpfsImpl: ipfsImpl,
 	}, nil
 }
 

docker-compose.yml

@@ -2,16 +2,15 @@ version: "3.9"
 name: tiros
 services:
   ipfs:
-    image: ipfs/kubo:v0.19.0
+    image: helia:latest
     restart: unless-stopped
     volumes:
       - ipfs_path:/data/ipfs
       - ipfs_fuse:/ipfs
       - ipns_fuse:/ipns
+    environment:
+      DEBUG: helia-server
     ports:
-      - "4001:4001/tcp"
-      - "4001:4001/udp"
-      - "0.0.0.0:5001:5001"
       - "0.0.0.0:8080:8080"
   chrome:
     image: browserless/chrome:latest

migrations/000004_add_providers_table.down.sql

@@ -0,0 +1 @@
+DROP TABLE providers;

migrations/000006_rename_kubo_measurement_type.down.sql

@@ -0,0 +1 @@
+ALTER TYPE measurement_type RENAME VALUE 'IPFS' TO 'KUBO';

migrations/000006_rename_kubo_measurement_type.up.sql

@@ -0,0 +1 @@
+ALTER TYPE measurement_type RENAME VALUE 'KUBO' TO 'IPFS';

migrations/000007_add_ipfs_impl_column_to_runs_table.down.sql

@@ -0,0 +1,6 @@
+BEGIN;
+
+ALTER TABLE runs
+    DROP COLUMN ipfs_impl;
+
+COMMIT;

migrations/000007_add_ipfs_impl_column_to_runs_table.up.sql

@@ -0,0 +1,11 @@
+BEGIN;
+
+ALTER TABLE runs
+    ADD COLUMN ipfs_impl TEXT;
+
+UPDATE runs SET ipfs_impl = 'KUBO';
+
+ALTER TABLE runs
+    ALTER COLUMN ipfs_impl SET NOT NULL;
+
+COMMIT;

probe.go

@@ -38,11 +38,11 @@ func (t *tiros) measureWebsites(c *cli.Context, websites []string, results chan<
 
 		sleepDur := time.Duration(settle) * time.Second
 
-		log.Infof("Letting Kubo settle for %s\n", sleepDur)
+		log.Infof("Letting the IPFS implementation settle for %s\n", sleepDur)
 		time.Sleep(sleepDur)
 
 		for i := 0; i < c.Int("times"); i++ {
-			for _, mType := range []string{models.MeasurementTypeKUBO, models.MeasurementTypeHTTP} {
+			for _, mType := range []string{models.MeasurementTypeIPFS, models.MeasurementTypeHTTP} {
 				for _, website := range websites {
 
 					pr, err := newProbe(c, website, mType).run()
@@ -67,9 +67,9 @@ func (t *tiros) measureWebsites(c *cli.Context, websites []string, results chan<
 
 					results <- pr
 
-					if mType == models.MeasurementTypeKUBO {
-						if err = t.KuboGC(c.Context); err != nil {
-							log.WithError(err).Warnln("error running kubo gc")
+					if mType == models.MeasurementTypeIPFS {
+						if err = t.GarbageCollect(c.Context); err != nil {
+							log.WithError(err).Warnln("error running ipfs gc")
 							continue
 						}
 					}
@@ -98,7 +98,7 @@ type probeResult struct {
 	url     string
 	website string
 
-	// measurement type (KUBO or HTTP)
+	// measurement type (IPFS or HTTP)
 	mType string
 	try   int
 
@@ -329,17 +329,17 @@ func (p *probe) close() {
 
 func websiteURL(c *cli.Context, website string, mType string) string {
 	switch mType {
-	case models.MeasurementTypeKUBO:
-		return fmt.Sprintf("http://%s:%d/ipns/%s", c.String("kubo-host"), c.Int("kubo-gateway-port"), website)
+	case models.MeasurementTypeIPFS:
+		return fmt.Sprintf("http://%s:%d/ipns/%s", c.String("ipfs-host"), c.Int("ipfs-gateway-port"), website)
 	case models.MeasurementTypeHTTP:
 		return fmt.Sprintf("https://%s", website)
 	default:
 		panic(fmt.Sprintf("unknown measurement type: %s", mType))
 	}
 }
 
-func (t *tiros) KuboGC(ctx context.Context) error {
-	return t.kubo.Request("repo/gc").Exec(ctx, nil)
+func (t *tiros) GarbageCollect(ctx context.Context) error {
+	return t.ipfs.Request("repo/gc").Exec(ctx, nil)
 }
 
 func p2f(ptr *float64) float64 {

providers.go

@@ -56,7 +56,7 @@ func (t *tiros) findProviders(ctx context.Context, website string, results chan<
 	logEntry := log.WithField("website", website)
 	logEntry.Infoln("Finding providers for", website)
 
-	nameResp, err := t.kubo.Request("name/resolve").
+	nameResp, err := t.ipfs.Request("name/resolve").
 		Option("arg", website).
 		Option("nocache", "true").
 		Option("dht-timeout", "30s").Send(ctx)
@@ -87,7 +87,7 @@ func (t *tiros) findProviders(ctx context.Context, website string, results chan<
 		return fmt.Errorf("unmarshal name/resolve response: %w", err)
 	}
 
-	findResp, err := t.kubo.
+	findResp, err := t.ipfs.
 		Request("routing/findprovs").
 		Option("arg", nrr.Path).
 		Option("num-providers", "1000").
@@ -183,7 +183,7 @@ func (t *tiros) idWorker(ctx context.Context, jobs <-chan *peer.AddrInfo, idResu
 		var out shell.IdOutput
 
 		tCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
-		err := t.kubo.Request("id", j.ID.String()).Exec(tCtx, &out)
+		err := t.ipfs.Request("id", j.ID.String()).Exec(tCtx, &out)
 		cancel()
 
 		idResults <- idResult{