package ipfscluster import ( "context" "errors" "github.com/ipfs-cluster/ipfs-cluster/api" "github.com/ipfs-cluster/ipfs-cluster/state" "github.com/ipfs-cluster/ipfs-cluster/version" peer "github.com/libp2p/go-libp2p-core/peer" rpc "github.com/libp2p/go-libp2p-gorpc" ocgorpc "github.com/lanzafame/go-libp2p-ocgorpc" "go.opencensus.io/trace" ) // RPC endpoint types w.r.t. trust level const ( // RPCClosed endpoints can only be called by the local cluster peer // on itself. RPCClosed RPCEndpointType = iota // RPCTrusted endpoints can be called by "trusted" peers. // It depends which peers are considered trusted. For example, // in "raft" mode, Cluster will all peers as "trusted". In "crdt" mode, // trusted peers are those specified in the configuration. RPCTrusted // RPCOpen endpoints can be called by any peer in the Cluster swarm. RPCOpen ) // RPCEndpointType controls how access is granted to an RPC endpoint type RPCEndpointType int const rpcStreamBufferSize = 1024 // A trick to find where something is used (i.e. Cluster.Pin): // grep -R -B 3 '"Pin"' | grep -C 1 '"Cluster"'. // This does not cover globalPinInfo*(...) broadcasts nor redirects to leader // in Raft. // newRPCServer returns a new RPC Server for Cluster. func newRPCServer(c *Cluster) (*rpc.Server, error) { var s *rpc.Server authF := func(pid peer.ID, svc, method string) bool { endpointType, ok := c.config.RPCPolicy[svc+"."+method] if !ok { return false } switch endpointType { case RPCTrusted: return c.consensus.IsTrustedPeer(c.ctx, pid) case RPCOpen: return true default: return false } } if c.config.Tracing { s = rpc.NewServer( c.host, version.RPCProtocol, rpc.WithServerStatsHandler(&ocgorpc.ServerHandler{}), rpc.WithAuthorizeFunc(authF), rpc.WithStreamBufferSize(rpcStreamBufferSize), ) } else { s = rpc.NewServer(c.host, version.RPCProtocol, rpc.WithAuthorizeFunc(authF)) } cl := &ClusterRPCAPI{c} err := s.RegisterName(RPCServiceID(cl), cl) if err != nil { return nil, err } pt := &PinTrackerRPCAPI{c.tracker} err = s.RegisterName(RPCServiceID(pt), pt) if err != nil { return nil, err } ic := &IPFSConnectorRPCAPI{c.ipfs} err = s.RegisterName(RPCServiceID(ic), ic) if err != nil { return nil, err } cons := &ConsensusRPCAPI{c.consensus} err = s.RegisterName(RPCServiceID(cons), cons) if err != nil { return nil, err } pm := &PeerMonitorRPCAPI{mon: c.monitor, pid: c.id} err = s.RegisterName(RPCServiceID(pm), pm) if err != nil { return nil, err } return s, nil } // RPCServiceID returns the Service ID for the given RPCAPI object. func RPCServiceID(rpcSvc interface{}) string { switch rpcSvc.(type) { case *ClusterRPCAPI: return "Cluster" case *PinTrackerRPCAPI: return "PinTracker" case *IPFSConnectorRPCAPI: return "IPFSConnector" case *ConsensusRPCAPI: return "Consensus" case *PeerMonitorRPCAPI: return "PeerMonitor" default: return "" } } // ClusterRPCAPI is a go-libp2p-gorpc service which provides the internal peer // API for the main cluster component. type ClusterRPCAPI struct { c *Cluster } // PinTrackerRPCAPI is a go-libp2p-gorpc service which provides the internal // peer API for the PinTracker component. type PinTrackerRPCAPI struct { tracker PinTracker } // IPFSConnectorRPCAPI is a go-libp2p-gorpc service which provides the // internal peer API for the IPFSConnector component. type IPFSConnectorRPCAPI struct { ipfs IPFSConnector } // ConsensusRPCAPI is a go-libp2p-gorpc service which provides the // internal peer API for the Consensus component. type ConsensusRPCAPI struct { cons Consensus } // PeerMonitorRPCAPI is a go-libp2p-gorpc service which provides the // internal peer API for the PeerMonitor component. type PeerMonitorRPCAPI struct { mon PeerMonitor pid peer.ID } /* Cluster component methods */ // ID runs Cluster.ID() func (rpcapi *ClusterRPCAPI) ID(ctx context.Context, in struct{}, out *api.ID) error { id := rpcapi.c.ID(ctx) *out = id return nil } // IDStream runs Cluster.ID() but in streaming form. func (rpcapi *ClusterRPCAPI) IDStream(ctx context.Context, in <-chan struct{}, out chan<- api.ID) error { defer close(out) id := rpcapi.c.ID(ctx) select { case <-ctx.Done(): return ctx.Err() case out <- id: } return nil } // Pin runs Cluster.pin(). func (rpcapi *ClusterRPCAPI) Pin(ctx context.Context, in api.Pin, out *api.Pin) error { // we do not call the Pin method directly since that method does not // allow to pin other than regular DataType pins. The adder will // however send Meta, Shard and ClusterDAG pins. pin, _, err := rpcapi.c.pin(ctx, in, []peer.ID{}) if err != nil { return err } *out = pin return nil } // Unpin runs Cluster.Unpin(). func (rpcapi *ClusterRPCAPI) Unpin(ctx context.Context, in api.Pin, out *api.Pin) error { pin, err := rpcapi.c.Unpin(ctx, in.Cid) if err != nil { return err } *out = pin return nil } // PinPath resolves path into a cid and runs Cluster.Pin(). func (rpcapi *ClusterRPCAPI) PinPath(ctx context.Context, in api.PinPath, out *api.Pin) error { pin, err := rpcapi.c.PinPath(ctx, in.Path, in.PinOptions) if err != nil { return err } *out = pin return nil } // UnpinPath resolves path into a cid and runs Cluster.Unpin(). func (rpcapi *ClusterRPCAPI) UnpinPath(ctx context.Context, in api.PinPath, out *api.Pin) error { pin, err := rpcapi.c.UnpinPath(ctx, in.Path) if err != nil { return err } *out = pin return nil } // Pins runs Cluster.Pins(). func (rpcapi *ClusterRPCAPI) Pins(ctx context.Context, in <-chan struct{}, out chan<- api.Pin) error { return rpcapi.c.Pins(ctx, out) } // PinGet runs Cluster.PinGet(). func (rpcapi *ClusterRPCAPI) PinGet(ctx context.Context, in api.Cid, out *api.Pin) error { pin, err := rpcapi.c.PinGet(ctx, in) if err != nil { return err } *out = pin return nil } // Version runs Cluster.Version(). func (rpcapi *ClusterRPCAPI) Version(ctx context.Context, in struct{}, out *api.Version) error { *out = api.Version{ Version: rpcapi.c.Version(), } return nil } // Peers runs Cluster.Peers(). func (rpcapi *ClusterRPCAPI) Peers(ctx context.Context, in <-chan struct{}, out chan<- api.ID) error { rpcapi.c.Peers(ctx, out) return nil } // PeersWithFilter runs Cluster.peersWithFilter(). func (rpcapi *ClusterRPCAPI) PeersWithFilter(ctx context.Context, in <-chan []peer.ID, out chan<- api.ID) error { peers := <-in rpcapi.c.peersWithFilter(ctx, peers, out) return nil } // PeerAdd runs Cluster.PeerAdd(). func (rpcapi *ClusterRPCAPI) PeerAdd(ctx context.Context, in peer.ID, out *api.ID) error { id, err := rpcapi.c.PeerAdd(ctx, in) if err != nil { return err } *out = *id return nil } // ConnectGraph runs Cluster.GetConnectGraph(). func (rpcapi *ClusterRPCAPI) ConnectGraph(ctx context.Context, in struct{}, out *api.ConnectGraph) error { graph, err := rpcapi.c.ConnectGraph() if err != nil { return err } *out = graph return nil } // PeerRemove runs Cluster.PeerRm(). func (rpcapi *ClusterRPCAPI) PeerRemove(ctx context.Context, in peer.ID, out *struct{}) error { return rpcapi.c.PeerRemove(ctx, in) } // Join runs Cluster.Join(). func (rpcapi *ClusterRPCAPI) Join(ctx context.Context, in api.Multiaddr, out *struct{}) error { return rpcapi.c.Join(ctx, in.Value()) } // StatusAll runs Cluster.StatusAll(). func (rpcapi *ClusterRPCAPI) StatusAll(ctx context.Context, in <-chan api.TrackerStatus, out chan<- api.GlobalPinInfo) error { filter := <-in return rpcapi.c.StatusAll(ctx, filter, out) } // StatusAllLocal runs Cluster.StatusAllLocal(). func (rpcapi *ClusterRPCAPI) StatusAllLocal(ctx context.Context, in <-chan api.TrackerStatus, out chan<- api.PinInfo) error { filter := <-in return rpcapi.c.StatusAllLocal(ctx, filter, out) } // Status runs Cluster.Status(). func (rpcapi *ClusterRPCAPI) Status(ctx context.Context, in api.Cid, out *api.GlobalPinInfo) error { pinfo, err := rpcapi.c.Status(ctx, in) if err != nil { return err } *out = pinfo return nil } // StatusLocal runs Cluster.StatusLocal(). func (rpcapi *ClusterRPCAPI) StatusLocal(ctx context.Context, in api.Cid, out *api.PinInfo) error { pinfo := rpcapi.c.StatusLocal(ctx, in) *out = pinfo return nil } // RecoverAll runs Cluster.RecoverAll(). func (rpcapi *ClusterRPCAPI) RecoverAll(ctx context.Context, in <-chan struct{}, out chan<- api.GlobalPinInfo) error { return rpcapi.c.RecoverAll(ctx, out) } // RecoverAllLocal runs Cluster.RecoverAllLocal(). func (rpcapi *ClusterRPCAPI) RecoverAllLocal(ctx context.Context, in <-chan struct{}, out chan<- api.PinInfo) error { return rpcapi.c.RecoverAllLocal(ctx, out) } // Recover runs Cluster.Recover(). func (rpcapi *ClusterRPCAPI) Recover(ctx context.Context, in api.Cid, out *api.GlobalPinInfo) error { pinfo, err := rpcapi.c.Recover(ctx, in) if err != nil { return err } *out = pinfo return nil } // RecoverLocal runs Cluster.RecoverLocal(). func (rpcapi *ClusterRPCAPI) RecoverLocal(ctx context.Context, in api.Cid, out *api.PinInfo) error { pinfo, err := rpcapi.c.RecoverLocal(ctx, in) if err != nil { return err } *out = pinfo return nil } // BlockAllocate returns allocations for blocks. This is used in the adders. // It's different from pin allocations when ReplicationFactor < 0. func (rpcapi *ClusterRPCAPI) BlockAllocate(ctx context.Context, in api.Pin, out *[]peer.ID) error { if rpcapi.c.config.FollowerMode { return errFollowerMode } // Allocating for a existing pin. Usually the adder calls this with // cid.Undef. existing, err := rpcapi.c.PinGet(ctx, in.Cid) if err != nil && err != state.ErrNotFound { return err } in, err = rpcapi.c.setupPin(ctx, in, existing) if err != nil { return err } // Return the current peer list. if in.ReplicationFactorMin < 0 { // Returned metrics are Valid and belong to current // Cluster peers. metrics := rpcapi.c.monitor.LatestMetrics(ctx, pingMetricName) peers := make([]peer.ID, len(metrics)) for i, m := range metrics { peers[i] = m.Peer } *out = peers return nil } allocs, err := rpcapi.c.allocate( ctx, in.Cid, existing, in.ReplicationFactorMin, in.ReplicationFactorMax, []peer.ID{}, // blacklist in.UserAllocations, // prio list ) if err != nil { return err } *out = allocs return nil } // RepoGC performs garbage collection sweep on all peers' repos. func (rpcapi *ClusterRPCAPI) RepoGC(ctx context.Context, in struct{}, out *api.GlobalRepoGC) error { res, err := rpcapi.c.RepoGC(ctx) if err != nil { return err } *out = res return nil } // RepoGCLocal performs garbage collection sweep only on the local peer's IPFS daemon. func (rpcapi *ClusterRPCAPI) RepoGCLocal(ctx context.Context, in struct{}, out *api.RepoGC) error { res, err := rpcapi.c.RepoGCLocal(ctx) if err != nil { return err } *out = res return nil } // SendInformerMetrics runs Cluster.sendInformerMetric(). func (rpcapi *ClusterRPCAPI) SendInformerMetrics(ctx context.Context, in struct{}, out *struct{}) error { return rpcapi.c.sendInformersMetrics(ctx) } // SendInformersMetrics runs Cluster.sendInformerMetric() on all informers. func (rpcapi *ClusterRPCAPI) SendInformersMetrics(ctx context.Context, in struct{}, out *struct{}) error { return rpcapi.c.sendInformersMetrics(ctx) } // Alerts runs Cluster.Alerts(). func (rpcapi *ClusterRPCAPI) Alerts(ctx context.Context, in struct{}, out *[]api.Alert) error { alerts := rpcapi.c.Alerts() *out = alerts return nil } // IPFSID returns the current cached IPFS ID for a peer. func (rpcapi *ClusterRPCAPI) IPFSID(ctx context.Context, in peer.ID, out *api.IPFSID) error { if in == "" { in = rpcapi.c.host.ID() } pingVal := pingValueFromMetric(rpcapi.c.monitor.LatestForPeer(ctx, pingMetricName, in)) i := api.IPFSID{ ID: pingVal.IPFSID, Addresses: pingVal.IPFSAddresses, } *out = i return nil } /* Tracker component methods */ // Track runs PinTracker.Track(). func (rpcapi *PinTrackerRPCAPI) Track(ctx context.Context, in api.Pin, out *struct{}) error { ctx, span := trace.StartSpan(ctx, "rpc/tracker/Track") defer span.End() return rpcapi.tracker.Track(ctx, in) } // Untrack runs PinTracker.Untrack(). func (rpcapi *PinTrackerRPCAPI) Untrack(ctx context.Context, in api.Pin, out *struct{}) error { ctx, span := trace.StartSpan(ctx, "rpc/tracker/Untrack") defer span.End() return rpcapi.tracker.Untrack(ctx, in.Cid) } // StatusAll runs PinTracker.StatusAll(). func (rpcapi *PinTrackerRPCAPI) StatusAll(ctx context.Context, in <-chan api.TrackerStatus, out chan<- api.PinInfo) error { ctx, span := trace.StartSpan(ctx, "rpc/tracker/StatusAll") defer span.End() select { case <-ctx.Done(): close(out) return ctx.Err() case filter := <-in: return rpcapi.tracker.StatusAll(ctx, filter, out) } } // Status runs PinTracker.Status(). func (rpcapi *PinTrackerRPCAPI) Status(ctx context.Context, in api.Cid, out *api.PinInfo) error { ctx, span := trace.StartSpan(ctx, "rpc/tracker/Status") defer span.End() pinfo := rpcapi.tracker.Status(ctx, in) *out = pinfo return nil } // RecoverAll runs PinTracker.RecoverAll().f func (rpcapi *PinTrackerRPCAPI) RecoverAll(ctx context.Context, in <-chan struct{}, out chan<- api.PinInfo) error { ctx, span := trace.StartSpan(ctx, "rpc/tracker/RecoverAll") defer span.End() return rpcapi.tracker.RecoverAll(ctx, out) } // Recover runs PinTracker.Recover(). func (rpcapi *PinTrackerRPCAPI) Recover(ctx context.Context, in api.Cid, out *api.PinInfo) error { ctx, span := trace.StartSpan(ctx, "rpc/tracker/Recover") defer span.End() pinfo, err := rpcapi.tracker.Recover(ctx, in) *out = pinfo return err } // PinQueueSize runs PinTracker.PinQueueSize(). func (rpcapi *PinTrackerRPCAPI) PinQueueSize(ctx context.Context, in struct{}, out *int64) error { size, err := rpcapi.tracker.PinQueueSize(ctx) *out = size return err } /* IPFS Connector component methods */ // Pin runs IPFSConnector.Pin(). func (rpcapi *IPFSConnectorRPCAPI) Pin(ctx context.Context, in api.Pin, out *struct{}) error { ctx, span := trace.StartSpan(ctx, "rpc/ipfsconn/IPFSPin") defer span.End() return rpcapi.ipfs.Pin(ctx, in) } // Unpin runs IPFSConnector.Unpin(). func (rpcapi *IPFSConnectorRPCAPI) Unpin(ctx context.Context, in api.Pin, out *struct{}) error { return rpcapi.ipfs.Unpin(ctx, in.Cid) } // PinLsCid runs IPFSConnector.PinLsCid(). func (rpcapi *IPFSConnectorRPCAPI) PinLsCid(ctx context.Context, in api.Pin, out *api.IPFSPinStatus) error { b, err := rpcapi.ipfs.PinLsCid(ctx, in) if err != nil { return err } *out = b return nil } // PinLs runs IPFSConnector.PinLs(). func (rpcapi *IPFSConnectorRPCAPI) PinLs(ctx context.Context, in <-chan []string, out chan<- api.IPFSPinInfo) error { select { case <-ctx.Done(): close(out) return ctx.Err() case pinTypes, ok := <-in: if !ok { close(out) return errors.New("no pinType provided for pin/ls") } return rpcapi.ipfs.PinLs(ctx, pinTypes, out) } } // ConfigKey runs IPFSConnector.ConfigKey(). func (rpcapi *IPFSConnectorRPCAPI) ConfigKey(ctx context.Context, in string, out *interface{}) error { res, err := rpcapi.ipfs.ConfigKey(in) if err != nil { return err } *out = res return nil } // RepoStat runs IPFSConnector.RepoStat(). func (rpcapi *IPFSConnectorRPCAPI) RepoStat(ctx context.Context, in struct{}, out *api.IPFSRepoStat) error { res, err := rpcapi.ipfs.RepoStat(ctx) if err != nil { return err } *out = res return err } // SwarmPeers runs IPFSConnector.SwarmPeers(). func (rpcapi *IPFSConnectorRPCAPI) SwarmPeers(ctx context.Context, in struct{}, out *[]peer.ID) error { res, err := rpcapi.ipfs.SwarmPeers(ctx) if err != nil { return err } *out = res return nil } // BlockStream runs IPFSConnector.BlockStream(). func (rpcapi *IPFSConnectorRPCAPI) BlockStream(ctx context.Context, in <-chan api.NodeWithMeta, out chan<- struct{}) error { defer close(out) // very important to do at the end return rpcapi.ipfs.BlockStream(ctx, in) } // BlockGet runs IPFSConnector.BlockGet(). func (rpcapi *IPFSConnectorRPCAPI) BlockGet(ctx context.Context, in api.Cid, out *[]byte) error { res, err := rpcapi.ipfs.BlockGet(ctx, in) if err != nil { return err } *out = res return nil } // Resolve runs IPFSConnector.Resolve(). func (rpcapi *IPFSConnectorRPCAPI) Resolve(ctx context.Context, in string, out *api.Cid) error { c, err := rpcapi.ipfs.Resolve(ctx, in) if err != nil { return err } *out = c return nil } /* Consensus component methods */ // LogPin runs Consensus.LogPin(). func (rpcapi *ConsensusRPCAPI) LogPin(ctx context.Context, in api.Pin, out *struct{}) error { ctx, span := trace.StartSpan(ctx, "rpc/consensus/LogPin") defer span.End() return rpcapi.cons.LogPin(ctx, in) } // LogUnpin runs Consensus.LogUnpin(). func (rpcapi *ConsensusRPCAPI) LogUnpin(ctx context.Context, in api.Pin, out *struct{}) error { ctx, span := trace.StartSpan(ctx, "rpc/consensus/LogUnpin") defer span.End() return rpcapi.cons.LogUnpin(ctx, in) } // AddPeer runs Consensus.AddPeer(). func (rpcapi *ConsensusRPCAPI) AddPeer(ctx context.Context, in peer.ID, out *struct{}) error { ctx, span := trace.StartSpan(ctx, "rpc/consensus/AddPeer") defer span.End() return rpcapi.cons.AddPeer(ctx, in) } // RmPeer runs Consensus.RmPeer(). func (rpcapi *ConsensusRPCAPI) RmPeer(ctx context.Context, in peer.ID, out *struct{}) error { ctx, span := trace.StartSpan(ctx, "rpc/consensus/RmPeer") defer span.End() return rpcapi.cons.RmPeer(ctx, in) } // Peers runs Consensus.Peers(). func (rpcapi *ConsensusRPCAPI) Peers(ctx context.Context, in struct{}, out *[]peer.ID) error { peers, err := rpcapi.cons.Peers(ctx) if err != nil { return err } *out = peers return nil } /* PeerMonitor */ // LatestMetrics runs PeerMonitor.LatestMetrics(). func (rpcapi *PeerMonitorRPCAPI) LatestMetrics(ctx context.Context, in string, out *[]api.Metric) error { *out = rpcapi.mon.LatestMetrics(ctx, in) return nil } // MetricNames runs PeerMonitor.MetricNames(). func (rpcapi *PeerMonitorRPCAPI) MetricNames(ctx context.Context, in struct{}, out *[]string) error { *out = rpcapi.mon.MetricNames(ctx) return nil }