monstera

package module
v0.0.0-...-56ba7c4 Latest Latest
Warning

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

 Go to latest Published: Aug 11, 2026 License: MIT Imports: 26 Imported by: 22

README

Monstera

Tests Go Reference

Monstera leaf

Monstera is a framework for writing stateful applications in pure Go with all data in memory or on disk without worrying about scalability and availability. Monstera takes care of replication, sharding, snapshotting, and rebalancing.

Monstera is a half-technical and half-mental framework. The tech part of it has a pretty small surface. It leaves a lot up to you to implement. But sticking to the framework principles will ensure:

  • Applications are insanely fast and efficient
  • Cluster is horizontally scalable
  • Business logic is easily testable
  • Local development is enjoyable

Data and compute are brought together into a single process. You are free to use any in-memory data structure or embedded database. By eliminating network calls and limited query languages for communication with external databases, you are able to solve problems which are known to be hard in distributed systems with just a few lines of your favorite programming language as you would do it on a whiteboard.

Go to documentation to learn more.

Example

It might be easier to understand how it works from examples rather than from the documentation.

  • evrblk/grackle is a complete production-ready application built with Monstera.

Installing

Use go get to install the latest version of the library.

$ go get -u github.com/evrblk/monstera@latest

There are also few CLI tools (such as codegen), so it also should be added as a tool:

$ go get -tool github.com/evrblk/monstera/cmd/monstera@latest

CLI

Monstera comes with CLI toolkit:

$ go tool github.com/evrblk/monstera/cmd/monstera
  • Code generation:
    • monstera code generate Generates stubs, core interfaces and adapters from monstera.yaml file.
  • Working with cluster configs (offline: edit a config file on disk):
    • monstera config init Creates a new cluster config.
    • monstera config add-node Adds a node to the cluster config.
    • monstera config add-application Adds an application to the cluster config.
  • Operating a running cluster (online: over the admin plane):
    • monstera cluster bootstrap-node Provisions a single unprovisioned node with its id and the initial cluster config, transitioning it to READY.
    • monstera cluster bootstrap-nodes Provisions every node listed in a cluster config, dialing each at its advertised gRPC address.
    • monstera cluster add-node Adds a node to a running cluster (rolls the new config out to all nodes, then bootstraps the newcomer).
    • monstera cluster move-shard Moves a shard replica from one node to another (add new replica, bake until caught up, then remove the old one).
    • monstera cluster split-shard Splits an active shard into two children at a given key (seed children, flip, bake).
    • monstera cluster get-config Downloads the cluster config a node is currently running and prints it (or writes it to a file).

Status

Monstera is being actively developed. There will be no version tagging before v0.1, just development in master branch. After it is tagged v0.1 it will be more or less stable and follow semver.

License

Monstera is released under MIT License.

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrNoClusterConfig   = errors.New("monstera client has no cluster config yet")
	ErrAllReplicasFailed = errors.New("all replicas failed")
	// ErrPayloadTooLarge is returned by Read/Update (and their *Shard variants)
	// when the payload exceeds the client's configured limit. It is checked on the
	// client before any node is contacted, so an oversized request never reaches a
	// Monstera node.
	ErrPayloadTooLarge = errors.New("payload exceeds maximum size")
)
View Source 
var DefaultMonsteraNodeConfig = NodeConfig{
	MaxHops:          defaultMaxHops,
	MaxReadTimeout:   defaultMaxReadTimeout,
	MaxUpdateTimeout: defaultMaxUpdateTimeout,

	UseInMemoryRaftStore: false,

	MembershipReconcileInterval: defaultMembershipReconcileInterval,
	MetricsSampleInterval:       defaultMetricsSampleInterval,
	SnapshotSessionTimeout:      defaultSnapshotSessionTimeout,
}

Functions

func RegisterMetrics 

func RegisterMetrics(registerer prometheus.Registerer)

RegisterMetrics registers all Prometheus metrics emitted by the Monstera framework (node, replica and Raft layers) with the given registerer. Call it once at startup, e.g. monstera.RegisterMetrics(prometheus.DefaultRegisterer), before serving. It panics if a metric is already registered.

