How to handle large file upload to Google Bucket?

I have the following Golang code to upload a file to Google bucket:

package main
import (
"context"
"fmt"
"io"
"net/http"
"os"
"time"
"cloud.google.com/go/storage"
)
var serverPort = ":8444"
var googleCredential = "./credential.json"
var googleBucket = "g1-BUCKET-001"
var googleFolder = "test_folder/"
func uploadToBucket(w http.ResponseWriter, r *http.Request) {
t1 := time.Now()
fmt.Println("File Upload Endpoint Hit")
// Parse our multipart form, 10 << 20 specifies a maximum
// upload of 10 MB files.
r.ParseMultipartForm(10 << 20)
file, handler, err := r.FormFile("myFile")
if err != nil {
fmt.Fprintf(w, fmt.Sprintf("Error uploading file: %v", err))
return
}
defer file.Close()
fmt.Printf("Uploaded File: %+v\n", handler.Filename)
fmt.Printf("File Size: %+v\n", handler.Size)
fmt.Printf("MIME Header: %+v\n", handler.Header)
// Upload file to bucket
os.Setenv("GOOGLE_APPLICATION_CREDENTIALS", googleCredential)
ctx := context.Background()
client, err := storage.NewClient(ctx)
if err != nil {
fmt.Fprintf(w, fmt.Sprintf("Error creating storage.NewClient: %v", err))
return
}
defer client.Close()
fmt.Println("Bucket client created.")
// Set timeout
ctx, cancel := context.WithTimeout(ctx, time.Second*7200)
defer cancel()
// Create bucket object for stream copy
destFilePath := googleFolder + handler.Filename
fmt.Printf("Target bucket: gs://" + googleBucket + "/.\n")
fmt.Printf("Destination file path: " + destFilePath + ".\n")
o := client.Bucket(googleBucket).Object(destFilePath)
o = o.If(storage.Conditions{DoesNotExist: true})
// Upload an object with storage.Writer.
wc := o.NewWriter(ctx)
if _, err = io.Copy(wc, file); err != nil {
fmt.Fprintf(w, fmt.Sprintf("io.Copy error: %v", err))
return
}
if err := wc.Close(); err != nil {
fmt.Fprintf(w, fmt.Sprintf("Writer.Close() error: %v", err))
return
}
fmt.Printf("%s uploaded to gs://%s/%s.", handler.Filename, googleBucket, googleFolder)
t2 := time.Now()
diff := t2.Sub(t1)
fmt.Printf("Time start: %+v\n", t1)
fmt.Printf("Time end: %+v\n", t2)
fmt.Printf("Time diff: %+v\n", diff)
// Return that we have successfully uploaded our file!
fmt.Fprintf(w, "Successfully Uploaded File\n")
}
func setupRoutes() {
http.HandleFunc("/upload", uploadToBucket)
http.ListenAndServe(serverPort, nil)
}
func main() {
setupRoutes()
}

It works fine with file around 100MB. However when it gets to 1GB+, the wait time is so long. The user may think the code stops working and quit before it can finish. For now, all they get is this line at the end:

fmt.Fprintf(w, "Successfully Uploaded File\n")

How can I implement a way to give the user some feedback, like a completion bar for instance?

Here's one option to add progress tracking to an arbitrary io.Writer such as you get from o.NewWriter(ctx). Using io.MultiWriter you can duplicate writes to multiple writers. One can be your o.NewWriter(ctx) and another can be an implementation of io.Writer that simply counts bytes and can compare against a total size.

You could then run a goroutine that could periodically update progress by writing to stdout (or something else).

Take a look at this example using This Proof of concept implementation:

package main
import (
"fmt"
"io"
"os"
"github.com/farrellit/writeprogress"
)
func main() {
in, err := os.Open("/dev/urandom")
if err != nil {
panic(err)
}
out, err := os.OpenFile("/dev/null", os.O_WRONLY, 0)
if err != nil {
panic(err)
}
defer in.Close()
defer out.Close()
length := int64(1e6)
wp := writeprogress.NewProgressWriter(uint64(length))
d, _ := wp.Watch(func(p float64) { fmt.Printf("\r%2.0f%%", p*100) })
if b, err := io.Copy(
io.MultiWriter(out, wp),
&io.LimitedReader{R: in, N: length},
); err != nil {
panic(err)
} else {
<-d
fmt.Printf("\n%d/%d %2.0f%%\n", b, length, wp.GetProgress()*100)
}
}

You can do the same thing with your o.NewWriter(ctx) and a file length determined by stat. you could also use it in conjunction with something like https://github.com/schollz/progressbar if you wanted something pretty.

here a working example that stream the response after each part upload. Change your bucket, file source, the object target name and the content Type.

Curl the localhost:8080, you will received chunk response. Then, inject that logic in your code, with JSON response or whatever.

My previous issue comes from the chunkSize below 5MiB in my tests.

package main

import (
	"bytes"
	"encoding/xml"
	"fmt"
	"golang.org/x/net/context"
	"golang.org/x/oauth2/google"
	"io/ioutil"
	"log"
	"net/http"
)

