improve monitor performance

.changelog/10368.txt

@@ -0,0 +1,3 @@
+```release-note:improvement
+monitoring: optimize the monitoring endpoint to avoid losing logs when under high load.
+```

agent/agent_endpoint.go

@@ -5,6 +5,7 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-memdb"
@@ -1202,6 +1203,9 @@ func (s *HTTPHandlers) AgentMonitor(resp http.ResponseWriter, req *http.Request)
 	// a gzip stream it will go ahead and write out the HTTP response header
 	resp.Write([]byte(""))
 	flusher.Flush()
+	const flushDelay = 200 * time.Millisecond
+	flushTicker := time.NewTicker(flushDelay)
+	defer flushTicker.Stop()
 
 	// Stream logs until the connection is closed.
 	for {
@@ -1211,9 +1215,13 @@ func (s *HTTPHandlers) AgentMonitor(resp http.ResponseWriter, req *http.Request)
 			if droppedCount > 0 {
 				s.agent.logger.Warn("Dropped logs during monitor request", "dropped_count", droppedCount)
 			}
+			flusher.Flush()
 			return nil, nil
+
 		case log := <-logsCh:
 			fmt.Fprint(resp, string(log))
+
+		case <-flushTicker.C:
 			flusher.Flush()
 		}
 	}

logging/monitor/monitor.go

@@ -80,9 +80,7 @@ func (d *monitor) Stop() int {
 // received log messages over the returned channel.
 func (d *monitor) Start() <-chan []byte {
 	// register our sink with the logger
-	d.logger.RegisterSink(d.sink)
 	streamCh := make(chan []byte, d.bufSize)
-
 	// run a go routine that listens for streamed
 	// log messages and sends them to streamCh
 	go func() {
@@ -101,7 +99,7 @@ func (d *monitor) Start() <-chan []byte {
 			}
 		}
 	}()
-
+	d.logger.RegisterSink(d.sink)
 	return streamCh
 }