functional/tester: clean up "broadcastOperation"
@@ -24,6 +24,7 @@ import (
"net/url"
"path/filepath"
"strings"
+ "sync"
"time"
"github.com/coreos/etcd/functional/rpcpb"
@@ -302,24 +303,41 @@ func (clus *Cluster) Restart() error {
}
func (clus *Cluster) broadcastOperation(op rpcpb.Operation) error {
+ var wg sync.WaitGroup
+ wg.Add(len(clus.agentStreams))
+
+ errc := make(chan error, len(clus.agentStreams))
for i := range clus.agentStreams {
- err := clus.sendOperation(i, op)
- if err != nil {
- if op == rpcpb.Operation_DestroyEtcdAgent &&
+ go func(idx int, o rpcpb.Operation) {
+ defer wg.Done()
+ errc <- clus.sendOperation(idx, o)
+ }(i, op)
+ }
+ wg.Wait()
+ close(errc)
+
+ errs := []string{}
+ for err := range errc {
+ if err == nil {
+ continue
+ }
+
+ if err != nil &&
+ op == rpcpb.Operation_DestroyEtcdAgent &&
strings.Contains(err.Error(), "rpc error: code = Unavailable desc = transport is closing") {
// agent server has already closed;
// so this error is expected
- clus.lg.Info(
- "successfully destroyed",
- zap.String("member", clus.Members[i].EtcdClientEndpoint),
- )
+ clus.lg.Info("successfully destroyed all")
continue
}
- return err
- }
+ errs = append(errs, err.Error())
}
+
+ if len(errs) == 0 {
return nil
}
+ return errors.New(strings.Join(errs, ", "))
+}
func (clus *Cluster) sendOperation(idx int, op rpcpb.Operation) error {
if op == rpcpb.Operation_InitialStartEtcd {
