gcs.go

/*
Copyright 2018 The Kubernetes Authors.

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.
*/

package gcs

import (
	"context"
	"errors"
	"fmt"
	"hash/crc32"
	"log"
	"net/url"
	"strings"

	"cloud.google.com/go/storage"
	"google.golang.org/api/option"
)

// ClientWithCreds returns a storage client, optionally authenticated with the specified .json creds
func ClientWithCreds(ctx context.Context, creds ...string) (*storage.Client, error) {
	var options []option.ClientOption
	switch l := len(creds); l {
	case 0: // Do nothing
	case 1:
		options = append(options, option.WithCredentialsFile(creds[0]))
	default:
		return nil, fmt.Errorf("%d creds files unsupported (at most 1)", l)
	}
	return storage.NewClient(ctx, options...)
}

// Path parses gs://bucket/obj urls
type Path struct {
	url url.URL
}

// String returns the gs://bucket/obj url
func (g Path) String() string {
	return g.url.String()
}

// Set updates value from a gs://bucket/obj string, validating errors.
func (g *Path) Set(v string) error {
	u, err := url.Parse(v)
	if err != nil {
		return fmt.Errorf("invalid gs:// url %s: %v", v, err)
	}
	return g.SetURL(u)
}

// SetURL updates value to the passed in gs://bucket/obj url
func (g *Path) SetURL(u *url.URL) error {
	switch {
	case u == nil:
		return errors.New("nil url")
	case u.Scheme != "gs":
		return fmt.Errorf("must use a gs:// url: %s", u)
	case strings.Contains(u.Host, ":"):
		return fmt.Errorf("gs://bucket may not contain a port: %s", u)
	case u.Opaque != "":
		return fmt.Errorf("url must start with gs://: %s", u)
	case u.User != nil:
		return fmt.Errorf("gs://bucket may not contain an user@ prefix: %s", u)
	case u.RawQuery != "":
		return fmt.Errorf("gs:// url may not contain a ?query suffix: %s", u)
	case u.Fragment != "":
		return fmt.Errorf("gs:// url may not contain a #fragment suffix: %s", u)
	}
	g.url = *u
	return nil
}

// ResolveReference returns the path relative to the current path
func (g Path) ResolveReference(ref *url.URL) (*Path, error) {
	var newP Path
	if err := newP.SetURL(g.url.ResolveReference(ref)); err != nil {
		return nil, err
	}
	return &newP, nil
}

// Bucket returns bucket in gs://bucket/obj
func (g Path) Bucket() string {
	return g.url.Host
}

// Object returns path/to/something in gs://bucket/path/to/something
func (g Path) Object() string {
	if g.url.Path == "" {
		return g.url.Path
	}
	return g.url.Path[1:]
}

func calcCRC(buf []byte) uint32 {
	return crc32.Checksum(buf, crc32.MakeTable(crc32.Castagnoli))
}

// Upload writes bytes to the specified Path
func Upload(ctx context.Context, client *storage.Client, path Path, buf []byte) error {
	crc := calcCRC(buf)
	w := client.Bucket(path.Bucket()).Object(path.Object()).NewWriter(ctx)
	w.SendCRC32C = true
	// Send our CRC32 to ensure google received the same data we sent.
	// See checksum example at:
	// https://godoc.org/cloud.google.com/go/storage#Writer.Write
	w.ObjectAttrs.CRC32C = crc
	w.ProgressFunc = func(bytes int64) {
		log.Printf("Uploading %s: %d/%d...", path, bytes, len(buf))
	}
	if n, err := w.Write(buf); err != nil {
		return fmt.Errorf("writing %s failed: %v", path, err)
	} else if n != len(buf) {
		return fmt.Errorf("partial write of %s: %d < %d", path, n, len(buf))
	}
	if err := w.Close(); err != nil {
		return fmt.Errorf("closing %s failed: %v", path, err)
	}
	return nil
}