Note: per-core RPC metrics live in the generated adapters, not here.

Types

type ApplicationCore 

type ApplicationCore interface {
	// Read is used to read a value directly from the application core.
	// Reads can be performed concurrently with updates, other reads,
	// and snapshots. Read must return internal errors, but all application
	// errors should be returned as part of the ReadResponse.
	Read(req []byte) (*ReadResponse, error)

	// Update is used to update the application core state.
	// All updates are applied to the application core sequentially,
	// in the order they are committed to the Raft log. This method is called
	// by the Raft thread. Update must return internal errors, but all application
	// errors should be returned as part of the UpdateResponse.
	Update(req []byte) (*UpdateResponse, error)

	// Snapshot returns an ApplicationCoreSnapshot used to support Raft log
	// compaction, state restoration, and follower catch-up.
	//
	// Snapshot must return quickly. Expensive I/O belongs in
	// ApplicationCoreSnapshot.Write. Update and Snapshot are always called
	// from the same thread, but Update will be called concurrently with
	// ApplicationCoreSnapshot.Write.
	Snapshot() ApplicationCoreSnapshot

	// Restore replaces the application core state with the data from the
	// given snapshot streams that belongs to this core's shard bounds —
	// "replace with the union of these streams". Callers pass one stream
	// (Raft restore on start, follower snapshot install, split seeding) or
	// two (merge seeding — one per merging parent). Streams are disjoint
	// after bounds filtering: the caller guarantees the producing shards'
	// ranges do not overlap, so no logical row appears in more than one
	// stream and stream order is irrelevant. It is not called concurrently
	// with any other command.
	Restore(readers ...io.ReadCloser) error

	// Close cleans up resources used by the application core. Do not clean up
	// resources shared by multiple cores. Close is called after a shard split
	// or move, and for each core after the Monstera node shuts down.
	Close()
}

ApplicationCore is the interface that must be implemented by clients to be used with the Monstera framework.

type ApplicationCoreDescriptor 

type ApplicationCoreDescriptor struct {
	// CoreFactoryFunc is a function that creates a new application core. It is called when
	// Monstera node starts for every replica on this node, and also for every new replica that
	// is added to the node while it is running.
	CoreFactoryFunc func(shard *cluster.Shard, replica *cluster.Replica) ApplicationCore

	// CoreType declares the storage model of this application's cores (see
	// the CoreType constants). Everything storage-dependent — restore on
	// start, shard-split seeding mechanism — is derived from it. Required;
	// the zero value is rejected at node start.
	CoreType CoreType
}

ApplicationCoreDescriptor is used to register an application core with Monstera.

type ApplicationCoreDescriptors 

type ApplicationCoreDescriptors = map[string]ApplicationCoreDescriptor

ApplicationCoreDescriptors map is used to register application cores with Monstera. Key: the name of the application core, it should match Application.Implementation in ClusterConfig. Value: application core descriptor.

type ApplicationCoreSnapshot 

type ApplicationCoreSnapshot interface {
	// Write should dump all necessary state to the Writer.
	Write(w io.Writer) error

	// Release is invoked when we are finished with the snapshot.
	Release()
}

ApplicationCoreSnapshot is returned by ApplicationCore.Snapshot and is written to persistent storage by the Raft snapshotting machinery.

type Client 

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

Client is a Monstera cluster client that routes reads and updates to the correct shard replicas and keeps replica leadership state up to date via periodic health checks. It is ok for leadership state to be stale here, because Monstera nodes can forward requests to the current leader.

func NewMonsteraClient 

func NewMonsteraClient(provider ClusterConfigProvider, trans transport.DataPlane, config ClientConfig) *Client

NewMonsteraClient creates a Client fed by the given config provider. Call Start to subscribe to config changes and begin background leader-state health checks.

If the provider already has a config (e.g. a StaticClusterConfigProvider), it is adopted eagerly here so the client is usable without Start; a PollingClusterConfigProvider has no config until Start, so gateways must call Start.

Non-positive config fields are replaced by their defaults, so a hand-built ClientConfig never has to set every knob.

