jpictl/pkg/cluster/regions.go

package cluster

import (
	"bytes"
	"path/filepath"

	"git.jpi.io/amery/jpictl/pkg/rings"
)

var (
	_ MachineIterator = (*Region)(nil)
	_ ZoneIterator    = (*Region)(nil)

	_ RegionIterator = (*Zone)(nil)
	_ RegionIterator = (*Cluster)(nil)
)

// A RegionIterator is a set of Regions we can iterate on
type RegionIterator interface {
	ForEachRegion(func(*Region) bool)
}

// Region represents a group of zones geographically related
type Region struct {
	m     *Cluster
	zones []*Zone

	Name    string
	ID      rings.RegionID `json:",omitempty" yaml:",omitempty"`
	Cluster *string        `json:",omitempty" yaml:",omitempty"`
	Regions []string       `json:",omitempty" yaml:",omitempty"`
}

// IsPrimary indicates the region is primary and corresponds
// to a kubernetes cluster.
func (r *Region) IsPrimary() bool {
	return r != nil && r.Cluster != nil
}

// ForEachRegion calls a function for each Region of the cluster
// until instructed to terminate the loop
func (m *Cluster) ForEachRegion(fn func(r *Region) bool) {
	for i := range m.Regions {
		r := &m.Regions[i]
		if fn(r) {
			return
		}
	}
}

// ForEachMachine calls a function for each Machine in the region
// until instructed to terminate the loop
func (r *Region) ForEachMachine(fn func(*Machine) bool) {
	r.ForEachZone(func(z *Zone) bool {
		var term bool

		z.ForEachMachine(func(p *Machine) bool {
			if p.IsActive() {
				term = fn(p)
			}
			return term
		})

		return term
	})
}

// ForEachZone calls a function for each Zone in the region
// until instructed to terminate the loop
func (r *Region) ForEachZone(fn func(*Zone) bool) {
	for _, p := range r.zones {
		if fn(p) {
			// terminate
			return
		}
	}
}

func (m *Cluster) initRegions(_ *ScanOptions) error {
	regions := make(map[string][]*Zone)

	// first regions defined by zones
	m.ForEachZone(func(z *Zone) bool {
		SortRegions(z.Regions)
		for _, region := range z.Regions {
			regions[region] = append(regions[region], z)
		}

		return false
	})

	// bind first level regions and their zones
	for name, zones := range regions {
		m.setRegionZones(name, zones...)
	}

	// and combine zones to produce larger regions
	for i := range m.Regions {
		r := &m.Regions[i]
		m.finishRegion(r)
	}

	m.sortRegions()
	m.scanRegionID()
	m.computeZonesRegion()
	return nil
}

func (m *Cluster) setRegionZones(name string, zones ...*Zone) {
	for i := range m.Regions {
		r := &m.Regions[i]

		if r.Name == name {
			// found
			r.m = m
			r.zones = zones
			return
		}
	}

	// new
	m.Regions = append(m.Regions, Region{
		m:     m,
		zones: zones,
		Name:  name,
	})
}

func (m *Cluster) setRegionClusterToken(name string, token string) error {
	for i := range m.Regions {
		r := &m.Regions[i]

		if r.Name == name {
			// found
			r.Cluster = &token
			return nil
		}
	}

	// new
	m.Regions = append(m.Regions, Region{
		m:       m,
		Name:    name,
		Cluster: &token,
	})
	return nil
}

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,
	})
}

// ForEachRegion calls a function on all regions this zone belongs to.
func (z *Zone) ForEachRegion(fn func(*Region) bool) {
	if fn == nil {
		return
	}

	z.zones.ForEachRegion(func(r *Region) bool {
		var match bool

		r.ForEachZone(func(z2 *Zone) bool {
			match = (z == z2)
			return match
		})

		if match && fn(r) {
			return true
		}

		return false
	})
}

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
		return
	}

	r.m = m
	sub := []string{}
	for _, name := range r.Regions {
		r2, ok := m.getFinishRegion(name)
		if !ok {
			m.warn(nil).WithField("region", name).Print("unknown region")
			continue
		}

		sub = append(sub, r2.Name)
		r.zones = append(r.zones, r2.zones...)
	}
	r.Regions = sub
}

