Add a ring-buffer reporter to libbeat (#28750)

* Add a ring-buffer reporter to libbeat

Add a ring buffer reporter that when enabled will store configured
namespaces in a buffer to allow operators to view recent metrics
history. Defaults are to gather the stats namespace every 10s for 10m.
Must be explicitly enabled, along with monitoring, and the HTTP
endpoint. The buffer endpoint is intended to be used for diagnostics
reporting.

* Add basic benchmarks

* Review feddback

libbeat/api/server.go

@@ -18,6 +18,7 @@
 package api
 
 import (
+	"errors"
 	"fmt"
 	"net"
 	"net/http"
@@ -72,6 +73,25 @@ func (s *Server) Stop() error {
 	return s.l.Close()
 }
 
+// AttachHandler will attach a handler at the specified route and return an error instead of panicing.
+func (s *Server) AttachHandler(route string, h http.Handler) (err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			switch r := r.(type) {
+			case error:
+				err = r
+			case string:
+				err = errors.New(r)
+			default:
+				err = fmt.Errorf("handle attempted to panic with %v", r)
+			}
+		}
+	}()
+	s.log.Infof("Attempting to attach %q to server.", route)
+	s.mux.Handle(route, h)
+	return
+}
+
 func parse(host string, port int) (string, string, error) {
 	url, err := url.Parse(host)
 	if err != nil {

libbeat/api/server_test.go

@@ -20,6 +20,7 @@ package api
 import (
 	"context"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"net"
 	"net/http"
@@ -184,3 +185,39 @@ func simpleMux() *http.ServeMux {
 	})
 	return mux
 }
+
+func TestAttachHandler(t *testing.T) {
+	url := "http://localhost:0"
+
+	cfg := common.MustNewConfigFrom(map[string]interface{}{
+		"host": url,
+	})
+
+	s, err := New(nil, simpleMux(), cfg)
+	require.NoError(t, err)
+	go s.Start()
+	defer s.Stop()
+
+	h := &testHandler{}
+
+	err = s.AttachHandler("/test", h)
+	require.NoError(t, err)
+
+	r, err := http.Get("http://" + s.l.Addr().String() + "/test")
+	require.NoError(t, err)
+	defer r.Body.Close()
+
+	body, err := io.ReadAll(r.Body)
+	require.NoError(t, err)
+
+	assert.Equal(t, "test!", string(body))
+
+	err = s.AttachHandler("/test", h)
+	assert.NotNil(t, err)
+}
+
+type testHandler struct{}
+
+func (t *testHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	fmt.Fprintf(w, "test!")
+}

libbeat/cmd/instance/beat.go

@@ -62,6 +62,7 @@ import (
 	"github.com/elastic/beats/v7/libbeat/metric/system/host"
 	"github.com/elastic/beats/v7/libbeat/monitoring"
 	"github.com/elastic/beats/v7/libbeat/monitoring/report"
+	"github.com/elastic/beats/v7/libbeat/monitoring/report/buffer"
 	"github.com/elastic/beats/v7/libbeat/monitoring/report/log"
 	"github.com/elastic/beats/v7/libbeat/outputs"
 	"github.com/elastic/beats/v7/libbeat/outputs/elasticsearch"
@@ -105,6 +106,7 @@ type beatConfig struct {
 	// beat internal components configurations
 	HTTP            *common.Config         `config:"http"`
 	HTTPPprof       *common.Config         `config:"http.pprof"`
+	BufferConfig    *common.Config         `config:"http.buffer"`
 	Path            paths.Path             `config:"path"`
 	Logging         *common.Config         `config:"logging"`
 	MetricLogging   *common.Config         `config:"logging.metrics"`
@@ -437,8 +439,9 @@ func (b *Beat) launch(settings Settings, bt beat.Creator) error {
 	// Start the API Server before the Seccomp lock down, we do this so we can create the unix socket
 	// set the appropriate permission on the unix domain file without having to whitelist anything
 	// that would be set at runtime.
+	var s *api.Server // buffer reporter may need to attach to the server.
 	if b.Config.HTTP.Enabled() {
-		s, err := api.NewWithDefaultRoutes(logp.NewLogger(""), b.Config.HTTP, monitoring.GetNamespace)
+		s, err = api.NewWithDefaultRoutes(logp.NewLogger(""), b.Config.HTTP, monitoring.GetNamespace)
 		if err != nil {
 			return errw.Wrap(err, "could not start the HTTP server for the API")
 		}
@@ -474,6 +477,19 @@ func (b *Beat) launch(settings Settings, bt beat.Creator) error {
 		defer reporter.Stop()
 	}
 
+	// only collect into a ring buffer if HTTP, and the ring buffer are explicitly enabled
+	if b.Config.HTTP.Enabled() && monitoring.IsBufferEnabled(b.Config.BufferConfig) {
+		buffReporter, err := buffer.MakeReporter(b.Info, b.Config.BufferConfig)
+		if err != nil {
+			return err
+		}
+		defer buffReporter.Stop()
+
+		if err := s.AttachHandler("/buffer", buffReporter); err != nil {
+			return err
+		}
+	}
+
 	ctx, cancel := context.WithCancel(context.Background())
 	var stopBeat = func() {
 		b.Instrumentation.Tracer().Close()

libbeat/monitoring/monitoring.go

@@ -108,3 +108,18 @@ func IsEnabled(monitoringCfg *common.Config) bool {
 
 	return monitoringCfg.Enabled()
 }
+
+// IsBufferEnabled will check if the monitoring buffer is explicitly enabled.
+func IsBufferEnabled(monitoringCfg *common.Config) bool {
+	if monitoringCfg == nil {
+		return false
+	}
+	fields := monitoringCfg.GetFields()
+	for _, field := range fields {
+		if field == "enabled" {
+			// default Enabled will return true, so we only return the value if it's defined.
+			return monitoringCfg.Enabled()
+		}
+	}
+	return false
+}

libbeat/monitoring/monitoring_test.go

@@ -0,0 +1,64 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package monitoring
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/beats/v7/libbeat/common"
+)
+
+func TestIsBufferEnabled(t *testing.T) {
+	tests := []struct {
+		name   string
+		input  map[string]interface{}
+		expect bool
+	}{{
+		name: "enabled",
+		input: map[string]interface{}{
+			"enabled": true,
+		},
+		expect: true,
+	}, {
+		name: "disabled",
+		input: map[string]interface{}{
+			"enabled": false,
+		},
+		expect: false,
+	}, {
+		name: "missing",
+		input: map[string]interface{}{
+			"size": 10,
+		},
+		expect: false,
+	}, {
+		name:   "nil",
+		input:  nil,
+		expect: false,
+	}}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cfg, err := common.NewConfigFrom(tt.input)
+			require.NoError(t, err)
+			assert.Equal(t, tt.expect, IsBufferEnabled(cfg))
+		})
+	}
+}

libbeat/monitoring/report/buffer/buffer.go

@@ -0,0 +1,65 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package buffer
+
+import "sync"
+
+// ringBuffer is a buffer with a fixed number of items that can be tracked.
+//
+// We assume that the size of the buffer is greater than one.
+// the buffer should be thread-safe.
+type ringBuffer struct {
+	mu      sync.Mutex
+	entries []interface{}
+	i       int
+	full    bool
+}
+
+// newBuffer returns a reference to a new ringBuffer with set size.
+func newBuffer(size int) *ringBuffer {
+	return &ringBuffer{
+		entries: make([]interface{}, size),
+	}
+}
+
+// add will add the passed entry to the buffer.
+func (r *ringBuffer) add(entry interface{}) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.entries[r.i] = entry
+	r.i = (r.i + 1) % len(r.entries)
+	if r.i == 0 {
+		r.full = true
+	}
+}
+
+// getAll returns all entries in the buffer in order
+func (r *ringBuffer) getAll() []interface{} {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.i == 0 && !r.full {
+		return []interface{}{}
+	}
+	if !r.full {
+		return r.entries[:r.i]
+	}
+	if r.full && r.i == 0 {
+		return r.entries
+	}
+	return append(r.entries[r.i:], r.entries[:r.i]...)
+}