pfp: recorder: implement maxSize option alongside maxCapacity

Allow to configure the NodeRecorder with maximum capacity to avoid
extreme growth in recorded PFP statuses size. This is yet another
sealing besides maxCapacity that the node recorder should not cross.
The maxSize takes precesdence over maxCapacity, meaning as long as the
current size of the nodeRecorder does not reach the maximum allowed,
`statuses` slice length is maintained and is allowed to reach the
maximum capacity. Once the new item pushed to the nodeRecorder is
expected to make the recorder reach the max allowed size, `statuses`
will witness reductions in older items to allow the new status to fit,
under the constraint that node recorder will at least have one status no
matter its size. So the boundary is important but the recorder goal
should still be on top.

Signed-off-by: Francesco Romani <fromani@redhat.com>

Author: ffromani

pkg/pfpstatus/pfpstatus.go

@@ -41,14 +41,16 @@ const (
 const (
 	defaultMaxNodes          = 5000
 	defaultMaxSamplesPerNode = 10
+	defaultMaxSizePerNode    = 0 // no constraints
 	defaultDumpPeriod        = 10 * time.Second
 )
 
 type StorageParams struct {
-	Enabled      bool
-	Directory    string
-	Period       time.Duration
-	CoalesceLast bool
+	Enabled        bool
+	Directory      string
+	Period         time.Duration
+	CoalesceLast   bool
+	MaxSizePerNode int
 }
 
 type Params struct {
@@ -64,10 +66,11 @@ type environ struct {
 func DefaultParams() Params {
 	return Params{
 		Storage: StorageParams{
-			Enabled:      false,
-			Directory:    DefaultDumpDirectory,
-			Period:       10 * time.Second,
-			CoalesceLast: false,
+			Enabled:        false,
+			Directory:      DefaultDumpDirectory,
+			Period:         10 * time.Second,
+			CoalesceLast:   false,
+			MaxSizePerNode: defaultMaxSizePerNode,
 		},
 	}
 }
@@ -100,6 +103,7 @@ func Setup(logh logr.Logger, params Params) {
 	rec, err := record.NewRecorder(
 		record.WithMaxNodes(defaultMaxNodes),
 		record.WithNodeCapacity(defaultMaxSamplesPerNode),
+		record.WithMaxSizePerNode(defaultMaxSizePerNode),
 		record.WithPFPCoalescing(params.Storage.CoalesceLast),
 	)
 	if err != nil {

pkg/pfpstatus/record/options.go

@@ -44,6 +44,12 @@ func WithPFPCoalescing(val bool) Option {
 	}
 }
 
+func WithMaxSizePerNode(maxSize int) Option {
+	return func(rr *Recorder) {
+		rr.maxSize = maxSize
+	}
+}
+
 type NodeOption func(*NodeRecorder)
 
 func WithCapacity(capacity int) NodeOption {
@@ -63,3 +69,9 @@ func WithCoalescing(val bool) NodeOption {
 		nr.coalesceLast = val
 	}
 }
+
+func WithMaxSize(maxSize int) NodeOption {
+	return func(nr *NodeRecorder) {
+		nr.maxSize = maxSize
+	}
+}

pkg/pfpstatus/record/recorder.go

@@ -18,6 +18,8 @@ package record
 
 import (
 	"errors"
+	"fmt"
+	"io"
 	"time"
 
 	"github.com/k8stopologyawareschedwg/podfingerprint"
@@ -40,6 +42,13 @@ type RecordedStatus struct {
 	podfingerprint.Status
 	// RecordTime is a timestamp of when the RecordedStatus was added to the record
 	RecordTime time.Time `json:"recordTime"`
+	// statusSize is approximate size of the string representation of the status in bytes
+	statusSize int `json:"-"`
+}
+
+func (rs RecordedStatus) Size() int {
+	byteCount, _ := fmt.Fprint(io.Discard, fmt.Sprintf("%+v\n%+v", rs.Pods, rs.RecordTime))
+	return byteCount
 }
 
 func (rs RecordedStatus) Equal(x RecordedStatus) bool {
@@ -55,6 +64,8 @@ type NodeRecorder struct {
 	timestamper  func() time.Time
 	nodeName     string // shortcut
 	capacity     int
+	maxSize      int
+	size         int
 	statuses     []RecordedStatus
 	coalesceLast bool
 }
@@ -88,14 +99,27 @@ func (nr *NodeRecorder) dropOldest() {
 	if nr.Len() < 1 {
 		return
 	}
+	nr.size -= nr.statuses[0].Size()
 	nr.statuses = nr.statuses[1:]
 }
 
-func (nr *NodeRecorder) makeRoom() {
-	if nr.Len() < nr.Cap() {
+func (nr *NodeRecorder) makeRoom(size int) {
+	if nr.capacity > 1 && nr.Len() == nr.Cap() {
+		nr.dropOldest()
+	}
+	if nr.maxSize == 0 {
+		return
+	}
+
+	if size >= nr.maxSize {
+		nr.size = 0
+		nr.statuses = []RecordedStatus{}
 		return
 	}
-	nr.dropOldest()
+
+	for (nr.size + size) > nr.maxSize {
+		nr.dropOldest()
+	}
 }
 
 // Push adds a new Status to the record, evicting the oldest Status if necessary.
@@ -119,15 +143,19 @@ func (nr *NodeRecorder) Push(st podfingerprint.Status) error {
 	}
 
 	item := RecordedStatus{
-		Status:     st.Clone(),
+		Status:     st,
 		RecordTime: ts,
 	}
+	item.statusSize = item.Size()
 	if nr.capacity == 1 { // handle common special case, avoid any resize
 		nr.statuses[0] = item
+		nr.size = item.Size()
+		// no maxSize constraint required
 		return nil
 	}
-	nr.makeRoom()
+	nr.makeRoom(item.statusSize)
 	nr.statuses = append(nr.statuses, item)
+	nr.size += item.Size()
 	return nil
 }
 
@@ -151,12 +179,18 @@ func (nr *NodeRecorder) IsCoalescing() bool {
 	return nr.coalesceLast
 }
 
+// MaxSize returns the maximum size allowed for nr
+func (nr *NodeRecorder) MaxSize() int {
+	return nr.maxSize
+}
+
 // Recorder stores all the recorded statuses, dividing them by node name.
 // There is a hard cap of how many nodes are managed, and how many Statuses are recorded per node.
 type Recorder struct {
 	nodes        map[string]*NodeRecorder
 	nodeCapacity int
 	maxNodes     int
+	maxSize      int
 	timestamper  Timestamper
 	coalesceLast bool
 }
@@ -186,7 +220,7 @@ func NewRecorder(opts ...Option) (*Recorder, error) {
 	return &rec, nil
 }
 
-// Cap returns the maximum nodes allowed in this Recorder
+// MaxNodes returns the maximum nodes allowed in this Recorder
 func (rr *Recorder) MaxNodes() int {
 	return rr.maxNodes
 }
@@ -224,6 +258,11 @@ func (rr *Recorder) IsCoalescing() bool {
 	return rr.coalesceLast
 }
 
+// MaxSize returns the maximum size allowed per NodeRecorder in rr
+func (rr *Recorder) MaxSize() int {
+	return rr.maxSize
+}
+
 // Push adds a new Status to the record for its node, evicting the oldest Status
 // belonging to the same node if necessary.
 // Per-node records are created lazily as needed, up to the configured maximum.
@@ -243,7 +282,7 @@ func (rr *Recorder) Push(st podfingerprint.Status) error {
 	}
 
 	if !ok {
-		nr, err = NewNodeRecorder(st.NodeName, WithTimestamper(rr.timestamper), WithCapacity(rr.nodeCapacity), WithCoalescing(rr.coalesceLast))
+		nr, err = NewNodeRecorder(st.NodeName, WithTimestamper(rr.timestamper), WithCapacity(rr.nodeCapacity), WithMaxSize(rr.maxSize), WithCoalescing(rr.coalesceLast))
 		if err != nil {
 			return err
 		}

pkg/pfpstatus/record/recorder_test.go

@@ -19,7 +19,9 @@ package record
 import (
 	"errors"
 	"fmt"
+	"io"
 	"testing"
+	"time"
 
 	"github.com/k8stopologyawareschedwg/podfingerprint"
 )
@@ -441,3 +443,91 @@ func TestRecorderWithCoalescing(t *testing.T) {
 		t.Fatalf("unexpected status count")
 	}
 }
+
+func TestRecorderWithMaxSize(t *testing.T) {
+	rr, err := NewRecorder(WithMaxNodes(8), WithNodeCapacity(5), WithMaxSizePerNode(1500))
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if rr.MaxSize() != 1500 {
+		t.Fatalf("unexpected max size: %d", rr.MaxSize())
+	}
+}
+
+func TestNodeRecorderWithMaxSize(t *testing.T) {
+	rt := time.Now() // to manage accurate size checks
+	rtf := func() time.Time {
+		return rt
+	}
+
+	st := podfingerprint.Status{
+		FingerprintExpected: "pfp-exp-st1",
+		FingerprintComputed: "pfp-comp-st1",
+		NodeName:            "node-0",
+	}
+	rs := RecordedStatus{
+		Status:     st,
+		RecordTime: rt,
+	}
+	stSize := rs.Size()
+	maxSize := 2*stSize + 1000
+
+	nr, err := NewNodeRecorder("node-0", WithCapacity(5), WithTimestamper(rtf), WithMaxSize(maxSize))
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if nr.MaxSize() != maxSize {
+		t.Fatalf("unexpected max size: %d vs %d", nr.MaxSize(), maxSize)
+	}
+
+	for i := 0; i < 2; i++ {
+		if err := nr.Push(st); err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	}
+
+	stSize += rs.Size()
+	if nr.Len() != 2 {
+		t.Fatalf("unexpected length: %d", nr.Len())
+	}
+	if nr.size != stSize {
+		t.Fatalf("unexpected size: %d vs %d", nr.size, stSize)
+	}
+
+	var pods []podfingerprint.NamespacedName
+	for i := 0; getSize(pods) < maxSize; i++ {
+		pods = append(pods, podfingerprint.NamespacedName{
+			Namespace: fmt.Sprintf("pod-name %d", i),
+			Name:      fmt.Sprintf("namespace-name %d", i),
+		})
+	}
+
+	rs2 := RecordedStatus{
+		Status: podfingerprint.Status{
+			FingerprintExpected: "pfp-exp-st2",
+			FingerprintComputed: "pfp-comp-st2",
+			Pods:                pods,
+			NodeName:            st.NodeName,
+		},
+		RecordTime: rt,
+	}
+
+	if err := nr.Push(rs2.Status); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if nr.Len() != 1 {
+		t.Fatalf("unexpected length: expected 1 found %d", nr.Len())
+	}
+
+	if nr.size != rs2.Size() {
+		t.Fatalf("unexpected size: expected %d found %d", rs2.Size(), nr.size)
+	}
+}
+
+func getSize(pods []podfingerprint.NamespacedName) int {
+	byteCount, _ := fmt.Fprint(io.Discard, fmt.Sprintf("%+v\n", pods))
+	return byteCount
+
+}