cipd: Refactor client to make it more readable.

go/src/infra/tools/cipd/client.go

Index: go/src/infra/tools/cipd/client.go
diff --git a/go/src/infra/tools/cipd/client.go b/go/src/infra/tools/cipd/client.go
new file mode 100644
index 0000000000000000000000000000000000000000..ab4b763ebd32efb2a9ab04909176abea2caa0fc6
--- /dev/null
+++ b/go/src/infra/tools/cipd/client.go
@@ -0,0 +1,629 @@
+// Copyright 2015 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+/*
+Package cipd implements client side of Chrome Infra Package Deployer.
+
+TODO: write more.
+
+Binary package file format (in free form representation):
+  <binary package> := <zipped data>
+  <zipped data> := DeterministicZip(<all input files> + <manifest json>)
+  <manifest json> := File{
+    name: ".cipdpkg/manifest.json",
+    data: JSON({
+      "FormatVersion": "1",
+      "PackageName": <name of the package>
+    }),
+  }
+  DeterministicZip = zip archive with deterministic ordering of files and stripped timestamps
+
+Main package data (<zipped data> above) is deterministic, meaning its content
+depends only on inputs used to built it (byte to byte): contents and names of
+all files added to the package (plus 'executable' file mode bit) and a package
+name (and all other data in the manifest).
+
+Binary package data MUST NOT depend on a timestamp, hostname of machine that
+built it, revision of the source code it was built from, etc. All that
+information will be distributed as a separate metadata packet associated with
+the package when it gets uploaded to the server.
+
+TODO: expand more when there's server-side package data model (labels
+and stuff).
+*/
+package cipd
+
+import (
+	"bufio"
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"infra/libs/build"
+	"infra/libs/logging"
+
+	"infra/tools/cipd/common"
+	"infra/tools/cipd/local"
+)
+
+// PackageACLChangeAction defines a flavor of PackageACLChange.
+type PackageACLChangeAction string
+
+const (
+	// GrantRole is used in PackageACLChange to request a role to be granted.
+	GrantRole PackageACLChangeAction = "GRANT"
+	// RevokeRole is used in PackageACLChange to request a role to be revoked.
+	RevokeRole PackageACLChangeAction = "REVOKE"
+
+	// CASFinalizationTimeout is how long to wait for CAS service to finalize the upload.
+	CASFinalizationTimeout = 1 * time.Minute
+	// TagAttachTimeout is how long to wait for instance to be processed when attaching tags.
+	TagAttachTimeout = 1 * time.Minute
+
+	// UserAgent is HTTP user agent string for CIPD client.
+	UserAgent = "cipd 1.0"
+
+	// ProdServiceURL is URL of a backend to connect to if client is build with +release tag.
+	ProdServiceURL = "https://chrome-infra-packages.appspot.com"
+	// TestingServiceURL is URL of a backend to connect to if client is build without +release tag.
+	TestingServiceURL = "https://chrome-infra-packages-dev.appspot.com"
+)
+
+var (
+	// ErrFinalizationTimeout is returned if CAS service can not finalize upload fast enough.
+	ErrFinalizationTimeout = errors.New("Timeout while waiting for CAS service to finalize the upload")
+	// ErrBadUpload is returned when a package file is uploaded, but servers asks us to upload it again.
+	ErrBadUpload = errors.New("Package file is uploaded, but servers asks us to upload it again")
+	// ErrBadUploadSession is returned by UploadToCAS if provided UploadSession is not valid.
+	ErrBadUploadSession = errors.New("UploadURL must be set if UploadSessionID is used")
+	// ErrUploadSessionDied is returned by UploadToCAS if upload session suddenly disappeared.
+	ErrUploadSessionDied = errors.New("Upload session is unexpectedly missing")
+	// ErrNoUploadSessionID is returned by UploadToCAS if server didn't provide upload session ID.
+	ErrNoUploadSessionID = errors.New("Server didn't provide upload session ID")
+	// ErrAttachTagsTimeout is returned when service refuses to accept tags for a long time.
+	ErrAttachTagsTimeout = errors.New("Timeout while attaching tags")
+	// ErrDownloadError is returned by FetchInstance on download errors.
+	ErrDownloadError = errors.New("Failed to download the package file after multiple attempts")
+	// ErrUploadError is returned by RegisterInstance and UploadToCAS on upload errors.
+	ErrUploadError = errors.New("Failed to upload the package file after multiple attempts")
+	// ErrAccessDenined is returned by calls talking to backend on 401 or 403 HTTP errors.
+	ErrAccessDenined = errors.New("Access denied (not authenticated or not enough permissions)")
+	// ErrBackendInaccessible is returned by calls talking to backed if it doesn't response.
+	ErrBackendInaccessible = errors.New("Request to the backend failed after multiple attempts")
+	// ErrEnsurePackagesFailed is returned by EnsurePackages if something is not right.
+	ErrEnsurePackagesFailed = errors.New("Failed to update packages, see the log")
+)
+
+// HTTPClientFactory lazily creates http.Client to use for making requests.
+type HTTPClientFactory func() (*http.Client, error)
+
+// Client provides high-level CIPD client interface.
+type Client struct {
+	// ServiceURL is root URL of the backend service.
+	ServiceURL string
+	// Log is a logger to use for logs, default is logging.DefaultLogger.
+	Log logging.Logger
+	// AuthenticatedClientFactory lazily creates http.Client to use for making RPC requests.
+	AuthenticatedClientFactory HTTPClientFactory
+	// AnonymousClientFactory lazily creates http.Client to use for making requests to storage.
+	AnonymousClientFactory HTTPClientFactory
+	// UserAgent is put into User-Agent HTTP header with each request.
+	UserAgent string
+
+	// clock provides current time and ability to sleep.
+	clock clock
+	// remote knows how to call backend REST API.
+	remote remote
+	// storage knows how to upload and download raw binaries using signed URLs.
+	storage storage
+
+	// authClient is a lazily created http.Client to use for authenticated requests.
+	authClient *http.Client
+	// anonClient is a lazily created http.Client to use for anonymous requests.
+	anonClient *http.Client
+}
+
+// PackageACL is per package path per role access control list that is a part of
+// larger overall ACL: ACL for package "a/b/c" is a union of PackageACLs for "a"
+// "a/b" and "a/b/c".
+type PackageACL struct {
+	// PackagePath is a package subpath this ACL is defined for.
+	PackagePath string
+	// Role is a role that listed users have, e.g. 'READER', 'WRITER', ...
+	Role string
+	// Principals list users and groups granted the role.
+	Principals []string
+	// ModifiedBy specifies who modified the list the last time.
+	ModifiedBy string
+	// ModifiedTs is a timestamp when the list was modified the last time.
+	ModifiedTs time.Time
+}
+
+// PackageACLChange is a mutation to some package ACL.
+type PackageACLChange struct {
+	// Action defines what action to perform: GrantRole or RevokeRole.
+	Action PackageACLChangeAction
+	// Role to grant or revoke to a user or group.
+	Role string
+	// Principal is a user or a group to grant or revoke a role for.
+	Principal string
+}
+
+// UploadSession describes open CAS upload session.
+type UploadSession struct {
+	// ID identifies upload session in the backend.
+	ID string
+	// URL is where to upload the data to.
+	URL string
+}
+
+// NewClient initializes default CIPD client object. Its fields can be further
+// tweaked after this call.
+func NewClient() *Client {
+	c := &Client{
+		ServiceURL: ProdServiceURL,
+		Log:        logging.DefaultLogger,
+		AuthenticatedClientFactory: func() (*http.Client, error) { return http.DefaultClient, nil },
+		AnonymousClientFactory:     func() (*http.Client, error) { return http.DefaultClient, nil },
+		UserAgent:                  UserAgent,
+		clock:                      &clockImpl{},
+	}
+	if !build.ReleaseBuild {
+		c.ServiceURL = TestingServiceURL
+		c.UserAgent += " testing"
+	}
+	c.remote = &remoteImpl{c}
+	c.storage = &storageImpl{c, uploadChunkSize}
+	return c
+}
+
+// doAuthenticatedHTTPRequest is used by remote implementation to make HTTP calls.
+func (client *Client) doAuthenticatedHTTPRequest(req *http.Request) (*http.Response, error) {
+	if client.authClient == nil {
+		var err error
+		client.authClient, err = client.AuthenticatedClientFactory()
+		if err != nil {
+			return nil, err
+		}
+	}
+	return client.authClient.Do(req)
+}
+
+// doAnonymousHTTPRequest is used by storage implementation to make HTTP calls.
+func (client *Client) doAnonymousHTTPRequest(req *http.Request) (*http.Response, error) {
+	if client.anonClient == nil {
+		var err error
+		client.anonClient, err = client.AnonymousClientFactory()
+		if err != nil {
+			return nil, err
+		}
+	}
+	return client.anonClient.Do(req)
+}
+
+// FetchACL returns a list of PackageACL objects (parent paths first) that
+// together define the access control list for the given package subpath.
+func (client *Client) FetchACL(packagePath string) ([]PackageACL, error) {
+	return client.remote.fetchACL(packagePath)
+}
+
+// ModifyACL applies a set of PackageACLChanges to a package path.
+func (client *Client) ModifyACL(packagePath string, changes []PackageACLChange) error {
+	return client.remote.modifyACL(packagePath, changes)
+}
+
+// UploadToCAS uploads package data blob to Content Addressed Store if it is not
+// there already. The data is addressed by SHA1 hash (also known as package's
+// InstanceID). It can be used as a standalone function (if 'session' is nil)
+// or as a part of more high level upload process (in that case upload session
+// can be opened elsewhere and its properties passed here via 'session'
+// argument). Returns nil on successful upload.
+func (client *Client) UploadToCAS(sha1 string, data io.ReadSeeker, session *UploadSession) error {
+	// Open new upload session if an existing is not provided.
+	var err error
+	if session == nil {
+		client.Log.Infof("cipd: uploading %s: initiating", sha1)
+		session, err = client.remote.initiateUpload(sha1)
+		if err != nil {
+			client.Log.Warningf("cipd: can't upload %s - %s", sha1, err)
+			return err
+		}
+		if session == nil {
+			client.Log.Infof("cipd: %s is already uploaded", sha1)
+			return nil
+		}
+	} else {
+		if session.ID == "" || session.URL == "" {
+			return ErrBadUploadSession
+		}
+	}
+
+	// Upload the file to CAS storage.
+	err = client.storage.upload(session.URL, data)
+	if err != nil {
+		return err
+	}
+
+	// Finalize the upload, wait until server verifies and publishes the file.
+	started := client.clock.now()
+	delay := time.Second
+	for {
+		published, err := client.remote.finalizeUpload(session.ID)
+		if err != nil {
+			client.Log.Warningf("cipd: upload of %s failed: %s", sha1, err)
+			return err
+		}
+		if published {
+			client.Log.Infof("cipd: successfully uploaded %s", sha1)
+			return nil
+		}
+		if client.clock.now().Sub(started) > CASFinalizationTimeout {
+			client.Log.Warningf("cipd: upload of %s failed: timeout", sha1)
+			return ErrFinalizationTimeout
+		}
+		client.Log.Infof("cipd: uploading - verifying")
+		client.clock.sleep(delay)
+		if delay < 4*time.Second {
+			delay += 500 * time.Millisecond
+		}
+	}
+}
+
+// RegisterInstance makes the package instance available for clients by
+// uploading it to the storage and registering it in the package repository.
+// 'instance' is a package instance to register.
+func (client *Client) RegisterInstance(instance local.PackageInstance) error {
+	// Attempt to register.
+	client.Log.Infof("cipd: registering %s", instance.Pin())
+	result, err := client.remote.registerInstance(instance.Pin())
+	if err != nil {
+		return err
+	}
+
+	// Asked to upload the package file to CAS first?
+	if result.uploadSession != nil {
+		err = client.UploadToCAS(instance.Pin().InstanceID, instance.DataReader(), result.uploadSession)
+		if err != nil {
+			return err
+		}
+		// Try again, now that file is uploaded.
+		client.Log.Infof("cipd: registering %s", instance.Pin())
+		result, err = client.remote.registerInstance(instance.Pin())
+		if err != nil {
+			return err
+		}
+		if result.uploadSession != nil {
+			return ErrBadUpload
+		}
+	}
+
+	if result.alreadyRegistered {
+		client.Log.Infof(
+			"cipd: instance %s is already registered by %s on %s",
+			instance.Pin(), result.registeredBy, result.registeredTs)
+	} else {
+		client.Log.Infof("cipd: instance %s was successfully registered", instance.Pin())
+	}
+
+	return nil
+}
+
+// AttachTagsWhenReady attaches tags to an instance retrying on "not yet processed" responses.
+func (client *Client) AttachTagsWhenReady(pin common.Pin, tags []string) error {
+	err := common.ValidatePin(pin)
+	if err != nil {
+		return err
+	}
+	if len(tags) == 0 {
+		return nil
+	}
+	for _, tag := range tags {
+		client.Log.Infof("cipd: attaching tag %s", tag)
+	}
+	deadline := client.clock.now().Add(TagAttachTimeout)
+	for client.clock.now().Before(deadline) {
+		err = client.remote.attachTags(pin, tags)
+		if err == nil {
+			client.Log.Infof("cipd: all tags attached")
+			return nil
+		}
+		if _, ok := err.(*pendingProcessingError); ok {
+			client.Log.Warningf("cipd: package instance is not ready yet - %s", err)
+			client.clock.sleep(5 * time.Second)
+		} else {
+			client.Log.Errorf("cipd: failed to attach tags - %s", err)
+			return err
+		}
+	}
+	client.Log.Errorf("cipd: failed to attach tags - deadline exceeded")
+	return ErrAttachTagsTimeout
+}
+
+// FetchInstance downloads package instance file from the repository.
+func (client *Client) FetchInstance(pin common.Pin, output io.WriteSeeker) error {
+	err := common.ValidatePin(pin)
+	if err != nil {
+		return err
+	}
+	client.Log.Infof("cipd: resolving fetch URL for %s", pin)
+	fetchInfo, err := client.remote.fetchInstance(pin)
+	if err == nil {
+		err = client.storage.download(fetchInfo.fetchURL, output)
+	}
+	if err != nil {
+		client.Log.Errorf("cipd: failed to fetch %s - %s", pin, err)
+		return err
+	}
+	client.Log.Infof("cipd: successfully fetched %s", pin)
+	return nil
+}
+
+// FetchAndDeployInstance fetches the package instance and deploys it into
+// a site root. It doesn't check whether the instance is already deployed.
+func (client *Client) FetchAndDeployInstance(root string, pin common.Pin) error {
+	err := common.ValidatePin(pin)
+	if err != nil {
+		return err
+	}
+
+	// Use temp file for storing package file. Delete it when done.
+	var instance local.PackageInstance
+	tempPath := filepath.Join(root, local.SiteServiceDir, "tmp")
+	err = os.MkdirAll(tempPath, 0777)
+	if err != nil {
+		return err
+	}
+	f, err := ioutil.TempFile(tempPath, pin.InstanceID)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		// Instance takes ownership of the file, no need to close it separately.
+		if instance == nil {
+			f.Close()
+		}
+		os.Remove(f.Name())
+	}()
+
+	// Fetch the package data to the provided storage.
+	err = client.FetchInstance(pin, f)
+	if err != nil {
+		return err
+	}
+
+	// Open the instance, verify the instance ID.
+	instance, err = local.OpenInstance(f, pin.InstanceID)
+	if err != nil {
+		return err
+	}
+	defer instance.Close()
+
+	// Deploy it. 'defer' will take care of removing the temp file if needed.
+	_, err = local.DeployInstance(root, instance)
+	return err
+}
+
+// ProcessEnsureFile parses text file that describes what should be installed
+// by EnsurePackages function. It is a text file where each line has a form:
+// <package name> <desired version>. Whitespaces are ignored. Lines that start
+// with '#' are ignored. Version can be specified as instance ID, tag or ref.
+// Will resolve tags and refs to concrete instance IDs.
+func (client *Client) ProcessEnsureFile(r io.Reader) ([]common.Pin, error) {
+	// TODO(vadimsh): Resolve tags to instance IDs.
+
+	lineNo := 0
+	makeError := func(msg string) error {
+		return fmt.Errorf("Failed to parse desired state (line %d): %s", lineNo, msg)
+	}
+
+	out := []common.Pin{}
+	scanner := bufio.NewScanner(r)
+	for scanner.Scan() {
+		lineNo++
+
+		// Split each line into words, ignore white space.
+		tokens := []string{}
+		for _, chunk := range strings.Split(scanner.Text(), " ") {
+			chunk = strings.TrimSpace(chunk)
+			if chunk != "" {
+				tokens = append(tokens, chunk)
+			}
+		}
+
+		// Skip empty lines or lines starting with '#'.
+		if len(tokens) == 0 || tokens[0][0] == '#' {
+			continue
+		}
+
+		// Each line has a format "<package name> <version>".
+		if len(tokens) != 2 {
+			return nil, makeError("expecting '<package name> <version>' line")
+		}
+		err := common.ValidatePackageName(tokens[0])
+		if err != nil {
+			return nil, makeError(err.Error())
+		}
+		err = common.ValidateInstanceID(tokens[1])
+		if err != nil {
+			return nil, makeError(err.Error())
+		}
+
+		// Good enough.
+		out = append(out, common.Pin{PackageName: tokens[0], InstanceID: tokens[1]})
+	}
+
+	return out, nil
+}
+
+// EnsurePackages is high-level interface for installation, removal and updates
+// of packages inside some installation site root. Given a description of
+// what packages (and versions) should be installed it will do all necessary
+// actions to bring the state of the site root to the desired one.
+func (client *Client) EnsurePackages(root string, pins []common.Pin) error {
+	// Make sure a package is specified only once.
+	seen := make(map[string]bool, len(pins))
+	for _, p := range pins {
+		if seen[p.PackageName] {
+			return fmt.Errorf("Package %s is specified twice", p.PackageName)
+		}
+		seen[p.PackageName] = true
+	}
+
+	// Ensure site root is a directory (or missing).
+	root, err := filepath.Abs(filepath.Clean(root))
+	if err != nil {
+		return err
+	}
+	stat, err := os.Stat(root)
+	if err == nil && !stat.IsDir() {
+		return fmt.Errorf("Path %s is not a directory", root)
+	}
+	if err != nil && !os.IsNotExist(err) {
+		return err
+	}
+	rootExists := (err == nil)
+
+	// Enumerate existing packages (only if root already exists).
+	existing := []common.Pin{}
+	if rootExists {
+		existing, err = local.FindDeployed(root)
+		if err != nil {
+			client.Log.Errorf("Failed to enumerate installed packages - %s", err)
+			return err
+		}
+	}
+
+	// Figure out what needs to be updated and deleted, log it.
+	toDeploy, toDelete := buildActionPlan(pins, existing)
+	if len(toDeploy) == 0 && len(toDelete) == 0 {
+		client.Log.Infof("Everything is up-to-date.")
+		return nil
+	}
+	if len(toDeploy) != 0 {
+		client.Log.Infof("Packages to be installed:")
+		for _, pin := range toDeploy {
+			client.Log.Infof("  %s", pin)
+		}
+	}
+	if len(toDelete) != 0 {
+		client.Log.Infof("Packages to be removed:")
+		for _, pin := range toDelete {
+			client.Log.Infof("  %s", pin)
+		}
+	}
+
+	// Create the site root directory before installing anything there.
+	if len(toDeploy) != 0 && !rootExists {
+		err = os.MkdirAll(root, 0777)
+		if err != nil {
+			return err
+		}
+	}
+
+	// Remove all unneeded stuff.
+	fail := false
+	for _, pin := range toDelete {
+		err = local.RemoveDeployed(root, pin.PackageName)
+		if err != nil {
+			client.Log.Errorf("Failed to remove %s - %s", pin.PackageName, err)
+			fail = true
+		}
+	}
+
+	// Install all new stuff.
+	for _, pin := range toDeploy {
+		err = client.FetchAndDeployInstance(root, pin)
+		if err != nil {
+			client.Log.Errorf("Failed to install %s - %s", pin, err)
+			fail = true
+		}
+	}
+
+	if !fail {
+		client.Log.Infof("All changes applied.")
+		return nil
+	}
+	return ErrEnsurePackagesFailed
+}
+
+////////////////////////////////////////////////////////////////////////////////
+// Private structs and interfaces.
+
+type clock interface {
+	now() time.Time
+	sleep(time.Duration)
+}
+
+type remote interface {
+	fetchACL(packagePath string) ([]PackageACL, error)
+	modifyACL(packagePath string, changes []PackageACLChange) error
+
+	initiateUpload(sha1 string) (*UploadSession, error)
+	finalizeUpload(sessionID string) (bool, error)
+	registerInstance(pin common.Pin) (*registerInstanceResponse, error)
+
+	attachTags(pin common.Pin, tags []string) error
+	fetchInstance(pin common.Pin) (*fetchInstanceResponse, error)
+}
+
+type storage interface {
+	upload(url string, data io.ReadSeeker) error
+	download(url string, output io.WriteSeeker) error
+}
+
+type registerInstanceResponse struct {
+	uploadSession     *UploadSession
+	alreadyRegistered bool
+	registeredBy      string
+	registeredTs      time.Time
+}
+
+type fetchInstanceResponse struct {
+	fetchURL     string
+	registeredBy string
+	registeredTs time.Time
+}
+
+// Private stuff.
+
+type clockImpl struct{}
+
+func (c *clockImpl) now() time.Time        { return time.Now() }
+func (c *clockImpl) sleep(d time.Duration) { time.Sleep(d) }
+
+// buildActionPlan is used by EnsurePackages to figure out what to install or remove.
+func buildActionPlan(desired, existing []common.Pin) (toDeploy, toDelete []common.Pin) {
+	// Figure out what needs to be installed or updated.
+	existingMap := buildInstanceIDMap(existing)
+	for _, d := range desired {
+		if existingMap[d.PackageName] != d.InstanceID {
+			toDeploy = append(toDeploy, d)
+		}
+	}
+
+	// Figure out what needs to be removed.
+	desiredMap := buildInstanceIDMap(desired)
+	for _, e := range existing {
+		if desiredMap[e.PackageName] == "" {
+			toDelete = append(toDelete, e)
+		}
+	}
+
+	return
+}
+
+// buildInstanceIDMap builds mapping {package name -> instance ID}.
+func buildInstanceIDMap(pins []common.Pin) map[string]string {
+	out := map[string]string{}
+	for _, p := range pins {
+		out[p.PackageName] = p.InstanceID
+	}
+	return out
+}