Move github.com/elastic/beats/v7/libbeat/monitoring/report/buffer (#42)

Author: kvch

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

monitoring/report/buffer/buffer_test.go

@@ -0,0 +1,146 @@
+// 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 (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_ringBuffer(t *testing.T) {
+	t.Run("Len 2 buffer", func(t *testing.T) {
+		r := newBuffer(2)
+		assert.Equal(t, 2, len(r.entries))
+		assert.False(t, r.full)
+		assert.Equal(t, 0, r.i)
+
+		assert.Empty(t, r.getAll())
+
+		r.add("1")
+		assert.False(t, r.full)
+		assert.Equal(t, 1, r.i)
+		assert.Equal(t, r.entries[0], "1")
+		assert.ElementsMatch(t, []string{"1"}, r.getAll())
+
+		r.add("2")
+		assert.True(t, r.full)
+		assert.Equal(t, 0, r.i)
+		assert.Equal(t, r.entries[1], "2")
+		assert.ElementsMatch(t, []string{"1", "2"}, r.getAll())
+
+		r.add("3")
+		assert.True(t, r.full)
+		assert.Equal(t, 1, r.i)
+		assert.Equal(t, r.entries[0], "3")
+		assert.ElementsMatch(t, []string{"2", "3"}, r.getAll())
+
+		r.add("4")
+		assert.True(t, r.full)
+		assert.Equal(t, 0, r.i)
+		assert.Equal(t, r.entries[1], "4")
+		assert.ElementsMatch(t, []string{"3", "4"}, r.getAll())
+	})
+
+	t.Run("Len 3 buffer", func(t *testing.T) {
+		r := newBuffer(3)
+		assert.Empty(t, r.getAll())
+
+		r.add("1")
+		assert.ElementsMatch(t, []string{"1"}, r.getAll())
+
+		r.add("2")
+		assert.ElementsMatch(t, []string{"1", "2"}, r.getAll())
+
+		r.add("3")
+		assert.ElementsMatch(t, []string{"1", "2", "3"}, r.getAll())
+
+		r.add("4")
+		assert.ElementsMatch(t, []string{"2", "3", "4"}, r.getAll())
+
+		r.add("5")
+		assert.ElementsMatch(t, []string{"3", "4", "5"}, r.getAll())
+
+		r.add("6")
+		assert.ElementsMatch(t, []string{"4", "5", "6"}, r.getAll())
+	})
+}
+
+func Benchmark_ringBuffer_add(b *testing.B) {
+	b.Run("size 6", func(b *testing.B) {
+		r := newBuffer(6)
+		for i := 0; i < b.N; i++ {
+			r.add(i)
+		}
+	})
+	b.Run("size 60", func(b *testing.B) {
+		r := newBuffer(60)
+		for i := 0; i < b.N; i++ {
+			r.add(i)
+		}
+	})
+	b.Run("size 600", func(b *testing.B) {
+		r := newBuffer(600)
+		for i := 0; i < b.N; i++ {
+			r.add(i)
+		}
+	})
+	b.Run("size 6000", func(b *testing.B) {
+		r := newBuffer(6000)
+		for i := 0; i < b.N; i++ {
+			r.add(i)
+		}
+	})
+}
+
+func Benchmark_ringBuffer_add_filled(b *testing.B) {
+	b.Run("size 6", func(b *testing.B) {
+		r := newFullBuffer(b, 6)
+		for i := 0; i < b.N; i++ {
+			r.add(i)
+		}
+	})
+	b.Run("size 60", func(b *testing.B) {
+		r := newFullBuffer(b, 60)
+		for i := 0; i < b.N; i++ {
+			r.add(i)
+		}
+	})
+	b.Run("size 600", func(b *testing.B) {
+		r := newFullBuffer(b, 600)
+		for i := 0; i < b.N; i++ {
+			r.add(i)
+		}
+	})
+	b.Run("size 6000", func(b *testing.B) {
+		r := newFullBuffer(b, 6000)
+		for i := 0; i < b.N; i++ {
+			r.add(i)
+		}
+	})
+}
+
+func newFullBuffer(b *testing.B, size int) *ringBuffer {
+	b.Helper()
+	r := newBuffer(size)
+	// fill size +1 so full flag is toggled
+	for i := 0; i < size+1; i++ {
+		r.add(-1)
+	}
+	return r
+}

