sync

package
v0.0.8 Latest Latest
Warning

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

 Go to latest Published: Jul 10, 2026 License: MIT Imports: 23 Imported by: 0

Documentation

Overview

Package sync implements industrial-grade distributed configuration sync based on go-libp2p with Gossip, WAL, Anti-Entropy, and Device Takeover.

Features:

  • ConfigStore with WAL + KV persistence for reliability
  • Gossip PubSub for fast propagation
  • mDNS + Peer Cache for robust discovery
  • Announce/Pull for efficient sync
  • Digest Anti-Entropy for eventual consistency
  • DeviceKey composite identity for IP drift resilience
  • Takeover lock for distributed control
  • Node roles (Leader/Follower/Readonly) for write control

Package sync provides multi-node config file synchronization (deprecated). Runtime configuration now lives in data/config.db; SyncManager is disabled in main.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func FormatNodeID 

func FormatNodeID(nodeID string) string

FormatNodeID formats a node ID to standard format

func GenBindingKey 

func GenBindingKey(ip, protocol string, port int) string

GenBindingKey generates a binding key for config

func GenDeviceKey 

func GenDeviceKey(ip, protocol string, port int, fingerprint *DeviceFingerprint) string

GenDeviceKey generates a device key from identity info

func GetDefaultNodeID 

func GetDefaultNodeID() string

GetDefaultNodeID returns a default node ID based on hostname

func GetNodeIDFromConfig 

func GetNodeIDFromConfig(configNodeID string) string

GetNodeIDFromConfig returns node ID from config or generates one

Types

type BucketConfig 

type BucketConfig struct {
	Pattern       string
	Description   string
	AccessMode    string
	RetentionDays int
	MaxRecords    int
	PartitionKey  string
}

type ClusterStore 

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

func NewClusterStore 

func NewClusterStore(dataPath string) (*ClusterStore, error)

NewClusterStore 创建 ClusterStore，复用现有 storage/ConfigStore 的 bbolt 模式。 dataPath: 数据目录路径（如 data/sync），数据库文件自动命名为 cluster.db。

func (*ClusterStore) Close 

func (cs *ClusterStore) Close() error

Close 关闭数据库（幂等）

func (*ClusterStore) DBPath 

func (cs *ClusterStore) DBPath() string

DBPath returns the database file path.

func (*ClusterStore) DeleteNode 

func (cs *ClusterStore) DeleteNode(nodeID string) error

DeleteNode 删除指定节点及其关联索引。

func (*ClusterStore) GetAllNodes 

func (cs *ClusterStore) GetAllNodes() ([]NodeMeta, error)

GetAllNodes 返回所有已知节点元数据。

func (*ClusterStore) GetClusterMeta 

func (cs *ClusterStore) GetClusterMeta(key string) (string, error)

GetClusterMeta 获取集群级别元数据。

func (*ClusterStore) GetClusterSummary 

func (cs *ClusterStore) GetClusterSummary() (*ClusterSummary, error)

GetClusterSummary 返回集群聚合统计信息。

func (*ClusterStore) GetDeviceOwners 

func (cs *ClusterStore) GetDeviceOwners(deviceID string) ([]string, error)

GetDeviceOwners 返回持有指定设备的所有节点ID。

func (*ClusterStore) GetDeviceSnapshot 

func (cs *ClusterStore) GetDeviceSnapshot(deviceID string) (*DeviceSnapshot, error)

GetDeviceSnapshot 按设备ID获取跨节点的设备快照。 返回该设备在各个节点上的配置。

func (*ClusterStore) GetNodeDevices 

func (cs *ClusterStore) GetNodeDevices(nodeID string) ([]TreeDevice, error)

GetNodeDevices 返回指定节点的所有设备列表。

func (*ClusterStore) GetNodeSnapshot 

func (cs *ClusterStore) GetNodeSnapshot(nodeID string) (*NodeSnapshot, error)

GetNodeSnapshot 从 bbolt 重建完整 NodeSnapshot。

func (*ClusterStore) ListKnownDevices 

func (cs *ClusterStore) ListKnownDevices() ([]string, error)

ListKnownDevices 返回所有已知的设备ID列表（去重）。

func (*ClusterStore) PutNodeSnapshot 

func (cs *ClusterStore) PutNodeSnapshot(nodeID string, snapshot *NodeSnapshot) error

PutNodeSnapshot 存储一个节点的完整快照到 bbolt。 使用单事务保证原子性：失败则整体回滚。

func (*ClusterStore) PutRemoteSnapshot 

func (cs *ClusterStore) PutRemoteSnapshot(nodeID string, snapshot *NodeSnapshot) error

PutRemoteSnapshot 存储远程节点的快照（与 PutNodeSnapshot 相同，语义区分）。

func (*ClusterStore) SetClusterMeta 

func (cs *ClusterStore) SetClusterMeta(key, value string) error

SetClusterMeta 设置集群级别的键值对元数据。

func (*ClusterStore) SnapshotExists 

func (cs *ClusterStore) SnapshotExists(nodeID string) bool

SnapshotExists 判断指定节点快照是否已存在。

type ClusterSummary 

type ClusterSummary struct {
	NodeCount    int       `json:"node_count"`
	DeviceCount  int       `json:"device_count"`
	ChannelCount int       `json:"channel_count"`
	PointCount   int       `json:"point_count"`
	KnownDevices int       `json:"known_devices"` // 去重后设备数
	UpdatedAt    time.Time `json:"updated_at"`
}

ClusterSummary 集群聚合统计

type ConfigFile 

type ConfigFile struct {
	Path    string         `json:"path"`
	Section string         `json:"section"`
	Hash    string         `json:"hash"`
	Content any            `json:"content,omitempty"`
	Meta    map[string]any `json:"meta,omitempty"`
}

ConfigFile captures one source config file for file-level diffing.

type ConfigMigrationRequest 

type ConfigMigrationRequest struct {
	DeviceCode     string `json:"device_code"`
	TargetDeviceID string `json:"target_device_id"`
	SourceNodeID   string `json:"source_node_id,omitempty"`
}

