Send structured data over the streaming log api.

Author: howbazaar

api/client.go

@@ -12,6 +12,7 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/juju/errors"
 	"github.com/juju/loggo"
@@ -593,25 +594,47 @@ func (args DebugLogParams) URLQuery() url.Values {
 	return attrs
 }
 
-// WatchDebugLog returns a ReadCloser that the caller can read the log
-// lines from. Only log lines that match the filtering specified in
-// the DebugLogParams are returned. It returns an error that satisfies
-// errors.IsNotImplemented when the API server does not support the
-// end-point.
-func (c *Client) WatchDebugLog(args DebugLogParams) (io.ReadCloser, error) {
-	// The websocket connection just hangs if the server doesn't have the log
-	// end point. So do a version check, as version was added at the same time
-	// as the remote end point.
-	_, err := c.AgentVersion()
-	if err != nil {
-		return nil, errors.NotSupportedf("WatchDebugLog")
-	}
+// LogMessage is a structured logging entry.
+type LogMessage struct {
+	Entity    string
+	Timestamp time.Time
+	Severity  string
+	Module    string
+	Location  string
+	Message   string
+}
+
+// WatchDebugLog returns a channel of structured Log Messages. Only log entries
+// that match the filtering specified in the DebugLogParams are returned.
+func (c *Client) WatchDebugLog(args DebugLogParams) (<-chan LogMessage, error) {
 	// Prepare URL query attributes.
 	attrs := args.URLQuery()
 
 	connection, err := c.st.ConnectStream("/log", attrs)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
-	return connection, nil
+
+	messages := make(chan LogMessage)
+	go func() {
+		defer close(messages)
+
+		for {
+			var msg params.LogMessage
+			err := connection.ReadJSON(&msg)
+			if err != nil {
+				return
+			}
+			messages <- LogMessage{
+				Entity:    msg.Entity,
+				Timestamp: msg.Timestamp,
+				Severity:  msg.Severity,
+				Module:    msg.Module,
+				Location:  msg.Location,
+				Message:   msg.Message,
+			}
+		}
+	}()
+
+	return messages, nil
 }

api/client_test.go

@@ -4,7 +4,6 @@
 package api_test
 
 import (
-	"bufio"
 	"bytes"
 	"fmt"
 	"io"
@@ -357,10 +356,9 @@ func (s *clientSuite) TestWatchDebugLogConnected(c *gc.C) {
 	// Use the no tail option so we don't try to start a tailing cursor
 	// on the oplog when there is no oplog configured in mongo as the tests
 	// don't set up mongo in replicaset mode.
-	reader, err := client.WatchDebugLog(api.DebugLogParams{NoTail: true})
+	messages, err := client.WatchDebugLog(api.DebugLogParams{NoTail: true})
 	c.Assert(err, jc.ErrorIsNil)
-	c.Assert(reader, gc.NotNil)
-	reader.Close()
+	c.Assert(messages, gc.NotNil)
 }
 
 func (s *clientSuite) TestConnectStreamRequiresSlashPathPrefix(c *gc.C) {
@@ -407,7 +405,8 @@ func (s *clientSuite) TestConnectStreamErrorReadError(c *gc.C) {
 }
 
 func (s *clientSuite) TestWatchDebugLogParamsEncoded(c *gc.C) {
-	s.PatchValue(api.WebsocketDialConfig, echoURL(c))
+	catcher := urlCatcher{}
+	s.PatchValue(api.WebsocketDialConfig, catcher.recordLocation)
 
 	params := api.DebugLogParams{
 		IncludeEntity: []string{"a", "b"},
@@ -422,10 +421,10 @@ func (s *clientSuite) TestWatchDebugLogParamsEncoded(c *gc.C) {
 	}
 
 	client := s.APIState.Client()
-	reader, err := client.WatchDebugLog(params)
+	_, err := client.WatchDebugLog(params)
 	c.Assert(err, jc.ErrorIsNil)
 
-	connectURL := connectURLFromReader(c, reader)
+	connectURL := catcher.location
 	values := connectURL.Query()
 	c.Assert(values, jc.DeepEquals, url.Values{
 		"includeEntity": params.IncludeEntity,
@@ -441,18 +440,18 @@ func (s *clientSuite) TestWatchDebugLogParamsEncoded(c *gc.C) {
 }
 
 func (s *clientSuite) TestConnectStreamAtUUIDPath(c *gc.C) {
-	s.PatchValue(api.WebsocketDialConfig, echoURL(c))
-	// If the server supports it, we should log at "/model/UUID/log"
+	catcher := urlCatcher{}
+	s.PatchValue(api.WebsocketDialConfig, catcher.recordLocation)
 	environ, err := s.State.Model()
 	c.Assert(err, jc.ErrorIsNil)
 	info := s.APIInfo(c)
 	info.ModelTag = environ.ModelTag()
 	apistate, err := api.Open(info, api.DialOpts{})
 	c.Assert(err, jc.ErrorIsNil)
 	defer apistate.Close()
-	reader, err := apistate.ConnectStream("/path", nil)
+	_, err = apistate.ConnectStream("/path", nil)
 	c.Assert(err, jc.ErrorIsNil)
-	connectURL := connectURLFromReader(c, reader)
+	connectURL := catcher.location
 	c.Assert(connectURL.Path, gc.Matches, fmt.Sprintf("/model/%s/path", environ.UUID()))
 }
 
@@ -529,25 +528,17 @@ func (r *badReader) Read(p []byte) (n int, err error) {
 	return 0, r.err
 }
 
-func echoURL(c *gc.C) func(*websocket.Config) (base.Stream, error) {
-	return func(config *websocket.Config) (base.Stream, error) {
-		pr, pw := io.Pipe()
-		go func() {
-			fmt.Fprintf(pw, "null\n")
-			fmt.Fprintf(pw, "%s\n", config.Location)
-		}()
-		return fakeStreamReader{pr}, nil
-	}
+type urlCatcher struct {
+	location *url.URL
 }
 
-func connectURLFromReader(c *gc.C, rc io.ReadCloser) *url.URL {
-	bufReader := bufio.NewReader(rc)
-	location, err := bufReader.ReadString('\n')
-	c.Assert(err, jc.ErrorIsNil)
-	connectURL, err := url.Parse(strings.TrimSpace(location))
-	c.Assert(err, jc.ErrorIsNil)
-	rc.Close()
-	return connectURL
+func (u *urlCatcher) recordLocation(config *websocket.Config) (base.Stream, error) {
+	u.location = config.Location
+	pr, pw := io.Pipe()
+	go func() {
+		fmt.Fprintf(pw, "null\n")
+	}()
+	return fakeStreamReader{pr}, nil
 }
 
 type fakeStreamReader struct {
@@ -562,13 +553,13 @@ func (s fakeStreamReader) Close() error {
 }
 
 func (s fakeStreamReader) Write([]byte) (int, error) {
-	panic("not implemented")
+	return 0, errors.NotImplementedf("Write")
 }
 
 func (s fakeStreamReader) ReadJSON(v interface{}) error {
-	panic("not implemented")
+	return errors.NotImplementedf("ReadJSON")
 }
 
 func (s fakeStreamReader) WriteJSON(v interface{}) error {
-	panic("not implemented")
+	return errors.NotImplementedf("WriteJSON")
 }

api/state_macaroon_test.go

@@ -60,7 +60,8 @@ func (s *macaroonLoginSuite) TestUnknownUserLogin(c *gc.C) {
 }
 
 func (s *macaroonLoginSuite) TestConnectStream(c *gc.C) {
-	s.PatchValue(api.WebsocketDialConfig, echoURL(c))
+	catcher := urlCatcher{}
+	s.PatchValue(api.WebsocketDialConfig, catcher.recordLocation)
 
 	dischargeCount := 0
 	s.DischargerLogin = func() string {
@@ -77,13 +78,14 @@ func (s *macaroonLoginSuite) TestConnectStream(c *gc.C) {
 	conn, err := s.client.ConnectStream("/path", nil)
 	c.Assert(err, gc.IsNil)
 	defer conn.Close()
-	connectURL := connectURLFromReader(c, conn)
+	connectURL := catcher.location
 	c.Assert(connectURL.Path, gc.Equals, "/model/"+s.State.ModelTag().Id()+"/path")
 	c.Assert(dischargeCount, gc.Equals, 1)
 }
 
 func (s *macaroonLoginSuite) TestConnectStreamWithoutLogin(c *gc.C) {
-	s.PatchValue(api.WebsocketDialConfig, echoURL(c))
+	catcher := urlCatcher{}
+	s.PatchValue(api.WebsocketDialConfig, catcher.recordLocation)
 
 	conn, err := s.client.ConnectStream("/path", nil)
 	c.Assert(err, gc.ErrorMatches, `cannot use ConnectStream without logging in`)

apiserver/debuglog.go

@@ -4,7 +4,6 @@
 package apiserver
 
 import (
-	"io"
 	"net"
 	"net/http"
 	"net/url"
@@ -71,7 +70,7 @@ func (h *debugLogHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	server := websocket.Server{
 		Handler: func(conn *websocket.Conn) {
 			socket := &debugLogSocketImpl{conn}
-			defer socket.Close()
+			defer conn.Close()
 
 			logger.Infof("debug log handler starting")
 			// Validate before authenticate because the authentication is
@@ -111,20 +110,21 @@ func isBrokenPipe(err error) bool {
 // debugLogSocket describes the functionality required for the
 // debuglog handlers to send logs to the client.
 type debugLogSocket interface {
-	io.Writer
-
 	// sendOk sends a nil error response, indicating there were no errors.
 	sendOk()
 
 	// sendError sends a JSON-encoded error response.
 	sendError(err error)
+
+	// sendLogRecord sends record JSON encoded.
+	sendLogRecord(record *params.LogMessage) error
 }
 
 // debugLogSocketImpl implements the debugLogSocket interface. It
 // wraps a websocket.Conn and provides a few debug-log specific helper
 // methods.
 type debugLogSocketImpl struct {
-	*websocket.Conn
+	conn *websocket.Conn
 }
 
 // sendOk implements debugLogSocket.
@@ -134,11 +134,15 @@ func (s *debugLogSocketImpl) sendOk() {
 
 // sendError implements debugLogSocket.
 func (s *debugLogSocketImpl) sendError(err error) {
-	sendJSON(s.Conn, &params.ErrorResult{
+	sendJSON(s.conn, &params.ErrorResult{
 		Error: common.ServerError(err),
 	})
 }
 
+func (s *debugLogSocketImpl) sendLogRecord(record *params.LogMessage) error {
+	return sendJSON(s.conn, record)
+}
+
 // debugLogParams contains the parsed debuglog API request parameters.
 type debugLogParams struct {
 	maxLines      uint

apiserver/debuglog_db.go

@@ -4,12 +4,11 @@
 package apiserver
 
 import (
-	"fmt"
 	"net/http"
-	"time"
 
 	"github.com/juju/errors"
 
+	"github.com/juju/juju/apiserver/params"
 	"github.com/juju/juju/state"
 )
 
@@ -43,9 +42,7 @@ func handleDebugLogDBRequest(
 				return errors.Annotate(tailer.Err(), "tailer stopped")
 			}
 
-			line := formatLogRecord(rec)
-			_, err := socket.Write([]byte(line))
-			if err != nil {
+			if err := socket.sendLogRecord(formatLogRecord(rec)); err != nil {
 				return errors.Annotate(err, "sending failed")
 			}
 
@@ -73,19 +70,15 @@ func makeLogTailerParams(reqParams *debugLogParams) *state.LogTailerParams {
 	return params
 }
 
-func formatLogRecord(r *state.LogRecord) string {
-	return fmt.Sprintf("%s: %s %s %s %s %s\n",
-		r.Entity,
-		formatTime(r.Time),
-		r.Level.String(),
-		r.Module,
-		r.Location,
-		r.Message,
-	)
-}
-
-func formatTime(t time.Time) string {
-	return t.In(time.UTC).Format("2006-01-02 15:04:05")
+func formatLogRecord(r *state.LogRecord) *params.LogMessage {
+	return &params.LogMessage{
+		Entity:    r.Entity.String(),
+		Timestamp: r.Time,
+		Severity:  r.Level.String(),
+		Module:    r.Module,
+		Location:  r.Location,
+		Message:   r.Message,
+	}
 }
 
 var newLogTailer = _newLogTailer // For replacing in tests

apiserver/debuglog_db_internal_test.go

@@ -12,6 +12,7 @@ import (
 	gc "gopkg.in/check.v1"
 	"gopkg.in/juju/names.v2"
 
+	"github.com/juju/juju/apiserver/params"
 	"github.com/juju/juju/state"
 	coretesting "github.com/juju/juju/testing"
 )
@@ -244,7 +245,17 @@ func (s *fakeDebugLogSocket) sendError(err error) {
 	s.writes <- fmt.Sprintf("err: %v", err)
 }
 
-func (s *fakeDebugLogSocket) Write(buf []byte) (int, error) {
-	s.writes <- string(buf)
-	return len(buf), nil
+func (s *fakeDebugLogSocket) sendLogRecord(r *params.LogMessage) error {
+	s.writes <- fmt.Sprintf("%s: %s %s %s %s %s\n",
+		r.Entity,
+		s.formatTime(r.Timestamp),
+		r.Severity,
+		r.Module,
+		r.Location,
+		r.Message)
+	return nil
+}
+
+func (c *fakeDebugLogSocket) formatTime(t time.Time) string {
+	return t.In(time.UTC).Format("2006-01-02 15:04:05")
 }

apiserver/httpcontext.go

@@ -180,14 +180,15 @@ func (ctxt *httpContext) stop() <-chan struct{} {
 
 // sendJSON writes a JSON-encoded response value
 // to the given writer along with a trailing newline.
-func sendJSON(w io.Writer, response interface{}) {
+func sendJSON(w io.Writer, response interface{}) error {
 	body, err := json.Marshal(response)
 	if err != nil {
 		logger.Errorf("cannot marshal JSON result %#v: %v", response, err)
-		return
+		return err
 	}
 	body = append(body, '\n')
-	w.Write(body)
+	_, err = w.Write(body)
+	return err
 }
 
 // sendStatusAndJSON sends an HTTP status code and

apiserver/params/internal.go

@@ -707,3 +707,13 @@ type GUIVersionRequest struct {
 	// Version holds the Juju GUI version number.
 	Version version.Number `json:"version"`
 }
+
+// LogMessage is a structured logging entry.
+type LogMessage struct {
+	Entity    string    `json:"tag"`
+	Timestamp time.Time `json:"ts"`
+	Severity  string    `json:"sev"`
+	Module    string    `json:"mod"`
+	Location  string    `json:"loc"`
+	Message   string    `json:"msg"`
+}