334 lines
8.7 KiB
Go
334 lines
8.7 KiB
Go
// Copyright (C) 2019 Storj Labs, Inc.
|
|
// See LICENSE for copying information
|
|
|
|
package testplanet
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"io/ioutil"
|
|
"strconv"
|
|
"time"
|
|
|
|
"github.com/zeebo/errs"
|
|
"go.uber.org/zap"
|
|
|
|
"storj.io/common/identity"
|
|
"storj.io/common/macaroon"
|
|
"storj.io/common/peertls/tlsopts"
|
|
"storj.io/common/rpc"
|
|
"storj.io/common/storj"
|
|
"storj.io/common/uuid"
|
|
"storj.io/uplink"
|
|
"storj.io/uplink/private/metainfo"
|
|
"storj.io/uplink/private/piecestore"
|
|
"storj.io/uplink/private/testuplink"
|
|
)
|
|
|
|
// Uplink is a registered user on all satellites,
|
|
// which contains the necessary accesses and project info.
|
|
type Uplink struct {
|
|
Log *zap.Logger
|
|
Identity *identity.FullIdentity
|
|
Dialer rpc.Dialer
|
|
|
|
APIKey map[storj.NodeID]*macaroon.APIKey
|
|
Access map[storj.NodeID]*uplink.Access
|
|
|
|
// Projects is indexed by the satellite number.
|
|
Projects []*Project
|
|
}
|
|
|
|
// Project contains all necessary information about a user.
|
|
type Project struct {
|
|
client *Uplink
|
|
|
|
ID uuid.UUID
|
|
Owner ProjectOwner
|
|
|
|
Satellite Peer
|
|
APIKey string
|
|
|
|
RawAPIKey *macaroon.APIKey
|
|
}
|
|
|
|
// ProjectOwner contains information about the project owner.
|
|
type ProjectOwner struct {
|
|
ID uuid.UUID
|
|
Email string
|
|
}
|
|
|
|
// DialMetainfo dials the satellite with the appropriate api key.
|
|
func (project *Project) DialMetainfo(ctx context.Context) (*metainfo.Client, error) {
|
|
return project.client.DialMetainfo(ctx, project.Satellite, project.RawAPIKey)
|
|
}
|
|
|
|
// newUplinks creates initializes uplinks, requires peer to have at least one satellite.
|
|
func (planet *Planet) newUplinks(prefix string, count int) ([]*Uplink, error) {
|
|
var xs []*Uplink
|
|
for i := 0; i < count; i++ {
|
|
uplink, err := planet.newUplink(prefix + strconv.Itoa(i))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
xs = append(xs, uplink)
|
|
}
|
|
|
|
return xs, nil
|
|
}
|
|
|
|
// newUplink creates a new uplink.
|
|
func (planet *Planet) newUplink(name string) (*Uplink, error) {
|
|
ctx := context.TODO()
|
|
|
|
identity, err := planet.NewIdentity()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
tlsOptions, err := tlsopts.NewOptions(identity, tlsopts.Config{
|
|
PeerIDVersions: strconv.Itoa(int(planet.config.IdentityVersion.Number)),
|
|
}, nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
planetUplink := &Uplink{
|
|
Log: planet.log.Named(name),
|
|
Identity: identity,
|
|
APIKey: map[storj.NodeID]*macaroon.APIKey{},
|
|
Access: map[storj.NodeID]*uplink.Access{},
|
|
}
|
|
|
|
planetUplink.Log.Debug("id=" + identity.ID.String())
|
|
|
|
planetUplink.Dialer = rpc.NewDefaultDialer(tlsOptions)
|
|
|
|
for j, satellite := range planet.Satellites {
|
|
console := satellite.API.Console
|
|
|
|
projectName := fmt.Sprintf("%s_%d", name, j)
|
|
user, err := satellite.AddUser(
|
|
ctx,
|
|
fmt.Sprintf("User %s", projectName),
|
|
fmt.Sprintf("user@%s.test", projectName),
|
|
10,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
project, err := satellite.AddProject(ctx, user.ID, projectName)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
authCtx, err := satellite.authenticatedContext(ctx, user.ID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
_, apiKey, err := console.Service.CreateAPIKey(authCtx, project.ID, "root")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
planetUplink.APIKey[satellite.ID()] = apiKey
|
|
|
|
planetUplink.Projects = append(planetUplink.Projects, &Project{
|
|
client: planetUplink,
|
|
|
|
ID: project.ID,
|
|
Owner: ProjectOwner{
|
|
ID: user.ID,
|
|
Email: user.Email,
|
|
},
|
|
|
|
Satellite: satellite,
|
|
APIKey: apiKey.Serialize(),
|
|
|
|
RawAPIKey: apiKey,
|
|
})
|
|
}
|
|
|
|
planet.Uplinks = append(planet.Uplinks, planetUplink)
|
|
|
|
return planetUplink, nil
|
|
}
|
|
|
|
// ID returns uplink id.
|
|
func (client *Uplink) ID() storj.NodeID { return client.Identity.ID }
|
|
|
|
// Addr returns uplink address.
|
|
func (client *Uplink) Addr() string { return "" }
|
|
|
|
// Shutdown shuts down all uplink dependencies.
|
|
func (client *Uplink) Shutdown() error { return nil }
|
|
|
|
// DialMetainfo dials destination with apikey and returns metainfo Client.
|
|
func (client *Uplink) DialMetainfo(ctx context.Context, destination Peer, apikey *macaroon.APIKey) (*metainfo.Client, error) {
|
|
return metainfo.DialNodeURL(ctx, client.Dialer, destination.NodeURL().String(), apikey, "Test/1.0")
|
|
}
|
|
|
|
// DialPiecestore dials destination storagenode and returns a piecestore client.
|
|
func (client *Uplink) DialPiecestore(ctx context.Context, destination Peer) (*piecestore.Client, error) {
|
|
return piecestore.DialNodeURL(ctx, client.Dialer, destination.NodeURL(), client.Log.Named("uplink>piecestore"), piecestore.DefaultConfig)
|
|
}
|
|
|
|
// Upload data to specific satellite.
|
|
func (client *Uplink) Upload(ctx context.Context, satellite *Satellite, bucket string, path storj.Path, data []byte) error {
|
|
return client.UploadWithExpiration(ctx, satellite, bucket, path, data, time.Time{})
|
|
}
|
|
|
|
// UploadWithExpiration data to specific satellite and expiration time.
|
|
func (client *Uplink) UploadWithExpiration(ctx context.Context, satellite *Satellite, bucketName string, path storj.Path, data []byte, expiration time.Time) error {
|
|
_, found := testuplink.GetMaxSegmentSize(ctx)
|
|
if !found {
|
|
ctx = testuplink.WithMaxSegmentSize(ctx, satellite.Config.Metainfo.MaxSegmentSize)
|
|
}
|
|
|
|
project, err := client.GetProject(ctx, satellite)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { err = errs.Combine(err, project.Close()) }()
|
|
|
|
_, err = project.EnsureBucket(ctx, bucketName)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
upload, err := project.UploadObject(ctx, bucketName, path, &uplink.UploadOptions{
|
|
Expires: expiration,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
_, err = io.Copy(upload, bytes.NewReader(data))
|
|
if err != nil {
|
|
abortErr := upload.Abort()
|
|
err = errs.Combine(err, abortErr)
|
|
return err
|
|
}
|
|
|
|
return upload.Commit()
|
|
}
|
|
|
|
// Download data from specific satellite.
|
|
func (client *Uplink) Download(ctx context.Context, satellite *Satellite, bucketName string, path storj.Path) ([]byte, error) {
|
|
project, err := client.GetProject(ctx, satellite)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { err = errs.Combine(err, project.Close()) }()
|
|
|
|
download, err := project.DownloadObject(ctx, bucketName, path, nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { err = errs.Combine(err, download.Close()) }()
|
|
|
|
data, err := ioutil.ReadAll(download)
|
|
if err != nil {
|
|
return []byte{}, err
|
|
}
|
|
return data, nil
|
|
}
|
|
|
|
// DownloadStream returns stream for downloading data.
|
|
func (client *Uplink) DownloadStream(ctx context.Context, satellite *Satellite, bucketName string, path storj.Path) (_ io.ReadCloser, cleanup func() error, err error) {
|
|
project, err := client.GetProject(ctx, satellite)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
|
|
cleanup = func() error {
|
|
err = errs.Combine(err,
|
|
project.Close(),
|
|
)
|
|
return err
|
|
}
|
|
|
|
downloader, err := project.DownloadObject(ctx, bucketName, path, nil)
|
|
return downloader, cleanup, err
|
|
}
|
|
|
|
// DownloadStreamRange returns stream for downloading data.
|
|
func (client *Uplink) DownloadStreamRange(ctx context.Context, satellite *Satellite, bucketName string, path storj.Path, start, limit int64) (_ io.ReadCloser, cleanup func() error, err error) {
|
|
project, err := client.GetProject(ctx, satellite)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
|
|
cleanup = func() error {
|
|
err = errs.Combine(err,
|
|
project.Close(),
|
|
)
|
|
return err
|
|
}
|
|
|
|
downloader, err := project.DownloadObject(ctx, bucketName, path, &uplink.DownloadOptions{
|
|
Offset: start,
|
|
Length: limit,
|
|
})
|
|
return downloader, cleanup, err
|
|
}
|
|
|
|
// DeleteObject deletes an object at the path in a bucket.
|
|
func (client *Uplink) DeleteObject(ctx context.Context, satellite *Satellite, bucketName string, path storj.Path) error {
|
|
project, err := client.GetProject(ctx, satellite)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { err = errs.Combine(err, project.Close()) }()
|
|
|
|
_, err = project.DeleteObject(ctx, bucketName, path)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return err
|
|
}
|
|
|
|
// CreateBucket creates a new bucket.
|
|
func (client *Uplink) CreateBucket(ctx context.Context, satellite *Satellite, bucketName string) error {
|
|
project, err := client.GetProject(ctx, satellite)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { err = errs.Combine(err, project.Close()) }()
|
|
|
|
_, err = project.CreateBucket(ctx, bucketName)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// DeleteBucket deletes a bucket.
|
|
func (client *Uplink) DeleteBucket(ctx context.Context, satellite *Satellite, bucketName string) error {
|
|
project, err := client.GetProject(ctx, satellite)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { err = errs.Combine(err, project.Close()) }()
|
|
|
|
_, err = project.DeleteBucket(ctx, bucketName)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// GetProject returns a uplink.Project which allows interactions with a specific project.
|
|
func (client *Uplink) GetProject(ctx context.Context, satellite *Satellite) (*uplink.Project, error) {
|
|
access := client.Access[satellite.ID()]
|
|
|
|
project, err := uplink.OpenProject(ctx, access)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return project, nil
|
|
}
|