ConfigMigrationRequest represents a request for device config migration

type ConfigRecord 

type ConfigRecord struct {
	Key        string      `json:"key"`
	Value      []byte      `json:"value"`
	Version    uint64      `json:"version"`
	NodeID     string      `json:"node_id"`
	Timestamp  int64       `json:"timestamp"`
	Hash       string      `json:"hash"`
	BindingKey string      `json:"binding_key"` // Device binding key
	Stage      ConfigStage `json:"stage"`       // draft/active/deprecated
}

ConfigRecord represents a single configuration record

type ConfigStage 

type ConfigStage string

ConfigStage defines the stage of a config 

const (
	StageDraft      ConfigStage = "draft"
	StageActive     ConfigStage = "active"
	StageDeprecated ConfigStage = "deprecated"
)

type ConfigStore 

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

func NewConfigStore 

func NewConfigStore(path string) (*ConfigStore, error)

func NewConfigStoreWithConfig 

func NewConfigStoreWithConfig(cfg StorageConfig) (*ConfigStore, error)

func (*ConfigStore) Close 

func (s *ConfigStore) Close() error

func (*ConfigStore) Delete 

func (s *ConfigStore) Delete(key string) error

func (*ConfigStore) DeleteFromBucket 

func (s *ConfigStore) DeleteFromBucket(bucketName, key string) error

func (*ConfigStore) EnsureBucket 

func (s *ConfigStore) EnsureBucket(bucketName string) error

func (*ConfigStore) GeneratePartitionKey 

func (s *ConfigStore) GeneratePartitionKey(deviceID string, timestamp time.Time) string

func (*ConfigStore) GenerateSecondaryKey 

func (s *ConfigStore) GenerateSecondaryKey(deviceID string, timestamp time.Time, sequence int64) string

func (*ConfigStore) Get 

func (s *ConfigStore) Get(key string) (*ConfigRecord, bool)

func (*ConfigStore) GetAll 

func (s *ConfigStore) GetAll() []*ConfigRecord

func (*ConfigStore) GetByBindingKey 

func (s *ConfigStore) GetByBindingKey(bindingKey string) []*ConfigRecord

func (*ConfigStore) GetDeviceBucketName 

func (s *ConfigStore) GetDeviceBucketName(channelID string) string

func (*ConfigStore) GetDigest 

func (s *ConfigStore) GetDigest(nodeID string) *Digest

func (*ConfigStore) GetFromBucket 

func (s *ConfigStore) GetFromBucket(bucketName, key string, result interface{}) error

func (*ConfigStore) GetSnapshotBucketName 

func (s *ConfigStore) GetSnapshotBucketName(deviceID string) string

func (*ConfigStore) GetSyncNodeBucketName 

func (s *ConfigStore) GetSyncNodeBucketName(nodeID string) string

func (*ConfigStore) ListBucketKeys 

func (s *ConfigStore) ListBucketKeys(bucketName string, prefix string) ([]string, error)

func (*ConfigStore) PruneOldRecords 

func (s *ConfigStore) PruneOldRecords(bucketName string, maxAge time.Duration) error

func (*ConfigStore) Put 

func (s *ConfigStore) Put(rec *ConfigRecord) error

func (*ConfigStore) PutToBucket 

func (s *ConfigStore) PutToBucket(bucketName, key string, data interface{}) error

type ConsistencyReport 

type ConsistencyReport struct {
	OverallStatus string                 `json:"overall_status"`
	Details       map[string]interface{} `json:"details"`
}

ConsistencyReport represents a consistency check report

type DeviceCode 

type DeviceCode struct {
	Protocol     string `json:"protocol"`
	VendorID     string `json:"vendor_id"`
	ModelID      string `json:"model_id"`
	SerialNumber string `json:"serial_number"`
	Extra        string `json:"extra,omitempty"`
}

DeviceCode represents a parsed device code

func ParseDeviceCode 

func ParseDeviceCode(deviceCode string) (*DeviceCode, error)

ParseDeviceCode parses a device code string Format: protocol-vendor-model-SN-serial-extra

type DeviceFingerprint 

type DeviceFingerprint struct {
	Vendor string `json:"vendor"`
	Model  string `json:"model"`
	SN     string `json:"sn"`
}

DeviceFingerprint for device identity

type DeviceIndexEntry 

type DeviceIndexEntry struct {
	DeviceID string   `json:"device_id"`
	NodeIDs  []string `json:"node_ids"`
}

DeviceIndexEntry 记录某个设备出现在哪些节点上

type DeviceSnapshot 

type DeviceSnapshot struct {
	DeviceID string                 `json:"device_id"`
	Nodes    map[string]*TreeDevice `json:"nodes"` // nodeID → device config
}

DeviceSnapshot 按设备ID检索的跨节点快照

type DiffItem 

type DiffItem struct {
	Key         string         `json:"key"`
	Kind        string         `json:"kind"` // file or structure
	Type        string         `json:"type"` // same, different, onlySource, onlyTarget
	Path        string         `json:"path,omitempty"`
	File        string         `json:"file,omitempty"`
	SourceValue any            `json:"source_value,omitempty"`
	TargetValue any            `json:"target_value,omitempty"`
	SourceHash  string         `json:"source_hash,omitempty"`
	TargetHash  string         `json:"target_hash,omitempty"`
	SourceMeta  map[string]any `json:"source_meta,omitempty"`
	TargetMeta  map[string]any `json:"target_meta,omitempty"`
}

DiffItem represents one diff entry.

type DiffResult 

type DiffResult struct {
	SourceNodeID   string      `json:"source_node_id"`
	TargetNodeID   string      `json:"target_node_id"`
	Summary        DiffSummary `json:"summary"`
	FileLevel      []DiffItem  `json:"file_level"`
	StructureLevel []DiffItem  `json:"structure_level"`
}

DiffResult contains file-level and structure-level diff details.

func CompareSnapshots 

func CompareSnapshots(source, target *NodeSnapshot) *DiffResult

