Merge pull request #7 from halkyon/add-retrieve-command

Add support for retrieving purged logs stored in Storj DCS

Author: ethanadams

cmd/oklog/main.go

@@ -33,6 +33,7 @@ func usage() {
 	fmt.Fprintf(os.Stderr, "  query        Querying commandline tool\n")
 	fmt.Fprintf(os.Stderr, "  stream       Streaming commandline tool\n")
 	fmt.Fprintf(os.Stderr, "  testsvc      Test service, emits log lines at a fixed rate\n")
+	fmt.Fprintf(os.Stderr, "  retrieve     Retrieve commandline tool to read purged logs stored in Storj DCS")
 	fmt.Fprintf(os.Stderr, "\n")
 	fmt.Fprintf(os.Stderr, "VERSION\n")
 	fmt.Fprintf(os.Stderr, "  %s (%s)\n", version, runtime.Version())
@@ -61,6 +62,8 @@ func main() {
 		run = runStream
 	case "testsvc":
 		run = runTestService
+	case "retrieve":
+		run = runRetrieve
 	default:
 		usage()
 		os.Exit(1)

cmd/oklog/retrieve.go

@@ -0,0 +1,104 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/oklog/oklog/pkg/store"
+	"github.com/pkg/errors"
+)
+
+func runRetrieve(args []string) error {
+	flagset := flag.NewFlagSet("retrieve", flag.ExitOnError)
+	var (
+		storeAddr = flagset.String("store", "localhost:7650", "address of store instance to query")
+		from      = flagset.String("from", "1h", "from, as RFC3339 timestamp or duration ago")
+		to        = flagset.String("to", "now", "to, as RFC3339 timestamp or duration ago")
+		withulid  = flagset.Bool("ulid", false, "include ULID prefix with each record")
+		withtime  = flagset.Bool("time", false, "include time prefix with each record")
+		verbose   = flagset.Bool("v", false, "verbose output to stderr")
+	)
+	flagset.Usage = usageFor(flagset, "oklog retrieve [flags]")
+	if err := flagset.Parse(args); err != nil {
+		return err
+	}
+
+	verbosePrintf := func(string, ...interface{}) {}
+	if *verbose {
+		verbosePrintf = func(format string, args ...interface{}) {
+			fmt.Fprintf(os.Stderr, format, args...)
+		}
+	}
+
+	_, hostport, _, _, err := parseAddr(*storeAddr, defaultAPIPort)
+	if err != nil {
+		return errors.Wrap(err, "couldn't parse -store")
+	}
+
+	fromDuration, durationErr := time.ParseDuration(*from)
+	fromTime, timeErr := time.Parse(time.RFC3339Nano, *from)
+	fromNow := strings.ToLower(*from) == "now"
+	var fromStr string
+	switch {
+	case fromNow:
+		fromStr = time.Now().Format(time.RFC3339)
+	case durationErr == nil && timeErr != nil:
+		fromStr = time.Now().Add(neg(fromDuration)).Format(time.RFC3339)
+	case durationErr != nil && timeErr == nil:
+		fromStr = fromTime.Format(time.RFC3339)
+	default:
+		return fmt.Errorf("couldn't parse -from (%q) as either duration or time", *from)
+	}
+
+	toDuration, durationErr := time.ParseDuration(*to)
+	toTime, timeErr := time.Parse(time.RFC3339, *to)
+	toNow := strings.ToLower(*to) == "now"
+	var toStr string
+	switch {
+	case toNow:
+		toStr = time.Now().Format(time.RFC3339)
+	case durationErr == nil && timeErr != nil:
+		toStr = time.Now().Add(neg(toDuration)).Format(time.RFC3339)
+	case durationErr != nil && timeErr == nil:
+		toStr = toTime.Format(time.RFC3339)
+	default:
+		return fmt.Errorf("couldn't parse -to (%q) as either duration or time", *to)
+	}
+
+	req, err := http.NewRequest(http.MethodGet, fmt.Sprintf(
+		"http://%s/store%s?from=%s&to=%s",
+		hostport,
+		store.APIPathDCSQuery,
+		url.QueryEscape(fromStr),
+		url.QueryEscape(toStr),
+	), nil)
+	if err != nil {
+		return err
+	}
+	verbosePrintf("GET %s\n", req.URL.String())
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return err
+	}
+	if resp.StatusCode != http.StatusOK {
+		req.URL.RawQuery = "" // for pretty print
+		return errors.Errorf("%s %s: %s", req.Method, req.URL.String(), resp.Status)
+	}
+
+	switch {
+	case *withulid:
+		io.Copy(os.Stdout, resp.Body)
+	case *withtime:
+		io.Copy(os.Stdout, parseTime(resp.Body))
+	default:
+		io.Copy(os.Stdout, strip(resp.Body))
+	}
+
+	return nil
+}

