datalayer

type AttributeMap ¶

type AttributeMap interface {
	Put(string, Cloneable)
	Get(string) (Cloneable, bool)
	Keys() []string
	Clone() AttributeMap
}

AttributeMap is used to store flexible metadata or traits across different aspects of an inference server. Stored values must be Cloneable.

func NewAttributes() AttributeMap

type Attributes ¶

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

Attributes provides a goroutine-safe implementation of AttributeMap.

func (a *Attributes) Clone() AttributeMap

func (a *Attributes) Get(key string) (Cloneable, bool)

func (a *Attributes) Keys() []string

func (a *Attributes) Put(key string, value Cloneable)

type Cloneable ¶

type Cloneable interface {
	Clone() Cloneable
}

Cloneable types support cloning of the value.

type DataSource ¶

type DataSource interface {
	plugin.Plugin
}

DataSource provides raw data to registered Extractors. Concrete variants are PollingDispatcher (poll-driven), NotificationSource (k8s-event-driven), and EndpointSource (lifecycle-event-driven).

type DiscoveryEndpointStore ¶

type DiscoveryEndpointStore interface {
	EndpointUpsert(ctx context.Context, meta *EndpointMetadata)
	EndpointDelete(id types.NamespacedName)
}

DiscoveryEndpointStore is the narrow interface required by NewDiscoveryNotifier. Any store that implements EndpointUpsert and EndpointDelete satisfies it.

type DiscoveryNotifier ¶

type DiscoveryNotifier interface {
	// Upsert adds or updates an endpoint in the datastore.
	Upsert(endpoint *EndpointMetadata)
	// Delete removes an endpoint from the datastore by its namespaced name.
	Delete(id types.NamespacedName)
}

DiscoveryNotifier is the callback through which EndpointDiscovery communicates endpoint state to the datastore.

DiscoveryNotifier is NOT goroutine-safe. All calls must be made sequentially from a single goroutine. This is the source of the ordering contract below.

Ordering contract: the datastore processes Upsert and Delete calls in the order they are received. Plugin implementations MUST preserve event order -- do not buffer or dispatch calls concurrently in a way that could reorder them. For example, an Upsert followed by a Delete for the same endpoint must arrive in that order, or the endpoint will be incorrectly left in the datastore.

func NewDiscoveryNotifier(store DiscoveryEndpointStore) DiscoveryNotifier

type DynamicAttribute ¶

type DynamicAttribute struct {
	Get func() Cloneable
}

DynamicAttribute wraps a getter function to allow on-demand resolution of attributes.

func (d *DynamicAttribute) Clone() Cloneable

type Endpoint ¶

type Endpoint interface {
	fmt.Stringer
	EndpointMetaState
	EndpointMetricsState
}

Endpoint represents an inference serving endpoint and its related attributes.

type EndpointDiscovery ¶

type EndpointDiscovery interface {
	fwkplugin.Plugin

	// Start begins discovery and blocks in the caller's goroutine until ctx is
	// cancelled or a fatal error occurs. It is the caller's responsibility to
	// invoke Start in a dedicated goroutine.
	//
	// Implementations SHOULD enumerate all currently known endpoints via
	// notifier.Upsert before entering the watch loop, to avoid serving an empty
	// datastore at startup. Implementations that guarantee no missed events
	// through their watch mechanism (e.g. a Kubernetes list+watch) may fold the
	// initial enumeration into the watch sequence instead.
	Start(ctx context.Context, notifier DiscoveryNotifier) error

	// Ready returns a channel that is closed once after the plugin has completed
	// its initial reconciliation with the underlying source, regardless of how
	// many endpoints that produced. Callers use it to gate request-serving
	// components until the datastore has been populated for the first time.
	//
	// Contract:
	//   - The channel is closed at most once per Start invocation.
	//   - It is closed only after a successful initial sync. If Start returns
	//     an error before that point, the channel remains open and callers
	//     waiting on it should also observe ctx cancellation.
	//   - "Initial sync" means whatever the plugin considers a complete first
	//     pass (e.g. file load for FileDiscovery; first list+watch reconcile
	//     for a Kubernetes plugin). Zero endpoints is a valid outcome.
	Ready() <-chan struct{}
}