monitoring/report/buffer/config.go

@@ -0,0 +1,37 @@
+// 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 (
+	"time"
+)
+
+type config struct {
+	Period     time.Duration `config:"period"`
+	Size       int           `config:"size" validate:"min=2"`
+	Namespaces []string      `config:"namespaces"`
+}
+
+// defaultConfig will gather 10m of data (every 10s) for the stats registry.
+func defaultConfig() config {
+	return config{
+		Period:     10 * time.Second,
+		Size:       60,
+		Namespaces: []string{"stats"},
+	}
+}

monitoring/report/buffer/reporter.go

@@ -0,0 +1,117 @@
+// 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 (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"sync"
+	"time"
+
+	c "github.com/elastic/elastic-agent-libs/config"
+	"github.com/elastic/elastic-agent-libs/monitoring"
+)
+
+// reporter is a struct that will fill a ring buffer for each monitored registry.
+type reporter struct {
+	config
+	wg         sync.WaitGroup
+	done       chan struct{}
+	registries map[string]*monitoring.Registry
+
+	// ring buffers for namespaces
+	entries map[string]*ringBuffer
+}
+
+// MakeReporter creates and starts a reporter with the given config.
+func MakeReporter(cfg *c.C) (*reporter, error) {
+	config := defaultConfig()
+	if cfg != nil {
+		if err := cfg.Unpack(&config); err != nil {
+			return nil, err
+		}
+	}
+
+	r := &reporter{
+		config:     config,
+		done:       make(chan struct{}),
+		registries: map[string]*monitoring.Registry{},
+		entries:    map[string]*ringBuffer{},
+	}
+
+	for _, ns := range r.config.Namespaces {
+		reg := monitoring.GetNamespace(ns).GetRegistry()
+		r.registries[ns] = reg
+		r.entries[ns] = newBuffer(r.config.Size)
+	}
+
+	r.wg.Add(1)
+	go func() {
+		defer r.wg.Done()
+		r.snapshotLoop()
+	}()
+	return r, nil
+}
+
+// Stop will stop the reporter from collecting new information.
+// It will not clear any previously collected data.
+func (r *reporter) Stop() {
+	close(r.done)
+	r.wg.Wait()
+}
+
+// snapshotLoop will collect a snapshot for each monitored registry for the configured period and store them in the correct buffer.
+func (r *reporter) snapshotLoop() {
+	ticker := time.NewTicker(r.config.Period)
+	defer ticker.Stop()
+
+	for {
+		var ts time.Time
+		select {
+		case <-r.done:
+			return
+		case ts = <-ticker.C:
+		}
+
+		for name, reg := range r.registries {
+			snap := monitoring.CollectStructSnapshot(reg, monitoring.Full, false)
+			if _, ok := snap["@timestamp"]; !ok {
+				snap["@timestamp"] = ts.UTC()
+			}
+			r.entries[name].add(snap)
+		}
+	}
+}
+
+// ServeHTTP is an http.Handler that will respond with the monitored registries buffer's contents in JSON.
+func (r *reporter) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	resp := make(map[string][]interface{}, len(r.entries))
+	for name, entries := range r.entries {
+		resp[name] = entries.getAll()
+	}
+
+	p, err := json.Marshal(resp)
+	if err != nil {
+		w.WriteHeader(500)
+		fmt.Fprintf(w, "Unable to encode JSON response: %v", err)
+		return
+	}
+	w.Header().Set("Content-Type", "application/json; charset=utf-8")
+	_, _ = w.Write(p)
+}