func (*Client) ListShards 

func (c *Client) ListShards(applicationName string) ([]*cluster.Shard, error)

ListShards returns the application's currently routable shards (active or splitting), sorted by lower bound. These are exactly the shards that serve the keyspace, so it is the set to fan a request out over (e.g. running GC on every shard); retired (inactive) and not-yet-serving (activating) shards are excluded. See Router.ListRoutableShards.

func (*Client) Read 

func (c *Client) Read(ctx context.Context, applicationName string, shardKey cluster.ShardKey, allowReadFromFollowers bool, payload []byte) ([]byte, error)

Read routes a read request to the shard responsible for shardKey.

func (*Client) ReadShard 

func (c *Client) ReadShard(ctx context.Context, applicationName string, shardId string, allowReadFromFollowers bool, payload []byte) ([]byte, error)

ReadShard sends a read request directly to the specified shard by ID, bypassing shard-key routing.

func (*Client) Start 

func (c *Client) Start(ctx context.Context) error

Start subscribes to the config provider (so topology changes flow into routing and the data plane), starts it, and launches the background goroutine that periodically polls nodes for replica states to identify shard leaders.

Start does not block on the initial config: with a PollingClusterConfigProvider the client comes up immediately and begins routing as soon as a config is adopted; requests made before then return an error. It returns an error only if the provider itself fails to start.

func (*Client) Stop 

func (c *Client) Stop()

Stop unsubscribes from the config provider, stops it and the background health-check goroutine, and closes the transport. It blocks until the refresh loop has fully exited so the transport is not closed out from under an in-flight ListReplicaStates call.

func (*Client) Update 

func (c *Client) Update(ctx context.Context, applicationName string, shardKey cluster.ShardKey, payload []byte) ([]byte, error)

Update routes a write request to the shard responsible for shardKey.

func (*Client) UpdateShard 

func (c *Client) UpdateShard(ctx context.Context, applicationName string, shardId string, payload []byte) ([]byte, error)

UpdateShard sends a write request directly to the specified shard by ID, bypassing shard-key routing.

type ClientConfig 

type ClientConfig struct {
	// MaxRetriesOnSingleReplica is the number of times to retry a request on the
	// same replica before moving on to the next one.
	MaxRetriesOnSingleReplica int
	// ListReplicaStatesTimeout is the per-node timeout for each replica-state
	// refresh RPC.
	ListReplicaStatesTimeout time.Duration
	// RefreshIntervalBase is the minimum wait between replica-state refresh
	// rounds.
	RefreshIntervalBase time.Duration
	// RefreshIntervalJitter is the upper bound of random jitter added to
	// RefreshIntervalBase to spread refresh load across clients. Non-positive
	// means default.
	RefreshIntervalJitter time.Duration
	// ReadRetryDelay is how long to wait before retrying a read on the same
	// replica.
	ReadRetryDelay time.Duration
	// UpdateRetryDelay is how long to wait before retrying an update on the same
	// replica.
	UpdateRetryDelay time.Duration

	// MaxReadPayloadBytes is the maximum size in bytes of a Read payload. A Read
	// (or ReadShard) with a larger payload fails locally with ErrPayloadTooLarge
	// before any node is contacted. Zero or negative means the default (1 MiB).
	MaxReadPayloadBytes int
	// MaxUpdatePayloadBytes is the maximum size in bytes of an Update payload. An
	// Update (or UpdateShard) with a larger payload fails locally with
	// ErrPayloadTooLarge before any node is contacted. Zero or negative means the
	// default (1 MiB).
	MaxUpdatePayloadBytes int
}

ClientConfig holds tunable parameters for Client behavior.

func DefaultClientConfig 

func DefaultClientConfig() ClientConfig

DefaultClientConfig returns a ClientConfig with sensible defaults.

type ClusterConfigProvider 