EndpointDiscovery discovers inference endpoints and drives their lifecycle in the datastore. Implementations are registered in the plugin registry and selected via EndpointPickerConfig.discovery.pluginRef.

type EndpointEvent ¶

type EndpointEvent struct {
	Type     EventType
	Endpoint Endpoint
}

EndpointEvent carries an endpoint lifecycle event.

type EndpointExtractor ¶

type EndpointExtractor = Extractor[EndpointEvent]

EndpointExtractor is the typed contract for endpoint-lifecycle extractors.

type EndpointMetaState ¶

type EndpointMetaState interface {
	GetMetadata() *EndpointMetadata
	UpdateMetadata(*EndpointMetadata)
	GetAttributes() AttributeMap
}

EndpointMetaState allows management of the EndpointMetadata related attributes.

type EndpointMetadata ¶

type EndpointMetadata struct {
	NamespacedName types.NamespacedName
	PodName        string
	Address        string
	Port           string
	MetricsHost    string
	Labels         map[string]string
	// RankIndex is this endpoint's position in the pool's TargetPorts,
	// identifying the pod-local rank in multi-port deployments.
	RankIndex int
}

EndpointMetadata represents the relevant Kubernetes Pod state of an inference server.

func (epm *EndpointMetadata) Clone() *EndpointMetadata

func (epm *EndpointMetadata) Equal(other *EndpointMetadata) bool

func (epm *EndpointMetadata) GetIPAddress() string

func (epm *EndpointMetadata) GetMetricsHost() string

func (epm *EndpointMetadata) GetNamespacedName() types.NamespacedName

func (epm *EndpointMetadata) GetPort() string

func (epm *EndpointMetadata) GetRankIndex() int

func (epm *EndpointMetadata) String() string

type EndpointMetricsState ¶

type EndpointMetricsState interface {
	GetMetrics() *Metrics
	UpdateMetrics(*Metrics)
}

EndpointMetricsState allows management of the Metrics related attributes.

type EndpointSource ¶

type EndpointSource interface {
	DataSource
	// NotifyEndpoint is called by the Runtime on each endpoint lifecycle event.
	// Returns nil event to skip extractor dispatch.
	NotifyEndpoint(ctx context.Context, event EndpointEvent) (*EndpointEvent, error)
}

EndpointSource is an event-driven DataSource driven by endpoint lifecycle changes.

type EventType ¶

type EventType int

EventType identifies the type of mutation that triggered a notification.

const (
	// EventAddOrUpdate is fired when a k8s object is created or updated.
	EventAddOrUpdate EventType = iota
	// EventDelete is fired when a k8s object is deleted.
	EventDelete
)

type Extractor ¶

type Extractor[T any] interface {
	plugin.Plugin
	Extract(ctx context.Context, in T) error
}

Extractor transforms typed input T into endpoint attributes. T pins the dispatch payload:

• Polling extractors: T = PollInput[D] (paired with a PollingDispatcher)
• Notification extractors: T = NotificationEvent (also implement NotificationExtractor for GVK)
• Endpoint extractors: T = EndpointEvent

type Metrics ¶

type Metrics struct {
	// ActiveModels is a set of models(including LoRA adapters) that are currently cached to GPU.
	ActiveModels  map[string]int
	WaitingModels map[string]int
	// MaxActiveModels is the maximum number of models that can be loaded to GPU.
	MaxActiveModels         int
	RunningRequestsSize     int
	WaitingQueueSize        int
	KVCacheUsagePercent     float64
	KvCacheMaxTokenCapacity int
	CacheBlockSize          int
	// Number of GPU blocks in the model server for KV Cache.
	CacheNumBlocks int

	// UpdateTime records the last time when the metrics were updated.
	UpdateTime time.Time
}

Metrics holds the latest metrics snapshot scraped from a pod.

func NewMetrics() *Metrics

func (m *Metrics) Clone() *Metrics

func (m *Metrics) String() string

type MissingPolicy ¶

type MissingPolicy int

MissingPolicy controls behavior when a required source type is absent from the config.

const (
	// Fail returns an error if the required source type is not configured (default).
	Fail MissingPolicy = iota
	// Warn logs a warning and skips wiring; the plugin degrades gracefully.
	Warn
)

