Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat: use lassie for retrievals #149

Merged
merged 4 commits into from
Dec 2, 2022
Merged

feat: use lassie for retrievals #149

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8146fd1
feat: use lassie for retrievals
rvagg Nov 25, 2022
d447ae7
fix: create new event types for lassie for proper matching
rvagg Nov 25, 2022
fac7000
fixup! feat: use lassie for retrievals
rvagg Dec 2, 2022
e63124a
fix: remove use of deprecated go-libp2p-core
rvagg Dec 2, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
85 changes: 74 additions & 11 deletions autoretrieve.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,23 +19,25 @@ import (
"github.com/application-research/autoretrieve/bitswap"
"github.com/application-research/autoretrieve/blocks"
"github.com/application-research/autoretrieve/endpoint"
"github.com/application-research/autoretrieve/filecoin"
"github.com/application-research/autoretrieve/filecoin/eventrecorder"
"github.com/application-research/filclient"
"github.com/application-research/filclient/keystore"
"github.com/application-research/filclient/rep"
"github.com/filecoin-project/go-address"
datatransfer "github.com/filecoin-project/go-data-transfer"
"github.com/filecoin-project/go-fil-markets/retrievalmarket"
lassieeventrecorder "github.com/filecoin-project/lassie/pkg/eventrecorder"
lassieretriever "github.com/filecoin-project/lassie/pkg/retriever"
"github.com/filecoin-project/lotus/chain/wallet"
lcli "github.com/filecoin-project/lotus/cli"
"github.com/ipfs/go-cid"
flatfs "github.com/ipfs/go-ds-flatfs"
leveldb "github.com/ipfs/go-ds-leveldb"
blockstore "github.com/ipfs/go-ipfs-blockstore"
"github.com/libp2p/go-libp2p"
"github.com/libp2p/go-libp2p-core/crypto"
"github.com/libp2p/go-libp2p-core/host"
"github.com/libp2p/go-libp2p-core/network"
peer "github.com/libp2p/go-libp2p-core/peer"
"github.com/libp2p/go-libp2p/core/crypto"
"github.com/libp2p/go-libp2p/core/host"
"github.com/libp2p/go-libp2p/core/network"
peer "github.com/libp2p/go-libp2p/core/peer"
"github.com/multiformats/go-multiaddr"
"github.com/urfave/cli/v2"
)
Expand All @@ -54,7 +56,7 @@ number of outbound streams: %d,

