[WIP] IPFS Cluster Integration

.gitignore

@@ -41,6 +41,7 @@ _testmain.go
 .gx/
 .ipfs
 .textile*
+.cluster*
 
 # Development environment files
 .ackrc

cluster/connector.go

@@ -0,0 +1,187 @@
+package cluster
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io/ioutil"
+
+	icid "github.com/ipfs/go-cid"
+	"github.com/ipfs/go-ipfs/core"
+	"github.com/ipfs/go-ipfs/core/coreapi"
+	"github.com/ipfs/go-ipfs/core/corerepo"
+	"github.com/ipfs/go-ipfs/pin"
+	iface "github.com/ipfs/interface-go-ipfs-core"
+	"github.com/ipfs/interface-go-ipfs-core/options"
+	"github.com/ipfs/interface-go-ipfs-core/path"
+	"github.com/ipfs/ipfs-cluster/api"
+	corepeer "github.com/libp2p/go-libp2p-core/peer"
+	rpc "github.com/libp2p/go-libp2p-gorpc"
+	peer "github.com/libp2p/go-libp2p-peer"
+	ma "github.com/multiformats/go-multiaddr"
+)
+
+type Connector struct {
+	node  *core.IpfsNode
+	api   iface.CoreAPI
+	peers func(ctx context.Context) []*api.ID
+}
+
+func NewConnector(node *core.IpfsNode, peers func(ctx context.Context) []*api.ID) (*Connector, error) {
+	capi, err := coreapi.NewCoreAPI(node)
+	if err != nil {
+		return nil, err
+	}
+	return &Connector{
+		node:  node,
+		api:   capi,
+		peers: peers,
+	}, nil
+}
+
+func (c *Connector) ID(context.Context) (*api.IPFSID, error) {
+	var addrs []api.Multiaddr
+	for _, addr := range c.node.PeerHost.Addrs() {
+		addrs = append(addrs, api.Multiaddr{Multiaddr: addr})
+	}
+	return &api.IPFSID{
+		ID:        c.node.Identity,
+		Addresses: addrs,
+	}, nil
+}
+
+func (c *Connector) SetClient(client *rpc.Client) {
+	// noop
+}
+
+func (c *Connector) Shutdown(ctx context.Context) error {
+	// noop
+	return nil
+}
+
+// @todo handle maxDepth
+func (c *Connector) Pin(ctx context.Context, cid icid.Cid, maxDepth int) error {
+	return c.api.Pin().Add(ctx, path.New(cid.String()))
+}
+
+func (c *Connector) Unpin(ctx context.Context, cid icid.Cid) error {
+	return c.api.Pin().Rm(ctx, path.New(cid.String()))
+}
+
+func (c *Connector) PinLsCid(ctx context.Context, cid icid.Cid) (api.IPFSPinStatus, error) {
+	pins, err := c.node.Pinning.CheckIfPinned(cid)
+	if err != nil {
+		return api.IPFSPinStatusError, err
+	}
+	if len(pins) == 0 {
+		return api.IPFSPinStatusError, fmt.Errorf("invalid pin check result")
+	}
+	return c.pinModeToStatus(pins[0].Mode), nil
+}
+
+func (c *Connector) PinLs(ctx context.Context, typeFilter string) (map[string]api.IPFSPinStatus, error) {
+	pins, err := c.api.Pin().Ls(ctx, c.pinFilterToOption(typeFilter))
+	if err != nil {
+		return nil, err
+	}
+	statusMap := make(map[string]api.IPFSPinStatus)
+	for _, p := range pins {
+		mode, ok := pin.StringToMode(p.Type())
+		if !ok {
+			continue
+		}
+		statusMap[p.Path().String()] = c.pinModeToStatus(mode)
+	}
+	return statusMap, nil
+}
+
+func (c *Connector) ConnectSwarms(ctx context.Context) error {
+	for _, p := range c.peers(ctx) {
+		log.Debugf("cluster dialing %s", p.ID.Pretty())
+		var addrs []ma.Multiaddr
+		for _, addr := range p.Addresses {
+			addrs = append(addrs, addr.Multiaddr)
+		}
+		err := c.api.Swarm().Connect(ctx, corepeer.AddrInfo{
+			ID:    p.ID,
+			Addrs: addrs,
+		})
+		if err != nil {
+			return err
+		}
+		log.Debugf("cluster connected to %s")
+	}
+	return nil
+}
+
+func (c *Connector) SwarmPeers(ctx context.Context) ([]peer.ID, error) {
+	conns, err := c.api.Swarm().Peers(ctx)
+	if err != nil {
+		return nil, err
+	}
+	var peers []peer.ID
+	for _, c := range conns {
+		peers = append(peers, c.ID())
+	}
+	return peers, nil
+}
+
+func (c *Connector) ConfigKey(keypath string) (interface{}, error) {
+	return c.node.Repo.GetConfigKey(keypath)
+}
+
+func (c *Connector) RepoStat(ctx context.Context) (*api.IPFSRepoStat, error) {
+	stat, err := corerepo.RepoStat(ctx, c.node)
+	if err != nil {
+		return nil, err
+	}
+	return &api.IPFSRepoStat{
+		RepoSize:   stat.RepoSize,
+		StorageMax: stat.StorageMax,
+	}, nil
+}
+
+func (c *Connector) Resolve(ctx context.Context, pth string) (icid.Cid, error) {
+	res, err := c.api.ResolvePath(ctx, path.New(pth))
+	if err != nil {
+		return icid.Undef, err
+	}
+	return res.Cid(), nil
+}
+
+func (c *Connector) BlockPut(ctx context.Context, b *api.NodeWithMeta) error {
+	_, err := c.api.Block().Put(ctx, bytes.NewReader(b.Data), options.Block.Format(b.Format))
+	return err
+}
+
+func (c *Connector) BlockGet(ctx context.Context, cid icid.Cid) ([]byte, error) {
+	r, err := c.api.Block().Get(ctx, path.New(cid.String()))
+	if err != nil {
+		return nil, err
+	}
+	return ioutil.ReadAll(r)
+}
+
+func (c *Connector) pinModeToStatus(mode pin.Mode) api.IPFSPinStatus {
+	switch mode {
+	case pin.Recursive:
+		return api.IPFSPinStatusRecursive
+	case pin.Direct:
+		return api.IPFSPinStatusDirect
+	case pin.Indirect:
+		return api.IPFSPinStatusIndirect
+	case pin.Internal:
+		return api.IPFSPinStatusDirect
+	case pin.NotPinned:
+		return api.IPFSPinStatusUnpinned
+	default:
+		return api.IPFSPinStatusError
+	}
+}
+
+func (c *Connector) pinFilterToOption(typeFilter string) options.PinLsOption {
+	return func(settings *options.PinLsSettings) error {
+		settings.Type = typeFilter
+		return nil
+	}
+}

