storage

type CachedLocalStorage ¶

type CachedLocalStorage[T proto.Message] struct {
	// contains filtered or unexported fields
}

CachedLocalStorage is a storage implementation that persists resources to the local filesystem and maintains an in-memory cache for fast access. Each resource is stored as a JSON file.

It is safe for concurrent access, and the guarantee covers the VALUES as well as the map: the RWMutex alone only ever protected the map, so handing out the cache's own pointers made every consumer of GetAll a potential writer of somebody else's read. That was a real race — the CNI server's termination watch set CNIPod.Terminating in place (under its own lifecycleMu) while the liveness ticker read the same pointer holding no lock at all, and AddResource's protojson.Marshal read it a third time. So every value crossing this boundary is deep-copied: AddResource caches a clone of what it is given, GetResource and GetAll return clones of what they hold. A caller owns what it gets back and may mutate it freely; to make the change stick it writes the value back with AddResource (copy-on-write, which is what the two CNI call sites do).

The copies are not free but they are cheap where it matters: the cached resources are node-local pod records (tens per node), the readers are a 5s liveness tick and a 60s sweep, and AddResource already marshals to JSON and fsyncs a file — a clone is noise next to that.

func NewCachedLocalStorage[T proto.Message](basePath string, newFunc func() T) *CachedLocalStorage[T]

func (f *CachedLocalStorage[T]) AddResource(_ context.Context, key types.ContainerID, resource T) error

func (f *CachedLocalStorage[T]) GetAll(_ context.Context) ([]T, error)

func (f *CachedLocalStorage[T]) GetResource(_ context.Context, key types.ContainerID) (T, error)

func (f *CachedLocalStorage[T]) Initialize(ctx context.Context) error

func (f *CachedLocalStorage[T]) Reload(_ context.Context) (Delta[T], error)

func (f *CachedLocalStorage[T]) RemoveResource(_ context.Context, key types.ContainerID) error

func (f *CachedLocalStorage[T]) WaitUntilReady(ctx context.Context) error

type Delta ¶

type Delta[T proto.Message] struct {
	// Added holds resources present on disk that the previous view did not have.
	Added []T
	// Updated holds resources present in both views whose content differs; the
	// value is the one now on disk.
	Updated []T
	// Removed holds resources the previous view had that are no longer on disk;
	// the value is the one the previous view held (what the caller built from).
	Removed []T
}

Delta is the difference between two views of a storage directory, by key.

func (d Delta[T]) Empty() bool

type MockStorage ¶

type MockStorage[T proto.Message] struct {
	// GetAllFunc is called by GetAll. If nil, GetAll iterates the resources map.
	GetAllFunc func(ctx context.Context) ([]T, error)

	// AddResourceFunc is called by AddResource. If nil, the resource is stored in the map.
	AddResourceFunc func(ctx context.Context, key types.ContainerID, resource T) error

	// GetResourceFunc is called by GetResource. If nil, the resource is looked up in the map.
	// Return os.ErrNotExist (or any error satisfying os.IsNotExist) to simulate a missing file.
	GetResourceFunc func(ctx context.Context, key types.ContainerID) (T, error)

	// RemoveResourceFunc is called by RemoveResource. If nil, the resource is deleted from the map.
	RemoveResourceFunc func(ctx context.Context, key types.ContainerID) error

	// ReloadFunc is called by Reload. If nil, Reload reports no difference.
	ReloadFunc func(ctx context.Context) (Delta[T], error)
	// contains filtered or unexported fields
}

MockStorage is an in-memory Storage implementation intended for use in tests outside the storage package. Because Storage contains the unexported loadAll method, types in other packages cannot implement the interface directly; MockStorage bridges that gap by living in package storage while exposing configurable behaviour via exported function fields.

Like LocalStorage — the implementation it stands in for — MockStorage is safe for concurrent use. Production callers (the CNI server's liveness loop, ghost sweep and drain timers) read and write storage from several goroutines, so a double without that guarantee turns any test of those paths into a data race on the backing map rather than a test of the code under test. The configurable function fields are NOT guarded: set them before the storage is shared.

It also honours the package's ownership contract: the map path deep-copies every value in and out, so a test double can never hand two goroutines the same proto pointer when the real storage would not. A GetAllFunc/ GetResourceFunc override is the test's own business and is returned verbatim.

func NewMockStorage[T proto.Message]() *MockStorage[T]

func NewMockStorageWithGetAll[T proto.Message](fn func(ctx context.Context) ([]T, error)) *MockStorage[T]

func (m *MockStorage[T]) AddResource(ctx context.Context, key types.ContainerID, resource T) error

func (m *MockStorage[T]) GetAll(ctx context.Context) ([]T, error)

func (m *MockStorage[T]) GetResource(ctx context.Context, key types.ContainerID) (T, error)

func (m *MockStorage[T]) Initialize(_ context.Context) error

func (m *MockStorage[T]) Reload(ctx context.Context) (Delta[T], error)

func (m *MockStorage[T]) RemoveResource(ctx context.Context, key types.ContainerID) error

func (m *MockStorage[T]) WaitUntilReady(_ context.Context) error

type Reloader ¶

type Reloader[T proto.Message] interface {
	// Reload re-reads every resource from disk, replaces the in-memory view with
	// it, and returns how it differs from the view it replaced. Every value in
	// the returned Delta is a deep copy the caller owns.
	Reload(ctx context.Context) (Delta[T], error)
}

Reloader is a Storage whose in-memory view can be re-synchronised with disk, reporting what another writer changed since this process last looked.

It exists for the node agent's surge handoff (proposal 041): a standby agent loads storage at its start, but until it takes the node over the OLD agent is the one serving CNI ADD/DEL and writing these files. On takeover the standby reloads and applies only the difference — a full listener rebuild costs 0.25-0.98 s on the reference cluster, inside the window the node's proxy has no ADS stream.

type Storage ¶

type Storage[T proto.Message] interface {
	// Initialize prepares the storage for use, including creating directories
	// and loading initial data if necessary.
	Initialize(ctx context.Context) error
	// WaitUntilReady waits until the storage is ready to serve requests.
	// This may involve waiting for file watching mechanisms to be initialized.
	WaitUntilReady(ctx context.Context) error
	// AddResource stores a new resource with the given key. The implementation
	// keeps a deep copy, so the caller may keep mutating its own value without
	// affecting what is stored.
	AddResource(ctx context.Context, key types.ContainerID, resource T) error
	// RemoveResource deletes a resource by key.
	RemoveResource(ctx context.Context, key types.ContainerID) error
	// GetResource retrieves a single resource by key. The returned message is a
	// deep copy the caller owns: mutating it does not affect the stored resource
	// (write it back with AddResource to persist a change).
	GetResource(ctx context.Context, key types.ContainerID) (T, error)
	// GetAll returns all stored resources from the in-memory cache. Each element
	// is a deep copy the caller owns — see GetResource.
	GetAll(_ context.Context) ([]T, error)
	// contains filtered or unexported methods
}

Storage defines an interface for storing and retrieving protobuf messages. Implementations manage the lifecycle of resources and maintain consistency between persistent storage and in-memory caches.