CompareSnapshots produces file-level and structure-level diffs.

type DiffSummary 

type DiffSummary struct {
	Total              int `json:"total"`
	Same               int `json:"same"`
	Different          int `json:"different"`
	OnlySource         int `json:"only_source"`
	OnlyTarget         int `json:"only_target"`
	FileDifferent      int `json:"file_different"`
	StructureDifferent int `json:"structure_different"`
}

DiffSummary aggregates diff counters.

type Digest 

type Digest struct {
	NodeID string            `json:"node_id"`
	Keys   map[string]uint64 `json:"keys"` // key -> version
	Hash   string            `json:"hash"` // Global merkle root
}

Digest for anti-entropy

type DiscoveryManager 

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

DiscoveryManager manages peer discovery with fallback

func NewDiscoveryManager 

func NewDiscoveryManager(dataPath string, gossip *GossipManager) (*DiscoveryManager, error)

NewDiscoveryManager creates a new DiscoveryManager

func (*DiscoveryManager) AddPeer 

func (d *DiscoveryManager) AddPeer(addr string)

AddPeer adds a peer to cache

func (*DiscoveryManager) Close 

func (d *DiscoveryManager) Close()

Close closes the discovery manager

func (*DiscoveryManager) DisableDiscovery 

func (d *DiscoveryManager) DisableDiscovery()

DisableDiscovery disables all discovery mechanisms

func (*DiscoveryManager) EnableDiscovery 

func (d *DiscoveryManager) EnableDiscovery()

EnableDiscovery enables all discovery mechanisms

func (*DiscoveryManager) GetAllPeers 

func (d *DiscoveryManager) GetAllPeers() []*PeerInfo

GetAllPeers returns all peers from all discovery mechanisms

func (*DiscoveryManager) GetKnownPeers 

func (d *DiscoveryManager) GetKnownPeers() []string

GetKnownPeers gets known peers

func (*DiscoveryManager) SetStaticSeeds 

func (d *DiscoveryManager) SetStaticSeeds(seeds []string)

SetStaticSeeds sets static seed nodes

func (*DiscoveryManager) TryConnectPeers 

func (d *DiscoveryManager) TryConnectPeers()

TryConnectPeers tries to connect to known peers

type GossipManager 

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

GossipManager manages gossip-based communication

func NewGossipManager 

func NewGossipManager(ctx context.Context, port int) (*GossipManager, error)

NewGossipManager creates a new GossipManager

func (*GossipManager) Close 

func (g *GossipManager) Close()

Close closes the gossip manager

func (*GossipManager) GetPeers 

func (g *GossipManager) GetPeers() []*PeerInfo

GetPeers returns connected peers

func (*GossipManager) HostID 

func (g *GossipManager) HostID() peer.ID

HostID returns the host ID

func (*GossipManager) Messages 

func (g *GossipManager) Messages() <-chan *SyncMessage

Messages returns the message channel

func (*GossipManager) Publish 

func (g *GossipManager) Publish(msg *SyncMessage) error

Publish publishes a message

func (*GossipManager) SetupMDNS 

func (g *GossipManager) SetupMDNS(serviceName string) error

SetupMDNS sets up mDNS discovery

type GroupManager 

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

GroupManager manages network groups

func NewGroupManager 

func NewGroupManager() *GroupManager

NewGroupManager creates a new group manager

func (*GroupManager) CreateGroup 

func (gm *GroupManager) CreateGroup(groupID, name, description string) (*NetworkGroup, error)

CreateGroup creates a new network group

func (*GroupManager) DeleteGroup 

func (gm *GroupManager) DeleteGroup(groupID string) error

DeleteGroup deletes a group

func (*GroupManager) DeserializeGroup 

func (gm *GroupManager) DeserializeGroup(data []byte) (*NetworkGroup, error)

DeserializeGroup deserializes a group from JSON

func (*GroupManager) GetGroup 

func (gm *GroupManager) GetGroup(groupID string) (*NetworkGroup, bool)

GetGroup gets a group by ID

func (*GroupManager) GetGroupMembers 

func (gm *GroupManager) GetGroupMembers(groupID string) ([]string, error)

GetGroupMembers returns all members of a group

func (*GroupManager) GetGroupsByPeer 

func (gm *GroupManager) GetGroupsByPeer(peerID peer.ID) []*NetworkGroup

GetGroupsByPeer returns all groups that a peer belongs to

func (*GroupManager) JoinGroup 

func (gm *GroupManager) JoinGroup(groupID string, peerID peer.ID) error

JoinGroup adds a peer to a group

func (*GroupManager) LeaveGroup 

func (gm *GroupManager) LeaveGroup(groupID string, peerID peer.ID) error

LeaveGroup removes a peer from a group

func (*GroupManager) ListGroups 

func (gm *GroupManager) ListGroups() []*NetworkGroup

ListGroups returns all groups

func (*GroupManager) SerializeGroup 

func (gm *GroupManager) SerializeGroup(groupID string) ([]byte, error)

SerializeGroup serializes a group to JSON

type GroupPeerInfo 

type GroupPeerInfo struct {
	PeerID       peer.ID    `json:"peer_id"`
	NodeID       string     `json:"node_id"`
	GroupID      string     `json:"group_id"`
	Online       bool       `json:"online"`
	LastSeen     time.Time  `json:"last_seen"`
	LastSync     time.Time  `json:"last_sync"`
	SyncVersion  uint64     `json:"sync_version"`
	OfflineSince *time.Time `json:"offline_since,omitempty"`
}

GroupPeerInfo contains peer information in a group

type GroupSyncManager 

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

GroupSyncManager manages group-based synchronization

func NewGroupSyncManager 

func NewGroupSyncManager(storage StorageInterface) *GroupSyncManager

NewGroupSyncManager creates a new group sync manager

func (*GroupSyncManager) CanSyncWithPeer 

func (gsm *GroupSyncManager) CanSyncWithPeer(peerID peer.ID) (bool, error)