type Autoretrieve struct {
host host.Host
retriever *filecoin.Retriever
retriever *lassieretriever.Retriever
provider *bitswap.Provider
apiCloser func()
}
Expand Down Expand Up @@ -156,9 +158,9 @@ func New(cctx *cli.Context, dataDir string, cfg Config) (*Autoretrieve, error) {
logger.Errorf("FilClient initialization failed: %v", err)
}
// Initialize Filecoin retriever
var retriever *filecoin.Retriever
var retriever *lassieretriever.Retriever
if !cfg.DisableRetrieval {
var ep filecoin.Endpoint
var ep lassieretriever.Endpoint
switch cfg.LookupEndpointType {
case EndpointTypeEstuary:
logger.Infof("Using Estuary endpoint type")
Expand All @@ -179,7 +181,7 @@ func New(cctx *cli.Context, dataDir string, cfg Config) (*Autoretrieve, error) {
return blockManager.Has(cctx.Context, c)
}

retriever, err = filecoin.NewRetriever(cctx.Context, retrieverCfg, fc, ep, confirmer)
retriever, err = lassieretriever.NewRetriever(cctx.Context, retrieverCfg, &filclientLassieWrapper{fc}, ep, confirmer)
if err != nil {
return nil, err
}
Expand All @@ -189,7 +191,7 @@ func New(cctx *cli.Context, dataDir string, cfg Config) (*Autoretrieve, error) {
if err != nil {
return nil, err
}
retriever.RegisterListener(eventrecorder.NewEventRecorder(cctx.Context, cfg.InstanceId, cfg.EventRecorderEndpointURL, eventRecorderEndpointAuthorization))
retriever.RegisterListener(lassieeventrecorder.NewEventRecorder(cctx.Context, cfg.InstanceId, cfg.EventRecorderEndpointURL, eventRecorderEndpointAuthorization))
}
if cfg.LogRetrievals {
w := tabwriter.NewWriter(os.Stdout, 5, 0, 3, ' ', 0)
Expand Down Expand Up @@ -423,3 +425,64 @@ func initHost(ctx context.Context, dataDir string, resourceManagerStats bool, li
}
return host, nil
}

// ---- transitional wrappers to bridge filclient into something usable by lassie in its current state ----- //

var _ rep.RetrievalSubscriber = &filclientLassieSubscriberWrapper{}
var _ lassieretriever.FilClient = &filclientLassieWrapper{}

type filclientLassieSubscriberWrapper struct {
sub lassieretriever.RetrievalSubscriber
}

func (flsw *filclientLassieSubscriberWrapper) OnRetrievalEvent(event rep.RetrievalEvent) {
switch ret := event.(type) {
case rep.RetrievalEventConnect:
flsw.sub.OnRetrievalEvent(lassieretriever.NewRetrievalEventConnect(lassieretriever.Phase(ret.Phase()), ret.PayloadCid(), ret.StorageProviderId(), ret.StorageProviderAddr()))
case rep.RetrievalEventQueryAsk:
flsw.sub.OnRetrievalEvent(lassieretriever.NewRetrievalEventQueryAsk(lassieretriever.Phase(ret.Phase()), ret.PayloadCid(), ret.StorageProviderId(), ret.StorageProviderAddr(), ret.QueryResponse()))
case rep.RetrievalEventProposed:
flsw.sub.OnRetrievalEvent(lassieretriever.NewRetrievalEventProposed(lassieretriever.Phase(ret.Phase()), ret.PayloadCid(), ret.StorageProviderId(), ret.StorageProviderAddr()))
case rep.RetrievalEventAccepted:
flsw.sub.OnRetrievalEvent(lassieretriever.NewRetrievalEventAccepted(lassieretriever.Phase(ret.Phase()), ret.PayloadCid(), ret.StorageProviderId(), ret.StorageProviderAddr()))
case rep.RetrievalEventFirstByte:
flsw.sub.OnRetrievalEvent(lassieretriever.NewRetrievalEventFirstByte(lassieretriever.Phase(ret.Phase()), ret.PayloadCid(), ret.StorageProviderId(), ret.StorageProviderAddr()))
case rep.RetrievalEventFailure:
flsw.sub.OnRetrievalEvent(lassieretriever.NewRetrievalEventFailure(lassieretriever.Phase(ret.Phase()), ret.PayloadCid(), ret.StorageProviderId(), ret.StorageProviderAddr(), ret.ErrorMessage()))
case rep.RetrievalEventSuccess:
flsw.sub.OnRetrievalEvent(lassieretriever.NewRetrievalEventSuccess(lassieretriever.Phase(ret.Phase()), ret.PayloadCid(), ret.StorageProviderId(), ret.StorageProviderAddr(), ret.ReceivedSize(), ret.ReceivedCids(), ret.Duration(), ret.TotalPayment()))
default:
logger.Errorf("unexpected retrieval event type %T: %v", ret, ret)
}
}

type filclientLassieWrapper struct {
fc *filclient.FilClient
}

func (flw *filclientLassieWrapper) RetrievalQueryToPeer(ctx context.Context, minerPeer peer.AddrInfo, pcid cid.Cid) (*retrievalmarket.QueryResponse, error) {
return flw.fc.RetrievalQueryToPeer(ctx, minerPeer, pcid)
}

func (flw *filclientLassieWrapper) RetrieveContentFromPeerAsync(
ctx context.Context,
peerID peer.ID,
minerWallet address.Address,
proposal *retrievalmarket.DealProposal,
) (<-chan lassieretriever.RetrievalResult, <-chan uint64, func()) {
lassieResultChan := make(chan lassieretriever.RetrievalResult, 1)

resultChan, onProgressChan, gracefulShutdownCb := flw.fc.RetrieveContentFromPeerAsync(ctx, peerID, minerWallet, proposal)
go func() {
result := <-resultChan
lassieResultChan <- lassieretriever.RetrievalResult{
RetrievalStats: (*lassieretriever.RetrievalStats)(result.RetrievalStats),
Err: result.Err,
}
}()
return lassieResultChan, onProgressChan, gracefulShutdownCb
}

func (flw *filclientLassieWrapper) SubscribeToRetrievalEvents(subscriber lassieretriever.RetrievalSubscriber) {
flw.fc.SubscribeToRetrievalEvents(&filclientLassieSubscriberWrapper{subscriber})
}
18 changes: 9 additions & 9 deletions bitswap/provider.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,8 @@ import (
"time"

"github.com/application-research/autoretrieve/blocks"
"github.com/application-research/autoretrieve/filecoin"
"github.com/application-research/autoretrieve/metrics"
lassieretriever "github.com/filecoin-project/lassie/pkg/retriever"
"github.com/ipfs/go-bitswap/message"
bitswap_message_pb "github.com/ipfs/go-bitswap/message/pb"
"github.com/ipfs/go-bitswap/network"
Expand All @@ -17,11 +17,11 @@ import (
"github.com/ipfs/go-log/v2"
"github.com/ipfs/go-peertaskqueue"
"github.com/ipfs/go-peertaskqueue/peertask"
"github.com/libp2p/go-libp2p-core/host"
"github.com/libp2p/go-libp2p-core/peer"
"github.com/libp2p/go-libp2p-core/routing"
dht "github.com/libp2p/go-libp2p-kad-dht"
"github.com/libp2p/go-libp2p-kad-dht/fullrt"
"github.com/libp2p/go-libp2p/core/host"
"github.com/libp2p/go-libp2p/core/peer"
"github.com/libp2p/go-libp2p/core/routing"
"go.opencensus.io/stats"
"go.opencensus.io/tag"
)
Expand Down Expand Up @@ -53,7 +53,7 @@ type Provider struct {
config ProviderConfig
network network.BitSwapNetwork
blockManager *blocks.Manager
retriever *filecoin.Retriever
retriever *lassieretriever.Retriever
taskQueue *peertaskqueue.PeerTaskQueue
workReady chan struct{}
}
Expand All @@ -64,7 +64,7 @@ func NewProvider(
host host.Host,
datastore datastore.Batching,
blockManager *blocks.Manager,
retriever *filecoin.Retriever,
retriever *lassieretriever.Retriever,
) (*Provider, error) {

var routing routing.ContentRouting
Expand Down Expand Up @@ -102,7 +102,7 @@ func NewProvider(
workReady: make(chan struct{}, config.MaxBitswapWorkers),
}

provider.network.SetDelegate(provider)
provider.network.Start(provider)

for i := uint(0); i < config.MaxBitswapWorkers; i++ {
go provider.runWorker()
Expand Down Expand Up @@ -195,7 +195,7 @@ func (provider *Provider) ReceiveMessage(ctx context.Context, sender peer.ID, in
// queue DONT_HAVE and move on
provider.queueDontHave(ctx, sender, entry, "failed_retriever_request")

if !errors.Is(err, filecoin.ErrNoCandidates) {
if !errors.Is(err, lassieretriever.ErrNoCandidates) {
logger.Warnf("Could not get candidates: %s", err.Error())
}

Expand All @@ -215,7 +215,7 @@ func (provider *Provider) ReceiveMessage(ctx context.Context, sender peer.ID, in
// queue DONT_HAVE and move on
provider.queueDontHave(ctx, sender, entry, "failed_retriever_request")

if !errors.Is(err, filecoin.ErrNoCandidates) {
if !errors.Is(err, lassieretriever.ErrNoCandidates) {
logger.Warnf("Could not get candidates: %s", err.Error())
}

Expand Down
20 changes: 10 additions & 10 deletions config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,12 +8,12 @@ import (
"time"

"github.com/application-research/autoretrieve/bitswap"
"github.com/application-research/autoretrieve/filecoin"
"github.com/application-research/filclient"
"github.com/dustin/go-humanize"
"github.com/filecoin-project/go-address"
lassieretriever "github.com/filecoin-project/lassie/pkg/retriever"
"github.com/ipfs/go-cid"
peer "github.com/libp2p/go-libp2p-core/peer"
peer "github.com/libp2p/go-libp2p/core/peer"
"gopkg.in/yaml.v3"
)

Expand Down Expand Up @@ -184,31 +184,31 @@ type Config struct {
}

// Extract relevant config points into a filecoin retriever config
func (cfg *Config) ExtractFilecoinRetrieverConfig(ctx context.Context, fc *filclient.FilClient) (filecoin.RetrieverConfig, error) {
convertedMinerCfgs := make(map[peer.ID]filecoin.MinerConfig)
func (cfg *Config) ExtractFilecoinRetrieverConfig(ctx context.Context, fc *filclient.FilClient) (lassieretriever.RetrieverConfig, error) {
convertedMinerCfgs := make(map[peer.ID]lassieretriever.MinerConfig)
for provider, minerCfg := range cfg.MinerConfigs {
peer, err := provider.GetPeerID(ctx, fc)
if err != nil {
return filecoin.RetrieverConfig{}, err
return lassieretriever.RetrieverConfig{}, err
}

convertedMinerCfgs[peer] = filecoin.MinerConfig(minerCfg)
convertedMinerCfgs[peer] = lassieretriever.MinerConfig(minerCfg)
}

blacklist, err := configStorageProviderListToPeerMap(ctx, fc, cfg.MinerBlacklist)
if err != nil {
return filecoin.RetrieverConfig{}, err
return lassieretriever.RetrieverConfig{}, err
}

whitelist, err := configStorageProviderListToPeerMap(ctx, fc, cfg.MinerWhitelist)
if err != nil {
return filecoin.RetrieverConfig{}, err
return lassieretriever.RetrieverConfig{}, err
}

return filecoin.RetrieverConfig{
return lassieretriever.RetrieverConfig{
MinerBlacklist: blacklist,
MinerWhitelist: whitelist,
DefaultMinerConfig: filecoin.MinerConfig(cfg.DefaultMinerConfig),
DefaultMinerConfig: lassieretriever.MinerConfig(cfg.DefaultMinerConfig),
MinerConfigs: convertedMinerCfgs,
PaidRetrievals: cfg.PaidRetrievals,
}, nil
Expand Down
8 changes: 4 additions & 4 deletions endpoint/estuary.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,9 +9,9 @@ import (
"net/url"
"path"

"github.com/application-research/autoretrieve/filecoin"
"github.com/application-research/filclient"
"github.com/filecoin-project/go-address"
lassieretriever "github.com/filecoin-project/lassie/pkg/retriever"
"github.com/ipfs/go-cid"
)

Expand Down Expand Up @@ -39,7 +39,7 @@ func NewEstuaryEndpoint(fc *filclient.FilClient, url string) *EstuaryEndpoint {
}
}

func (ee *EstuaryEndpoint) FindCandidates(ctx context.Context, cid cid.Cid) ([]filecoin.RetrievalCandidate, error) {
func (ee *EstuaryEndpoint) FindCandidates(ctx context.Context, cid cid.Cid) ([]lassieretriever.RetrievalCandidate, error) {
// Create URL with CID
endpointURL, err := url.Parse(ee.url)
if err != nil {
Expand All @@ -63,13 +63,13 @@ func (ee *EstuaryEndpoint) FindCandidates(ctx context.Context, cid cid.Cid) ([]f
return nil, ErrEndpointBodyInvalid
}

converted := make([]filecoin.RetrievalCandidate, 0, len(unfiltered))
converted := make([]lassieretriever.RetrievalCandidate, 0, len(unfiltered))
for _, original := range unfiltered {
minerPeer, err := ee.fc.MinerPeer(ctx, original.Miner)
if err != nil {
return nil, fmt.Errorf("%w: failed to get miner peer: %v", ErrEndpointRequestFailed, err)
}
converted = append(converted, filecoin.RetrievalCandidate{
converted = append(converted, lassieretriever.RetrievalCandidate{
MinerPeer: minerPeer,
RootCid: original.RootCid,
})
Expand Down
8 changes: 4 additions & 4 deletions endpoint/indexer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,8 @@ import (
"net/http"
"time"

"github.com/application-research/autoretrieve/filecoin"
"github.com/filecoin-project/index-provider/metadata"
lassieretriever "github.com/filecoin-project/lassie/pkg/retriever"
"github.com/filecoin-project/storetheindex/api/v0/finder/model"
"github.com/ipfs/go-cid"
)
Expand Down Expand Up @@ -51,7 +51,7 @@ func (idxf *IndexerEndpoint) sendRequest(req *http.Request) (*model.FindResponse
return model.UnmarshalFindResponse(b)
}

func (idxf *IndexerEndpoint) FindCandidates(ctx context.Context, cid cid.Cid) ([]filecoin.RetrievalCandidate, error) {
func (idxf *IndexerEndpoint) FindCandidates(ctx context.Context, cid cid.Cid) ([]lassieretriever.RetrievalCandidate, error) {
u := fmt.Sprint(idxf.baseUrl, "/multihash/", cid.Hash().B58String())
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
if err != nil {
Expand All @@ -64,7 +64,7 @@ func (idxf *IndexerEndpoint) FindCandidates(ctx context.Context, cid cid.Cid) ([
}
hash := string(cid.Hash())
// turn parsedResp into records.
var matches []filecoin.RetrievalCandidate
var matches []lassieretriever.RetrievalCandidate

indices := rand.Perm(len(parsedResp.MultihashResults))
for _, i := range indices {
Expand All @@ -80,7 +80,7 @@ func (idxf *IndexerEndpoint) FindCandidates(ctx context.Context, cid cid.Cid) ([
continue
}

matches = append(matches, filecoin.RetrievalCandidate{
matches = append(matches, lassieretriever.RetrievalCandidate{
RootCid: cid,
MinerPeer: val.Provider,
})
Expand Down
Loading