type ClusterConfigProvider interface {
	// Latest returns the most recently adopted config, or nil before the first one
	// is available. Cheap and safe for concurrent use.
	Latest() *cluster.Config

	// Watch registers fn. If a config is already available, fn is called with it
	// synchronously before Watch returns; thereafter fn is called on every strictly
	// newer version adopted (monotonic by Config.Version). fn must not block. The
	// returned func unregisters the callback.
	Watch(fn func(*cluster.Config)) (unwatch func())

	// Start begins background discovery/polling. It does not block on, or fail for,
	// unreachable nodes: a PollingClusterConfigProvider returns immediately and adopts a
	// config as soon as one becomes available (Latest stays nil until then, and
	// callers should treat that as "not ready yet" rather than an error). Static
	// providers no-op.
	Start(ctx context.Context) error

	// Stop halts background work started by Start.
	Stop()
}

ClusterConfigProvider supplies the current cluster config to a non-node process (a gateway, worker, or admin tool) and notifies it when a newer version is adopted. It is the single owner that a Client uses to keep both its routing table and its data plane in sync with the cluster topology.

type CoreType 

type CoreType int

CoreType declares the storage model of an application core. The framework derives all storage-dependent behavior from it: whether the latest Raft snapshot is restored into the core on start, and how children of a splitting shard are seeded. The zero value is invalid: the type must be declared explicitly. 

const (
	// CoreTypeInMemory: core state lives only in RAM; the Raft log and
	// snapshots are its durability. On start, the latest snapshot is restored
	// (via ApplicationCore.Restore) and the log tail is replayed. Split
	// seeding copies the parent snapshot as the child's base plus the routed
	// log tail; activation replays it through the bounds-filtered Restore.
	CoreTypeInMemory CoreType = iota + 1

	// CoreTypePersistedShared: core state is durable, keyed by shard-key
	// range only, in a store shared across cores — cores with overlapping
	// bounds alias the same physical rows. No restore on start. Split
	// seeding is nothing at all: the children's rows are the parent's live
	// rows, and the split is just the cutoff.
	CoreTypePersistedShared

	// CoreTypePersistedExclusive: core state is durable and every row lives
	// under a shard-unique prefix — no row is readable or writable by more
	// than one core (the physical store may still be shared per node). No
	// restore on start. Split seeding restores the parent snapshot into the
	// child core plus a live routed Update tail until the cutoff.
	CoreTypePersistedExclusive
)

func (CoreType) RestoreSnapshotOnStart 

func (t CoreType) RestoreSnapshotOnStart() bool

RestoreSnapshotOnStart reports whether cores of this type restore their state from the latest Raft snapshot on start (true only for in-memory cores, whose snapshots+log are the durable state).

func (CoreType) String 

func (t CoreType) String() string

type Event 

type Event struct {
	// Data is the marshaled form of the event data.
	Data []byte

	// Topic that the event is published on.
	Topic string
}

Event is an event that is emitted by the application core after an update is applied.

type FileNodeDiscovery 

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

FileNodeDiscovery reads addresses from a file on every call — one "host:port" per line; blank lines and lines starting with '#' are ignored. Reading on each call lets an operator edit the file to change the candidate set without a restart.

func NewFileNodeDiscovery 

func NewFileNodeDiscovery(path string) *FileNodeDiscovery

func (*FileNodeDiscovery) Endpoints 

func (d *FileNodeDiscovery) Endpoints(ctx context.Context) ([]string, error)

type Node 

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

Node is a single Monstera server process. It hosts the shard replicas assigned to it by the cluster config, exposes Read/Update entry points that route to the replica owning a shard (forwarding to the Raft leader when necessary), and carries Raft traffic between replicas of the same shard. A Node moves through the INITIAL -> READY -> STOPPED lifecycle (see NodeState).

func NewNode 

func NewNode(baseDir string, coreDescriptors ApplicationCoreDescriptors, nodeConfig NodeConfig, trans transport.DataPlane) (*Node, error)

NewNode creates a Node backed by baseDir. Identity and config are discovered from disk. config/node.json is the provisioning marker: when it is present the node loads its applied cluster config (config/cluster.json) and runs as that node. When it is absent the node comes up UNPROVISIONED and serves only Bootstrap, which assigns its identity and installs the initial config. It opens the shared Raft store (durable on disk, or in-memory when NodeConfig.UseInMemoryRaftStore is set). Call Start to load replicas and begin serving.

