cloudflare/cloudflared
Publicmirrored from https://github.com/cloudflare/cloudflaredAvailable
cmd/cloudflared/tail/cmd.go
428lines · modecode
| 1 | package tail |
| 2 | |
| 3 | import ( |
| 4 | "encoding/json" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "net/http" |
| 8 | "net/url" |
| 9 | "os" |
| 10 | "os/signal" |
| 11 | "syscall" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/google/uuid" |
| 15 | "github.com/mattn/go-colorable" |
| 16 | "github.com/rs/zerolog" |
| 17 | "github.com/urfave/cli/v2" |
| 18 | "nhooyr.io/websocket" |
| 19 | |
| 20 | "github.com/cloudflare/cloudflared/cmd/cloudflared/cliutil" |
| 21 | "github.com/cloudflare/cloudflared/credentials" |
| 22 | "github.com/cloudflare/cloudflared/logger" |
| 23 | "github.com/cloudflare/cloudflared/management" |
| 24 | ) |
| 25 | |
| 26 | var ( |
| 27 | buildInfo *cliutil.BuildInfo |
| 28 | ) |
| 29 | |
| 30 | func Init(bi *cliutil.BuildInfo) { |
| 31 | buildInfo = bi |
| 32 | } |
| 33 | |
| 34 | func Command() *cli.Command { |
| 35 | subcommands := []*cli.Command{ |
| 36 | buildTailManagementTokenSubcommand(), |
| 37 | } |
| 38 | |
| 39 | return buildTailCommand(subcommands) |
| 40 | } |
| 41 | |
| 42 | func buildTailManagementTokenSubcommand() *cli.Command { |
| 43 | return &cli.Command{ |
| 44 | Name: "token", |
| 45 | Action: cliutil.ConfiguredAction(managementTokenCommand), |
| 46 | Usage: "Get management access jwt", |
| 47 | UsageText: "cloudflared tail token TUNNEL_ID", |
| 48 | Description: `Get management access jwt for a tunnel`, |
| 49 | Hidden: true, |
| 50 | } |
| 51 | } |
| 52 | |
| 53 | func managementTokenCommand(c *cli.Context) error { |
| 54 | log := createLogger(c) |
| 55 | token, err := getManagementToken(c, log) |
| 56 | if err != nil { |
| 57 | return err |
| 58 | } |
| 59 | var tokenResponse = struct { |
| 60 | Token string `json:"token"` |
| 61 | }{Token: token} |
| 62 | |
| 63 | return json.NewEncoder(os.Stdout).Encode(tokenResponse) |
| 64 | } |
| 65 | |
| 66 | func buildTailCommand(subcommands []*cli.Command) *cli.Command { |
| 67 | return &cli.Command{ |
| 68 | Name: "tail", |
| 69 | Action: Run, |
| 70 | Usage: "Stream logs from a remote cloudflared", |
| 71 | UsageText: "cloudflared tail [tail command options] [TUNNEL-ID]", |
| 72 | Flags: []cli.Flag{ |
| 73 | &cli.StringFlag{ |
| 74 | Name: "connector-id", |
| 75 | Usage: "Access a specific cloudflared instance by connector id (for when a tunnel has multiple cloudflared's)", |
| 76 | Value: "", |
| 77 | EnvVars: []string{"TUNNEL_MANAGEMENT_CONNECTOR"}, |
| 78 | }, |
| 79 | &cli.StringSliceFlag{ |
| 80 | Name: "event", |
| 81 | Usage: "Filter by specific Events (cloudflared, http, tcp, udp) otherwise, defaults to send all events", |
| 82 | EnvVars: []string{"TUNNEL_MANAGEMENT_FILTER_EVENTS"}, |
| 83 | }, |
| 84 | &cli.StringFlag{ |
| 85 | Name: "level", |
| 86 | Usage: "Filter by specific log levels (debug, info, warn, error). Filters by debug log level by default.", |
| 87 | EnvVars: []string{"TUNNEL_MANAGEMENT_FILTER_LEVEL"}, |
| 88 | Value: "debug", |
| 89 | }, |
| 90 | &cli.Float64Flag{ |
| 91 | Name: "sample", |
| 92 | Usage: "Sample log events by percentage (0.0 .. 1.0). No sampling by default.", |
| 93 | EnvVars: []string{"TUNNEL_MANAGEMENT_FILTER_SAMPLE"}, |
| 94 | Value: 1.0, |
| 95 | }, |
| 96 | &cli.StringFlag{ |
| 97 | Name: "token", |
| 98 | Usage: "Access token for a specific tunnel", |
| 99 | Value: "", |
| 100 | EnvVars: []string{"TUNNEL_MANAGEMENT_TOKEN"}, |
| 101 | }, |
| 102 | &cli.StringFlag{ |
| 103 | Name: "output", |
| 104 | Usage: "Output format for the logs (default, json)", |
| 105 | Value: "default", |
| 106 | EnvVars: []string{"TUNNEL_MANAGEMENT_OUTPUT"}, |
| 107 | }, |
| 108 | &cli.StringFlag{ |
| 109 | Name: "management-hostname", |
| 110 | Usage: "Management hostname to signify incoming management requests", |
| 111 | EnvVars: []string{"TUNNEL_MANAGEMENT_HOSTNAME"}, |
| 112 | Hidden: true, |
| 113 | Value: "management.argotunnel.com", |
| 114 | }, |
| 115 | &cli.StringFlag{ |
| 116 | Name: "trace", |
| 117 | Usage: "Set a cf-trace-id for the request", |
| 118 | Hidden: true, |
| 119 | Value: "", |
| 120 | }, |
| 121 | &cli.StringFlag{ |
| 122 | Name: logger.LogLevelFlag, |
| 123 | Value: "info", |
| 124 | Usage: "Application logging level {debug, info, warn, error, fatal}", |
| 125 | EnvVars: []string{"TUNNEL_LOGLEVEL"}, |
| 126 | }, |
| 127 | &cli.StringFlag{ |
| 128 | Name: credentials.OriginCertFlag, |
| 129 | Usage: "Path to the certificate generated for your origin when you run cloudflared login.", |
| 130 | EnvVars: []string{"TUNNEL_ORIGIN_CERT"}, |
| 131 | Value: credentials.FindDefaultOriginCertPath(), |
| 132 | }, |
| 133 | }, |
| 134 | Subcommands: subcommands, |
| 135 | } |
| 136 | } |
| 137 | |
| 138 | // Middleware validation error struct for returning to the eyeball |
| 139 | type managementError struct { |
| 140 | Code int `json:"code,omitempty"` |
| 141 | Message string `json:"message,omitempty"` |
| 142 | } |
| 143 | |
| 144 | // Middleware validation error HTTP response JSON for returning to the eyeball |
| 145 | type managementErrorResponse struct { |
| 146 | Success bool `json:"success,omitempty"` |
| 147 | Errors []managementError `json:"errors,omitempty"` |
| 148 | } |
| 149 | |
| 150 | func handleValidationError(resp *http.Response, log *zerolog.Logger) { |
| 151 | if resp.StatusCode == 530 { |
| 152 | log.Error().Msgf("no cloudflared connector available or reachable via management request (a recent version of cloudflared is required to use streaming logs)") |
| 153 | } |
| 154 | var managementErr managementErrorResponse |
| 155 | err := json.NewDecoder(resp.Body).Decode(&managementErr) |
| 156 | if err != nil { |
| 157 | log.Error().Msgf("unable to start management log streaming session: http response code returned %d", resp.StatusCode) |
| 158 | return |
| 159 | } |
| 160 | if managementErr.Success || len(managementErr.Errors) == 0 { |
| 161 | log.Error().Msgf("management tunnel validation returned success with invalid HTTP response code to convert to a WebSocket request") |
| 162 | return |
| 163 | } |
| 164 | for _, e := range managementErr.Errors { |
| 165 | log.Error().Msgf("management request failed validation: (%d) %s", e.Code, e.Message) |
| 166 | } |
| 167 | } |
| 168 | |
| 169 | // logger will be created to emit only against the os.Stderr as to not obstruct with normal output from |
| 170 | // management requests |
| 171 | func createLogger(c *cli.Context) *zerolog.Logger { |
| 172 | level, levelErr := zerolog.ParseLevel(c.String(logger.LogLevelFlag)) |
| 173 | if levelErr != nil { |
| 174 | level = zerolog.InfoLevel |
| 175 | } |
| 176 | log := zerolog.New(zerolog.ConsoleWriter{ |
| 177 | Out: colorable.NewColorable(os.Stderr), |
| 178 | TimeFormat: time.RFC3339, |
| 179 | }).With().Timestamp().Logger().Level(level) |
| 180 | return &log |
| 181 | } |
| 182 | |
| 183 | // parseFilters will attempt to parse provided filters to send to with the EventStartStreaming |
| 184 | func parseFilters(c *cli.Context) (*management.StreamingFilters, error) { |
| 185 | var level *management.LogLevel |
| 186 | var events []management.LogEventType |
| 187 | var sample float64 |
| 188 | |
| 189 | argLevel := c.String("level") |
| 190 | argEvents := c.StringSlice("event") |
| 191 | argSample := c.Float64("sample") |
| 192 | |
| 193 | if argLevel != "" { |
| 194 | l, ok := management.ParseLogLevel(argLevel) |
| 195 | if !ok { |
| 196 | return nil, fmt.Errorf("invalid --level filter provided, please use one of the following Log Levels: debug, info, warn, error") |
| 197 | } |
| 198 | level = &l |
| 199 | } |
| 200 | |
| 201 | for _, v := range argEvents { |
| 202 | t, ok := management.ParseLogEventType(v) |
| 203 | if !ok { |
| 204 | return nil, fmt.Errorf("invalid --event filter provided, please use one of the following EventTypes: cloudflared, http, tcp, udp") |
| 205 | } |
| 206 | events = append(events, t) |
| 207 | } |
| 208 | |
| 209 | if argSample <= 0.0 || argSample > 1.0 { |
| 210 | return nil, fmt.Errorf("invalid --sample value provided, please make sure it is in the range (0.0 .. 1.0)") |
| 211 | } |
| 212 | sample = argSample |
| 213 | |
| 214 | if level == nil && len(events) == 0 && argSample != 1.0 { |
| 215 | // When no filters are provided, do not return a StreamingFilters struct |
| 216 | return nil, nil |
| 217 | } |
| 218 | |
| 219 | return &management.StreamingFilters{ |
| 220 | Level: level, |
| 221 | Events: events, |
| 222 | Sampling: sample, |
| 223 | }, nil |
| 224 | } |
| 225 | |
| 226 | // getManagementToken will make a call to the Cloudflare API to acquire a management token for the requested tunnel. |
| 227 | func getManagementToken(c *cli.Context, log *zerolog.Logger) (string, error) { |
| 228 | userCreds, err := credentials.Read(c.String(credentials.OriginCertFlag), log) |
| 229 | if err != nil { |
| 230 | return "", err |
| 231 | } |
| 232 | |
| 233 | client, err := userCreds.Client(c.String("api-url"), buildInfo.UserAgent(), log) |
| 234 | if err != nil { |
| 235 | return "", err |
| 236 | } |
| 237 | |
| 238 | tunnelIDString := c.Args().First() |
| 239 | if tunnelIDString == "" { |
| 240 | return "", errors.New("no tunnel ID provided") |
| 241 | } |
| 242 | tunnelID, err := uuid.Parse(tunnelIDString) |
| 243 | if err != nil { |
| 244 | return "", errors.New("unable to parse provided tunnel id as a valid UUID") |
| 245 | } |
| 246 | |
| 247 | token, err := client.GetManagementToken(tunnelID) |
| 248 | if err != nil { |
| 249 | return "", err |
| 250 | } |
| 251 | |
| 252 | return token, nil |
| 253 | } |
| 254 | |
| 255 | // buildURL will build the management url to contain the required query parameters to authenticate the request. |
| 256 | func buildURL(c *cli.Context, log *zerolog.Logger) (url.URL, error) { |
| 257 | var err error |
| 258 | managementHostname := c.String("management-hostname") |
| 259 | token := c.String("token") |
| 260 | if token == "" { |
| 261 | token, err = getManagementToken(c, log) |
| 262 | if err != nil { |
| 263 | return url.URL{}, fmt.Errorf("unable to acquire management token for requested tunnel id: %w", err) |
| 264 | } |
| 265 | } |
| 266 | query := url.Values{} |
| 267 | query.Add("access_token", token) |
| 268 | connector := c.String("connector-id") |
| 269 | if connector != "" { |
| 270 | connectorID, err := uuid.Parse(connector) |
| 271 | if err != nil { |
| 272 | return url.URL{}, fmt.Errorf("unabled to parse 'connector-id' flag into a valid UUID: %w", err) |
| 273 | } |
| 274 | query.Add("connector_id", connectorID.String()) |
| 275 | } |
| 276 | return url.URL{Scheme: "wss", Host: managementHostname, Path: "/logs", RawQuery: query.Encode()}, nil |
| 277 | } |
| 278 | |
| 279 | func printLine(log *management.Log, logger *zerolog.Logger) { |
| 280 | fields, err := json.Marshal(log.Fields) |
| 281 | if err != nil { |
| 282 | fields = []byte("unable to parse fields") |
| 283 | logger.Debug().Msgf("unable to parse fields from event %+v", log) |
| 284 | } |
| 285 | fmt.Printf("%s %s %s %s %s\n", log.Time, log.Level, log.Event, log.Message, fields) |
| 286 | } |
| 287 | |
| 288 | func printJSON(log *management.Log, logger *zerolog.Logger) { |
| 289 | output, err := json.Marshal(log) |
| 290 | if err != nil { |
| 291 | logger.Debug().Msgf("unable to parse event to json %+v", log) |
| 292 | } else { |
| 293 | fmt.Println(string(output)) |
| 294 | } |
| 295 | } |
| 296 | |
| 297 | // Run implements a foreground runner |
| 298 | func Run(c *cli.Context) error { |
| 299 | log := createLogger(c) |
| 300 | |
| 301 | signals := make(chan os.Signal, 10) |
| 302 | signal.Notify(signals, syscall.SIGTERM, syscall.SIGINT) |
| 303 | defer signal.Stop(signals) |
| 304 | |
| 305 | output := "default" |
| 306 | switch c.String("output") { |
| 307 | case "default", "": |
| 308 | output = "default" |
| 309 | case "json": |
| 310 | output = "json" |
| 311 | default: |
| 312 | log.Err(errors.New("invalid --output value provided, please make sure it is one of: default, json")).Send() |
| 313 | } |
| 314 | |
| 315 | filters, err := parseFilters(c) |
| 316 | if err != nil { |
| 317 | log.Error().Err(err).Msgf("invalid filters provided") |
| 318 | return nil |
| 319 | } |
| 320 | |
| 321 | u, err := buildURL(c, log) |
| 322 | if err != nil { |
| 323 | log.Err(err).Msg("unable to construct management request URL") |
| 324 | return nil |
| 325 | } |
| 326 | |
| 327 | header := make(http.Header) |
| 328 | header.Add("User-Agent", buildInfo.UserAgent()) |
| 329 | trace := c.String("trace") |
| 330 | if trace != "" { |
| 331 | header["cf-trace-id"] = []string{trace} |
| 332 | } |
| 333 | ctx := c.Context |
| 334 | conn, resp, err := websocket.Dial(ctx, u.String(), &websocket.DialOptions{ |
| 335 | HTTPHeader: header, |
| 336 | }) |
| 337 | if err != nil { |
| 338 | if resp != nil && resp.StatusCode != http.StatusSwitchingProtocols { |
| 339 | handleValidationError(resp, log) |
| 340 | return nil |
| 341 | } |
| 342 | log.Error().Err(err).Msgf("unable to start management log streaming session") |
| 343 | return nil |
| 344 | } |
| 345 | defer conn.Close(websocket.StatusInternalError, "management connection was closed abruptly") |
| 346 | |
| 347 | // Once connection is established, send start_streaming event to begin receiving logs |
| 348 | err = management.WriteEvent(conn, ctx, &management.EventStartStreaming{ |
| 349 | ClientEvent: management.ClientEvent{Type: management.StartStreaming}, |
| 350 | Filters: filters, |
| 351 | }) |
| 352 | if err != nil { |
| 353 | log.Error().Err(err).Msg("unable to request logs from management tunnel") |
| 354 | return nil |
| 355 | } |
| 356 | log.Debug(). |
| 357 | Str("tunnel-id", c.Args().First()). |
| 358 | Str("connector-id", c.String("connector-id")). |
| 359 | Interface("filters", filters). |
| 360 | Msg("connected") |
| 361 | |
| 362 | readerDone := make(chan struct{}) |
| 363 | |
| 364 | go func() { |
| 365 | defer close(readerDone) |
| 366 | for { |
| 367 | select { |
| 368 | case <-ctx.Done(): |
| 369 | return |
| 370 | default: |
| 371 | event, err := management.ReadServerEvent(conn, ctx) |
| 372 | if err != nil { |
| 373 | if closeErr := management.AsClosed(err); closeErr != nil { |
| 374 | // If the client (or the server) already closed the connection, don't continue to |
| 375 | // attempt to read from the client. |
| 376 | if closeErr.Code == websocket.StatusNormalClosure { |
| 377 | return |
| 378 | } |
| 379 | // Only log abnormal closures |
| 380 | log.Error().Msgf("received remote closure: (%d) %s", closeErr.Code, closeErr.Reason) |
| 381 | return |
| 382 | } |
| 383 | log.Err(err).Msg("unable to read event from server") |
| 384 | return |
| 385 | } |
| 386 | switch event.Type { |
| 387 | case management.Logs: |
| 388 | logs, ok := management.IntoServerEvent(event, management.Logs) |
| 389 | if !ok { |
| 390 | log.Error().Msgf("invalid logs event") |
| 391 | continue |
| 392 | } |
| 393 | // Output all the logs received to stdout |
| 394 | for _, l := range logs.Logs { |
| 395 | if output == "json" { |
| 396 | printJSON(l, log) |
| 397 | } else { |
| 398 | printLine(l, log) |
| 399 | } |
| 400 | } |
| 401 | case management.UnknownServerEventType: |
| 402 | fallthrough |
| 403 | default: |
| 404 | log.Debug().Msgf("unexpected log event type: %s", event.Type) |
| 405 | } |
| 406 | } |
| 407 | } |
| 408 | }() |
| 409 | |
| 410 | for { |
| 411 | select { |
| 412 | case <-ctx.Done(): |
| 413 | return nil |
| 414 | case <-readerDone: |
| 415 | return nil |
| 416 | case <-signals: |
| 417 | log.Debug().Msg("closing management connection") |
| 418 | // Cleanly close the connection by sending a close message and then |
| 419 | // waiting (with timeout) for the server to close the connection. |
| 420 | conn.Close(websocket.StatusNormalClosure, "") |
| 421 | select { |
| 422 | case <-readerDone: |
| 423 | case <-time.After(time.Second): |
| 424 | } |
| 425 | return nil |
| 426 | } |
| 427 | } |
| 428 | } |
| 429 | |