diff --git a/CHANGELOG.md b/CHANGELOG.md index e2d5c1148..2804368eb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ ### Enhancements: - feat(auth): add `auth revoke` subcommand for revoking API tokens via `--current`, `--name`, `--token-value`, `--id`, or `--file` (bulk) [#1717](https://github.com/fastly/cli/pull/1717) +- feat(service/logging/debug): add support for logging endpoint error streaming via the `service logging debug` subcommand [#1721](https://github.com/fastly/cli/pull/1721) - feat(stats): accept `--json` / `-j` as an alias for `--format=json` on all stats and help subcommands, matching the flag style used by the rest of the CLI [#1719](https://github.com/fastly/cli/pull/1719) ### Dependencies: diff --git a/pkg/api/interface.go b/pkg/api/interface.go index 56e55af60..e7741dbaa 100644 --- a/pkg/api/interface.go +++ b/pkg/api/interface.go @@ -250,6 +250,7 @@ type Interface interface { GetOriginMetricsForServiceJSON(context.Context, *fastly.GetOriginMetricsInput, any) error CreateManagedLogging(context.Context, *fastly.CreateManagedLoggingInput) (*fastly.ManagedLogging, error) + GetLoggingEndpointErrors(context.Context, *fastly.LoggingEndpointErrorsInput) (*fastly.LoggingEndpointErrorsResponse, error) GetGeneratedVCL(context.Context, *fastly.GetGeneratedVCLInput) (*fastly.VCL, error) diff --git a/pkg/commands/commands.go b/pkg/commands/commands.go index f683d11fc..6fff7917e 100644 --- a/pkg/commands/commands.go +++ b/pkg/commands/commands.go @@ -121,6 +121,7 @@ import ( serviceloggingbigquery "github.com/fastly/cli/pkg/commands/service/logging/bigquery" serviceloggingcloudfiles "github.com/fastly/cli/pkg/commands/service/logging/cloudfiles" serviceloggingdatadog "github.com/fastly/cli/pkg/commands/service/logging/datadog" + serviceloggingdebug "github.com/fastly/cli/pkg/commands/service/logging/debug" serviceloggingdigitalocean "github.com/fastly/cli/pkg/commands/service/logging/digitalocean" serviceloggingelasticsearch "github.com/fastly/cli/pkg/commands/service/logging/elasticsearch" serviceloggingftp "github.com/fastly/cli/pkg/commands/service/logging/ftp" @@ -550,6 +551,7 @@ func Define( // nolint:revive // function-length servicevclSnippetList := servicevclsnippet.NewListCommand(servicevclSnippetCmdRoot.CmdClause, data) servicevclSnippetUpdate := servicevclsnippet.NewUpdateCommand(servicevclSnippetCmdRoot.CmdClause, data) serviceloggingCmdRoot := servicelogging.NewRootCommand(serviceCmdRoot.CmdClause, data) + serviceloggingDebugCmd := serviceloggingdebug.NewDebugCommand(serviceloggingCmdRoot.CmdClause, data) serviceloggingAzureblobCmdRoot := serviceloggingazureblob.NewRootCommand(serviceloggingCmdRoot.CmdClause, data) serviceloggingAzureblobCreate := serviceloggingazureblob.NewCreateCommand(serviceloggingAzureblobCmdRoot.CmdClause, data) serviceloggingAzureblobDelete := serviceloggingazureblob.NewDeleteCommand(serviceloggingAzureblobCmdRoot.CmdClause, data) @@ -1172,6 +1174,7 @@ func Define( // nolint:revive // function-length kvstoreentryDescribe, kvstoreentryList, logtailCmdRoot, + serviceloggingDebugCmd, serviceloggingAzureblobCmdRoot, serviceloggingAzureblobCreate, serviceloggingAzureblobDelete, diff --git a/pkg/commands/service/logging/debug/debug_test.go b/pkg/commands/service/logging/debug/debug_test.go new file mode 100644 index 000000000..109159f52 --- /dev/null +++ b/pkg/commands/service/logging/debug/debug_test.go @@ -0,0 +1,34 @@ +package debug + +import ( + "encoding/json" + "testing" + + "github.com/google/go-cmp/cmp" + + "github.com/fastly/go-fastly/v14/fastly" +) + +// TestParseLoggingError validates we're correctly decoding individual logging error JSON. +func TestParseLoggingError(t *testing.T) { + data := []byte(`{"sequence_number":1,"error_time_us":1601645172164,"stream":"logging_error","message":"Failed to send log","endpoint":"my-s3-endpoint","details":"connection refused"}`) + + var got fastly.LoggingEndpointError + err := json.Unmarshal(data, &got) + if err != nil { + t.Fatalf("error parsing response data: %v", err) + } + + want := fastly.LoggingEndpointError{ + SequenceNumber: 1, + Timestamp: 1601645172164, + Stream: "logging_error", + Message: "Failed to send log", + Endpoint: "my-s3-endpoint", + Details: "connection refused", + } + + if diff := cmp.Diff(want, got); diff != "" { + t.Errorf("JSON unmarshal mismatch (-want +got):\n%s", diff) + } +} diff --git a/pkg/commands/service/logging/debug/doc.go b/pkg/commands/service/logging/debug/doc.go new file mode 100644 index 000000000..ef7391a08 --- /dev/null +++ b/pkg/commands/service/logging/debug/doc.go @@ -0,0 +1,3 @@ +// Package debug contains the command to stream live logging endpoint errors, +// providing visibility into logging pipeline issues for troubleshooting and resolution. +package debug diff --git a/pkg/commands/service/logging/debug/root.go b/pkg/commands/service/logging/debug/root.go new file mode 100644 index 000000000..d57a47221 --- /dev/null +++ b/pkg/commands/service/logging/debug/root.go @@ -0,0 +1,252 @@ +package debug + +import ( + "context" + "encoding/json" + "fmt" + "io" + "os" + "os/signal" + "strconv" + "strings" + "syscall" + "time" + + "github.com/fastly/go-fastly/v14/fastly" + + "github.com/fastly/cli/pkg/argparser" + "github.com/fastly/cli/pkg/global" + "github.com/fastly/cli/pkg/text" +) + +// batch wraps errors for sending to the output loop. +type batch struct { + Errors []fastly.LoggingEndpointError +} + +// Command is the command for streaming logging endpoint errors. +type Command struct { + argparser.Base + + serviceName argparser.OptionalServiceNameID + serviceID string + from uint64 + to uint64 + filter string + printTimestamps bool + jsonOutput bool + batchCh chan batch + dieCh chan struct{} + doneCh chan struct{} +} + +// CommandName is the string to be used to invoke this command. +const CommandName = "debug" + +// NewDebugCommand returns a new command registered in the parent. +func NewDebugCommand(parent argparser.Registerer, g *global.Data) *Command { + var c Command + c.Globals = g + c.CmdClause = parent.Command(CommandName, "Stream live logging endpoint errors") + c.RegisterFlag(argparser.StringFlagOpts{ + Name: argparser.FlagServiceIDName, + Description: argparser.FlagServiceIDDesc, + Dst: &g.Manifest.Flag.ServiceID, + Short: 's', + }) + c.RegisterFlag(argparser.StringFlagOpts{ + Action: c.serviceName.Set, + Name: argparser.FlagServiceName, + Description: argparser.FlagServiceNameDesc, + Dst: &c.serviceName.Value, + }) + c.CmdClause.Flag("from", "From time, in Unix seconds").Uint64Var(&c.from) + c.CmdClause.Flag("to", "To time, in Unix seconds").Uint64Var(&c.to) + c.CmdClause.Flag("filter", "Filter errors by logging endpoint name").StringVar(&c.filter) + c.CmdClause.Flag("timestamps", "Print full timestamps instead of compact time").BoolVar(&c.printTimestamps) + c.CmdClause.Flag("json", "Output error stream as JSON").BoolVar(&c.jsonOutput) + return &c +} + +// Exec implements the command interface. +func (c *Command) Exec(_ io.Reader, out io.Writer) error { + serviceID, source, flag, err := argparser.ServiceID(c.serviceName, *c.Globals.Manifest, c.Globals.APIClient, c.Globals.ErrLog) + if err != nil { + return err + } + if c.Globals.Verbose() { + argparser.DisplayServiceID(serviceID, flag, source, out) + } + + c.serviceID = serviceID + + c.dieCh = make(chan struct{}) + c.batchCh = make(chan batch) + c.doneCh = make(chan struct{}) + + text.Info(out, "Streaming logging endpoint errors for service %s\n\n", c.serviceID) + + failure := make(chan error) + sigs := make(chan os.Signal, 2) + signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP) + + // Start the output loop. + go c.outputLoop(out) + + // Start streaming the errors. + go func() { + failure <- c.stream(out) + }() + + select { + case asyncErr := <-failure: + close(c.dieCh) + return asyncErr + case <-c.doneCh: + return nil + case <-sigs: + close(c.dieCh) + } + + return nil +} + +// stream fetches error data from the API and sends it to the output loop. +func (c *Command) stream(out io.Writer) error { + var curWindow *uint64 + if c.from != 0 { + curWindow = &c.from + } + var toWindow *uint64 + if c.to != 0 { + toWindow = &c.to + } + + // Prepare filter slice + var filter []string + if c.filter != "" { + filter = []string{c.filter} + } + + ctx := context.Background() + + for { + // Check if we've passed the "to" requirement. + if toWindow != nil && curWindow != nil && *curWindow > *toWindow { + text.Info(out, "Reached window: %v which is newer than the requested 'to': %v", *curWindow, *toWindow) + close(c.doneCh) + break + } + + // Use go-fastly to fetch logging endpoint errors + resp, err := c.Globals.APIClient.GetLoggingEndpointErrors(ctx, &fastly.LoggingEndpointErrorsInput{ + ServiceID: c.serviceID, + From: curWindow, + To: toWindow, + Filter: filter, + }) + if err != nil { + c.Globals.ErrLog.Add(err) + return fmt.Errorf("unable to fetch logging endpoint errors: %w", err) + } + + // Send errors to the output loop + if len(resp.Errors) > 0 { + c.batchCh <- batch{Errors: resp.Errors} + } + + // Check for next link to continue streaming + if resp.NextFrom != "" { + // Parse the next link value (it's already the from parameter value) + nextFrom, err := strconv.ParseUint(resp.NextFrom, 10, 64) + if err != nil { + c.Globals.ErrLog.AddWithContext(err, map[string]any{ + "NextFrom": resp.NextFrom, + }) + text.Error(out, "error parsing next from") + continue + } + curWindow = &nextFrom + } else { + // No next link, we're done + close(c.doneCh) + break + } + } + return nil +} + +// outputLoop processes the errors out of band from the request/response loop. +func (c *Command) outputLoop(out io.Writer) { + for { + select { + case <-c.dieCh: + return + case batch := <-c.batchCh: + c.printErrors(out, batch.Errors) + } + } +} + +// printErrors prints error entries. +func (c *Command) printErrors(out io.Writer, errors []fastly.LoggingEndpointError) { + if len(errors) == 0 { + return + } + + if c.jsonOutput { + // Output as JSON array + encoder := json.NewEncoder(out) + for _, e := range errors { + if err := encoder.Encode(e); err != nil { + c.Globals.ErrLog.Add(err) + } + } + } else { + // Find the longest endpoint name in this batch for dynamic width + maxEndpointLen := 0 + for _, e := range errors { + if len(e.Endpoint) > maxEndpointLen { + maxEndpointLen = len(e.Endpoint) + } + } + + // Human-readable format - match log-tail style + for _, e := range errors { + // Format timestamp + // #nosec G115 -- Timestamp is in microseconds, multiplication by 1000 for nanoseconds is safe for reasonable time values + timestamp := time.Unix(0, int64(e.Timestamp)*1000) // Convert microseconds to nanoseconds + var timeStr string + if c.printTimestamps { + // Full timestamp with --timestamps flag + timeStr = timestamp.UTC().Format(time.RFC3339) + } else { + // Compact time by default (HH:MM:SS) + timeStr = timestamp.UTC().Format("15:04:05") + } + + // Extract clean error message from details JSON if present + errorSummary := e.Message + if e.Details != "" { + var detailsJSON map[string]interface{} + if err := json.Unmarshal([]byte(e.Details), &detailsJSON); err == nil { + // Try to extract a cleaner error message + if errorMsg, ok := detailsJSON["error"].(string); ok { + // Simplify common error patterns + errorMsg = strings.TrimPrefix(errorMsg, "non-temporary request err: ") + errorMsg = strings.TrimPrefix(errorMsg, "temporary request err: ") + errorSummary = errorMsg + } + } + } + + // Format: time | endpoint | message + fmt.Fprintf(out, "%s | %-*s | %s\n", timeStr, maxEndpointLen, e.Endpoint, errorSummary) + } + } + + // Flush output immediately + if f, ok := out.(*os.File); ok { + _ = f.Sync() + } +} diff --git a/pkg/mock/api.go b/pkg/mock/api.go index 5803a0c29..2b56aff1b 100644 --- a/pkg/mock/api.go +++ b/pkg/mock/api.go @@ -239,7 +239,8 @@ type API struct { GetOriginMetricsForServiceFn func(context.Context, *fastly.GetOriginMetricsInput) (*fastly.OriginInspector, error) GetOriginMetricsForServiceJSONFn func(context.Context, *fastly.GetOriginMetricsInput, any) error - CreateManagedLoggingFn func(context.Context, *fastly.CreateManagedLoggingInput) (*fastly.ManagedLogging, error) + CreateManagedLoggingFn func(context.Context, *fastly.CreateManagedLoggingInput) (*fastly.ManagedLogging, error) + GetLoggingEndpointErrorsFn func(context.Context, *fastly.LoggingEndpointErrorsInput) (*fastly.LoggingEndpointErrorsResponse, error) GetGeneratedVCLFn func(context.Context, *fastly.GetGeneratedVCLInput) (*fastly.VCL, error) @@ -1373,6 +1374,11 @@ func (m API) CreateManagedLogging(ctx context.Context, i *fastly.CreateManagedLo return m.CreateManagedLoggingFn(ctx, i) } +// GetLoggingEndpointErrors implements Interface. +func (m API) GetLoggingEndpointErrors(ctx context.Context, i *fastly.LoggingEndpointErrorsInput) (*fastly.LoggingEndpointErrorsResponse, error) { + return m.GetLoggingEndpointErrorsFn(ctx, i) +} + // GetGeneratedVCL implements Interface. func (m API) GetGeneratedVCL(ctx context.Context, i *fastly.GetGeneratedVCLInput) (*fastly.VCL, error) { return m.GetGeneratedVCLFn(ctx, i)