Non-positive nodeConfig fields are replaced by their defaults, so a hand-built NodeConfig never has to set every knob.

func (*Node) Bootstrap 

func (n *Node) Bootstrap(ctx context.Context, nodeId string, config *cluster.Config) error

Bootstrap provisions an UNPROVISIONED node: it assigns the node its id, installs the cluster config, creates the node's replicas, and transitions to READY. The id is assigned here (this is the only place a node's identity is set) and must name a node present in the config. Subsequent config changes go through UpdateClusterConfig.

Bootstrap is idempotent with respect to identity: calling it on a node already provisioned as the same nodeId is a no-op success (so it is safe for an operator to bootstrap manually and for a control action to retry). It does not change the applied config in that case — the node may already be at a newer version. A bootstrap for a different nodeId is rejected: identity is immutable once set.

func (*Node) GetClusterConfig 

func (n *Node) GetClusterConfig() *cluster.Config

GetClusterConfig returns an independent deep copy of the cluster config this node is currently running with, so callers may freely retain, serialize or mutate it without racing the node's wholesale config swaps. Returns nil if the node has no config yet (unprovisioned). Useful for inspecting the applied config version on each node.

func (*Node) LeadershipTransfer 

func (n *Node) LeadershipTransfer(replicaId string) error

LeadershipTransfer asks the replica with the given id to hand off Raft leadership to another replica in its group (used for graceful node drain).

func (*Node) ListSnapshots 

func (n *Node) ListSnapshots(replicaId string) ([]raft.SnapshotMetadata, error)

ListSnapshots returns the snapshots stored for the replica with the given id. It reads the replica's snapshot store from disk, so it is meant for on-demand admin/ops use rather than frequent polling.

func (*Node) NodeId 

func (n *Node) NodeId() string

NodeId returns this node's id in the cluster config.

func (*Node) NodeState 

func (n *Node) NodeState() NodeState

NodeState returns the node's current lifecycle state.

func (*Node) RaftMessage 

func (n *Node) RaftMessage(ctx context.Context, req *transport.RaftMessageRequest) (*transport.RaftMessageResponse, error)

RaftMessage delivers a raw Raft protocol message to the target replica hosted on this node. It is the receiving end of the Raft transport between replicas of the same shard.

func (*Node) Read 

func (n *Node) Read(ctx context.Context, req *transport.ReadRequest) (*transport.ReadResponse, error)

Read serves a read for the shard that owns req. When follower reads are allowed it is served from the local replica directly (possibly stale); otherwise it is served locally only if this replica is the Raft leader, and forwarded to the leader's node otherwise. It returns errLeaderUnknown once the forwarding hop budget (MaxHops) is exhausted.

func (*Node) ReplicaStates 

func (n *Node) ReplicaStates() []*transport.ReplicaState

ListReplicas returns a snapshot of the replicas currently hosted on this node. ReplicaStates returns the observed state of every replica hosted on this node: serving replicas with their live Raft state and stats, and dormant (seeding) replicas of activating shards with their seeding progress. It is the single producer behind ListReplicaStates on both transports.

func (*Node) SplitCutoff 

func (n *Node) SplitCutoff(ctx context.Context, shardId string) (uint64, error)

SplitCutoff proposes the shard-split CUTOFF command through this node's replica of the given shard, which must be the Raft leader (callers locate the leader via ListReplicaStates). It freezes the shard at the returned log index. Idempotent: an already-frozen shard returns its original cutoff index.

func (*Node) Start 

func (n *Node) Start()

Start loads the replicas assigned to this node from the cluster config, bootstraps their Raft groups where needed, and marks the node READY. It panics if loading or bootstrapping fails: the node cannot serve without its replicas.

func (*Node) Stop 

func (n *Node) Stop()

Stop shuts the node down: it stops serving, closes the transport and every hosted replica, and closes the shared Raft store. It is safe to call more than once.

func (*Node) TriggerSnapshot 

