v0522

Author: giventocode

LICENSE

@@ -31,3 +31,11 @@ Unless required by applicable law or agreed to in writing, software distributed
 See the License for specific language governing permissions and limitations under the License. 
 --------------------------------------------------------End of License---------------------------------------------------------------------------------------
 
+Minio Client SDK for Go 
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License. You may obtain a copy of the License at
+http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and limitations under the License. 

README.md

@@ -15,12 +15,12 @@ Sources and targets are decoupled, this design enables the composition of variou
 | ---------------  | -----------------| -----------------|-----------------|
 | File (Upload)    | Yes              | Yes              | NA              |
 | HTTP/HTTPS*      | Yes              | Yes              | Yes             |
-| Azure Block Blob | Yes**            | Yes**            | Yes             |
-| Azure Page Blob  | Yes**            | Yes**            | Yes             |
+| Azure Block Blob | Yes              | Yes              | Yes             |
+| Azure Page Blob  | Yes              | Yes              | Yes             |
+| S3 Endpoint      | Yes              | Yes              | No              |
 
 *\*   The HTTP/HTTPS source must support HTTP byte ranges and return the file size as a response to a HTTP HEAD request.*
 
-*\*\* Using the Blob URL with a valid SAS Token with read access .*
 
 ## Getting Started
 
@@ -29,7 +29,7 @@ Sources and targets are decoupled, this design enables the composition of variou
 Download, extract and set permissions:
 
 ```bash
-wget -O bp_linux.tar.gz https://github.com/Azure/blobporter/releases/download/v0.5.14/bp_linux.tar.gz
+wget -O bp_linux.tar.gz https://github.com/Azure/blobporter/releases/download/v0.5.22/bp_linux.tar.gz
 tar -xvf bp_linux.tar.gz linux_amd64/blobporter
 chmod +x ~/linux_amd64/blobporter
 cd ~/linux_amd64
@@ -46,7 +46,7 @@ export ACCOUNT_KEY=<STORAGE_ACCOUNT_KEY>
 
 ### Windows
 
-Download [BlobPorter.exe](https://github.com/Azure/blobporter/releases/download/v0.5.14/bp_windows.zip)
+Download [BlobPorter.exe](https://github.com/Azure/blobporter/releases/download/v0.5.22/bp_windows.zip)
 
 Set environment variables (if using the command prompt):
 
@@ -94,6 +94,25 @@ For example, a single file upload to page blob:
 
 >Note: The file size and block size must be a multiple of 512 (bytes). The maximum block size is 4MB.
 
+### Upload from an S3 compatible endpoint to Azure Blob Storage
+
+You can upload data directly from a S3 compatible endpoint. First you must specify the access and secret keys via environment variables.
+
+```bash
+export S3_ACCESS_KEY=<YOUR ACCESS KEY>
+export S3_SECRET_KEY=<YOUR_SECRET_KEY>
+```
+
+Then you can specify a S3 URI, with the following format:
+
+[URL]/[BUCKET][PREFIX]
+
+For example:
+
+`./blobporter -f s3://s3.amazonaws.com/bpperftest/mydata -c froms3 -t s3-blockblob -p`
+
+Note: It is recommended to perform this operation from a VM running on the cloud. This is a network bound operation where is uploaded as it is received from the source.
+
 ### Upload from an HTTP/HTTPS source to Azure Blob Storage
 
 To block blob storage:
@@ -104,13 +123,22 @@ To page blob storage:
 
 `./blobporter -f "http://mysource/my.vhd"  -c mycontainer -n my.vhd -t http-pageblob`
 
-You can use this approach to transfer data between Azure Storage accounts and blob types - e.g. transfer a blob from one account to another or from a page blob to block blob.
 
-The source is a page blob with a SAS token and the target is block blob:
+### Synchronously Copy data in Azure Blob Storage
+
+You can synchronously transfer data between Azure Storage accounts, containers and blob types.
+
+First, you must set the account key of the source storage account.
+
+```bash
+export SOURCE_ACCOUNT_KEY=<YOUR  KEY>
+```
+
+Then you can specify the URI of the source. Prefixes supported.
 
