blobserver/google/cloudstorage: make it a SubFetcher

For blobpacked.

Change-Id: Ibca3d8a6976d8e1d4058e8d7fcdca6b63f8d8b87

pkg/blobserver/google/cloudstorage/storage.go

@@ -50,8 +50,9 @@ type Storage struct {
 }
 
 var (
-	_ blobserver.MaxEnumerateConfig = (*Storage)(nil)
+	_ blob.SubFetcher               = (*Storage)(nil)
 	_ blobserver.Generationer       = (*Storage)(nil)
+	_ blobserver.MaxEnumerateConfig = (*Storage)(nil)
 )
 
 func (gs *Storage) MaxEnumerate() int { return 1000 }
@@ -195,7 +196,10 @@ func (s *Storage) Fetch(br blob.Ref) (rc io.ReadCloser, size uint32, err error)
 		err = errors.New("object too big")
 	}
 	return rc, uint32(sz), err
+}
 
+func (s *Storage) SubFetch(br blob.Ref, offset, length int64) (rc io.ReadCloser, err error) {
+	return s.client.GetPartialObject(googlestorage.Object{Bucket: s.bucket, Key: br.String()}, offset, length)
 }
 
 func (s *Storage) RemoveBlobs(blobs []blob.Ref) error {

pkg/blobserver/storagetest/storagetest.go

@@ -200,7 +200,10 @@ func testSubFetcher(t *testing.T, sto blobserver.Storage) {
 	for _, tt := range regions {
 		r, err := sf.SubFetch(big.BlobRef(), tt.off, tt.limit)
 		if err != nil {
-			t.Fatal("Error fetching big blob for SubFetch: %v", err)
+			t.Fatalf("Error fetching big blob for SubFetch: %v", err)
+		}
+		if r == nil {
+			t.Fatal("SubFetch returned nil, nil")
 		}
 		all, err := ioutil.ReadAll(r)
 		r.Close()

pkg/googlestorage/googlestorage.go

@@ -162,24 +162,64 @@ func (gsa *Client) simpleRequest(method, url_ string) (resp *http.Response, err
 
 // GetObject fetches a Google Cloud Storage object.
 // The caller must close rc.
-func (gsa *Client) GetObject(obj *Object) (rc io.ReadCloser, size int64, err error) {
+func (c *Client) GetObject(obj *Object) (rc io.ReadCloser, size int64, err error) {
 	if err = obj.valid(); err != nil {
 		return
 	}
-	resp, err := gsa.simpleRequest("GET", gsAccessURL+"/"+obj.Bucket+"/"+obj.Key)
+	resp, err := c.simpleRequest("GET", gsAccessURL+"/"+obj.Bucket+"/"+obj.Key)
 	if err != nil {
 		return nil, 0, fmt.Errorf("GS GET request failed: %v\n", err)
 	}
 	if resp.StatusCode == http.StatusNotFound {
+		resp.Body.Close()
 		return nil, 0, os.ErrNotExist
 	}
 	if resp.StatusCode != http.StatusOK {
+		resp.Body.Close()
 		return nil, 0, fmt.Errorf("GS GET request failed status: %v\n", resp.Status)
 	}
 
 	return resp.Body, resp.ContentLength, nil
 }
 
+// GetPartialObject fetches part of a Google Cloud Storage object.
+// If length is negative, the rest of the object is returned.
+// The caller must close rc.
+func (c *Client) GetPartialObject(obj Object, offset, length int64) (rc io.ReadCloser, err error) {
+	if offset < 0 {
+		return nil, errors.New("invalid negative length")
+	}
+	if err = obj.valid(); err != nil {
+		return
+	}
+
+	req, err := http.NewRequest("GET", gsAccessURL+"/"+obj.Bucket+"/"+obj.Key, nil)
+	if err != nil {
+		return
+	}
+	req.Header.Set("x-goog-api-version", "2")
+	if length >= 0 {
+		req.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", offset, offset+length-1))
+	} else {
+		req.Header.Set("Range", fmt.Sprintf("bytes=%d-", offset))
+	}
+
+	resp, err, _ := c.doRequest(req, true)
+	if err != nil {
+		return nil, fmt.Errorf("GS GET request failed: %v\n", err)
+	}
+	if resp.StatusCode == http.StatusNotFound {
+		resp.Body.Close()
+		return nil, os.ErrNotExist
+	}
+	if !(resp.StatusCode == http.StatusPartialContent || (offset == 0 && resp.StatusCode == http.StatusOK)) {
+		resp.Body.Close()
+		return nil, fmt.Errorf("GS GET request failed status: %v\n", resp.Status)
+	}
+
+	return resp.Body, nil
+}
+
 // StatObject checks for the size & existence of a Google Cloud Storage object.
 // Non-existence of a file is not an error.
 func (gsa *Client) StatObject(obj *Object) (size int64, exists bool, err error) {