manager

package
v0.0.0-...-e9fa201 Latest Latest
Warning

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

 Go to latest Published: Mar 16, 2026 License: Apache-2.0 Imports: 55 Imported by: 0

Documentation

Overview

Package manager provides stub region configuration for community edition. Multi-region support is an enterprise feature.

Package manager provides the manager server for ZapFS cluster coordination.

The manager server handles: - Service registry (file and metadata services) - Placement decisions (where to store chunks) - Raft consensus for high availability - IAM credential management and sync to metadata services - Collection (bucket) management - GC coordination

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrNotLeader          = status.Errorf(codes.FailedPrecondition, "not the leader")
	ErrRaftNotInitialized = status.Errorf(codes.FailedPrecondition, "raft not initialized")
)
View Source 
var (
	// ErrUserNotFound is returned when a user is not found.
	ErrUserNotFound = errors.New("user not found")
	// ErrUserExists is returned when trying to create a user that already exists.
	ErrUserExists = errors.New("user already exists")
	// ErrCredentialNotFound is returned when a credential is not found.
	ErrCredentialNotFound = errors.New("credential not found")
)
View Source 
var (
	// RaftState tracks the current Raft state (0=Follower, 1=Candidate, 2=Leader, 3=Shutdown)
	RaftState = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: "zapfs",
		Subsystem: "raft",
		Name:      "state",
		Help:      "Current Raft state (0=Follower, 1=Candidate, 2=Leader, 3=Shutdown)",
	})

	// RaftIsLeader is 1 if this node is the leader, 0 otherwise
	RaftIsLeader = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: "zapfs",
		Subsystem: "raft",
		Name:      "is_leader",
		Help:      "1 if this node is the Raft leader, 0 otherwise",
	})

	// RaftTerm tracks the current Raft term
	RaftTerm = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: "zapfs",
		Subsystem: "raft",
		Name:      "term",
		Help:      "Current Raft term",
	})

	// RaftCommitIndex tracks the commit index
	RaftCommitIndex = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: "zapfs",
		Subsystem: "raft",
		Name:      "commit_index",
		Help:      "Current Raft commit index",
	})

	// RaftAppliedIndex tracks the last applied index
	RaftAppliedIndex = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: "zapfs",
		Subsystem: "raft",
		Name:      "applied_index",
		Help:      "Last applied Raft index",
	})

	// RaftLastLogIndex tracks the last log index
	RaftLastLogIndex = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: "zapfs",
		Subsystem: "raft",
		Name:      "last_log_index",
		Help:      "Last Raft log index",
	})

	// RaftLeaderChanges counts leader election changes
	RaftLeaderChanges = prometheus.NewCounter(prometheus.CounterOpts{
		Namespace: "zapfs",
		Subsystem: "raft",
		Name:      "leader_changes_total",
		Help:      "Total number of leader changes observed",
	})

	// RaftPeers tracks the number of peers in the cluster
	RaftPeers = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: "zapfs",
		Subsystem: "raft",
		Name:      "peers",
		Help:      "Number of peers in the Raft cluster",
	})

	// RaftFSMApplyLatency tracks FSM apply latency
	RaftFSMApplyLatency = prometheus.NewHistogram(prometheus.HistogramOpts{
		Namespace: "zapfs",
		Subsystem: "raft",
		Name:      "fsm_apply_duration_seconds",
		Help:      "Duration of FSM apply operations",
		Buckets:   []float64{0.0001, 0.0005, 0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1},
	})

	// RaftSnapshotLatency tracks snapshot creation latency
	RaftSnapshotLatency = prometheus.NewHistogram(prometheus.HistogramOpts{
		Namespace: "zapfs",
		Subsystem: "raft",
		Name:      "snapshot_duration_seconds",
		Help:      "Duration of Raft snapshot operations",
		Buckets:   []float64{0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120},
	})

	// ManagerDataLossDetected is 1 if manager detected potential data loss
	// (metadata services have buckets but manager has none)
	ManagerDataLossDetected = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: "zapfs",
		Subsystem: "manager",
		Name:      "data_loss_detected",
		Help:      "1 if potential data loss detected (metadata has buckets but manager has 0 collections), 0 otherwise",
	})

	// ManagerRecoveryRequired is 1 if bucket creates are frozen pending recovery
	ManagerRecoveryRequired = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: "zapfs",
		Subsystem: "manager",
		Name:      "recovery_required",
		Help:      "1 if bucket creates are frozen pending data recovery, 0 otherwise",
	})

	// ManagerCollectionsTotal tracks the total number of collections
	ManagerCollectionsTotal = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: "zapfs",
		Subsystem: "manager",
		Name:      "collections_total",
		Help:      "Total number of collections (buckets) in the manager",
	})

	// TopologyEventsDropped counts events dropped due to slow subscribers
	TopologyEventsDropped = prometheus.NewCounter(prometheus.CounterOpts{
		Namespace: "zapfs",
		Subsystem: "manager",
		Name:      "topology_events_dropped_total",
		Help:      "Total number of topology events dropped due to slow subscribers (channel full)",
	})

	// IAMEventsDropped counts IAM events dropped due to slow subscribers
	IAMEventsDropped = prometheus.NewCounter(prometheus.CounterOpts{
		Namespace: "zapfs",
		Subsystem: "manager",
		Name:      "iam_events_dropped_total",
		Help:      "Total number of IAM credential events dropped due to slow subscribers",
	})

	// CollectionEventsDropped counts collection events dropped due to slow subscribers
	CollectionEventsDropped = prometheus.NewCounter(prometheus.CounterOpts{
		Namespace: "zapfs",
		Subsystem: "manager",
		Name:      "collection_events_dropped_total",
		Help:      "Total number of collection events dropped due to slow subscribers",
	})
)
View Source 
var ErrMultiRegionNotAvailable = errors.New("multi-region support requires enterprise license")

ErrMultiRegionNotAvailable is returned when multi-region features are used in community edition.

Functions

func AdminProtectedMethods 

func AdminProtectedMethods() []string

AdminProtectedMethods returns a copy of the protected methods list.

func HasExistingRaftState 

func HasExistingRaftState(dataDir string) bool

HasExistingRaftState checks if there's existing Raft state in the data directory. This is used to determine if a node is restarting (has state) vs joining fresh (no state).

func NewAdminAuthInterceptor 

func NewAdminAuthInterceptor(token string, protectedMethods []string) grpc.UnaryServerInterceptor

NewAdminAuthInterceptor creates a gRPC unary interceptor that requires a bearer token for admin-level RPCs. Non-admin RPCs pass through. If token is empty, all requests pass through (auth disabled).

func TrackLeaderChange 

func TrackLeaderChange(newLeader string)

TrackLeaderChange should be called when leader changes are observed

func UpdateRaftMetrics 

func UpdateRaftMetrics(stats map[string]string, isLeader bool, numPeers int)

UpdateRaftMetrics updates Raft metrics from the Raft stats map. This should be called periodically (e.g., every second) by the manager server.

Types

type BackupManifest 

type BackupManifest struct {
	ID                 string    `json:"id"`
	CreatedAt          time.Time `json:"created_at"`
	TopologyVersion    uint64    `json:"topology_version"`
	CollectionsVersion uint64    `json:"collections_version"`
	FileServicesCount  int       `json:"file_services_count"`
	MetadataServices   int       `json:"metadata_services_count"`
	CollectionsCount   int       `json:"collections_count"`
	SizeBytes          int64     `json:"size_bytes"`
	Filename           string    `json:"filename"`
	Checksum           string    `json:"checksum,omitempty"`
	Description        string    `json:"description,omitempty"`
	Scheduled          bool      `json:"scheduled"` // true if created by scheduler
}