const (
	bucket      = "gib-multiregion-us"
	fileName    = "./20220909_ListePieces.csv"
	objectName  = "test-multipart.csv"
	contentType = "text/csv"

	// Min 5MiB with multiple of 256 KiB
	chunckSize = 5242880 + (256 * 1024 * 0)

	// API key values
	storageapi          = "storage.googleapis.com"
	uploadsExtension    = "uploads"
	partNumberExtension = "partNumber"
	uploadIdExtension   = "uploadId"
	etagKey             = "ETag"
)

var httpClient *http.Client

func main() {
	ctx := context.Background()
	c, err := google.DefaultClient(ctx)
	if err != nil {
		log.Fatal(err)
	}
	httpClient = c

	http.HandleFunc("/", multipartUpload)
	http.ListenAndServe(":8080", nil)
}

func multipartUpload(w http.ResponseWriter, r *http.Request) {

	url := fmt.Sprintf("https://%s.%s/%s", bucket, storageapi, objectName)

	// init
	resp, err := httpClient.Post(fmt.Sprintf("%s?%s", url, uploadsExtension), contentType, nil)
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		fmt.Println("error during init: ", err)
		fmt.Fprintf(w, "error during init: %s\n", err)
		return
	}

	respbody, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		fmt.Println("error during reading init resp: ", err)
		fmt.Fprintf(w, "error during reading init resp: %s\n", err)
		return
	}
	defer resp.Body.Close()

	var initMultiPart InitiateMultipartUploadResult
	err = xml.Unmarshal(respbody, &initMultiPart)
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		fmt.Println("error during unmarshalling init resp: ", err)
		fmt.Fprintf(w, "error during unmarshalling init resp: %s\n", err)
		return
	}
	uploadId := initMultiPart.UploadId

	// send content
	content, err := ioutil.ReadFile(fileName)
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		fmt.Println("error during reading input file: ", err)
		fmt.Fprintf(w, "error during reading input file: %s\n", err)
		return
	}

	// Prepare the multipart response
	flusher, ok := w.(http.Flusher)
	if !ok {
		w.WriteHeader(http.StatusInternalServerError)
		fmt.Println("Impossible to create the stream flusher: ", err)
		fmt.Fprintf(w, "Impossible to create the stream flusher: %s\n", err)
		return
	}
	w.Header().Set("Transfer-Encoding", "chunked")
	flusher.Flush()

	complMulti := CompleteMultipartUpload{}
	counter := 0
	for {
		if len(content) <= counter*chunckSize {
			break
		}
		var toSend []byte
		if len(content) >= (counter+1)*chunckSize {
			toSend = content[counter*chunckSize : (counter+1)*chunckSize]
		} else {
			toSend = content[counter*chunckSize:]
		}
		req, err := http.NewRequest(http.MethodPut, fmt.Sprintf("%s?%s=%d&%s=%s", url, partNumberExtension, counter+1, uploadIdExtension, uploadId), bytes.NewReader(toSend))
		if err != nil {
			w.WriteHeader(http.StatusInternalServerError)
			fmt.Println("error chunk req creation: ", err)
			fmt.Fprintf(w, "error chunk req creation: %s\n", err)
			return
		}
		resp, err = httpClient.Do(req)
		if err != nil {
			w.WriteHeader(http.StatusInternalServerError)
			fmt.Println("error chunk submission: ", err)
			fmt.Fprintf(w, "error chunk submission: %s\n", err)
			return
		}

		etag := resp.Header.Get(etagKey)
		p := Part{
			PartNumber: counter + 1,
			ETag:       etag,
		}

		fmt.Printf("upload is done at %d%%\n", 100*counter*chunckSize/len(content))

		fmt.Fprintf(w, "upload is done at %d%%\n", 100*counter*chunckSize/len(content))
		flusher.Flush()

		complMulti.Parts = append(complMulti.Parts, p)
		counter++
	}

	toSend, err := xml.Marshal(complMulti)
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		fmt.Println("error marshal complete multi: ", err)
		fmt.Fprintf(w, "error marshal complete multi: %s\n", err)
		return
	}

	// Complete transfert
	resp, err = httpClient.Post(fmt.Sprintf("%s?%s=%s", url, uploadIdExtension, uploadId), "application/xml", bytes.NewReader(toSend))
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		fmt.Println("error during send complete: ", err)
		fmt.Fprintf(w, "error during send complete: %s\n", err)
		return
	}

	if resp.StatusCode == http.StatusOK {
		fmt.Fprintf(w, "done")
		fmt.Println("done")
	} else {
		b, _ := ioutil.ReadAll(resp.Body)
		fmt.Printf("error %s with body %s\n", resp.Status, string(b))
		fmt.Fprintf(w, "error %s with body %s\n", resp.Status, string(b))
	}

}

type InitiateMultipartUploadResult struct {
	InitiateMultipartUploadResult xml.Name `xml:"InitiateMultipartUploadResult"`
	Bucket                        string   `xml:"Bucket"`
	Key                           string   `xml:"Key"`
	UploadId                      string   `xml:"UploadId"`
}

type CompleteMultipartUpload struct {
	completeMultipartUpload xml.Name `xml:"CompleteMultipartUpload"`
	Parts                   []Part   `xml:"Part"`
}

type Part struct {
	part       xml.Name `xml:"Part"`
	PartNumber int      `xml:"PartNumber"`
	ETag       string   `xml:"ETag"`
}