❯ go-clikit
v0.10.0 pkg.go.dev ↗

guides › work on many hosts

Work on many hosts

Run a function on each of many targets, a bounded number at a time, with every target reported, and the failures summed up.

A command for a cluster works on hundreds of nodes. fanout.Map runs a function on each, a bounded number at a time. It reports a step with a target for each item, folded, so that the tree shows node sets.

reinstall exe[01-12] --progress=tty 
✓ fetching the base image  5.0s  2.0 GiB at 410 MiB/s
reinstall ◦ 0:10.5
╎ reinstalling the nodes  ◉◉◉◉◉◌◌◌◌◌ 6/12 ◦ 4 running ◦ 2 queued
  ╎ ◡ exe07  2.8s/30s  POST /redfish/v1/Systems/1
  ╎ ◡ exe08  2.1s/30s  POST /redfish/v1/Systems/1
  ╎ ◡ exe09  1.4s/30s  POST /redfish/v1/Systems/1
  ╎ ◡ exe10  0.7s/30s  POST /redfish/v1/Systems/1
  ╎ ✓ exe[01-06]
Twelve nodes, four at a time. The ended targets fold into a node set.

Procedure#

  1. Give the step a name and a Limit in fanout.MapOptions.
  2. Give Describe when a display should name an item by its node, its host or its role.
  3. Write the function for one item. Return its result and its error.
  4. Read the outcomes in the order of the items. Each has the value and the error of its item.
  5. Return the error of Map. By default it is a fanout.FailureError, which names the items that failed.
capture := &progresstest.Capture{}
bus := progress.NewBus(progress.BusOptions{Sinks: []progress.Sink{capture}})
ctx := progress.WithBus(context.Background(), bus)

nodes := []string{"exe01", "exe02", "exe03", "exe04"}
outcomes, err := fanout.Map(ctx, nodes, fanout.MapOptions[string]{Step: "uptime", Limit: 2},
	func(_ context.Context, node string) (string, error) {
		if node == "exe03" {
			return "", fmt.Errorf("%s: no answer", node)
		}
		return strings.ToUpper(node), nil
	})
for i, o := range outcomes {
	fmt.Println(nodes[i], o.Value, o.Err)
}
fmt.Println(err)
bus.Close()
fmt.Print(capture.Tree())
exe01 EXE01 <nil>
exe02 EXE02 <nil>
exe03  exe03: no answer
exe04 EXE04 <nil>
1 of 4 failed: exe03
step uptime total=4 limit=2 [fold]: failed (target): 1 of 4 failed: exe03
  target exe03: failed (target): {}: no answer
  target exe[01-02,04]: ok
ExampleMap in fanout/example_test.go. CI runs it and checks its output.

More than a limit#

NeedUse
No more than two calls at a time to each gateway, whatever the limitfanout.Keyed, and MapOptions.Acquire
Batches, one after the other, with a pause; stop at the first that failsfanout.Batches
Leave an item out on purposeprogress.Skip: its target ends skipped
Your own exit code for a failed stepMapOptions.Summarize, which builds on FailureError
Bytes or items for each targetItemInfo.Unit and .Size, and progress.AdvanceWriter
Prove the limit in a testfanouttest.InFlight
gateways := []string{"gw1.example.org", "gw2.example.org"}
keys := &fanout.Keyed{Limit: 2}
var mu sync.Mutex
now, peak := map[string]int{}, map[string]int{}

fanout.Each(context.Background(), 8, 8, func(i int) {
	gateway := gateways[i%len(gateways)]
	release, err := keys.Acquire(context.Background(), gateway)
	if err != nil {
		return
	}
	defer release()

	mu.Lock()
	now[gateway]++
	peak[gateway] = max(peak[gateway], now[gateway])
	mu.Unlock()
	time.Sleep(time.Millisecond)
	mu.Lock()
	now[gateway]--
	mu.Unlock()
})
for _, gateway := range gateways {
	fmt.Printf("%s: at most 2 at once: %t\n", gateway, peak[gateway] <= 2)
}
gw1.example.org: at most 2 at once: true
gw2.example.org: at most 2 at once: true
ExampleKeyed in fanout/example_test.go. CI runs it and checks its output.
CAUTION

A panic in the function of an item does not end the program. Map recovers it, ends the target as failed with a fanout.PanicError, and writes the stack to the PanicLog.

Prove the limit#

A test that only counts calls does not prove that four ran at once. fanouttest.InFlight holds each call until the number you expect is under way together.

fake := &uptime{}
fake.calls.Hold = 3
fake.calls.Patience = time.Minute

fanout.Each(context.Background(), 3, 3, func(i int) {
	fake.Ask(fmt.Sprintf("exe%02d", i+1))
})
fmt.Println(fake.calls.Peak(), fake.calls.Started())
3 3
ExampleInFlight in fanout/fanouttest/example_test.go. CI runs it and checks its output.