CanSyncWithPeer checks if sync can proceed with a peer

func (*GroupSyncManager) Close 

func (gsm *GroupSyncManager) Close() error

Close closes the group sync manager

func (*GroupSyncManager) CreateGroup 

func (gsm *GroupSyncManager) CreateGroup(groupID, name, description string) error

CreateGroup creates a new network group

func (*GroupSyncManager) DeleteGroup 

func (gsm *GroupSyncManager) DeleteGroup(groupID string) error

DeleteGroup deletes a group

func (*GroupSyncManager) DeserializeGroup 

func (gsm *GroupSyncManager) DeserializeGroup(data []byte) error

DeserializeGroup deserializes group configuration from sync

func (*GroupSyncManager) GetGroupInfo 

func (gsm *GroupSyncManager) GetGroupInfo(groupID string) (*NetworkGroup, bool)

GetGroupInfo returns group information

func (*GroupSyncManager) GetGroupMembers 

func (gsm *GroupSyncManager) GetGroupMembers(groupID string) ([]*GroupPeerInfo, error)

GetGroupMembers returns all members of a group

func (*GroupSyncManager) GetPeerInfo 

func (gsm *GroupSyncManager) GetPeerInfo(peerID peer.ID) (*GroupPeerInfo, bool)

GetPeerInfo returns peer information

func (*GroupSyncManager) JoinGroup 

func (gsm *GroupSyncManager) JoinGroup(groupID, nodeID string, peerID peer.ID) error

JoinGroup makes the local peer join a group

func (*GroupSyncManager) LeaveGroup 

func (gsm *GroupSyncManager) LeaveGroup(groupID string, peerID peer.ID) error

LeaveGroup makes the local peer leave a group

func (*GroupSyncManager) ListGroups 

func (gsm *GroupSyncManager) ListGroups() []*NetworkGroup

ListGroups lists all groups

func (*GroupSyncManager) RequestSync 

func (gsm *GroupSyncManager) RequestSync(groupID string, targetPeer peer.ID, fullSync bool) error

RequestSync requests sync with a specific peer in the group

func (*GroupSyncManager) SerializeGroup 

func (gsm *GroupSyncManager) SerializeGroup(groupID string) ([]byte, error)

SerializeGroup serializes group configuration for sync

func (*GroupSyncManager) UpdatePeerStatus 

func (gsm *GroupSyncManager) UpdatePeerStatus(peerID peer.ID, online bool)

UpdatePeerStatus updates peer online/offline status

type LegacyPeerInfo 

type LegacyPeerInfo struct {
	ID       string
	Addr     string
	Status   string
	Version  uint64
	IsLeader bool
}

LegacyPeerInfo matches old PeerInfo

type LegacySyncAdapter 

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

LegacySyncAdapter provides compatibility with existing SyncManager

func NewLegacySyncAdapter 

func NewLegacySyncAdapter(syncMgr *SyncManager) *LegacySyncAdapter

NewLegacySyncAdapter creates a new adapter

func (*LegacySyncAdapter) GetConfig 

func (a *LegacySyncAdapter) GetConfig(key string, out interface{}) error

GetConfig gets config in legacy format

func (*LegacySyncAdapter) GetLocalNodeID 

func (a *LegacySyncAdapter) GetLocalNodeID() string

GetLocalNodeID returns local node ID

func (*LegacySyncAdapter) GetNodeStatus 

func (a *LegacySyncAdapter) GetNodeStatus() map[string]interface{}

GetNodeStatus returns node status info

func (*LegacySyncAdapter) GetPeers 

func (a *LegacySyncAdapter) GetPeers() []*LegacyPeerInfo

GetPeers returns peers in legacy format

func (*LegacySyncAdapter) PutConfig 

func (a *LegacySyncAdapter) PutConfig(key string, value interface{}, bindingKey string) error

PutConfig puts config in legacy format

func (*LegacySyncAdapter) StartSync 

func (a *LegacySyncAdapter) StartSync() error

StartSync starts sync (compatibility)

type NetworkGroup 

type NetworkGroup struct {
	GroupID     string    `json:"group_id"`
	Name        string    `json:"name"`
	Description string    `json:"description"`
	Members     []string  `json:"members"`
	CreatedAt   time.Time `json:"created_at"`
	UpdatedAt   time.Time `json:"updated_at"`
	// contains filtered or unexported fields
}

NetworkGroup represents a sync group that multiple gateways can join

func NewNetworkGroup 

func NewNetworkGroup(groupID, name, description string) *NetworkGroup

NewNetworkGroup creates a new network group

func (*NetworkGroup) GetMembers 

func (ng *NetworkGroup) GetMembers() []string

GetMembers returns a copy of members list

func (*NetworkGroup) IsMember 

func (ng *NetworkGroup) IsMember(peerID peer.ID) bool

IsMember checks if a peer is a member of this group

func (*NetworkGroup) JoinGroup 

func (ng *NetworkGroup) JoinGroup(peerID peer.ID) error

JoinGroup adds a peer to this group

func (*NetworkGroup) LeaveGroup 

func (ng *NetworkGroup) LeaveGroup(peerID peer.ID)

LeaveGroup removes a peer from this group

type NodeIDGenerator 

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

NodeIDGenerator handles node ID generation and management

func NewNodeIDGenerator 

func NewNodeIDGenerator() *NodeIDGenerator

NewNodeIDGenerator creates a new NodeIDGenerator

func (*NodeIDGenerator) GenerateNodeID 

func (n *NodeIDGenerator) GenerateNodeID() string

GenerateNodeID generates a unique node ID based on system information

func (*NodeIDGenerator) GetHostname 

func (n *NodeIDGenerator) GetHostname() string

GetHostname returns the hostname

func (*NodeIDGenerator) GetMACAddresses 

func (n *NodeIDGenerator) GetMACAddresses() []string

GetMACAddresses returns MAC addresses

type NodeMeta 

