Extracted the combined recorder from rpc.Conn

rpc.Conn now accepts a RecorderFactory rather than an ObserverFactory. The Recorder interface has the same methods as Observer, but they can return errors to stop the handling of the request, which is needed for auditing. I've kept the Observer interface, since the lack of errors makes multiplexing simpler. The observer is now embedded in a combined recorder that forwards messages to it but also passes them on to the auditlog recorder, which has the opportunity to interrupt the request.
sdg0919 · Dec 11, 2017 · bae76b8 · bae76b8
1 parent c388ea9
commit bae76b8
Show file tree

Hide file tree

Showing 12 changed files with 157 additions and 129 deletions.
diff --git a/api/apiclient.go b/api/apiclient.go
@@ -31,7 +31,6 @@ import (
 	"gopkg.in/retry.v1"
 
 	"github.com/juju/juju/api/base"
-	"github.com/juju/juju/apiserver/observer"
 	"github.com/juju/juju/apiserver/params"
 	"github.com/juju/juju/network"
 	"github.com/juju/juju/rpc"
@@ -196,7 +195,7 @@ func Open(info *Info, opts DialOpts) (Connection, error) {
 		return nil, errors.Trace(err)
 	}
 
-	client := rpc.NewConn(jsoncodec.New(dialResult.conn), observer.None())
+	client := rpc.NewConn(jsoncodec.New(dialResult.conn), nil)
 	client.Start()
 
 	bakeryClient := opts.BakeryClient

diff --git a/api/testing/fakeserver.go b/api/testing/fakeserver.go
@@ -17,7 +17,7 @@ func FakeAPIServer(root interface{}) net.Conn {
 	c0, c1 := net.Pipe()
 	serverCodec := jsoncodec.NewNet(c1)
 	serverRPC := rpc.NewConn(serverCodec, nil)
-	serverRPC.Serve(root, nil)
+	serverRPC.Serve(root, nil, nil)
 	serverRPC.Start()
 	go func() {
 		<-serverRPC.Dead()

diff --git a/apiserver/admin.go b/apiserver/admin.go
@@ -136,10 +136,10 @@ func (a *admin) login(req params.LoginRequest, loginVersion int) (params.LoginRe
 		modelTag = a.root.model.Tag().String()
 	}
 
-	var recorder *auditlog.Recorder
+	var auditRecorder *auditlog.Recorder
 	if authResult.userLogin {
 		// We only audit connections from humans.
-		recorder, err = auditlog.NewRecorder(
+		auditRecorder, err = auditlog.NewRecorder(
 			a.srv.auditLogger,
 			auditlog.ConversationArgs{
 				Who:          req.AuthTag,
@@ -155,7 +155,10 @@ func (a *admin) login(req params.LoginRequest, loginVersion int) (params.LoginRe
 		}
 	}
 
-	a.root.rpcConn.ServeRoot(apiRoot, recorder, serverError)
+	recorderFactory := observer.NewRecorderFactory(
+		a.apiObserver, auditRecorder)
+
+	a.root.rpcConn.ServeRoot(apiRoot, recorderFactory, serverError)
 	return params.LoginResult{
 		Servers:       params.FromNetworkHostsPorts(hostPorts),
 		ControllerTag: a.root.model.ControllerTag().String(),

diff --git a/apiserver/apiserver.go b/apiserver/apiserver.go
@@ -967,7 +967,7 @@ func (srv *Server) serveConn(
 	host string,
 ) error {
 	codec := jsoncodec.NewWebsocket(wsConn.Conn)
-	conn := rpc.NewConn(codec, apiObserver)
+	conn := rpc.NewConn(codec, observer.NewRecorderFactory(apiObserver, nil))
 
 	// Note that we don't overwrite modelUUID here because
 	// newAPIHandler treats an empty modelUUID as signifying

diff --git a/apiserver/export_test.go b/apiserver/export_test.go
@@ -111,7 +111,7 @@ func TestingAPIHandler(c *gc.C, pool *state.StatePool, st *state.State) (*apiHan
 		statePool:     pool,
 		tag:           names.NewMachineTag("0"),
 	}
-	h, err := newAPIHandler(srv, st, nil, st.ModelUUID(), "testing.invalid:1234")
+	h, err := newAPIHandler(srv, st, nil, st.ModelUUID(), 6543, "testing.invalid:1234")
 	c.Assert(err, jc.ErrorIsNil)
 	return h, h.getResources()
 }

diff --git a/apiserver/observer/recorder.go b/apiserver/observer/recorder.go
@@ -0,0 +1,83 @@
+// Copyright 2017 Canonical Ltd.
+// Licensed under the AGPLv3, see LICENCE file for details.
+
+package observer
+
+import (
+	"encoding/json"
+
+	"github.com/juju/errors"
+
+	"github.com/juju/juju/core/auditlog"
+	"github.com/juju/juju/rpc"
+)
+
+// NewRecorderFactory makes a new rpc.RecorderFactory to make
+// recorders that that will update the observer and the auditlog
+// recorder when it records a request or reply. The auditlog recorder
+// can be nil.
+func NewRecorderFactory(observerFactory rpc.ObserverFactory, recorder *auditlog.Recorder) rpc.RecorderFactory {
+	return func() rpc.Recorder {
+		return &combinedRecorder{
+			observer: observerFactory.RPCObserver(),
+			recorder: recorder,
+		}
+	}
+}
+
+// combinedRecorder wraps an observer (which might be a multiplexer)
+// up with an auditlog recorder into an rpc.Recorder.
+type combinedRecorder struct {
+	observer rpc.Observer
+	recorder *auditlog.Recorder
+}
+
+// ServerRequest implements rpc.Recorder.
+func (cr *combinedRecorder) ServerRequest(hdr *rpc.Header, body interface{}) error {
+	cr.observer.ServerRequest(hdr, body)
+	if cr.recorder == nil {
+		return nil
+	}
+	// TODO(babbageclunk): make this configurable.
+	jsonArgs, err := json.Marshal(body)
+	if err != nil {
+		return errors.Trace(err)
+	}
+	return errors.Trace(cr.recorder.AddRequest(auditlog.RequestArgs{
+		RequestID: hdr.RequestId,
+		Facade:    hdr.Request.Type,
+		Method:    hdr.Request.Action,
+		Version:   hdr.Request.Version,
+		Args:      string(jsonArgs),
+	}))
+}
+
+// ServerReply implements rpc.Recorder.
+func (cr *combinedRecorder) ServerReply(req rpc.Request, replyHdr *rpc.Header, body interface{}) error {
+	cr.observer.ServerReply(req, replyHdr, body)
+	if cr.recorder == nil {
+		return nil
+	}
+	var responseErrors []*auditlog.Error
+	if replyHdr.Error == "" {
+		var err error
+		responseErrors, err = extractErrors(body)
+		if err != nil {
+			return errors.Trace(err)
+		}
+	} else {
+		responseErrors = []*auditlog.Error{{
+			Message: replyHdr.Error,
+			Code:    replyHdr.ErrorCode,
+		}}
+	}
+	return errors.Trace(cr.recorder.AddResponse(auditlog.ResponseErrorsArgs{
+		RequestID: replyHdr.RequestId,
+		Errors:    responseErrors,
+	}))
+}
+
+func extractErrors(body interface{}) ([]*auditlog.Error, error) {
+	// TODO(babbageclunk): use reflection to find errors in the response body.
+	return nil, nil
+}
diff --git a/apiserver/testing/fakeapi.go b/apiserver/testing/fakeapi.go
@@ -11,6 +11,7 @@ import (
 	"github.com/gorilla/websocket"
 	"github.com/juju/utils"
 
+	"github.com/juju/juju/apiserver/observer"
 	"github.com/juju/juju/apiserver/observer/fakeobserver"
 	"github.com/juju/juju/rpc"
 	"github.com/juju/juju/rpc/jsoncodec"
@@ -77,7 +78,7 @@ func (srv *Server) serveAPI(w http.ResponseWriter, req *http.Request) {
 
 func (srv *Server) serveConn(wsConn *websocket.Conn, modelUUID string) {
 	codec := jsoncodec.NewWebsocket(wsConn)
-	conn := rpc.NewConn(codec, &fakeobserver.Instance{})
+	conn := rpc.NewConn(codec, observer.NewRecorderFactory(&fakeobserver.Instance{}, nil))
 
 	root := allVersions{
 		rpcreflect.ValueOf(reflect.ValueOf(srv.newRoot(modelUUID))),

diff --git a/core/auditlog/auditlog.go b/core/auditlog/auditlog.go
@@ -138,9 +138,6 @@ func NewRecorder(log AuditLog, c ConversationArgs) (*Recorder, error) {
 
 // AddRequest records a method call to the API.
 func (r *Recorder) AddRequest(m RequestArgs) error {
-	if r == nil {
-		return nil
-	}
 	return errors.Trace(r.log.AddRequest(Request{
 		ConversationID: r.callID,
 		ConnectionID:   r.connectionID,
@@ -154,9 +151,6 @@ func (r *Recorder) AddRequest(m RequestArgs) error {
 
 // AddResponse records the result of a method call to the API.
 func (r *Recorder) AddResponse(m ResponseErrorsArgs) error {
-	if r == nil {
-		return nil
-	}
 	return errors.Trace(r.log.AddResponse(ResponseErrors{
 		ConversationID: r.callID,
 		ConnectionID:   r.connectionID,

diff --git a/rpc/dispatch_test.go b/rpc/dispatch_test.go
@@ -32,9 +32,9 @@ func (s *dispatchSuite) SetUpSuite(c *gc.C) {
 	s.BaseSuite.SetUpSuite(c)
 	rpcServer := func(ws *websocket.Conn) {
 		codec := jsoncodec.NewWebsocket(ws)
-		conn := rpc.NewConn(codec, &notifier{})
+		conn := rpc.NewConn(codec, nil)
 
-		conn.Serve(&DispatchRoot{}, nil)
+		conn.Serve(&DispatchRoot{}, nil, nil)
 		conn.Start()
 
 		<-conn.Dead()

diff --git a/rpc/observers.go b/rpc/observers.go
@@ -28,6 +28,14 @@ type Observer interface {
 	ServerReply(req Request, hdr *Header, body interface{})
 }
 
+// ObserverFactory is a type which can construct a new Observer.
+type ObserverFactory interface {
+	// RPCObserver will return a new Observer usually constructed
+	// from the state previously built up in the Observer. The
+	// returned instance will be utilized per RPC request.
+	RPCObserver() Observer
+}
+
 // NewObserverMultiplexer returns a new ObserverMultiplexer
 // with the provided RequestNotifiers.
 func NewObserverMultiplexer(rpcObservers ...Observer) *ObserverMultiplexer {

diff --git a/rpc/rpc_test.go b/rpc/rpc_test.go
@@ -237,11 +237,11 @@ func (a *CallbackMethods) Factorial(x int64val) (int64val, error) {
 }
 
 func (a *ChangeAPIMethods) ChangeAPI() {
-	a.r.conn.Serve(&changedAPIRoot{}, nil)
+	a.r.conn.Serve(&changedAPIRoot{}, nil, nil)
 }
 
 func (a *ChangeAPIMethods) RemoveAPI() {
-	a.r.conn.Serve(nil, nil)
+	a.r.conn.Serve(nil, nil, nil)
 }
 
 type changedAPIRoot struct{}
@@ -987,7 +987,7 @@ func (*rpcSuite) TestBidirectional(c *gc.C) {
 	client, srvDone, _ := newRPCClientServer(c, srvRoot, nil, true)
 	defer closeClient(c, client, srvDone)
 	clientRoot := &Root{conn: client}
-	client.Serve(clientRoot, nil)
+	client.Serve(clientRoot, nil, nil)
 	var r int64val
 	err := client.Call(rpc.Request{"CallbackMethods", 0, "", "Factorial"}, int64val{12}, &r)
 	c.Assert(err, jc.ErrorIsNil)
@@ -1105,12 +1105,13 @@ func newRPCClientServer(
 		if bidir {
 			role = roleBoth
 		}
-		rpcConn := rpc.NewConn(NewJSONCodec(conn, role), serverNotifier)
+		recorderFactory := func() rpc.Recorder { return serverNotifier }
+		rpcConn := rpc.NewConn(NewJSONCodec(conn, role), recorderFactory)
 		if custroot, ok := root.(*CustomRoot); ok {
-			rpcConn.ServeRoot(custroot, nil, tfErr)
+			rpcConn.ServeRoot(custroot, recorderFactory, tfErr)
 			custroot.root.conn = rpcConn
 		} else {
-			rpcConn.Serve(root, tfErr)
+			rpcConn.Serve(root, recorderFactory, tfErr)
 		}
 		if root, ok := root.(*Root); ok {
 			root.conn = rpcConn
@@ -1125,9 +1126,9 @@ func newRPCClientServer(
 	if bidir {
 		role = roleBoth
 	}
-	client = rpc.NewConn(NewJSONCodec(conn, role), &notifier{})
+	client = rpc.NewConn(NewJSONCodec(conn, role), nil)
 	client.Start()
-	return client, srvDone, serverNotifier
+	return client, server, srvDone, serverNotifier
 }
 
 func closeClient(c *gc.C, client *rpc.Conn, srvDone <-chan error) {
@@ -1225,37 +1226,30 @@ type notifier struct {
 	serverReplies  []replyEvent
 }
 
-func (n *notifier) RPCObserver() rpc.Observer {
-	// For testing, we usually won't want an actual copy of the
-	// stub. To avoid confusing test failures (e.g. wondering why your
-	// calls aren't showing up on your stub because the underlying
-	// code has called DeepCopy) and immense complexity, just return
-	// the same value.
-	return n
-}
-
 func (n *notifier) reset() {
 	n.mu.Lock()
 	defer n.mu.Unlock()
 	n.serverRequests = nil
 	n.serverReplies = nil
 }
 
-func (n *notifier) ServerRequest(hdr *rpc.Header, body interface{}) {
+func (n *notifier) ServerRequest(hdr *rpc.Header, body interface{}) error {
 	n.mu.Lock()
 	defer n.mu.Unlock()
 	n.serverRequests = append(n.serverRequests, requestEvent{
 		hdr:  *hdr,
 		body: body,
 	})
+	return nil
 }
 
-func (n *notifier) ServerReply(req rpc.Request, hdr *rpc.Header, body interface{}) {
+func (n *notifier) ServerReply(req rpc.Request, hdr *rpc.Header, body interface{}) error {
 	n.mu.Lock()
 	defer n.mu.Unlock()
 	n.serverReplies = append(n.serverReplies, replyEvent{
 		req:  req,
 		hdr:  *hdr,
 		body: body,
 	})
+	return nil
 }