cluster

package
v0.0.0-...-dc8f43e Latest Latest
Warning

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

Go to latest
Published: Aug 13, 2025 License: Apache-2.0 Imports: 16 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Alert

type Alert struct {
	ID        string                 `json:"id"`
	Type      AlertType              `json:"type"`
	Severity  AlertSeverity          `json:"severity"`
	Message   string                 `json:"message"`
	Data      map[string]interface{} `json:"data"`
	Timestamp time.Time              `json:"timestamp"`
	Resolved  bool                   `json:"resolved"`
}

Alert represents an alert message

type AlertChannel

type AlertChannel interface {
	Send(alert Alert) error
	GetType() string
}

AlertChannel defines an alerting channel

type AlertManager

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

AlertManager handles alerting for failover events

func (*AlertManager) SendAlert

func (am *AlertManager) SendAlert(alert Alert)

SendAlert sends an alert through all configured channels

type AlertSeverity

type AlertSeverity string

AlertSeverity defines alert severity levels

const (
	AlertSeverityLow      AlertSeverity = "low"
	AlertSeverityMedium   AlertSeverity = "medium"
	AlertSeverityHigh     AlertSeverity = "high"
	AlertSeverityCritical AlertSeverity = "critical"
)

type AlertTemplate

type AlertTemplate struct {
	Subject string `yaml:"subject"`
	Body    string `yaml:"body"`
}

AlertTemplate defines alert message templates

type AlertType

type AlertType string

AlertType defines types of alerts

const (
	AlertTypeNodeDown         AlertType = "node_down"
	AlertTypeNodeRecovered    AlertType = "node_recovered"
	AlertTypeFailoverStarted  AlertType = "failover_started"
	AlertTypeFailoverComplete AlertType = "failover_complete"
	AlertTypeClusterDegraded  AlertType = "cluster_degraded"
)

type AppendEntriesRequest

type AppendEntriesRequest struct {
	Term         uint64     `json:"term"`
	LeaderID     string     `json:"leader_id"`
	PrevLogIndex uint64     `json:"prev_log_index"`
	PrevLogTerm  uint64     `json:"prev_log_term"`
	Entries      []LogEntry `json:"entries"`
	LeaderCommit uint64     `json:"leader_commit"`
}

AppendEntriesRequest represents a Raft AppendEntries RPC request

type AppendEntriesResponse

type AppendEntriesResponse struct {
	Term    uint64 `json:"term"`
	Success bool   `json:"success"`
}

AppendEntriesResponse represents a Raft AppendEntries RPC response

type Cluster

type Cluster struct {
	ID        string            `json:"id"`
	Name      string            `json:"name"`
	Nodes     map[string]*Node  `json:"nodes"`
	Shards    map[string]*Shard `json:"shards"`
	Config    ClusterConfig     `json:"config"`
	State     ClusterState      `json:"state"`
	CreatedAt time.Time         `json:"created_at"`
	UpdatedAt time.Time         `json:"updated_at"`
	// contains filtered or unexported fields
}

Cluster represents a govc distributed cluster

func NewCluster

func NewCluster(id, name string, config ClusterConfig, dataDir string) (*Cluster, error)

NewCluster creates a new cluster

func (*Cluster) AddNode

func (c *Cluster) AddNode(node *Node) error

AddNode adds a node to the cluster

func (*Cluster) CreateShard

func (c *Cluster) CreateShard(keyRange ShardKeyRange) (*Shard, error)

CreateShard creates a new shard

func (*Cluster) DistributeRepository

func (c *Cluster) DistributeRepository(repoID string, repo *govc.Repository) error

DistributeRepository distributes a repository to the appropriate shard

func (*Cluster) GetClusterHealth

func (c *Cluster) GetClusterHealth() ClusterHealth

GetClusterHealth returns the overall health of the cluster

func (*Cluster) GetID

func (c *Cluster) GetID() string

GetID returns the cluster ID

func (*Cluster) GetLeaderNode

func (c *Cluster) GetLeaderNode() *Node

GetLeaderNode returns the current leader node

func (*Cluster) GetNodeByID