type NodeMeta struct {
	NodeID     string      `json:"node_id"`
	NodeName   string      `json:"node_name"`
	CapturedAt time.Time   `json:"captured_at"`
	Summary    TreeSummary `json:"summary"`
	Status     string      `json:"status"` // online, offline, syncing
	Version    uint64      `json:"version"`
}

NodeMeta 存储节点的基础元信息

type NodeRole 

type NodeRole string

NodeRole defines the role of a node 

const (
	RoleLeader   NodeRole = "leader"
	RoleFollower NodeRole = "follower"
	RoleReadonly NodeRole = "readonly"
)

type NodeSnapshot 

type NodeSnapshot struct {
	NodeID     string        `json:"node_id"`
	NodeName   string        `json:"node_name,omitempty"`
	CapturedAt time.Time     `json:"captured_at"`
	Files      []ConfigFile  `json:"files"`
	Channels   []TreeChannel `json:"channels"`
	Northbound []TreeSection `json:"northbound"`
	System     []TreeSection `json:"system"`
	Summary    TreeSummary   `json:"summary"`
}

NodeSnapshot represents one node's configuration tree and file view.

func BuildNodeSnapshot 

func BuildNodeSnapshot(nodeID string, cfg *config.Config) *NodeSnapshot

BuildNodeSnapshot converts a loaded config into a tree snapshot.

type PeerCache 

type PeerCache struct {
	KnownPeers []string `json:"known_peers"`
}

PeerCache stores known peers for discovery recovery

type PeerInfo 

type PeerInfo struct {
	ID       peer.ID
	Addr     string
	LastSeen time.Time
	Status   string // online, offline, syncing
	Version  uint64
	IsLeader bool
	Role     NodeRole
}

PeerInfo holds information about a connected peer

type RoleManager 

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

RoleManager manages node roles

func NewRoleManager 

func NewRoleManager(selfID peer.ID) *RoleManager

NewRoleManager creates a new RoleManager

func (*RoleManager) CanWrite 

func (r *RoleManager) CanWrite() bool

CanWrite checks if current node can write

func (*RoleManager) ElectLeader 

func (r *RoleManager) ElectLeader() NodeRole

ElectLeader elects a leader based on node ID

func (*RoleManager) GetRole 

func (r *RoleManager) GetRole() NodeRole

GetRole gets current role

func (*RoleManager) IsLeader 

func (r *RoleManager) IsLeader() bool

IsLeader checks if current node is leader

func (*RoleManager) SetReadonly 

func (r *RoleManager) SetReadonly()

SetReadonly sets node to readonly

func (*RoleManager) UpdatePeer 

func (r *RoleManager) UpdatePeer(peerID peer.ID, info *PeerInfo)

UpdatePeer updates peer info

type SimpleSyncRequest 

type SimpleSyncRequest struct {
	NodeID     string `json:"node_id"`
	DeviceCode string `json:"device_code"`
}

SimpleSyncRequest represents a simple sync request

type Snapshot 

type Snapshot struct {
	ID          string        `json:"id"`
	Name        string        `json:"name"`
	NodeID      string        `json:"node_id"`
	NodeName    string        `json:"node_name"`
	CapturedAt  time.Time     `json:"captured_at"`
	Size        int64         `json:"size"`
	Description string        `json:"description"`
	Tags        []string      `json:"tags"`
	Data        *NodeSnapshot `json:"data"`
	FilePath    string        `json:"file_path,omitempty"`
}

Snapshot represents a saved configuration snapshot

type SnapshotManager 

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

SnapshotManager handles snapshot backup, restore, and remote sync operations

func NewSnapshotManager 

func NewSnapshotManager(syncDir string, syncMgr *SyncManager) (*SnapshotManager, error)

NewSnapshotManager creates a new SnapshotManager

func (*SnapshotManager) ClearNodeConfig 

func (sm *SnapshotManager) ClearNodeConfig(nodeID string) error

ClearNodeConfig clears a node's configuration (simulates device replacement)

func (*SnapshotManager) CreateSnapshot 

func (sm *SnapshotManager) CreateSnapshot(nodeID, name, description string, tags []string) (*Snapshot, error)

CreateSnapshot creates a new snapshot from the current node configuration

func (*SnapshotManager) DeleteSnapshot 

func (sm *SnapshotManager) DeleteSnapshot(snapshotID string) error

DeleteSnapshot deletes a snapshot

func (*SnapshotManager) ExportSnapshot 

func (sm *SnapshotManager) ExportSnapshot(snapshotID, exportPath string) error

ExportSnapshot exports a snapshot to a file

func (*SnapshotManager) GetSnapshot 

func (sm *SnapshotManager) GetSnapshot(snapshotID string) (*Snapshot, bool)

GetSnapshot returns a snapshot by ID

func (*SnapshotManager) GetSnapshotStats 

func (sm *SnapshotManager) GetSnapshotStats() map[string]interface{}

GetSnapshotStats returns statistics about snapshots

func (*SnapshotManager) GetSnapshots 

func (sm *SnapshotManager) GetSnapshots() []*Snapshot

GetSnapshots returns all snapshots

func (*SnapshotManager) GetSnapshotsByNode 

func (sm *SnapshotManager) GetSnapshotsByNode(nodeID string) []*Snapshot

GetSnapshotsByNode returns snapshots for a specific node

func (*SnapshotManager) ImportSnapshot 

func (sm *SnapshotManager) ImportSnapshot(importPath string) (*Snapshot, error)

ImportSnapshot imports a snapshot from a file

func (*SnapshotManager) PullFromRemote 

func (sm *SnapshotManager) PullFromRemote(peerID string) (*NodeSnapshot, error)

PullFromRemote pulls configuration from a remote node

func (*SnapshotManager) RestoreSnapshot 

func (sm *SnapshotManager) RestoreSnapshot(snapshotID string) error

RestoreSnapshot restores a snapshot to the sync manager

func (*SnapshotManager) RestoreToRemote 

func (sm *SnapshotManager) RestoreToRemote(peerID string, snapshotID string) error

RestoreToRemote restores configuration to a remote node (triggers sync)

type StaticSeedDiscovery 

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

