Compare commits

..

2 Commits

Author SHA1 Message Date
amery a725b66d14 cluster: mark Machine as Inactive if the "region" file contains "none"
Signed-off-by: Alejandro Mery <amery@jpi.io>
2023-10-30 18:52:38 +00:00
amery 851274954e cluster: introduce Machine.Inactive flag
if a Machine is Inactive, it won't be included on the DNS
aliases for the zone or it's regions.

Signed-off-by: Alejandro Mery <amery@jpi.io>
2023-10-30 18:52:34 +00:00
7 changed files with 33 additions and 240 deletions
+1 -1
View File
@@ -52,7 +52,7 @@ func populateDNSManager(mgr *dns.Manager, m *cluster.Cluster) error {
m.ForEachZone(func(z *cluster.Zone) bool {
z.ForEachMachine(func(p *cluster.Machine) bool {
err = mgr.AddHost(ctx, z.Name, p.ID, p.IsActive(), p.PublicAddresses...)
err = mgr.AddHost(ctx, z.Name, p.ID, p.Active(), p.PublicAddresses...)
return err != nil
})
+26 -87
View File
@@ -2,25 +2,17 @@ package cluster
import (
"io/fs"
"path"
"sort"
"darvaza.org/core"
)
const (
// ZoneRegionsFileName indicates the file containing
// region names as references
ZoneRegionsFileName = "regions"
)
func (m *Cluster) scan(opts *ScanOptions) error {
for _, fn := range []func(*ScanOptions) error{
m.scanDirectory,
m.scanMachines,
m.scanZoneIDs,
m.scanSort,
m.initRegions,
m.scanGateways,
m.scanCephMonitors,
} {
@@ -32,7 +24,7 @@ func (m *Cluster) scan(opts *ScanOptions) error {
return nil
}
func (m *Cluster) scanDirectory(opts *ScanOptions) error {
func (m *Cluster) scanDirectory(_ *ScanOptions) error {
// each directory is a zone
entries, err := fs.ReadDir(m.dir, ".")
if err != nil {
@@ -41,14 +33,16 @@ func (m *Cluster) scanDirectory(opts *ScanOptions) error {
for _, e := range entries {
if e.IsDir() {
ok, err := m.scanSubdirectory(opts, e.Name())
z, err := m.newZone(e.Name())
switch {
case err != nil:
return core.Wrap(err, e.Name())
case !ok:
m.warn(nil).
WithField("zone", e.Name()).
case z.Machines.Len() == 0:
z.warn(nil).
WithField("zone", z.Name).
Print("empty")
default:
m.Zones = append(m.Zones, z)
}
}
}
@@ -56,27 +50,6 @@ func (m *Cluster) scanDirectory(opts *ScanOptions) error {
return nil
}
func (m *Cluster) scanSubdirectory(_ *ScanOptions, name string) (bool, error) {
z, err := m.newZone(name)
switch {
case err != nil:
// somewhere went wrong scanning the subdirectory
return false, err
case z.Machines.Len() > 0:
// zones have machines and the regions they belong
m.Zones = append(m.Zones, z)
return true, nil
case len(z.Regions) > 0:
// regions have no machines but can include
// other regions
m.appendRegionRegions(name, z.Regions...)
return true, nil
default:
// empty
return false, nil
}
}
func (m *Cluster) newZone(name string) (*Zone, error) {
z := &Zone{
zones: m,
@@ -181,68 +154,34 @@ func (z *Zone) scan() error {
}
for _, e := range entries {
name := e.Name()
if e.IsDir() {
m := &Machine{
zone: z,
logger: z,
Name: e.Name(),
}
switch {
case name == ZoneRegionsFileName:
err = z.loadRegions()
case e.IsDir():
err = z.scanSubdirectory(name)
default:
z.warn(nil).
m.debug().
WithField("node", m.Name).
WithField("zone", z.Name).
WithField("filename", name).
Print("unknown")
}
Print("found")
if err != nil {
return err
if err := m.init(); err != nil {
m.error(err).
WithField("node", m.Name).
WithField("zone", z.Name).
Print()
return err
}
z.Machines = append(z.Machines, m)
}
}
return nil
}
func (z *Zone) loadRegions() error {
filename := path.Join(z.Name, ZoneRegionsFileName)
regions, err := z.zones.ReadLines(filename)
if err == nil {
// parsed
err = z.appendRegions(regions...)
if err != nil {
err = core.Wrap(err, filename)
}
}
return err
}
func (z *Zone) scanSubdirectory(name string) error {
m := &Machine{
zone: z,
logger: z,
Name: name,
}
m.debug().
WithField("node", m.Name).
WithField("zone", z.Name).
Print("found")
if err := m.init(); err != nil {
m.error(err).
WithField("node", m.Name).
WithField("zone", z.Name).
Print()
return err
}
z.Machines = append(z.Machines, m)
return nil
}
// GetGateway returns the first gateway found, if none
// files will be created to enable the first [Machine] to
// be one
+2 -2
View File
@@ -44,8 +44,8 @@ func (m *Machine) FullName() string {
return strings.Join(name, ".")
}
// IsActive indicates the machine is to be included in regions' DNS entries
func (m *Machine) IsActive() bool {
// Active indicates the machine is to be included in regions' DNS entries
func (m *Machine) Active() bool {
return !m.Inactive
}
+4 -101
View File
@@ -1,10 +1,5 @@
package cluster
import (
"bytes"
"path/filepath"
)
var (
_ MachineIterator = (*Region)(nil)
_ ZoneIterator = (*Region)(nil)
@@ -37,7 +32,7 @@ func (r *Region) ForEachMachine(fn func(*Machine) bool) {
var term bool
z.ForEachMachine(func(p *Machine) bool {
if p.IsActive() {
if p.Active() {
term = fn(p)
}
return term
@@ -73,7 +68,7 @@ func (m *Cluster) initRegions(_ *ScanOptions) error {
// bind first level regions and their zones
for name, zones := range regions {
m.setRegionZones(name, zones...)
m.syncRegions(name, zones...)
}
// and combine zones to produce larger regions
@@ -86,10 +81,8 @@ func (m *Cluster) initRegions(_ *ScanOptions) error {
return nil
}
func (m *Cluster) setRegionZones(name string, zones ...*Zone) {
for i := range m.Regions {
r := &m.Regions[i]
func (m *Cluster) syncRegions(name string, zones ...*Zone) {
for _, r := range m.Regions {
if r.Name == name {
// found
r.m = m
@@ -106,38 +99,6 @@ func (m *Cluster) setRegionZones(name string, zones ...*Zone) {
})
}
func (m *Cluster) appendRegionRegions(name string, subs ...string) {
for i := range m.Regions {
r := &m.Regions[i]
if name == r.Name {
// found
r.Regions = append(r.Regions, subs...)
return
}
}
// new
m.Regions = append(m.Regions, Region{
Name: name,
Regions: subs,
})
}
func (z *Zone) appendRegions(regions ...string) error {
for _, s := range regions {
// TODO: validate
z.debug().
WithField("zone", z.Name).
WithField("region", s).
Print("attached")
z.Regions = append(z.Regions, s)
}
return nil
}
func (m *Cluster) finishRegion(r *Region) {
if r.m != nil {
// ready
@@ -171,61 +132,3 @@ func (m *Cluster) getRegion(name string) (*Region, bool) {
return nil, false
}
// SyncRegions writes to the file system the regions this [Zone]
// belongs to.
func (z *Zone) SyncRegions() error {
err := z.syncZoneRegions()
if err == nil {
z.ForEachMachine(func(p *Machine) bool {
if p.IsActive() {
err = p.RemoveFile("region")
} else {
err = p.WriteStringFile("none\n", "region")
}
return err != nil
})
}
return err
}
func (z *Zone) syncZoneRegions() error {
name := filepath.Join(z.Name, "regions")
if len(z.Regions) > 0 {
var buf bytes.Buffer
for _, s := range z.Regions {
_, _ = buf.WriteString(s)
_, _ = buf.WriteRune('\n')
}
return z.zones.WriteStringFile(buf.String(), name)
}
return z.zones.RemoveFile(name)
}
// SyncRegions writes to the file system the regions covered
// by this meta-region
func (r *Region) SyncRegions() error {
name := filepath.Join(r.Name, "regions")
if len(r.Regions) > 0 {
var buf bytes.Buffer
for _, s := range r.Regions {
_, _ = buf.WriteString(s)
_, _ = buf.WriteRune('\n')
}
if err := r.m.MkdirAll(r.Name); err != nil {
return err
}
return r.m.WriteStringFile(buf.String(), name)
}
return r.m.RemoveFile(name)
}
-22
View File
@@ -6,7 +6,6 @@ func (m *Cluster) SyncAll() error {
m.SyncMkdirAll,
m.SyncAllWireguard,
m.SyncAllCeph,
m.SyncAllRegions,
m.WriteHosts,
} {
if err := fn(); err != nil {
@@ -59,24 +58,3 @@ func (m *Cluster) SyncAllCeph() error {
return m.WriteCephConfig(cfg)
}
// SyncAllRegions rewrites all region data
func (m *Cluster) SyncAllRegions() error {
var err error
m.ForEachZone(func(z *Zone) bool {
err := z.SyncRegions()
return err != nil
})
if err != nil {
return err
}
m.ForEachRegion(func(r *Region) bool {
err = r.SyncRegions()
return err != nil
})
return err
}
-5
View File
@@ -107,11 +107,6 @@ func (ep EndpointAddress) String() string {
}
}
// UnmarshalText loads an endpoint address from text data
func (ep *EndpointAddress) UnmarshalText(b []byte) error {
return ep.FromString(string(b))
}
// FromString sets the EndpointAddress from a given "[host]:port"
func (ep *EndpointAddress) FromString(s string) error {
host, port, err := core.SplitHostPort(s)
-22
View File
@@ -51,28 +51,6 @@ func (pub PublicKey) String() string {
}
}
// UnmarshalText loads the value from base64
func (key *PrivateKey) UnmarshalText(b []byte) error {
v, err := PrivateKeyFromBase64(string(b))
if err != nil {
return err
}
*key = v
return nil
}
// UnmarshalText loads the value from base64
func (pub *PublicKey) UnmarshalText(b []byte) error {
v, err := PublicKeyFromBase64(string(b))
if err != nil {
return err
}
*pub = v
return nil
}
// MarshalJSON encodes the key for JSON, omitting empty.
func (key PrivateKey) MarshalJSON() ([]byte, error) {
return encodeKeyJSON(key.String())