BackupManifest describes a backup file

type BackupScheduler 

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

BackupScheduler handles automatic periodic backups

func NewBackupScheduler 

func NewBackupScheduler(manager *ManagerServer, config BackupSchedulerConfig) *BackupScheduler

NewBackupScheduler creates a new backup scheduler

func (*BackupScheduler) DeleteBackup 

func (s *BackupScheduler) DeleteBackup(backupID string) error

DeleteBackup deletes a backup by ID

func (*BackupScheduler) GetBackup 

func (s *BackupScheduler) GetBackup(backupID string) (*BackupManifest, error)

GetBackup returns a specific backup manifest

func (*BackupScheduler) ListBackups 

func (s *BackupScheduler) ListBackups() []BackupManifest

ListBackups returns all known backups

func (*BackupScheduler) Start 

func (s *BackupScheduler) Start() error

Start begins the backup scheduler loop

func (*BackupScheduler) Stop 

func (s *BackupScheduler) Stop()

Stop stops the backup scheduler and waits for completion

type BackupSchedulerConfig 

type BackupSchedulerConfig struct {
	// Enabled flag
	Enabled bool

	// Schedule interval for automatic backups (default: 24h)
	Interval time.Duration

	// Directory to store backups (local path)
	// Future: support s3://bucket/prefix or gs://bucket/prefix
	DestinationDir string

	// Retention policy
	RetainCount int   // Number of backups to retain (default: 7)
	RetainDays  int   // Days to retain backups (default: 30)
	MaxSize     int64 // Max total backup size in bytes (0 = unlimited)

	// Backup naming
	Prefix string // Prefix for backup files (default: "manager")
}

BackupSchedulerConfig configures automatic backup scheduling

func DefaultBackupSchedulerConfig 

func DefaultBackupSchedulerConfig() BackupSchedulerConfig

DefaultBackupSchedulerConfig returns the default backup scheduler configuration

type ClusterCapacity 

type ClusterCapacity struct {
	TotalBytes   uint64    // Sum of all storage backend capacities
	UsedBytes    uint64    // Sum of all used storage
	UpdatedAt    time.Time // Last update time
	FileServices int       // Number of active file services
}

ClusterCapacity holds cached cluster-wide capacity metrics. Updated via heartbeats from file services (30s stale max).

type CommandType 

type CommandType string

Commands that go through Raft 

const (
	CommandRegisterFederation          CommandType = "register_federation"
	CommandUnregisterFederation        CommandType = "unregister_federation"
	CommandSetFederationMode           CommandType = "set_federation_mode"
	CommandPauseFederation             CommandType = "pause_federation"
	CommandResumeFederation            CommandType = "resume_federation"
	CommandSetFederationDualWrite      CommandType = "set_federation_dual_write"
	CommandUpdateFederationCredentials CommandType = "update_federation_credentials"
)
const (
	CommandRegisterService     CommandType = "register_service"
	CommandUnregisterService   CommandType = "unregister_service"
	CommandUpdatePolicy        CommandType = "update_policy"
	CommandCreateCollection    CommandType = "create_collection"
	CommandDeleteCollection    CommandType = "delete_collection"
	CommandUpdateCollection    CommandType = "update_collection"
	CommandUpdateServiceStatus CommandType = "update_service_status"

	// IAM commands
	CommandIAMCreateUser   CommandType = "iam_create_user"
	CommandIAMUpdateUser   CommandType = "iam_update_user"
	CommandIAMDeleteUser   CommandType = "iam_delete_user"
	CommandIAMCreateKey    CommandType = "iam_create_key"
	CommandIAMDeleteKey    CommandType = "iam_delete_key"
	CommandIAMCreatePolicy CommandType = "iam_create_policy"
	CommandIAMDeletePolicy CommandType = "iam_delete_policy"
)

type Config 

type Config struct {
	// Node identity
	NodeID   string
	BindAddr string
	DataDir  string

	// Raft tuning
	HeartbeatTimeout time.Duration
	ElectionTimeout  time.Duration
	CommitTimeout    time.Duration
	MaxAppendEntries int

	// Snapshot
	SnapshotRetain int

	// Cluster
	Bootstrap       bool
	BootstrapExpect int

	// TLS for Raft transport (optional — nil means plaintext)
	TLSConfig *tls.Config
}

type CredentialNotifier 

type CredentialNotifier interface {
	NotifyCredentialChange(eventType iam_pb.CredentialEvent_EventType, identity *iam.Identity, cred *iam.Credential)
}

CredentialNotifier is called when credentials are created/updated/deleted to notify streaming subscribers (metadata services)

type ExternalS3ClientPool 

type ExternalS3ClientPool struct {
	*s3client.Pool
}

ExternalS3ClientPool manages a pool of S3 clients for different endpoints. This is used by the Manager to verify external bucket credentials and perform bucket operations during federation registration.

func NewExternalS3ClientPool 

func NewExternalS3ClientPool(timeout time.Duration, maxIdleConns int) *ExternalS3ClientPool

NewExternalS3ClientPool creates a new client pool with the given timeout and max idle connections.

func (*ExternalS3ClientPool) CountObjects 

func (p *ExternalS3ClientPool) CountObjects(ctx context.Context, cfg *ExternalS3Config, bucket string) (int64, error)

CountObjects returns the number of objects in a bucket. This is a rough count for migration progress estimation.

func (*ExternalS3ClientPool) DeleteBucket 

func (p *ExternalS3ClientPool) DeleteBucket(ctx context.Context, cfg *ExternalS3Config, bucket string) error

DeleteBucket deletes a bucket on the external S3 service. Returns nil if successful or if bucket doesn't exist (404).

func (*ExternalS3ClientPool) HeadBucket 

func (p *ExternalS3ClientPool) HeadBucket(ctx context.Context, cfg *ExternalS3Config, bucket string) error

HeadBucket checks if a bucket exists and the credentials are valid. Returns nil if successful, error otherwise.

func (*ExternalS3ClientPool) ListObjects 

func (p *ExternalS3ClientPool) ListObjects(ctx context.Context, cfg *ExternalS3Config, bucket, prefix, startAfter string, maxKeys int32) ([]string, string, error)

ListObjects lists objects in a bucket (for discovery during migration). Returns object keys and a continuation token for pagination.

type ExternalS3Config 

type ExternalS3Config struct {
	Endpoint        string
	Region          string
	AccessKeyID     string
	SecretAccessKey string
	PathStyle       bool
}

ExternalS3Config holds configuration for connecting to an external S3 service.

type FSMState 

type FSMState struct {

	// Service registry - tracks registered file and metadata services
	FileServices     map[string]*ServiceRegistration
	MetadataServices map[string]*ServiceRegistration
	TopologyVersion  uint64

	// Collections (storage namespaces / buckets)
	Collections        map[string]*manager_pb.Collection
	CollectionsVersion uint64

	// Derived indexes - rebuilt from Collections on restore
	// These are not persisted in snapshots but reconstructed
	CollectionsByOwner   map[string][]string
	OwnerCollectionCount map[string]int
	CollectionsByTier    map[string][]string
	CollectionsByTime    *btree.BTree

	// Configuration replicated via Raft
	PlacementPolicy *manager_pb.PlacementPolicy
	RegionID        string

	// IAM state - users, groups, policies replicated via Raft
	// Credentials are stored with encrypted secrets (EncryptedSecret field)
	IAMUsers    map[string]*iam.Identity // name -> identity
	IAMPolicies map[string]*iam.Policy   // name -> policy
	IAMVersion  uint64                   // Monotonic version counter for IAM changes

	// Derived IAM indexes - rebuilt on restore
	IAMCredentialIndex map[string]string // accessKey -> userName (for fast lookup)

	// Federation state - stores external S3 connection configs for federated buckets
	FederationConfigs map[string]*FederationInfo // bucket -> federation config
	// contains filtered or unexported fields
}

