run

Documentation

Index

• Variables

Variables

var (
	APICmd = &cli.Command{
		Name:  "api",
		Usage: "Run the singularity API",
		Flags: []cli.Flag{
			&cli.StringFlag{
				Name:  "bind",
				Usage: "Bind address for the API server",
				Value: ":9090",
			},
		},
		Action: api.Run,
	}
)

var ContentProviderCmd = &cli.Command{
	Name:  "content-provider",
	Usage: "Start a content provider that serves retrieval requests",
	Flags: []cli.Flag{
		&cli.StringFlag{
			Category: "HTTP Retrieval",
			Name:     "http-bind",
			Usage:    "Address to bind the HTTP server to",
			Value:    "127.0.0.1:7777",
		},
		&cli.BoolFlag{
			Category: "HTTP Piece Retrieval",
			Name:     "enable-http-piece",
			Usage:    "Enable HTTP Piece retrieval",
			Aliases:  []string{"enable-http"},
			Value:    true,
		},
		&cli.BoolFlag{
			Category: "HTTP Piece Metadata Retrieval",
			Name:     "enable-http-piece-metadata",
			Usage:    "Enable HTTP Piece Metadata, this is to be used with the download server",
			Value:    true,
		},
		&cli.BoolFlag{
			Category: "Bitswap Retrieval",
			Name:     "enable-bitswap",
			Usage:    "Enable bitswap retrieval",
			Value:    false,
		},
		&cli.StringFlag{
			Category:    "Bitswap Retrieval",
			Name:        "libp2p-identity-key",
			Usage:       "The base64 encoded private key for libp2p peer",
			Value:       "",
			DefaultText: "AutoGenerated",
		},
		&cli.StringSliceFlag{
			Category: "Bitswap Retrieval",
			Name:     "libp2p-listen",
			Usage:    "Addresses to listen on for libp2p connections",
		},
	},
	Action: func(c *cli.Context) error {
		db, closer, err := database.OpenFromCLI(c)
		if err != nil {
			return errors.WithStack(err)
		}
		defer closer.Close()

		config := contentprovider.Config{
			HTTP: contentprovider.HTTPConfig{
				EnablePiece:         c.Bool("enable-http-piece"),
				EnablePieceMetadata: c.Bool("enable-http-piece-metadata"),
				Bind:                c.String("http-bind"),
			},
			Bitswap: contentprovider.BitswapConfig{
				Enable:           c.Bool("enable-bitswap"),
				IdentityKey:      c.String("libp2p-identity-key"),
				ListenMultiAddrs: c.StringSlice("libp2p-listen"),
			},
		}

		s, err := contentprovider.NewService(db, config)
		if err != nil {
			return errors.WithStack(err)
		}
		return s.Start(c.Context)
	},
}

var DatasetWorkerCmd = &cli.Command{
	Name:  "dataset-worker",
	Usage: "Start a dataset preparation worker to process dataset scanning and preparation tasks",
	Flags: []cli.Flag{
		&cli.IntFlag{
			Name:  "concurrency",
			Usage: "Number of concurrent workers to run",
			Value: 1,
		},
		&cli.BoolFlag{
			Name:  "enable-scan",
			Usage: "Enable scanning of datasets",
			Value: true,
		},
		&cli.BoolFlag{
			Name:  "enable-pack",
			Usage: "Enable packing of datasets that calculates CIDs and packs them into CAR files",
			Value: true,
		},
		&cli.BoolFlag{
			Name:  "enable-dag",
			Usage: "Enable dag generation of datasets that maintains the directory structure of datasets",
			Value: true,
		},
		&cli.BoolFlag{
			Name:  "exit-on-complete",
			Usage: "Exit the worker when there is no more work to do",
			Value: false,
		},
		&cli.BoolFlag{
			Name:  "exit-on-error",
			Usage: "Exit the worker when there is any error",
			Value: false,
		},
		&cli.DurationFlag{
			Name:  "min-interval",
			Usage: "How often to check for new jobs (minimum)",
			Value: 5 * time.Second,
		},
		&cli.DurationFlag{
			Name:  "max-interval",
			Usage: "How often to check for new jobs (maximum)",
			Value: 160 * time.Second,
		},
	},
	Action: func(c *cli.Context) error {
		db, closer, err := database.OpenFromCLI(c)
		if err != nil {
			return errors.WithStack(err)
		}
		defer closer.Close()
		worker := datasetworker.NewWorker(
			db,
			datasetworker.Config{
				Concurrency:    c.Int("concurrency"),
				EnableScan:     c.Bool("enable-scan"),
				EnablePack:     c.Bool("enable-pack"),
				EnableDag:      c.Bool("enable-dag"),
				ExitOnComplete: c.Bool("exit-on-complete"),
				ExitOnError:    c.Bool("exit-on-error"),
				MinInterval:    c.Duration("min-interval"),
				MaxInterval:    c.Duration("max-interval"),
			})
		err = worker.Run(c.Context)
		if err != nil {
			return errors.WithStack(err)
		}
		return nil
	},
}

