Implement log retrieval

4 files changed, 252 insertions, 36 deletions

diff --git a/client/go/vespa/log.go b/client/go/vespa/log.go
new file mode 100644
index 00000000000..94505004b9c
--- /dev/null
+++ b/client/go/vespa/log.go
@@ -0,0 +1,99 @@
+package vespa
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"strconv"
+	"strings"
+	"time"
+)
+
+var dequoter = strings.NewReplacer("\\n", "\n", "\\t", "\t")
+
+// LogEntry represents a Vespa log entry.
+type LogEntry struct {
+	Time      time.Time
+	Host      string
+	Service   string
+	Component string
+	Level     string
+	Message   string
+}
+
+func (le *LogEntry) Format(dequote bool) string {
+	t := le.Time.Format("2006-01-02 15:04:05.000000")
+	msg := le.Message
+	if dequote {
+		msg = dequoter.Replace(msg)
+	}
+	return fmt.Sprintf("[%s] %-8s %-7s %-16s %s\t%s", t, le.Host, le.Level, le.Service, le.Component, msg)
+}
+
+// ParseLogEntry parses a Vespa log entry from string s.
+func ParseLogEntry(s string) (LogEntry, error) {
+	parts := strings.SplitN(s, "\t", 7)
+	if len(parts) != 7 {
+		return LogEntry{}, fmt.Errorf("invalid number of log parts: %d: %q", len(parts), s)
+	}
+	time, err := parseLogTimestamp(parts[0])
+	if err != nil {
+		return LogEntry{}, err
+	}
+	return LogEntry{
+		Time:      time,
+		Host:      parts[1],
+		Service:   parts[3],
+		Component: parts[4],
+		Level:     parts[5],
+		Message:   parts[6],
+	}, nil
+}
+
+// ReadLogEntries reads and parses all log entries from reader r.
+func ReadLogEntries(r io.Reader) ([]LogEntry, error) {
+	var entries []LogEntry
+	scanner := bufio.NewScanner(r)
+	for scanner.Scan() {
+		line := scanner.Text()
+		logEntry, err := ParseLogEntry(line)
+		if err != nil {
+			return nil, err
+		}
+		entries = append(entries, logEntry)
+	}
+	if err := scanner.Err(); err != nil {
+		return nil, err
+	}
+	return entries, nil
+}
+
+// LogLevel returns an int representing a named log level.
+func LogLevel(name string) int {
+	switch name {
+	case "error":
+		return 0
+	case "warning":
+		return 1
+	case "info":
+		return 2
+	default: // everything else, e.g. debug
+		return 3
+	}
+}
+
+func parseLogTimestamp(s string) (time.Time, error) {
+	parts := strings.Split(s, ".")
+	if len(parts) != 2 {
+		return time.Time{}, fmt.Errorf("invalid number of log timestamp parts: %d", len(parts))
+	}
+	unixSecs, err := strconv.ParseInt(parts[0], 10, 64)
+	if err != nil {
+		return time.Time{}, fmt.Errorf("invalid timestamp seconds: %s", parts[0])
+	}
+	unixMicros, err := strconv.ParseInt(parts[1], 10, 64)
+	if err != nil {
+		return time.Time{}, fmt.Errorf("invalid timestamp microseconds: %s", parts[1])
+	}
+	return time.Unix(unixSecs, unixMicros*1000).UTC(), nil
+}
diff --git a/client/go/vespa/log_test.go b/client/go/vespa/log_test.go
new file mode 100644
index 00000000000..f8e66db0474
--- /dev/null
+++ b/client/go/vespa/log_test.go
@@ -0,0 +1,31 @@
+package vespa
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestParseLogEntry(t *testing.T) {
+	expected := LogEntry{
+		Time:      time.Date(2021, 9, 27, 10, 31, 30, 905535000, time.UTC),
+		Host:      "host1a.dev.aws-us-east-1c",
+		Service:   "logserver-container",
+		Component: "Container.com.yahoo.container.jdisc.ConfiguredApplication",
+		Level:     "info",
+		Message:   "Switching to the latest deployed set of configurations and components. Application config generation: 52532",
+	}
+	in := "1632738690.905535	host1a.dev.aws-us-east-1c	806/53	logserver-container	Container.com.yahoo.container.jdisc.ConfiguredApplication	info	Switching to the latest deployed set of configurations and components. Application config generation: 52532"
+	logEntry, err := ParseLogEntry(in)
+	assert.Nil(t, err)
+	assert.Equal(t, expected, logEntry)
+
+	formatted := "[2021-09-27 10:31:30.905535] host1a.dev.aws-us-east-1c info    logserver-container Container.com.yahoo.container.jdisc.ConfiguredApplication\tSwitching to the latest deployed set of configurations and components. Application config generation: 52532"
+	assert.Equal(t, formatted, logEntry.Format(false))
+
+	in = "1632738690.905535	host1a.dev.aws-us-east-1c	806/53	logserver-container	Container.com.yahoo.container.jdisc.ConfiguredApplication	info	message containing newline\\nand\\ttab"
+	logEntry, err = ParseLogEntry(in)
+	assert.Nil(t, err)
+	assert.Equal(t, "[2021-09-27 10:31:30.905535] host1a.dev.aws-us-east-1c info    logserver-container Container.com.yahoo.container.jdisc.ConfiguredApplication\tmessage containing newline\nand\ttab", logEntry.Format(true))
+}
diff --git a/client/go/vespa/target.go b/client/go/vespa/target.go
index aa0ddb8babb..f05d109aeaa 100644
--- a/client/go/vespa/target.go
+++ b/client/go/vespa/target.go
@@ -1,11 +1,13 @@
 package vespa
 
 import (
+	"bytes"
 	"crypto/tls"
 	"encoding/json"
 	"fmt"
 	"io"
 	"io/ioutil"
+	"math"
 	"net/http"
 	"net/url"
 	"sort"
@@ -41,6 +43,9 @@ type Target interface {
 
 	// Service returns the service for given name. If timeout is non-zero, wait for the service to converge.
 	Service(name string, timeout time.Duration, sessionOrRunID int64) (*Service, error)
+
+	// PrintLog writes the logs of this deployment using given options to control output.
+	PrintLog(options LogOptions) error
 }
 
 // TLSOptions configures the certificate to use for service requests.
@@ -50,10 +55,14 @@ type TLSOptions struct {
 	PrivateKeyFile  string
 }
 
-// LogOptions configures the log output to produce when waiting for services.
+// LogOptions configures the log output to produce when writing log messages.
 type LogOptions struct {
-	Writer io.Writer
-	Level  int
+	From    time.Time
+	To      time.Time
+	Follow  bool
+	Dequote bool
+	Writer  io.Writer
+	Level   int
 }
 
 type customTarget struct {
@@ -119,6 +128,10 @@ func (t *customTarget) Service(name string, timeout time.Duration, sessionID int
 	return nil, fmt.Errorf("unknown service: %s", name)
 }
 
+func (t *customTarget) PrintLog(options LogOptions) error {
+	return fmt.Errorf("reading logs from non-cloud deployment is currently unsupported")
+}
+
 func (t *customTarget) urlWithPort(serviceName string) (string, error) {
 	u, err := url.Parse(t.baseURL)
 	if err != nil {
@@ -207,6 +220,64 @@ func (t *cloudTarget) Service(name string, timeout time.Duration, runID int64) (
 	return nil, fmt.Errorf("unknown service: %s", name)
 }
 
+func (t *cloudTarget) logsURL() string {
+	return fmt.Sprintf("%s/application/v4/tenant/%s/application/%s/instance/%s/environment/%s/region/%s/logs",
+		t.apiURL,
+		t.deployment.Application.Tenant, t.deployment.Application.Application, t.deployment.Application.Instance,
+		t.deployment.Zone.Environment, t.deployment.Zone.Region)
+}
+
+func (t *cloudTarget) PrintLog(options LogOptions) error {
+	req, err := http.NewRequest("GET", t.logsURL(), nil)
+	if err != nil {
+		return err
+	}
+	signer := NewRequestSigner(t.deployment.Application.SerializedForm(), t.apiKey)
+	lastFrom := options.From
+	requestFunc := func() *http.Request {
+		fromMillis := lastFrom.Unix() * 1000
+		q := req.URL.Query()
+		q.Set("from", strconv.FormatInt(fromMillis, 10))
+		if !options.To.IsZero() {
+			toMillis := options.To.Unix() * 1000
+			q.Set("to", strconv.FormatInt(toMillis, 10))
+		}
+		req.URL.RawQuery = q.Encode()
+		if err := signer.SignRequest(req); err != nil {
+			panic(err)
+		}
+		return req
+	}
+	logFunc := func(status int, response []byte) (bool, error) {
+		if ok, err := isOK(status); !ok {
+			return ok, err
+		}
+		logEntries, err := ReadLogEntries(bytes.NewReader(response))
+		if err != nil {
+			return true, err
+		}
+		for _, le := range logEntries {
+			if !le.Time.After(lastFrom) {
+				continue
+			}
+			if LogLevel(le.Level) > options.Level {
+				continue
+			}
+			fmt.Fprintln(options.Writer, le.Format(options.Dequote))
+		}
+		if len(logEntries) > 0 {
+			lastFrom = logEntries[len(logEntries)-1].Time
+		}
+		return false, nil
+	}
+	var timeout time.Duration
+	if options.Follow {
+		timeout = math.MaxInt64 // No timeout
+	}
+	_, err = wait(logFunc, requestFunc, &t.tlsOptions.KeyPair, timeout)
+	return err
+}
+
 func (t *cloudTarget) waitForEndpoints(timeout time.Duration, runID int64) error {
 	signer := NewRequestSigner(t.deployment.Application.SerializedForm(), t.apiKey)
 	if runID > 0 {
@@ -348,20 +419,6 @@ func CloudTarget(apiURL string, deployment Deployment, apiKey []byte, tlsOptions
 	}
 }
 
-// LogLevel returns an int representing a named log level.
-func LogLevel(name string) int {
-	switch name {
-	case "error":
-		return 0
-	case "warning":
-		return 1
-	case "info":
-		return 2
-	default: // everything else, e.g. debug
-		return 3
-	}
-}
-
 type deploymentEndpoint struct {
 	URL string `json:"url"`
 }
diff --git a/client/go/vespa/target_test.go b/client/go/vespa/target_test.go
index 2c90baefbbc..b5f967b935a 100644
--- a/client/go/vespa/target_test.go
+++ b/client/go/vespa/target_test.go
@@ -4,6 +4,8 @@ import (
 	"bytes"
 	"crypto/tls"
 	"fmt"
+	"io"
+	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
 	"testing"
@@ -40,6 +42,11 @@ func (v *mockVespaApi) mockVespaHandler(w http.ResponseWriter, req *http.Request
 	case "/application/v2/tenant/default/application/default/environment/prod/region/default/instance/default/serviceconverge":
 		response := fmt.Sprintf(`{"converged": %t}`, v.deploymentConverged)
 		w.Write([]byte(response))
+	case "/application/v4/tenant/t1/application/a1/instance/i1/environment/dev/region/us-north-1/logs":
+		log := `1632738690.905535	host1a.dev.aws-us-east-1c	806/53	logserver-container	Container.com.yahoo.container.jdisc.ConfiguredApplication	info	Switching to the latest deployed set of configurations and components. Application config generation: 52532
+1632738698.600189	host1a.dev.aws-us-east-1c	1723/33590	config-sentinel	sentinel.sentinel.config-owner	config	Sentinel got 3 service elements [tenant(vespa-team), application(music), instance(mpolden)] for config generation 52532
+`
+		w.Write([]byte(log))
 	case "/status.html":
 		w.Write([]byte("OK"))
 	case "/ApplicationStatus":
@@ -92,6 +99,44 @@ func TestCloudTargetWait(t *testing.T) {
 	defer srv.Close()
 	vc.serverURL = srv.URL
 
+	var logWriter bytes.Buffer
+	target := createCloudTarget(t, srv.URL, &logWriter)
+	assertServiceWait(t, 200, target, "deploy")
+
+	_, err := target.Service("query", time.Millisecond, 42)
+	assert.NotNil(t, err)
+
+	vc.deploymentConverged = true
+	_, err = target.Service("query", time.Millisecond, 42)
+	assert.Nil(t, err)
+
+	assertServiceWait(t, 500, target, "query")
+	assertServiceWait(t, 500, target, "document")
+
+	// Log timestamp is converted to local time, do the same here in case the local time where tests are run varies
+	tm := time.Unix(1631707708, 431000)
+	expectedTime := tm.Format("[15:04:05]")
+	assert.Equal(t, expectedTime+" info    Deploying platform version 7.465.17 and application version 1.0.2 ...\n", logWriter.String())
+}
+
+func TestLog(t *testing.T) {
+	vc := mockVespaApi{}
+	srv := httptest.NewServer(http.HandlerFunc(vc.mockVespaHandler))
+	defer srv.Close()
+	vc.serverURL = srv.URL
+	vc.deploymentConverged = true
+
+	var buf bytes.Buffer
+	target := createCloudTarget(t, srv.URL, ioutil.Discard)
+	if err := target.PrintLog(LogOptions{Writer: &buf, Level: 3}); err != nil {
+		t.Fatal(err)
+	}
+	expected := "[2021-09-27 10:31:30.905535] host1a.dev.aws-us-east-1c info    logserver-container Container.com.yahoo.container.jdisc.ConfiguredApplication\tSwitching to the latest deployed set of configurations and components. Application config generation: 52532\n" +
+		"[2021-09-27 10:31:38.600189] host1a.dev.aws-us-east-1c config  config-sentinel  sentinel.sentinel.config-owner\tSentinel got 3 service elements [tenant(vespa-team), application(music), instance(mpolden)] for config generation 52532\n"
+	assert.Equal(t, expected, buf.String())
+}
+
+func createCloudTarget(t *testing.T, url string, logWriter io.Writer) Target {
 	kp, err := CreateKeyPair()
 	assert.Nil(t, err)
 
@@ -100,7 +145,6 @@ func TestCloudTargetWait(t *testing.T) {
 	apiKey, err := CreateAPIKey()
 	assert.Nil(t, err)
 
-	var logWriter bytes.Buffer
 	target := CloudTarget(
 		"https://example.com",
 		Deployment{
@@ -109,28 +153,13 @@ func TestCloudTargetWait(t *testing.T) {
 		},
 		apiKey,
 		TLSOptions{KeyPair: x509KeyPair},
-		LogOptions{Writer: &logWriter})
+		LogOptions{Writer: logWriter})
 	if ct, ok := target.(*cloudTarget); ok {
-		ct.apiURL = srv.URL
+		ct.apiURL = url
 	} else {
 		t.Fatalf("Wrong target type %T", ct)
 	}
-	assertServiceWait(t, 200, target, "deploy")
-
-	_, err = target.Service("query", time.Millisecond, 42)
-	assert.NotNil(t, err)
-
-	vc.deploymentConverged = true
-	_, err = target.Service("query", time.Millisecond, 42)
-	assert.Nil(t, err)
-
-	assertServiceWait(t, 500, target, "query")
-	assertServiceWait(t, 500, target, "document")
-
-	// Log timestamp is converted to local time, do the same here in case the local time where tests are run varies
-	tm := time.Unix(1631707708, 431000)
-	expectedTime := tm.Format("[15:04:05]")
-	assert.Equal(t, expectedTime+" info    Deploying platform version 7.465.17 and application version 1.0.2 ...\n", logWriter.String())
+	return target
 }
 
 func assertServiceURL(t *testing.T, url string, target Target, service string) {