Merge pull request #30 from filecoin-project/feat/index-service

feat: add index service to resolve latest snapshot to static resource

Author: travisperson

cmd/filecoin-chain-archiver/cmds/cmds.go

@@ -9,7 +9,7 @@ import (
 
 var logger = log.Logger("filecoin-chain-archiver/cmds")
 
-var Commands = []*cli.Command{cmdCreate, cmdDefaultConfig, cmdService}
+var Commands = []*cli.Command{cmdCreate, cmdDefaultConfig, cmdService, cmdIndexService}
 
 func TrimDescription(desc string) string {
 	lines := strings.Split(desc, "\n")

cmd/filecoin-chain-archiver/cmds/config.go

@@ -15,7 +15,7 @@ var cmdDefaultConfig = &cli.Command{
 	Action: func(cctx *cli.Context) error {
 		var icfg interface{}
 
-		cfg := config.DefaultConfig()
+		cfg := config.DefaultExportWorkerConfig()
 		cfg.Nodes = append(cfg.Nodes, config.Node{
 			Address:   "/ip4/127.0.0.1/1234",
 			TokenPath: "/path/to/token",

cmd/filecoin-chain-archiver/cmds/create.go

@@ -153,12 +153,12 @@ var cmdCreate = &cli.Command{
 			return err
 		}
 
-		icfg, err := config.FromFile(flagConfigPath, &config.Config{})
+		icfg, err := config.FromFile(flagConfigPath, &config.ExportWorkerConfig{})
 		if err != nil {
 			return err
 		}
 
-		cfg := icfg.(*config.Config)
+		cfg := icfg.(*config.ExportWorkerConfig)
 
 		addrs, err := NodeMultiaddrs(cfg)
 		if err != nil {
@@ -338,7 +338,7 @@ var cmdCreate = &cli.Command{
 				ContentType:        "application/octet-stream",
 			})
 			if err != nil {
-				return fmt.Errorf("failed to upload", "object", fmt.Sprintf("%s%s.car", flagNamePrefix, name), "err", err)
+				return fmt.Errorf("failed to upload object (%s): %w", fmt.Sprintf("%s%s.car", flagNamePrefix, name), err)
 			}
 
 			logger.Infow("upload",

cmd/filecoin-chain-archiver/cmds/index_service.go

@@ -0,0 +1,177 @@
+package cmds
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"net/http"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/urfave/cli/v2"
+
+	"github.com/filecoin-project/filecoin-chain-archiver/build"
+	"github.com/filecoin-project/filecoin-chain-archiver/pkg/config"
+	"github.com/filecoin-project/filecoin-chain-archiver/pkg/index/service"
+)
+
+var cmdIndexService = &cli.Command{
+	Name:        "index-resolver-service",
+	Usage:       "Commands for the index resolver service",
+	Description: "The index resolver service provides a way to resolve snapshots",
+	Flags:       []cli.Flag{},
+	Subcommands: []*cli.Command{
+		{
+			Name:        "default-config",
+			Usage:       "prints the default configuration",
+			Description: TrimDescription(``),
+			Flags:       []cli.Flag{},
+			Action: func(cctx *cli.Context) error {
+				var icfg interface{}
+
+				cfg := config.DefaultIndexServiceConfig()
+				icfg = cfg
+
+				bs, err := config.ConfigComment(icfg)
+				if err != nil {
+					return err
+				}
+
+				fmt.Printf("%s", string(bs))
+
+				return nil
+			},
+		},
+		{
+			Name:  "run",
+			Usage: "start the service",
+			Flags: []cli.Flag{
+				&cli.StringFlag{
+					Name:    "service-listen",
+					Usage:   "host and port to listen on",
+					EnvVars: []string{"FCA_INDEX_RESOLVER_SERVICE_LISTEN"},
+					Value:   "localhost:5200",
+				},
+				&cli.StringFlag{
+					Name:    "operator-listen",
+					Usage:   "host and port to listen on",
+					EnvVars: []string{"FCA_INDEX_RESOLVER_OPERATOR_LISTEN"},
+					Value:   "localhost:5201",
+				},
+				&cli.StringFlag{
+					Name:    "config-path",
+					Usage:   "path to configuration file",
+					EnvVars: []string{"FCA_INDEX_RESOLVER_CONFIG_PATH"},
+					Value:   "./config.toml",
+				},
+			},
+			Action: func(cctx *cli.Context) error {
+				ctx, cancelFunc := context.WithCancel(context.Background())
+				ctx = context.WithValue(ctx, versionKey{}, build.Version())
+
+				signalChan := make(chan os.Signal, 1)
+				signal.Notify(signalChan, syscall.SIGQUIT, syscall.SIGINT, syscall.SIGHUP, syscall.SIGTERM)
+
+				s := service.NewIndexService(ctx)
+
+				if err := s.SetupService(cctx.String("config-path")); err != nil {
+					return err
+				}
+
+				svr := &http.Server{
+					Addr:    cctx.String("service-listen"),
+					Handler: s.ServiceRouter,
+					BaseContext: func(listener net.Listener) context.Context {
+						return context.Background()
+					},
+				}
+
+				go func() {
+					logger.Debugw("service running")
+					err := svr.ListenAndServe()
+					switch err {
+					case nil:
+					case http.ErrServerClosed:
+						logger.Infow("server closed")
+					case context.Canceled:
+						logger.Infow("context cancled")
+					default:
+						logger.Errorw("error shutting down service server", "err", err)
+					}
+				}()
+
+				if err := s.SetupOperator(); err != nil {
+					return err
+				}
+
+				osvr := http.Server{
+					Addr:    cctx.String("operator-listen"),
+					Handler: s.OperatorRouter,
+					BaseContext: func(listener net.Listener) context.Context {
+						return context.Background()
+					},
+				}
+
+				go func() {
+					logger.Debugw("operator running")
+					err := osvr.ListenAndServe()
+					switch err {
+					case nil:
+					case http.ErrServerClosed:
+						logger.Infow("server closed")
+					case context.Canceled:
+						logger.Infow("context cancled")
+					default:
+						logger.Errorw("error shutting down internal server", "err", err)
+					}
+				}()
+
+				logger.Infow("waiting for signal")
+				<-signalChan
+				s.Shutdown()
+
+				t := time.NewTimer(svrShutdownTimeout)
+
+				shutdownChan := make(chan error)
+				go func() {
+					shutdownChan <- svr.Shutdown(ctx)
+				}()
+
+				select {
+				case err := <-shutdownChan:
+					if err != nil {
+						logger.Errorw("shutdown finished with an error", "err", err)
+					} else {
+						logger.Infow("shutdown finished successfully")
+					}
+				case <-t.C:
+					logger.Warnw("shutdown timed out")
+				}
+
+				cancelFunc()
+				time.Sleep(ctxCancelWait)
+
+				logger.Infow("closing down database connections")
+				s.Close()
+
+				if err := osvr.Shutdown(ctx); err != nil {
+					switch err {
+					case nil:
+					case http.ErrServerClosed:
+						logger.Infow("server closed")
+					case context.Canceled:
+						logger.Infow("context cancled")
+					default:
+						logger.Errorw("error shutting down operator server", "err", err)
+					}
+				}
+
+				logger.Infow("existing")
+
+				return nil
+			},
+		},
+	},
+}

cmd/filecoin-chain-archiver/cmds/utils.go

@@ -10,11 +10,11 @@ import (
 
 	"github.com/filecoin-project/go-jsonrpc"
 
-	cliutil "github.com/filecoin-project/lotus/cli/util"
 	"github.com/filecoin-project/filecoin-chain-archiver/pkg/config"
+	cliutil "github.com/filecoin-project/lotus/cli/util"
 )
 
-func NodeMultiaddrs(cfg *config.Config) ([]string, error) {
+func NodeMultiaddrs(cfg *config.ExportWorkerConfig) ([]string, error) {
 	var multiaddrs []string
 
 	for _, node := range cfg.Nodes {

pkg/config/config.go

@@ -10,8 +10,12 @@ import (
 	"golang.org/x/xerrors"
 )
 
-func DefaultConfig() *Config {
-	return &Config{}
+func DefaultIndexServiceConfig() *IndexServiceConfig {
+	return &IndexServiceConfig{}
+}
+
+func DefaultExportWorkerConfig() *ExportWorkerConfig {
+	return &ExportWorkerConfig{}
 }
 
 type URL url.URL
@@ -36,10 +40,21 @@ type Node struct {
 	TokenPath string
 }
 
-type Config struct {
+type ExportWorkerConfig struct {
 	Nodes []Node
 }
 
+type S3ResolverConfig struct {
+	Endpoint      string
+	Bucket        string
+	AccessKeyPath string
+	SecretKeyPath string
+}
+
+type IndexServiceConfig struct {
+	S3Resolver S3ResolverConfig
+}
+
 func FromFile(path string, def interface{}) (interface{}, error) {
 	file, err := os.Open(path)
 	switch {

pkg/index/index.go

@@ -0,0 +1,108 @@
+package index
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/minio/minio-go/v7"
+
+	"github.com/ipfs/go-log/v2"
+)
+
+var (
+	expiryLength = 5 * time.Minute
+)
+
+var logger = log.Logger("filecoin-chain-archiver/pkg/index-resolver")
+
+type IndexS3Resolver struct {
+	client s3ClientInterface
+	bucket string
+}
+
+type s3ClientInterface interface {
+	StatObject(ctx context.Context, bucketName, objectName string, opts minio.StatObjectOptions) (minio.ObjectInfo, error)
+}
+
+type Resolver interface {
+	Resolve(context.Context, string) (string, error)
+}
+
+func NewIndexS3Resolver(client s3ClientInterface, bucket string) *IndexS3Resolver {
+	return &IndexS3Resolver{
+		client: client,
+		bucket: bucket,
+	}
+}
+
+func (i *IndexS3Resolver) Resolve(ctx context.Context, obj string) (string, error) {
+	objInfo, err := i.client.StatObject(ctx, i.bucket, obj, minio.StatObjectOptions{})
+	if err != nil {
+		return "", err
+	}
+
+	if v, ok := objInfo.Metadata["X-Amz-Website-Redirect-Location"]; ok {
+		return v[0], nil
+	}
+
+	return "", fmt.Errorf("failed to resolve link")
+}
+
+type cacheMetadata struct {
+	value  string
+	expiry time.Time
+}
+
+type CachedResolver struct {
+	resolver Resolver
+
+	cacheMu sync.Mutex
+	cache   map[string]cacheMetadata
+}
+
+func NewCachedResolver(resolver Resolver) *CachedResolver {
+	return &CachedResolver{
+		resolver: resolver,
+		cache:    make(map[string]cacheMetadata),
+	}
+}
+
+func (i *CachedResolver) read(obj string) (string, bool) {
+	i.cacheMu.Lock()
+	defer i.cacheMu.Unlock()
+	if v, ok := i.cache[obj]; ok {
+		if time.Now().Before(v.expiry) {
+			return v.value, true
+		}
+	}
+
+	return "", false
+}
+
+func (i *CachedResolver) set(obj, value string, expiry time.Time) {
+	i.cacheMu.Lock()
+	defer i.cacheMu.Unlock()
+	i.cache[obj] = cacheMetadata{
+		expiry: expiry,
+		value:  value,
+	}
+}
+
+func (i *CachedResolver) Resolve(ctx context.Context, obj string) (string, error) {
+	if v, ok := i.read(obj); ok {
+		logger.Debugw("cache hit")
+		return v, nil
+	}
+
+	value, err := i.resolver.Resolve(ctx, obj)
+	if err != nil {
+		return "", err
+	}
+
+	logger.Debugw("cache miss")
+	i.set(obj, value, time.Now().Add(expiryLength))
+
+	return value, nil
+}