func (c *Cluster) GetNodeByID(nodeID string) *Node

GetNodeByID returns a node by its ID

func (*Cluster) GetNodes

func (c *Cluster) GetNodes() []*Node

GetNodes returns all nodes in the cluster

func (*Cluster) GetShardForRepository

func (c *Cluster) GetShardForRepository(repoID string) *Shard

GetShardForRepository returns the shard that should contain a repository

func (*Cluster) RemoveNode

func (c *Cluster) RemoveNode(nodeID string) error

RemoveNode removes a node from the cluster

type ClusterConfig

type ClusterConfig struct {
	ReplicationFactor int           `yaml:"replication_factor"`
	ShardSize         int           `yaml:"shard_size"`
	ElectionTimeout   time.Duration `yaml:"election_timeout"`
	HeartbeatInterval time.Duration `yaml:"heartbeat_interval"`
	MaxLogEntries     int           `yaml:"max_log_entries"`
	SnapshotThreshold int           `yaml:"snapshot_threshold"`
	AutoRebalance     bool          `yaml:"auto_rebalance"`
	ConsistencyLevel  string        `yaml:"consistency_level"`
}

ClusterConfig contains cluster-wide configuration

type ClusterEvent

type ClusterEvent struct {
	Type      ClusterEventType       `json:"type"`
	NodeID    string                 `json:"node_id"`
	ShardID   string                 `json:"shard_id"`
	Data      map[string]interface{} `json:"data"`
	Timestamp time.Time              `json:"timestamp"`
}

ClusterEvent represents events that occur in the cluster

type ClusterEventType

type ClusterEventType string

ClusterEventType defines types of cluster events

const (
	EventNodeJoined        ClusterEventType = "node_joined"
	EventNodeLeft          ClusterEventType = "node_left"
	EventNodeFailed        ClusterEventType = "node_failed"
	EventShardMoved        ClusterEventType = "shard_moved"
	EventRebalanceStarted  ClusterEventType = "rebalance_started"
	EventRebalanceComplete ClusterEventType = "rebalance_complete"
	EventLeaderElected     ClusterEventType = "leader_elected"
)

type ClusterHealth

type ClusterHealth struct {
	Status       ClusterState `json:"status"`
	TotalNodes   int          `json:"total_nodes"`
	HealthyNodes int          `json:"healthy_nodes"`
	TotalShards  int          `json:"total_shards"`
	ActiveShards int          `json:"active_shards"`
	Timestamp    time.Time    `json:"timestamp"`
}

ClusterHealth represents the health status of the cluster

type ClusterState

type ClusterState string

ClusterState represents the overall state of the cluster

const (
	ClusterStateHealthy     ClusterState = "healthy"
	ClusterStateDegraded    ClusterState = "degraded"
	ClusterStateUnavailable ClusterState = "unavailable"
	ClusterStateRebalancing ClusterState = "rebalancing"
	ClusterStateRecovering  ClusterState = "recovering"
)

type Command

type Command struct {
	Type       CommandType            `json:"type"`
	Repository string                 `json:"repository"`
	Data       map[string]interface{} `json:"data"`
	Metadata   map[string]string      `json:"metadata"`
}

Command represents a distributed command that can be replicated

type CommandType

type CommandType string

CommandType defines the types of commands that can be replicated

const (
	CommandTypeCreateRepo   CommandType = "create_repo"
	CommandTypeDeleteRepo   CommandType = "delete_repo"
	CommandTypeCommit       CommandType = "commit"
	CommandTypeCreateBranch CommandType = "create_branch"
	CommandTypeDeleteBranch CommandType = "delete_branch"
	CommandTypeCreateTag    CommandType = "create_tag"
	CommandTypeDeleteTag    CommandType = "delete_tag"
	CommandTypeShardMove    CommandType = "shard_move"
	CommandTypeNodeJoin     CommandType = "node_join"
	CommandTypeNodeLeave    CommandType = "node_leave"
)

type ConflictResolution

