get local rpc endpoint working

api/agent_test.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	"github.com/hashicorp/go-uuid"
 	"github.com/hashicorp/nomad/api/internal/testutil"
 	"github.com/stretchr/testify/assert"
 )
@@ -261,7 +262,7 @@ func TestAgent_Health(t *testing.T) {
 	assert.True(health.Server.Ok)
 }
 
-func TestAgent_Monitor(t *testing.T) {
+func TestAgent_MonitorServer(t *testing.T) {
 	t.Parallel()
 	c, s := makeClient(t, nil, nil)
 	defer s.Stop()
@@ -269,7 +270,40 @@ func TestAgent_Monitor(t *testing.T) {
 	agent := c.Agent()
 
 	doneCh := make(chan struct{})
-	logCh, err := agent.Monitor("debug", doneCh, nil)
+	logCh, err := agent.Monitor("debug", "", doneCh, nil)
+	defer close(doneCh)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	// make a request to generate some logs
+	_, err = agent.Region()
+	require.NoError(t, err)
+
+	// Wait for the first log message and validate it
+	for {
+		select {
+		case log := <-logCh:
+			if log == " " {
+				return
+			}
+			require.Contains(t, log, "[DEBUG]")
+		case <-time.After(10 * time.Second):
+			require.Fail(t, "failed to get a log message")
+		}
+	}
+}
+func TestAgent_MonitorWithNode(t *testing.T) {
+	t.Parallel()
+	c, s := makeClient(t, nil, nil)
+	defer s.Stop()
+
+	agent := c.Agent()
+	id, _ := uuid.GenerateUUID()
+
+	doneCh := make(chan struct{})
+	// todo need to create or stub a nodeid?
+	logCh, err := agent.Monitor("debug", id, doneCh, nil)
 	defer close(doneCh)
 	if err != nil {
 		t.Fatalf("err: %v", err)

client/monitor_endpoint.go

@@ -23,13 +23,13 @@ type Monitor struct {
 
 func NewMonitorEndpoint(c *Client) *Monitor {
 	m := &Monitor{c: c}
-	m.c.streamingRpcs.Register("Client.Monitor", m.monitor)
+	m.c.streamingRpcs.Register("Agent.Monitor", m.monitor)
 	return m
 }
 
 func (m *Monitor) monitor(conn io.ReadWriteCloser) {
 	defer metrics.MeasureSince([]string{"client", "monitor", "monitor"}, time.Now())
-	// defer conn.Close()
+	defer conn.Close()
 
 	// Decode arguments
 	var req cstructs.MonitorRequest
@@ -62,8 +62,6 @@ func (m *Monitor) monitor(conn io.ReadWriteCloser) {
 		return
 	}
 
-	// var buf bytes.Buffer
-
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
@@ -96,11 +94,12 @@ OUTER:
 		select {
 		case log := <-streamWriter.logCh:
 			var resp cstructs.StreamErrWrapper
+
+			resp.Payload = log
 			if err := encoder.Encode(resp); err != nil {
 				streamErr = err
 				break OUTER
 			}
-			resp.Payload = []byte(log)
 			encoder.Reset(conn)
 		case <-ctx.Done():
 			break OUTER
@@ -117,7 +116,7 @@ OUTER:
 type streamWriter struct {
 	sync.Mutex
 	logs         []string
-	logCh        chan string
+	logCh        chan []byte
 	index        int
 	droppedCount int
 }
@@ -125,7 +124,7 @@ type streamWriter struct {
 func newStreamWriter(buf int) *streamWriter {
 	return &streamWriter{
 		logs:  make([]string, buf),
-		logCh: make(chan string, buf),
+		logCh: make(chan []byte, buf),
 		index: 0,
 	}
 }
@@ -136,16 +135,16 @@ func (d *streamWriter) Write(p []byte) (n int, err error) {
 
 	// Strip off newlines at the end if there are any since we store
 	// individual log lines in the agent.
-	n = len(p)
-	if p[n-1] == '\n' {
-		p = p[:n-1]
-	}
+	// n = len(p)
+	// if p[n-1] == '\n' {
+	// 	p = p[:n-1]
+	// }
 
 	d.logs[d.index] = string(p)
 	d.index = (d.index + 1) % len(d.logs)
 
 	select {
-	case d.logCh <- string(p):
+	case d.logCh <- p:
 	default:
 		d.droppedCount++
 	}

client/rpc.go

@@ -236,7 +236,6 @@ func (c *Client) setupClientRpcServer(server *rpc.Server) {
 	server.Register(c.endpoints.ClientStats)
 	server.Register(c.endpoints.FileSystem)
 	server.Register(c.endpoints.Allocations)
-	server.Register(c.endpoints.Monitor)
 }
 
 // rpcConnListener is a long lived function that listens for new connections

command/agent/agent_endpoint.go

@@ -195,11 +195,11 @@ func (s *HTTPServer) AgentMonitor(resp http.ResponseWriter, req *http.Request) (
 		var handler structs.StreamingRpcHandler
 		var handlerErr error
 		if useLocalClient {
-			handler, handlerErr = s.agent.Client().StreamingRpcHandler("Client.Monitor")
+			handler, handlerErr = s.agent.Client().StreamingRpcHandler("Agent.Monitor")
 		} else if useClientRPC {
-			handler, handlerErr = s.agent.Client().RemoteStreamingRpcHandler("Client.Monitor")
+			handler, handlerErr = s.agent.Client().RemoteStreamingRpcHandler("Agent.Monitor")
 		} else if useServerRPC {
-			handler, handlerErr = s.agent.Server().StreamingRpcHandler("Client.Monitor")
+			handler, handlerErr = s.agent.Server().StreamingRpcHandler("Agent.Monitor")
 		} else {
 			handlerErr = CodedError(400, "No local Node and node_id not provided")
 		}

command/agent/agent_endpoint_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"io/ioutil"
 	"net"
 	"net/http"
 	"net/http/httptest"
@@ -318,21 +319,26 @@ func TestHTTP_AgentMonitor(t *testing.T) {
 			// fully set up
 			maxLogAttempts := 10
 			tried := 0
+			out := ""
 			testutil.WaitForResult(func() (bool, error) {
 				if tried < maxLogAttempts {
 					s.Server.logger.Debug("log that should not be sent")
 					s.Server.logger.Warn("log that should be sent")
 					tried++
 				}
+				output, err := ioutil.ReadAll(resp.Body)
+				if err != nil {
+					return false, err
+				}
 
-				got := resp.Body.String()
+				out += string(output)
 				want := "[WARN]  http: log that should be sent"
-				if strings.Contains(got, want) {
+				if strings.Contains(out, want) {
 					require.NotContains(t, resp.Body.String(), "[DEBUG]")
 					return true, nil
 				}
 
-				return false, fmt.Errorf("missing expected log, got: %v, want: %v", got, want)
+				return false, fmt.Errorf("missing expected log, got: %v, want: %v", out, want)
 			}, func(err error) {
 				require.Fail(t, err.Error())
 			})