Skip to content

Commit

Permalink
#461: experimenting with a way of doing prefixes which works for pull…
Browse files Browse the repository at this point in the history 
…ing, too
  • Loading branch information
Charlotte committed Jul 13, 2018
1 parent 8a0623e commit e0f9751
Show file tree
Hide file tree
Showing 7 changed files with 172 additions and 163 deletions.
1 change: 1 addition & 0 deletions cmd/dm/pkg/commands/clone.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ Online help: https://docs.dotmesh.com/references/cli/#clone-dm-clone-local-name-
"pull", peer,
cloneLocalVolume, branchName,
filesystemName, branchName,
nil,
// TODO also switch to the remote?
)
if err != nil {
Expand Down
1 change: 1 addition & 0 deletions cmd/dm/pkg/commands/pull.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@ Online help: https://docs.dotmesh.com/references/cli/#pull-dm-pull-remote-dot-br
"pull", peer,
filesystemName, branchName,
pullRemoteVolume, branchName,
nil,
)
if err != nil {
return err
Expand Down
2 changes: 1 addition & 1 deletion cmd/dm/pkg/commands/push.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ Online help: https://docs.dotmesh.com/references/cli/#push-dm-push-remote-remote
return err
}
transferId, err := dm.RequestTransfer(
"push", peer, filesystemName, branchName, pushRemoteVolume, "",
"push", peer, filesystemName, branchName, pushRemoteVolume, "", nil,
)
if err != nil {
return err
Expand Down
4 changes: 2 additions & 2 deletions cmd/dm/pkg/commands/s3.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ func NewCmdS3(out io.Writer) *cobra.Command {
cmd.AddCommand(&cobra.Command{
Use: "remote add <remote-name> <key-id:secret-key>[@endpoint]",
Short: "Add an S3 remote",
Long: "Online help: https://docs.dotmesh.com/references/cli/#add-a-new-S3-remote-dm-s3-remote-add-access-key-secret-key-host-port",
Long: "Online help: https://docs.dotmesh.com/references/cli/#add-a-new-s3-remote-dm-s3-remote-add-access-key-secret-key-host-port",

Run: func(cmd *cobra.Command, args []string) {
runHandlingError(func() error {
Expand Down Expand Up @@ -96,7 +96,7 @@ func NewCmdS3(out io.Writer) *cobra.Command {
if err != nil {
return err
}
transferId, err := dm.RequestS3SubsetTransfer(
transferId, err := dm.RequestTransfer(
"pull", peer,
localVolumeName, branchName,
filesystemName, branchName,
Expand Down
210 changes: 82 additions & 128 deletions cmd/dm/pkg/remotes/api.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -906,6 +906,7 @@ func (dm *DotmeshAPI) RequestTransfer(
direction, peer,
localFilesystemName, localBranchName,
remoteFilesystemName, remoteBranchName string,
prefixes []string,
) (string, error) {
connectionInitiator := dm.Configuration.CurrentRemote

Expand All @@ -927,110 +928,20 @@ func (dm *DotmeshAPI) RequestTransfer(

// Let's replace any missing things with defaults.
// The defaults depend on whether we're pushing or pulling.
localNamespace, localVolume, localBranchName, remoteNamespace, remoteVolume, remoteBranchName, err := dm.parseVolumeDetails(remote, peer, direction, localFilesystemName, localBranchName, remoteFilesystemName, remoteBranchName)
if err != nil {
return "", err
}
if direction == "push" {
fmt.Printf("Pushing %s/%s to %s:%s/%s\n",
localNamespace, localVolume,
peer,
remoteNamespace, remoteVolume,
)
} else {
fmt.Printf("Pulling %s/%s from %s:%s/%s\n",
localNamespace, localVolume,
peer,
remoteNamespace, remoteVolume,
)
}

// connect to connectionInitiator
client, err := dm.Configuration.ClusterFromRemote(connectionInitiator, dm.verbose)
if err != nil {
return "", err
}
var transferId string
// TODO make ApiKey time- and domain- (filesystem?) limited
// cryptographically somehow
dmRemote, ok := remote.(*DMRemote)

if ok {
transferRequest := TransferRequest{
Peer: dmRemote.Hostname,
User: dmRemote.User,
Port: dmRemote.Port,
ApiKey: dmRemote.ApiKey,
Direction: direction,
LocalNamespace: localNamespace,
LocalName: localVolume,
LocalBranchName: deMasterify(localBranchName),
RemoteNamespace: remoteNamespace,
RemoteName: remoteVolume,
RemoteBranchName: deMasterify(remoteBranchName),
// TODO add TargetSnapshot here, to support specifying "push to a given
// snapshot" rather than just "push all snapshots up to the latest"
}

if debugMode {
fmt.Printf("[DEBUG] TransferRequest: %#v\n", transferRequest)
}

err = client.CallRemote(context.Background(),
"DotmeshRPC.Transfer", transferRequest, &transferId)
if err != nil {
return "", err
}
} else {
s3Remote, ok := remote.(*S3Remote)
if ok {

transferRequest := S3TransferRequest{
KeyID: s3Remote.KeyID,
SecretKey: s3Remote.SecretKey,
Endpoint: s3Remote.Endpoint,
Direction: direction,
LocalNamespace: localNamespace,
LocalName: localVolume,
LocalBranchName: deMasterify(localBranchName),
RemoteName: remoteVolume,
// TODO add TargetSnapshot here, to support specifying "push to a given
// snapshot" rather than just "push all snapshots up to the latest"
}

if debugMode {
fmt.Printf("[DEBUG] S3TransferRequest: %#v\n", transferRequest)
}

err = client.CallRemote(context.Background(),
"DotmeshRPC.S3Transfer", transferRequest, &transferId)
if err != nil {
return "", err
}
} else {
return "", fmt.Errorf("Unknown remote type %#v\n", remote)
}
}
return transferId, nil

}

func (dm *DotmeshAPI) parseVolumeDetails(remote Remote, peer, direction, localFilesystemName, localBranchName, remoteFilesystemName, remoteBranchName string) (string, string, string, string, string, string, error) {
var err error
if direction == "push" {
// We are pushing, so if no local filesystem/branch is
// specified, take the current one.
if localFilesystemName == "" {
localFilesystemName, err = dm.Configuration.CurrentVolume()
if err != nil {
return "", "", "", "", "", "", err
return "", err
}
}

if localBranchName == "" {
localBranchName, err = dm.Configuration.CurrentBranch()
if err != nil {
return "", "", "", "", "", "", err
return "", err
}
}
} else if direction == "pull" {
Expand All @@ -1041,15 +952,15 @@ func (dm *DotmeshAPI) parseVolumeDetails(remote Remote, peer, direction, localFi
if localFilesystemName == "" && remoteFilesystemName != "" {
_, localFilesystemName, err = ParseNamespacedVolume(remoteFilesystemName)
if err != nil {
return "", "", "", "", "", "", err
return "", err
}
}
}

// Split the local volume name's namespace out
localNamespace, localVolume, err := ParseNamespacedVolume(localFilesystemName)
if err != nil {
return "", "", "", "", "", "", err
return "", err
}

// Guess defaults for the remote filesystem
Expand All @@ -1072,7 +983,7 @@ func (dm *DotmeshAPI) parseVolumeDetails(remote Remote, peer, direction, localFi
// Default namespace for remote volume is the username on this remote
remoteNamespace, remoteVolume, err = ParseNamespacedVolumeWithDefault(remoteFilesystemName, remote.DefaultNamespace())
if err != nil {
return "", "", "", "", "", "", err
return "", err
}
}

Expand All @@ -1087,61 +998,98 @@ func (dm *DotmeshAPI) parseVolumeDetails(remote Remote, peer, direction, localFi
}

if remoteBranchName != "" && remoteVolume == "" {
return "", "", "", "", "", "", fmt.Errorf(
return "", fmt.Errorf(
"It's dubious to specify a remote branch name " +
"without specifying a remote filesystem name.",
)
}
return localNamespace, localVolume, localBranchName, remoteNamespace, remoteVolume, remoteBranchName, nil
}

func (dm *DotmeshAPI) RequestS3SubsetTransfer(
direction, peer,
localFilesystemName, localBranchName,
remoteFilesystemName, remoteBranchName string,
prefixes []string,
) (string, error) {
var transferId string
if direction == "push" {
fmt.Printf("Pushing %s/%s to %s:%s/%s\n",
localNamespace, localVolume,
peer,
remoteNamespace, remoteVolume,
)
} else {
fmt.Printf("Pulling %s/%s from %s:%s/%s\n",
localNamespace, localVolume,
peer,
remoteNamespace, remoteVolume,
)
}

connectionInitiator := dm.Configuration.CurrentRemote
// connect to connectionInitiator
client, err := dm.Configuration.ClusterFromRemote(connectionInitiator, dm.verbose)
if err != nil {
return "", err
}
remote, err := dm.Configuration.GetRemote(peer)
if err != nil {
return "", err
}
localNamespace, localVolume, localBranchName, _, remoteVolume, _, err := dm.parseVolumeDetails(remote, peer, direction, localFilesystemName, localBranchName, remoteFilesystemName, remoteBranchName)
if err != nil {
return "", err
}
s3Remote, ok := remote.(*S3Remote)
if ok {
var transferId string
// TODO make ApiKey time- and domain- (filesystem?) limited
// cryptographically somehow
dmRemote, ok := remote.(*DMRemote)

transferRequest := S3TransferRequest{
KeyID: s3Remote.KeyID,
SecretKey: s3Remote.SecretKey,
Prefixes: prefixes,
Endpoint: s3Remote.Endpoint,
Direction: direction,
LocalNamespace: localNamespace,
LocalName: localVolume,
LocalBranchName: localBranchName,
RemoteName: remoteVolume,
if ok {
transferRequest := TransferRequest{
Peer: dmRemote.Hostname,
User: dmRemote.User,
Port: dmRemote.Port,
ApiKey: dmRemote.ApiKey,
Direction: direction,
LocalNamespace: localNamespace,
LocalName: localVolume,
LocalBranchName: deMasterify(localBranchName),
RemoteNamespace: remoteNamespace,
RemoteName: remoteVolume,
RemoteBranchName: deMasterify(remoteBranchName),
// TODO add TargetSnapshot here, to support specifying "push to a given
// snapshot" rather than just "push all snapshots up to the latest"
}

if debugMode {
fmt.Printf("[DEBUG] TransferRequest: %#v\n", transferRequest)
}

err = client.CallRemote(context.Background(),
"DotmeshRPC.S3Transfer", transferRequest, &transferId)
"DotmeshRPC.Transfer", transferRequest, &transferId)
if err != nil {
return "", err
}
} else {
return "", fmt.Errorf("Unknown remote type %#v\n", remote)
s3Remote, ok := remote.(*S3Remote)
if ok {
if prefixes != nil {
dm.Configuration.SetPrefixesFor(peer, localNamespace, localVolume, prefixes)
}
prefixes, _ = s3Remote.PrefixesFor(localNamespace, localVolume)
transferRequest := S3TransferRequest{
KeyID: s3Remote.KeyID,
SecretKey: s3Remote.SecretKey,
Endpoint: s3Remote.Endpoint,
Prefixes: prefixes,
Direction: direction,
LocalNamespace: localNamespace,
LocalName: localVolume,
LocalBranchName: deMasterify(localBranchName),
RemoteName: remoteVolume,
// TODO add TargetSnapshot here, to support specifying "push to a given
// snapshot" rather than just "push all snapshots up to the latest"
}

if debugMode {
fmt.Printf("[DEBUG] S3TransferRequest: %#v\n", transferRequest)
}

err = client.CallRemote(context.Background(),
"DotmeshRPC.S3Transfer", transferRequest, &transferId)
if err != nil {
return "", err
}
} else {
return "", fmt.Errorf("Unknown remote type %#v\n", remote)
}
}
return transferId, nil

}

// FIXME: Put this in a shared library, as it duplicates the copy in
Expand All @@ -1152,6 +1100,12 @@ type VolumeName struct {
Name string
}

type S3VolumeName struct {
Namespace string
Name string
Prefixes []string
}

func (v VolumeName) String() string {
if v.Namespace == "admin" {
return v.Name
Expand Down
Loading

0 comments on commit e0f9751

Please sign in to comment.