[CDTOOL-1193] Support for Logging Endpoint Errors

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:

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)
 

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,

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)
+	}
+}

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

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()
+	}
+}

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)