func (n *Node) TriggerSnapshot(replicaId string) error

TriggerSnapshot asks the replica with the given id to take a Raft snapshot.

func (*Node) Update 

func (n *Node) Update(ctx context.Context, req *transport.UpdateRequest) (*transport.UpdateResponse, error)

Update applies a write to the shard that owns req. Writes must go through the Raft leader: if this replica is the leader the write is applied (and replicated) locally, otherwise the request is forwarded to the leader's node. It returns errLeaderUnknown once the forwarding hop budget (MaxHops) is exhausted.

func (*Node) UpdateClusterConfig 

func (n *Node) UpdateClusterConfig(ctx context.Context, newConfig *cluster.Config) error

UpdateClusterConfig installs a new cluster config: it persists the config, swaps it in, and reconciles this node's replicas and Raft group membership to match. This is the only place replicas and clusterConfig change after startup; the persist + swap + replica reconcile happen under mu so readers always observe a config that matches the replica map. It also refreshes the transport's view of the cluster so it can dial added nodes and drop connections to removed ones.

type NodeConfig 

type NodeConfig struct {
	// MaxHops bounds how many times a read/update may be forwarded between nodes
	// while chasing the current leader before giving up with errLeaderUnknown.
	MaxHops int32

	// MaxReadTimeout bounds the total time a Node.Read may take, including leader
	// discovery and forwarding to the leader.
	MaxReadTimeout time.Duration

	// MaxUpdateTimeout bounds the total time a Node.Update may take. It is also
	// the timeout passed to Raft when applying a committed log entry.
	MaxUpdateTimeout time.Duration

	// UseInMemoryRaftStore set to `true` should be used only in unit tests or dev
	// environment and is not recommended for production use, since in-memory Raft
	// store is not durable.
	UseInMemoryRaftStore bool

	// MembershipReconcileInterval is how often a node re-checks, for each shard it
	// leads, that the Raft group membership matches the cluster config (adding or
	// removing voters as needed). The reconcile is idempotent and cheap when there
	// is nothing to do.
	MembershipReconcileInterval time.Duration

	// MetricsSampleInterval is how often the node samples per-replica gauge metrics
	// that reflect live Raft state (e.g. monstera_raft_replica_commit_lag).
	MetricsSampleInterval time.Duration

	// SnapshotSessionTimeout is the per-replica inactivity deadline for receiving
	// an InstallSnapshot: if no new chunk arrives within this window the follower
	// abandons the transfer and frees the main Raft goroutine blocked on it (see
	// the InstallSnapshot session handling in internal/raft). It is reset on every
	// chunk, so it never aborts a snapshot that is actively streaming.
	SnapshotSessionTimeout time.Duration
}

NodeConfig holds tunable parameters for Node behavior.

type NodeDiscovery 

type NodeDiscovery interface {
	// Endpoints returns candidate gRPC addresses ("host:port").
	Endpoints(ctx context.Context) ([]string, error)
}

NodeDiscovery answers "which node addresses should I ask for the cluster config?". It is deliberately separate from ConfigProvider — who to ask versus what they say — so the bootstrap source is pluggable per deployment (a static list, a file, a DNS SRV record, ...).

Discovery only needs to be good enough to reach one live node: once a PollingClusterConfigProvider has a config, the authoritative node set comes from that config, and discovery is just the durable "who do I ask" seed and ongoing membership hint. Endpoints is called once per poll round, so implementations may re-resolve (DNS, file) on each call.

type NodeState 

type NodeState int

NodeState is the lifecycle state of a Node. 

const (
	// NodeStateInitial is the state before Start finishes; the node does not serve yet.
	NodeStateInitial NodeState = iota
	// NodeStateUnprovisioned means the node started without a cluster config: it
	// serves only Bootstrap (and read-only status calls), awaiting provisioning.
	NodeStateUnprovisioned
	// NodeStateReady means replicas are loaded and the node serves reads and updates.
	NodeStateReady
	// NodeStateStopped means the node has been shut down and rejects further requests.
	NodeStateStopped
)

func (NodeState) String 

func (s NodeState) String() string

