Support volume reservation on storagepool

README.md

@@ -97,3 +97,10 @@ However, it is important to note that LiteIO does not currently support data rep
 
 - [x] Disk-Agent exposes metric service
 - [ ] SPDK volume replica
+
+
+## Contact
+
+Wechat Group QRCode
+
+![Wechat Group](doc/image/wechat_group.JPG)

cmd/controller/antplugins/localstorage.go

@@ -5,7 +5,9 @@ import (
  "fmt"
  "strconv"
 
+ corev1 "k8s.io/api/core/v1"
  metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "sigs.k8s.io/controller-runtime/pkg/client"
 
  v1 "code.alipay.com/dbplatform/node-disk-controller/pkg/api/volume.antstor.alipay.com/v1"
  "code.alipay.com/dbplatform/node-disk-controller/pkg/controller/kubeutil"
@@ -41,6 +43,7 @@ func NewReportLocalStoragePlugin(h *controllers.PluginHandle) (p plugin.Plugin,
  }
 
  p = &ReportLocalStoragePlugin{
+ Client: h.Client,
  NodeUpdater: kubeutil.NewKubeNodeInfoGetter(h.Req.KubeCli),
  PoolUtil: kubeutil.NewStoragePoolUtil(h.Client),
  ReportLocalConfigs: pluginCfg.DefaultLocalSpaceRules,
@@ -50,7 +53,7 @@ func NewReportLocalStoragePlugin(h *controllers.PluginHandle) (p plugin.Plugin,
 
 // ReportLocalStoragePlugin is a AntstorVolume plugin.
 type ReportLocalStoragePlugin struct {
- // NodeGetter kubeutil.NodeInfoGetterIface
+ Client client.Client
  NodeUpdater kubeutil.NodeUpdaterIface
  PoolUtil kubeutil.StoragePoolUpdater
 
@@ -96,18 +99,44 @@ func (r *ReportLocalStoragePlugin) Reconcile(ctx *plugin.Context) (result plugin
 
  // report the local storage when the StoragePool is created in the first place.
  if isPool && pool != nil {
- totalBs := pool.GetAvailableBytes()
- if _, has := pool.Labels[v1.PoolLocalStorageBytesKey]; !has {
- log.Info("update node/status capacity", "local-storage", totalBs)
- // update Pool Label "obnvmf/node-local-storage-size" = totalBs
- err = r.PoolUtil.SavePoolLocalStorageMark(pool, uint64(totalBs))
+ var (
+ localBS uint64
+ node corev1.Node
+ snode *state.Node
+ hasNodeRes bool
+ hasPoolLabel bool
+ )
+
+ // calculate local storage
+ snode, err = stateObj.GetNodeByNodeID(pool.Name)
+ if err != nil {
+ log.Error(err, "find node failed")
+ return plugin.Result{Error: err}
+ }
+ localBS = CalculateLocalStorageCapacity(snode)
+
+ // get node
+ err = r.Client.Get(ctx.ReqCtx.Ctx, client.ObjectKey{Name: pool.Name}, &node)
+ if err != nil {
+ log.Error(err, "getting Node failed")
+ return plugin.Result{Error: err}
+ }
+
+ _, hasNodeRes = node.Status.Allocatable[kubeutil.SdsLocalStorageResourceKey]
+ _, hasPoolLabel = pool.Labels[v1.PoolLocalStorageBytesKey]
+ log.Info("check pool PoolLocalStorageBytesKey and node SdsLocalStorageResourceKey", "nodeResource", hasNodeRes, "hasPoolLabel", hasPoolLabel)
+
+ if !hasPoolLabel || !hasNodeRes {
+ log.Info("update node/status capacity", "local-storage", localBS)
+ // update Pool Label "obnvmf/local-storage-bytes" = totalBs
+ err = r.PoolUtil.SavePoolLocalStorageMark(pool, localBS)
  if err != nil {
  log.Error(err, "SavePoolLocalStorageMark failed")
  return plugin.Result{Error: err}
  }
 
  // update node/status capacity = totalBs
- _, err = r.NodeUpdater.ReportLocalDiskResource(pool.Name, uint64(totalBs))
+ _, err = r.NodeUpdater.ReportLocalDiskResource(pool.Name, localBS)
  if err != nil {
  log.Error(err, "ReportLocalDiskResource failed")
  return plugin.Result{Error: err}
@@ -126,28 +155,6 @@ func (r *ReportLocalStoragePlugin) Reconcile(ctx *plugin.Context) (result plugin
  }
  var sp = node.Pool
 
- /*
- var localStorePct int
- var volInState *v1.AntstorVolume
- for _, item := range r.ReportLocalConfigs {
- selector, err := metav1.LabelSelectorAsSelector(&item.LabelSelector)
- if err != nil {
- log.Error(err, "LabelSelectorAsSelector failed", "selector", item.LabelSelector)
- continue
- }
- if selector.Matches(labels.Set(sp.Spec.NodeInfo.Labels)) && item.EnableDefault {
- localStorePct = item.DefaultLocalStoragePct
- log.Info("matched local-storage percentage", "pct", localStorePct)
- }
- }
-
- volInState, err = node.GetVolumeByID(volume.Spec.Uuid)
- if err == nil {
- log.Info("copy volume into state")
- *volInState = *volume
- }
- */
-
  var expectLocalSize = CalculateLocalStorageCapacity(node)
  var localSizeStr = strconv.Itoa(int(expectLocalSize))
  log.Info("compare local storage size", "in label", sp.Labels[v1.PoolLocalStorageBytesKey], "expect", localSizeStr, "delTS", volume.DeletionTimestamp)

hack/deploy/lvm/050-configmap.yaml

@@ -25,6 +25,9 @@ data:
  nodeTaints:
  - key: node.sigma.ali/lifecycle
  operator: Exists
+ #nodeReservations:
+ #- id: obnvmf/app-vol
+ # size: 107374182400 # 100Gi
  pluginConfigs:
  defaultLocalSpaceRules:
  - enableDefault: true

pkg/controller/manager/config/config.go

@@ -30,6 +30,13 @@ type SchedulerConfig struct {
  NodeCacheSelector map[string]string `json:"nodeCacheSelector" yaml:"nodeCacheSelector"`
  // MinLocalStoragePct defines the minimun percentage of local storage to be reserved on one node.
  MinLocalStoragePct int `json:"minLocalStoragePct" yaml:"minLocalStoragePct"`
+ // NodeReservations defines the reservations on each node
+ NodeReservations []NodeReservation `json:"nodeReservations" yaml:"nodeReservations"`
+}
+
+type NodeReservation struct {
+ ID string `json:"id" yaml:"id"`
+ Size int64 `json:"size" yaml:"size"`
 }
 
 type NoScheduleConfig struct {

pkg/controller/manager/controllers/manager.go

@@ -100,6 +100,7 @@ func NewAndInitControllerManager(req NewManagerRequest) manager.Manager {
  Concurrency: 4,
  MainHandler: &reconciler.StoragePoolReconcileHandler{
  Client: mgr.GetClient(),
+ Cfg: req.ControllerConfig,
  State: stateObj,
  PoolUtil: poolUtil,
  KubeCli: kubeClient,

pkg/controller/manager/reconciler/pool_reconciler.go

@@ -20,6 +20,7 @@ import (
 
  v1 "code.alipay.com/dbplatform/node-disk-controller/pkg/api/volume.antstor.alipay.com/v1"
  "code.alipay.com/dbplatform/node-disk-controller/pkg/controller/kubeutil"
+ "code.alipay.com/dbplatform/node-disk-controller/pkg/controller/manager/config"
  "code.alipay.com/dbplatform/node-disk-controller/pkg/controller/manager/reconciler/plugin"
  "code.alipay.com/dbplatform/node-disk-controller/pkg/controller/manager/state"
  "code.alipay.com/dbplatform/node-disk-controller/pkg/util/misc"
@@ -39,6 +40,7 @@ var (
 type StoragePoolReconcileHandler struct {
  client.Client
 
+ Cfg config.Config
  State state.StateIface
  PoolUtil kubeutil.StoragePoolUpdater
  KubeCli kubernetes.Interface
@@ -227,6 +229,7 @@ func (r *StoragePoolReconcileHandler) handleDeletion(pCtx *plugin.Context) (resu
 func (r *StoragePoolReconcileHandler) saveToState(sp *v1.StoragePool, log logr.Logger) (result plugin.Result) {
  var patch = client.MergeFrom(sp.DeepCopy())
  var err error
+ var node *state.Node
 
  r.State.SetStoragePool(sp)
 
@@ -244,6 +247,19 @@ func (r *StoragePoolReconcileHandler) saveToState(sp *v1.StoragePool, log logr.L
  }
  }
 
+ // try to add reservation by config
+ node, err = r.State.GetNodeByNodeID(sp.Name)
+ if err != nil {
+ log.Error(err, "GetNodeByNodeID error")
+ }
+ for _, item := range r.Cfg.Scheduler.NodeReservations {
+ if node != nil {
+ if _, has := node.GetReservation(item.ID); !has {
+ node.Reserve(state.NewReservation(item.ID, item.Size))
+ }
+ }
+ }
+
  return plugin.Result{}
 }
 

pkg/controller/manager/scheduler/filter/basic.go

@@ -21,6 +21,11 @@ func BasicFilterFunc(ctx *FilterContext, n *state.Node, vol *v1.AntstorVolume) b
  )
  // check if err is nil
 
+ // if voume matches reservation, then do not do following checks
+ if pass, hasErr := matchReservationFilter(ctx, n, vol); hasErr || pass {
+ return pass
+ }
+
  // consider Pool FreeSpace
  var freeRes = n.GetFreeResourceNonLock()
  var freeDisk = freeRes[v1.ResourceDiskPoolByte]
@@ -100,3 +105,23 @@ func BasicFilterFunc(ctx *FilterContext, n *state.Node, vol *v1.AntstorVolume) b
 
  return true
 }
+
+func matchReservationFilter(ctx *FilterContext, n *state.Node, vol *v1.AntstorVolume) (pass, hasError bool) {
+ if resvId, has := vol.Annotations[v1.ReservationIDKey]; has {
+ free := n.FreeResource.Storage()
+ if free.CmpInt64(0) < 0 {
+ ctx.Error.AddReason(ReasonPoolFreeSize)
+ return false, true
+ }
+
+ if r, has := n.GetReservation(resvId); has {
+ if r.Size() < int64(vol.Spec.SizeByte) {
+ ctx.Error.AddReason(ReasonReservationSize)
+ return false, true
+ }
+ return true, false
+ }
+ }
+
+ return false, false
+}

pkg/controller/manager/scheduler/filter/error.go

@@ -16,6 +16,7 @@ const (
  ReasonNodeAffinity = "NodeAffinity"
  ReasonPoolAffinity = "PoolAffinity"
  ReasonPoolUnschedulable = "PoolUnschedulable"
+ ReasonReservationSize = "ReservationTooSmall"
 
  NoStoragePoolAvailable = "NoStoragePoolAvailable"
  //

pkg/controller/manager/state/http.go

@@ -13,20 +13,31 @@ type NodeStateAPI struct {
  KernelLVM *v1.KernelLVM `json:"kernelLVM,omitempty"`
  SpdkLVS *v1.SpdkLVStore `json:"spdkLVS,omitempty"`
  // Volumes breif info
- Volumes []VolumeBreif `json:"volumes"`
- // FreeSize of the pool
- FreeSize int64 `json:"freeSize"`
+ Volumes []VolumeBrief `json:"volumes"`
+ // VgFreeSize of the pool
+ VgFreeSize int64 `json:"vgFreeSize"`
+ // MemFreeSize in controller memory
+ MemFreeSize int64 `json:"memFreeSize"`
+ // MemFreeSizeStr readable size in controller memory
+ MemFreeSizeStr string `json:"memFreeSizeStr"`
  // Conditions of the pool status
  Conditions map[v1.PoolConditionType]v1.ConditionStatus `json:"conditions"`
+ // Resvervations on the node
+ Resvervations []ReservationBreif `json:"reservations"`
 }
 
-type VolumeBreif struct {
+type VolumeBrief struct {
  Namespace string `json:"ns"`
  Name string `json:"name"`
  DataHolder string `json:"dataHolder"`
  Size int64 `json:"size"`
 }
 
+type ReservationBreif struct {
+ ID string `json:"id"`
+ Size int64 `json:"size"`
+}
+
 func NewStateHandler(s StateIface) *StateHandler {
  return &StateHandler{state: s}
 }
@@ -49,11 +60,14 @@ func (h *StateHandler) ServeHTTP(writer http.ResponseWriter, req *http.Request)
  }
 
  var api = NodeStateAPI{
- Name: spName,
- PoolLabels: node.Pool.Labels,
- KernelLVM: &node.Pool.Spec.KernelLVM,
- SpdkLVS: &node.Pool.Spec.SpdkLVStore,
- FreeSize: node.Pool.Status.VGFreeSize.Value(),
+ Name: spName,
+ PoolLabels: node.Pool.Labels,
+ KernelLVM: &node.Pool.Spec.KernelLVM,
+ SpdkLVS: &node.Pool.Spec.SpdkLVStore,
+ VgFreeSize: node.Pool.Status.VGFreeSize.Value(),
+ MemFreeSize: int64(node.FreeResource.Storage().AsApproximateFloat64()),
+ MemFreeSizeStr: node.FreeResource.Storage().String(),
+
  Conditions: make(map[v1.PoolConditionType]v1.ConditionStatus),
  }
 
@@ -62,14 +76,23 @@ func (h *StateHandler) ServeHTTP(writer http.ResponseWriter, req *http.Request)
  }
 
  for _, vol := range node.Volumes {
- api.Volumes = append(api.Volumes, VolumeBreif{
+ api.Volumes = append(api.Volumes, VolumeBrief{
  Namespace: vol.Namespace,
  Name: vol.Name,
  Size: int64(vol.Spec.SizeByte),
  DataHolder: vol.Labels[v1.VolumeDataHolderKey],
  })
  }
 
+ if node.resvSet != nil {
+ for _, resv := range node.resvSet.Items() {
+ api.Resvervations = append(api.Resvervations, ReservationBreif{
+ ID: resv.ID(),
+ Size: resv.Size(),
+ })
+ }
+ }
+
  bs, err := json.Marshal(api)
  if err != nil {
  writer.Write([]byte(err.Error()))

pkg/controller/manager/state/node.go

@@ -63,6 +63,13 @@ func (n *Node) AddVolume(vol *v1.AntstorVolume) (err error) {
  defer n.volLock.Unlock()
 
  var nodeID = n.Info.ID
+ var duplicate bool
+
+ // delete reservation if volume has reservation id
+ if resvID := getVolumeReservationID(vol); resvID != "" {
+ n.resvSet.Unreserve(resvID)
+ }
+
  // check duplicate
  for _, item := range n.Volumes {
  if item.Name == vol.Name {
@@ -77,19 +84,16 @@ func (n *Node) AddVolume(vol *v1.AntstorVolume) (err error) {
  // save the newer volume
  *item = *vol.DeepCopy()
  klog.Infof("vol %s already in node %s. type and sizes equal to each other", vol.Name, nodeID)
- return
+ duplicate = true
  }
  }
 
- // delete reservation if volume has reservation id
- if resvID := getVolumeReservationID(vol); resvID != "" {
- n.resvSet.Unreserve(resvID)
+ if !duplicate {
+ // volume reside on Node
+ vol.Spec.TargetNodeId = n.Pool.Spec.NodeInfo.ID
+ n.Volumes = append(n.Volumes, vol)
  }
 
- n.Volumes = append(n.Volumes, vol)
- // volume reside on Node
- vol.Spec.TargetNodeId = n.Pool.Spec.NodeInfo.ID
-
  // update free resource
  n.FreeResource = n.GetFreeResourceNonLock()
 
@@ -231,14 +235,40 @@ func (n *Node) GetFreeResourceNonLock() (free corev1.ResourceList) {
 
 // Reserve storage resource for Node
 func (n *Node) Reserve(r ReservationIface) {
+ // if volume is already binded, then skip reservation.
+ var resvID = r.ID()
+ for _, vol := range n.Volumes {
+ if resvID == getVolumeReservationID(vol) {
+ return
+ }
+ }
+
+ // check free resource
+ if free := n.FreeResource.Storage(); free != nil {
+ if free.CmpInt64(r.Size()) < 0 {
+ klog.Errorf("node %s have no enough disk pool space for reservation %s", n.Info.ID, resvID)
+ return
+ }
+ }
+
+ n.volLock.Lock()
+ defer n.volLock.Unlock()
+
  n.resvSet.Reserve(r)
  // update free resource
  n.FreeResource = n.GetFreeResourceNonLock()
 }
 
 // Unreserve storage resource
 func (n *Node) Unreserve(id string) {
+ n.volLock.Lock()
+ defer n.volLock.Unlock()
+
  n.resvSet.Unreserve(id)
  // update free resource
  n.FreeResource = n.GetFreeResourceNonLock()
 }
+
+func (n *Node) GetReservation(id string) (r ReservationIface, has bool) {
+ return n.resvSet.GetById(id)
+}