Documentation
¶
Index ¶
- Constants
- Variables
- func QueuedContainerProfileBuilder() any
- type ErrorCallback
- type QueueConfig
- type QueueData
- func (qd *QueueData) Close() error
- func (qd *QueueData) EmptyQueue() error
- func (qd *QueueData) Enqueue(profile *v1beta1.ContainerProfile, containerID string) error
- func (qd *QueueData) EnqueueWithSizeLimit(profile *v1beta1.ContainerProfile, containerID string, maxProfileSize int64) error
- func (qd *QueueData) GetQueueSize() int
- func (qd *QueueData) GetQueueStats() map[string]any
- func (qd *QueueData) Start()
- type QueuedContainerProfile
Constants ¶
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 )
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 ¶
var ErrQueueNotRunning = errors.New("queue is not running")
ErrQueueNotRunning is returned by enqueueLocked once the queue has been closed.
Functions ¶
func QueuedContainerProfileBuilder ¶
func QueuedContainerProfileBuilder() any
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) EmptyQueue ¶
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) EnqueueWithSizeLimit ¶ added in v0.3.254
func (qd *QueueData) EnqueueWithSizeLimit(profile *v1beta1.ContainerProfile, containerID string, maxProfileSize int64) error
EnqueueWithSizeLimit applies an estimated size budget before sending to storage. The existing splitter preserves the materialized data and report chain; profiles that cannot be split within the depth limit are still offered to storage.
func (*QueueData) GetQueueSize ¶
GetQueueSize returns the current number of items in the queue
func (*QueueData) GetQueueStats ¶
GetQueueStats returns basic statistics about 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"`
// MaxProfileSize is an optional in-memory size estimate for profiles whose
// network peers expanded after collection. Zero disables proactive splitting.
MaxProfileSize int64 `json:"maxProfileSize,omitempty"`
// SplitDepth counts how many times this item's lineage has been halved, either
// proactively for MaxProfileSize or 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
Source Files