type ModelServer ¶

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

ModelServer is an implementation of the Endpoint interface.

func NewEndpoint(meta *EndpointMetadata, metrics *Metrics) *ModelServer

func (srv *ModelServer) Clone() *ModelServer

func (srv *ModelServer) GetAttributes() AttributeMap

func (srv *ModelServer) GetMetadata() *EndpointMetadata

func (srv *ModelServer) GetMetrics() *Metrics

func (srv *ModelServer) String() string

func (srv *ModelServer) UpdateMetadata(pod *EndpointMetadata)

func (srv *ModelServer) UpdateMetrics(metrics *Metrics)

type NotificationEvent ¶

type NotificationEvent struct {
	Type   EventType
	Object *unstructured.Unstructured
}

NotificationEvent carries the event type and the affected object. Object is deep-copied by the framework core before delivery.

type NotificationExtractor ¶

type NotificationExtractor interface {
	Extractor[NotificationEvent]
	GVK() schema.GroupVersionKind
}

NotificationExtractor is the typed contract for k8s-event extractors. GVK identifies the kind this extractor handles; it must match the paired NotificationSource's GVK.

type NotificationSource ¶

type NotificationSource interface {
	DataSource
	// GVK returns the GroupVersionKind this source watches.
	GVK() schema.GroupVersionKind
	// Notify is called by the framework core when a mutation event fires.
	// The event object is already deep-copied.
	// Returns the event (possibly modified) for Runtime to dispatch to extractors.
	// Returns nil event to signal Runtime to skip extractor dispatch.
	// TODO: why accept event but return *event?
	Notify(ctx context.Context, event NotificationEvent) (*NotificationEvent, error)
}

NotificationSource is an event-driven DataSource for a single k8s GVK.

type PendingRegistration ¶

type PendingRegistration struct {
	Owner         plugin.TypedName // registering plugin; used as map key + error context
	SourceType    string           // TypedName.Type to match, e.g. "endpoint-notification-source"
	Extractor     plugin.Plugin    // required; framework type-asserts to the matched source's variant
	DefaultSource DataSource       // nil → no auto-create; non-nil → create if absent
	IfMissing     MissingPolicy    // applies only when DefaultSource is nil
}

PendingRegistration describes a (source-type, extractor) dependency. Extractor must be non-nil; registering a DataSource without an Extractor is not supported. If DefaultSource is nil, IfMissing governs behavior when no matching source exists. If DefaultSource is non-nil, it is registered as a new source when no match is found; for NotificationSources its GVK() narrows the match.

type PollInput ¶

type PollInput[D any] struct {
	Payload  D
	Endpoint Endpoint
}

PollInput pairs the typed poll payload with the endpoint being polled. Payload is the parser's result and is expected to be usable when Poll returns a nil error.

type PollingDispatcher ¶

type PollingDispatcher interface {
	plugin.Plugin
	Dispatch(ctx context.Context, ep Endpoint) error
	AppendExtractor(ext plugin.Plugin) error
}

PollingDispatcher is the framework's contract for polling sources. The source owns its extractors and runs them with typed input each tick.

Contract:

• Dispatch runs bound extractors in AppendExtractor-insertion order.
• Each Poll and each Extract step runs under its own timeout.
• Non-nil return = poll failure; per-extractor failures record DataLayerExtractErrorsTotal and do NOT surface as the return error.
• AppendExtractor is a pure append; duplicate-Type detection is the caller's responsibility (see runtime.Configure).

type PollingExtractor ¶

type PollingExtractor[T any] = Extractor[PollInput[T]]

PollingExtractor is the typed contract for poll-based extractors.

type Registrant ¶

type Registrant interface {
	RegisterDependencies(Registrar) error
}

Registrant is an optional interface any Plugin may implement to declare its datalayer dependencies. The runner calls RegisterDependencies on all eligible plugins after instantiation, before Runtime.Configure().

type Registrar ¶

type Registrar interface {
	Register(PendingRegistration) error
}

Registrar accepts pending source/extractor dependency declarations from plugins. Runtime.Configure() resolves them after processing user config. Register returns an error if the registration is invalid (e.g. nil Extractor).