FSMState holds all state that is replicated through Raft consensus. This struct isolates replicated state from server concerns, making snapshots cleaner and state management more explicit.

All fields in this struct are persisted in Raft snapshots and must be consistent across all nodes in the cluster.

func NewFSMState 

func NewFSMState(regionID string, defaultNumReplicas uint32) *FSMState

NewFSMState creates a new FSMState with initialized maps.

func (*FSMState) AddCollectionToIndexes 

func (s *FSMState) AddCollectionToIndexes(col *manager_pb.Collection)

AddCollectionToIndexes adds a collection to the derived indexes. Must be called with the write lock held.

func (*FSMState) AddIAMPolicy 

func (s *FSMState) AddIAMPolicy(policy *iam.Policy)

AddIAMPolicy adds a policy to the IAM state. Must be called with the write lock held.

func (*FSMState) AddIAMUser 

func (s *FSMState) AddIAMUser(identity *iam.Identity)

AddIAMUser adds a user to the IAM state. Must be called with the write lock held.

func (*FSMState) GetCollectionsByOwner 

func (s *FSMState) GetCollectionsByOwner(owner string) []string

GetCollectionsByOwner returns collection names for a given owner. Must be called with at least a read lock held.

func (*FSMState) GetCollectionsByTier 

func (s *FSMState) GetCollectionsByTier(tier string) []string

GetCollectionsByTier returns collection names for a given tier. Must be called with at least a read lock held.

func (*FSMState) GetCollectionsByTime 

func (s *FSMState) GetCollectionsByTime() *btree.BTree

GetCollectionsByTime returns the btree for time-based lookups. Must be called with at least a read lock held.

func (*FSMState) GetIAMPolicy 

func (s *FSMState) GetIAMPolicy(id string) (*iam.Policy, bool)

GetIAMPolicy returns a policy by ID. Must be called with at least a read lock held.

func (*FSMState) GetIAMUser 

func (s *FSMState) GetIAMUser(name string) (*iam.Identity, bool)

GetIAMUser returns a user by name. Must be called with at least a read lock held.

func (*FSMState) GetIAMUserByAccessKey 

func (s *FSMState) GetIAMUserByAccessKey(accessKey string) (string, bool)

GetIAMUserByAccessKey returns the user name for an access key. Must be called with at least a read lock held.

func (*FSMState) GetIAMVersion 

func (s *FSMState) GetIAMVersion() uint64

GetIAMVersion returns the current IAM version counter. Must be called with at least a read lock held.

func (*FSMState) GetOwnerCollectionCount 

func (s *FSMState) GetOwnerCollectionCount(owner string) int

GetOwnerCollectionCount returns the number of collections for a given owner. Must be called with at least a read lock held.

func (*FSMState) GetRegistry 

func (s *FSMState) GetRegistry(serviceType manager_pb.ServiceType) map[string]*ServiceRegistration

GetRegistry returns the appropriate service registry based on service type.

func (*FSMState) HasIAMUsers 

func (s *FSMState) HasIAMUsers() bool

HasIAMUsers returns true if there are any IAM users in the state. Must be called with at least a read lock held.

func (*FSMState) Lock 

func (s *FSMState) Lock()

Lock acquires the write lock on the state.

func (*FSMState) RLock 

func (s *FSMState) RLock()

RLock acquires the read lock on the state.

func (*FSMState) RUnlock 

func (s *FSMState) RUnlock()

RUnlock releases the read lock on the state.

func (*FSMState) RemoveCollectionFromIndexes 

func (s *FSMState) RemoveCollectionFromIndexes(col *manager_pb.Collection)

RemoveCollectionFromIndexes removes a collection from the derived indexes. Must be called with the write lock held.

func (*FSMState) RemoveIAMPolicy 

func (s *FSMState) RemoveIAMPolicy(policyID string)

RemoveIAMPolicy removes a policy from the IAM state. Must be called with the write lock held.

func (*FSMState) RemoveIAMUser 

func (s *FSMState) RemoveIAMUser(userName string)

RemoveIAMUser removes a user from the IAM state. Must be called with the write lock held.

func (*FSMState) Restore 

func (s *FSMState) Restore(snapshot *fsmSnapshot)

Restore restores state from a snapshot. Must be called with the write lock held.

func (*FSMState) Snapshot 

func (s *FSMState) Snapshot() *fsmSnapshot

Snapshot creates a point-in-time snapshot of the Raft state. This is called by the FSM Snapshot method. Must be called with at least a read lock held.

func (*FSMState) Unlock 

func (s *FSMState) Unlock()

Unlock releases the write lock on the state.

func (*FSMState) UpdateIAMUser 

func (s *FSMState) UpdateIAMUser(identity *iam.Identity)

UpdateIAMUser updates an existing user in the IAM state. Must be called with the write lock held.

type FederationConfig 

type FederationConfig struct {
	Enabled              bool
	ExternalTimeout      time.Duration
	ExternalMaxIdleConns int
}

FederationConfig holds federation configuration for the manager.

type FederationCredentialsUpdate 

type FederationCredentialsUpdate struct {
	Bucket          string `json:"bucket"`
	AccessKeyID     string `json:"access_key_id"`
	SecretAccessKey string `json:"secret_access_key"`
	UpdatedAt       int64  `json:"updated_at"`
}

FederationCredentialsUpdate is the Raft command for updating credentials.

type FederationDualWrite 

type FederationDualWrite struct {
	Bucket    string `json:"bucket"`
	Enabled   bool   `json:"enabled"`
	ChangedAt int64  `json:"changed_at"`
}

FederationDualWrite is the Raft command for enabling/disabling dual-write.

type FederationInfo 

type FederationInfo struct {
	External          *manager_pb.ExternalS3Config `json:"external"`
	ObjectsDiscovered int64                        `json:"objects_discovered"`
	ObjectsSynced     int64                        `json:"objects_synced"`
	BytesSynced       int64                        `json:"bytes_synced"`
	MigrationPaused   bool                         `json:"migration_paused"`
	DualWriteEnabled  bool                         `json:"dual_write_enabled"`
	CreatedAt         int64                        `json:"created_at"`
	UpdatedAt         int64                        `json:"updated_at"`
}

FederationInfo stores federation configuration in Raft state.

type FederationModeChange 

type FederationModeChange struct {
	Bucket                   string                          `json:"bucket"`
	NewMode                  manager_pb.FederationBucketMode `json:"new_mode"`
	DeleteExternalOnComplete bool                            `json:"delete_external_on_complete"`
	ChangedAt                int64                           `json:"changed_at"`
}

FederationModeChange is the Raft command for changing bucket mode.

type FederationPauseResume 

type FederationPauseResume struct {
	Bucket    string `json:"bucket"`
	Paused    bool   `json:"paused"`
	ChangedAt int64  `json:"changed_at"`
}

FederationPauseResume is the Raft command for pausing/resuming migration.

type FederationRegistration 

type FederationRegistration struct {
	LocalBucket          string                          `json:"local_bucket"`
	Mode                 manager_pb.FederationBucketMode `json:"mode"`
	External             *manager_pb.ExternalS3Config    `json:"external"`
	ObjectsDiscovered    int64                           `json:"objects_discovered"`
	StartActiveMigration bool                            `json:"start_active_migration"`
	RegisteredAt         int64                           `json:"registered_at"`
}

