internal

package
v0.4.0 Latest Latest
Warning

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

 Go to latest Published: Jul 13, 2026 License: Apache-2.0 Imports: 7 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	DefaultDLQMaxSizeBytes   = 10 << 20 // 10 MB
	DefaultRaftTickMs        = 100
	DefaultRaftHeartbeatTick = 5
	DefaultRaftElectionTick  = 20
)

Variables

View Source 
var (
	// ErrServiceUnavailable returned when node is not active (not leader or follower)
	ErrServiceUnavailable = errors.New("service unavailable: node is not active")

	// ErrUnknownCommand returned when command type is not recognized
	ErrUnknownCommand = errors.New("unknown command type")

	// ErrUnknownRequestType returned when cluster request type is not recognized
	ErrUnknownRequestType = errors.New("unknown request type")

	// ErrInvalidRaftMessage returned when Raft message decoding fails
	ErrInvalidRaftMessage = errors.New("invalid raft message")

	// ErrDeadlineExceeded returned when request exceeds deadline
	ErrDeadlineExceeded = errors.New("deadline exceeded")

	// ErrMaxRetriesExceeded returned when max retries are exhausted
	ErrMaxRetriesExceeded = errors.New("max retries exceeded")

	ErrUnknownNodeID = errors.New("unknown node ID for raft ID")

	ErrDuplicateJobID = errors.New("job ID already exist")

	ErrInvalidPayload = errors.New("invalid payload")

	ErrQueueFull = errors.New("queue is full")

	ErrTopicNotFound = errors.New("topic not found")
	ErrTopicExists   = errors.New("topic already exists")
)
View Source 
var CfgFile string

Functions

This section is empty.

Types

type ApiConfig 

type ApiConfig struct {
	ListenAddr     string `mapstructure:"listen_addr"`
	ApiPort        uint16 `mapstructure:"api_port"`
	TokenPath      string `mapstructure:"token_path"`
	TimeoutSeconds int    `mapstructure:"timeout_seconds"`
}

type ClusterConfig 

type ClusterConfig struct {
	InitialVoters        []string `mapstructure:"initial_voters"`
	ListenAddr           string   `mapstructure:"listen_addr"`
	QUICPort             uint16   `mapstructure:"quic_port"`
	SnapshotIntervalSec  uint64   `mapstructure:"snapshot_interval_sec"`
	SnapshotTriggerCount uint64   `mapstructure:"snapshot_trigger_count"`
	WALFlushThreshold    int      `mapstructure:"wal_flush_threshold"`
	CertPath             string   `mapstructure:"cert_path"`
	KeyPath              string   `mapstructure:"key_path"`
	CACertPath           string   `mapstructure:"ca_path"`
	DLQMaxSizeBytes      int64    `mapstructure:"dlq_max_size_bytes"`
	RaftTickMs           int      `mapstructure:"raft_tick_ms"`
	RaftHeartbeatTick    int      `mapstructure:"raft_heartbeat_tick"`
	RaftElectionTick     int      `mapstructure:"raft_election_tick"`

	// Discovery configuration
	DiscoveryKind     string `mapstructure:"discovery_kind"`
	DiscoveryYMLPath  string `mapstructure:"discovery_yml_path"`  // Required for DiscoveryKindStatic
	DiscoveryHTTPHost string `mapstructure:"discovery_http_host"` // Required for DiscoveryKindHttp

}

ClusterConfig - matches what NewClusterAgent expects

type ClusterMetrics 

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

ClusterMetrics caches metric references for cluster-level operations

func GetClusterMetrics 

func GetClusterMetrics() *ClusterMetrics

GetClusterMetrics returns the singleton ClusterMetrics instance

func (*ClusterMetrics) AddBytesReceived 

func (m *ClusterMetrics) AddBytesReceived(bytes uint64)

func (*ClusterMetrics) AddBytesSent 

func (m *ClusterMetrics) AddBytesSent(bytes uint64)

func (*ClusterMetrics) ConnectionAccepted 

func (m *ClusterMetrics) ConnectionAccepted()

func (*ClusterMetrics) ConnectionDropped 

func (m *ClusterMetrics) ConnectionDropped()

func (*ClusterMetrics) ConnectionError 

func (m *ClusterMetrics) ConnectionError()

func (*ClusterMetrics) ConnectionOpened 

func (m *ClusterMetrics) ConnectionOpened()

Connection methods

func (*ClusterMetrics) ConnectionRejected 

func (m *ClusterMetrics) ConnectionRejected()

func (*ClusterMetrics) LeaderChanged 

func (m *ClusterMetrics) LeaderChanged()

Condition methods

func (*ClusterMetrics) MessageReceived 

func (m *ClusterMetrics) MessageReceived()

func (*ClusterMetrics) MessageSent 

func (m *ClusterMetrics) MessageSent()

Request/Response methods

func (*ClusterMetrics) ReceiveError 

func (m *ClusterMetrics) ReceiveError()

func (*ClusterMetrics) SendError 

func (m *ClusterMetrics) SendError()

func (*ClusterMetrics) SetLastAppliedWalIndex 

func (m *ClusterMetrics) SetLastAppliedWalIndex(index uint64)

func (*ClusterMetrics) WalFlush 

func (m *ClusterMetrics) WalFlush()

WAL methods

type Config 

type Config struct {
	NodeID    string          `mapstructure:"node_id"`
	DataDir   string          `mapstructure:"data_dir"`
	Cluster   ClusterConfig   `mapstructure:"cluster"`
	Proxy     ProxyConfig     `mapstructure:"proxy"`
	WAL       WALConfig       `mapstructure:"wal"`
	Partition PartitionConfig `mapstructure:"partition"`
	Logging   LoggingConfig   `mapstructure:"logging"`
	ApiConfig ApiConfig       `mapstructure:"api"`
}

