195 lines
7.1 KiB
Go
195 lines
7.1 KiB
Go
package main
|
|
|
|
import (
|
|
"fmt"
|
|
"log"
|
|
"sort"
|
|
"strings"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
"git.novox.be/novox/mesh-sdk/go/announce"
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"git.novox.be/novox/mesh-sdk/go/membership"
|
|
"git.novox.be/novox/mesh-sdk/go/process"
|
|
"git.novox.be/novox/mesh-sdk/go/provisions"
|
|
|
|
"git.novox.be/novox/mesh-controller/modules/route-proxy/contract"
|
|
)
|
|
|
|
// keptStates is every state route-proxy keeps, as both its halves write them: the routes and the refusals this
|
|
// proxy writes, readable by others, and the bound entries the provider helper writes.
|
|
func keptStates() []announce.State {
|
|
return []announce.State{
|
|
{Name: contract.BoundState},
|
|
{Name: contract.RoutesState, Readable: true},
|
|
{Name: contract.UnroutedState, Readable: true},
|
|
}
|
|
}
|
|
|
|
// What the mesh says this proxy routes, and who is inside the mesh, read on the bus.
|
|
//
|
|
// **The routes are the seat mesh-provisioner's plans, alone** (issue 774, research 072 row 34): every plan's entry
|
|
// naming this proxy as a route's provider, followed, so a route added or withdrawn reaches a running proxy without a
|
|
// restart, and said on the bus as it changes (contract.EventServed, EventWithdrawn, EventRefused). **Who is inside**
|
|
// is every node's address on the private network as the seat mesh-placement decides it (mesh-sdk
|
|
// delivery.InsideAddresses), which is who may be served an internal name; until it is read, an internal name is
|
|
// served to this machine alone.
|
|
|
|
// follow connects with the credential in path, follows the plans and the placement, and retries the connection for
|
|
// as long as it takes: until the bus answers, the proxy routes nothing, and says so.
|
|
func follow(path string, held *table) {
|
|
for {
|
|
err := followOnce(path, held)
|
|
if err == nil {
|
|
return
|
|
}
|
|
log.Printf("cannot follow the plans that say what this proxy routes, so it routes what it last read: %v", err)
|
|
time.Sleep(30 * time.Second)
|
|
}
|
|
}
|
|
|
|
func followOnce(path string, held *table) error {
|
|
account, err := process.ReadAccount(path)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
conn, err := process.Connect(account, nats.Name(account.Node+"."+account.Module))
|
|
if err != nil {
|
|
return fmt.Errorf("connecting to the bus at %s: %w", membership.BareAddress(account.URL), err)
|
|
}
|
|
// What this proxy keeps, said on the bus (novox/hq ADR 0495; issue 774): the names it serves and the names it
|
|
// refused, readable by any module granted a read of them; the provider helper's entries, read by the machines'
|
|
// clients through their own grant. Said before the first write, so no write of an unannounced state is refused.
|
|
if err := conn.Keeps(keptStates()...); err != nil {
|
|
log.Printf("cannot say on the bus what this proxy keeps, so it keeps nothing there: %v", err)
|
|
} else {
|
|
held.ledger.attach(account.Node, conn.State(contract.RoutesState), conn.State(contract.UnroutedState))
|
|
}
|
|
say := func(eventType string, e contract.Event) {
|
|
if err := conn.Emit(eventType, e); err != nil {
|
|
log.Printf("cannot say %s of %q on the bus: %v", eventType, e.Name, err)
|
|
}
|
|
}
|
|
// Who is inside first, so the first table is served to the machines it is meant for.
|
|
if err := followInside(conn.Read(delivery.PlacementRead), held); err != nil {
|
|
log.Printf("cannot follow %s, so an internal name is served to this machine alone: %v", delivery.PlacementRead, err)
|
|
}
|
|
if err := followPlans(conn.Read(provisions.PlansRead), newPlanned(account.Node, say), held, say); err != nil {
|
|
conn.Close()
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// planWatch is a state this proxy watches: every current key, then every change.
|
|
type planWatch interface {
|
|
Watch(key string, handle func(stdio.StateChange) error) error
|
|
}
|
|
|
|
// followPlans serves what the plans give this proxy: the whole table once every current plan is read, then again on
|
|
// every change that alters it. A table built from part of the plans would withdraw every route not yet read.
|
|
func followPlans(plans planWatch, given *planned, held *table, say func(string, contract.Event)) error {
|
|
var read atomic.Bool // every current plan is read: from here, each change is served as it comes
|
|
err := plans.Watch("", func(c stdio.StateChange) error {
|
|
var value []byte
|
|
if c.Op != "delete" && c.Op != "purge" {
|
|
value = c.Value
|
|
}
|
|
if changed := given.take(c.Key, value); changed && read.Load() {
|
|
serve(given, held, say)
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("watching %s: %w", provisions.PlansRead, err)
|
|
}
|
|
read.Store(true)
|
|
serve(given, held, say)
|
|
return nil
|
|
}
|
|
|
|
// serve replaces what the proxy serves with what the plans give it, and says each name it now serves or serves no
|
|
// more.
|
|
func serve(given *planned, held *table, say func(string, contract.Event)) {
|
|
before := held.served()
|
|
routes, public := routesOf(given.contributions())
|
|
held.set(routes, public)
|
|
after := held.served()
|
|
for _, e := range servedChanges(given.node, before, after) {
|
|
say(e.kind, e.Event)
|
|
}
|
|
log.Printf("serving %d route(s) from the seat's plans: %s", len(routes), strings.Join(held.names(), ", "))
|
|
}
|
|
|
|
// change is one event serve says.
|
|
type change struct {
|
|
kind string
|
|
contract.Event
|
|
}
|
|
|
|
// servedChanges is what changed between two tables' names, each name once, in order: a name now served, or now
|
|
// served at another reach or by another module, is served; a name no longer served is withdrawn.
|
|
func servedChanges(proxy string, before, after map[string]contract.Event) []change {
|
|
var out []change
|
|
for _, name := range sortedNames(after) {
|
|
if was, ok := before[name]; !ok || was != after[name] {
|
|
e := after[name]
|
|
e.Proxy = proxy
|
|
out = append(out, change{contract.EventServed, e})
|
|
}
|
|
}
|
|
for _, name := range sortedNames(before) {
|
|
if _, ok := after[name]; !ok {
|
|
out = append(out, change{contract.EventWithdrawn, contract.Event{Proxy: proxy, Name: name}})
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
func sortedNames(m map[string]contract.Event) []string {
|
|
out := make([]string, 0, len(m))
|
|
for k := range m {
|
|
out = append(out, k)
|
|
}
|
|
sort.Strings(out)
|
|
return out
|
|
}
|
|
|
|
// followInside follows the seat mesh-placement's state and serves an internal name to every node's address on the
|
|
// private network it decides, each time it changes (mesh-sdk delivery.InsideAddresses; research 072 rows 33 and 34:
|
|
// the membership that carried them went with the assignments stream). A key that does not read keeps who is inside
|
|
// as it was.
|
|
func followInside(placement planWatch, held *table) error {
|
|
keys := delivery.NewPlacementKeys()
|
|
var read atomic.Bool
|
|
apply := func() {
|
|
inside, err := sourcesOf(delivery.InsideAddresses(keys.All()))
|
|
if err != nil {
|
|
log.Printf("the placement says an address that is not one, so who is inside stays as it was: %v", err)
|
|
return
|
|
}
|
|
held.setInside(inside)
|
|
log.Printf("internal names are served to %d machine(s)", len(inside))
|
|
}
|
|
err := placement.Watch("", func(c stdio.StateChange) error {
|
|
if _, err := keys.Put(c.Key, c.Value, c.Revision, c.Op == "delete" || c.Op == "purge"); err != nil {
|
|
log.Printf("the placement's key %s does not read, so who is inside stays as it was: %v", c.Key, err)
|
|
return nil
|
|
}
|
|
if read.Load() {
|
|
apply()
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
read.Store(true)
|
|
apply()
|
|
return nil
|
|
}
|