func NewStaticSeedDiscovery 

func NewStaticSeedDiscovery(gossip *GossipManager, seeds []string) *StaticSeedDiscovery

func (*StaticSeedDiscovery) ConnectToSeeds 

func (s *StaticSeedDiscovery) ConnectToSeeds()

func (*StaticSeedDiscovery) Disable 

func (s *StaticSeedDiscovery) Disable()

func (*StaticSeedDiscovery) Enable 

func (s *StaticSeedDiscovery) Enable()

func (*StaticSeedDiscovery) GetPeers 

func (s *StaticSeedDiscovery) GetPeers() []*PeerInfo

func (*StaticSeedDiscovery) GetSeeds 

func (s *StaticSeedDiscovery) GetSeeds() []string

func (*StaticSeedDiscovery) SetSeeds 

func (s *StaticSeedDiscovery) SetSeeds(seeds []string)

type StorageConfig 

type StorageConfig struct {
	Path         string
	PageSize     int
	CacheSize    int
	Timeout      time.Duration
	NoGrowSync   bool
	FreelistType bolt.FreelistType
	WriteBuffer  int
}

func DefaultStorageConfig 

func DefaultStorageConfig() StorageConfig

type StorageInterface 

type StorageInterface interface {
	Close() error
}

StorageInterface defines the storage interface

type SyncManager 

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

SyncManager is the main manager for the sync system

func NewSyncManager 

func NewSyncManager(ctx context.Context, dataPath string, port int) (*SyncManager, error)

NewSyncManager creates a new SyncManager

func (*SyncManager) AddMemberToGroup 

func (s *SyncManager) AddMemberToGroup(groupID, peerID string) error

AddMemberToGroup adds a member to a group

func (*SyncManager) AutoJoinGroup 

func (s *SyncManager) AutoJoinGroup()

AutoJoinGroup automatically joins groups

func (*SyncManager) CheckConsistency 

func (s *SyncManager) CheckConsistency() (*ConsistencyReport, error)

CheckConsistency checks data consistency across peers

func (*SyncManager) ClearNodeConfig 

func (s *SyncManager) ClearNodeConfig(nodeID string) error

ClearNodeConfig clears a node's configuration

func (*SyncManager) CompareSnapshots 

func (s *SyncManager) CompareSnapshots(sourceNodeID, targetNodeID string) (*DiffResult, error)

CompareSnapshots compares two stored snapshots.

func (*SyncManager) ConnectToPeerByID 

func (s *SyncManager) ConnectToPeerByID(peerID string) error

ConnectToPeerByID connects to a peer by ID

func (*SyncManager) CreateGroup 

func (s *SyncManager) CreateGroup(groupID, name, description string) error

CreateGroup creates a new network group

func (*SyncManager) CreateSnapshot 

func (s *SyncManager) CreateSnapshot(nodeID, name, description string, tags []string) (*Snapshot, error)

CreateSnapshot creates a new snapshot for a node

func (*SyncManager) DeleteGroup 

func (s *SyncManager) DeleteGroup(groupID string) error

DeleteGroup deletes a group

func (*SyncManager) DeleteSnapshot 

func (s *SyncManager) DeleteSnapshot(snapshotID string) error

DeleteSnapshot deletes a snapshot

func (*SyncManager) DeviceHello 

func (s *SyncManager) DeviceHello(deviceKey string)

DeviceHello sends a hello for device takeover

func (*SyncManager) DisableDiscovery 

func (s *SyncManager) DisableDiscovery()

DisableDiscovery disables discovery

func (*SyncManager) DisconnectFromPeer 

func (s *SyncManager) DisconnectFromPeer(peerID string) error

DisconnectFromPeer disconnects from a peer

func (*SyncManager) EnableDiscovery 

func (s *SyncManager) EnableDiscovery()

EnableDiscovery enables discovery

func (*SyncManager) GetAllClusterNodes 

func (s *SyncManager) GetAllClusterNodes() ([]NodeMeta, error)

GetAllClusterNodes 获取所有已知节点

func (*SyncManager) GetAllConfigs 

func (s *SyncManager) GetAllConfigs() []*ConfigRecord

GetAllConfigs gets all configs

func (*SyncManager) GetAllGroups 

func (s *SyncManager) GetAllGroups() []*NetworkGroup

GetAllGroups returns all groups

func (*SyncManager) GetClusterDevices 

func (s *SyncManager) GetClusterDevices() ([]string, error)

GetClusterDevices 获取所有已知设备列表（去重）

func (*SyncManager) GetClusterStore 

func (s *SyncManager) GetClusterStore() *ClusterStore

GetClusterStore 返回底层 ClusterStore（供 HTTP handler 使用）

func (*SyncManager) GetClusterSummary 

func (s *SyncManager) GetClusterSummary() (*ClusterSummary, error)

GetClusterSummary 获取集群聚合统计

func (*SyncManager) GetConfig 

func (s *SyncManager) GetConfig(key string) (*ConfigRecord, bool)

GetConfig gets a config

func (*SyncManager) GetConnectedPeers 

func (s *SyncManager) GetConnectedPeers() []*PeerInfo

GetConnectedPeers returns connected peers

func (*SyncManager) GetDeviceClusterSnapshot 

func (s *SyncManager) GetDeviceClusterSnapshot(deviceID string) (*DeviceSnapshot, error)

GetDeviceClusterSnapshot 按设备ID获取跨节点快照

func (*SyncManager) GetGroup 

func (s *SyncManager) GetGroup(groupID string) (*NetworkGroup, error)

GetGroup returns a group by ID

func (*SyncManager) GetGroupInfo 

func (s *SyncManager) GetGroupInfo(groupID string) (*NetworkGroup, bool)

GetGroupInfo gets group info

func (*SyncManager) GetGroupMembers 

func (s *SyncManager) GetGroupMembers(groupID string) ([]*GroupPeerInfo, error)

GetGroupMembers gets group members

func (*SyncManager) GetHost 

func (s *SyncManager) GetHost() interface{}

