vm/proxyapp: logging over rpc

3 files changed, 110 insertions, 26 deletions

diff --git a/vm/proxyapp/proxyappclient.go b/vm/proxyapp/proxyappclient.go
index b17e045d0..e5f1fbcf4 100644
--- a/vm/proxyapp/proxyappclient.go
+++ b/vm/proxyapp/proxyappclient.go
@@ -42,11 +42,8 @@ func ctor(params *proxyAppParams, env *vmimpl.Env) (vmimpl.Pool, error) {
 			select {
 			case <-p.close:
 				p.mu.Lock()
-				if p.proxy != nil {
-					p.proxy.terminate()
-					<-p.proxy.onTerminated
-				}
-				p.proxy = nil
+				p.closeProxy()
+
 				p.onClosed <- nil
 				p.mu.Unlock()
 				return
@@ -78,16 +75,14 @@ func (p *pool) init(params *proxyAppParams, cfg *Config) error {
 		return fmt.Errorf("failed to run ProxyApp: %w", err)
 	}
 
+	p.proxy.doLogPooling(params.LogOutput)
+
 	count, err := p.proxy.CreatePool(string(cfg.ProxyAppConfig), p.env.Debug)
 	if err != nil || count == 0 || (p.count != 0 && p.count != count) {
 		if err == nil {
 			err = fmt.Errorf("wrong pool size %v, prev was %v", count, p.count)
 		}
-
-		p.proxy.terminate()
-		<-p.proxy.onTerminated
-		p.proxy = nil
-
+		p.closeProxy()
 		return fmt.Errorf("failed to construct pool: %w", err)
 	}
 
@@ -97,6 +92,23 @@ func (p *pool) init(params *proxyAppParams, cfg *Config) error {
 	return nil
 }
 
+func (p *pool) closeProxy() {
+	if p.proxy != nil {
+		if p.proxy.stopLogPooling != nil {
+			p.proxy.stopLogPooling <- true
+			<-p.proxy.logPoolingDone
+		}
+		if p.proxy.Client != nil {
+			p.proxy.Client.Close()
+		}
+		if p.proxy.terminate != nil {
+			p.proxy.terminate()
+			<-p.proxy.onTerminated
+		}
+	}
+	p.proxy = nil
+}
+
 func (p *pool) Count() int {
 	return p.count
 }
@@ -122,8 +134,10 @@ func (p *pool) Close() error {
 
 type ProxyApp struct {
 	*rpc.Client
-	terminate    context.CancelFunc
-	onTerminated chan bool
+	terminate      context.CancelFunc
+	onTerminated   chan bool
+	stopLogPooling chan bool
+	logPoolingDone chan bool
 }
 
 func runProxyApp(params *proxyAppParams, cmd string) (*ProxyApp, error) {
@@ -185,6 +199,38 @@ func runProxyApp(params *proxyAppParams, cmd string) (*ProxyApp, error) {
 	}, nil
 }
 
+func (proxy *ProxyApp) doLogPooling(writer io.Writer) {
+	proxy.stopLogPooling = make(chan bool, 1)
+	proxy.logPoolingDone = make(chan bool, 1)
+	go func() {
+		defer func() { proxy.logPoolingDone <- true }()
+		for {
+			var reply proxyrpc.PoolLogsReply
+			call := proxy.Go(
+				"ProxyVM.PoolLogs",
+				&proxyrpc.PoolLogsParam{},
+				&reply,
+				nil,
+			)
+			select {
+			case <-proxy.stopLogPooling:
+				return
+			case c := <-call.Done:
+				if c.Error != nil {
+					// possible errors here are:
+					// "unexpected EOF"
+					// "read tcp 127.0.0.1:56886->127.0.0.1:34603: use of closed network connection"
+					log.Logf(0, "error pooling ProxyApp logs: %v", c.Error)
+					return
+				}
+				if log.V(reply.Verbosity) {
+					writer.Write([]byte(fmt.Sprintf("ProxyAppLog: %v", reply.Log)))
+				}
+			}
+		}
+	}()
+}
+
 func (proxy *ProxyApp) CreatePool(config string, debug bool) (int, error) {
 	var reply proxyrpc.CreatePoolResult
 	err := proxy.Call(
diff --git a/vm/proxyapp/proxyappclient_mocks_test.go b/vm/proxyapp/proxyappclient_mocks_test.go
index 149c37c0c..e40e898b2 100644
--- a/vm/proxyapp/proxyappclient_mocks_test.go
+++ b/vm/proxyapp/proxyappclient_mocks_test.go
@@ -43,3 +43,15 @@ func (cmd *mockCommandRunner) Wait() error {
 	cmd.onWaitCalled <- true
 	return cmd.SubProcessCmd.Wait()
 }
+
+type mockProxyAppInterface struct {
+	*mocks.ProxyAppInterface
+	OnLogsReceived chan bool
+}
+
+func makeMockProxyAppInterface(t mocks.NewProxyAppInterfaceT) *mockProxyAppInterface {
+	return &mockProxyAppInterface{
+		ProxyAppInterface: mocks.NewProxyAppInterface(t),
+		OnLogsReceived:    make(chan bool, 1), // 1 is enough as we read it just once
+	}
+}
diff --git a/vm/proxyapp/proxyappclient_test.go b/vm/proxyapp/proxyappclient_test.go
index db9649672..86ce1b324 100644
--- a/vm/proxyapp/proxyappclient_test.go
+++ b/vm/proxyapp/proxyappclient_test.go
@@ -14,7 +14,6 @@ import (
 	"time"
 
 	"github.com/google/syzkaller/pkg/report"
-	"github.com/google/syzkaller/vm/proxyapp/mocks"
 	"github.com/google/syzkaller/vm/proxyapp/proxyrpc"
 	"github.com/google/syzkaller/vm/vmimpl"
 	"github.com/stretchr/testify/assert"
@@ -35,18 +34,19 @@ func makeTestParams() *proxyAppParams {
 	return &proxyAppParams{
 		CommandRunner:  osutilCommandContext,
 		InitRetryDelay: 0,
+		LogOutput:      io.Discard,
 	}
 }
 
 func makeMockProxyAppProcess(t *testing.T) (
-	*mock.Mock, io.WriteCloser, io.ReadCloser, io.ReadCloser) {
+	*mockProxyAppInterface, io.WriteCloser, io.ReadCloser, io.ReadCloser) {
 	rStdin, wStdin := io.Pipe()
 	rStdout, wStdout := io.Pipe()
 	rStderr, wStderr := io.Pipe()
 	wStderr.Close()
 
 	server := rpc.NewServer()
-	handler := mocks.NewProxyAppInterface(t)
+	handler := makeMockProxyAppInterface(t)
 	server.RegisterName("ProxyVM", struct{ proxyrpc.ProxyAppInterface }{handler})
 
 	go server.ServeCodec(jsonrpc.NewServerCodec(stdInOutCloser{
@@ -54,7 +54,7 @@ func makeMockProxyAppProcess(t *testing.T) (
 		wStdout,
 	}))
 
-	return &handler.Mock, wStdin, rStdout, rStderr
+	return handler, wStdin, rStdout, rStderr
 }
 
 type nopWriteCloser struct {
@@ -142,7 +142,10 @@ func TestCtor_FailedConstructPool(t *testing.T) {
 
 	mProxyAppServer.
 		On("CreatePool", mock.Anything, mock.Anything).
-		Return(fmt.Errorf("failed to construct pool"))
+		Return(fmt.Errorf("failed to construct pool")).
+		On("PoolLogs", mock.Anything, mock.Anything).
+		Return(nil).
+		Maybe() // on CreatePool error we close logger. This close makes PoolLogs racy.
 
 	mCmdRunner, params := makeMockCommandRunner(t)
 	mCmdRunner.
@@ -165,19 +168,37 @@ func TestCtor_FailedConstructPool(t *testing.T) {
 	assert.Nil(t, p)
 }
 
-// TODO: to remove duplicate see TestCtor_FailedConstructPool() comment
-//  nolint: dupl
-func proxyAppServerFixture(t *testing.T) (*mock.Mock, *mockCommandRunner, *proxyAppParams) {
-	mProxyAppServer, stdin, stdout, stderr :=
-		makeMockProxyAppProcess(t)
-
+func initProxyAppServerFixture(mProxyAppServer *mockProxyAppInterface) *mockProxyAppInterface {
 	mProxyAppServer.
 		On("CreatePool", mock.Anything, mock.Anything).
 		Run(func(args mock.Arguments) {
 			out := args.Get(1).(*proxyrpc.CreatePoolResult)
 			out.Count = 2
 		}).
-		Return(nil)
+		Return(nil).
+		Once().
+		On("PoolLogs", mock.Anything, mock.Anything).
+		Run(func(args mock.Arguments) {
+			select {
+			case mProxyAppServer.OnLogsReceived <- true:
+			default:
+			}
+		}).
+		Return(nil).
+		// PoolLogs is optional as we can call .closeProxy any time.
+		// If PoolLogs call is expected we are checking for OnLogsReceived.
+		// TODO: refactor it once Mock.Unset() is available.
+		Maybe()
+
+	return mProxyAppServer
+}
+
+// TODO: to remove duplicate see TestCtor_FailedConstructPool() comment
+//  nolint: dupl
+func proxyAppServerFixture(t *testing.T) (*mockProxyAppInterface, *mockCommandRunner, *proxyAppParams) {
+	mProxyAppServer, stdin, stdout, stderr :=
+		makeMockProxyAppProcess(t)
+	initProxyAppServerFixture(mProxyAppServer)
 
 	mCmdRunner, params := makeMockCommandRunner(t)
 	mCmdRunner.
@@ -200,7 +221,7 @@ func proxyAppServerFixture(t *testing.T) (*mock.Mock, *mockCommandRunner, *proxy
 	return mProxyAppServer, mCmdRunner, params
 }
 
-func poolFixture(t *testing.T) (*mock.Mock, *mockCommandRunner, vmimpl.Pool) {
+func poolFixture(t *testing.T) (*mockProxyAppInterface, *mockCommandRunner, vmimpl.Pool) {
 	mProxyAppServer, mCmdRunner, params := proxyAppServerFixture(t)
 	p, _ := ctor(params, testEnv)
 	return mProxyAppServer, mCmdRunner, p
@@ -217,6 +238,11 @@ func TestPool_Create_Ok(t *testing.T) {
 	assert.Nil(t, err)
 }
 
+func TestPool_Logs_Ok(t *testing.T) {
+	mockServer, _, _ := poolFixture(t)
+	<-mockServer.OnLogsReceived
+}
+
 func TestPool_Create_ProxyNilError(t *testing.T) {
 	_, mCmdRunner, p := poolFixture(t)
 	mCmdRunner.
@@ -272,7 +298,7 @@ func createInstanceFixture(t *testing.T) (*mock.Mock, vmimpl.Instance) {
 	assert.Nil(t, err)
 	assert.NotNil(t, inst)
 
-	return mockServer, inst
+	return &mockServer.Mock, inst
 }
 
 func TestInstance_Close(t *testing.T) {