Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions packages/api/internal/orchestrator/nodemanager/mock.go
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,13 @@ func WithFeatureFlags(ff *featureflags.Client) TestOptions {
}
}

// WithAllocatedMemoryBytes sets the initial allocated memory metric for the test node
func WithAllocatedMemoryBytes(bytes uint64) TestOptions {
return func(node *TestNode) {
node.metrics.MemoryAllocatedBytes = bytes
}
}

// MockSandboxClientCustom allows custom error logic per call
type MockSandboxClientCustom struct {
orchestrator.SandboxServiceClient
Expand Down
15 changes: 12 additions & 3 deletions packages/api/internal/orchestrator/nodemanager/node.go
Original file line number Diff line number Diff line change
Expand Up @@ -215,7 +215,16 @@ func (n *Node) OptimisticRemove(ctx context.Context, res SandboxResources) {
n.metricsMu.Lock()
defer n.metricsMu.Unlock()

// Directly subtract from the current metrics view
n.metrics.CpuAllocated -= uint32(res.CPUs)
n.metrics.MemoryAllocatedBytes -= uint64(res.MiBMemory) * 1024 * 1024
cpu := uint32(res.CPUs)
memory := uint64(res.MiBMemory) * 1024 * 1024

// Prevent underflow due to race condition (the sandbox was most likely already removed by the node sync)
if cpu > n.metrics.CpuAllocated || memory > n.metrics.MemoryAllocatedBytes {
logger.L().Warn(ctx, "OptimisticRemove would cause underflow, skipping", logger.WithNodeID(n.ID), zap.Uint32("cpuAllocated", n.metrics.CpuAllocated), zap.Uint64("memoryAllocatedBytes", n.metrics.MemoryAllocatedBytes), zap.Uint32("cpuToRemove", cpu), zap.Uint64("memoryToRemove", memory))

return
}
Comment thread
jakubno marked this conversation as resolved.

n.metrics.CpuAllocated -= cpu
n.metrics.MemoryAllocatedBytes -= memory
}
46 changes: 45 additions & 1 deletion packages/api/internal/orchestrator/nodemanager/node_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ func TestNode_OptimisticRemove_FlagEnabled(t *testing.T) {
require.NoError(t, err)

// 4. Initialize Node with the injected ffClient - some resources are already allocated at initialization
node := NewTestNode("test-node", api.NodeStatusReady, 4, 8192, WithFeatureFlags(ffClient))
node := NewTestNode("test-node", api.NodeStatusReady, 4, 8192, WithFeatureFlags(ffClient), WithAllocatedMemoryBytes(8192*1024*1024))
initialMetrics := node.Metrics()

// 5. Call the method
Expand All @@ -101,6 +101,50 @@ func TestNode_OptimisticRemove_FlagEnabled(t *testing.T) {
assert.Equal(t, initialMetrics.MemoryAllocatedBytes-uint64(res.MiBMemory)*1024*1024, newMetrics.MemoryAllocatedBytes)
}

func TestNode_OptimisticRemove_SkipsWhenItWouldUnderflow(t *testing.T) {
t.Parallel()

td := ldtestdata.DataSource()
td.Update(td.Flag(featureflags.OptimisticResourceAccountingFlag.Key()).VariationForAll(true))

ffClient, err := featureflags.NewClientWithDatasource(td)
require.NoError(t, err)

// Node with less allocated than what will be removed: 1 CPU, 512 MiB
node := NewTestNode("test-node", api.NodeStatusReady, 1, 8192, WithFeatureFlags(ffClient), WithAllocatedMemoryBytes(512*1024*1024))
initialMetrics := node.Metrics()

res := SandboxResources{
CPUs: 2,
MiBMemory: 1024,
}
node.OptimisticRemove(t.Context(), res)

// Counters must never wrap to ~2^32/2^64; subtraction is skipped instead
newMetrics := node.Metrics()
assert.Equal(t, initialMetrics.CpuAllocated, newMetrics.CpuAllocated)
assert.Equal(t, initialMetrics.MemoryAllocatedBytes, newMetrics.MemoryAllocatedBytes)
}

func TestNode_OptimisticRemove_FreshNodeDoesNotUnderflow(t *testing.T) {
t.Parallel()

td := ldtestdata.DataSource()
td.Update(td.Flag(featureflags.OptimisticResourceAccountingFlag.Key()).VariationForAll(true))

ffClient, err := featureflags.NewClientWithDatasource(td)
require.NoError(t, err)

// Fresh node: nothing allocated yet (e.g. poll overwrote counters after sandbox already left the orchestrator)
node := NewTestNode("test-node", api.NodeStatusReady, 0, 8192, WithFeatureFlags(ffClient))

node.OptimisticRemove(t.Context(), SandboxResources{CPUs: 2, MiBMemory: 1024})

newMetrics := node.Metrics()
assert.Equal(t, uint32(0), newMetrics.CpuAllocated)
assert.Equal(t, uint64(0), newMetrics.MemoryAllocatedBytes)
}

func TestNode_OptimisticRemove_FlagDisabled(t *testing.T) {
t.Parallel()

Expand Down
Loading