runDERPMapSubscriber runs a coordinator and returns if a reconnect should occur.
(ctx context.Context, tClient tailnetproto.DRPCTailnetClient24, network *tailnet.Conn)
| 1968 | |
| 1969 | // runDERPMapSubscriber runs a coordinator and returns if a reconnect should occur. |
| 1970 | func (a *agent) runDERPMapSubscriber(ctx context.Context, tClient tailnetproto.DRPCTailnetClient24, network *tailnet.Conn) (retErr error) { |
| 1971 | ctx, cancel := context.WithCancel(ctx) |
| 1972 | defer cancel() |
| 1973 | stream, err := tClient.StreamDERPMaps(ctx, &tailnetproto.StreamDERPMapsRequest{}) |
| 1974 | if err != nil { |
| 1975 | return xerrors.Errorf("stream DERP Maps: %w", err) |
| 1976 | } |
| 1977 | defer func() { |
| 1978 | cErr := stream.Close() |
| 1979 | if cErr != nil { |
| 1980 | a.logger.Debug(ctx, "error closing DERPMap stream", slog.Error(err)) |
| 1981 | } |
| 1982 | |
| 1983 | reason, initiator := classifyCoordinatorRPCExit(ctx, retErr) |
| 1984 | a.logger.Debug(ctx, "disconnected from derp map RPC", |
| 1985 | codersdk.ConnectionDirectionServerToAgent.SlogField(), |
| 1986 | reason.SlogField(), |
| 1987 | reason.SlogExpectedField(), |
| 1988 | initiator.SlogField(), |
| 1989 | slog.Error(retErr), |
| 1990 | ) |
| 1991 | }() |
| 1992 | a.logger.Info(ctx, "connected to derp map RPC") |
| 1993 | for { |
| 1994 | dmp, err := stream.Recv() |
| 1995 | if err != nil { |
| 1996 | return xerrors.Errorf("recv DERPMap error: %w", err) |
| 1997 | } |
| 1998 | dm := tailnet.DERPMapFromProto(dmp) |
| 1999 | a.client.RewriteDERPMap(dm) |
| 2000 | network.SetDERPMap(dm) |
| 2001 | } |
| 2002 | } |
| 2003 | |
| 2004 | // Collect collects additional stats from the agent |
| 2005 | func (a *agent) Collect(ctx context.Context, networkStats map[netlogtype.Connection]netlogtype.Counts) *proto.Stats { |
no test coverage detected