GetHost returns the underlying host

func (*SyncManager) GetJoinedGroups 

func (s *SyncManager) GetJoinedGroups() []*NetworkGroup

GetJoinedGroups returns joined groups

func (*SyncManager) GetNodeClusterDevices 

func (s *SyncManager) GetNodeClusterDevices(nodeID string) ([]TreeDevice, error)

GetNodeClusterDevices 获取指定节点的设备列表

func (*SyncManager) GetNodeInfo 

func (s *SyncManager) GetNodeInfo() map[string]interface{}

GetNodeInfo gets node info

func (*SyncManager) GetPeerIDString 

func (s *SyncManager) GetPeerIDString() string

GetPeerIDString returns the peer ID as string

func (*SyncManager) GetSnapshot 

func (s *SyncManager) GetSnapshot(nodeID string) (*NodeSnapshot, bool)

GetSnapshot returns a node snapshot — 先从内存取，再尝试 bbolt 恢复。

func (*SyncManager) GetSnapshotByID 

func (s *SyncManager) GetSnapshotByID(snapshotID string) (*Snapshot, bool)

GetSnapshotByID returns a snapshot by ID

func (*SyncManager) GetSnapshotStats 

func (s *SyncManager) GetSnapshotStats() map[string]interface{}

GetSnapshotStats returns snapshot statistics

func (*SyncManager) GetSnapshots 

func (s *SyncManager) GetSnapshots() []*Snapshot

GetSnapshots returns all snapshots

func (*SyncManager) GetSnapshotsByNode 

func (s *SyncManager) GetSnapshotsByNode(nodeID string) []*Snapshot

GetSnapshotsByNode returns snapshots for a specific node

func (*SyncManager) GetStatus 

func (s *SyncManager) GetStatus() map[string]interface{}

GetStatus returns the sync status

func (*SyncManager) GetTakeoverEvents 

func (s *SyncManager) GetTakeoverEvents(deviceKey string) []*TakeoverEvent

GetTakeoverEvents returns takeover history.

func (*SyncManager) HandleSyncMessage 

func (s *SyncManager) HandleSyncMessage(msg *SyncMessage) error

HandleSyncMessage handles sync messages

func (*SyncManager) JoinGroup 

func (s *SyncManager) JoinGroup(groupID, nodeID string) error

JoinGroup joins a group

func (*SyncManager) LeaveGroup 

func (s *SyncManager) LeaveGroup(groupID string) error

LeaveGroup leaves a group

func (*SyncManager) ListGroups 

func (s *SyncManager) ListGroups() []*NetworkGroup

ListGroups lists all groups

func (*SyncManager) OnPeerConnected 

func (s *SyncManager) OnPeerConnected(peerID peer.ID)

OnPeerConnected handles peer connection

func (*SyncManager) OnPeerDisconnected 

func (s *SyncManager) OnPeerDisconnected(peerID peer.ID)

OnPeerDisconnected handles peer disconnection

func (*SyncManager) PullFromRemote 

func (s *SyncManager) PullFromRemote(peerID string) (*NodeSnapshot, error)

PullFromRemote pulls configuration from a remote node

func (*SyncManager) PutConfig 

func (s *SyncManager) PutConfig(key string, value []byte, bindingKey string) error

PutConfig puts a config

func (*SyncManager) RestoreSnapshot 

func (s *SyncManager) RestoreSnapshot(snapshotID string) error

RestoreSnapshot restores a snapshot

func (*SyncManager) RestoreToRemote 

func (s *SyncManager) RestoreToRemote(peerID string, snapshotID string) error

RestoreToRemote restores configuration to a remote node

func (*SyncManager) SeedSnapshot 

func (s *SyncManager) SeedSnapshot(nodeID string, cfg *config.Config)

SeedSnapshot stores a tree snapshot for a node, usually the local node. 同时持久化到 bbolt ClusterStore + 内存缓存。

func (*SyncManager) Start 

func (s *SyncManager) Start() error

Start starts the sync manager

func (*SyncManager) StartDeviceTakeover 

func (s *SyncManager) StartDeviceTakeover(deviceKey, targetPeer string) error

StartDeviceTakeover triggers the HELLO -> TAKEOVER -> FULL_CONFIG flow.

func (*SyncManager) Stop 

func (s *SyncManager) Stop()

Stop stops the sync manager

func (*SyncManager) TriggerSync 

func (s *SyncManager) TriggerSync(syncType string) error

TriggerSync triggers a sync operation

func (*SyncManager) ValidateDeviceCode 

func (s *SyncManager) ValidateDeviceCode(deviceCode string) (*DeviceCode, error)

ValidateDeviceCode validates a device code

type SyncMessage 

type SyncMessage struct {
	Version     string                 `json:"version"`
	MessageType string                 `json:"message_type"` // announce, pull, full_config, hello, takeover, digest
	MessageID   string                 `json:"message_id"`
	SourcePeer  string                 `json:"source_peer"`
	TargetPeer  string                 `json:"target_peer"`
	Timestamp   time.Time              `json:"timestamp"`
	PayloadType string                 `json:"payload_type"`
	Payload     map[string]interface{} `json:"payload"`
}

SyncMessage defines the synchronization message format

type SyncMeta 

type SyncMeta struct {
	Operation  string    `json:"operation"`
	Version    uint64    `json:"version"`
	Checksum   string    `json:"checksum"`
	LastSyncAt time.Time `json:"last_sync_at"`
}

SyncMeta contains metadata for synchronization

type TakeoverEvent 

type TakeoverEvent struct {
	ID         string        `json:"id"`
	DeviceKey  string        `json:"device_key"`
	SourcePeer string        `json:"source_peer"`
	TargetPeer string        `json:"target_peer"`
	Stage      TakeoverStage `json:"stage"`
	Status     string        `json:"status"`
	Message    string        `json:"message,omitempty"`
	Timestamp  time.Time     `json:"timestamp"`
}

TakeoverEvent describes a takeover state transition.

type TakeoverLock 

