runControllersOnce uses the provided clients to call into the controllers once. It is combined into one function so that a problem with one tears down the other and triggers a retry (if appropriate). We typically multiplex all RPCs over the same websocket, so we want them to share the same fate.
(clients ControlProtocolClients)
| 1620 | // appropriate). We typically multiplex all RPCs over the same websocket, so we want them to share |
| 1621 | // the same fate. |
| 1622 | func (c *Controller) runControllersOnce(clients ControlProtocolClients) { |
| 1623 | // clients.Closer.Close should nominally be idempotent, but let's not press our luck |
| 1624 | closeOnce := sync.Once{} |
| 1625 | closeClients := func() { |
| 1626 | closeOnce.Do(func() { |
| 1627 | closeErr := clients.Closer.Close() |
| 1628 | if closeErr != nil && |
| 1629 | !xerrors.Is(closeErr, io.EOF) && |
| 1630 | !xerrors.Is(closeErr, context.Canceled) && |
| 1631 | !xerrors.Is(closeErr, context.DeadlineExceeded) { |
| 1632 | c.logger.Error(c.ctx, "error closing tailnet clients", slog.Error(closeErr)) |
| 1633 | } |
| 1634 | }) |
| 1635 | } |
| 1636 | defer closeClients() |
| 1637 | |
| 1638 | if c.TelemetryCtrl != nil { |
| 1639 | c.TelemetryCtrl.New(clients.Telemetry) // synchronous, doesn't need a goroutine |
| 1640 | } |
| 1641 | |
| 1642 | wg := sync.WaitGroup{} |
| 1643 | |
| 1644 | if c.CoordCtrl != nil { |
| 1645 | wg.Add(1) |
| 1646 | go func() { |
| 1647 | defer wg.Done() |
| 1648 | c.coordinate(clients.Coordinator) |
| 1649 | if c.ctx.Err() == nil { |
| 1650 | // Main context is still active, but our coordination exited, due to some error. |
| 1651 | // Close down all the rest of the clients so we'll exit and retry. |
| 1652 | closeClients() |
| 1653 | } |
| 1654 | }() |
| 1655 | } |
| 1656 | if c.DERPCtrl != nil { |
| 1657 | wg.Add(1) |
| 1658 | go func() { |
| 1659 | defer wg.Done() |
| 1660 | dErr := c.derpMap(clients.DERP) |
| 1661 | if dErr != nil && c.ctx.Err() == nil { |
| 1662 | // The main context is still active, meaning that we want the tailnet data plane to stay |
| 1663 | // up, even though we hit some error getting DERP maps on the control plane. That means |
| 1664 | // we do NOT want to gracefully disconnect on the coordinate() routine. So, we'll just |
| 1665 | // close the underlying connection. This will trigger a retry of the control plane in |
| 1666 | // run(). |
| 1667 | closeClients() |
| 1668 | } |
| 1669 | }() |
| 1670 | } |
| 1671 | if c.WorkspaceUpdatesCtrl != nil { |
| 1672 | wg.Add(1) |
| 1673 | go func() { |
| 1674 | defer wg.Done() |
| 1675 | c.workspaceUpdates(clients.WorkspaceUpdates) |
| 1676 | if c.ctx.Err() == nil { |
| 1677 | // Main context is still active, but our workspace updates stream exited, due to |
| 1678 | // some error. Close down all the rest of the clients so we'll exit and retry. |
| 1679 | closeClients() |
no test coverage detected