// revive:disable:cognitive-complexity
func (m *Cluster) scanRegionID() {
	// revive:enable:cognitive-complexity
	var max rings.RegionID
	var missing bool

	// check IDs
	ids := make(map[rings.RegionID]bool)
	fn := func(r *Region) bool {
		var term bool

		switch {
		case !r.IsPrimary():
			// secondary, no ID.
			r.ID = 0
		case !r.ID.Valid():
			// primary without ID
			missing = true
		case ids[r.ID]:
			// duplicate
			m.error(nil).WithField("region", r.Name).Print("duplicate ID")
			missing = true
			r.ID = 0
		default:
			ids[r.ID] = true

			if r.ID > max {
				max = r.ID
			}
		}

		return term
	}
	m.ForEachRegion(fn)

	if missing {
		// assign missing IDs
		fn := func(r *Region) bool {
			var term bool

			switch {
			case !r.IsPrimary():
				// ignore secondary
			case r.ID.Valid():
				// already has an ID
			default:
				r.ID = max + 1
				max = r.ID
			}

			return term
		}

		m.ForEachRegion(fn)
	}
}

func (m *Cluster) computeZonesRegion() {
	fn := func(r *Region, z *Zone) {
		if z.region != nil {
			m.error(nil).
				WithField("zone", z.Name).
				WithField("region", []string{
					z.region.Name,
					r.Name,
				}).
				Print("zone in two regions")
		} else {
			z.region = r
		}
	}

	m.ForEachRegion(func(r *Region) bool {
		var term bool

		if r.IsPrimary() {
			r.ForEachZone(func(z *Zone) bool {
				fn(r, z)
				return term
			})
		}

		return term
	})
}

func (m *Cluster) getRegion(name string) (*Region, bool) {
	for i := range m.Regions {
		r := &m.Regions[i]
		if name == r.Name {
			return r, true
		}
	}

	return nil, false
}

func (m *Cluster) getFinishRegion(name string) (*Region, bool) {
	if r, ok := m.getRegion(name); ok {
		m.finishRegion(r)
		return r, true
	}

	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 {
			err = z.syncMachineRegions(p)
			return err != nil
		})
	}

	return err
}

func (*Zone) syncMachineRegions(p *Machine) error {
	var err error

	if p.IsActive() {
		err = p.RemoveFile("region")
	} else {
		err = p.WriteStringFile("none\n", "region")
	}

	if err == nil {
		err = p.RemoveFile(RegionClusterTokenFileName)
	}

	return err
}

func (z *Zone) syncZoneRegions() error {
	name := filepath.Join(z.Name, ZoneRegionsFileName)

	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 {
	if err := r.syncRegionsFile(); err != nil {
		return err
	}

	return r.syncClusterFile()
}

func (r *Region) mkdir() error {
	return r.m.MkdirAll(r.Name)
}

func (r *Region) syncRegionsFile() error {
	var err error

	name := filepath.Join(r.Name, ZoneRegionsFileName)

	if len(r.Regions) == 0 {
		err = r.m.RemoveFile(name)
	} else if err = r.mkdir(); err == nil {
		var buf bytes.Buffer

		for _, s := range r.Regions {
			_, _ = buf.WriteString(s)
			_, _ = buf.WriteRune('\n')
		}

		err = r.m.WriteStringFile(buf.String(), name)
	}

	return err
}

func (r *Region) syncClusterFile() error {
	var err error

	name := filepath.Join(r.Name, RegionClusterTokenFileName)

	if r.Cluster == nil {
		err = r.m.RemoveFile(name)
	} else if err = r.mkdir(); err == nil {
		var buf bytes.Buffer

		_, _ = buf.WriteString(*r.Cluster)
		if buf.Len() > 0 {
			_, _ = buf.WriteRune('\n')
		}

		err = r.m.WriteStringFile(buf.String(), name)
	}

	return err
}