type TakeoverLock struct {
	DeviceKey string        `json:"device_key"`
	Owner     peer.ID       `json:"owner"`
	TTL       time.Duration `json:"ttl"`
	ExpiresAt time.Time     `json:"expires_at"`
}

TakeoverLock for distributed takeover control

type TakeoverManager 

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

TakeoverManager manages device takeover

func NewTakeoverManager 

func NewTakeoverManager() *TakeoverManager

NewTakeoverManager creates a new TakeoverManager

func (*TakeoverManager) CleanupExpiredLocks 

func (t *TakeoverManager) CleanupExpiredLocks()

CleanupExpiredLocks cleans up expired locks

func (*TakeoverManager) GetEvents 

func (t *TakeoverManager) GetEvents(deviceKey string) []*TakeoverEvent

GetEvents returns takeover events for a given device or all events when deviceKey is empty.

func (*TakeoverManager) GetLockStatus 

func (t *TakeoverManager) GetLockStatus(deviceKey string) (*TakeoverLock, bool)

GetLockStatus gets the lock status

func (*TakeoverManager) RecordEvent 

func (t *TakeoverManager) RecordEvent(event *TakeoverEvent)

RecordEvent stores a takeover lifecycle entry.

func (*TakeoverManager) ReleaseLock 

func (t *TakeoverManager) ReleaseLock(deviceKey string)

ReleaseLock releases a takeover lock

func (*TakeoverManager) TryLock 

func (t *TakeoverManager) TryLock(deviceKey string, owner peer.ID, ttl time.Duration) bool

TryLock tries to acquire a takeover lock

type TakeoverStage 

type TakeoverStage string

TakeoverStage records the flow stage for one device takeover attempt. 

const (
	TakeoverStageHello      TakeoverStage = "hello"
	TakeoverStageTakeover   TakeoverStage = "takeover"
	TakeoverStageFullConfig TakeoverStage = "full_config"
	TakeoverStageCompleted  TakeoverStage = "completed"
	TakeoverStageFailed     TakeoverStage = "failed"
)

type TreeChannel 

type TreeChannel struct {
	Type       string         `json:"type"`
	ID         string         `json:"id"`
	Label      string         `json:"label"`
	Name       string         `json:"name"`
	Protocol   string         `json:"protocol"`
	Status     string         `json:"status"`
	Enabled    bool           `json:"enabled"`
	HasDiff    bool           `json:"has_diff"`
	SourceFile string         `json:"source_file,omitempty"`
	Config     map[string]any `json:"config,omitempty"`
	Devices    []TreeDevice   `json:"devices,omitempty"`
}

TreeChannel is the top-level channel node.

type TreeDevice 

type TreeDevice struct {
	Type       string         `json:"type"`
	ID         string         `json:"id"`
	Label      string         `json:"label"`
	Name       string         `json:"name"`
	Status     string         `json:"status"`
	Enabled    bool           `json:"enabled"`
	HasDiff    bool           `json:"has_diff"`
	PointCount int            `json:"point_count"`
	SourceFile string         `json:"source_file,omitempty"`
	Config     map[string]any `json:"config,omitempty"`
	Points     []TreePoint    `json:"points,omitempty"`
}

TreeDevice is a device node.

type TreePoint 

type TreePoint struct {
	Type       string         `json:"type"`
	ID         string         `json:"id"`
	Label      string         `json:"label"`
	Name       string         `json:"name"`
	Status     string         `json:"status"`
	HasDiff    bool           `json:"has_diff"`
	SourceFile string         `json:"source_file,omitempty"`
	Config     map[string]any `json:"config,omitempty"`
}

TreePoint is a point node.

type TreeSection 

type TreeSection struct {
	Type       string         `json:"type"`
	ID         string         `json:"id"`
	Label      string         `json:"label"`
	Name       string         `json:"name"`
	Section    string         `json:"section"`
	Status     string         `json:"status"`
	Enabled    bool           `json:"enabled"`
	HasDiff    bool           `json:"has_diff"`
	SourceFile string         `json:"source_file,omitempty"`
	Config     map[string]any `json:"config,omitempty"`
	Children   []TreeSection  `json:"children,omitempty"`
}

TreeSection represents a node in northbound/system trees.

type TreeSummary 

type TreeSummary struct {
	ChannelCount    int `json:"channel_count"`
	DeviceCount     int `json:"device_count"`
	PointCount      int `json:"point_count"`
	NorthboundCount int `json:"northbound_count"`
	SystemCount     int `json:"system_count"`
	FileCount       int `json:"file_count"`
}

TreeSummary contains counters used by the UI.

type UDPDiscovery 

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

func NewUDPDiscovery 

func NewUDPDiscovery(gossip *GossipManager) (*UDPDiscovery, error)

func (*UDPDiscovery) Close 

func (u *UDPDiscovery) Close()

func (*UDPDiscovery) Disable 

func (u *UDPDiscovery) Disable()

func (*UDPDiscovery) Enable 

func (u *UDPDiscovery) Enable()

func (*UDPDiscovery) GetPeers 

func (u *UDPDiscovery) GetPeers() []*PeerInfo

type VectorClock 

type VectorClock map[string]uint64

func NewVectorClock 

func NewVectorClock() VectorClock

func (VectorClock) Clone 

func (vc VectorClock) Clone() VectorClock

func (VectorClock) Compare 

func (vc VectorClock) Compare(other VectorClock) int

func (VectorClock) Get 

func (vc VectorClock) Get(nodeID string) uint64

func (VectorClock) Increment 

func (vc VectorClock) Increment(nodeID string)

func (VectorClock) IsEmpty 

func (vc VectorClock) IsEmpty() bool

func (VectorClock) MarshalJSON 

func (vc VectorClock) MarshalJSON() ([]byte, error)

func (VectorClock) Merge 

func (vc VectorClock) Merge(other VectorClock)

func (VectorClock) String 

func (vc VectorClock) String() string

func (*VectorClock) UnmarshalJSON 

func (vc *VectorClock) UnmarshalJSON(data []byte) error

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.