diff --git a/config/light-testnet.yaml b/config/light-testnet.yaml new file mode 100644 index 00000000..e0669c2f --- /dev/null +++ b/config/light-testnet.yaml @@ -0,0 +1,75 @@ +# bee-light-testnet: 20 full nodes on an isolated testnet (swarm1), used for A/B runs. +# The bootnode has no ingress and is not part of the cluster definition. +clusters: + bee-light-testnet: + _inherit: "" + name: bee + namespace: bee-light-testnet + disable-namespace: true + api-domain: lightnet.testnet.internal + api-domain-internal: bee-light-testnet.svc.swarm1.local:1633 + api-insecure-tls: true + api-scheme: http + funding: + eth: 0.1 + gbzz: 2.0 + bzz: 5.0 + node-groups: + bee-1: + mode: node + bee-config: light-testnet + config: ng-light-testnet + count: 5 + bee-2: + mode: node + bee-config: light-testnet + config: ng-light-testnet + count: 5 + bee-3: + mode: node + bee-config: light-testnet + config: ng-light-testnet + count: 5 + bee-4: + mode: node + bee-config: light-testnet + config: ng-light-testnet + count: 5 + +node-groups: + ng-light-testnet: + _inherit: "" + +bee-configs: + light-testnet: + _inherit: "" + full-node: true + +checks: + lt-pushsync-placement: + options: + chunks-per-node: 2 # per neighborhood, per uploader + mode: placement + postage-ttl: 24h + postage-depth: 22 + postage-label: test-label + retry-delay: 5s + storage-radius: 2 # the radius the nodes report in /status + upload-nodes: [bee-1-2, bee-2-0, bee-3-0, bee-3-1, bee-4-3, bee-4-4] + upload-timeout: 60s + timeout: 60m + type: pushsync + lt-pushsync-placement-deferred: + options: + chunks-per-node: 2 + deferred: true + mode: placement + postage-ttl: 24h + postage-depth: 22 + postage-label: test-label + retry-delay: 30s # the pusher queue is busy with the background load + storage-radius: 2 + upload-nodes: [bee-1-2, bee-2-0, bee-3-0, bee-3-1, bee-4-3, bee-4-4] + upload-timeout: 60s + timeout: 60m + type: pushsync diff --git a/config/local.yaml b/config/local.yaml index d389ee2b..25914d8c 100644 --- a/config/local.yaml +++ b/config/local.yaml @@ -40,7 +40,7 @@ clusters: bee: bee-config: bee-local-dns config: local-dns - count: 3 + count: 5 mode: node light: bee-config: bee-local-light @@ -207,10 +207,14 @@ bee-configs: bootnode-local: _inherit: "bee-local" bootnode-mode: true + # a forced storage radius splits the small cluster into neighborhoods, some + # of them empty, so that out-of-depth storing and shallow receipts can occur bee-local-dns: _inherit: "bee-local" + minimum-storage-radius: 3 bootnode-local-dns: _inherit: "bee-local" + minimum-storage-radius: 3 bootnode-mode: true bee-local-light: _inherit: "bee-local" @@ -304,6 +308,33 @@ checks: - bootnode timeout: 5m type: pushsync + ci-pushsync-placement: + options: + chunks-per-node: 2 # per neighborhood, per uploader + mode: placement + postage-ttl: 24h + postage-depth: 21 + postage-label: test-label + retry-delay: 3s + storage-radius: 3 # must match minimum-storage-radius of the full nodes + upload-node-count: 2 + upload-timeout: 30s + timeout: 30m + type: pushsync + ci-pushsync-placement-deferred: + options: + chunks-per-node: 2 # per neighborhood, per uploader + deferred: true + mode: placement + postage-ttl: 24h + postage-depth: 21 + postage-label: test-label + retry-delay: 3s + storage-radius: 3 # must match minimum-storage-radius of the full nodes + upload-node-count: 2 + upload-timeout: 30s + timeout: 30m + type: pushsync ci-pushsync-light-chunks: options: chunks-per-node: 3 diff --git a/pkg/bee/api/chunks.go b/pkg/bee/api/chunks.go index e384b6e3..04b9d4b1 100644 --- a/pkg/bee/api/chunks.go +++ b/pkg/bee/api/chunks.go @@ -5,6 +5,7 @@ import ( "context" "io" "net/http" + "strconv" "github.com/ethersphere/bee/v2/pkg/swarm" ) @@ -36,6 +37,9 @@ func (c *ChunksService) Upload(ctx context.Context, data []byte, o UploadOptions if o.Direct { h.Add(deferredUploadHeader, "false") } + if o.Tag != 0 { + h.Add(swarmTagHeader, strconv.FormatUint(o.Tag, 10)) + } h.Add(postageStampBatchHeader, o.BatchID) err := c.client.requestWithHeader(ctx, http.MethodPost, "/"+apiVersion+"/chunks", h, bytes.NewReader(data), &resp) return resp, err diff --git a/pkg/bee/api/node.go b/pkg/bee/api/node.go index 789eeccc..a767bb78 100644 --- a/pkg/bee/api/node.go +++ b/pkg/bee/api/node.go @@ -3,6 +3,7 @@ package api import ( "context" "fmt" + "io" "net/http" "time" @@ -95,6 +96,36 @@ func (n *NodeService) HasChunk(ctx context.Context, a swarm.Address) (bool, erro return true, nil } +// HasChunkLocal reports whether the chunk is in the node's local store. Unlike +// HasChunk it uses HEAD, which never retrieves the chunk from the network and +// so does not change what the node stores. +func (n *NodeService) HasChunkLocal(ctx context.Context, a swarm.Address) (bool, error) { + err := n.client.request(ctx, http.MethodHead, "/chunks/"+a.String(), nil, nil) + if IsHTTPStatusErrorCode(err, http.StatusNotFound) { + return false, nil + } else if err != nil { + return false, err + } + + return true, nil +} + +// Metrics returns the node's metrics in the Prometheus text format. +func (n *NodeService) Metrics(ctx context.Context) (string, error) { + r, err := n.client.requestData(ctx, http.MethodGet, "/metrics", nil, nil) + if err != nil { + return "", err + } + defer r.Close() + + b, err := io.ReadAll(r) + if err != nil { + return "", err + } + + return string(b), nil +} + // Health represents node's health type Health struct { Status string `json:"status"` diff --git a/pkg/bee/client.go b/pkg/bee/client.go index 096c68da..2aa66704 100644 --- a/pkg/bee/client.go +++ b/pkg/bee/client.go @@ -270,6 +270,17 @@ func (c *Client) HasChunk(ctx context.Context, a swarm.Address) (bool, error) { return c.api.Node.HasChunk(ctx, a) } +// HasChunkLocal reports whether the node has the chunk in its local store, +// without retrieving it from the network. +func (c *Client) HasChunkLocal(ctx context.Context, a swarm.Address) (bool, error) { + return c.api.Node.HasChunkLocal(ctx, a) +} + +// Metrics returns the node's metrics in the Prometheus text format. +func (c *Client) Metrics(ctx context.Context) (string, error) { + return c.api.Node.Metrics(ctx) +} + func (c *Client) HasChunks(ctx context.Context, a []swarm.Address) (has []bool, count int, err error) { has = make([]bool, len(a)) for i, addr := range a { diff --git a/pkg/check/pushsync/check_placement.go b/pkg/check/pushsync/check_placement.go new file mode 100644 index 00000000..26e39c58 --- /dev/null +++ b/pkg/check/pushsync/check_placement.go @@ -0,0 +1,356 @@ +package pushsync + +import ( + "context" + "errors" + "fmt" + "math/rand" + "sort" + "strconv" + "strings" + "time" + + "github.com/ethersphere/bee/v2/pkg/swarm" + "github.com/ethersphere/beekeeper/pkg/bee" + "github.com/ethersphere/beekeeper/pkg/bee/api" + "github.com/ethersphere/beekeeper/pkg/logging" + "github.com/ethersphere/beekeeper/pkg/orchestration" + "github.com/ethersphere/beekeeper/pkg/random" +) + +// placement outcomes of a single uploaded chunk +const ( + // upload succeeded and at least one node holds the chunk within its AOR + outcomeStoredInAOR = "stored_in_aor" + // upload succeeded but no node holds the chunk within its AOR + outcomeFalseSuccess = "false_success" + // upload failed or timed out and no node holds the chunk within its AOR + outcomeReportedFailure = "reported_failure" + // upload failed or timed out although a node holds the chunk within its AOR + outcomeFailedButStored = "failed_but_stored" +) + +// errNotSynced marks a deferred upload whose tag does not count the chunk as synced. +var errNotSynced = errors.New("not synced") + +// bee counters summed over all nodes and reported as a delta over the run +var placementBeeMetrics = []string{ + "bee_pushsync_storer", + "bee_pushsync_shallow_receipt", + "bee_pushsync_out_of_depth_storing", + "bee_pusher_total_to_push", + "bee_pusher_total_synced", + "bee_pusher_total_could_not_sync", +} + +type placementResult struct { + chunk swarm.Address + uploader string + populated bool // some storer node has the chunk within its AOR + uploadErr error + uploadTime time.Duration + inAOR []string // holders with the chunk within their AOR + outOfAOR []string // holders with the chunk outside their AOR + queued bool // deferred: not synced and still in the uploader's queue + outcomeName string +} + +// checkPlacement uploads chunks into every neighborhood of the configured +// storage radius and then asks every node, with a HEAD request that never +// touches the network, whether it holds the chunk. A chunk only counts as +// stored when a holder has it within its area of responsibility. The check +// fails when an upload is reported as successful while no such holder exists, +// or when a chunk whose neighborhood is populated does not end up there. +// +// A direct upload reports its outcome with the response of the upload request. +// A deferred upload reports it with its tag: it is successful once the tag +// counts the chunk as synced. +func checkPlacement(ctx context.Context, c orchestration.Cluster, o Options, l logging.Logger) error { + l.Infof("running pushsync (placement mode), storage radius %d, deferred %t", o.StorageRadius, o.Deferred) + l.Infof("seed: %d", o.Seed) + + if o.StorageRadius > 8 { + return fmt.Errorf("storage radius %d is too high for the placement check", o.StorageRadius) + } + + overlays, err := c.FlattenOverlays(ctx) + if err != nil { + return err + } + clients, err := c.NodesClients(ctx) + if err != nil { + return err + } + + storers := c.FullNodeNames() + sort.Strings(storers) + if o.UploadNodeCount > len(storers) { + return fmt.Errorf("not enough full nodes in the cluster, required %d, available %d", o.UploadNodeCount, len(storers)) + } + uploaders := storers[:o.UploadNodeCount] + if len(o.UploadNodes) > 0 { + uploaders = o.UploadNodes + } + isStorer := make(map[string]bool, len(storers)) + for _, name := range storers { + isStorer[name] = true + } + + nodeNames := make([]string, 0, len(overlays)) + for name := range overlays { + nodeNames = append(nodeNames, name) + } + sort.Strings(nodeNames) + + for _, name := range nodeNames { + status, err := clients[name].Status(ctx) + if err != nil { + return fmt.Errorf("node %s: status: %w", name, err) + } + l.Infof("node %s: overlay %s neighborhood %s mode %s reported radius %d storer %t", + name, overlays[name], neighborhood(overlays[name], o.StorageRadius), status.BeeMode, status.StorageRadius, isStorer[name]) + } + + // holdsInAOR tells whether the node is responsible for storing the chunk + holdsInAOR := func(name string, chunk swarm.Address) bool { + return isStorer[name] && swarm.Proximity(overlays[name].Bytes(), chunk.Bytes()) >= o.StorageRadius + } + + before, err := sumBeeMetrics(ctx, clients, nodeNames) + if err != nil { + return err + } + + rnd := random.PseudoGenerator(o.Seed) + var results []placementResult + + for _, uploaderName := range uploaders { + uploader, ok := clients[uploaderName] + if !ok { + return fmt.Errorf("upload node %s not found in the cluster", uploaderName) + } + + batchID, err := uploader.GetOrCreateMutableBatch(ctx, o.PostageTTL, o.PostageDepth, o.PostageLabel) + if err != nil { + return fmt.Errorf("node %s: batch id %w", uploaderName, err) + } + l.Infof("node %s: batch id %s", uploaderName, batchID) + + for prefix := range 1 << o.StorageRadius { + for range o.ChunksPerNode { + chunk, err := randomChunkInNeighborhood(rnd, l, prefix, o.StorageRadius) + if err != nil { + return fmt.Errorf("node %s: %w", uploaderName, err) + } + + res := placementResult{chunk: chunk.Address(), uploader: uploaderName} + for _, name := range storers { + if holdsInAOR(name, res.chunk) { + res.populated = true + break + } + } + + uploadOpts := api.UploadOptions{BatchID: batchID, Direct: !o.Deferred} + if o.Deferred { + tag, err := uploader.CreateTag(ctx) + if err != nil { + return fmt.Errorf("node %s: %w", uploaderName, err) + } + uploadOpts.Tag = tag.Uid + } + + uploadCtx, cancel := context.WithTimeout(ctx, o.UploadTimeout) + t0 := time.Now() + _, res.uploadErr = uploader.UploadChunk(uploadCtx, chunk.Data(), uploadOpts) + res.uploadTime = time.Since(t0) + cancel() + if ctx.Err() != nil { + return ctx.Err() + } + + // give the nodes time to sync and replicate before looking for the chunk + time.Sleep(o.RetryDelay) + + if o.Deferred && res.uploadErr == nil { + tag, err := uploader.GetTag(ctx, uploadOpts.Tag) + if err != nil { + return fmt.Errorf("node %s: %w", uploaderName, err) + } + if tag.Synced == 0 { + res.uploadErr = errNotSynced + } + } + + for _, name := range nodeNames { + has, err := clients[name].HasChunkLocal(ctx, res.chunk) + if err != nil { + return fmt.Errorf("node %s: has chunk %s: %w", name, res.chunk, err) + } + if !has { + continue + } + switch { + case holdsInAOR(name, res.chunk): + res.inAOR = append(res.inAOR, name) + case name == uploaderName && errors.Is(res.uploadErr, errNotSynced): + // the chunk waits in the upload queue, it is not stored + res.queued = true + default: + res.outOfAOR = append(res.outOfAOR, name) + } + } + + switch { + case res.uploadErr == nil && len(res.inAOR) > 0: + res.outcomeName = outcomeStoredInAOR + case res.uploadErr == nil: + res.outcomeName = outcomeFalseSuccess + case len(res.inAOR) > 0: + res.outcomeName = outcomeFailedButStored + default: + res.outcomeName = outcomeReportedFailure + } + + l.Infof("chunk %s neighborhood %s populated %t uploader %s upload %s in %s: %s, in AOR %v, out of AOR %v, queued %t", + res.chunk, neighborhood(res.chunk, o.StorageRadius), res.populated, uploaderName, + uploadStatus(res.uploadErr), res.uploadTime.Round(time.Millisecond), res.outcomeName, res.inAOR, res.outOfAOR, res.queued) + + results = append(results, res) + } + } + } + + after, err := sumBeeMetrics(ctx, clients, nodeNames) + if err != nil { + return err + } + + return reportPlacement(l, results, before, after) +} + +// reportPlacement logs the summary and returns an error when the placement +// guarantees do not hold. +func reportPlacement(l logging.Logger, results []placementResult, before, after map[string]float64) error { + type key struct { + populated bool + outcome string + } + counts := make(map[key]int) + total := map[bool]int{} + strayCopies, queued := 0, 0 + for _, r := range results { + counts[key{r.populated, r.outcomeName}]++ + total[r.populated]++ + if len(r.outOfAOR) > 0 { + strayCopies++ + } + if r.queued { + queued++ + } + } + + l.Info("placement summary") + for _, populated := range []bool{true, false} { + label := "populated neighborhood" + if !populated { + label = "empty neighborhood" + } + l.Infof(" %s: %d chunks", label, total[populated]) + for _, outcome := range []string{outcomeStoredInAOR, outcomeFalseSuccess, outcomeReportedFailure, outcomeFailedButStored} { + l.Infof(" %-18s %d", outcome, counts[key{populated, outcome}]) + } + } + l.Infof(" chunks with a copy outside of the holder's AOR: %d", strayCopies) + l.Infof(" chunks not synced and still in the uploader's queue: %d", queued) + + l.Info("bee metrics delta, summed over all nodes") + for _, name := range placementBeeMetrics { + l.Infof(" %-36s %.0f", name, after[name]-before[name]) + } + + falseSuccess := counts[key{true, outcomeFalseSuccess}] + counts[key{false, outcomeFalseSuccess}] + lostInPopulated := total[true] - counts[key{true, outcomeStoredInAOR}] + + var errs []error + if falseSuccess > 0 { + errs = append(errs, fmt.Errorf("%d of %d uploads reported success with no node holding the chunk within its AOR", falseSuccess, len(results))) + } + if lostInPopulated > 0 { + errs = append(errs, fmt.Errorf("%d of %d chunks with a populated neighborhood were not stored there", lostInPopulated, total[true])) + } + + return errors.Join(errs...) +} + +// randomChunkInNeighborhood generates a random chunk whose address starts +// with the given prefix of radius bits. +func randomChunkInNeighborhood(rnd *rand.Rand, l logging.Logger, prefix int, radius uint8) (bee.Chunk, error) { + for { + chunk, err := bee.NewRandomChunk(rnd, l) + if err != nil { + return bee.Chunk{}, err + } + if addressPrefix(chunk.Address(), radius) == prefix { + return chunk, nil + } + } +} + +// addressPrefix returns the first radius bits of the address, radius <= 8. +func addressPrefix(a swarm.Address, radius uint8) int { + return int(a.Bytes()[0] >> (8 - radius)) +} + +// neighborhood returns the first radius bits of the address in binary. +func neighborhood(a swarm.Address, radius uint8) string { + if radius == 0 { + return "-" + } + return fmt.Sprintf("%0*b", int(radius), addressPrefix(a, radius)) +} + +func uploadStatus(err error) string { + switch { + case err == nil: + return "ok" + case errors.Is(err, context.DeadlineExceeded): + return "timeout" + case errors.Is(err, errNotSynced): + return "not synced" + default: + return fmt.Sprintf("failed (%v)", err) + } +} + +// sumBeeMetrics sums the observed bee counters over the given nodes. +func sumBeeMetrics(ctx context.Context, clients map[string]*bee.Client, nodeNames []string) (map[string]float64, error) { + sums := make(map[string]float64, len(placementBeeMetrics)) + for _, name := range nodeNames { + text, err := clients[name].Metrics(ctx) + if err != nil { + return nil, fmt.Errorf("node %s: metrics: %w", name, err) + } + for line := range strings.SplitSeq(text, "\n") { + if line == "" || strings.HasPrefix(line, "#") { + continue + } + i := strings.LastIndexByte(line, ' ') + if i < 0 { + continue + } + metric, _, _ := strings.Cut(line[:i], "{") + for _, want := range placementBeeMetrics { + if metric != want { + continue + } + v, err := strconv.ParseFloat(line[i+1:], 64) + if err != nil { + return nil, fmt.Errorf("node %s: metric %s: %w", name, metric, err) + } + sums[metric] += v + } + } + } + return sums, nil +} diff --git a/pkg/check/pushsync/pushsync.go b/pkg/check/pushsync/pushsync.go index 689e2624..04f737c5 100644 --- a/pkg/check/pushsync/pushsync.go +++ b/pkg/check/pushsync/pushsync.go @@ -26,6 +26,10 @@ type Options struct { Seed int64 UploadNodeCount int ExcludeNodeGroups []string + StorageRadius uint8 // placement mode: storage radius the cluster nodes run with + UploadTimeout time.Duration // placement mode: time after which a direct upload counts as timed out + Deferred bool // placement mode: use deferred uploads and read the outcome from the tag + UploadNodes []string // placement mode: uploader names; overrides UploadNodeCount when set } // NewDefaultOptions returns new default options @@ -42,6 +46,8 @@ func NewDefaultOptions() Options { Seed: random.Int64(), UploadNodeCount: 1, ExcludeNodeGroups: []string{}, + StorageRadius: 0, + UploadTimeout: 30 * time.Second, } } @@ -73,6 +79,8 @@ func (c *Check) Run(ctx context.Context, cluster orchestration.Cluster, opts any return checkChunks(ctx, cluster, o, c.logger, c.metrics) case "light-chunks": return checkLightChunks(ctx, cluster, o, c.logger, c.metrics) + case "placement": + return checkPlacement(ctx, cluster, o, c.logger) default: return c.defaultCheck(ctx, cluster, o) } diff --git a/pkg/config/check.go b/pkg/config/check.go index 541afb7e..1978f875 100644 --- a/pkg/config/check.go +++ b/pkg/config/check.go @@ -373,6 +373,10 @@ var Checks = map[string]CheckType{ Seed *int64 `yaml:"seed"` UploadNodeCount *int `yaml:"upload-node-count"` ExcludeNodeGroups *[]string `yaml:"exclude-node-group"` + StorageRadius *uint8 `yaml:"storage-radius"` + UploadTimeout *time.Duration `yaml:"upload-timeout"` + Deferred *bool `yaml:"deferred"` + UploadNodes *[]string `yaml:"upload-nodes"` }) if err := check.Options.Decode(checkOpts); err != nil { return nil, fmt.Errorf("decoding check %s options: %w", check.Type, err)