Represents a client for communicating with a single cluster.
type ClusterClient interface {
// Gets the name of the cluster.
ClusterName() string
// Commits the scheduling decision to the cluster.
CommitSchedulingDecision(ctx context.Context, schedulingDecision *ClusterSchedulingDecision) (*CommitSchedulingDecisionSuccess, error)
}
Manages the clients for multiple clusters.
type ClusterClientsManager interface {
// Gets the ClusterClient for the specified cluster or an error, if the specified cluster cannot be found.
GetClusterClient(clusterName string) (ClusterClient, error)
// Gets the number of clusters known to this ClusterClientsManager.
ClustersCount() int
// Calls the specified function for each cluster.
// If the function returns an error, the iteration is stopped immediately and returns that error.
ForEach(fn func(clusterName string, clusterClient ClusterClient) error) error
}
Augments a node with information computed by the Polaris Scheduling framework.
type ClusterNode struct {
*core.Node `json:",inline" yaml:",inline"`
// The pods that are already scheduled on (bound to) this node.
Pods []*ClusterPod
// The pods that are queued to be bound to this node.
// These pods are currently in the binding pipeline, but their resources are already accounted for in
// the node's AvailableResources, because committing a scheduling decision may take some time.
QueuedPods []*ClusterPod
// The resources that are currently available for allocation on the node.
//
// Unlike the Kubernetes Allocatable field, these AvailableResources already accounts for
// resources consumed by other pods.
AvailableResources *util.Resources `json:"availableResources" yaml:"availableResources"`
// The total amount of resources that are available on the node.
TotalResources *util.Resources `json:"totalResources" yaml:"totalResources"`
// The Unix timestamp that indicates when the last pod was added to this node.
// If no pods are scheduled on the node, this value is 0.
LastPodAddedTimestamp int64 `json:"lastPodAddedTimestamp" yaml:"lastPodAddedTimestamp"`
}
func NewClusterNode(node *core.Node) *ClusterNode
Creates a new cluster node, based on the specified node object, assuming that no pods are scheduled on it yet.
func NewClusterNodeWithPods(node *core.Node, pods []*ClusterPod, queuedPods []*ClusterPod, lastPodAddedTimestamp int64) *ClusterNode
Creates a new cluster node, based on the specified node object and the pods that are already scheduled and queued on it.
func (cn *ClusterNode) ShallowCopy() *ClusterNode
Creates a shallow copy of this ClusterNode, i.e., a new object, whose fields point to the same objects as the source object.
Condensed information about an existing pod in a cluster.
type ClusterPod struct {
// The namespace of the pod.
Namespace string
// The name of the pod.
Name string
// The total resources consumed by this pod.
TotalResources *util.Resources
// Affinity/anti-affinity information.
// This is nil, if not present on the pod.
Affinity *core.Affinity
}
func NewClusterPod(pod *core.Pod) *ClusterPod
Creates a new ClusterPod, based on the specified pod object.
Contains the scheduling decision for a pod within a cluster.
type ClusterSchedulingDecision struct {
// The Pod to be scheduled.
Pod *core.Pod `json:"pod" yaml:"pod"`
// The name of the node, to which the pod has been assigned.
NodeName string `json:"nodeName" yaml:"nodeName"`
}
Encapsulates the success result of committing a SchedulingDecision.
type CommitSchedulingDecisionSuccess struct {
// The namespace of the pod.
Namespace string `json:"namespace" yaml:"namespace"`
// The name of the pod.
PodName string `json:"podName" yaml:"podName"`
// The name of the target node, to which the pod was bound.
NodeName string `json:"nodeName" yaml:"nodeName"`
// Timings of the commit operation on the polaris-cluster-agent.
// Note that when using a LocalClusterClient, only the CreatePod and CreateBinding fields are filled.
Timings *CommitSchedulingDecisionTimings `json:"timings" yaml:"timings"`
}
Describes timings (in milliseconds) of various phases of the CommitSchedulingDecision operation on the polaris-cluster-agent.
type CommitSchedulingDecisionTimings struct {
// The time spent in the queue waiting for a binding pipeline to become available.
QueueTime int64 `json:"queueTime" yaml:"queueTime"`
// The time spent waiting for the node to be locked for binding.
NodeLockTime int64 `json:"nodeLockTime" yaml:"nodeLockTime"`
// The time it takes to fetch the target node and its assigned pods.
FetchNodeInfo int64 `json:"fetchNodeInfo" yaml:"fetchNodeInfo"`
// The duration of the binding pipeline
BindingPipeline int64 `json:"bindingPipeline" yaml:"bindingPipeline"`
// Commit decision is the entire time it takes to commit a decision using the local cluster client.
// The commit involves CreatePod, CreateBinding, and any calling overheads.
CommitDecision int64 `json:"commitDecision" yaml:"commitDecision"`
// The duration of the request to create a Pod in the orchestrator.
CreatePod int64 `json:"createPod" yaml:"createPod"`
// The duration of the request to bind the pod to the target node in the orchestrator.
CreateBinding int64 `json:"createBinding" yaml:"createBinding"`
}
A generic default implementation of the ClusterClientsManager.
type GenericClusterClientsManager[T ClusterClient] struct { // contains filtered or unexported fields }
func NewGenericClusterClientsManager[T ClusterClient](clients map[string]T) *GenericClusterClientsManager[T]
func (mgr *GenericClusterClientsManager[T]) ClustersCount() int
func (mgr *GenericClusterClientsManager[T]) ForEach(fn func(clusterName string, clusterClient ClusterClient) error) error
func (mgr *GenericClusterClientsManager[T]) GetClusterClient(clusterName string) (ClusterClient, error)
A superset of ClusterClient with more capabilities and which is only available in the ClusterAgent.
type LocalClusterClient interface {
ClusterClient
// Fetches the node with the specified name.
FetchNode(ctx context.Context, name string) (*core.Node, error)
// Fetches all pods that are currently scheduled on the node with the specified name.
//
// ToDo: Refactor this into a fetch method with more generic search criteria and maybe return an array of pointers.
FetchPodsScheduledOnNode(ctx context.Context, nodeName string) ([]core.Pod, error)
}
Tracks and caches the full list of nodes in a cluster and their available resources. All nodes obtained through the cache must be treated as IMMUTABLE, because they are shared objects. If modifications need to be made to a node, a partially deep copy must be made.
The node objects in the cache are always the most recent ones received through the watch.
As pods are assigned to nodes by the scheduler, it is recommended to use the QueuePodOnNode() method to update the node's resources in the cache before committing the scheduling decision to the cluster, as this may take some time, during which the node's status in the cache would be outdated.
type NodesCache interface {
// Starts watching the nodes.
// This method returns once the initial list has been retrieved and added to the store.
//
// The passed context can be used to stop the watch.
StartWatch(ctx context.Context) error
// Gets the cache of all nodes.
Nodes() collections.ConcurrentObjectStore[*ClusterNode]
// Adds the pod as to be bound to the specified node and updates the node's available resources.
//
// Use the returned PodQueuedOnNode to inform the cache of the outcome of the commit scheduling decision operation.
QueuePodOnNode(pod *core.Pod, nodeName string) PodQueuedOnNode
}
Handle to a pod that was queued to be bound to a node. This object is returned by NodesCache.QueuePodOnNode() and must be used to inform the cache of the final result of the commit operation.
type PodQueuedOnNode interface {
// The pod that was queued (immutable).
Pod() *ClusterPod
// The name of the node, on which the pod was queued.
NodeName() string
// Removes this pod from the node's queue, without marking it as committed,
// and updates the node's resources.
RemoveFromQueue()
// Marks the pod as committed, i.e., moves it from the queue to the list of pods running on the node.
MarkAsCommitted()
}
A generic DTO for transmitting error information.
type PolarisErrorDto struct {
Message string `json:"message" yaml:"message"`
}
func NewPolarisErrorDto(err error) *PolarisErrorDto
func (e *PolarisErrorDto) Error() string
Error implements error