String returns the lowercase name of the state.

type PollingClusterConfigProvider 

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

PollingClusterConfigProvider learns the config from the cluster itself: each round it asks a set of candidate nodes (from a NodeDiscovery, unioned with the nodes in the config it already holds) for their config over the AdminPlane and adopts the highest Version seen. Config versions are monotonic and transitions are validated cluster-side, so "highest version wins" is correct even mid-rollout when nodes briefly disagree. This is what makes a gateway self-healing — the cluster is the single source of truth, with no config files to keep in sync.

func NewPollingClusterConfigProvider 

func NewPollingClusterConfigProvider(discovery NodeDiscovery, admin transport.AdminPlane, opts PollingOptions) *PollingClusterConfigProvider

func (*PollingClusterConfigProvider) Latest 

func (p *PollingClusterConfigProvider) Latest() *cluster.Config

func (*PollingClusterConfigProvider) Start 

func (p *PollingClusterConfigProvider) Start(ctx context.Context) error

func (*PollingClusterConfigProvider) Stop 

func (p *PollingClusterConfigProvider) Stop()

func (*PollingClusterConfigProvider) Watch 

func (p *PollingClusterConfigProvider) Watch(fn func(*cluster.Config)) func()

type PollingOptions 

type PollingOptions struct {
	// Interval is the wait between poll rounds. Defaults to 5s.
	Interval time.Duration
	// Timeout bounds each per-endpoint GetClusterConfig call. Defaults to 1s.
	Timeout time.Duration
}

PollingOptions tunes a PollingClusterConfigProvider.

type ReadResponse 

type ReadResponse struct {
	// Data is the marshaled form of the response that the application core
	// returns as the result of a read.
	Data []byte
}

ReadResponse is the response returned by ApplicationCore.Read.

type Router 

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

Router is an immutable, index-backed view of a cluster.Config used for the request serving path: it resolves a shard key to its owning shard and looks up shards, replicas and nodes by id in O(1)/O(log n) instead of scanning the config.

A Router is built once from a config (see NewRouter) and never mutated, so — unlike the cluster.Config it is built from — it is safe for concurrent reads. Callers that swap in a new config (a node applying a cluster config update, a client adopting a polled config) build a fresh Router alongside it and publish the pair together.

func NewRouter 

func NewRouter(cfg *cluster.Config) *Router

NewRouter builds a Router from cfg. cfg is only read, never retained beyond construction (the Router aliases cfg's shard/replica/node pointers, which are treated as read-only, but does not depend on cfg's slice ordering afterwards).

A nil cfg yields an empty Router whose lookups all report "not found"; this lets a not-yet-provisioned node or a client without a config hold a non-nil Router without special-casing every call site.

func (*Router) FindShardByShardKey 

func (r *Router) FindShardByShardKey(applicationName string, shardKey cluster.ShardKey) (*cluster.Shard, error)

FindShardByShardKey returns the routable (active or splitting) shard whose [LowerKey, UpperKey] range contains shardKey. Inactive and activating shards may overlap that range and are never returned. Every ShardKey value is valid; on a validated config (routable shards cover the whole keyspace) the lookup only fails when the application is unknown.

func (*Router) GetNode 

func (r *Router) GetNode(nodeId string) (*cluster.Node, error)

GetNode returns the node with the given id, or errRouteNodeNotFound.

func (*Router) GetReplica 

func (r *Router) GetReplica(replicaId string) (*cluster.Replica, error)

GetReplica returns the replica with the given id, or errRouteReplicaNotFound.

func (*Router) GetShard 

func (r *Router) GetShard(shardId string) (*cluster.Shard, error)

GetShard returns the shard with the given id, or errRouteShardNotFound. It finds shards in any state (routing state is irrelevant for a by-id lookup).

func (*Router) ListRoutableShards 

func (r *Router) ListRoutableShards(applicationName string) ([]*cluster.Shard, error)