cluster/main.go

@@ -0,0 +1,192 @@
+package cluster
+
+import (
+	"context"
+	"fmt"
+	"path/filepath"
+
+	ds "github.com/ipfs/go-datastore"
+	logging "github.com/ipfs/go-log"
+	ipfscluster "github.com/ipfs/ipfs-cluster"
+	"github.com/ipfs/ipfs-cluster/allocator/ascendalloc"
+	"github.com/ipfs/ipfs-cluster/allocator/descendalloc"
+	"github.com/ipfs/ipfs-cluster/config"
+	"github.com/ipfs/ipfs-cluster/consensus/crdt"
+	"github.com/ipfs/ipfs-cluster/informer/disk"
+	"github.com/ipfs/ipfs-cluster/informer/numpin"
+	"github.com/ipfs/ipfs-cluster/monitor/pubsubmon"
+	"github.com/ipfs/ipfs-cluster/observations"
+	"github.com/ipfs/ipfs-cluster/pintracker/maptracker"
+	"github.com/ipfs/ipfs-cluster/pintracker/stateless"
+	host "github.com/libp2p/go-libp2p-host"
+	dht "github.com/libp2p/go-libp2p-kad-dht"
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
+	ma "github.com/multiformats/go-multiaddr"
+)
+
+var log = logging.Logger("tex-cluster")
+
+func InitCluster(repoPath, listenAddr string) error {
+	cfgMgr, cfgs := makeClusterConfigs()
+	err := cfgMgr.Default()
+	if err != nil {
+		return err
+	}
+
+	if listenAddr != "" {
+		addr, err := ma.NewMultiaddr(listenAddr)
+		if err != nil {
+			return err
+		}
+		cfgs.ClusterCfg.ListenAddr = addr
+	}
+
+	return cfgMgr.SaveJSON(ConfigPath(repoPath))
+}
+
+func ConfigPath(repoPath string) string {
+	return filepath.Join(repoPath, "service.json")
+}
+
+func MakeAndLoadConfigs(repoPath string) (*config.Manager, *cfgs, error) {
+	cfgMgr, cfgs := makeClusterConfigs()
+	err := cfgMgr.LoadJSONFromFile(ConfigPath(repoPath))
+	if err != nil {
+		return nil, nil, err
+	}
+	return cfgMgr, cfgs, nil
+}
+
+func ParseBootstraps(addrs []string) ([]ma.Multiaddr, error) {
+	var parsed []ma.Multiaddr
+	for _, a := range addrs {
+		p, err := ma.NewMultiaddr(a)
+		if err != nil {
+			return nil, err
+		}
+		parsed = append(parsed, p)
+	}
+	return parsed, nil
+}
+
+func Bootstrap(ctx context.Context, cluster *ipfscluster.Cluster, cons ipfscluster.Consensus, bootstraps []ma.Multiaddr) {
+	for _, bstrap := range bootstraps {
+		log.Infof("Bootstrapping to %s", bstrap)
+		err := cluster.Join(ctx, bstrap)
+		if err != nil {
+			log.Errorf("bootstrap to %s failed: %s", bstrap, err)
+		} else {
+			for _, p := range cluster.Peers(ctx) {
+				err = cons.Trust(ctx, p.ID)
+				if err != nil {
+					log.Errorf("failed to trust %s: %s", p.ID.Pretty(), err)
+				}
+			}
+		}
+	}
+}
+
+func SetupAllocation(
+	name string,
+	diskInfCfg *disk.Config,
+	numpinInfCfg *numpin.Config,
+) (ipfscluster.Informer, ipfscluster.PinAllocator, error) {
+	switch name {
+	case "disk", "disk-freespace":
+		informer, err := disk.NewInformer(diskInfCfg)
+		if err != nil {
+			return nil, nil, err
+		}
+		return informer, descendalloc.NewAllocator(), nil
+	case "disk-reposize":
+		informer, err := disk.NewInformer(diskInfCfg)
+		if err != nil {
+			return nil, nil, err
+		}
+		return informer, ascendalloc.NewAllocator(), nil
+	case "numpin", "pincount":
+		informer, err := numpin.NewInformer(numpinInfCfg)
+		if err != nil {
+			return nil, nil, err
+		}
+		return informer, ascendalloc.NewAllocator(), nil
+	default:
+		return nil, nil, fmt.Errorf("unknown allocation strategy")
+	}
+}
+
+func SetupPinTracker(
+	name string,
+	h host.Host,
+	mapCfg *maptracker.Config,
+	statelessCfg *stateless.Config,
+	peerName string,
+) (ipfscluster.PinTracker, error) {
+	switch name {
+	case "map":
+		ptrk := maptracker.NewMapPinTracker(mapCfg, h.ID(), peerName)
+		log.Debug("map pintracker loaded")
+		return ptrk, nil
+	case "stateless":
+		ptrk := stateless.New(statelessCfg, h.ID(), peerName)
+		log.Debug("stateless pintracker loaded")
+		return ptrk, nil
+	default:
+		return nil, fmt.Errorf("unknown pintracker type")
+	}
+}
+
+func SetupConsensus(
+	h host.Host,
+	dht *dht.IpfsDHT,
+	pubsub *pubsub.PubSub,
+	crdtCfg *crdt.Config,
+	store ds.Datastore,
+) (ipfscluster.Consensus, error) {
+	convrdt, err := crdt.New(h, dht, pubsub, crdtCfg, store)
+	if err != nil {
+		return nil, fmt.Errorf("error creating CRDT component: %s", err)
+	}
+	return convrdt, nil
+}
+
+type cfgs struct {
+	ClusterCfg          *ipfscluster.Config
+	CrdtCfg             *crdt.Config
+	MaptrackerCfg       *maptracker.Config
+	StatelessTrackerCfg *stateless.Config
+	PubsubmonCfg        *pubsubmon.Config
+	DiskInfCfg          *disk.Config
+	NumpinInfCfg        *numpin.Config
+	TracingCfg          *observations.TracingConfig
+}
+
+func makeClusterConfigs() (*config.Manager, *cfgs) {
+	cfg := config.NewManager()
+	clusterCfg := &ipfscluster.Config{}
+	crdtCfg := &crdt.Config{}
+	maptrackerCfg := &maptracker.Config{}
+	statelessCfg := &stateless.Config{}
+	pubsubmonCfg := &pubsubmon.Config{}
+	diskInfCfg := &disk.Config{}
+	numpinInfCfg := &numpin.Config{}
+	tracingCfg := &observations.TracingConfig{}
+	cfg.RegisterComponent(config.Cluster, clusterCfg)
+	cfg.RegisterComponent(config.Consensus, crdtCfg)
+	cfg.RegisterComponent(config.PinTracker, maptrackerCfg)
+	cfg.RegisterComponent(config.PinTracker, statelessCfg)
+	cfg.RegisterComponent(config.Monitor, pubsubmonCfg)
+	cfg.RegisterComponent(config.Informer, diskInfCfg)
+	cfg.RegisterComponent(config.Informer, numpinInfCfg)
+	cfg.RegisterComponent(config.Observations, tracingCfg)
+	return cfg, &cfgs{
+		clusterCfg,
+		crdtCfg,
+		maptrackerCfg,
+		statelessCfg,
+		pubsubmonCfg,
+		diskInfCfg,
+		numpinInfCfg,
+		tracingCfg,
+	}
+}