prototype of custom resource scheduling

Platforms other than Linux, especially non-Unix platforms, may have a different
view of resources that don't model well as `resources.cpu` or
`resources.memory`. And users with unusual deployment environments have asked
about the possibility of scheduling resources specific to those environments.

This PR is a prototype of the kind of interface we might want to be able to
support. Nodes get a `client.custom_resource` block where they can define a
resource the node will make available in its fingerprint. Resources are one of
five types, designed to make it possible to model all our existing resources as
custom resources:
* `ratio`: The resource is used as a value relative to other tasks with the same
resource. An example of this would be Linux `cpu.weight` as implemented by a
systemd unit file.
* `capped-ratio`: Like a ratio, except all the quantity used by tasks has a
maximum total value. An example of this would be Linux `cpu.weight` as currently
implemented in Nomad, where the "cap" is derived from the total Mhz of CPUs on
the host.
* `countable`: The resource has a fixed but fungible amount. An example of this
would be memory allocation (ignoring NUMA), where there's a certain amount of
memory available on the host and it's "used up" by allocations, but we don't
care about identity of the individual blocks of memory.
* `dynamic`: The resource has a fixed set of items with exclusive access, but
the job doesn't care which ones it gets. An example of this would be Nomad's
current dynamic port assignment or `resource.cores`.
* `static`: The resource has a fixed set of items with exclusive access, and the
job wants a specific one of those items. An example of this woul dbe Nomad's
current static port assignment.

We can use this prototype to anchor discussions about how we might implement a
set of custom resource scheduling features but it's nowhere near
production-ready. I've included enough plumbing to implement these five resource
types, fingerprint them on clients via config files, and schedule them for
allocations. What's not included, all of which we'd want to solve in any
productionized version of this work:

* Support for exposing the custom resource allocation to a task driver. We'd
need to thread custom resources into the protobufs we send via `go-plugin` and
then the user's task driver would need to consume that data.
* Support for preemption
* Support for quotas
* Support for dynamically adding custom resources to a node

Ref: #1081

Author: tgross

api/allocations.go