type ConflictResolution struct {
	Strategy     ConflictStrategy       `json:"strategy"`
	WinnerNode   string                 `json:"winner_node"`
	ConflictData map[string]interface{} `json:"conflict_data"`
	Resolution   string                 `json:"resolution"`
	ResolvedAt   time.Time              `json:"resolved_at"`
}

ConflictResolution handles replication conflicts

type ConflictStrategy

type ConflictStrategy string

ConflictStrategy defines how to resolve replication conflicts

const (
	ConflictStrategyLastWrite  ConflictStrategy = "last_write_wins"
	ConflictStrategyMerge      ConflictStrategy = "merge"
	ConflictStrategyManual     ConflictStrategy = "manual"
	ConflictStrategySourceWins ConflictStrategy = "source_wins"
	ConflictStrategyTargetWins ConflictStrategy = "target_wins"
)

type ConsistentHashRing

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

ConsistentHashRing implements consistent hashing for shard distribution

func NewConsistentHashRing

func NewConsistentHashRing(replicas int) *ConsistentHashRing

NewConsistentHashRing creates a new consistent hash ring

func (*ConsistentHashRing) AddNode

func (chr *ConsistentHashRing) AddNode(nodeID string)

AddNode adds a node to the hash ring

func (*ConsistentHashRing) GetNode

func (chr *ConsistentHashRing) GetNode(key string) string

GetNode returns the node responsible for a given key

func (*ConsistentHashRing) GetNodes

func (chr *ConsistentHashRing) GetNodes(key string, count int) []string

GetNodes returns the top N nodes responsible for a key (for replication)

func (*ConsistentHashRing) RemoveNode

func (chr *ConsistentHashRing) RemoveNode(nodeID string)

RemoveNode removes a node from the hash ring

type FailoverEvent

type FailoverEvent struct {
	ID          string            `json:"id"`
	Type        FailoverType      `json:"type"`
	SourceNode  string            `json:"source_node"`
	TargetNode  string            `json:"target_node"`
	Reason      string            `json:"reason"`
	Status      FailoverStatus    `json:"status"`
	StartedAt   time.Time         `json:"started_at"`
	CompletedAt time.Time         `json:"completed_at"`
	Duration    time.Duration     `json:"duration"`
	Errors      []string          `json:"errors"`
	Metadata    map[string]string `json:"metadata"`
}

FailoverEvent represents a failover event

type FailoverManager

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

FailoverManager handles automatic failover and recovery

func NewFailoverManager

func NewFailoverManager(cluster *Cluster, policy FailoverPolicy) *FailoverManager

NewFailoverManager creates a new failover manager

func (*FailoverManager) GetClusterHealth

func (fm *FailoverManager) GetClusterHealth() map[string]interface{}

GetClusterHealth returns overall cluster health

func (*FailoverManager) GetFailoverHistory

func (fm *FailoverManager) GetFailoverHistory() []FailoverEvent

GetFailoverHistory returns the failover event history

func (*FailoverManager) Start

func (fm *FailoverManager) Start(ctx context.Context) error

Start starts the failover manager

type FailoverPolicy

type FailoverPolicy struct {
	AutoFailoverEnabled   bool          `yaml:"auto_failover_enabled"`
	FailoverTimeout       time.Duration `yaml:"failover_timeout"`
	MinHealthyNodes       int           `yaml:"min_healthy_nodes"`
	RequireQuorum         bool          `yaml:"require_quorum"`
	PreventSplitBrain     bool          `yaml:"prevent_split_brain"`
	MaxFailoversPerMinute int           `yaml:"max_failovers_per_minute"`
	CooldownPeriod        time.Duration `yaml:"cooldown_period"`
}

FailoverPolicy defines failover behavior

type FailoverStatus

type FailoverStatus string

FailoverStatus represents failover operation status

const (
	FailoverStatusPending   FailoverStatus = "pending"
	FailoverStatusActive    FailoverStatus = "active"
	FailoverStatusCompleted FailoverStatus = "completed"
	FailoverStatusFailed    FailoverStatus = "failed"
	FailoverStatusAborted   FailoverStatus = "aborted"
)

type FailoverType

type FailoverType string

FailoverType defines types of failover operations

