X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/0ad80a1594e583be9821feb8369e8dc4f619ab65..7af506a0e9712ca22096ebd56df8867a427dae96:/sdk/go/arvados/keep_service.go

diff --git a/sdk/go/arvados/keep_service.go b/sdk/go/arvados/keep_service.go
index a2263d1791..5b6d71a4fb 100644
--- a/sdk/go/arvados/keep_service.go
+++ b/sdk/go/arvados/keep_service.go
@@ -12,7 +12,10 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"time"
+
+	"git.arvados.org/arvados.git/sdk/go/ctxlog"
 )
 
 // KeepService is an arvados#keepService record
@@ -104,6 +107,24 @@ func (s *KeepService) Mounts(c *Client) ([]KeepMount, error) {
 	return mounts, nil
 }
 
+// Touch updates the timestamp on the given block.
+func (s *KeepService) Touch(ctx context.Context, c *Client, blk string) error {
+	req, err := http.NewRequest("TOUCH", s.url(blk), nil)
+	if err != nil {
+		return err
+	}
+	resp, err := c.Do(req.WithContext(ctx))
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		body, _ := ioutil.ReadAll(resp.Body)
+		return fmt.Errorf("%s %s: %s", resp.Proto, resp.Status, body)
+	}
+	return nil
+}
+
 // Untrash moves/copies the given block out of trash.
 func (s *KeepService) Untrash(ctx context.Context, c *Client, blk string) error {
 	req, err := http.NewRequest("PUT", s.url("untrash/"+blk), nil)
@@ -122,21 +143,21 @@ func (s *KeepService) Untrash(ctx context.Context, c *Client, blk string) error
 	return nil
 }
 
-// Index returns an unsorted list of blocks at the given mount point.
-func (s *KeepService) IndexMount(c *Client, mountUUID string, prefix string) ([]KeepServiceIndexEntry, error) {
-	return s.index(c, s.url("mounts/"+mountUUID+"/blocks?prefix="+prefix))
+// IndexMount returns an unsorted list of blocks at the given mount point.
+func (s *KeepService) IndexMount(ctx context.Context, c *Client, mountUUID string, prefix string) ([]KeepServiceIndexEntry, error) {
+	return s.index(ctx, c, prefix, s.url("mounts/"+mountUUID+"/blocks?prefix="+prefix))
 }
 
 // Index returns an unsorted list of blocks that can be retrieved from
 // this server.
-func (s *KeepService) Index(c *Client, prefix string) ([]KeepServiceIndexEntry, error) {
-	return s.index(c, s.url("index/"+prefix))
+func (s *KeepService) Index(ctx context.Context, c *Client, prefix string) ([]KeepServiceIndexEntry, error) {
+	return s.index(ctx, c, prefix, s.url("index/"+prefix))
 }
 
-func (s *KeepService) index(c *Client, url string) ([]KeepServiceIndexEntry, error) {
-	req, err := http.NewRequest("GET", url, nil)
+func (s *KeepService) index(ctx context.Context, c *Client, prefix, url string) ([]KeepServiceIndexEntry, error) {
+	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
 	if err != nil {
-		return nil, fmt.Errorf("NewRequest(%v): %v", url, err)
+		return nil, fmt.Errorf("NewRequestWithContext(%v): %v", url, err)
 	}
 	resp, err := c.Do(req)
 	if err != nil {
@@ -146,10 +167,30 @@ func (s *KeepService) index(c *Client, url string) ([]KeepServiceIndexEntry, err
 	}
 	defer resp.Body.Close()
 
+	var progress int64
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	go func() {
+		log := ctxlog.FromContext(ctx)
+		logticker := time.NewTicker(5 * time.Minute)
+		defer logticker.Stop()
+		for {
+			select {
+			case <-logticker.C:
+				log.Printf("index progress: received %d blocks from %s", atomic.LoadInt64(&progress), url)
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
 	var entries []KeepServiceIndexEntry
 	scanner := bufio.NewScanner(resp.Body)
 	sawEOF := false
 	for scanner.Scan() {
+		if ctx.Err() != nil {
+			return nil, ctx.Err()
+		}
 		if scanner.Err() != nil {
 			// If we encounter a read error (timeout,
 			// connection failure), stop now and return it
@@ -169,6 +210,9 @@ func (s *KeepService) index(c *Client, url string) ([]KeepServiceIndexEntry, err
 		if len(fields) != 2 {
 			return nil, fmt.Errorf("Malformed index line %q: %d fields", line, len(fields))
 		}
+		if !strings.HasPrefix(fields[0], prefix) {
+			return nil, fmt.Errorf("Index response included block %q despite asking for prefix %q", fields[0], prefix)
+		}
 		mtime, err := strconv.ParseInt(fields[1], 10, 64)
 		if err != nil {
 			return nil, fmt.Errorf("Malformed index line %q: mtime: %v", line, err)
@@ -185,6 +229,7 @@ func (s *KeepService) index(c *Client, url string) ([]KeepServiceIndexEntry, err
 			SizedDigest: SizedDigest(fields[0]),
 			Mtime:       mtime,
 		})
+		atomic.AddInt64(&progress, 1)
 	}
 	if err := scanner.Err(); err != nil {
 		return nil, fmt.Errorf("Error scanning index response: %v", err)