queue

package
v0.3.215 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Aug 28, 2026 License: Apache-2.0 Imports: 23 Imported by: 0

Documentation

Index

Constants

View Source
const (
	// DefaultQueueName is the default name for the queue.
	DefaultQueueName = "container-profiles-queue"
	// DefaultRetryInterval is the default interval between retries for processing a queue item.
	DefaultRetryInterval = 5 * time.Second
	// DefaultQueueDir is the default directory for the queue.
	DefaultQueueDir = "/profiles"
	// ItemsPerSegment is the number of items per segment in the queue.
	ItemsPerSegment = 100
	// DefaultMaxQueueSize is the default maximum size of the queue
	DefaultMaxQueueSize = 1000
	// DefaultMaxAttempts is the default number of times a single queued profile is retried
	// before it is dropped. Without a bound, a profile that can never be accepted is
	// retried forever, and because the queue is disk-persistent it survives pod restarts.
	//
	// The bound exists to shed permanently-failing items, not to give up on a reachable
	// storage, so it is deliberately generous: at DefaultRetryInterval this is roughly
	// 30 minutes, which comfortably outlasts a storage rollout, image pull or node
	// eviction. Dropping a profile loses data the container has already stopped tracking,
	// so a tight bound would trade the infinite-retry bug for silent loss during an
	// ordinary restart.
	DefaultMaxAttempts = 360
)
View Source
const DefaultMaxSplitDepth = 4

DefaultMaxSplitDepth bounds how many times a single original chunk may be halved.

With the byte-progress guard in splitProfile this is a tripwire rather than the primary bound: needing depth > 4 implies the chunk overshot storage's cap by more than 16x (a ~40MB delta against a 2.5MB cap), which is a node-agent estimator bug that should surface as a Warning promptly instead of being absorbed by split storms. It also caps one lineage's peak queue occupancy at 16 leaves; enforceMaxSize evicts FIFO from the head while splits append to the tail, so an unbounded lineage would silently evict the oldest in-flight halves of *other* lineages.

Variables

View Source
var ErrQueueNotRunning = errors.New("queue is not running")

ErrQueueNotRunning is returned by enqueueLocked once the queue has been closed.

Functions

func QueuedContainerProfileBuilder

func QueuedContainerProfileBuilder() interface{}

QueuedContainerProfileBuilder creates a new QueuedContainerProfile instance for dque

Types

type ErrorCallback

type ErrorCallback interface {
	OnQueueError(profile *v1beta1.ContainerProfile, containerID string, err error)
}

ErrorCallback defines the interface for handling queue processing errors

type QueueConfig

type QueueConfig struct {
	QueueName       string
	QueueDir        string
	MaxQueueSize    int
	MaxAttempts     int
	MaxSplitDepth   int
	RetryInterval   time.Duration
	ItemsPerSegment int
	ErrorCallback   ErrorCallback
	MetricsManager  metricsmanager.MetricsManager
}

QueueConfig holds configuration for the queue

type QueueData

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

QueueData holds the data and configuration for the queue processing.

func NewQueueData

func NewQueueData(ctx context.Context, creator storage.ProfileCreator, config QueueConfig) (*QueueData, error)

NewQueueData creates a new QueueData instance with simple LRU behavior

func (*QueueData) Close

func (qd *QueueData) Close() error

Close gracefully shuts down the queue

func (*QueueData) EmptyQueue

func (qd *QueueData) EmptyQueue() error

EmptyQueue clears all items from the queue

func (*QueueData) Enqueue

func (qd *QueueData) Enqueue(profile *v1beta1.ContainerProfile, containerID string) error

Enqueue adds a new container profile to the queue with LRU eviction

func (*QueueData) GetQueueSize

func (qd *QueueData) GetQueueSize() int

GetQueueSize returns the current number of items in the queue

func (*QueueData) GetQueueStats

func (qd *QueueData) GetQueueStats() map[string]interface{}

GetQueueStats returns basic statistics about the queue

func (*QueueData) Start

func (qd *QueueData) Start()

Start begins processing the queue

type QueuedContainerProfile

type QueuedContainerProfile struct {
	Profile     *v1beta1.ContainerProfile `json:"profile"`
	ContainerID string                    `json:"containerID"`
	// Attempts counts how many times creation of this profile has failed with a
	// retryable error. Items persisted before this field existed decode with Attempts
	// at zero, so they simply get a full budget of retries.
	Attempts int `json:"attempts"`
	// SplitDepth counts how many times this item's lineage has been halved after an HTTP 413.
	// Items persisted before this field existed decode with SplitDepth at zero.
	SplitDepth int `json:"splitDepth"`
	// IsStitch marks a metadata-only chunk emitted in place of a chunk that was dropped or
	// LRU-evicted (see dropChunk and enforceMaxSize) to repair the report chain. A stitch is
	// never split and is never itself stitched - dropping a stitch leaves the chain forked,
	// which is strictly better than an unbounded stitch->drop->stitch loop.
	IsStitch bool `json:"isStitch"`
}

QueuedContainerProfile represents a container profile queued for creation

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL