254 lines
9.3 KiB
Go
254 lines
9.3 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"sort"
|
|
"sync"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/compose"
|
|
sdkdecl "git.novox.be/novox/mesh-sdk/go/declaration"
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"github.com/novox/mesh-controller/internal/broker"
|
|
"github.com/novox/mesh-controller/internal/conditions"
|
|
)
|
|
|
|
// **A declaration's large files travel by their digest to a node that can read one** (novox/hq ADR 0486 §1, §5, issue 1696).
|
|
//
|
|
// The declaration to a node that reports sdkdecl.FileReferencesCapability carries, in place of the content of every
|
|
// file resource of more than sdkdecl.FileReferenceThreshold bytes that may travel so (File.TravelsByDigest: plain content or
|
|
// bytes, not sealed, not filled with secrets, not written into another file), its sha256 digest and size. The content is put
|
|
// into the bucket of the bus under the hex of the digest before the declaration is sent. A node that does not report the
|
|
// capability is sent every content inline, as before: an older node-engine refuses a field it does not know, whole.
|
|
|
|
// readsFileReferences says the node reports the capability of reading a file given by digest (ADR 0486 §5). A node the mesh has
|
|
// no profile of reads none.
|
|
func readsFileReferences(in compose.Input, node string) bool {
|
|
n, ok := nodeOf(in, node)
|
|
return ok && n.Capabilities[sdkdecl.FileReferencesCapability]
|
|
}
|
|
|
|
// referencing is the resources as a node that reads references is sent them, and the content of each file that went by
|
|
// digest, by the name of its object (the hex of its digest). The resources it is given are not changed. A file that does not
|
|
// travel by digest is passed as it is.
|
|
func referencing(resources []map[string]any) ([]map[string]any, map[string][]byte, error) {
|
|
out := make([]map[string]any, len(resources))
|
|
var objects map[string][]byte
|
|
for i, r := range resources {
|
|
out[i] = r
|
|
if kind, _ := r["type"].(string); kind != string(sdkdecl.TypeFile) {
|
|
continue
|
|
}
|
|
// Cheap first: only a file with a content or bytes of the size is worth reading whole.
|
|
content, _ := r["content"].(string)
|
|
bytesField, _ := r["bytes"].(string)
|
|
if len(content) <= sdkdecl.FileReferenceThreshold && len(bytesField) <= sdkdecl.FileReferenceThreshold {
|
|
continue
|
|
}
|
|
raw, err := json.Marshal(r)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("resource %v: %w", r["id"], err)
|
|
}
|
|
var f sdkdecl.File
|
|
if err := json.Unmarshal(raw, &f); err != nil {
|
|
return nil, nil, fmt.Errorf("resource %v is not a file the controller can read: %w", r["id"], err)
|
|
}
|
|
if !f.TravelsByDigest() {
|
|
continue
|
|
}
|
|
ref, body, err := f.AsReference()
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
name, err := sdkdecl.ObjectName(ref.Digest)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
with := make(map[string]any, len(r)+2)
|
|
for k, v := range r {
|
|
with[k] = v
|
|
}
|
|
delete(with, "content")
|
|
delete(with, "bytes")
|
|
with["digest"], with["size"] = ref.Digest, ref.Size
|
|
out[i] = with
|
|
if objects == nil {
|
|
objects = map[string][]byte{}
|
|
}
|
|
objects[name] = body
|
|
}
|
|
return out, objects, nil
|
|
}
|
|
|
|
// namedObjects are the objects of the bucket a body names: the object name of every file resource given by reference,
|
|
// with its size. It reads a declaration as the node does, so it also reads the latest one the bus holds.
|
|
func namedObjects(body []byte) (map[string]int64, error) {
|
|
var envelope struct {
|
|
Resources []struct {
|
|
Type string `json:"type"`
|
|
Digest string `json:"digest"`
|
|
Size int64 `json:"size"`
|
|
} `json:"resources"`
|
|
}
|
|
if err := json.Unmarshal(body, &envelope); err != nil {
|
|
return nil, err
|
|
}
|
|
var out map[string]int64
|
|
for _, r := range envelope.Resources {
|
|
if r.Type != string(sdkdecl.TypeFile) || r.Digest == "" {
|
|
continue
|
|
}
|
|
name, err := sdkdecl.ObjectName(r.Digest)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if out == nil {
|
|
out = map[string]int64{}
|
|
}
|
|
out[name] = r.Size
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// objectStore is what putting the objects needs of a node's bucket.
|
|
type objectStore interface {
|
|
// Has says whether an object of that name is there, complete, and whether it carries the sweep's mark of being named by no
|
|
// node's latest declaration.
|
|
Has(ctx context.Context, name string) (have, marked bool, err error)
|
|
// Put writes the object under that name, replacing one of the same name (and so erasing its mark).
|
|
Put(ctx context.Context, name string, content []byte) error
|
|
}
|
|
|
|
// putObjects makes sure every object of the declaration is in the bucket before the declaration that names it is sent: a node
|
|
// that reads the declaration finds each object. An object already there (the same content, the same name) is not written again,
|
|
// **unless the sweep marked it unnamed**: then it is written again, so the mark that would have it removed is gone with the
|
|
// old object (issue 1696, review: a mark read just before a send must not take an object the send relies on). Refuses on the
|
|
// first it cannot put, naming it, so the declaration is not sent.
|
|
func putObjects(ctx context.Context, store objectStore, objects map[string][]byte) error {
|
|
names := make([]string, 0, len(objects))
|
|
for name := range objects {
|
|
names = append(names, name)
|
|
}
|
|
sort.Strings(names)
|
|
for _, name := range names {
|
|
have, marked, err := store.Has(ctx, name)
|
|
if err != nil {
|
|
return fmt.Errorf("whether the object %s is in the bucket: %w", name[:12], err)
|
|
}
|
|
if have && !marked {
|
|
continue
|
|
}
|
|
if err := store.Put(ctx, name, objects[name]); err != nil {
|
|
return fmt.Errorf("putting the object %s (%d bytes) in the bucket: %w", name[:12], len(objects[name]), err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// busObjects is a node's bucket on the controller's own connection.
|
|
type busObjects struct{ store jetstream.ObjectStore }
|
|
|
|
// openBucket opens the bucket of a node's files, making it first if it is not there (it is asserted for every node at the
|
|
// controller's start; a node adopted since is covered here).
|
|
func openBucket(ctx context.Context, conn *nats.Conn, node string) (busObjects, error) {
|
|
js, err := jetstream.New(conn)
|
|
if err != nil {
|
|
return busObjects{}, err
|
|
}
|
|
bucket, err := broker.DeclarationFilesBucketOf(node)
|
|
if err != nil {
|
|
return busObjects{}, fmt.Errorf("the object bucket of node %q: %w", node, err)
|
|
}
|
|
store, err := js.ObjectStore(ctx, bucket)
|
|
if errors.Is(err, jetstream.ErrBucketNotFound) {
|
|
if err = broker.EnsureDeclarationFiles(ctx, js, node); err == nil {
|
|
store, err = js.ObjectStore(ctx, bucket)
|
|
}
|
|
}
|
|
if err != nil {
|
|
return busObjects{}, fmt.Errorf("the object bucket %s: %w", bucket, err)
|
|
}
|
|
return busObjects{store}, nil
|
|
}
|
|
|
|
func (b busObjects) Has(ctx context.Context, name string) (bool, bool, error) {
|
|
info, err := b.store.GetInfo(ctx, name)
|
|
switch {
|
|
case errors.Is(err, jetstream.ErrObjectNotFound):
|
|
return false, false, nil
|
|
case err != nil:
|
|
return false, false, err
|
|
}
|
|
_, marked := info.Metadata[declarationUnnamed]
|
|
return !info.Deleted, marked, nil
|
|
}
|
|
|
|
func (b busObjects) Put(ctx context.Context, name string, content []byte) error {
|
|
_, err := b.store.PutBytes(ctx, name, content)
|
|
return err
|
|
}
|
|
|
|
// filesLocks hold a node's bucket for one act at a time in this process: a send puts its objects and publishes the declaration that
|
|
// names them under the node's lock, and the sweep judges the node's bucket under the same one, so the sweep never removes an
|
|
// object between a send's put and its declaration. Another process (a person's push) is covered by the put erasing a mark and by
|
|
// the sweep looking at an object again just before it removes it.
|
|
var filesLocks sync.Map
|
|
|
|
func lockFilesOf(node string) func() {
|
|
l, _ := filesLocks.LoadOrStore(node, &sync.Mutex{})
|
|
mu := l.(*sync.Mutex)
|
|
mu.Lock()
|
|
return mu.Unlock
|
|
}
|
|
|
|
// kindFilesNotPut is a node whose files that travel by digest could not be put in its bucket, send after send.
|
|
const kindFilesNotPut = "declaration-files-not-put"
|
|
|
|
// filesNotPutAfter is how many sends in a row may fail to put before the condition is raised: the first send to the machine holding
|
|
// the bus fails for want of the grant that very send delivers, and says so, and is not yet a fault.
|
|
const filesNotPutAfter = 2
|
|
|
|
var (
|
|
filesFailMu sync.Mutex
|
|
filesFailed = map[string]int{}
|
|
)
|
|
|
|
func filesNotPutObservation(node, summary string) conditions.Observation {
|
|
return conditions.Observation{Scope: conditions.ScopeMachine, ID: node, Kind: kindFilesNotPut, Token: kindFilesNotPut,
|
|
Machine: node, Severity: conditions.Warning, Source: "sends", Summary: summary}
|
|
}
|
|
|
|
// notePut records how putting a send's objects went: a failure counts, the second in a row raises the condition, a success
|
|
// clears the count and the condition.
|
|
func notePut(ctx context.Context, node string, put error) {
|
|
filesFailMu.Lock()
|
|
if put == nil {
|
|
delete(filesFailed, node)
|
|
} else {
|
|
filesFailed[node]++
|
|
}
|
|
n := filesFailed[node]
|
|
filesFailMu.Unlock()
|
|
keeper := conditionsFrom
|
|
if keeper == nil {
|
|
return
|
|
}
|
|
key := filesNotPutObservation(node, "").Key()
|
|
switch {
|
|
case put == nil:
|
|
if _, err := keeper.Clear(ctx, key, node+" was sent its files by digest"); err != nil {
|
|
fmt.Printf("the condition %s could not be cleared: %v\n", key, err)
|
|
}
|
|
case n >= filesNotPutAfter:
|
|
summary := fmt.Sprintf("%s's files that travel by digest could not be put in its bucket %d sends in a row, so it is sent them "+
|
|
"inline: %v", node, n, put)
|
|
if _, err := keeper.Observe(ctx, filesNotPutObservation(node, summary)); err != nil {
|
|
fmt.Printf("the condition %s could not be raised: %v\n", key, err)
|
|
}
|
|
}
|
|
}
|