Add a ring-buffer reporter to libbeat

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]...)
+}