-
Notifications
You must be signed in to change notification settings - Fork 46
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add s3 lens for working against s3 data (#222)
* Add s3 lens for working against s3 data * factor out common lens code
- Loading branch information
Showing
7 changed files
with
369 additions
and
399 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,16 @@ | ||
| package s3repo | ||
|
|
||
| import ( | ||
| "github.com/filecoin-project/sentinel-visor/lens" | ||
| "github.com/filecoin-project/sentinel-visor/lens/util" | ||
| "github.com/urfave/cli/v2" | ||
| ) | ||
|
|
||
| func NewAPIOpener(c *cli.Context) (lens.APIOpener, lens.APICloser, error) { | ||
| bs, err := NewBlockStore(c.String("repo")) | ||
| if err != nil { | ||
| return nil, nil, err | ||
| } | ||
|
|
||
| return util.NewAPIOpener(c, bs, bs.(*S3Blockstore).getMasterTsKey) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,119 @@ | ||
| package s3repo | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "io/ioutil" | ||
| "net/http" | ||
| "strings" | ||
|
|
||
| logging "github.com/ipfs/go-log/v2" | ||
|
|
||
| "github.com/filecoin-project/lotus/chain/types" | ||
| "github.com/filecoin-project/lotus/lib/blockstore" | ||
| blocks "github.com/ipfs/go-block-format" | ||
| "github.com/ipfs/go-cid" | ||
| ) | ||
|
|
||
| var log = logging.Logger("sql") | ||
|
|
||
| type S3Blockstore struct { | ||
| prefix string | ||
| client http.Client | ||
| } | ||
|
|
||
| func NewBlockStore(connstr string) (blockstore.Blockstore, error) { | ||
|
|
||
| sbs := &S3Blockstore{ | ||
| prefix: connstr, | ||
| client: http.Client{}, | ||
| } | ||
|
|
||
| // we do not currently use the Identity codec, but just in case... | ||
| return blockstore.WrapIDStore(sbs), nil | ||
| } | ||
|
|
||
| func (sbs *S3Blockstore) Has(c cid.Cid) (has bool, err error) { | ||
| resp, err := sbs.client.Head(sbs.prefix + c.String() + "/data.raw") | ||
| if err != nil { | ||
| return false, err | ||
| } | ||
| return resp.StatusCode == 200, nil | ||
| } | ||
|
|
||
| func (sbs *S3Blockstore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) { | ||
| return nil, fmt.Errorf("Not implemented") | ||
| } | ||
|
|
||
| func (sbs *S3Blockstore) Get(c cid.Cid) (blocks.Block, error) { | ||
| resp, err := sbs.client.Get(sbs.prefix + c.String() + "/data.raw") | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| if resp.StatusCode != http.StatusOK { | ||
| return nil, fmt.Errorf("Failed to fetch: %v", resp.StatusCode) | ||
| } | ||
| defer resp.Body.Close() | ||
| buf, err := ioutil.ReadAll(resp.Body) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return blocks.NewBlockWithCid(buf, c) | ||
| } | ||
|
|
||
| func (sbs *S3Blockstore) GetSize(c cid.Cid) (size int, err error) { | ||
| resp, err := sbs.client.Head(sbs.prefix + c.String() + "/data.raw") | ||
| if err != nil { | ||
| return -1, err | ||
| } | ||
| if resp.StatusCode == 200 { | ||
| return int(resp.ContentLength), nil | ||
| } | ||
| return -1, fmt.Errorf("does not exist") | ||
| } | ||
|
|
||
| func (sbs *S3Blockstore) getMasterTsKey(ctx context.Context, lookback int) (*types.TipSetKey, error) { | ||
| resp, err := sbs.client.Get(sbs.prefix + "/head") | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| defer resp.Body.Close() | ||
| buf, err := ioutil.ReadAll(resp.Body) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| cidStrs := strings.Split(string(buf), " ") | ||
| cids := make([]cid.Cid, len(cidStrs)) | ||
| for _, cs := range cidStrs { | ||
| c, err := cid.Parse(cs) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| cids = append(cids, c) | ||
| } | ||
|
|
||
| tk := types.NewTipSetKey(cids...) | ||
| return &tk, nil | ||
| } | ||
|
|
||
| // BEGIN UNIMPLEMENTED | ||
|
|
||
| // HashOnRead specifies if every read block should be | ||
| // rehashed to make sure it matches its CID. | ||
| func (sbs *S3Blockstore) HashOnRead(enabled bool) { | ||
| log.Warn("HashOnRead toggle not implemented, ignoring") | ||
| } | ||
|
|
||
| // Put puts a given block to the underlying datastore | ||
| func (sbs *S3Blockstore) Put(b blocks.Block) (err error) { | ||
| return fmt.Errorf("Not Implemented") | ||
| } | ||
|
|
||
| func (sbs *S3Blockstore) PutMany(blks []blocks.Block) error { | ||
| return fmt.Errorf("Not Implemented") | ||
| } | ||
|
|
||
| func (sbs *S3Blockstore) DeleteBlock(cid.Cid) error { | ||
| return fmt.Errorf("Not Implemented") | ||
| } |
Oops, something went wrong.