var DealPusherCmd = &cli.Command{
	Name:  "deal-pusher",
	Usage: "Start a deal pusher that monitors deal schedules and pushes deals to storage providers",
	Flags: []cli.Flag{
		&cli.UintFlag{
			Name:    "deal-attempts",
			Usage:   "Number of times to attempt a deal before giving up",
			Aliases: []string{"d"},
			Value:   3,
		},
		&cli.UintFlag{
			Name:        "max-replication-factor",
			Usage:       "Max number of replicas for each individual PieceCID across all clients and providers",
			Aliases:     []string{"M"},
			DefaultText: "Unlimited",
		},
	},
	Action: func(c *cli.Context) error {
		db, closer, err := database.OpenFromCLI(c)
		if err != nil {
			return errors.WithStack(err)
		}
		defer closer.Close()
		lotusAPI := c.String("lotus-api")
		lotusToken := c.String("lotus-token")
		err = epochutil.Initialize(c.Context, lotusAPI, lotusToken)
		if err != nil {
			return errors.WithStack(err)
		}

		dm, err := dealpusher.NewDealPusher(db, c.String("lotus-api"), c.String("lotus-token"), c.Uint("deal-attempts"), c.Uint("max-replication-factor"))
		if err != nil {
			return errors.WithStack(err)
		}
		return service.StartServers(c.Context, dealpusher.Logger, dm)
	},
}

var DealTrackerCmd = &cli.Command{
	Name:  "deal-tracker",
	Usage: "Start a deal tracker that tracks the deal for all relevant wallets",
	Flags: []cli.Flag{
		&cli.StringFlag{
			Name:    "market-deal-url",
			Usage:   "The URL for ZST compressed state market deals json. Set to empty to use Lotus API.",
			Aliases: []string{"m"},
			EnvVars: []string{"MARKET_DEAL_URL"},
			Value:   "https://marketdeals.s3.amazonaws.com/StateMarketDeals.json.zst",
		},
		&cli.DurationFlag{
			Name:    "interval",
			Usage:   "How often to check for new deals",
			Aliases: []string{"i"},
			Value:   1 * time.Hour,
		},
		&cli.BoolFlag{
			Name:  "once",
			Usage: "Run once and exit",
			Value: false,
		},
	},
	Action: func(c *cli.Context) error {
		db, closer, err := database.OpenFromCLI(c)
		if err != nil {
			return errors.WithStack(err)
		}
		defer closer.Close()

		lotusAPI := c.String("lotus-api")
		lotusToken := c.String("lotus-token")
		err = epochutil.Initialize(c.Context, lotusAPI, lotusToken)
		if err != nil {
			return errors.WithStack(err)
		}

		tracker := dealtracker.NewDealTracker(db,
			c.Duration("interval"),
			c.String("market-deal-url"),
			c.String("lotus-api"),
			c.String("lotus-token"),
			c.Bool("once"),
		)

		return service.StartServers(c.Context, dealtracker.Logger, &tracker)
	},
}

var DownloadServerCmd = &cli.Command{
	Name:        "download-server",
	Usage:       "An HTTP server connecting to remote metadata API to offer CAR file downloads",
	Description: "Example Usage:\n  singularity run download-server --metadata-api \"http://remote-metadata-api:7777\" --bind \"127.0.0.1:8888\"",
	Flags: func() []cli.Flag {
		flags := []cli.Flag{
			&cli.StringFlag{
				Name:     "metadata-api",
				Usage:    "URL of the metadata API",
				Value:    "http://127.0.0.1:7777",
				Category: "General Config",
			},
			&cli.StringFlag{
				Name:     "bind",
				Usage:    "Address to bind the HTTP server to",
				Value:    "127.0.0.1:8888",
				Category: "General Config",
			},
		}

		flags = append(flags, storage.HTTPClientConfigFlagsForUpdate...)
		flags = append(flags, storage.CommonConfigFlags...)

		keys := make(map[string]struct{})
		for _, backend := range storagesystem.Backends {
			for _, providerOptions := range backend.ProviderOptions {
				for _, option := range providerOptions.Options {
					if !model.IsSecretConfigName(option.Name) {
						continue
					}
					flag := option.ToCLIFlag(backend.Prefix+"-", false, backend.Description)
					if _, ok := keys[flag.Names()[0]]; ok {
						continue
					}
					keys[flag.Names()[0]] = struct{}{}
					flags = append(flags, flag)
				}
			}
		}
		return flags
	}(),
	Action: func(c *cli.Context) error {
		api := c.String("metadata-api")
		bind := c.String("bind")
		config := map[string]string{}
		for _, key := range c.LocalFlagNames() {
			if c.IsSet(key) {
				if slices.Contains([]string{"api", "bind"}, key) {
					continue
				}
				value := c.String(key)
				config[key] = value
			}
		}
		clientConfig, err := storage.GetClientConfigForUpdate(c)
		if err != nil {
			return errors.WithStack(err)
		}

		server := downloadserver.NewDownloadServer(bind, api, config, *clientConfig)
		return service.StartServers(c.Context, downloadserver.Logger, server)
	},
}