Merge pull request #42 from sapcc/transferred-bytes

report the number of bytes transferred

Author: talal

pkg/actors/report.go

@@ -39,6 +39,7 @@ type ReportEvent struct {
 
 	IsFile             bool
 	FileTransferResult objects.TransferResult
+	FileTransferBytes  int64
 
 	IsCleanup            bool
 	CleanedUpObjectCount int64
@@ -70,6 +71,7 @@ func (r *Report) Run() {
 		filesFailed        int64
 		filesTransferred   int64
 		filesCleanedUp     int64
+		bytesTransferred   int64
 		jobsSkipped        int64
 		statter            statsd.Statter
 	)
@@ -100,6 +102,7 @@ func (r *Report) Run() {
 			switch mark.FileTransferResult {
 			case objects.TransferSuccess:
 				filesTransferred++
+				bytesTransferred += mark.FileTransferBytes
 			case objects.TransferFailed:
 				filesFailed++
 			}
@@ -125,6 +128,7 @@ func (r *Report) Run() {
 	gauge("last_run.files_transfered", filesTransferred, 1.0)
 	gauge("last_run.files_failed", filesFailed, 1.0)
 	gauge("last_run.files_cleaned_up", filesCleanedUp, 1.0)
+	gauge("last_run.bytes_transfered", bytesTransferred, 1.0)
 	if filesFailed > 0 || directoriesFailed > 0 {
 		gauge("last_run.success", 0, 1.0)
 		r.ExitCode = 1
@@ -145,6 +149,7 @@ func (r *Report) Run() {
 	if filesCleanedUp > 0 {
 		logg.Info("%d old files cleaned up", filesCleanedUp)
 	}
+	logg.Info("%d bytes transferred", bytesTransferred)
 
 	duration := time.Since(r.StartTime)
 	gauge("last_run.duration_seconds", int64(duration.Seconds()), 1.0)

pkg/actors/transfer.go

@@ -55,12 +55,12 @@ LOOP:
 			if !ok {
 				break LOOP
 			}
-			result := file.PerformTransfer()
+			result, size := file.PerformTransfer()
 			if result == objects.TransferFailed {
 				filesToRetry = append(filesToRetry, file)
 			} else {
 				t.Output <- FileInfoForCleaner{File: file, Failed: false}
-				t.Report <- ReportEvent{IsFile: true, FileTransferResult: result}
+				t.Report <- ReportEvent{IsFile: true, FileTransferResult: result, FileTransferBytes: size}
 			}
 		}
 	}
@@ -71,14 +71,15 @@ LOOP:
 	}
 	for _, file := range filesToRetry {
 		result := objects.TransferFailed
+		var size int64
 		//...but only if we were not aborted (this is checked in every loop
 		//iteration because the abort signal (i.e. Ctrl-C) could also happen
 		//during this loop)
 		if !aborted && t.Context.Err() == nil {
-			result = file.PerformTransfer()
+			result, size = file.PerformTransfer()
 		}
 		t.Output <- FileInfoForCleaner{File: file, Failed: result == objects.TransferFailed}
-		t.Report <- ReportEvent{IsFile: true, FileTransferResult: result}
+		t.Report <- ReportEvent{IsFile: true, FileTransferResult: result, FileTransferBytes: size}
 	}
 
 	//if interrupt was received, consume all remaining input to get the Scraper

pkg/objects/file.go

@@ -73,15 +73,16 @@ const (
 )
 
 //PerformTransfer transfers this file from the source to the target.
-//The return value indicates if the transfer finished successfully.
-func (f File) PerformTransfer() TransferResult {
+//It returns the TransferResult (which indicates if the transfer finished successfully)
+//and the number of bytes transferred.
+func (f File) PerformTransfer() (TransferResult, int64) {
 	object := f.TargetObject()
 
 	//check if this file needs transfer
 	if f.Job.Matcher.ImmutableFileRx != nil && f.Job.Matcher.ImmutableFileRx.MatchString(f.Spec.Path) {
 		if f.Job.Target.FileExists[object.Name()] {
 			logg.Debug("skipping %s: already transferred", object.FullName())
-			return TransferSkipped
+			return TransferSkipped, 0
 		}
 	}
 
@@ -122,14 +123,14 @@ func (f File) PerformTransfer() TransferResult {
 			logg.Error("skipping target %s: HEAD failed: %s",
 				object.FullName(), err.Error(),
 			)
-			return TransferFailed
+			return TransferFailed, 0
 		}
 	}
 
 	//if we want to upload a symlink, we can skip the whole Last-Modified/Etag
 	//shebang and straight-up compare the symlink target
 	if f.Spec.SymlinkTargetPath != "" {
-		return f.uploadSymlink(currentSymlinkTarget, hdr.IsLargeObject())
+		return f.uploadSymlink(currentSymlinkTarget, hdr.IsLargeObject()), 0
 	}
 
 	//retrieve object from source, taking advantage of Etag and Last-Modified where possible
@@ -160,13 +161,13 @@ func (f File) PerformTransfer() TransferResult {
 	}
 	if err != nil {
 		logg.Error("GET %s failed: %s", f.Spec.Path, err.Error())
-		return TransferFailed
+		return TransferFailed, 0
 	}
 	if body != nil {
 		defer body.Close()
 	}
 	if sourceState.SkipTransfer { // 304 Not Modified
-		return TransferSkipped
+		return TransferSkipped, 0
 	}
 
 	if util.LogIndividualTransfers {
@@ -198,9 +199,9 @@ func (f File) PerformTransfer() TransferResult {
 	}
 
 	if ok {
-		return TransferSuccess
+		return TransferSuccess, size
 	}
-	return TransferFailed
+	return TransferFailed, 0
 }
 
 func (f File) uploadSymlink(previousTarget *schwift.Object, cleanupOldSegments bool) TransferResult {