const (
	FailoverTypeNodeFailure    FailoverType = "node_failure"
	FailoverTypeLeaderElection FailoverType = "leader_election"
	FailoverTypeShardMigration FailoverType = "shard_migration"
	FailoverTypeLoadBalancing  FailoverType = "load_balancing"
)

type HealthCallback

type HealthCallback func(nodeID string, oldStatus, newStatus HealthStatus)

HealthCallback is called when node health changes

type HealthChecker

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

HealthChecker monitors node health

func (*HealthChecker) RegisterCallback

func (hc *HealthChecker) RegisterCallback(callback HealthCallback)

RegisterCallback registers a health change callback

func (*HealthChecker) Start

func (hc *HealthChecker) Start(ctx context.Context, cluster *Cluster)

Start starts the health checker

type HealthStatus

type HealthStatus string

HealthStatus represents node health status

const (
	HealthStatusHealthy   HealthStatus = "healthy"
	HealthStatusDegraded  HealthStatus = "degraded"
	HealthStatusUnhealthy HealthStatus = "unhealthy"
	HealthStatusUnknown   HealthStatus = "unknown"
)

type JoinRequest

type JoinRequest struct {
	NodeID  string `json:"node_id"`
	Address string `json:"address"`
	Port    int    `json:"port"`
	Version string `json:"version"`
}

JoinRequest represents a request to join the cluster

type JoinResponse

type JoinResponse struct {
	Success   bool     `json:"success"`
	Message   string   `json:"message"`
	LeaderID  string   `json:"leader_id"`
	ClusterID string   `json:"cluster_id"`
	Nodes     []string `json:"nodes"`
}

JoinResponse represents a response to a join request

type LeaveRequest

type LeaveRequest struct {
	NodeID string `json:"node_id"`
	Reason string `json:"reason"`
}

LeaveRequest represents a request to leave the cluster

type LeaveResponse

type LeaveResponse struct {
	Success bool   `json:"success"`
	Message string `json:"message"`
}

LeaveResponse represents a response to a leave request

type LogEntry

type LogEntry struct {
	Term      uint64    `json:"term"`
	Index     uint64    `json:"index"`
	Command   Command   `json:"command"`
	Timestamp time.Time `json:"timestamp"`
	ClientID  string    `json:"client_id"`
}

LogEntry represents a single entry in the Raft log

type MigrationQueue

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

MigrationQueue manages shard migration operations

type MigrationState

type MigrationState string

MigrationState represents the state of a migration task

const (
	MigrationStatePending   MigrationState = "pending"
	MigrationStateActive    MigrationState = "active"
	MigrationStateCompleted MigrationState = "completed"
	MigrationStateFailed    MigrationState = "failed"
	MigrationStateCancelled MigrationState = "cancelled"
)

type MigrationTask

type MigrationTask struct {
	ID           string            `json:"id"`
	ShardID      string            `json:"shard_id"`
	SourceNode   string            `json:"source_node"`
	TargetNode   string            `json:"target_node"`
	Repositories []string          `json:"repositories"`
	State        MigrationState    `json:"state"`
	Progress     int               `json:"progress"`
	Error        string            `json:"error"`
	StartedAt    time.Time         `json:"started_at"`
	CompletedAt  time.Time         `json:"completed_at"`
	Metadata     map[string]string `json:"metadata"`
}

MigrationTask represents a shard migration operation

type Node

type Node struct {
	ID       string    `json:"id"`
	Address  string    `json:"address"`
	Port     int       `json:"port"`
	State    NodeState `json:"state"`
	LastSeen time.Time `json:"last_seen"`
	Version  string    `json:"version"`

	// Raft-specific fields
	Term        uint64   `json:"term"`
	VotedFor    string   `json:"voted_for"`
	IsLeader    bool     `json:"is_leader"`
	LeaderID    string   `json:"leader_id"`
	CommitIndex uint64   `json:"commit_index"`
	LastApplied uint64   `json:"last_applied"`
	NextIndex   []uint64 `json:"next_index"`
	MatchIndex  []uint64 `json:"match_index"`

	Stats    NodeStats         `json:"stats"`
	Metadata map[string]string `json:"metadata"`
	// contains filtered or unexported fields
}

