1// Copyright 2018 The etcd Authors 2// 3// Licensed under the Apache License, Version 2.0 (the "License"); 4// you may not use this file except in compliance with the License. 5// You may obtain a copy of the License at 6// 7// http://www.apache.org/licenses/LICENSE-2.0 8// 9// Unless required by applicable law or agreed to in writing, software 10// distributed under the License is distributed on an "AS IS" BASIS, 11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12// See the License for the specific language governing permissions and 13// limitations under the License. 14 15package agent 16 17import ( 18 "errors" 19 "fmt" 20 "io/ioutil" 21 "net/url" 22 "os" 23 "os/exec" 24 "path/filepath" 25 "syscall" 26 "time" 27 28 "github.com/coreos/etcd/functional/rpcpb" 29 "github.com/coreos/etcd/pkg/fileutil" 30 "github.com/coreos/etcd/pkg/proxy" 31 32 "go.uber.org/zap" 33) 34 35// return error for system errors (e.g. fail to create files) 36// return status error in response for wrong configuration/operation (e.g. start etcd twice) 37func (srv *Server) handleTesterRequest(req *rpcpb.Request) (resp *rpcpb.Response, err error) { 38 defer func() { 39 if err == nil && req != nil { 40 srv.last = req.Operation 41 srv.lg.Info("handler success", zap.String("operation", req.Operation.String())) 42 } 43 }() 44 if req != nil { 45 srv.Member = req.Member 46 srv.Tester = req.Tester 47 } 48 49 switch req.Operation { 50 case rpcpb.Operation_INITIAL_START_ETCD: 51 return srv.handle_INITIAL_START_ETCD(req) 52 case rpcpb.Operation_RESTART_ETCD: 53 return srv.handle_RESTART_ETCD() 54 55 case rpcpb.Operation_SIGTERM_ETCD: 56 return srv.handle_SIGTERM_ETCD() 57 case rpcpb.Operation_SIGQUIT_ETCD_AND_REMOVE_DATA: 58 return srv.handle_SIGQUIT_ETCD_AND_REMOVE_DATA() 59 60 case rpcpb.Operation_SAVE_SNAPSHOT: 61 return srv.handle_SAVE_SNAPSHOT() 62 case rpcpb.Operation_RESTORE_RESTART_FROM_SNAPSHOT: 63 return srv.handle_RESTORE_RESTART_FROM_SNAPSHOT() 64 case rpcpb.Operation_RESTART_FROM_SNAPSHOT: 65 return srv.handle_RESTART_FROM_SNAPSHOT() 66 67 case rpcpb.Operation_SIGQUIT_ETCD_AND_ARCHIVE_DATA: 68 return srv.handle_SIGQUIT_ETCD_AND_ARCHIVE_DATA() 69 case rpcpb.Operation_SIGQUIT_ETCD_AND_REMOVE_DATA_AND_STOP_AGENT: 70 return srv.handle_SIGQUIT_ETCD_AND_REMOVE_DATA_AND_STOP_AGENT() 71 72 case rpcpb.Operation_BLACKHOLE_PEER_PORT_TX_RX: 73 return srv.handle_BLACKHOLE_PEER_PORT_TX_RX() 74 case rpcpb.Operation_UNBLACKHOLE_PEER_PORT_TX_RX: 75 return srv.handle_UNBLACKHOLE_PEER_PORT_TX_RX() 76 case rpcpb.Operation_DELAY_PEER_PORT_TX_RX: 77 return srv.handle_DELAY_PEER_PORT_TX_RX() 78 case rpcpb.Operation_UNDELAY_PEER_PORT_TX_RX: 79 return srv.handle_UNDELAY_PEER_PORT_TX_RX() 80 81 default: 82 msg := fmt.Sprintf("operation not found (%v)", req.Operation) 83 return &rpcpb.Response{Success: false, Status: msg}, errors.New(msg) 84 } 85} 86 87func (srv *Server) handle_INITIAL_START_ETCD(req *rpcpb.Request) (*rpcpb.Response, error) { 88 if srv.last != rpcpb.Operation_NOT_STARTED { 89 return &rpcpb.Response{ 90 Success: false, 91 Status: fmt.Sprintf("%q is not valid; last server operation was %q", rpcpb.Operation_INITIAL_START_ETCD.String(), srv.last.String()), 92 Member: req.Member, 93 }, nil 94 } 95 96 err := fileutil.TouchDirAll(srv.Member.BaseDir) 97 if err != nil { 98 return nil, err 99 } 100 srv.lg.Info("created base directory", zap.String("path", srv.Member.BaseDir)) 101 102 if err = srv.createEtcdLogFile(); err != nil { 103 return nil, err 104 } 105 106 srv.creatEtcdCmd(false) 107 108 if err = srv.saveTLSAssets(); err != nil { 109 return nil, err 110 } 111 if err = srv.startEtcdCmd(); err != nil { 112 return nil, err 113 } 114 srv.lg.Info("started etcd", zap.String("command-path", srv.etcdCmd.Path)) 115 if err = srv.loadAutoTLSAssets(); err != nil { 116 return nil, err 117 } 118 119 // wait some time for etcd listener start 120 // before setting up proxy 121 time.Sleep(time.Second) 122 if err = srv.startProxy(); err != nil { 123 return nil, err 124 } 125 126 return &rpcpb.Response{ 127 Success: true, 128 Status: "start etcd PASS", 129 Member: srv.Member, 130 }, nil 131} 132 133func (srv *Server) startProxy() error { 134 if srv.Member.EtcdClientProxy { 135 advertiseClientURL, advertiseClientURLPort, err := getURLAndPort(srv.Member.Etcd.AdvertiseClientURLs[0]) 136 if err != nil { 137 return err 138 } 139 listenClientURL, _, err := getURLAndPort(srv.Member.Etcd.ListenClientURLs[0]) 140 if err != nil { 141 return err 142 } 143 144 srv.advertiseClientPortToProxy[advertiseClientURLPort] = proxy.NewServer(proxy.ServerConfig{ 145 Logger: srv.lg, 146 From: *advertiseClientURL, 147 To: *listenClientURL, 148 }) 149 select { 150 case err = <-srv.advertiseClientPortToProxy[advertiseClientURLPort].Error(): 151 return err 152 case <-time.After(2 * time.Second): 153 srv.lg.Info("started proxy on client traffic", zap.String("url", advertiseClientURL.String())) 154 } 155 } 156 157 if srv.Member.EtcdPeerProxy { 158 advertisePeerURL, advertisePeerURLPort, err := getURLAndPort(srv.Member.Etcd.AdvertisePeerURLs[0]) 159 if err != nil { 160 return err 161 } 162 listenPeerURL, _, err := getURLAndPort(srv.Member.Etcd.ListenPeerURLs[0]) 163 if err != nil { 164 return err 165 } 166 167 srv.advertisePeerPortToProxy[advertisePeerURLPort] = proxy.NewServer(proxy.ServerConfig{ 168 Logger: srv.lg, 169 From: *advertisePeerURL, 170 To: *listenPeerURL, 171 }) 172 select { 173 case err = <-srv.advertisePeerPortToProxy[advertisePeerURLPort].Error(): 174 return err 175 case <-time.After(2 * time.Second): 176 srv.lg.Info("started proxy on peer traffic", zap.String("url", advertisePeerURL.String())) 177 } 178 } 179 return nil 180} 181 182func (srv *Server) stopProxy() { 183 if srv.Member.EtcdClientProxy && len(srv.advertiseClientPortToProxy) > 0 { 184 for port, px := range srv.advertiseClientPortToProxy { 185 if err := px.Close(); err != nil { 186 srv.lg.Warn("failed to close proxy", zap.Int("port", port)) 187 continue 188 } 189 select { 190 case <-px.Done(): 191 // enough time to release port 192 time.Sleep(time.Second) 193 case <-time.After(time.Second): 194 } 195 srv.lg.Info("closed proxy", 196 zap.Int("port", port), 197 zap.String("from", px.From()), 198 zap.String("to", px.To()), 199 ) 200 } 201 srv.advertiseClientPortToProxy = make(map[int]proxy.Server) 202 } 203 if srv.Member.EtcdPeerProxy && len(srv.advertisePeerPortToProxy) > 0 { 204 for port, px := range srv.advertisePeerPortToProxy { 205 if err := px.Close(); err != nil { 206 srv.lg.Warn("failed to close proxy", zap.Int("port", port)) 207 continue 208 } 209 select { 210 case <-px.Done(): 211 // enough time to release port 212 time.Sleep(time.Second) 213 case <-time.After(time.Second): 214 } 215 srv.lg.Info("closed proxy", 216 zap.Int("port", port), 217 zap.String("from", px.From()), 218 zap.String("to", px.To()), 219 ) 220 } 221 srv.advertisePeerPortToProxy = make(map[int]proxy.Server) 222 } 223} 224 225func (srv *Server) createEtcdLogFile() error { 226 var err error 227 srv.etcdLogFile, err = os.Create(srv.Member.EtcdLogPath) 228 if err != nil { 229 return err 230 } 231 srv.lg.Info("created etcd log file", zap.String("path", srv.Member.EtcdLogPath)) 232 return nil 233} 234 235func (srv *Server) creatEtcdCmd(fromSnapshot bool) { 236 etcdPath, etcdFlags := srv.Member.EtcdExecPath, srv.Member.Etcd.Flags() 237 if fromSnapshot { 238 etcdFlags = srv.Member.EtcdOnSnapshotRestore.Flags() 239 } 240 u, _ := url.Parse(srv.Member.FailpointHTTPAddr) 241 srv.lg.Info("creating etcd command", 242 zap.String("etcd-exec-path", etcdPath), 243 zap.Strings("etcd-flags", etcdFlags), 244 zap.String("failpoint-http-addr", srv.Member.FailpointHTTPAddr), 245 zap.String("failpoint-addr", u.Host), 246 ) 247 srv.etcdCmd = exec.Command(etcdPath, etcdFlags...) 248 srv.etcdCmd.Env = []string{"GOFAIL_HTTP=" + u.Host} 249 srv.etcdCmd.Stdout = srv.etcdLogFile 250 srv.etcdCmd.Stderr = srv.etcdLogFile 251} 252 253// if started with manual TLS, stores TLS assets 254// from tester/client to disk before starting etcd process 255func (srv *Server) saveTLSAssets() error { 256 if srv.Member.PeerCertPath != "" { 257 if srv.Member.PeerCertData == "" { 258 return fmt.Errorf("got empty data for %q", srv.Member.PeerCertPath) 259 } 260 if err := ioutil.WriteFile(srv.Member.PeerCertPath, []byte(srv.Member.PeerCertData), 0644); err != nil { 261 return err 262 } 263 } 264 if srv.Member.PeerKeyPath != "" { 265 if srv.Member.PeerKeyData == "" { 266 return fmt.Errorf("got empty data for %q", srv.Member.PeerKeyPath) 267 } 268 if err := ioutil.WriteFile(srv.Member.PeerKeyPath, []byte(srv.Member.PeerKeyData), 0644); err != nil { 269 return err 270 } 271 } 272 if srv.Member.PeerTrustedCAPath != "" { 273 if srv.Member.PeerTrustedCAData == "" { 274 return fmt.Errorf("got empty data for %q", srv.Member.PeerTrustedCAPath) 275 } 276 if err := ioutil.WriteFile(srv.Member.PeerTrustedCAPath, []byte(srv.Member.PeerTrustedCAData), 0644); err != nil { 277 return err 278 } 279 } 280 if srv.Member.PeerCertPath != "" && 281 srv.Member.PeerKeyPath != "" && 282 srv.Member.PeerTrustedCAPath != "" { 283 srv.lg.Info( 284 "wrote", 285 zap.String("peer-cert", srv.Member.PeerCertPath), 286 zap.String("peer-key", srv.Member.PeerKeyPath), 287 zap.String("peer-trusted-ca", srv.Member.PeerTrustedCAPath), 288 ) 289 } 290 291 if srv.Member.ClientCertPath != "" { 292 if srv.Member.ClientCertData == "" { 293 return fmt.Errorf("got empty data for %q", srv.Member.ClientCertPath) 294 } 295 if err := ioutil.WriteFile(srv.Member.ClientCertPath, []byte(srv.Member.ClientCertData), 0644); err != nil { 296 return err 297 } 298 } 299 if srv.Member.ClientKeyPath != "" { 300 if srv.Member.ClientKeyData == "" { 301 return fmt.Errorf("got empty data for %q", srv.Member.ClientKeyPath) 302 } 303 if err := ioutil.WriteFile(srv.Member.ClientKeyPath, []byte(srv.Member.ClientKeyData), 0644); err != nil { 304 return err 305 } 306 } 307 if srv.Member.ClientTrustedCAPath != "" { 308 if srv.Member.ClientTrustedCAData == "" { 309 return fmt.Errorf("got empty data for %q", srv.Member.ClientTrustedCAPath) 310 } 311 if err := ioutil.WriteFile(srv.Member.ClientTrustedCAPath, []byte(srv.Member.ClientTrustedCAData), 0644); err != nil { 312 return err 313 } 314 } 315 if srv.Member.ClientCertPath != "" && 316 srv.Member.ClientKeyPath != "" && 317 srv.Member.ClientTrustedCAPath != "" { 318 srv.lg.Info( 319 "wrote", 320 zap.String("client-cert", srv.Member.ClientCertPath), 321 zap.String("client-key", srv.Member.ClientKeyPath), 322 zap.String("client-trusted-ca", srv.Member.ClientTrustedCAPath), 323 ) 324 } 325 326 return nil 327} 328 329func (srv *Server) loadAutoTLSAssets() error { 330 if srv.Member.Etcd.PeerAutoTLS { 331 // in case of slow disk 332 time.Sleep(time.Second) 333 334 fdir := filepath.Join(srv.Member.Etcd.DataDir, "fixtures", "peer") 335 336 srv.lg.Info( 337 "loading client auto TLS assets", 338 zap.String("dir", fdir), 339 zap.String("endpoint", srv.EtcdClientEndpoint), 340 ) 341 342 certPath := filepath.Join(fdir, "cert.pem") 343 if !fileutil.Exist(certPath) { 344 return fmt.Errorf("cannot find %q", certPath) 345 } 346 certData, err := ioutil.ReadFile(certPath) 347 if err != nil { 348 return fmt.Errorf("cannot read %q (%v)", certPath, err) 349 } 350 srv.Member.PeerCertData = string(certData) 351 352 keyPath := filepath.Join(fdir, "key.pem") 353 if !fileutil.Exist(keyPath) { 354 return fmt.Errorf("cannot find %q", keyPath) 355 } 356 keyData, err := ioutil.ReadFile(keyPath) 357 if err != nil { 358 return fmt.Errorf("cannot read %q (%v)", keyPath, err) 359 } 360 srv.Member.PeerKeyData = string(keyData) 361 362 srv.lg.Info( 363 "loaded peer auto TLS assets", 364 zap.String("peer-cert-path", certPath), 365 zap.Int("peer-cert-length", len(certData)), 366 zap.String("peer-key-path", keyPath), 367 zap.Int("peer-key-length", len(keyData)), 368 ) 369 } 370 371 if srv.Member.Etcd.ClientAutoTLS { 372 // in case of slow disk 373 time.Sleep(time.Second) 374 375 fdir := filepath.Join(srv.Member.Etcd.DataDir, "fixtures", "client") 376 377 srv.lg.Info( 378 "loading client TLS assets", 379 zap.String("dir", fdir), 380 zap.String("endpoint", srv.EtcdClientEndpoint), 381 ) 382 383 certPath := filepath.Join(fdir, "cert.pem") 384 if !fileutil.Exist(certPath) { 385 return fmt.Errorf("cannot find %q", certPath) 386 } 387 certData, err := ioutil.ReadFile(certPath) 388 if err != nil { 389 return fmt.Errorf("cannot read %q (%v)", certPath, err) 390 } 391 srv.Member.ClientCertData = string(certData) 392 393 keyPath := filepath.Join(fdir, "key.pem") 394 if !fileutil.Exist(keyPath) { 395 return fmt.Errorf("cannot find %q", keyPath) 396 } 397 keyData, err := ioutil.ReadFile(keyPath) 398 if err != nil { 399 return fmt.Errorf("cannot read %q (%v)", keyPath, err) 400 } 401 srv.Member.ClientKeyData = string(keyData) 402 403 srv.lg.Info( 404 "loaded client TLS assets", 405 zap.String("peer-cert-path", certPath), 406 zap.Int("peer-cert-length", len(certData)), 407 zap.String("peer-key-path", keyPath), 408 zap.Int("peer-key-length", len(keyData)), 409 ) 410 } 411 412 return nil 413} 414 415// start but do not wait for it to complete 416func (srv *Server) startEtcdCmd() error { 417 return srv.etcdCmd.Start() 418} 419 420func (srv *Server) handle_RESTART_ETCD() (*rpcpb.Response, error) { 421 var err error 422 if !fileutil.Exist(srv.Member.BaseDir) { 423 err = fileutil.TouchDirAll(srv.Member.BaseDir) 424 if err != nil { 425 return nil, err 426 } 427 } 428 429 srv.creatEtcdCmd(false) 430 431 if err = srv.saveTLSAssets(); err != nil { 432 return nil, err 433 } 434 if err = srv.startEtcdCmd(); err != nil { 435 return nil, err 436 } 437 srv.lg.Info("restarted etcd", zap.String("command-path", srv.etcdCmd.Path)) 438 if err = srv.loadAutoTLSAssets(); err != nil { 439 return nil, err 440 } 441 442 // wait some time for etcd listener start 443 // before setting up proxy 444 // TODO: local tests should handle port conflicts 445 // with clients on restart 446 time.Sleep(time.Second) 447 if err = srv.startProxy(); err != nil { 448 return nil, err 449 } 450 451 return &rpcpb.Response{ 452 Success: true, 453 Status: "restart etcd PASS", 454 Member: srv.Member, 455 }, nil 456} 457 458func (srv *Server) handle_SIGTERM_ETCD() (*rpcpb.Response, error) { 459 srv.stopProxy() 460 461 err := stopWithSig(srv.etcdCmd, syscall.SIGTERM) 462 if err != nil { 463 return nil, err 464 } 465 srv.lg.Info("killed etcd", zap.String("signal", syscall.SIGTERM.String())) 466 467 return &rpcpb.Response{ 468 Success: true, 469 Status: "killed etcd", 470 }, nil 471} 472 473func (srv *Server) handle_SIGQUIT_ETCD_AND_REMOVE_DATA() (*rpcpb.Response, error) { 474 srv.stopProxy() 475 476 err := stopWithSig(srv.etcdCmd, syscall.SIGQUIT) 477 if err != nil { 478 return nil, err 479 } 480 srv.lg.Info("killed etcd", zap.String("signal", syscall.SIGQUIT.String())) 481 482 srv.etcdLogFile.Sync() 483 srv.etcdLogFile.Close() 484 485 // for debugging purposes, rename instead of removing 486 if err = os.RemoveAll(srv.Member.BaseDir + ".backup"); err != nil { 487 return nil, err 488 } 489 if err = os.Rename(srv.Member.BaseDir, srv.Member.BaseDir+".backup"); err != nil { 490 return nil, err 491 } 492 srv.lg.Info( 493 "renamed", 494 zap.String("base-dir", srv.Member.BaseDir), 495 zap.String("new-dir", srv.Member.BaseDir+".backup"), 496 ) 497 498 // create a new log file for next new member restart 499 if !fileutil.Exist(srv.Member.BaseDir) { 500 err = fileutil.TouchDirAll(srv.Member.BaseDir) 501 if err != nil { 502 return nil, err 503 } 504 } 505 if err = srv.createEtcdLogFile(); err != nil { 506 return nil, err 507 } 508 509 return &rpcpb.Response{ 510 Success: true, 511 Status: "killed etcd and removed base directory", 512 }, nil 513} 514 515func (srv *Server) handle_SAVE_SNAPSHOT() (*rpcpb.Response, error) { 516 err := srv.Member.SaveSnapshot(srv.lg) 517 if err != nil { 518 return nil, err 519 } 520 return &rpcpb.Response{ 521 Success: true, 522 Status: "saved snapshot", 523 SnapshotInfo: srv.Member.SnapshotInfo, 524 }, nil 525} 526 527func (srv *Server) handle_RESTORE_RESTART_FROM_SNAPSHOT() (resp *rpcpb.Response, err error) { 528 err = srv.Member.RestoreSnapshot(srv.lg) 529 if err != nil { 530 return nil, err 531 } 532 resp, err = srv.handle_RESTART_FROM_SNAPSHOT() 533 if resp != nil && err == nil { 534 resp.Status = "restored snapshot and " + resp.Status 535 } 536 return resp, err 537} 538 539func (srv *Server) handle_RESTART_FROM_SNAPSHOT() (resp *rpcpb.Response, err error) { 540 srv.creatEtcdCmd(true) 541 542 if err = srv.saveTLSAssets(); err != nil { 543 return nil, err 544 } 545 if err = srv.startEtcdCmd(); err != nil { 546 return nil, err 547 } 548 srv.lg.Info("restarted etcd", zap.String("command-path", srv.etcdCmd.Path)) 549 if err = srv.loadAutoTLSAssets(); err != nil { 550 return nil, err 551 } 552 553 // wait some time for etcd listener start 554 // before setting up proxy 555 // TODO: local tests should handle port conflicts 556 // with clients on restart 557 time.Sleep(time.Second) 558 if err = srv.startProxy(); err != nil { 559 return nil, err 560 } 561 562 return &rpcpb.Response{ 563 Success: true, 564 Status: "restarted etcd from snapshot", 565 SnapshotInfo: srv.Member.SnapshotInfo, 566 }, nil 567} 568 569func (srv *Server) handle_SIGQUIT_ETCD_AND_ARCHIVE_DATA() (*rpcpb.Response, error) { 570 srv.stopProxy() 571 572 // exit with stackstrace 573 err := stopWithSig(srv.etcdCmd, syscall.SIGQUIT) 574 if err != nil { 575 return nil, err 576 } 577 srv.lg.Info("killed etcd", zap.String("signal", syscall.SIGQUIT.String())) 578 579 srv.etcdLogFile.Sync() 580 srv.etcdLogFile.Close() 581 582 // TODO: support separate WAL directory 583 if err = archive( 584 srv.Member.BaseDir, 585 srv.Member.EtcdLogPath, 586 srv.Member.Etcd.DataDir, 587 ); err != nil { 588 return nil, err 589 } 590 srv.lg.Info("archived data", zap.String("base-dir", srv.Member.BaseDir)) 591 592 if err = srv.createEtcdLogFile(); err != nil { 593 return nil, err 594 } 595 596 srv.lg.Info("cleaning up page cache") 597 if err := cleanPageCache(); err != nil { 598 srv.lg.Warn("failed to clean up page cache", zap.String("error", err.Error())) 599 } 600 srv.lg.Info("cleaned up page cache") 601 602 return &rpcpb.Response{ 603 Success: true, 604 Status: "cleaned up etcd", 605 }, nil 606} 607 608// stop proxy, etcd, delete data directory 609func (srv *Server) handle_SIGQUIT_ETCD_AND_REMOVE_DATA_AND_STOP_AGENT() (*rpcpb.Response, error) { 610 srv.stopProxy() 611 612 err := stopWithSig(srv.etcdCmd, syscall.SIGQUIT) 613 if err != nil { 614 return nil, err 615 } 616 srv.lg.Info("killed etcd", zap.String("signal", syscall.SIGQUIT.String())) 617 618 srv.etcdLogFile.Sync() 619 srv.etcdLogFile.Close() 620 621 err = os.RemoveAll(srv.Member.BaseDir) 622 if err != nil { 623 return nil, err 624 } 625 srv.lg.Info("removed base directory", zap.String("dir", srv.Member.BaseDir)) 626 627 // stop agent server 628 srv.Stop() 629 630 return &rpcpb.Response{ 631 Success: true, 632 Status: "destroyed etcd and agent", 633 }, nil 634} 635 636func (srv *Server) handle_BLACKHOLE_PEER_PORT_TX_RX() (*rpcpb.Response, error) { 637 for port, px := range srv.advertisePeerPortToProxy { 638 srv.lg.Info("blackholing", zap.Int("peer-port", port)) 639 px.BlackholeTx() 640 px.BlackholeRx() 641 srv.lg.Info("blackholed", zap.Int("peer-port", port)) 642 } 643 return &rpcpb.Response{ 644 Success: true, 645 Status: "blackholed peer port tx/rx", 646 }, nil 647} 648 649func (srv *Server) handle_UNBLACKHOLE_PEER_PORT_TX_RX() (*rpcpb.Response, error) { 650 for port, px := range srv.advertisePeerPortToProxy { 651 srv.lg.Info("unblackholing", zap.Int("peer-port", port)) 652 px.UnblackholeTx() 653 px.UnblackholeRx() 654 srv.lg.Info("unblackholed", zap.Int("peer-port", port)) 655 } 656 return &rpcpb.Response{ 657 Success: true, 658 Status: "unblackholed peer port tx/rx", 659 }, nil 660} 661 662func (srv *Server) handle_DELAY_PEER_PORT_TX_RX() (*rpcpb.Response, error) { 663 lat := time.Duration(srv.Tester.UpdatedDelayLatencyMs) * time.Millisecond 664 rv := time.Duration(srv.Tester.DelayLatencyMsRv) * time.Millisecond 665 666 for port, px := range srv.advertisePeerPortToProxy { 667 srv.lg.Info("delaying", 668 zap.Int("peer-port", port), 669 zap.Duration("latency", lat), 670 zap.Duration("random-variable", rv), 671 ) 672 px.DelayTx(lat, rv) 673 px.DelayRx(lat, rv) 674 srv.lg.Info("delayed", 675 zap.Int("peer-port", port), 676 zap.Duration("latency", lat), 677 zap.Duration("random-variable", rv), 678 ) 679 } 680 681 return &rpcpb.Response{ 682 Success: true, 683 Status: "delayed peer port tx/rx", 684 }, nil 685} 686 687func (srv *Server) handle_UNDELAY_PEER_PORT_TX_RX() (*rpcpb.Response, error) { 688 for port, px := range srv.advertisePeerPortToProxy { 689 srv.lg.Info("undelaying", zap.Int("peer-port", port)) 690 px.UndelayTx() 691 px.UndelayRx() 692 srv.lg.Info("undelayed", zap.Int("peer-port", port)) 693 } 694 return &rpcpb.Response{ 695 Success: true, 696 Status: "undelayed peer port tx/rx", 697 }, nil 698} 699