fix(inventory): readjust inventory on startup for existing leases

Signed-off-by: Artur Troian <[email protected]>
akash-network · Mar 27, 2024 · efe9fb2 · efe9fb2
1 parent 7a06ee4
commit efe9fb2
Show file tree

Hide file tree

Showing 8 changed files with 755 additions and 345 deletions.
diff --git a/cluster/inventory.go b/cluster/inventory.go
@@ -478,6 +478,10 @@ func (is *inventoryService) run(ctx context.Context, reservationsArg []*reservat
 	}
 
 	var runch <-chan runner.Result
+	var currinv ctypes.Inventory
+
+	invupch := make(chan ctypes.Inventory, 1)
+
 	invch := is.clients.inventory.ResultChan()
 	var reserveChLocal <-chan inventoryRequest
 
@@ -543,6 +547,13 @@ loop:
 							"order", res.OrderID(),
 							"resource-group", res.Resources().GetName(),
 							"allocated", res.allocated)
+
+						if currinv != nil {
+							select {
+							case invupch <- currinv:
+							default:
+							}
+						}
 					}
 
 					break
@@ -606,12 +617,19 @@ loop:
 				res: resp,
 				err: err,
 			}
-			inventoryRequestsCounter.WithLabelValues("status", "success").Inc()
+			if err == nil {
+				inventoryRequestsCounter.WithLabelValues("status", "success").Inc()
+			} else {
+				inventoryRequestsCounter.WithLabelValues("status", "error").Inc()
+			}
 		case inv, open := <-invch:
 			if !open {
 				continue
 			}
 
+			invupch <- inv
+		case inv := <-invupch:
+			currinv = inv.Dup()
 			state.inventory = inv
 			updateIPs()
 

diff --git a/cluster/inventory_test.go b/cluster/inventory_test.go
@@ -67,7 +67,7 @@ func TestInventory_reservationAllocatable(t *testing.T) {
 		}
 	}
 