Node represents a single node in the govc cluster

func NewNode

func NewNode(config NodeConfig) (*Node, error)

NewNode creates a new cluster node

func (*Node) AddRepository

func (n *Node) AddRepository(repoID string, repo *govc.Repository) error

AddRepository adds a repository to the node

func (*Node) GetAddress

func (n *Node) GetAddress() string

GetAddress returns the node address

func (*Node) GetCapacity

func (n *Node) GetCapacity() NodeCapacity

GetCapacity returns the node's capacity information

func (*Node) GetID

func (n *Node) GetID() string

GetID returns the node ID

func (*Node) GetLeaderID

func (n *Node) GetLeaderID() string

GetLeaderID returns the ID of the current leader

func (*Node) GetRepositories

func (n *Node) GetRepositories() []string

GetRepositories returns all repository IDs on this node

func (*Node) GetRepository

func (n *Node) GetRepository(repoID string) *govc.Repository

GetRepository retrieves a repository from the node

func (*Node) GetState

func (n *Node) GetState() NodeState

GetState returns the current node state

func (*Node) IsHealthy

func (n *Node) IsHealthy() bool

IsHealthy returns true if the node is healthy

func (*Node) IsLeaderNode

func (n *Node) IsLeaderNode() bool

IsLeaderNode returns true if this node is the current leader

func (*Node) RemoveRepository

func (n *Node) RemoveRepository(repoID string) error

RemoveRepository removes a repository from the node

func (*Node) Start

func (n *Node) Start(ctx context.Context) error

Start starts the node and joins the cluster

func (*Node) Stop

func (n *Node) Stop() error

Stop gracefully stops the node

func (*Node) UpdateStats

func (n *Node) UpdateStats()

UpdateStats updates the node's statistics

type NodeCapacity

type NodeCapacity struct {
	RepositoryCount   int64   `json:"repository_count"`
	MaxRepositories   int64   `json:"max_repositories"`
	AvailableCapacity float64 `json:"available_capacity"`
}

NodeCapacity represents node capacity information

type NodeConfig

type NodeConfig struct {
	ID                string        `yaml:"id"`
	Address           string        `yaml:"address"`
	Port              int           `yaml:"port"`
	DataDir           string        `yaml:"data_dir"`
	ClusterPeers      []string      `yaml:"cluster_peers"`
	ElectionTimeout   time.Duration `yaml:"election_timeout"`
	HeartbeatTimeout  time.Duration `yaml:"heartbeat_timeout"`
	MaxLogEntries     int           `yaml:"max_log_entries"`
	SnapshotThreshold int           `yaml:"snapshot_threshold"`
}

NodeConfig contains configuration for a cluster node

type NodeHealth

type NodeHealth struct {
	NodeID           string        `json:"node_id"`
	Status           HealthStatus  `json:"status"`
	LastSeen         time.Time     `json:"last_seen"`
	ConsecutiveFails int           `json:"consecutive_fails"`
	ResponseTime     time.Duration `json:"response_time"`
	LoadAverage      float64       `json:"load_average"`
	MemoryUsage      float64       `json:"memory_usage"`
	DiskUsage        float64       `json:"disk_usage"`
	NetworkLatency   time.Duration `json:"network_latency"`
	Errors           []string      `json:"errors"`
}

NodeHealth represents the health status of a node

type NodeState

type NodeState string

NodeState represents the current state of a cluster node

const (
	NodeStateFollower  NodeState = "follower"
	NodeStateCandidate NodeState = "candidate"
	NodeStateLeader    NodeState = "leader"
	NodeStateOffline   NodeState = "offline"
	NodeStateJoining   NodeState = "joining"
	NodeStateLeaving   NodeState = "leaving"
)

type NodeStats

type NodeStats struct {
	CPUUsage        float64   `json:"cpu_usage"`
	MemoryUsage     float64   `json:"memory_usage"`
	DiskUsage       float64   `json:"disk_usage"`
	RepositoryCount int64     `json:"repository_count"`
	LastUpdate      time.Time `json:"last_update"`
}