pkg/store/api.go

@@ -30,6 +30,7 @@ const (
 	APIPathInternalStream = "/_stream"
 	APIPathReplicate      = "/replicate"
 	APIPathClusterState   = "/_clusterstate"
+	APIPathDCSQuery       = "/dcsquery"
 )
 
 // ClusterPeer models cluster.Peer.
@@ -111,6 +112,8 @@ func (a *API) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		a.handleReplicate(w, r)
 	case method == "GET" && path == APIPathClusterState:
 		a.handleClusterState(w, r)
+	case method == "GET" && path == APIPathDCSQuery:
+		a.handleDCSQuery(w, r)
 	default:
 		http.NotFound(w, r)
 	}
@@ -449,6 +452,28 @@ func (a *API) handleClusterState(w http.ResponseWriter, r *http.Request) {
 	w.Write(buf)
 }
 
+func (a *API) handleDCSQuery(w http.ResponseWriter, r *http.Request) {
+	var qp QueryParams
+	if err := qp.DecodeFrom(r.URL, rangeNotRequired); err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	log, ok := a.log.(*fileLogDCS)
+	if !ok {
+		http.Error(w, "unsupported log type, must be fileLogDCS", http.StatusNotImplemented)
+		return
+	}
+
+	lr := log.DCSReader(qp)
+
+	_, err := io.Copy(w, lr)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+}
+
 func teeRecords(src io.Reader, dst ...io.Writer) (lo, hi ulid.ULID, n int, err error) {
 	var (
 		first = true

pkg/store/file_log_dcs.go

@@ -11,7 +11,10 @@ import (
 	"storj.io/uplink"
 )
 
-const uploadTimeout = 15 * time.Second
+const (
+	uploadTimeout   = 15 * time.Second
+	downloadTimeout = 15 * time.Second
+)
 
 var (
 	_ Log          = (*fileLogDCS)(nil)
@@ -133,3 +136,80 @@ func (t fileTrashSegmentDCS) Purge() error {
 
 	return t.fileTrashSegment.Purge() // The segment is in DCS now, and we're safe to delete it.
 }
+
+func (fl *fileLogDCS) DCSReader(qp QueryParams) io.Reader {
+	pr, pw := io.Pipe()
+
+	go func() {
+		defer pw.Close()
+
+		ctx, cancel := context.WithTimeout(context.Background(), downloadTimeout)
+		defer cancel()
+
+		var totalBytes int64
+
+		// TODO: sorting the segment files from the bucket, an equivalent to sorting in queryMatchingSegments()
+		iterator := fl.project.ListObjects(ctx, fl.bucketName, nil)
+		for iterator.Next() {
+			low, high, err := parseFilename(iterator.Item().Key)
+			if err != nil {
+				fl.reportDCSWarning(err, iterator.Item().Key)
+				continue
+			}
+
+			if !overlap(qp.From.ULID, qp.To.ULID, low, high) {
+				continue
+			}
+
+			download, err := fl.project.DownloadObject(ctx, fl.bucketName, iterator.Item().Key, nil)
+			if err != nil {
+				fl.reportDCSError(err, iterator.Item().Key)
+				continue
+			}
+
+			gzDownload, err := gzip.NewReader(download)
+			if err != nil {
+				download.Close()
+				fl.reportDCSError(err, iterator.Item().Key)
+				continue
+			}
+
+			n, err := io.Copy(pw, gzDownload)
+			if err != nil {
+				gzDownload.Close()
+				download.Close()
+				fl.reportDCSError(err, iterator.Item().Key)
+				continue
+			}
+
+			gzDownload.Close()
+			download.Close()
+
+			totalBytes += n
+		}
+
+		fl.reporter.ReportEvent(Event{
+			Debug: true,
+			Op:    "DCSReader",
+			Msg:   fmt.Sprintf("Downloaded %d bytes from DCS", totalBytes),
+		})
+	}()
+
+	return pr
+}
+
+func (fl *fileLogDCS) reportDCSWarning(err error, filename string) {
+	fl.reporter.ReportEvent(Event{
+		Op:      "DCSReader",
+		File:    fmt.Sprintf("sj://%s/%s", fl.bucketName, filename),
+		Warning: err,
+	})
+}
+
+func (fl *fileLogDCS) reportDCSError(err error, filename string) {
+	fl.reporter.ReportEvent(Event{
+		Op:    "DCSReader",
+		File:  fmt.Sprintf("sj://%s/%s", fl.bucketName, filename),
+		Error: err,
+	})
+}