provisioning

Documentation

Overview

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Index

• Variables
• type Batcher
  • func NewBatcher(running context.Context) *Batcher
  • func (b *Batcher) Add(item interface{}) <-chan struct{}
  • func (b *Batcher) Flush()
  • func (b *Batcher) Wait() (items []interface{}, window time.Duration)
• type Controller
  • func NewController(ctx context.Context, kubeClient client.Client, ...) *Controller
  • func (c *Controller) Apply(ctx context.Context, provisioner *v1alpha5.Provisioner) error
  • func (c *Controller) Delete(name string)
  • func (c *Controller) List(ctx context.Context) []*Provisioner
  • func (c *Controller) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error)
  • func (c *Controller) Register(_ context.Context, m manager.Manager) error
• type Provisioner
  • func NewProvisioner(ctx context.Context, provisioner *v1alpha5.Provisioner, ...) *Provisioner
  • func (p *Provisioner) Add(pod *v1.Pod) <-chan struct{}

Variables

var (
	MaxBatchDuration  = time.Second * 10
	BatchIdleDuration = time.Second * 1
	// MaxItemsPerBatch limits the number of items we process at one time to avoid using too much memory
	MaxItemsPerBatch = 2_000
)

Types

type Batcher

type Batcher struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Batcher separates a stream of Add(item) calls into windowed slices. The window is dynamic and will be extended if additional items are added up to a maximum batch duration or maximum items per batch.

func NewBatcher

func NewBatcher(running context.Context) *Batcher

NewBatcher is a constructor

func (*Batcher) Add

func (b *Batcher) Add(item interface{}) <-chan struct{}

Add an item to the batch, returning the next gate which the caller may block on. The gate is protected by a read-write mutex, and may be modified by Flush(), which makes a new gate.

In rare scenarios, if a goroutine hangs after enqueueing but before acquiring the gate lock, the batch could be flushed, resulting in the pod waiting on the next gate. This will be flushed on the next batch, and may result in delayed retries for the individual pod if the provisioning loop fails. In practice, this won't be encountered because this time window is O(seconds).

func (*Batcher) Flush

func (b *Batcher) Flush()

Flush all goroutines blocking on the current gate and create a new gate.

func (*Batcher) Wait

func (b *Batcher) Wait() (items []interface{}, window time.Duration)

Wait starts a batching window and returns a slice of items when closed.

type Controller

type Controller struct {
	// contains filtered or unexported fields
}

Controller for the resource

func NewController

func NewController(ctx context.Context, kubeClient client.Client, coreV1Client corev1.CoreV1Interface, cloudProvider cloudprovider.CloudProvider) *Controller

NewController is a constructor

func (*Controller) Apply

func (c *Controller) Apply(ctx context.Context, provisioner *v1alpha5.Provisioner) error

Apply creates or updates the provisioner to the latest configuration

func (*Controller) Delete

func (c *Controller) Delete(name string)

Delete stops and removes a provisioner. Enqueued pods will be provisioned.

func (*Controller) List

func (c *Controller) List(ctx context.Context) []*Provisioner

List active provisioners in order of priority

func (*Controller) Reconcile

func (c *Controller) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error)

Reconcile a control loop for the resource

func (*Controller) Register

func (c *Controller) Register(_ context.Context, m manager.Manager) error

Register the controller to the manager

type Provisioner

type Provisioner struct {
	// State
	*v1alpha5.Provisioner

	Stop context.CancelFunc
	// contains filtered or unexported fields
}

Provisioner waits for enqueued pods, batches them, creates capacity and binds the pods to the capacity.

func NewProvisioner

func NewProvisioner(ctx context.Context, provisioner *v1alpha5.Provisioner, kubeClient client.Client, coreV1Client corev1.CoreV1Interface, cloudProvider cloudprovider.CloudProvider) *Provisioner

func (*Provisioner) Add

func (p *Provisioner) Add(pod *v1.Pod) <-chan struct{}

Add a pod to the provisioner and return a channel to block on. The caller is responsible for verifying that the pod was scheduled correctly.