Config is the single source of truth for the whole application

func LoadConfig 

func LoadConfig() (*Config, error)

LoadConfig loads configuration from file, env vars, and flags

func NewConfig 

func NewConfig() *Config

func (*Config) GetClusterAddr 

func (c *Config) GetClusterAddr() string

GetClusterAddr returns the full cluster listen address

func (*Config) GetProxyAddr 

func (c *Config) GetProxyAddr() string

GetProxyAddr returns the full proxy listen address

func (*Config) Validate 

func (c *Config) Validate() error

type DiscoveryKind added in v0.3.0 

type DiscoveryKind uint8
const (
	DiscoveryKindStatic DiscoveryKind = iota
	DiscoveryKindHttp
)

func ParseDiscoveryKind added in v0.3.0 

func ParseDiscoveryKind(s string) (DiscoveryKind, error)

ParseDiscoveryKind converts a string to DiscoveryKind

func (DiscoveryKind) String added in v0.3.0 

func (d DiscoveryKind) String() string

String returns the string representation of DiscoveryKind

type GatewayMetrics 

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

GatewayMetrics - Only production-realistic failure modes

func GetGatewayMetrics 

func GetGatewayMetrics() *GatewayMetrics

func (*GatewayMetrics) ConnectionClosed 

func (m *GatewayMetrics) ConnectionClosed()

func (*GatewayMetrics) ConnectionFailed 

func (m *GatewayMetrics) ConnectionFailed()

func (*GatewayMetrics) ConnectionOpened 

func (m *GatewayMetrics) ConnectionOpened()

func (*GatewayMetrics) MessageReceived 

func (m *GatewayMetrics) MessageReceived()

func (*GatewayMetrics) MessageSent 

func (m *GatewayMetrics) MessageSent()

func (*GatewayMetrics) NetworkError 

func (m *GatewayMetrics) NetworkError()

func (*GatewayMetrics) ProxyConnected 

func (m *GatewayMetrics) ProxyConnected()

Methods

func (*GatewayMetrics) ProxyDisconnected 

func (m *GatewayMetrics) ProxyDisconnected()

type LoggingConfig 

type LoggingConfig struct {
	Level      string `mapstructure:"level"`
	Format     string `mapstructure:"format"`
	OutputPath string `mapstructure:"output_path"`
}

type PartitionConfig 

type PartitionConfig struct {
	ActiveQueueCapacity int   `mapstructure:"active_queue_capacity"`
	MaxRetries          int   `mapstructure:"max_retries"`
	MaxBackoffSec       int64 `mapstructure:"max_backoff_sec"`
	DispatchBatchSize   int   `mapstructure:"dispatch_batch_size"`
	DLQMaxBytes         int64 `mapstructure:"dlq_max_bytes"`
	DLQMaxAgeMs         int64 `mapstructure:"dlq_max_age_ms"`
	PartitionTickMs     int   `mapstructure:"-"`
	ProxyCleanupTickSec int   `mapstructure:"-"`
	HeartbeatTickMs     int   `mapstructure:"-"`
}

PartitionConfig

type PartitionMetrics 

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

============================================================================= Partition Metrics ============================================================================= PartitionMetrics - SINGLETON shared across all partitions

func GetPartitionMetrics 

func GetPartitionMetrics() *PartitionMetrics

func (*PartitionMetrics) JobAdded 

func (m *PartitionMetrics) JobAdded(topic string, n uint64)

Methods now take topic as parameter

func (*PartitionMetrics) JobCompleted 

func (m *PartitionMetrics) JobCompleted(topic string, n uint64)

func (*PartitionMetrics) JobDLQ 

func (m *PartitionMetrics) JobDLQ(topic string, n uint64)

func (*PartitionMetrics) JobRetried 

func (m *PartitionMetrics) JobRetried(topic string, n uint64)

func (*PartitionMetrics) RemoveTopic 

func (m *PartitionMetrics) RemoveTopic(topic string)

cleanup when topic is removed

func (*PartitionMetrics) SetActiveDepth 

func (m *PartitionMetrics) SetActiveDepth(topic string, v uint32)

type ProxyConfig 

type ProxyConfig struct {
	Addr     string `mapstructure:"addr"`
	Port     int    `mapstructure:"port"`
	CertPath string `mapstructure:"cert_path"`
	KeyPath  string `mapstructure:"key_path"`
	CAPath   string `mapstructure:"ca_path"`
}

ProxyConfig for proxy connections

type TopicManagerMetrics 

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

TopicManagerMetrics - tracks topic lifecycle only

func GetTopicManagerMetrics 

func GetTopicManagerMetrics() *TopicManagerMetrics

func (*TopicManagerMetrics) SetActiveTopics 

func (m *TopicManagerMetrics) SetActiveTopics(count int)

func (*TopicManagerMetrics) TopicCreated 

func (m *TopicManagerMetrics) TopicCreated()

Methods

func (*TopicManagerMetrics) TopicRemoved 

func (m *TopicManagerMetrics) TopicRemoved()

type WALConfig 

type WALConfig struct {
	CompactAfterBytes int64  `mapstructure:"compact_after_bytes"`
	SyncInterval      string `mapstructure:"sync_interval"`
}

Source Files

View all Source files

Directories

Path Synopsis
internal/gateway/server.go
 internal/gateway/server.go

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.