247 lines
11 KiB
Go
247 lines
11 KiB
Go
package main
|
|
|
|
// The two tools of this module that change the bus (novox/hq ADR 0564 clause 5, issue 1952 notes 6, 7 and 9):
|
|
// remove one named consumer, and remove one node's keys under a prefix of a key-value bucket. Everything else
|
|
// here reads.
|
|
//
|
|
// **Narrow by what the tool needs, and narrower by what the bus grants.** The tools reach the bus on the
|
|
// credential the module's manifest names (NATS_TOOLS_CREDENTIAL); the bus refuses whatever that credential is not
|
|
// granted, so a wrong stream, consumer or bucket is refused by the server and said here, not trusted to the
|
|
// arguments. Each tool refuses without `why`, reads what it would remove first (and says it, so a person can ask
|
|
// with dry_run and then for real), removes, reads that it is gone, and says what it did as an event of the module
|
|
// (the audit trail: who asked is in the call the controller audits, why and what are in the event).
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"regexp"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
)
|
|
|
|
// Bus opens JetStream on the bus as the module's credential for one call; close ends the connection.
|
|
type Bus func(ctx context.Context) (js jetstream.JetStream, close func(), err error)
|
|
|
|
// Said keeps the audit line: an event of this module.
|
|
type Said func(event string, body any) error
|
|
|
|
const (
|
|
consumerRemoved = "consumer-removed"
|
|
bucketKeysRemoved = "bucket-keys-removed"
|
|
// maxKeysShown bounds the keys an answer lists; the count is always whole.
|
|
maxKeysShown = 200
|
|
)
|
|
|
|
var (
|
|
// A stream, consumer or bucket is one subject token: no dot, no wildcard, no space.
|
|
name = regexp.MustCompile(`^[A-Za-z0-9_-]+$`)
|
|
// A key prefix is subject tokens and their dots; no wildcard, so it names keys and not a pattern.
|
|
keyPrefix = regexp.MustCompile(`^[A-Za-z0-9_./=-]+$`)
|
|
)
|
|
|
|
func why(a map[string]any) (string, error) {
|
|
w := strings.TrimSpace(arg(a, "why"))
|
|
if w == "" {
|
|
return "", errors.New("why is required: say why this is removed, for the audit trail (novox/hq ADR 0564 clause 5); nothing was removed")
|
|
}
|
|
return w, nil
|
|
}
|
|
|
|
// answerWithin is how long the bus is given to answer one request. **The bus does not refuse a request its
|
|
// credential is not granted: it drops it** (a permissions violation is told to the connection's error handler,
|
|
// never answered), so an unanswered request is how a refusal shows here.
|
|
var answerWithin = 10 * time.Second
|
|
|
|
// unanswered says what a request with no answer most likely is.
|
|
func unanswered(err error, what string) error {
|
|
if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, nats.ErrTimeout) {
|
|
return fmt.Errorf("%s: the bus did not answer within %s; it drops, without an answer, what the module's credential is not granted, "+
|
|
"so this is most likely outside its grant (or the bus is not reachable): %w", what, answerWithin, err)
|
|
}
|
|
return fmt.Errorf("%s: %w", what, err)
|
|
}
|
|
|
|
func step(ctx context.Context) (context.Context, context.CancelFunc) {
|
|
return context.WithTimeout(ctx, answerWithin)
|
|
}
|
|
|
|
// ConsumerRemoved is what nats_consumer_remove answers and says.
|
|
type ConsumerRemoved struct {
|
|
Stream string `json:"stream"`
|
|
Consumer string `json:"consumer"`
|
|
Why string `json:"why"`
|
|
Was map[string]any `json:"was"`
|
|
Removed bool `json:"removed"`
|
|
At time.Time `json:"at"`
|
|
}
|
|
|
|
// RemoveConsumer removes one named consumer of one stream, after reading it, and says so.
|
|
func RemoveConsumer(ctx context.Context, bus Bus, said Said, a map[string]any) (any, error) {
|
|
w, err := why(a)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
stream, consumer := arg(a, "stream"), arg(a, "consumer")
|
|
if !name.MatchString(stream) || !name.MatchString(consumer) {
|
|
return nil, errors.New("stream and consumer are required, each one name (letters, digits, - and _): nothing was removed")
|
|
}
|
|
js, closeBus, err := bus(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer closeBus()
|
|
sctx, cancel := step(ctx)
|
|
c, err := js.Consumer(sctx, stream, consumer)
|
|
cancel()
|
|
if err != nil {
|
|
if errors.Is(err, jetstream.ErrConsumerNotFound) || errors.Is(err, jetstream.ErrStreamNotFound) {
|
|
return nil, fmt.Errorf("%s has no consumer %s (%v): nothing was removed", stream, consumer, err)
|
|
}
|
|
return nil, unanswered(err, fmt.Sprintf("the consumer %s of %s could not be read, so it was not removed", consumer, stream))
|
|
}
|
|
info := c.CachedInfo()
|
|
was := map[string]any{
|
|
"durable": info.Config.Durable, "filter_subject": info.Config.FilterSubject, "filter_subjects": info.Config.FilterSubjects,
|
|
"deliver_subject": info.Config.DeliverSubject, "pending": info.NumPending, "ack_pending": info.NumAckPending,
|
|
"redelivered": info.NumRedelivered, "created": info.Created,
|
|
}
|
|
sctx, cancel = step(ctx)
|
|
err = js.DeleteConsumer(sctx, stream, consumer)
|
|
cancel()
|
|
if err != nil {
|
|
return nil, unanswered(err, fmt.Sprintf("consumer %s of %s was not removed", consumer, stream))
|
|
}
|
|
sctx, cancel = step(ctx)
|
|
_, err = js.Consumer(sctx, stream, consumer)
|
|
cancel()
|
|
if err == nil {
|
|
return nil, fmt.Errorf("consumer %s of %s is still there after the removal was accepted", consumer, stream)
|
|
}
|
|
out := ConsumerRemoved{Stream: stream, Consumer: consumer, Why: w, Was: was, Removed: true, At: time.Now().UTC()}
|
|
if err := said(consumerRemoved, out); err != nil {
|
|
return nil, fmt.Errorf("consumer %s of %s WAS removed (%s) but the audit event could not be said: %w", consumer, stream, w, err)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// KeysRemoved is what nats_bucket_remove answers and says.
|
|
type KeysRemoved struct {
|
|
Bucket string `json:"bucket"`
|
|
Prefix string `json:"prefix"`
|
|
Why string `json:"why"`
|
|
DryRun bool `json:"dry_run"`
|
|
Count int `json:"count"`
|
|
Keys []string `json:"keys"`
|
|
Cut bool `json:"keys_cut,omitempty"`
|
|
Removed int `json:"removed"`
|
|
At time.Time `json:"at"`
|
|
}
|
|
|
|
// RemoveKeys removes the keys of a key-value bucket under a prefix: counted first, and only those.
|
|
func RemoveKeys(ctx context.Context, bus Bus, said Said, a map[string]any) (any, error) {
|
|
w, err := why(a)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
bucket, node, prefix := arg(a, "bucket"), arg(a, "node"), arg(a, "prefix")
|
|
if !name.MatchString(bucket) {
|
|
return nil, errors.New("bucket is required, a bucket's name as nats_buckets lists it: nothing was removed")
|
|
}
|
|
if (node == "") == (prefix == "") {
|
|
return nil, errors.New("give exactly one of node (that node's keys: those starting `<node>.`) or prefix (keys starting so): " +
|
|
"a removal of a whole bucket is not this tool's; nothing was removed")
|
|
}
|
|
if node != "" {
|
|
if !name.MatchString(node) {
|
|
return nil, errors.New("node is one machine's name: nothing was removed")
|
|
}
|
|
prefix = node + "."
|
|
}
|
|
if !keyPrefix.MatchString(prefix) {
|
|
return nil, errors.New("prefix is key text without wildcards or spaces: nothing was removed")
|
|
}
|
|
dry, _ := a["dry_run"].(bool)
|
|
js, closeBus, err := bus(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer closeBus()
|
|
sctx, cancel := step(ctx)
|
|
kv, err := js.KeyValue(sctx, bucket)
|
|
cancel()
|
|
if err != nil {
|
|
return nil, unanswered(err, fmt.Sprintf("bucket %s could not be opened, nothing was removed", bucket))
|
|
}
|
|
sctx, cancel = step(ctx)
|
|
stream, err := js.Stream(sctx, "KV_"+bucket)
|
|
cancel()
|
|
if err != nil {
|
|
return nil, unanswered(err, fmt.Sprintf("bucket %s has no stream to read its keys from", bucket))
|
|
}
|
|
// The server narrows to the prefix's tokens when the prefix ends on a token; any other prefix is cut here.
|
|
filter := "$KV." + bucket + ".>"
|
|
if strings.HasSuffix(prefix, ".") {
|
|
filter = "$KV." + bucket + "." + prefix + ">"
|
|
}
|
|
sctx, cancel = step(ctx)
|
|
info, err := stream.Info(sctx, jetstream.WithSubjectFilter(filter))
|
|
cancel()
|
|
if err != nil {
|
|
return nil, unanswered(err, fmt.Sprintf("the keys of %s could not be read, so none were removed", bucket))
|
|
}
|
|
var keys []string
|
|
for subject := range info.State.Subjects {
|
|
if key := strings.TrimPrefix(subject, "$KV."+bucket+"."); strings.HasPrefix(key, prefix) {
|
|
keys = append(keys, key)
|
|
}
|
|
}
|
|
sort.Strings(keys)
|
|
out := KeysRemoved{Bucket: bucket, Prefix: prefix, Why: w, DryRun: dry, Count: len(keys), Keys: keys, At: time.Now().UTC()}
|
|
if len(keys) > maxKeysShown {
|
|
out.Keys, out.Cut = keys[:maxKeysShown], true
|
|
}
|
|
if want, ok := a["expect"].(float64); ok && int(want) != len(keys) {
|
|
return nil, fmt.Errorf("%d keys of %s start with %q, not the %d expected: nothing was removed", len(keys), bucket, prefix, int(want))
|
|
}
|
|
if dry || len(keys) == 0 {
|
|
return out, nil
|
|
}
|
|
for _, k := range keys {
|
|
sctx, cancel := step(ctx)
|
|
err := kv.Purge(sctx, k)
|
|
cancel()
|
|
if err != nil {
|
|
return nil, unanswered(err, fmt.Sprintf("removing %q from %s failed after %d of %d keys, the rest were left", k, bucket, out.Removed, len(keys)))
|
|
}
|
|
out.Removed++
|
|
}
|
|
if err := said(bucketKeysRemoved, out); err != nil {
|
|
return nil, fmt.Errorf("%d keys of %s under %q WERE removed (%s) but the audit event could not be said: %w", out.Removed, bucket, prefix, w, err)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// removalTools are the module's two tools that change the bus; every other tool here reads.
|
|
func removalTools(bus Bus, said Said) []stdio.Tool {
|
|
run := func(f func(context.Context, Bus, Said, map[string]any) (any, error)) func(map[string]any) (any, error) {
|
|
return func(a map[string]any) (any, error) { return f(context.Background(), bus, said, a) }
|
|
}
|
|
return []stdio.Tool{
|
|
{Name: "nats_consumer_remove",
|
|
Description: "Removes (deletes) one named JetStream consumer of one stream, e.g. the consumer `controller` of DEAD_LETTER_NOTICES at the lease write (novox/hq ADR 0564 clause 5). Needs `why`, kept in the audit trail; refuses an unknown stream or consumer; reads the consumer first and answers what it was (filters, pending, unacknowledged); the stream and its other consumers are not touched. The bus refuses any consumer the module's credential is not granted.",
|
|
Input: map[string]any{"stream": str("the stream, e.g. DEAD_LETTER_NOTICES"), "consumer": str("the consumer's name, e.g. controller"), "why": str("why it is removed, for the audit trail (required)")},
|
|
Run: run(RemoveConsumer)},
|
|
{Name: "nats_bucket_remove",
|
|
Description: "Removes (purges) keys from a key-value bucket: one node's keys (`node`, the keys starting `<node>.`) or the keys under a `prefix`, e.g. a full acts bucket of the node-engine. Needs `why`, kept in the audit trail; never the whole bucket. Counts the keys first and lists them (`dry_run` only counts; `expect` refuses when the count differs); removes only those keys, and their history. See the buckets and their keys with nats_buckets. The bus refuses any bucket the module's credential is not granted.",
|
|
Input: map[string]any{"bucket": str("the bucket as nats_buckets lists it, e.g. mesh-host_acts"), "node": str("a node whose keys (`<node>.`) go"), "prefix": str("key prefix instead of node"),
|
|
"why": str("why they are removed, for the audit trail (required)"), "dry_run": map[string]any{"type": "boolean", "description": "count and list the keys, remove none"}, "expect": num("the count of keys expected; refuses when it differs")},
|
|
Run: run(RemoveKeys)},
|
|
}
|
|
}
|