-	inv := <-cinventory.NewNull("a", "b").ResultChan()
+	inv := <-cinventory.NewNull(context.Background(), "a", "b").ResultChan()
 
 	reservations := []*reservation{
 		mkres(true, mkrg(750, 0, 3*unit.Gi, 1*unit.Gi, 0, 1)),
@@ -109,7 +109,7 @@ func TestInventory_ClusterDeploymentNotDeployed(t *testing.T) {
 
 	ctx = context.WithValue(ctx, fromctx.CtxKeyKubeClientSet, kubernetes.Interface(kc))
 	ctx = context.WithValue(ctx, fromctx.CtxKeyAkashClientSet, aclient.Interface(ac))
-	ctx = context.WithValue(ctx, cfromctx.CtxKeyClientInventory, cinventory.NewNull("nodeA"))
+	ctx = context.WithValue(ctx, cfromctx.CtxKeyClientInventory, cinventory.NewNull(ctx, "nodeA"))
 
 	inv, err := newInventoryService(
 		ctx,
@@ -205,7 +205,7 @@ func TestInventory_ClusterDeploymentDeployed(t *testing.T) {
 	ctx = context.WithValue(ctx, fromctx.CtxKeyPubSub, tpubsub.New(ctx, 1000))
 	ctx = context.WithValue(ctx, fromctx.CtxKeyKubeClientSet, kubernetes.Interface(kc))
 	ctx = context.WithValue(ctx, fromctx.CtxKeyAkashClientSet, aclient.Interface(ac))
-	ctx = context.WithValue(ctx, cfromctx.CtxKeyClientInventory, cinventory.NewNull("nodeA"))
+	ctx = context.WithValue(ctx, cfromctx.CtxKeyClientInventory, cinventory.NewNull(ctx, "nodeA"))
 
 	inv, err := newInventoryService(
 		ctx,
@@ -419,7 +419,7 @@ func TestInventory_ReserveIPNoIPOperator(t *testing.T) {
 	ctx = context.WithValue(ctx, fromctx.CtxKeyPubSub, tpubsub.New(ctx, 1000))
 	ctx = context.WithValue(ctx, fromctx.CtxKeyKubeClientSet, kubernetes.Interface(kc))
 	ctx = context.WithValue(ctx, fromctx.CtxKeyAkashClientSet, aclient.Interface(ac))
-	ctx = context.WithValue(ctx, cfromctx.CtxKeyClientInventory, cinventory.NewNull("nodeA"))
+	ctx = context.WithValue(ctx, cfromctx.CtxKeyClientInventory, cinventory.NewNull(ctx, "nodeA"))
 
 	inv, err := newInventoryService(
 		ctx,
@@ -473,7 +473,7 @@ func TestInventory_ReserveIPUnavailableWithIPOperator(t *testing.T) {
 	ctx = context.WithValue(ctx, fromctx.CtxKeyPubSub, tpubsub.New(ctx, 1000))
 	ctx = context.WithValue(ctx, fromctx.CtxKeyKubeClientSet, kubernetes.Interface(kc))
 	ctx = context.WithValue(ctx, fromctx.CtxKeyAkashClientSet, aclient.Interface(ac))
-	ctx = context.WithValue(ctx, cfromctx.CtxKeyClientInventory, cinventory.NewNull("nodeA"))
+	ctx = context.WithValue(ctx, cfromctx.CtxKeyClientInventory, cinventory.NewNull(ctx, "nodeA"))
 	ctx = context.WithValue(ctx, cfromctx.CtxKeyClientIP, cip.Client(mockIP))
 
 	inv, err := newInventoryService(
@@ -548,7 +548,7 @@ func TestInventory_ReserveIPAvailableWithIPOperator(t *testing.T) {
 	ctx = context.WithValue(ctx, fromctx.CtxKeyPubSub, tpubsub.New(ctx, 1000))
 	ctx = context.WithValue(ctx, fromctx.CtxKeyKubeClientSet, kubernetes.Interface(kc))
 	ctx = context.WithValue(ctx, fromctx.CtxKeyAkashClientSet, aclient.Interface(ac))
-	ctx = context.WithValue(ctx, cfromctx.CtxKeyClientInventory, cinventory.NewNull("nodeA", "nodeB"))
+	ctx = context.WithValue(ctx, cfromctx.CtxKeyClientInventory, cinventory.NewNull(ctx, "nodeA", "nodeB"))
 	ctx = context.WithValue(ctx, cfromctx.CtxKeyClientIP, cip.Client(mockIP))
 
 	inv, err := newInventoryService(
@@ -596,6 +596,7 @@ func TestInventory_ReserveIPAvailableWithIPOperator(t *testing.T) {
 	mockIP.AssertNumberOfCalls(t, "GetIPAddressStatus", 2)
 }
 
+// following test needs refactoring it reports incorrect inventory
 func TestInventory_OverReservations(t *testing.T) {
 	scaffold := makeInventoryScaffold(t, 10)
 	defer scaffold.bus.Close()
@@ -659,10 +660,12 @@ func TestInventory_OverReservations(t *testing.T) {
 	ac := afake.NewSimpleClientset()
 
 	ctx, cancel := context.WithCancel(context.Background())
+	nullInv := cinventory.NewNull(ctx, "nodeA")
+
 	ctx = context.WithValue(ctx, fromctx.CtxKeyPubSub, tpubsub.New(ctx, 1000))
 	ctx = context.WithValue(ctx, fromctx.CtxKeyKubeClientSet, kubernetes.Interface(kc))
 	ctx = context.WithValue(ctx, fromctx.CtxKeyAkashClientSet, aclient.Interface(ac))
-	ctx = context.WithValue(ctx, cfromctx.CtxKeyClientInventory, cinventory.NewNull("nodeA"))
+	ctx = context.WithValue(ctx, cfromctx.CtxKeyClientInventory, nullInv)
 
 	inv, err := newInventoryService(
 		ctx,
@@ -685,6 +688,8 @@ func TestInventory_OverReservations(t *testing.T) {
 	require.Error(t, err)
 	require.ErrorIs(t, err, ctypes.ErrInsufficientCapacity)
 
+	nullInv.Commit(reservation.Resources())
+
 	// Send the event immediately to indicate it was deployed
 	err = scaffold.bus.Publish(event.ClusterDeployment{
 		LeaseID: lid0,

diff --git a/cluster/kube/operators/clients/inventory/client.go b/cluster/kube/operators/clients/inventory/client.go
@@ -19,8 +19,6 @@ import (
 	"k8s.io/client-go/tools/portforward"
 	"k8s.io/client-go/transport/spdy"
 
-	"github.com/tendermint/tendermint/libs/log"
-
 	inventoryV1 "github.com/akash-network/akash-api/go/inventory/v1"
 
 	kutil "github.com/akash-network/provider/cluster/kube/util"
@@ -48,7 +46,6 @@ type client struct {
 
 type inventory struct {
 	inventoryV1.Cluster
-	log log.Logger
 }
 
 type inventoryState struct {
@@ -209,22 +206,20 @@ func (cl *client) subscriber(in <-chan inventoryV1.Cluster, out chan<- ctypes.In
 	var msg ctypes.Inventory
 	var och chan<- ctypes.Inventory
 
-	ilog := fromctx.LogcFromCtx(cl.ctx).With("inventory.adjuster")
-
 	for {
 		select {
 		case <-cl.ctx.Done():
 			return cl.ctx.Err()
 		case inv := <-in:
 			pending = append(pending, inv)
 			if och == nil {
-				msg = newInventory(ilog, pending[0])
+				msg = newInventory(pending[0])
 				och = out
 			}
 		case och <- msg:
 			pending = pending[1:]
 			if len(pending) > 0 {
-				msg = newInventory(ilog, pending[0])
+				msg = newInventory(pending[0])
 			} else {
 				och = nil
 				msg = nil

diff --git a/cluster/kube/operators/clients/inventory/inventory.go b/cluster/kube/operators/clients/inventory/inventory.go
@@ -1,13 +1,10 @@
 package inventory
 
 import (
-	"encoding/json"
 	"fmt"
 	"reflect"
 	"strings"
 
-	"github.com/tendermint/tendermint/libs/log"
-
 	inventoryV1 "github.com/akash-network/akash-api/go/inventory/v1"
 	dtypes "github.com/akash-network/akash-api/go/node/deployment/v1beta3"
 	types "github.com/akash-network/akash-api/go/node/types/v1beta3"
@@ -20,10 +17,9 @@ import (
 
 var _ ctypes.Inventory = (*inventory)(nil)
 
-func newInventory(log log.Logger, clState inventoryV1.Cluster) *inventory {
+func newInventory(clState inventoryV1.Cluster) *inventory {
 	inv := &inventory{
 		Cluster: clState,
-		log:     log,
 	}
 
 	return inv
@@ -32,12 +28,17 @@ func newInventory(log log.Logger, clState inventoryV1.Cluster) *inventory {
 func (inv *inventory) dup() inventory {
 	dup := inventory{
 		Cluster: *inv.Cluster.Dup(),
-		log:     inv.log,
 	}
 
 	return dup
 }
 
+func (inv *inventory) Dup() ctypes.Inventory {
+	dup := inv.dup()
+
+	return &dup
+}
+
 // tryAdjust cluster inventory
 // It returns two boolean values. First indicates if node-wide resources satisfy (true) requirements
 // Seconds indicates if cluster-wide resources satisfy (true) requirements
@@ -84,11 +85,6 @@ func (inv *inventory) tryAdjust(node int, res *types.Resources) (*crd.SchedulerP
 			return nil, false, true
 		}
 
-		// if !nd.tryAdjustVolumesAttached(types.NewResourceValue(1)) {
-		// 	return nil, false, true
-
-		// }
-
 		storageAdjusted := false
 
 		for idx := range storageClasses {
@@ -272,27 +268,13 @@ nodes:
 				// same adjusted resource units as well as cluster params
 				if adjustedGroup {
 					if !reflect.DeepEqual(adjusted, &adjustedResources[i].Resources) {
-						jFirstAdjusted, _ := json.Marshal(&adjustedResources[i].Resources)
-						jCurrAdjusted, _ := json.Marshal(adjusted)
-
-						inv.log.Error(fmt.Sprintf("resource mismatch between replicas within group:\n"+
-							"\tfirst adjusted replica: %s\n"+
-							"\tcurr adjusted replica: %s", string(jFirstAdjusted), string(jCurrAdjusted)))
-
 						err = ctypes.ErrGroupResourceMismatch
 						break nodes
 					}
 
 					// all replicas of the same service are expected to have same node selectors and runtimes
 					// if they don't match then provider cannot bid
 					if !reflect.DeepEqual(sparams, cparams[adjusted.ID]) {
-						jFirstSparams, _ := json.Marshal(cparams[adjusted.ID])
-						jCurrSparams, _ := json.Marshal(sparams)
-
-						inv.log.Error(fmt.Sprintf("scheduler params mismatch between replicas within group:\n"+
-							"\tfirst replica: %s\n"+
-							"\tcurr replica: %s", string(jFirstSparams), string(jCurrSparams)))
-
 						err = ctypes.ErrGroupResourceMismatch
 						break nodes
 					}