NodeStats holds node statistics

type RaftState

type RaftState struct {
	// Persistent state
	CurrentTerm uint64     `json:"current_term"`
	VotedFor    string     `json:"voted_for"`
	Log         []LogEntry `json:"log"`

	// Volatile state
	CommitIndex uint64 `json:"commit_index"`
	LastApplied uint64 `json:"last_applied"`

	// Leader state (reinitialized after election)
	NextIndex  map[string]uint64 `json:"next_index"`
	MatchIndex map[string]uint64 `json:"match_index"`

	// Timing
	ElectionTimeout  time.Duration `json:"election_timeout"`
	HeartbeatTimeout time.Duration `json:"heartbeat_timeout"`
	LastHeartbeat    time.Time     `json:"last_heartbeat"`
	// contains filtered or unexported fields
}

RaftState manages Raft consensus state for the node

type RebalanceTask

type RebalanceTask struct {
	ID         string    `json:"id"`
	Type       string    `json:"type"`
	SourceNode string    `json:"source_node"`
	TargetNode string    `json:"target_node"`
	ShardID    string    `json:"shard_id"`
	Priority   int       `json:"priority"`
	CreatedAt  time.Time `json:"created_at"`
}

RebalanceTask represents a rebalancing operation

type RecoveryManager

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

RecoveryManager handles node recovery

func (*RecoveryManager) Start

func (rm *RecoveryManager) Start(ctx context.Context)

Start starts the recovery manager

type RecoveryStatus

type RecoveryStatus string

RecoveryStatus represents recovery operation status

const (
	RecoveryStatusPending   RecoveryStatus = "pending"
	RecoveryStatusActive    RecoveryStatus = "active"
	RecoveryStatusCompleted RecoveryStatus = "completed"
	RecoveryStatusFailed    RecoveryStatus = "failed"
)

type RecoveryTask

type RecoveryTask struct {
	ID          string            `json:"id"`
	NodeID      string            `json:"node_id"`
	Type        RecoveryType      `json:"type"`
	Status      RecoveryStatus    `json:"status"`
	StartedAt   time.Time         `json:"started_at"`
	CompletedAt time.Time         `json:"completed_at"`
	Progress    int               `json:"progress"`
	Errors      []string          `json:"errors"`
	Metadata    map[string]string `json:"metadata"`
}

RecoveryTask represents a node recovery operation

type RecoveryType

type RecoveryType string

RecoveryType defines types of recovery operations

const (
	RecoveryTypeNodeRestart     RecoveryType = "node_restart"
	RecoveryTypeDataSync        RecoveryType = "data_sync"
	RecoveryTypeShardRecovery   RecoveryType = "shard_recovery"
	RecoveryTypeLeaderPromotion RecoveryType = "leader_promotion"
)

type ReplicationManager

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

ReplicationManager handles cross-node replication

func NewReplicationManager

func NewReplicationManager(cluster *Cluster) *ReplicationManager

NewReplicationManager creates a new replication manager

func (*ReplicationManager) GetActiveReplications

func (rm *ReplicationManager) GetActiveReplications() []ReplicationTask

GetActiveReplications returns currently active replication tasks

func (*ReplicationManager) GetReplicationMetrics

func (rm *ReplicationManager) GetReplicationMetrics() ReplicationMetrics

GetReplicationMetrics returns current replication metrics

func (*ReplicationManager) ReplicateRepository

func (rm *ReplicationManager) ReplicateRepository(repoID string, sourceNode string, targetNodes []string, strategy ReplicationStrategy) error

ReplicateRepository replicates a repository to target nodes

func (*ReplicationManager) ReplicateShard

func (rm *ReplicationManager) ReplicateShard(shardID string, sourceNode string, targetNodes []string, strategy ReplicationStrategy) error

ReplicateShard replicates an entire shard to target nodes

func (*ReplicationManager) Start

func (rm *ReplicationManager) Start(ctx context.Context) error

Start starts the replication manager

func (*ReplicationManager) Stop

func (rm *ReplicationManager) Stop()

Stop stops the replication manager