FederationRegistration is the Raft command for registering a federated bucket.

type FederationService 

type FederationService struct {
	manager_pb.UnimplementedFederationServiceServer
	// contains filtered or unexported fields
}

FederationService implements the gRPC FederationService for bucket federation and migration control. All write operations are forwarded to the Raft leader.

func NewFederationService 

func NewFederationService(ms *ManagerServer, cfg *FederationConfig) *FederationService

NewFederationService creates a new FederationService.

func (*FederationService) GetBucketStatus 

func (s *FederationService) GetBucketStatus(ctx context.Context, req *manager_pb.FederationGetBucketStatusRequest) (*manager_pb.FederationGetBucketStatusResponse, error)

GetBucketStatus returns the federation status for a bucket.

func (*FederationService) ListFederatedBuckets 

func (s *FederationService) ListFederatedBuckets(ctx context.Context, req *manager_pb.FederationListBucketsRequest) (*manager_pb.FederationListBucketsResponse, error)

ListFederatedBuckets returns all federated buckets.

func (*FederationService) PauseMigration 

func (s *FederationService) PauseMigration(ctx context.Context, req *manager_pb.FederationPauseMigrationRequest) (*manager_pb.FederationPauseMigrationResponse, error)

PauseMigration pauses an active migration.

func (*FederationService) RegisterBucket 

func (s *FederationService) RegisterBucket(ctx context.Context, req *manager_pb.FederationRegisterBucketRequest) (*manager_pb.FederationRegisterBucketResponse, error)

RegisterBucket registers an existing external S3 bucket for federation.

func (*FederationService) ResumeMigration 

func (s *FederationService) ResumeMigration(ctx context.Context, req *manager_pb.FederationResumeMigrationRequest) (*manager_pb.FederationResumeMigrationResponse, error)

ResumeMigration resumes a paused migration.

func (*FederationService) SetBucketMode 

func (s *FederationService) SetBucketMode(ctx context.Context, req *manager_pb.FederationSetBucketModeRequest) (*manager_pb.FederationSetBucketModeResponse, error)

SetBucketMode changes the bucket mode (e.g., passthrough → migrating → local).

func (*FederationService) SetDualWrite 

func (s *FederationService) SetDualWrite(ctx context.Context, req *manager_pb.FederationSetDualWriteRequest) (*manager_pb.FederationSetDualWriteResponse, error)

SetDualWrite enables or disables dual-write for a bucket in migrating mode.

func (*FederationService) UnregisterBucket 

func (s *FederationService) UnregisterBucket(ctx context.Context, req *manager_pb.FederationUnregisterBucketRequest) (*manager_pb.FederationUnregisterBucketResponse, error)

UnregisterBucket removes federation configuration from a bucket.

func (*FederationService) UpdateCredentials 

func (s *FederationService) UpdateCredentials(ctx context.Context, req *manager_pb.FederationUpdateCredentialsRequest) (*manager_pb.FederationUpdateCredentialsResponse, error)

UpdateCredentials updates the external S3 credentials for a federated bucket.

type FederationUnregistration 

type FederationUnregistration struct {
	Bucket          string `json:"bucket"`
	DeleteExternal  bool   `json:"delete_external"`
	DeleteLocalData bool   `json:"delete_local_data"`
	UnregisteredAt  int64  `json:"unregistered_at"`
}

FederationUnregistration is the Raft command for unregistering a federated bucket.

type IAMAdminConfig 

type IAMAdminConfig struct {
	// EnableAuth enables basic authentication for the admin UI/API
	EnableAuth bool
	// AdminUsername is the username for basic auth (default: admin)
	AdminUsername string
	// AdminPassword is the password for basic auth (required if EnableAuth is true)
	AdminPassword string
}

IAMAdminConfig holds configuration for the IAM admin handler

type IAMAdminHandler 

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