-`./blobporter -f "https://myaccount.blob.core.windows.net/vhds/my.vhd?st=2015-03-23T03%3A59%3A00Z&se=2015-03-24T03%3A59%3A00Z&sp=rl&sv=2015-12-11&sr=b&sig=123rfdAsYyqOxxOGe28%3Fp4H6r5reR8pdSBdlchi64wgs3D"  -c mycontainer -n my.vhd -t http-blockblob`
+`./blobporter -f "https://mysourceaccount.blob.core.windows.net/container/myblob"  -c mycontainer -t blob-blockblob`
 
->Note: In HTTP/HTTPS to blob transfers, data is downloaded and uploaded as it is received without disk IO.
+>Note: It is recommended to perform this operation from a VM running in the same region as source or the target. As with all HTTP based transfers, data is uploaded as it is downloaded from the source, therefore the transfer is primarily network bound.
 
 ### Download from Azure Blob Storage
 
@@ -139,7 +167,7 @@ By default files are downloaded to the same directory where you are running blob
 
 ## Command Options
 
-- `-f`, `--source_file` *string* URL, file or files (e.g. /data/*.gz) to upload.
+- `-f`, `--source_file` *string* URL, Azure Blob or S3 Endpoint, file or files (e.g. /data/*.gz) to upload.
 
 - `-c`, `--container_name` *string* container name (e.g. `mycontainer`).
 
@@ -167,7 +195,7 @@ By default files are downloaded to the same directory where you are running blob
 
 - `-d`, `--dup_check_level` *string* desired level of effort to detect duplicate data blocks to minimize upload size. Must be one of None, ZeroOnly, Full (default "None")
 
-- `-t`, `--transfer_type` *string*  defines the source and target of the transfer. Must be one of file-blockblob, file-pageblob, http-blockblob, http-pageblob, blob-file, pageblock-file (alias of blob-file), blockblob-file (alias of blob-file) or http-file
+- `-t`, `--transfer_type` *string*  defines the source and target of the transfer. Must be one of file-blockblob, file-pageblob, http-blockblob, http-pageblob, blob-file, pageblock-file (alias of blob-file), blockblob-file (alias of blob-file), http-file, blob-pageblob, blob-blockblob, s3-pageblob and s3-blockblob.
 
 - `m`, `--compute_blockmd5` *bool* if present or true, block level MD5 has will be computed and included as a header when the block is sent to blob storage. Default is false.
 
@@ -189,7 +217,7 @@ By default, BlobPorter creates 5 readers and 8 workers for each core on the comp
 
 - For transfers from fast disks (SSD) or HTTP sources reducing the number readers or workers could provide better performance than the default values. Reduce these values if you want to minimize resource utilization. Lowering these numbers reduces contention and the likelihood of experiencing throttling conditions.
 
-- Transfers can be batched. Each batch transfer will concurrently read and transfer up to 200 files (default value) from the source. The batch size can be modified using the -x option, the maximum value is 500.
+- Transfers can be batched. Each batch transfer will concurrently read and transfer up to 500 files (default value) from the source. The batch size can be modified using the -x option.
 
 - Blobs smaller than the block size are transferred in a single operation. With relatively small files (<32MB) performance may be better if you set a block size equal to the size of the files. Setting the number of workers and readers to the number of files could yield performance gains.
 

blobporter.go

@@ -54,12 +54,12 @@ const (
 	s3AccessKeyEnvVar = "S3_ACCESS_KEY"
 	s3SecretKeyEnvVar = "S3_SECRET_KEY"
 
-	programVersion = "0.5.21" // version number to show in help
+	programVersion = "0.5.22" // version number to show in help
 )
 
 const numOfWorkersFactor = 8
 const numOfReadersFactor = 5
-const defaultNumberOfFilesInBatch = 200
+const defaultNumberOfFilesInBatch = 500
 const defaultNumberOfHandlesPerFile = 2
 
 func init() {

inttest._m.sh

@@ -100,4 +100,13 @@ dd if=/dev/urandom of=$F1 bs=36M count=1 iflag=fullblock
 
 #Download file
 ./blobporter -n $DOWN_F1 -c $CONT -t blob-file -b 32MB
-calculateMD5 $F1 $DOWN_F1
+calculateMD5 $F1 $DOWN_F1
+
+
+#Scenario 9 - Synchronous copy from one container to another in the same storage account.
+CONT2="syncopy"
+SRC_URL="https://"$ACCOUNT_NAME".blob.core.windows.net/"$CONT
+SRC_ACCOUNT_KEY=$ACCOUNT_KEY
+
+./blobporter -f $SRC_URL -c $CONT2 -t blob-blockblob
+

sources/http.go

@@ -1,7 +1,7 @@
 package sources
 
 import (
-	"io"
+	"io/ioutil"
 	"log"
 	"strconv"
 	"strings"
@@ -201,12 +201,17 @@ func (f *HTTPPipeline) ExecuteReader(partitionsQ chan pipeline.PartsPartition, p
 					err = fmt.Errorf("Invalid status code in the response. Status: %v Bytes: %v", status, header)
 				}
 
+				if res != nil && res.Body != nil {
+					res.Body.Close()
+				}
+				f.HTTPClient = util.NewHTTPClient()
+
 				util.PrintfIfDebug("ExecuteReader -> |%v|%v|%v|%v|%v", p.BlockID, p.BytesToRead, status, err, header)
 
 				return err
 			}
-			//p.Data, err = ioutil.ReadAll(res.Body)
-			_, err = io.ReadFull(res.Body, p.Data[:p.BytesToRead])
+			p.Data, err = ioutil.ReadAll(res.Body)
+			//_, err = io.ReadFull(res.Body, p.Data[:p.BytesToRead])
 
 			res.Body.Close()
 			if err != nil {

targets/multifile.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/Azure/blobporter/pipeline"
+	"github.com/Azure/blobporter/util"
 )
 
 ////////////////////////////////////////////////////////////
@@ -18,7 +19,7 @@ import (
 //MultiFile represents an OS file(s) target
 type MultiFile struct {
 	Container       string
-	FileHandles     map[string]chan *os.File
+	FileHandles     sync.Map
 	NumberOfHandles int
 	OverWrite       bool
 	sync.Mutex
@@ -27,51 +28,43 @@ type MultiFile struct {
 //NewMultiFile creates a new multi file target and 'n' number of handles for concurrent writes to a file.
 func NewMultiFile(overwrite bool, numberOfHandles int) pipeline.TargetPipeline {
 
-	return &MultiFile{FileHandles: make(map[string]chan *os.File), NumberOfHandles: numberOfHandles, OverWrite: overwrite}
+	//return &MultiFile{FileHandles: make(map[string]chan *os.File), NumberOfHandles: numberOfHandles, OverWrite: overwrite}
+	return &MultiFile{NumberOfHandles: numberOfHandles, OverWrite: overwrite}
 }
 
 //PreProcessSourceInfo implementation of PreProcessSourceInfo from the pipeline.TargetPipeline interface.
 //Passthrough no need to pre-process for a file target.
 func (t *MultiFile) PreProcessSourceInfo(source *pipeline.SourceInfo, blockSize uint64) (err error) {
-	t.Lock()
-	defer t.Unlock()
+	return nil
+}
 
+func (t *MultiFile) createFileIfNotExists(targetAlias string) error {
 	var fh *os.File
+	var err error
 
-	path := filepath.Dir(source.TargetAlias)
+	path := filepath.Dir(targetAlias)
 
 	if path != "" {
 		err = os.MkdirAll(path, 0777)
 
 		if err != nil {
-			return
+			return err
 		}
 	}
-
-	if fh, err = os.Create(source.TargetAlias); os.IsExist(err) {
+	defer fh.Close()
+	if fh, err = os.Create(targetAlias); os.IsExist(err) {
 		if !t.OverWrite {
 			return fmt.Errorf("The file already exists and file overwrite is disabled")
 		}
-		if err = os.Remove(source.TargetAlias); err != nil {
+		if err = os.Remove(targetAlias); err != nil {
 			return err
 		}
 
-		if fh, err = os.Create(source.TargetAlias); err != nil {
+		if fh, err = os.Create(targetAlias); err != nil {
 			return err
 		}
-	}
-
-	fhQ := make(chan *os.File, t.NumberOfHandles)
-
-	for i := 0; i < t.NumberOfHandles; i++ {
 
-		if fh, err = os.OpenFile(source.TargetAlias, os.O_WRONLY, os.ModeAppend); err != nil {
-			log.Fatal(err)
-		}
-
-		fhQ <- fh
 	}
-	t.FileHandles[source.TargetAlias] = fhQ
 
 	return nil
 }
@@ -80,10 +73,18 @@ func (t *MultiFile) PreProcessSourceInfo(source *pipeline.SourceInfo, blockSize
 //For a file download a final commit is not required and this implementation closes all the filehandles.
 func (t *MultiFile) CommitList(listInfo *pipeline.TargetCommittedListInfo, numberOfBlocks int, targetName string) (msg string, err error) {
 
-	close(t.FileHandles[targetName])
+	value, ok := t.FileHandles.Load(targetName)
+	util.PrintfIfDebug("CommitList -> targetname:%v listinfo:%+v\n", targetName, *listInfo)
+
+	if !ok {
+		return "", nil
+	}
+
+	fhq := value.(chan *os.File)
+	close(fhq)
 	for {
 
-		fh, ok := <-t.FileHandles[targetName]
+		fh, ok := <-fhq
 
 		if !ok {
 			break
@@ -104,21 +105,90 @@ func (t *MultiFile) CommitList(listInfo *pipeline.TargetCommittedListInfo, numbe
 //ProcessWrittenPart implements ProcessWrittenPart from the pipeline.TargetPipeline interface.
 //Passthrough implementation as no post-written-processing is required (e.g. maintain a list) when files are downloaded.
 func (t *MultiFile) ProcessWrittenPart(result *pipeline.WorkerResult, listInfo *pipeline.TargetCommittedListInfo) (requeue bool, err error) {
+	//fmt.Printf("WrittenPart->%+v \n", *result)
 	return false, nil
 }
 
+func (t *MultiFile) loadHandle(part *pipeline.Part) (*os.File, error) {
+	var fh *os.File
+	var fhQ chan *os.File
+	var err error
+
+	//if this is a small file open the handle but not leave open
+	if part.NumberOfBlocks == 1 {
+		if err = t.createFileIfNotExists(part.TargetAlias); err != nil {
+			return nil, err
+		}
+		if fh, err = os.OpenFile(part.TargetAlias, os.O_WRONLY, os.ModeAppend); err != nil {
+			return nil, err
+		}
+		return fh, nil
+	}
+
+	t.Lock()
+	value, _ := t.FileHandles.Load(part.TargetAlias)
+	if value == nil {
+		fhQ = make(chan *os.File, t.NumberOfHandles)
+		if err = t.createFileIfNotExists(part.TargetAlias); err != nil {
+			return nil, err
+		}
+		for i := 0; i < t.NumberOfHandles; i++ {
+			if fh, err = os.OpenFile(part.TargetAlias, os.O_WRONLY, os.ModeAppend); err != nil {
+				return nil, err
+			}
+			fhQ <- fh
+		}
+		t.FileHandles.Store(part.TargetAlias, fhQ)
+
+	} else {
+		fhQ = value.(chan *os.File)
+	}
+	t.Unlock()
+
+	fh = <-fhQ
+	return fh, nil
+}
+
+func (t *MultiFile) closeOrKeepHandle(part *pipeline.Part, fh *os.File) error {
+	if part.NumberOfBlocks == 1 {
+		return fh.Close()
+	}
+
+	t.Lock()
+	defer t.Unlock()
+	value, ok := t.FileHandles.Load(part.TargetAlias)
+
+	if ok {
+		fhq := value.(chan *os.File)
+		fhq <- fh
+		t.FileHandles.Store(part.TargetAlias, fhq)
+		return nil
+	}
+
+	return fmt.Errorf("File handle channel not found in the map")
+
+}
+
 //WritePart implements WritePart from the pipeline.TargetPipeline interface.
 //Writes to a local file using a filehandle received from a channel.
 func (t *MultiFile) WritePart(part *pipeline.Part) (duration time.Duration, startTime time.Time, numOfRetries int, err error) {
 	startTime = time.Now()
+	var fh *os.File
 
-	fh := <-t.FileHandles[part.TargetAlias]
-	if _, err := fh.WriteAt((*part).Data, int64((*part).Offset)); err != nil {
+	if fh, err = t.loadHandle(part); err != nil {
 		log.Fatal(err)
 	}
-	duration = time.Now().Sub(startTime)
 
-	t.FileHandles[part.TargetAlias] <- fh
+	if _, err = fh.WriteAt((*part).Data, int64((*part).Offset)); err != nil {
+		log.Fatal(err)
+	}
+
+	//fmt.Printf("Part Written:%+v\n", part.Ordinal)
+
+	if err = t.closeOrKeepHandle(part, fh); err != nil {
+		log.Fatal(err)
+	}
+	duration = time.Now().Sub(startTime)
 
 	return
 }