type ReplicationMetrics

type ReplicationMetrics struct {
	TotalReplications      uint64            `json:"total_replications"`
	SuccessfulReplications uint64            `json:"successful_replications"`
	FailedReplications     uint64            `json:"failed_replications"`
	AverageLatency         time.Duration     `json:"average_latency"`
	ThroughputBytesPerSec  uint64            `json:"throughput_bytes_per_sec"`
	ActiveReplications     int               `json:"active_replications"`
	ReplicationsByNode     map[string]uint64 `json:"replications_by_node"`
	ErrorsByType           map[string]uint64 `json:"errors_by_type"`
}

ReplicationMetrics contains replication performance metrics

type ReplicationState

type ReplicationState string

ReplicationState represents the state of a replication task

const (
	ReplicationStatePending   ReplicationState = "pending"
	ReplicationStateActive    ReplicationState = "active"
	ReplicationStateCompleted ReplicationState = "completed"
	ReplicationStateFailed    ReplicationState = "failed"
	ReplicationStateRetrying  ReplicationState = "retrying"
)

type ReplicationStrategy

type ReplicationStrategy struct {
	Type               ReplicationType `yaml:"type"`
	ConsistencyLevel   string          `yaml:"consistency_level"`
	SyncTimeout        time.Duration   `yaml:"sync_timeout"`
	AsyncBufferSize    int             `yaml:"async_buffer_size"`
	CompressionEnabled bool            `yaml:"compression_enabled"`
	EncryptionEnabled  bool            `yaml:"encryption_enabled"`
	BatchSize          int             `yaml:"batch_size"`
	RetryPolicy        RetryPolicy     `yaml:"retry_policy"`
}

ReplicationStrategy defines replication behavior

type ReplicationTask

type ReplicationTask struct {
	ID           string            `json:"id"`
	Type         ReplicationType   `json:"type"`
	SourceNode   string            `json:"source_node"`
	TargetNodes  []string          `json:"target_nodes"`
	RepositoryID string            `json:"repository_id"`
	ShardID      string            `json:"shard_id"`
	Data         []byte            `json:"data"`
	Checksum     string            `json:"checksum"`
	State        ReplicationState  `json:"state"`
	Progress     int               `json:"progress"`
	StartedAt    time.Time         `json:"started_at"`
	CompletedAt  time.Time         `json:"completed_at"`
	RetryCount   int               `json:"retry_count"`
	MaxRetries   int               `json:"max_retries"`
	Error        string            `json:"error"`
	Metadata     map[string]string `json:"metadata"`
}

ReplicationTask represents a replication operation

type ReplicationType

type ReplicationType string

ReplicationType defines types of replication operations

const (
	ReplicationTypeSync        ReplicationType = "sync"
	ReplicationTypeAsync       ReplicationType = "async"
	ReplicationTypeIncremental ReplicationType = "incremental"
	ReplicationTypeFull        ReplicationType = "full"
	ReplicationTypeSnapshot    ReplicationType = "snapshot"
)

type RepositoryOperation

type RepositoryOperation struct {
	Operation string                 `json:"operation"`
	RepoID    string                 `json:"repo_id"`
	Data      map[string]interface{} `json:"data"`
}

RepositoryOperation represents a repository operation

type RequestVoteRequest

type RequestVoteRequest struct {
	Term         uint64 `json:"term"`
	CandidateID  string `json:"candidate_id"`
	LastLogIndex uint64 `json:"last_log_index"`
	LastLogTerm  uint64 `json:"last_log_term"`
}

RequestVoteRequest represents a Raft RequestVote RPC request

type RequestVoteResponse

type RequestVoteResponse struct {
	Term        uint64 `json:"term"`
	VoteGranted bool   `json:"vote_granted"`
}

RequestVoteResponse represents a Raft RequestVote RPC response

type RetryPolicy

type RetryPolicy struct {
	MaxRetries    int           `yaml:"max_retries"`
	InitialDelay  time.Duration `yaml:"initial_delay"`
	MaxDelay      time.Duration `yaml:"max_delay"`
	BackoffFactor float64       `yaml:"backoff_factor"`
	JitterEnabled bool          `yaml:"jitter_enabled"`
}