IAMAdminHandler provides HTTP endpoints for IAM administration. The admin UI is served at /v1/iam/ui/ and the API at /v1/iam/*.

Security considerations: - The admin port (8060) should not be exposed publicly - Use network policies to restrict access to internal IPs - Optionally enable basic auth with IAMAdminConfig

func NewIAMAdminHandler 

func NewIAMAdminHandler(iamService *iam.Service) *IAMAdminHandler

NewIAMAdminHandler creates a new IAM admin handler

func NewIAMAdminHandlerWithConfig 

func NewIAMAdminHandlerWithConfig(iamService *iam.Service, config IAMAdminConfig) *IAMAdminHandler

NewIAMAdminHandlerWithConfig creates a new IAM admin handler with custom config

func (*IAMAdminHandler) ServeHTTP 

func (h *IAMAdminHandler) ServeHTTP(w http.ResponseWriter, r *http.Request)

func (*IAMAdminHandler) SetNotifier 

func (h *IAMAdminHandler) SetNotifier(n CredentialNotifier)

SetNotifier sets the credential notifier for streaming updates

func (*IAMAdminHandler) SetRaftStore 

func (h *IAMAdminHandler) SetRaftStore(store *RaftCredentialStore)

SetRaftStore sets the Raft-backed credential store for mutations. When set, all user/credential mutations go through Raft consensus.

type IAMCreateKeyRequest 

type IAMCreateKeyRequest struct {
	UserName   string          `json:"user_name"`
	Credential *iam.Credential `json:"credential"`
}

IAMCreateKeyRequest is the Raft command data for creating an access key.

type IAMCreatePolicyRequest 

type IAMCreatePolicyRequest struct {
	Policy *iam.Policy `json:"policy"`
}

IAMCreatePolicyRequest is the Raft command data for creating a policy.

type IAMCreateUserRequest 

type IAMCreateUserRequest struct {
	Identity *iam.Identity `json:"identity"`
}

IAMCreateUserRequest is the Raft command data for creating a user.

type IAMDeleteKeyRequest 

type IAMDeleteKeyRequest struct {
	UserName  string `json:"user_name"`
	AccessKey string `json:"access_key"`
}

IAMDeleteKeyRequest is the Raft command data for deleting an access key.

type IAMDeletePolicyRequest 

type IAMDeletePolicyRequest struct {
	PolicyID string `json:"policy_id"`
}

IAMDeletePolicyRequest is the Raft command data for deleting a policy.

type IAMDeleteUserRequest 

type IAMDeleteUserRequest struct {
	UserName string `json:"user_name"`
}

IAMDeleteUserRequest is the Raft command data for deleting a user.

type LeaderForwarder 

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

LeaderForwarder manages gRPC connections to the leader for request forwarding. It caches the leader connection and refreshes it when the leader changes.

func NewLeaderForwarder 

func NewLeaderForwarder(adminToken string) *LeaderForwarder

NewLeaderForwarder creates a new leader forwarder. adminToken is optional; if non-empty it will be sent as a Bearer token when forwarding admin RPCs to the leader.

func (*LeaderForwarder) Close 

func (lf *LeaderForwarder) Close()

Close closes any open connections

func (*LeaderForwarder) ForwardCreateCollection 

func (lf *LeaderForwarder) ForwardCreateCollection(ctx context.Context, leaderAddr string, req *manager_pb.CreateCollectionRequest) (*manager_pb.CreateCollectionResponse, error)

ForwardCreateCollection forwards a CreateCollection request to the leader

func (*LeaderForwarder) ForwardDeleteCollection 

func (lf *LeaderForwarder) ForwardDeleteCollection(ctx context.Context, leaderAddr string, req *manager_pb.DeleteCollectionRequest) (*manager_pb.DeleteCollectionResponse, error)

ForwardDeleteCollection forwards a DeleteCollection request to the leader

func (*LeaderForwarder) ForwardFederationPauseMigration 

func (lf *LeaderForwarder) ForwardFederationPauseMigration(ctx context.Context, leaderAddr string, req *manager_pb.FederationPauseMigrationRequest) (*manager_pb.FederationPauseMigrationResponse, error)

ForwardFederationPauseMigration forwards a PauseMigration request to the leader

func (*LeaderForwarder) ForwardFederationRegisterBucket 

func (lf *LeaderForwarder) ForwardFederationRegisterBucket(ctx context.Context, leaderAddr string, req *manager_pb.FederationRegisterBucketRequest) (*manager_pb.FederationRegisterBucketResponse, error)

ForwardFederationRegisterBucket forwards a RegisterBucket request to the leader

func (*LeaderForwarder) ForwardFederationResumeMigration 

func (lf *LeaderForwarder) ForwardFederationResumeMigration(ctx context.Context, leaderAddr string, req *manager_pb.FederationResumeMigrationRequest) (*manager_pb.FederationResumeMigrationResponse, error)

ForwardFederationResumeMigration forwards a ResumeMigration request to the leader

func (*LeaderForwarder) ForwardFederationSetBucketMode 

func (lf *LeaderForwarder) ForwardFederationSetBucketMode(ctx context.Context, leaderAddr string, req *manager_pb.FederationSetBucketModeRequest) (*manager_pb.FederationSetBucketModeResponse, error)

ForwardFederationSetBucketMode forwards a SetBucketMode request to the leader

func (*LeaderForwarder) ForwardFederationSetDualWrite 

func (lf *LeaderForwarder) ForwardFederationSetDualWrite(ctx context.Context, leaderAddr string, req *manager_pb.FederationSetDualWriteRequest) (*manager_pb.FederationSetDualWriteResponse, error)

ForwardFederationSetDualWrite forwards a SetDualWrite request to the leader

func (*LeaderForwarder) ForwardFederationUnregisterBucket 

func (lf *LeaderForwarder) ForwardFederationUnregisterBucket(ctx context.Context, leaderAddr string, req *manager_pb.FederationUnregisterBucketRequest) (*manager_pb.FederationUnregisterBucketResponse, error)

ForwardFederationUnregisterBucket forwards an UnregisterBucket request to the leader

func (*LeaderForwarder) ForwardFederationUpdateCredentials 

func (lf *LeaderForwarder) ForwardFederationUpdateCredentials(ctx context.Context, leaderAddr string, req *manager_pb.FederationUpdateCredentialsRequest) (*manager_pb.FederationUpdateCredentialsResponse, error)

ForwardFederationUpdateCredentials forwards an UpdateCredentials request to the leader

func (*LeaderForwarder) ForwardRaftAddServer 

func (lf *LeaderForwarder) ForwardRaftAddServer(ctx context.Context, leaderAddr string, req *manager_pb.RaftAddServerRequest) (*manager_pb.RaftAddServerResponse, error)

ForwardRaftAddServer forwards a RaftAddServer request to the leader

func (*LeaderForwarder) ForwardRaftRemoveServer 

func (lf *LeaderForwarder) ForwardRaftRemoveServer(ctx context.Context, leaderAddr string, req *manager_pb.RaftRemoveServerRequest) (*manager_pb.RaftRemoveServerResponse, error)

ForwardRaftRemoveServer forwards a RaftRemoveServer request to the leader

func (*LeaderForwarder) ForwardRegisterService 

func (lf *LeaderForwarder) ForwardRegisterService(ctx context.Context, leaderAddr string, req *manager_pb.RegisterServiceRequest) (*manager_pb.RegisterServiceResponse, error)

ForwardRegisterService forwards a RegisterService request to the leader

func (*LeaderForwarder) ForwardTriggerReconciliation 

func (lf *LeaderForwarder) ForwardTriggerReconciliation(ctx context.Context, leaderAddr string, req *manager_pb.TriggerReconciliationRequest) (*manager_pb.TriggerReconciliationResponse, error)

ForwardTriggerReconciliation forwards a TriggerReconciliation request to the leader

func (*LeaderForwarder) ForwardUnregisterService 

func (lf *LeaderForwarder) ForwardUnregisterService(ctx context.Context, leaderAddr string, req *manager_pb.UnregisterServiceRequest) (*manager_pb.UnregisterServiceResponse, error)

ForwardUnregisterService forwards an UnregisterService request to the leader

type ManagerServer 

type ManagerServer struct {
	manager_pb.UnimplementedManagerServiceServer
	iam_pb.UnimplementedIAMServiceServer
	// contains filtered or unexported fields
}

ManagerServer is the main manager server that coordinates the ZapFS cluster. It implements both ManagerServiceServer (cluster management) and IAMServiceServer (credential sync to metadata services).

Lock ordering (to prevent deadlocks): 

state.Lock() > collectionsRecoveryRequiredMu > clusterCapacityMu

The state lock must always be acquired first. The secondary mutexes (collectionsRecoveryRequiredMu, clusterCapacityMu) protect independent caches and may be acquired while state is held, but never the reverse.

func NewManagerServer 

func NewManagerServer(regionID string, raftConfig *Config, leaderTimeout time.Duration, iamService *iam.Service, licenseChecker license.Checker, defaultNumReplicas uint32, adminToken string) (*ManagerServer, error)

NewManagerServer creates a new manager server with the given configuration

func (*ManagerServer) Apply 

func (ms *ManagerServer) Apply(l *raft.Log) interface{}

Apply applies a Raft log entry to the FSM (state machine)

func (*ManagerServer) BootstrapIAMFromConfig 

func (ms *ManagerServer) BootstrapIAMFromConfig(ctx context.Context) error

BootstrapIAMFromConfig bootstraps IAM credentials from the legacy config (TOML) into Raft. This should be called once on the leader after cluster initialization. If Raft already has IAM users, this is a no-op.

func (*ManagerServer) CalculateRebalancePlan 

func (ms *ManagerServer) CalculateRebalancePlan(ctx context.Context, req *manager_pb.CalculateRebalancePlanRequest) (*manager_pb.CalculateRebalancePlanResponse, error)

CalculateRebalancePlan calculates a plan to rebalance chunks across file servers.

func (*ManagerServer) ClearRecoveryMode 

func (ms *ManagerServer) ClearRecoveryMode()

ClearRecoveryMode clears the recovery required flag after successful recovery. This should only be called after explicit recovery operation.

func (*ManagerServer) ConfigureBackupScheduler 

func (ms *ManagerServer) ConfigureBackupScheduler(config BackupSchedulerConfig) error

ConfigureBackupScheduler sets up automatic backup scheduling. This is an enterprise feature requiring FeatureBackup license.

func (*ManagerServer) ConfigureMultiRegion 

func (ms *ManagerServer) ConfigureMultiRegion(regionConfig *RegionConfig) error

ConfigureMultiRegion sets up multi-region support for the manager server. This enables cross-region bucket forwarding and cache synchronization. This is an enterprise feature - in community edition, this is a no-op.

func (*ManagerServer) CreateBackup 

func (ms *ManagerServer) CreateBackup(ctx context.Context, req *manager_pb.CreateBackupRequest) (*manager_pb.CreateBackupResponse, error)

CreateBackup exports the current Raft FSM state as a JSON snapshot. Enterprise feature - requires FeatureBackup license.

func (*ManagerServer) CreateCollection 

func (ms *ManagerServer) CreateCollection(ctx context.Context, req *manager_pb.CreateCollectionRequest) (*manager_pb.CreateCollectionResponse, error)

CreateCollection creates a new storage collection (similar to S3 bucket) This is a write operation that forwards to leader if not leader. In multi-region mode, secondary regions forward to the primary region.

func (*ManagerServer) DecryptFederationCredentials 

func (ms *ManagerServer) DecryptFederationCredentials(fedInfo *FederationInfo) (accessKeyID, secretAccessKey string, err error)

DecryptFederationCredentials decrypts the AccessKeyId and SecretAccessKey from a FederationInfo that was stored in encrypted form in Raft state. Returns the decrypted accessKeyID and secretAccessKey. If no masterKey is set or the values are not encrypted, returns them as-is.

Note: The metadata service's DB layer (vitess/postgres) has its own independent encryption via iam.EncryptCredentialSecret/DecryptCredentialSecret. This method is for code that reads directly from the Raft FSM state (e.g., manager-side federation operations, diagnostics, or snapshot inspection).

func (*ManagerServer) DeleteBackup 

func (ms *ManagerServer) DeleteBackup(ctx context.Context, req *manager_pb.DeleteBackupRequest) (*manager_pb.DeleteBackupResponse, error)

DeleteBackup deletes a backup by ID

func (*ManagerServer) DeleteCollection 

func (ms *ManagerServer) DeleteCollection(ctx context.Context, req *manager_pb.DeleteCollectionRequest) (*manager_pb.DeleteCollectionResponse, error)

DeleteCollection deletes a collection This is a write operation that forwards to leader if not leader. In multi-region mode, secondary regions forward to the primary region.

func (*ManagerServer) ExecuteRebalance 

func (ms *ManagerServer) ExecuteRebalance(req *manager_pb.ExecuteRebalanceRequest, stream manager_pb.ManagerService_ExecuteRebalanceServer) error

ExecuteRebalance executes a rebalance plan, streaming progress updates.

func (*ManagerServer) GetActiveFileServiceCount 

func (ms *ManagerServer) GetActiveFileServiceCount() int

GetActiveFileServiceCount returns the number of active file services.

func (*ManagerServer) GetBackupStatus 

func (ms *ManagerServer) GetBackupStatus(ctx context.Context, req *manager_pb.GetBackupStatusRequest) (*manager_pb.GetBackupStatusResponse, error)

GetBackupStatus returns current FSM state info. Useful for monitoring and backup verification.

func (*ManagerServer) GetClusterCapacity 

func (ms *ManagerServer) GetClusterCapacity() ClusterCapacity

GetClusterCapacity returns the cached cluster capacity metrics. This is O(1) and safe for frequent reads.

func (*ManagerServer) GetClusterState 

func (ms *ManagerServer) GetClusterState() map[string]interface{}

GetClusterState returns information about the current cluster state

func (*ManagerServer) GetClusterStatus 

func (ms *ManagerServer) GetClusterStatus(ctx context.Context, req *manager_pb.GetClusterStatusRequest) (*manager_pb.GetClusterStatusResponse, error)

GetClusterStatus returns the current status of all file servers in the cluster.

func (*ManagerServer) GetCollection 

func (ms *ManagerServer) GetCollection(ctx context.Context, req *manager_pb.GetCollectionRequest) (*manager_pb.GetCollectionResponse, error)

GetCollection retrieves a specific collection by name This is a read operation and can be served by any node

func (*ManagerServer) GetCollectionsByOwner 

func (ms *ManagerServer) GetCollectionsByOwner(owner string) []string

GetCollectionsByOwner returns collection names for an owner (must hold state lock) Results are already sorted

func (*ManagerServer) GetCredential 

func (ms *ManagerServer) GetCredential(ctx context.Context, req *iam_pb.GetCredentialRequest) (*iam_pb.GetCredentialResponse, error)

GetCredential retrieves a credential by access key

func (*ManagerServer) GetExpectedChunks 

func (ms *ManagerServer) GetExpectedChunks(req *manager_pb.GetExpectedChunksRequest, stream manager_pb.ManagerService_GetExpectedChunksServer) error

GetExpectedChunks streams chunk IDs expected on a file server. The file server calls this to get the list of chunks it should have. Manager queries a metadata service and streams the results.

func (*ManagerServer) GetIAMService 

func (ms *ManagerServer) GetIAMService() *iam.Service

GetIAMService returns the IAM service

func (*ManagerServer) GetIAMVersion 

func (ms *ManagerServer) GetIAMVersion(ctx context.Context, req *iam_pb.GetIAMVersionRequest) (*iam_pb.GetIAMVersionResponse, error)

GetIAMVersion returns the current IAM data version

func (*ManagerServer) GetOwnerCollectionCount 

func (ms *ManagerServer) GetOwnerCollectionCount(owner string) int

GetOwnerCollectionCount returns number of collections owned by an owner (must hold state lock)

func (*ManagerServer) GetRaftCredentialStore 

func (ms *ManagerServer) GetRaftCredentialStore() *RaftCredentialStore

GetRaftCredentialStore returns the Raft-backed credential store

func (*ManagerServer) GetReplicationTargets 

func (ms *ManagerServer) GetReplicationTargets(ctx context.Context, req *manager_pb.GetReplicationTargetsRequest) (*manager_pb.GetReplicationTargetsResponse, error)

func (*ManagerServer) HasLeader 

func (ms *ManagerServer) HasLeader() bool

HasLeader returns true if the Raft cluster currently has an elected leader.

func (*ManagerServer) Heartbeat 

func (ms *ManagerServer) Heartbeat(ctx context.Context, req *manager_pb.HeartbeatRequest) (*manager_pb.HeartbeatResponse, error)

func (*ManagerServer) IsClusterReady 

func (ms *ManagerServer) IsClusterReady() bool

IsClusterReady returns true if the Raft cluster has an elected leader This is used for readiness checks - the service should not accept requests until the cluster is operational

func (*ManagerServer) IsLeader 

func (ms *ManagerServer) IsLeader() bool

IsLeader returns true if this node is the current Raft leader.

func (*ManagerServer) IsPrimaryRegion 

func (ms *ManagerServer) IsPrimaryRegion() bool

IsPrimaryRegion returns true if this manager is in the primary region. In single-region mode (no multi-region config), this always returns true.

func (*ManagerServer) IsRecoveryRequired 

func (ms *ManagerServer) IsRecoveryRequired() bool

IsRecoveryRequired returns true if the manager detected data loss and bucket creates are frozen pending recovery.

func (*ManagerServer) ListBackups 

func (ms *ManagerServer) ListBackups(ctx context.Context, req *manager_pb.ListBackupsRequest) (*manager_pb.ListBackupsResponse, error)

ListBackups returns all available backups

func (*ManagerServer) ListCollections 

func (ms *ManagerServer) ListCollections(req *manager_pb.ListCollectionsRequest, stream manager_pb.ManagerService_ListCollectionsServer) error

ListCollections lists all collections, optionally filtered by owner This is a read operation and can be served by any node (leader or follower)

func (*ManagerServer) ListCredentials 

func (ms *ManagerServer) ListCredentials(req *iam_pb.ListCredentialsRequest, stream iam_pb.IAMService_ListCredentialsServer) error

ListCredentials returns all credentials for initial sync

func (*ManagerServer) LogClusterCapacity 

func (ms *ManagerServer) LogClusterCapacity()

LogClusterCapacity logs the current cluster capacity in human-readable format.

func (*ManagerServer) NotifyCredentialChange 

func (ms *ManagerServer) NotifyCredentialChange(eventType iam_pb.CredentialEvent_EventType, identity *iam.Identity, cred *iam.Credential)

NotifyCredentialChange broadcasts a credential change to all subscribers. Called by IAM admin handlers when credentials are created/updated/deleted.

func (*ManagerServer) Ping 

func (ms *ManagerServer) Ping(ctx context.Context, req *manager_pb.PingRequest) (*manager_pb.PingResponse, error)

func (*ManagerServer) RaftAddServer 

func (ms *ManagerServer) RaftAddServer(ctx context.Context, req *manager_pb.RaftAddServerRequest) (*manager_pb.RaftAddServerResponse, error)

func (*ManagerServer) RaftListClusterServers 

func (ms *ManagerServer) RaftListClusterServers(ctx context.Context, req *manager_pb.RaftListClusterServersRequest) (*manager_pb.RaftListClusterServersResponse, error)

func (*ManagerServer) RaftRemoveServer 

func (ms *ManagerServer) RaftRemoveServer(ctx context.Context, req *manager_pb.RaftRemoveServerRequest) (*manager_pb.RaftRemoveServerResponse, error)

func (*ManagerServer) RecoverCollections 

func (ms *ManagerServer) RecoverCollections(ctx context.Context, req *manager_pb.RecoverCollectionsRequest) (*manager_pb.RecoverCollectionsResponse, error)

RecoverCollections implements the disaster recovery RPC. It queries metadata services for their bucket lists and rebuilds the manager's collection registry through Raft.

func (*ManagerServer) RegisterService 

func (ms *ManagerServer) RegisterService(ctx context.Context, req *manager_pb.RegisterServiceRequest) (*manager_pb.RegisterServiceResponse, error)

func (*ManagerServer) ReportReconciliation 

func (ms *ManagerServer) ReportReconciliation(ctx context.Context, req *manager_pb.ReconciliationReport) (*manager_pb.ReconciliationAck, error)

ReportReconciliation receives reconciliation results from a file server.

func (*ManagerServer) Restore 

func (ms *ManagerServer) Restore(rc io.ReadCloser) error

Restore restores the FSM from a snapshot

func (*ManagerServer) SetMasterKey 

func (ms *ManagerServer) SetMasterKey(key []byte)

SetMasterKey sets the encryption key for federation secrets in Raft state. When set, federation credentials (AccessKeyId, SecretAccessKey) are encrypted before being stored in the FSM state and decrypted when read. Must be called before the Raft node processes any federation commands. Can only be called once; subsequent calls are no-ops. Pass nil to disable encryption (credentials stored in plaintext).

func (*ManagerServer) Shutdown 

func (ms *ManagerServer) Shutdown()

Shutdown gracefully shuts down the manager server

func (*ManagerServer) Snapshot 

func (ms *ManagerServer) Snapshot() (raft.FSMSnapshot, error)

Snapshot returns a snapshot of the FSM state

func (*ManagerServer) StreamAllCollections 

func (ms *ManagerServer) StreamAllCollections(callback func(*manager_pb.Collection) bool)

StreamAllCollections streams all collections in time order

func (*ManagerServer) StreamCollectionsSince 

func (ms *ManagerServer) StreamCollectionsSince(since time.Time, callback func(*manager_pb.Collection) bool)

StreamCollectionsSince streams collections created since a given timestamp Used for gRPC streaming API

func (*ManagerServer) StreamCredentials 

func (ms *ManagerServer) StreamCredentials(req *iam_pb.StreamCredentialsRequest, stream iam_pb.IAMService_StreamCredentialsServer) error

StreamCredentials streams credential changes to metadata services in real-time

func (*ManagerServer) TriggerReconciliation 

func (ms *ManagerServer) TriggerReconciliation(ctx context.Context, req *manager_pb.TriggerReconciliationRequest) (*manager_pb.TriggerReconciliationResponse, error)

TriggerReconciliation manually triggers reconciliation for file servers. If server_id is empty, triggers for all servers.

func (*ManagerServer) UnregisterService 

func (ms *ManagerServer) UnregisterService(ctx context.Context, req *manager_pb.UnregisterServiceRequest) (*manager_pb.UnregisterServiceResponse, error)

func (*ManagerServer) WaitForClusterReady 

func (ms *ManagerServer) WaitForClusterReady(timeout time.Duration) error

WaitForClusterReady blocks until the cluster has a leader or context is cancelled

func (*ManagerServer) WatchCollections 

func (ms *ManagerServer) WatchCollections(req *manager_pb.WatchCollectionsRequest, stream manager_pb.ManagerService_WatchCollectionsServer) error

WatchCollections implements server-streaming PUSH-based collection updates. Used by secondary regions to keep their bucket cache synchronized with primary.

func (*ManagerServer) WatchTopology 

func (ms *ManagerServer) WatchTopology(req *manager_pb.WatchTopologyRequest, stream manager_pb.ManagerService_WatchTopologyServer) error

WatchTopology implements server-streaming PUSH-based topology updates. Works on any node (leader or follower) - events are broadcast when Raft applies changes.

type RaftCommand 

type RaftCommand struct {
	Type CommandType     `json:"type"`
	Data json.RawMessage `json:"data"`
}

type RaftCredentialStore 

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

RaftCredentialStore implements iam.CredentialStore backed by Raft consensus. All mutations go through Raft, ensuring credentials are consistent across the cluster. Credentials are stored with encrypted secrets (EncryptedSecret field).

func NewRaftCredentialStore 

func NewRaftCredentialStore(ms *ManagerServer) *RaftCredentialStore

NewRaftCredentialStore creates a new RaftCredentialStore.

func (*RaftCredentialStore) CreateAccessKey 

func (s *RaftCredentialStore) CreateAccessKey(ctx context.Context, userName string, credential *iam.Credential) error

CreateAccessKey creates a new access key for a user. The credential should have EncryptedSecret set.

func (*RaftCredentialStore) CreatePolicy 

func (s *RaftCredentialStore) CreatePolicy(ctx context.Context, policy *iam.Policy) error

CreatePolicy creates a new policy.

func (*RaftCredentialStore) CreateUser 

func (s *RaftCredentialStore) CreateUser(ctx context.Context, identity *iam.Identity) error

CreateUser creates a new user with the given identity. Credentials should have EncryptedSecret set (call iam.SecureCredential before).

func (*RaftCredentialStore) DeleteAccessKey 

func (s *RaftCredentialStore) DeleteAccessKey(ctx context.Context, userName, accessKey string) error

DeleteAccessKey deletes an access key from a user.

func (*RaftCredentialStore) DeletePolicy 

func (s *RaftCredentialStore) DeletePolicy(ctx context.Context, policyID string) error

DeletePolicy deletes a policy.

func (*RaftCredentialStore) DeleteUser 

func (s *RaftCredentialStore) DeleteUser(ctx context.Context, userName string) error

DeleteUser deletes a user by name.

func (*RaftCredentialStore) GetCredential 

func (s *RaftCredentialStore) GetCredential(ctx context.Context, accessKey string) (*iam.Identity, *iam.Credential, error)

GetCredential retrieves a credential by access key. Returns the identity and credential if found, or ErrCredentialNotFound if not.

func (*RaftCredentialStore) GetPolicy 

func (s *RaftCredentialStore) GetPolicy(ctx context.Context, policyID string) (*iam.Policy, error)

GetPolicy retrieves a policy by ID.

func (*RaftCredentialStore) GetUser 

func (s *RaftCredentialStore) GetUser(ctx context.Context, userName string) (*iam.Identity, error)

GetUser retrieves a user by name.

func (*RaftCredentialStore) GetVersion 

func (s *RaftCredentialStore) GetVersion() uint64

GetVersion returns the current IAM version counter.

func (*RaftCredentialStore) HasUsers 

func (s *RaftCredentialStore) HasUsers() bool

HasUsers returns true if there are any users in the store.

func (*RaftCredentialStore) ListCredentials 

func (s *RaftCredentialStore) ListCredentials(ctx context.Context) ([]*iam.Identity, error)

ListCredentials returns all credentials (for streaming to metadata services).

func (*RaftCredentialStore) ListUsers 

func (s *RaftCredentialStore) ListUsers(ctx context.Context) ([]string, error)

ListUsers returns the names of all users.

func (*RaftCredentialStore) UpdateUser 

func (s *RaftCredentialStore) UpdateUser(ctx context.Context, identity *iam.Identity) error

UpdateUser updates an existing user.

type RaftNode 

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

func NewRaftNode 

func NewRaftNode(fsm raft.FSM, config *Config) (*RaftNode, error)

func (*RaftNode) AddNonvoter 

func (rn *RaftNode) AddNonvoter(id, address string, timeout time.Duration) error

AddNonvoter adds a non-voting member to the cluster

func (*RaftNode) AddVoter 

func (rn *RaftNode) AddVoter(id, address string, timeout time.Duration) error

AddVoter adds a voting member to the cluster

func (*RaftNode) Apply 

func (rn *RaftNode) Apply(data []byte, timeout time.Duration) error

Apply applies a log entry

func (*RaftNode) GetConfiguration 

func (rn *RaftNode) GetConfiguration() (raft.Configuration, error)

GetConfiguration returns the cluster configuration

func (*RaftNode) HasLeader 

func (rn *RaftNode) HasLeader() bool

HasLeader returns true if the cluster has an elected leader

func (*RaftNode) IsLeader 

func (rn *RaftNode) IsLeader() bool

IsLeader returns true if this node is the leader

func (*RaftNode) Leader 

func (rn *RaftNode) Leader() string

Leader returns the current leader address

func (*RaftNode) RemoveServer 

func (rn *RaftNode) RemoveServer(id string, timeout time.Duration) error

RemoveServer removes a server from the cluster

func (*RaftNode) Shutdown 

func (rn *RaftNode) Shutdown() error

Shutdown gracefully shuts down the Raft node

func (*RaftNode) Snapshot 

func (rn *RaftNode) Snapshot() error

Snapshot forces the Raft layer to take a snapshot. Used after recovery to persist the restored state.

func (*RaftNode) State 

func (rn *RaftNode) State() string

State returns the current Raft state (Follower, Candidate, Leader, Shutdown)

func (*RaftNode) Stats 

func (rn *RaftNode) Stats() map[string]string

Stats returns Raft stats

func (*RaftNode) WaitForLeader 

func (rn *RaftNode) WaitForLeader(timeout time.Duration) error

WaitForLeader blocks until a leader is elected or timeout

type RegionCacheConfig 

type RegionCacheConfig struct {
	SyncInterval       int
	CrossRegionTimeout int
}

RegionCacheConfig is a stub for community edition.

type RegionClient 

type RegionClient struct{}

RegionClient is a stub for community edition.

func NewRegionClient 

func NewRegionClient(_ *RegionConfig) (*RegionClient, error)

NewRegionClient returns nil in community edition (multi-region not supported).

func (*RegionClient) Close 

func (rc *RegionClient) Close() error

Close is a no-op in community edition.

func (*RegionClient) ForwardCreateCollection 

func (rc *RegionClient) ForwardCreateCollection(_ context.Context, _ *manager_pb.CreateCollectionRequest) (*manager_pb.CreateCollectionResponse, error)

ForwardCreateCollection returns nil in community edition (handle locally).

func (*RegionClient) ForwardDeleteCollection 

func (rc *RegionClient) ForwardDeleteCollection(_ context.Context, _ *manager_pb.DeleteCollectionRequest) (*manager_pb.DeleteCollectionResponse, error)

ForwardDeleteCollection returns nil in community edition (handle locally).

func (*RegionClient) GetClient 

func (rc *RegionClient) GetClient(_ context.Context, _ string) (interface{}, error)

GetClient always returns an error in community edition.

func (*RegionClient) IsRegionHealthy 

func (rc *RegionClient) IsRegionHealthy(_ context.Context, _ string) bool

IsRegionHealthy always returns false in community edition.

func (*RegionClient) ListPrimaryCollections 

func (rc *RegionClient) ListPrimaryCollections(_ context.Context) (manager_pb.ManagerService_ListCollectionsClient, error)

ListPrimaryCollections always returns an error in community edition.

type RegionConfig 

type RegionConfig struct {
	Name           string
	PrimaryRegions []string // Stub field for compatibility
	Peers          []RegionPeer
	Cache          RegionCacheConfig
}

RegionConfig is a stub for community edition. Multi-region support requires an enterprise license.

func (*RegionConfig) GetCrossRegionTimeout 

func (c *RegionConfig) GetCrossRegionTimeout() time.Duration

GetCrossRegionTimeout returns the default timeout.

func (*RegionConfig) GetCurrentPrimary 

func (c *RegionConfig) GetCurrentPrimary(_ context.Context, _ func(context.Context, string) bool) (string, error)

GetCurrentPrimary returns this region's name in community edition.

func (*RegionConfig) GetPeerAddresses 

func (c *RegionConfig) GetPeerAddresses(_ string) []string

GetPeerAddresses always returns nil in community edition.

func (*RegionConfig) GetSyncInterval 

func (c *RegionConfig) GetSyncInterval() time.Duration

GetSyncInterval returns the default sync interval.

func (*RegionConfig) IsConfigured 

func (c *RegionConfig) IsConfigured() bool

IsConfigured always returns false in community edition.

func (*RegionConfig) IsPrimary 

func (c *RegionConfig) IsPrimary() bool

IsPrimary always returns true in community edition (single-region mode).

func (*RegionConfig) Validate 

func (c *RegionConfig) Validate() error

Validate always succeeds in community edition.

type RegionPeer 

type RegionPeer struct {
	Name             string
	ManagerAddresses []string
}

RegionPeer is a stub for community edition.

type RegionSyncer 

type RegionSyncer struct{}

RegionSyncer is a stub for community edition. Multi-region cache sync requires an enterprise license.

func NewRegionSyncer 

func NewRegionSyncer(_ *ManagerServer) *RegionSyncer

NewRegionSyncer returns nil in community edition.

func (*RegionSyncer) Start 

func (rs *RegionSyncer) Start()

Start is a no-op in community edition.

func (*RegionSyncer) Stop 

func (rs *RegionSyncer) Stop()

Stop is a no-op in community edition.

type ServiceRegistration 

type ServiceRegistration struct {
	ServiceType              manager_pb.ServiceType
	Location                 *common_pb.Location
	Status                   ServiceStatus
	RegisteredAt             time.Time
	LastHeartbeat            time.Time
	LastKnownTopologyVersion uint64
	StorageBackends          []*manager_pb.StorageBackend
}

ServiceRegistration holds information about a registered service

type ServiceStatus 

type ServiceStatus int

ServiceStatus represents the health status of a service 

const (
	ServiceActive ServiceStatus = iota
	ServiceOffline
)

type ServiceStatusUpdate 

type ServiceStatusUpdate struct {
	ServiceType manager_pb.ServiceType `json:"service_type"`
	ServiceID   string                 `json:"service_id"`
	NewStatus   ServiceStatus          `json:"new_status"`
}

ServiceStatusUpdate represents a status change for a service (goes through Raft)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.
⌂ 换网址