ListRoutableShards returns the application's routable shards (active or splitting), sorted by lower bound. These are exactly the shards that currently serve the keyspace, which is what fanout operations (e.g. running GC on every shard) must target: inactive shards are retired and serve nothing, and activating shards are not serving yet (their range is still covered by the splitting parent). The returned slice is a copy the caller may retain; the shard pointers in it alias the config and are read-only.

type SRVNodeDiscovery 

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

SRVNodeDiscovery resolves a DNS SRV record to node addresses on every call — the natural fit for Kubernetes headless services or Consul, where membership is already published in DNS. The name is looked up directly (service and proto are empty), so pass the full SRV record name.

func NewSRVNodeDiscovery 

func NewSRVNodeDiscovery(name string, resolver *net.Resolver) *SRVNodeDiscovery

func (*SRVNodeDiscovery) Endpoints 

func (d *SRVNodeDiscovery) Endpoints(ctx context.Context) ([]string, error)

type StaticClusterConfigProvider 

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

StaticClusterConfigProvider always yields the same config. This is the pre-provider behavior: use it when the config is fixed for the process lifetime (tests, or a gateway that is restarted on config change).

func NewStaticClusterConfigProvider 

func NewStaticClusterConfigProvider(cfg *cluster.Config) *StaticClusterConfigProvider

func (*StaticClusterConfigProvider) Latest 

func (p *StaticClusterConfigProvider) Latest() *cluster.Config

func (*StaticClusterConfigProvider) Start 

func (p *StaticClusterConfigProvider) Start(ctx context.Context) error

func (*StaticClusterConfigProvider) Stop 

func (p *StaticClusterConfigProvider) Stop()

func (*StaticClusterConfigProvider) Watch 

func (p *StaticClusterConfigProvider) Watch(fn func(*cluster.Config)) func()

type StaticNodeDiscovery 

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

StaticNodeDiscovery returns a fixed list of addresses (e.g. from a flag).

func NewStaticNodeDiscovery 

func NewStaticNodeDiscovery(addrs []string) *StaticNodeDiscovery

func (*StaticNodeDiscovery) Endpoints 

func (d *StaticNodeDiscovery) Endpoints(ctx context.Context) ([]string, error)

type UpdateResponse 

type UpdateResponse struct {
	// Data is the marshaled form of the response that the application core
	// produces as the result of an update.
	Data []byte

	// Events are emitted by the application core after an update is applied.
	// Can be zero or more events. The events are related to the update and are
	// used to notify subscribers of the changes that happened as a result of
	// the update.
	Events []Event
}

UpdateResponse is the response returned by ApplicationCore.Update.

Source Files

View all Source files

Directories

Path Synopsis
Package cluster models a monstera cluster's topology (nodes, applications, shards and replicas) as a Config, and provides loading, validation and mutation helpers for it.
 Package cluster models a monstera cluster's topology (nodes, applications, shards and replicas) as a Config, and provides loading, validation and mutation helpers for it.
cmd
monstera command
monstera/commands
Package control drives deterministic, resumable cluster reconfiguration sequences (add node, move shard, ...) over the transport admin plane.
 Package control drives deterministic, resumable cluster reconfiguration sequences (add node, move shard, ...) over the transport admin plane.
internal
integration_test/codegen
integration_test/codegen/types
integration_test/testcore
Package testcore provides application cores and client stubs shared by the Monstera integration tests.
 Package testcore provides application cores and client stubs shared by the Monstera integration tests.
integration_test/testutils
Package testutils holds the shared harness for integration tests: network helpers, gRPC and local-transport cluster startup, cluster config builders, and replica-state assertions.
 Package testutils holds the shared harness for integration tests: network helpers, gRPC and local-transport cluster startup, cluster config builders, and replica-state assertions.
raft
raft/raftpb
replication
replication/replicationpb
rpc
codegen
grpc
grpc/monsterapb
local
Package utils holds small helpers for building the keys that address data in a Monstera cluster: deriving shard keys from entity ids, concatenating typed values into a byte key, and fixed-width big-endian integer conversions.
 Package utils holds small helpers for building the keys that address data in a Monstera cluster: deriving shard keys from entity ids, concatenating typed values into a byte key, and fixed-width big-endian integer conversions.

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.