package predastore import ( "cmp" "slices" "time" "github.com/mulgadc/predastore/internal/config" "github.com/mulgadc/predastore/internal/gate" "github.com/mulgadc/predastore/internal/gate/auth" "github.com/mulgadc/predastore/internal/gate/handlers" ) // The TOML file is this package's contract with operators, so every type it // names is re-exported here. They are aliases rather than copies: the shape an // operator writes is the shape the cluster is derived from, and a mirror would // only be something to keep in sync. The host and node tables are named for // the file because Host is already this package's running process. // NodeID and HostID identify the two levels of the cluster. They are distinct // types so that a host id cannot be passed where a node id belongs. type ( NodeID = config.NodeID HostID = config.HostID ) // Role is the function a node performs in the cluster, as written under // [[host.node]]. type Role = config.Role const ( // RoleGate serves the S3 API in front of the cluster. RoleGate = config.RoleGate // RoleBlob stores erasure-coded object shards. RoleBlob = config.RoleBlob // RoleMeta participates in Raft consensus over global state. RoleMeta = config.RoleMeta ) // HostConfig is one machine, as written under [[host]]: a data directory, an // address, a TLS identity and the nodes it runs in one process. type HostConfig = config.Host // NodeConfig is a logical role pinned to a host, as written under // [[host.node]]. type NodeConfig = config.Node // The remaining tables of the configuration file. type ( RS = config.RS Repair = config.Repair Reaper = config.Reaper Compaction = config.Compaction S3 = config.S3 Bucket = config.Bucket AuthEntry = config.AuthEntry PolicyRule = config.PolicyRule IAMConfig = config.IAMConfig ) // Config is predastore's on-disk configuration. type Config = config.Config // LoadConfig reads and validates a TOML configuration file. func LoadConfig(path string) (*Config, error) { return config.Load(path) } // HostBindAddr is where a host's cluster traffic listens, and NodeBindAddr // where a gate serves S3. Both are re-exported because an embedder settling // host-local fields has to know which plane it is settling: the S3 address // belongs to the gate, and putting it on the host would move raft and blob // traffic onto the public interface with it. func HostBindAddr(h HostConfig) string { return config.HostBindAddr(h) } // NodeBindAddr resolves a gate's S3 listen address against its host. func NodeBindAddr(h HostConfig, n NodeConfig) string { return config.NodeBindAddr(h, n) } // The queries below answer what the cluster is made of. They live here rather // than in the config package because they are inventory questions about a // parsed file, not part of parsing it. // // nodesByRole sorts by id. Raft bootstraps from the meta set and treats an // identically ordered set as idempotent across replicas, so the order is part // of the contract rather than a tidiness. // hostOf is the [[host]] with this id, and whether the file names one. func hostOf(c *Config, hostID HostID) (HostConfig, bool) { for _, h := range c.Hosts { if h.ID == hostID { return h, true } } return HostConfig{}, false } // nodesByRole is every node with the role, wherever it runs. func nodesByRole(c *Config, role Role) []NodeConfig { var out []NodeConfig for _, h := range c.Hosts { for _, n := range h.Nodes { if n.Role == role { out = append(out, n) } } } slices.SortFunc(out, func(a, b NodeConfig) int { return cmp.Compare(a.ID, b.ID) }) return out } // hasRemoteNodes reports whether any node runs off hostID, which is the only // reason a node on it opens a network socket. func hasRemoteNodes(c *Config, hostID HostID) bool { return slices.ContainsFunc(c.Hosts, func(h HostConfig) bool { return h.ID != hostID && len(h.Nodes) > 0 }) } // hostNodesByRole is every node with the role that runs on this host. func hostNodesByRole(h HostConfig, role Role) []NodeConfig { var out []NodeConfig for _, n := range h.Nodes { if n.Role == role { out = append(out, n) } } return out } // nodeIDs names a set of nodes by id, which is how every client addresses one. func nodeIDs(nodes []NodeConfig) []NodeID { ids := make([]NodeID, 0, len(nodes)) for _, n := range nodes { ids = append(ids, n.ID) } return ids } // The conversions below turn the file into each subsystem's own settings. They // live here rather than in the config package because gate.Config names a // NodeID and so imports config; converting there would close the cycle. func bucketConfigs(c *Config) []handlers.BucketConfig { buckets := make([]handlers.BucketConfig, len(c.Buckets)) for i, b := range c.Buckets { buckets[i] = handlers.BucketConfig{ Name: b.Name, Region: b.Region, Public: b.Public, AccountID: b.AccountID, } } return buckets } func authEntries(c *Config) []auth.Entry { entries := make([]auth.Entry, len(c.Auth)) for i, a := range c.Auth { rules := make([]auth.PolicyRule, len(a.Policy)) for j, p := range a.Policy { rules[j] = auth.PolicyRule{Bucket: p.Bucket, Actions: p.Actions} } entries[i] = auth.Entry{ AccessKeyID: a.AccessKeyID, SecretAccessKey: a.SecretAccessKey, AccountID: a.AccountID, Policy: rules, } } return entries } func iamConfig(c *Config) *auth.IAMConfig { if c.IAM == nil { return nil } return &auth.IAMConfig{ NATSUrl: c.IAM.NATSUrl, NATSToken: c.IAM.NATSToken, MasterKeyPath: c.IAM.MasterKeyPath, AccessKeysBucket: c.IAM.AccessKeysBucket, } } // gateConfig is everything the S3 frontend runs on: the slice of the file it // reads, plus the wiring its host supplies — where it listens, the TLS // identity it serves under, and the cluster clients it works through. func gateConfig( c *Config, host HostConfig, n NodeConfig, mc gate.MetaClient, bc gate.BlobClient, ) gate.Config { return gate.Config{ Region: c.Region, NodeID: n.ID, RS: gate.RS{ Data: c.RS.Data, Parity: c.RS.Parity, DegradedWrites: c.RS.DegradedWritesEnabled(), HintedHandoff: c.RS.HintedHandoffEnabled(), }, Buckets: bucketConfigs(c), Auth: authEntries(c), IAM: iamConfig(c), RateLimit: c.RateLimit, BlobNodeIDs: nodeIDs(nodesByRole(c, RoleBlob)), // Repair is scoped to the blob nodes sharing this process, so each is // swept by exactly one coordinator without anything having to elect one. LocalBlobNodeIDs: nodeIDs(hostNodesByRole(host, RoleBlob)), MetaNodeIDs: nodeIDs(nodesByRole(c, RoleMeta)), LocalMetaNodeIDs: nodeIDs(hostNodesByRole(host, RoleMeta)), Repair: gate.RepairConfig{ Enabled: c.Repair.IsEnabled(), Workers: c.Repair.Workers, PageSize: c.Repair.PageSize, Interval: time.Duration(c.Repair.IntervalSeconds) * time.Second, }, // Reaper has no ownership scoping: every gate that runs repair also // runs it, and a tombstone racing two gates costs redundant RPCs and // nothing else. Reaper: gate.ReaperConfig{ Enabled: c.Reaper.IsEnabled(), Workers: c.Reaper.Workers, PageSize: c.Reaper.PageSize, Interval: time.Duration(c.Reaper.IntervalSeconds) * time.Second, }, Addr: config.NodeBindAddr(host, n), Port: n.Port, TLSCert: host.TLSCert, TLSKey: host.TLSKey, EnableHTTP2: c.S3.HTTP2Enabled(), Meta: mc, Blob: bc, } }