You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

317 lines
5.5 KiB

package cluster
import (
"io/fs"
"path"
"sort"
"strings"
"darvaza.org/core"
"git.jpi.io/amery/jpictl/pkg/rings"
)
const (
// ZoneRegionsFileName indicates the file containing
// region names as references
ZoneRegionsFileName = "regions"
// RegionClusterTokenFileName contains the kubernetes
// token of the cluster this region represents
RegionClusterTokenFileName = "k8s_token"
)
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,
} {
if err := fn(opts); err != nil {
return err
}
}
return nil
}
func (m *Cluster) scanDirectory(opts *ScanOptions) error {
// each directory is a zone
entries, err := fs.ReadDir(m.dir, ".")
if err != nil {
return err
}
for _, e := range entries {
if e.IsDir() {
ok, err := m.scanSubdirectory(opts, e.Name())
switch {
case err != nil:
return core.Wrap(err, e.Name())
case !ok:
m.warn(nil).
WithField("zone", e.Name()).
Print("empty")
}
}
}
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,
logger: m,
Name: name,
}
z.debug().
WithField("zone", z.Name).
Print("found")
if err := z.scan(); err != nil {
return nil, err
}
return z, nil
}
func (m *Cluster) scanMachines(opts *ScanOptions) error {
var err error
m.ForEachMachine(func(p *Machine) bool {
err = p.scan(opts)
return err != nil
})
m.ForEachMachine(func(p *Machine) bool {
err = p.scanWrapUp(opts)
return err != nil
})
return err
}
func (m *Cluster) scanZoneIDs(_ *ScanOptions) error {
var hasMissing bool
var lastZoneID rings.ZoneID
m.ForEachZone(func(z *Zone) bool {
switch {
case z.ID == 0:
hasMissing = true
case z.ID > lastZoneID:
lastZoneID = z.ID
}
return false
})
if hasMissing {
next := lastZoneID + 1
m.ForEachZone(func(z *Zone) bool {
if z.ID == 0 {
z.ID, next = next, next+1
}
return false
})
}
return nil
}
func (m *Cluster) scanSort(_ *ScanOptions) error {
sort.SliceStable(m.Zones, func(i, j int) bool {
id1 := m.Zones[i].ID
id2 := m.Zones[j].ID
return id1 < id2
})
m.ForEachZone(func(z *Zone) bool {
sort.Sort(z)
return false
})
m.ForEachMachine(func(p *Machine) bool {
sort.SliceStable(p.Rings, func(i, j int) bool {
ri1 := p.Rings[i]
ri2 := p.Rings[j]
return ri1.Ring < ri2.Ring
})
return false
})
return nil
}
func (m *Cluster) scanGateways(_ *ScanOptions) error {
var err error
m.ForEachZone(func(z *Zone) bool {
_, _, err = z.GetGateway()
return err != nil
})
return err
}
func (z *Zone) scan() error {
// each directory is a machine
entries, err := fs.ReadDir(z.zones.dir, z.Name)
if err != nil {
return err
}
for _, e := range entries {
name := e.Name()
switch {
case name == ZoneRegionsFileName:
err = z.loadRegions()
case name == RegionClusterTokenFileName:
err = z.loadClusterToken()
case e.IsDir():
err = z.scanSubdirectory(name)
default:
z.warn(nil).
WithField("zone", z.Name).
WithField("filename", name).
Print("unknown")
}
if err != nil {
return err
}
}
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) loadClusterToken() error {
var token string
filename := path.Join(z.Name, RegionClusterTokenFileName)
lines, err := z.zones.ReadLines(filename)
if err != nil {
return err
}
// first non-empty line
for _, s := range lines {
s = strings.TrimSpace(s)
if s != "" {
token = s
break
}
}
err = z.zones.setRegionClusterToken(z.Name, token)
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
func (z *Zone) GetGateway() (*Machine, bool, error) {
var first *Machine
var gateway *Machine
z.zones.ForEachMachine(func(p *Machine) bool {
switch {
case p.IsGateway():
// found
gateway = p
case first == nil:
// remember
first = p
default:
// keep looking
}
return gateway != nil
})
switch {
case gateway != nil:
// found one
return gateway, false, nil
case first != nil:
// make one
if err := first.SetGateway(true); err != nil {
return first, false, err
}
return first, true, nil
default:
// Zone without nodes?
panic("unreachable")
}
}