@@ -436,12 +436,14 @@ type AllocatedTaskResources struct {
 	Memory   AllocatedMemoryResources
 	Networks []*NetworkResource
 	Devices  []*AllocatedDeviceResource
+	Custom   []*CustomResource
 }
 
 type AllocatedSharedResources struct {
 	DiskMB   int64
 	Networks []*NetworkResource
 	Ports    []PortMapping
+	Custom   []*CustomResource
 }
 
 type PortMapping struct {

api/nodes.go

@@ -587,6 +587,7 @@ type NodeResources struct {
 	Disk     NodeDiskResources
 	Networks []*NetworkResource
 	Devices  []*NodeDeviceResource
+	Custom   []*NodeCustomResource
 
 	MinDynamicPort int
 	MaxDynamicPort int
@@ -629,6 +630,18 @@ type NodeReservedNetworkResources struct {
 	ReservedHostPorts string
 }
 
+type NodeCustomResource struct {
+	Name    string              `hcl:"name,label"`
+	Version uint64              `hcl:"version,optional"`
+	Type    CustomResourceType  `hcl:"type,optional"`
+	Scope   CustomResourceScope `hcl:"scope,optional"`
+
+	Quantity int64             `hcl:"quantity,optional"`
+	Range    string            `hcl:"range,optional"`
+	Items    []any             `hcl:"items,optional"`
+	Meta     map[string]string `hcl:"meta,block"`
+}
+
 type CSITopologyRequest struct {
 	Required  []*CSITopology `hcl:"required"`
 	Preferred []*CSITopology `hcl:"preferred"`

api/resources.go

@@ -20,6 +20,7 @@ type Resources struct {
 	Devices     []*RequestedDevice `hcl:"device,block"`
 	NUMA        *NUMAResource      `hcl:"numa,block"`
 	SecretsMB   *int               `mapstructure:"secrets" hcl:"secrets,optional"`
+	Custom      []*CustomResource  `mapstructure:"custom" hcl:"custom,block"`
 
 	// COMPAT(0.10)
 	// XXX Deprecated. Please do not use. The field will be removed in Nomad
@@ -330,3 +331,32 @@ func (d *RequestedDevice) Canonicalize() {
 		a.Canonicalize()
 	}
 }
+
+type CustomResource struct {
+	Name    string              `hcl:"name,label"`
+	Version uint64              `hcl:"version,optional"`
+	Type    CustomResourceType  `hcl:"type,optional"`
+	Scope   CustomResourceScope `hcl:"scope,optional"`
+
+	Quantity    int64         `hcl:"quantity,optional"`
+	Range       string        `hcl:"range,optional"`
+	Items       []any         `hcl:"items,optional"`
+	Constraints []*Constraint `hcl:"constraint,block"`
+}
+
+type CustomResourceScope string
+
+const (
+	CustomResourceScopeGroup CustomResourceScope = "group"
+	CustomResourceScopeTask  CustomResourceScope = "task"
+)
+
+type CustomResourceType string
+
+const (
+	CustomResourceTypeRatio           CustomResourceType = "ratio"        // ex. weight
+	CustomResourceTypeCappedRatio     CustomResourceType = "capped-ratio" // ex. resource.cpu
+	CustomResourceTypeCountable       CustomResourceType = "countable"    // ex. memory, disk
+	CustomResourceTypeDynamicInstance CustomResourceType = "dynamic"      // ex. ports, cores
+	CustomResourceTypeStaticInstance  CustomResourceType = "static"       // ex. ports, devices
+)

client/client.go

@@ -1649,12 +1649,14 @@ func (c *Client) setupNode() error {
 		node.NodeResources.MinDynamicPort = newConfig.MinDynamicPort
 		node.NodeResources.MaxDynamicPort = newConfig.MaxDynamicPort
 		node.NodeResources.Processors = newConfig.Node.NodeResources.Processors
+		node.NodeResources.Custom = newConfig.CustomResources
 
 		if node.NodeResources.Processors.Empty() {
 			node.NodeResources.Processors = structs.NodeProcessorResources{
 				Topology: &numalib.Topology{},
 			}
 		}
+		node.NodeResources.Custom = newConfig.CustomResources
 	}
 	if node.ReservedResources == nil {
 		node.ReservedResources = &structs.NodeReservedResources{}
@@ -1813,6 +1815,8 @@ func (c *Client) updateNodeFromFingerprint(response *fingerprint.FingerprintResp
 		if cpu := response.NodeResources.Processors.TotalCompute(); cpu > 0 {
 			newConfig.CpuCompute = cpu
 		}
+
+		response.NodeResources.Custom = newConfig.CustomResources
 	}
 
 	if nodeHasChanged {

client/config/config.go

@@ -395,6 +395,8 @@ type Config struct {
 
 	// LogFile is used by MonitorExport to stream a server's log file
 	LogFile string `hcl:"log_file"`
+
+	CustomResources structs.CustomResources
 }
 
 type APIListenerRegistrar interface {
@@ -896,6 +898,7 @@ func (c *Config) Copy() *Config {
 	nc.ReservableCores = slices.Clone(c.ReservableCores)
 	nc.Artifact = c.Artifact.Copy()
 	nc.Users = c.Users.Copy()
+	nc.CustomResources = c.CustomResources.Copy()
 	return &nc
 }
 

command/agent/agent.go

@@ -158,6 +158,7 @@ func NewAgent(config *Config, logger log.InterceptLogger, logOutput io.Writer, i
 	if err := a.setupServer(); err != nil {
 		return nil, err
 	}
+
 	if err := a.setupClient(); err != nil {
 		return nil, err
 	}
@@ -936,6 +937,7 @@ func convertClientConfig(agentConfig *Config) (*clientconfig.Config, error) {
 	conf.Node.Meta = agentConfig.Client.Meta
 	conf.Node.NodeClass = agentConfig.Client.NodeClass
 	conf.Node.NodePool = agentConfig.Client.NodePool
+	conf.CustomResources = agentConfig.Client.CustomResources
 
 	// Set up the HTTP advertise address
 	conf.Node.HTTPAddr = agentConfig.AdvertiseAddrs.HTTP

command/agent/config.go

@@ -443,6 +443,10 @@ type ClientConfig struct {
 
 	// LogFile is used by MonitorExport to stream a client's log file
 	LogFile string `hcl:"log_file"`
+
+	// CustomResources allows the user to define custom schedulable resources on
+	// this node
+	CustomResources structs.CustomResources `hcl:"custom_resource,block"`
 }
 
 func (c *ClientConfig) Copy() *ClientConfig {
@@ -466,6 +470,7 @@ func (c *ClientConfig) Copy() *ClientConfig {
 	nc.Drain = c.Drain.Copy()
 	nc.Users = c.Users.Copy()
 	nc.ExtraKeysHCL = slices.Clone(c.ExtraKeysHCL)
+	nc.CustomResources = c.CustomResources.Copy()
 	return &nc
 }
 
@@ -2883,6 +2888,10 @@ func (c *ClientConfig) Merge(b *ClientConfig) *ClientConfig {
 		result.IntroToken = b.IntroToken
 	}
 
+	if b.CustomResources != nil {
+		result.CustomResources.Merge(b.CustomResources)
+	}
+
 	return &result
 }
 

command/agent/config_parse.go

@@ -330,6 +330,11 @@ func extraKeys(c *Config) error {
 		helper.RemoveEqualFold(&c.Client.ExtraKeysHCL, "host_network")
 	}
 
+	for _, cr := range c.Client.CustomResources {
+		helper.RemoveEqualFold(&c.Client.ExtraKeysHCL, "custom_resource")
+		helper.RemoveEqualFold(&c.Client.ExtraKeysHCL, cr.Name)
+	}
+
 	// Remove Template extra keys
 	for _, t := range []string{"function_denylist", "disable_file_sandbox", "max_stale", "wait", "wait_bounds", "block_query_wait", "consul_retry", "vault_retry", "nomad_retry"} {
 		helper.RemoveEqualFold(&c.Client.ExtraKeysHCL, t)

command/agent/job_endpoint.go

@@ -1646,6 +1646,22 @@ func ApiResourcesToStructs(in *api.Resources) *structs.Resources {
 		out.SecretsMB = *in.SecretsMB
 	}
 
+	if len(in.Custom) > 0 {
+		out.Custom = []*structs.CustomResource{}
+		for _, apiCr := range in.Custom {
+			out.Custom = append(out.Custom, &structs.CustomResource{
+				Name:        apiCr.Name,
+				Version:     apiCr.Version,
+				Type:        structs.CustomResourceType(apiCr.Type),
+				Scope:       structs.CustomResourceScope(apiCr.Scope),
+				Quantity:    apiCr.Quantity,
+				Range:       apiCr.Range,
+				Items:       apiCr.Items,
+				Constraints: ApiConstraintsToStructs(apiCr.Constraints),
+			})
+		}
+	}
+
 	return out
 }
 

command/node_status.go

@@ -569,6 +569,7 @@ func (c *NodeStatusCommand) formatNode(client *api.Client, node *api.Node) int {
 		c.outputNodeNetworkInfo(node)
 		c.outputNodeCSIVolumeInfo(client, node, runningAllocs)
 		c.outputNodeDriverInfo(node)
+		c.outputNodeCustomResourceInfo(node)
 	}
 
 	// Emit node events
@@ -788,6 +789,35 @@ func (c *NodeStatusCommand) outputNodeDriverInfo(node *api.Node) {
 	c.Ui.Output(formatList(nodeDrivers))
 }
 
+func (c *NodeStatusCommand) outputNodeCustomResourceInfo(node *api.Node) {
+	c.Ui.Output(c.Colorize().Color("\n[bold]Custom Resources"))
+
+	resources := node.NodeResources.Custom
+	if len(resources) == 0 {
+		c.Ui.Output("<none>")
+		return
+	}
+
+	out := make([]string, 0, len(resources)+1)
+	out = append(out, "Resource|Version|Type|Available")
+
+	sort.Slice(resources, func(i int, j int) bool { return resources[i].Name < resources[j].Name })
+	for _, resource := range resources {
+		switch resource.Type {
+		case api.CustomResourceTypeRatio, api.CustomResourceTypeCappedRatio,
+			api.CustomResourceTypeCountable:
+			out = append(out, fmt.Sprintf("%s|%d|%s|%d",
+				resource.Name, resource.Version, resource.Type, resource.Quantity))
+
+		case api.CustomResourceTypeStaticInstance, api.CustomResourceTypeDynamicInstance:
+			out = append(out, fmt.Sprintf("%s|%d|%s|%v",
+				resource.Name, resource.Version, resource.Type, resource.Items))
+
+		}
+	}
+	c.Ui.Output(formatList(out))
+}
+
 func (c *NodeStatusCommand) outputNodeStatusEvents(node *api.Node) {
 	c.Ui.Output(c.Colorize().Color("\n[bold]Node Events"))
 	c.outputNodeEvent(node.Events)