diff --git a/cmd/skpr/logs/command.go b/cmd/skpr/logs/command.go index d601b03..64f83a0 100644 --- a/cmd/skpr/logs/command.go +++ b/cmd/skpr/logs/command.go @@ -4,6 +4,8 @@ import ( "github.com/spf13/cobra" "github.com/skpr/cli/cmd/skpr/logs/list" + "github.com/skpr/cli/cmd/skpr/logs/query" + "github.com/skpr/cli/cmd/skpr/logs/summarise" "github.com/skpr/cli/cmd/skpr/logs/tail" skprcommand "github.com/skpr/cli/internal/command" ) @@ -24,6 +26,8 @@ func NewCommand() *cobra.Command { } cmd.AddCommand(list.NewCommand()) + cmd.AddCommand(query.NewCommand()) + cmd.AddCommand(summarise.NewCommand()) cmd.AddCommand(tail.NewCommand()) return cmd diff --git a/cmd/skpr/logs/query/command.go b/cmd/skpr/logs/query/command.go new file mode 100644 index 0000000..6cc5a86 --- /dev/null +++ b/cmd/skpr/logs/query/command.go @@ -0,0 +1,54 @@ +package query + +import ( + "github.com/spf13/cobra" + + "github.com/skpr/cli/internal/command/logs/filter" + v1query "github.com/skpr/cli/internal/command/logs/query" +) + +var ( + cmdLong = `Run a bounded query over the logs of a running application.` + + cmdExample = ` + # Query the last hour of the default streams. + skpr logs query dev + + # Query specific streams over the last 24 hours. + skpr logs query dev nginx fpm --timeframe 24h + + # Query an absolute time range. + skpr logs query dev --from "2 days ago" --to now + + # Query for events containing a substring, excluding another. + skpr logs query dev --contains error --exclude healthcheck` +) + +// NewCommand creates a new cobra.Command for 'query' sub command +func NewCommand() *cobra.Command { + command := v1query.Command{} + + cmd := &cobra.Command{ + Use: "query ", + Args: cobra.MinimumNArgs(1), + DisableFlagsInUseLine: true, + Short: "Run a bounded query over the logs of a running application", + Long: cmdLong, + Example: cmdExample, + RunE: func(cmd *cobra.Command, args []string) error { + command.Environment = args[0] + + if len(args) > 1 { + command.Streams = args[1:] + } + + return command.Run(cmd.Context()) + }, + } + + filter.AddFlags(cmd.Flags(), &command.Options) + cmd.Flags().Int32Var(&command.Limit, "limit", 0, "Cap on returned events (0 for server default)") + cmd.Flags().BoolVar(&command.Indent, "indent", false, "Enable indenting for pretty printed logs") + + return cmd +} diff --git a/cmd/skpr/logs/summarise/command.go b/cmd/skpr/logs/summarise/command.go new file mode 100644 index 0000000..d377e76 --- /dev/null +++ b/cmd/skpr/logs/summarise/command.go @@ -0,0 +1,51 @@ +package summarise + +import ( + "github.com/spf13/cobra" + + "github.com/skpr/cli/internal/command/logs/filter" + v1summarise "github.com/skpr/cli/internal/command/logs/summarise" +) + +var ( + cmdLong = `Summarise a window of an application's logs using a natural-language prompt.` + + cmdExample = ` + # Summarise the last hour of the default streams. + skpr logs summarise dev + + # Summarise specific streams over the last 24 hours. + skpr logs summarise dev nginx fpm --timeframe 24h + + # Ask a specific question of the summariser. + skpr logs summarise dev --prompt "What caused the 500 errors?"` +) + +// NewCommand creates a new cobra.Command for 'summarise' sub command +func NewCommand() *cobra.Command { + command := v1summarise.Command{} + + cmd := &cobra.Command{ + Use: "summarise ", + Aliases: []string{"summarize"}, + Args: cobra.MinimumNArgs(1), + DisableFlagsInUseLine: true, + Short: "Summarise a window of an application's logs", + Long: cmdLong, + Example: cmdExample, + RunE: func(cmd *cobra.Command, args []string) error { + command.Environment = args[0] + + if len(args) > 1 { + command.Streams = args[1:] + } + + return command.Run(cmd.Context()) + }, + } + + filter.AddFlags(cmd.Flags(), &command.Options) + cmd.Flags().StringVar(&command.Prompt, "prompt", "", "Natural-language question for the summariser") + + return cmd +} diff --git a/go.mod b/go.mod index 87ecd69..01e70d2 100644 --- a/go.mod +++ b/go.mod @@ -30,7 +30,7 @@ require ( github.com/moby/patternmatcher v0.6.1 github.com/olekukonko/ts v0.0.0-20171002115256-78ecb04241c0 github.com/pkg/errors v0.9.1 - github.com/skpr/api v1.6.0 + github.com/skpr/api v1.7.0 github.com/skpr/compass/tracing v0.0.0-20251208094547-dafe383c3926 github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966 github.com/spf13/cobra v1.10.2 diff --git a/go.sum b/go.sum index bd78e10..70d83bd 100644 --- a/go.sum +++ b/go.sum @@ -219,6 +219,8 @@ github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= github.com/skpr/api v1.6.0 h1:LzbNm97wXvnWAcm+iyV1r36YKGMFWNn65y+EHhBxgqo= github.com/skpr/api v1.6.0/go.mod h1:LMyIte2bWjGuO2e8XOA6FKNFUX10itF2qYVTckM7dwg= +github.com/skpr/api v1.7.0 h1:J9ZRTKDyUlSHXUGbkppB22fic/gC40g1+LjlWuxRijs= +github.com/skpr/api v1.7.0/go.mod h1:HWzGU9hY392K8N4I7xdk+anSRLoouJM2NtSjnYfIj1s= github.com/skpr/compass/tracing v0.0.0-20251208094547-dafe383c3926 h1:g8m8qqehB6/GMeM81BaIXQOkD2LZbCQE0HPRGear9Jw= github.com/skpr/compass/tracing v0.0.0-20251208094547-dafe383c3926/go.mod h1:Rokt2mnHteBaBmsSgiNk4P7laV0E8/BLQsSLyGUJLtk= github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966 h1:JIAuq3EEf9cgbU6AtGPK4CTG3Zf6CKMNqf0MHTggAUA= diff --git a/internal/command/logs/filter/filter.go b/internal/command/logs/filter/filter.go new file mode 100644 index 0000000..46e12b5 --- /dev/null +++ b/internal/command/logs/filter/filter.go @@ -0,0 +1,104 @@ +package filter + +import ( + "fmt" + "time" + + "github.com/skpr/api/pb" + "google.golang.org/protobuf/types/known/durationpb" + "google.golang.org/protobuf/types/known/timestamppb" + + skprtime "github.com/skpr/cli/internal/time" +) + +// Options holds the shared filter inputs used by bounded log operations such as +// query and summarise. +type Options struct { + Environment string + Streams []string + Timeframe time.Duration + From string + To string + Contains []string + Exclude []string +} + +// Build converts the options into a pb.LogFilter. An absolute time range +// (--from/--to) takes precedence over the relative --timeframe. +func (o Options) Build() (*pb.LogFilter, error) { + filter := &pb.LogFilter{ + Environment: o.Environment, + Streams: o.Streams, + } + + // Build the substring filters. An event must satisfy every entry. + for _, value := range o.Contains { + filter.Contains = append(filter.Contains, &pb.LogContainsFilter{ + Value: value, + }) + } + + for _, value := range o.Exclude { + filter.Contains = append(filter.Contains, &pb.LogContainsFilter{ + Value: value, + Exclude: true, + }) + } + + if o.From != "" || o.To != "" { + window, err := buildTimeRange(o.From, o.To) + if err != nil { + return nil, err + } + + filter.Window = window + } else { + filter.Window = &pb.LogFilter_Timeframe{ + Timeframe: durationpb.New(o.Timeframe), + } + } + + return filter, nil +} + +// buildTimeRange constructs an absolute time range window from the provided +// --from and --to values. Either bound may be omitted. +func buildTimeRange(from, to string) (*pb.LogFilter_TimeRange, error) { + timeRange := &pb.LogTimeRange{} + + if from != "" { + t, err := skprtime.ParseString(from) + if err != nil { + return nil, fmt.Errorf("failed to parse --from: %w", err) + } + + timeRange.From = timestamppb.New(t) + } + + if to != "" { + t, err := skprtime.ParseString(to) + if err != nil { + return nil, fmt.Errorf("failed to parse --to: %w", err) + } + + timeRange.To = timestamppb.New(t) + } + + return &pb.LogFilter_TimeRange{TimeRange: timeRange}, nil +} + +// AddFlags registers the shared filter flags on a command's flag set. +func AddFlags(flags flagSet, o *Options) { + flags.DurationVar(&o.Timeframe, "timeframe", time.Hour, "Window duration relative to now (ignored when --from or --to is set)") + flags.StringVar(&o.From, "from", "", "Start of an absolute time range (e.g. \"2 days ago\", RFC3339 timestamp)") + flags.StringVar(&o.To, "to", "", "End of an absolute time range (e.g. \"now\", RFC3339 timestamp)") + flags.StringArrayVar(&o.Contains, "contains", nil, "Only include events whose message contains this substring (repeatable)") + flags.StringArrayVar(&o.Exclude, "exclude", nil, "Exclude events whose message contains this substring (repeatable)") +} + +// flagSet is the subset of *pflag.FlagSet used to register filter flags. +type flagSet interface { + DurationVar(p *time.Duration, name string, value time.Duration, usage string) + StringVar(p *string, name, value, usage string) + StringArrayVar(p *[]string, name string, value []string, usage string) +} diff --git a/internal/command/logs/list/command.go b/internal/command/logs/list/command.go index 1451caa..4e9d327 100644 --- a/internal/command/logs/list/command.go +++ b/internal/command/logs/list/command.go @@ -4,18 +4,22 @@ import ( "context" "fmt" "os" - - "github.com/skpr/api/pb" + "strings" "github.com/skpr/cli/internal/client" + "github.com/skpr/cli/internal/command/logs/streams" "github.com/skpr/cli/internal/components/tooltip" skprtable "github.com/skpr/cli/internal/table" ) // Helpful text provided by the tooltip. -const tooltipText = `To view logs for a specific stream, use the tail command. +const tooltipText = `To tail a stream in real-time, use the tail command. + +$ skpr logs tail ENVIRONMENT STREAM STREAM + +To run a bounded query over a time window, use the query command. -$ skpr logs tail ENVIRONMENT STREAM STREAM` +$ skpr logs query ENVIRONMENT STREAM STREAM` // Command to list all log sources. type Command struct { @@ -29,21 +33,26 @@ func (cmd *Command) Run(ctx context.Context) error { return err } - resp, err := client.Logs().ListStreams(ctx, &pb.LogListStreamsRequest{ - Environment: cmd.Environment, - }) + resp, err := streams.List(ctx, client.Logs(), cmd.Environment) if err != nil { return fmt.Errorf("failed to list streams: %w", err) } header := []string{ - "Streams", + "Stream", + "Capabilities", } var rows [][]string for _, stream := range resp.Streams { - rows = append(rows, []string{stream}) + var capabilities []string + + for _, streamType := range stream.Types { + capabilities = append(capabilities, streamType.String()) + } + + rows = append(rows, []string{stream.Name, strings.Join(capabilities, ", ")}) } err = skprtable.Print(os.Stdout, header, rows) diff --git a/internal/command/logs/query/command.go b/internal/command/logs/query/command.go new file mode 100644 index 0000000..4541e97 --- /dev/null +++ b/internal/command/logs/query/command.go @@ -0,0 +1,103 @@ +package query + +import ( + "context" + "encoding/json" + "fmt" + "io" + "os" + "time" + + "github.com/TylerBrock/colorjson" + faithcolor "github.com/fatih/color" + "github.com/jwalton/gchalk" + "github.com/skpr/api/pb" + + "github.com/skpr/cli/internal/client" + "github.com/skpr/cli/internal/color" + "github.com/skpr/cli/internal/command/logs/filter" +) + +// Command to run a bounded query over an environment's logs. +type Command struct { + filter.Options + Limit int32 + Indent bool +} + +// Run the command. +func (cmd *Command) Run(ctx context.Context) error { + ctx, client, err := client.New(ctx) + if err != nil { + return err + } + + logFilter, err := cmd.Options.Build() + if err != nil { + return err + } + + stream, err := client.Logs().Query(ctx, &pb.LogQueryRequest{ + Filter: logFilter, + Limit: cmd.Limit, + }) + if err != nil { + return fmt.Errorf("failed to run query: %w", err) + } + + for { + resp, err := stream.Recv() + if err == io.EOF { + break + } + + if err != nil { + return fmt.Errorf("failed to receive query response: %w", err) + } + + if batch := resp.GetBatch(); batch != nil { + for _, event := range batch.Events { + printEvent(event, cmd.Indent) + } + } + + if meta := resp.GetMeta(); meta != nil { + fmt.Fprintf(os.Stderr, "Scanned %d events\n", meta.Scanned) + } + } + + return nil +} + +// printEvent prints a single log event prefixed with its timestamp and stream. +func printEvent(event *pb.LogEvent, indent bool) { + prefix := gchalk.WithHex(color.HexOrange).Bold(event.Stream) + timestamp := event.Timestamp.AsTime().Format(time.RFC3339) + message := prettyPrint(event.Message, indent) + + fmt.Println(timestamp, prefix, message) +} + +// Returns a pretty output for JSON messages. +func prettyPrint(message string, indent bool) string { + var obj map[string]interface{} + + err := json.Unmarshal([]byte(message), &obj) + if err != nil { + return message + } + + formatter := colorjson.NewFormatter() + formatter.KeyColor = faithcolor.New(faithcolor.FgWhite).Add(faithcolor.Bold) + + if indent { + formatter.Indent = 2 + } + + raw, err := formatter.Marshal(obj) + if err != nil { + return message + } + + return string(raw) +} diff --git a/internal/command/logs/streams/streams.go b/internal/command/logs/streams/streams.go new file mode 100644 index 0000000..f0a0fb4 --- /dev/null +++ b/internal/command/logs/streams/streams.go @@ -0,0 +1,54 @@ +package streams + +import ( + "context" + "fmt" + + "github.com/skpr/api/pb" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// List returns the streams available for an environment using the ListStreamsV2 +// RPC. Older servers may not implement ListStreamsV2, in which case this falls +// back to the original ListStreams RPC and adapts the response to the V2 shape. +func List(ctx context.Context, client pb.LogsClient, environment string, types ...pb.LogStreamType) (*pb.LogListStreamsV2Response, error) { + resp, err := client.ListStreamsV2(ctx, &pb.LogListStreamsV2Request{ + Environment: environment, + Types: types, + }) + if err == nil { + return resp, nil + } + + // Any error other than the RPC being missing is a genuine failure. + if status.Code(err) != codes.Unimplemented { + return nil, err + } + + // Fall back to the original ListStreams RPC for older servers. It has no + // concept of stream capabilities, so every stream is returned without types + // and the single default is treated as the default for tailing. + legacy, err := client.ListStreams(ctx, &pb.LogListStreamsRequest{ + Environment: environment, + }) + if err != nil { + return nil, fmt.Errorf("failed to list streams: %w", err) + } + + out := &pb.LogListStreamsV2Response{ + Defaults: &pb.LogStreamDefaults{}, + } + + for _, name := range legacy.Streams { + out.Streams = append(out.Streams, &pb.LogStream{ + Name: name, + }) + } + + if legacy.Default != "" { + out.Defaults.Tail = []string{legacy.Default} + } + + return out, nil +} diff --git a/internal/command/logs/streams/streams_test.go b/internal/command/logs/streams/streams_test.go new file mode 100644 index 0000000..a9b3cef --- /dev/null +++ b/internal/command/logs/streams/streams_test.go @@ -0,0 +1,81 @@ +package streams + +import ( + "context" + "testing" + + "github.com/skpr/api/pb" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// mockLogsClient is a stub pb.LogsClient that only implements the two list RPCs. +type mockLogsClient struct { + pb.LogsClient + + v2Resp *pb.LogListStreamsV2Response + v2Err error + + v1Resp *pb.LogListStreamsResponse + listStreamsCalled bool +} + +func (m *mockLogsClient) ListStreamsV2(_ context.Context, _ *pb.LogListStreamsV2Request, _ ...grpc.CallOption) (*pb.LogListStreamsV2Response, error) { + return m.v2Resp, m.v2Err +} + +func (m *mockLogsClient) ListStreams(_ context.Context, _ *pb.LogListStreamsRequest, _ ...grpc.CallOption) (*pb.LogListStreamsResponse, error) { + m.listStreamsCalled = true + return m.v1Resp, nil +} + +func TestListUsesV2WhenAvailable(t *testing.T) { + mock := &mockLogsClient{ + v2Resp: &pb.LogListStreamsV2Response{ + Streams: []*pb.LogStream{ + {Name: "nginx", Types: []pb.LogStreamType{pb.LogStreamType_Tail}}, + }, + Defaults: &pb.LogStreamDefaults{Tail: []string{"nginx"}}, + }, + } + + resp, err := List(context.Background(), mock, "dev") + require.NoError(t, err) + assert.False(t, mock.listStreamsCalled, "should not fall back when V2 succeeds") + assert.Equal(t, "nginx", resp.Streams[0].Name) + assert.Equal(t, []string{"nginx"}, resp.Defaults.Tail) +} + +func TestListFallsBackWhenV2Unimplemented(t *testing.T) { + mock := &mockLogsClient{ + v2Err: status.Error(codes.Unimplemented, "method ListStreamsV2 not implemented"), + v1Resp: &pb.LogListStreamsResponse{ + Streams: []string{"nginx", "fpm"}, + Default: "nginx", + }, + } + + resp, err := List(context.Background(), mock, "dev") + require.NoError(t, err) + assert.True(t, mock.listStreamsCalled, "should fall back when V2 is unimplemented") + + require.Len(t, resp.Streams, 2) + assert.Equal(t, "nginx", resp.Streams[0].Name) + assert.Equal(t, "fpm", resp.Streams[1].Name) + assert.Empty(t, resp.Streams[0].Types, "legacy streams have no capability metadata") + assert.Equal(t, []string{"nginx"}, resp.Defaults.Tail) +} + +func TestListPropagatesOtherErrors(t *testing.T) { + mock := &mockLogsClient{ + v2Err: status.Error(codes.PermissionDenied, "nope"), + } + + _, err := List(context.Background(), mock, "dev") + require.Error(t, err) + assert.False(t, mock.listStreamsCalled, "should not fall back on non-Unimplemented errors") + assert.Equal(t, codes.PermissionDenied, status.Code(err)) +} diff --git a/internal/command/logs/summarise/command.go b/internal/command/logs/summarise/command.go new file mode 100644 index 0000000..c45065a --- /dev/null +++ b/internal/command/logs/summarise/command.go @@ -0,0 +1,76 @@ +package summarise + +import ( + "context" + "fmt" + "io" + "os" + + "github.com/jwalton/gchalk" + "github.com/skpr/api/pb" + + "github.com/skpr/cli/internal/client" + "github.com/skpr/cli/internal/color" + "github.com/skpr/cli/internal/command/logs/filter" +) + +// Command to summarise a window of an environment's logs. +type Command struct { + filter.Options + Prompt string +} + +// Run the command. +func (cmd *Command) Run(ctx context.Context) error { + ctx, client, err := client.New(ctx) + if err != nil { + return err + } + + logFilter, err := cmd.Options.Build() + if err != nil { + return err + } + + resp, err := client.Logs().Summarise(ctx, &pb.LogSummariseRequest{ + Filter: logFilter, + Prompt: cmd.Prompt, + }) + if err != nil { + return fmt.Errorf("failed to summarise logs: %w", err) + } + + print(os.Stdout, resp) + + return nil +} + +// print renders the summary response to the given writer. +func print(w io.Writer, resp *pb.LogSummariseResponse) { + heading := func(text string) string { + return gchalk.WithHex(color.HexOrange).Bold(text) + } + + if resp.Overview != "" { + fmt.Fprintln(w, heading("Overview")) + fmt.Fprintln(w, resp.Overview) + } + + if len(resp.Bullets) > 0 { + fmt.Fprintln(w) + fmt.Fprintln(w, heading("Notable events")) + + for _, bullet := range resp.Bullets { + fmt.Fprintf(w, " - %s\n", bullet) + } + } + + if len(resp.SuggestedActions) > 0 { + fmt.Fprintln(w) + fmt.Fprintln(w, heading("Suggested actions")) + + for _, action := range resp.SuggestedActions { + fmt.Fprintf(w, " - %s\n", action) + } + } +} diff --git a/internal/command/logs/tail/command.go b/internal/command/logs/tail/command.go index 626cbf9..75b7e2a 100644 --- a/internal/command/logs/tail/command.go +++ b/internal/command/logs/tail/command.go @@ -16,6 +16,7 @@ import ( "github.com/skpr/cli/internal/client" "github.com/skpr/cli/internal/color" + "github.com/skpr/cli/internal/command/logs/streams" ) // Command to stream the logs for an environment. @@ -32,18 +33,21 @@ func (cmd *Command) Run(ctx context.Context) error { return err } - list, err := client.Logs().ListStreams(ctx, &pb.LogListStreamsRequest{ - Environment: cmd.Environment, - }) + list, err := streams.List(ctx, client.Logs(), cmd.Environment, pb.LogStreamType_Tail) if err != nil { return fmt.Errorf("failed to get stream list: %w", err) } - // If none are provided use the default stream provided by the API. - if len(cmd.Streams) == 0 && list.Default != "" { - cmd.Streams = []string{ - list.Default, - } + // Collect the streams which support tailing. + var available []string + + for _, stream := range list.Streams { + available = append(available, stream.Name) + } + + // If none are provided use the default streams provided by the API. + if len(cmd.Streams) == 0 && list.Defaults != nil { + cmd.Streams = list.Defaults.Tail } if len(cmd.Streams) == 0 { @@ -52,7 +56,7 @@ func (cmd *Command) Run(ctx context.Context) error { // Validate that the provided streams are correct. for _, stream := range cmd.Streams { - if !slices.Contains(list.Streams, stream) { + if !slices.Contains(available, stream) { return fmt.Errorf("stream not found: %s", stream) } }