RetryPolicy defines retry behavior for failed replications

type Shard

type Shard struct {
	ID           string          `json:"id"`
	KeyRange     ShardKeyRange   `json:"key_range"`
	PrimaryNode  string          `json:"primary_node"`
	ReplicaNodes []string        `json:"replica_nodes"`
	State        ShardState      `json:"state"`
	Repositories map[string]bool `json:"repositories"`
	Size         int64           `json:"size"`
	LastAccessed time.Time       `json:"last_accessed"`
	CreatedAt    time.Time       `json:"created_at"`
}

Shard represents a data shard in the cluster

type ShardKeyRange

type ShardKeyRange struct {
	Start string `json:"start"`
	End   string `json:"end"`
}

ShardKeyRange defines the key range for a shard

type ShardMetrics

type ShardMetrics struct {
	TotalShards          int            `json:"total_shards"`
	ShardsPerNode        map[string]int `json:"shards_per_node"`
	RepositoriesPerShard map[string]int `json:"repositories_per_shard"`
	LoadBalance          float64        `json:"load_balance"`
	Imbalance            float64        `json:"imbalance"`
	HotSpots             []string       `json:"hot_spots"`
}

ShardMetrics contains metrics for shard distribution

type ShardRebalancer

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

ShardRebalancer handles automatic rebalancing of shards

type ShardState

type ShardState string

ShardState represents the state of a shard

const (
	ShardStateActive      ShardState = "active"
	ShardStateRebalancing ShardState = "rebalancing"
	ShardStateOffline     ShardState = "offline"
	ShardStateMigrating   ShardState = "migrating"
)

type ShardingManager

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

ShardingManager manages repository sharding across cluster nodes

func NewShardingManager

func NewShardingManager(cluster *Cluster) *ShardingManager

NewShardingManager creates a new sharding manager

func (*ShardingManager) CalculateShardMetrics

func (sm *ShardingManager) CalculateShardMetrics() ShardMetrics

CalculateShardMetrics calculates sharding metrics for the cluster

func (*ShardingManager) DistributeRepository

func (sm *ShardingManager) DistributeRepository(repoID string, repo *govc.Repository) error

DistributeRepository distributes a repository to the appropriate shard

func (*ShardingManager) GetMigrationStatus

func (sm *ShardingManager) GetMigrationStatus() map[string]interface{}

GetMigrationStatus returns the status of all migrations

func (*ShardingManager) GetShardForRepository

func (sm *ShardingManager) GetShardForRepository(repoID string) *Shard

GetShardForRepository determines which shard should contain a repository

func (*ShardingManager) MigrateShard

func (sm *ShardingManager) MigrateShard(shardID, sourceNode, targetNode string) error

MigrateShard migrates a shard from one node to another

type SnapshotRequest

type SnapshotRequest struct {
	Term              uint64 `json:"term"`
	LeaderID          string `json:"leader_id"`
	LastIncludedIndex uint64 `json:"last_included_index"`
	LastIncludedTerm  uint64 `json:"last_included_term"`
	Offset            uint64 `json:"offset"`
	Data              []byte `json:"data"`
	Done              bool   `json:"done"`
}

SnapshotRequest represents a Raft InstallSnapshot RPC request

type SnapshotResponse

type SnapshotResponse struct {
	Term uint64 `json:"term"`
}

SnapshotResponse represents a Raft InstallSnapshot RPC response

type StatusResponse

type StatusResponse struct {
	NodeID         string    `json:"node_id"`
	State          NodeState `json:"state"`
	IsLeader       bool      `json:"is_leader"`
	LeaderID       string    `json:"leader_id"`
	Term           uint64    `json:"term"`
	CommitIndex    uint64    `json:"commit_index"`
	LastApplied    uint64    `json:"last_applied"`
	LogLength      int       `json:"log_length"`
	ClusterSize    int       `json:"cluster_size"`
	ConnectedNodes []string  `json:"connected_nodes"`
}

StatusResponse represents cluster status information

Jump to

Keyboard shortcuts

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