diff --git a/go.mod b/go.mod index 8db434d..552f055 100644 --- a/go.mod +++ b/go.mod @@ -30,8 +30,8 @@ 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.8.0 - github.com/skpr/compass/tracing v0.0.0-20251208094547-dafe383c3926 + github.com/skpr/api v1.8.3 + github.com/skpr/compass v1.12.0 github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966 github.com/spf13/cobra v1.10.2 github.com/stretchr/testify v1.12.1 @@ -61,12 +61,12 @@ require ( github.com/aws/smithy-go v1.28.1 // indirect github.com/aymanbagabas/go-osc52/v2 v2.0.1 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect - github.com/charmbracelet/bubbles v0.21.0 // indirect + github.com/charmbracelet/bubbles v1.0.0 // indirect github.com/charmbracelet/colorprofile v0.4.3 // indirect github.com/charmbracelet/lipgloss v1.1.0 // indirect github.com/charmbracelet/ultraviolet v0.0.0-20260811164956-006e29f97886 // indirect github.com/charmbracelet/x/ansi v0.11.8 // indirect - github.com/charmbracelet/x/cellbuf v0.0.14 // indirect + github.com/charmbracelet/x/cellbuf v0.0.15 // indirect github.com/charmbracelet/x/exp/charmtone v0.0.0-20251215102626-e0db08df7383 // indirect github.com/charmbracelet/x/exp/golden v0.0.0-20251215102626-e0db08df7383 // indirect github.com/charmbracelet/x/term v0.2.2 // indirect @@ -109,7 +109,6 @@ require ( github.com/opencontainers/go-digest v1.0.0 // indirect github.com/opencontainers/image-spec v1.1.1 // indirect github.com/rivo/uniseg v0.4.7 // indirect - github.com/sahilm/fuzzy v0.1.1 // indirect github.com/sirupsen/logrus v1.9.4 // indirect github.com/spf13/pflag v1.0.10 // indirect github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect diff --git a/go.sum b/go.sum index b1cd49a..33883e8 100644 --- a/go.sum +++ b/go.sum @@ -54,8 +54,8 @@ github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1x github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= -github.com/charmbracelet/bubbles v0.21.0 h1:9TdC97SdRVg/1aaXNVWfFH3nnLAwOXr8Fn6u6mfQdFs= -github.com/charmbracelet/bubbles v0.21.0/go.mod h1:HF+v6QUR4HkEpz62dx7ym2xc71/KBHg+zKwJtMw+qtg= +github.com/charmbracelet/bubbles v1.0.0 h1:12J8/ak/uCZEMQ6KU7pcfwceyjLlWsDLAxB5fXonfvc= +github.com/charmbracelet/bubbles v1.0.0/go.mod h1:9d/Zd5GdnauMI5ivUIVisuEm3ave1XwXtD1ckyV6r3E= github.com/charmbracelet/bubbletea v1.3.10 h1:otUDHWMMzQSB0Pkc87rm691KZ3SWa4KUlvF9nRvCICw= github.com/charmbracelet/bubbletea v1.3.10/go.mod h1:ORQfo0fk8U+po9VaNvnV95UPWA1BitP1E0N6xJPlHr4= github.com/charmbracelet/colorprofile v0.4.3 h1:QPa1IWkYI+AOB+fE+mg/5/4HRMZcaXex9t5KX76i20Q= @@ -68,8 +68,8 @@ github.com/charmbracelet/ultraviolet v0.0.0-20260811164956-006e29f97886 h1:rdnVW github.com/charmbracelet/ultraviolet v0.0.0-20260811164956-006e29f97886/go.mod h1:nAw0d9PhFp1qdzi2xhQU5YOu5sVpDIHWlaW2Uz/bCro= github.com/charmbracelet/x/ansi v0.11.8 h1:JMFwp0CgDC2+jcOB162HH5k7I3FVbgFSMMYg7dSPBQQ= github.com/charmbracelet/x/ansi v0.11.8/go.mod h1:ZNN+3mXny/516oTQPLMPIBeSINvNJJQ8uQXDgbeJxY0= -github.com/charmbracelet/x/cellbuf v0.0.14 h1:iUEMryGyFTelKW3THW4+FfPgi4fkmKnnaLOXuc+/Kj4= -github.com/charmbracelet/x/cellbuf v0.0.14/go.mod h1:P447lJl49ywBbil/KjCk2HexGh4tEY9LH0/1QrZZ9rA= +github.com/charmbracelet/x/cellbuf v0.0.15 h1:ur3pZy0o6z/R7EylET877CBxaiE1Sp1GMxoFPAIztPI= +github.com/charmbracelet/x/cellbuf v0.0.15/go.mod h1:J1YVbR7MUuEGIFPCaaZ96KDl5NoS0DAWkskup+mOY+Q= github.com/charmbracelet/x/exp/charmtone v0.0.0-20251215102626-e0db08df7383 h1:xGojlO6kHCDB1k6DolME79LG0u90TzVd8atGhmxFRIo= github.com/charmbracelet/x/exp/charmtone v0.0.0-20251215102626-e0db08df7383/go.mod h1:nsExn0DGyX0lh9LwLHTn2Gg+hafdzfSXnC+QmEJTZFY= github.com/charmbracelet/x/exp/golden v0.0.0-20251215102626-e0db08df7383 h1:R0iAuPE4yU0omOM9ANVmxYqW+ktB9xMDMyxx6prkrA0= @@ -145,8 +145,6 @@ github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= -github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= -github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/lucasb-eyer/go-colorful v1.4.1 h1:1EO+WB73+EH8EVbzlrG3KLAfEypQWVHIBqlTf+2hNss= github.com/lucasb-eyer/go-colorful v1.4.1/go.mod h1:R4dSotOR9KMtayYi1e77YzuveK+i7ruzyGqttikkLy0= github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE= @@ -210,14 +208,12 @@ github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUc github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= -github.com/sahilm/fuzzy v0.1.1 h1:ceu5RHF8DGgoi+/dR5PsECjCDH1BE3Fnmpo7aVXOdRA= -github.com/sahilm/fuzzy v0.1.1/go.mod h1:VFvziUEIMCrT6A6tw2RFIXPXXmzXbOsSHF0DOI8ZK9Y= github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w= github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g= -github.com/skpr/api v1.8.0 h1:mOfpy6W+8THIaRG2nxIFpQ8wdtTawSoUZbwhcs0VS0M= -github.com/skpr/api v1.8.0/go.mod h1:KMGPWD5+HwbmqqX6HNtdpDylQ1lFfTS/qiSv/bolfdc= -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/skpr/api v1.8.3 h1:nlPR0gfS02BJOWkvwuRbj6CyAku9ETHOnXSV8oEGNxY= +github.com/skpr/api v1.8.3/go.mod h1:3ujFy8YadnZ/eEOCoeTnmtIFeVoBu6sQjTLLUKkFVFY= +github.com/skpr/compass v1.12.0 h1:3MwUg1Sg+qVitZjz9C2EVKfZwgvldFwXfbuqQRWBQEo= +github.com/skpr/compass v1.12.0/go.mod h1:VaD+bWzckT40JGeAPPkyu9Ce30tcSS2rFCbeO1buLu0= github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966 h1:JIAuq3EEf9cgbU6AtGPK4CTG3Zf6CKMNqf0MHTggAUA= github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966/go.mod h1:sUM3LWHvSMaG192sy56D9F7CNvL7jUJVXoqM1QKLnog= github.com/spf13/cobra v1.10.2 h1:DMTTonx5m65Ic0GOoRY2c16WCbHxOOw6xxezuLaBpcU= diff --git a/internal/command/trace/client.go b/internal/command/trace/client.go new file mode 100644 index 0000000..9a91ecc --- /dev/null +++ b/internal/command/trace/client.go @@ -0,0 +1,61 @@ +package trace + +import ( + "context" + "fmt" + + "github.com/skpr/api/pb" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/skpr/cli/internal/client" +) + +type traceStream interface { + Recv() (*pb.StreamTracesResponse, error) +} + +type commandAPI interface { + GetEnvironment(context.Context, *pb.EnvironmentGetRequest) (*pb.EnvironmentGetResponse, error) + StreamTraces(context.Context, *pb.StreamTracesRequest) (traceStream, error) +} + +type connectFunc func(context.Context) (context.Context, commandAPI, error) + +type skprAPI struct { + client *client.Client +} + +func connectAPI(ctx context.Context) (context.Context, commandAPI, error) { + ctx, apiClient, err := client.New(ctx) + if err != nil { + return ctx, nil, err + } + + return ctx, skprAPI{client: apiClient}, nil +} + +func (api skprAPI) GetEnvironment(ctx context.Context, request *pb.EnvironmentGetRequest) (*pb.EnvironmentGetResponse, error) { + return api.client.Environment().Get(ctx, request) +} + +func (api skprAPI) StreamTraces(ctx context.Context, request *pb.StreamTracesRequest) (traceStream, error) { + return api.client.Trace().StreamTraces(ctx, request) +} + +func (cmd *Command) preflight(ctx context.Context, connect connectFunc) (context.Context, commandAPI, error) { + ctx, api, err := connect(ctx) + if err != nil { + return ctx, nil, fmt.Errorf("failed to connect to Skpr API: %w", err) + } + + _, err = api.GetEnvironment(ctx, &pb.EnvironmentGetRequest{Name: cmd.Environment}) + if status.Code(err) == codes.NotFound { + return ctx, nil, fmt.Errorf("environment %q does not exist", cmd.Environment) + } + if err != nil { + return ctx, nil, fmt.Errorf("failed to verify environment %q: %w", cmd.Environment, err) + } + + return ctx, api, nil +} diff --git a/internal/command/trace/client_test.go b/internal/command/trace/client_test.go new file mode 100644 index 0000000..905f6a9 --- /dev/null +++ b/internal/command/trace/client_test.go @@ -0,0 +1,92 @@ +package trace + +import ( + "context" + "errors" + "testing" + + "github.com/skpr/api/pb" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +type fakeCommandAPI struct { + getEnvironment func(context.Context, *pb.EnvironmentGetRequest) (*pb.EnvironmentGetResponse, error) + streamTraces func(context.Context, *pb.StreamTracesRequest) (traceStream, error) +} + +func (api *fakeCommandAPI) GetEnvironment(ctx context.Context, request *pb.EnvironmentGetRequest) (*pb.EnvironmentGetResponse, error) { + return api.getEnvironment(ctx, request) +} + +func (api *fakeCommandAPI) StreamTraces(ctx context.Context, request *pb.StreamTracesRequest) (traceStream, error) { + return api.streamTraces(ctx, request) +} + +func TestPreflightConnectionFailure(t *testing.T) { + cmd := Command{Environment: "staging"} + + _, api, err := cmd.preflight(context.Background(), func(ctx context.Context) (context.Context, commandAPI, error) { + return ctx, nil, errors.New("connection refused") + }) + + require.EqualError(t, err, "failed to connect to Skpr API: connection refused") + assert.Nil(t, api) +} + +func TestPreflightEnvironmentNotFound(t *testing.T) { + cmd := Command{Environment: "missing"} + api := &fakeCommandAPI{ + getEnvironment: func(_ context.Context, request *pb.EnvironmentGetRequest) (*pb.EnvironmentGetResponse, error) { + assert.Equal(t, "missing", request.GetName()) + return nil, status.Error(codes.NotFound, "environment not found") + }, + } + + _, validatedAPI, err := cmd.preflight(context.Background(), func(ctx context.Context) (context.Context, commandAPI, error) { + return ctx, api, nil + }) + + require.EqualError(t, err, `environment "missing" does not exist`) + assert.Nil(t, validatedAPI) +} + +func TestPreflightEnvironmentLookupFailure(t *testing.T) { + cmd := Command{Environment: "staging"} + api := &fakeCommandAPI{ + getEnvironment: func(context.Context, *pb.EnvironmentGetRequest) (*pb.EnvironmentGetResponse, error) { + return nil, status.Error(codes.PermissionDenied, "permission denied") + }, + } + + _, validatedAPI, err := cmd.preflight(context.Background(), func(ctx context.Context) (context.Context, commandAPI, error) { + return ctx, api, nil + }) + + require.EqualError(t, err, `failed to verify environment "staging": rpc error: code = PermissionDenied desc = permission denied`) + assert.Nil(t, validatedAPI) +} + +func TestPreflightSuccess(t *testing.T) { + type contextKey string + const key contextKey = "authenticated" + + cmd := Command{Environment: "staging"} + api := &fakeCommandAPI{ + getEnvironment: func(ctx context.Context, request *pb.EnvironmentGetRequest) (*pb.EnvironmentGetResponse, error) { + assert.Equal(t, true, ctx.Value(key)) + assert.Equal(t, "staging", request.GetName()) + return &pb.EnvironmentGetResponse{}, nil + }, + } + + ctx, validatedAPI, err := cmd.preflight(context.Background(), func(ctx context.Context) (context.Context, commandAPI, error) { + return context.WithValue(ctx, key, true), api, nil + }) + + require.NoError(t, err) + assert.Equal(t, true, ctx.Value(key)) + assert.Same(t, api, validatedAPI) +} diff --git a/internal/command/trace/collector.go b/internal/command/trace/collector.go new file mode 100644 index 0000000..f81b217 --- /dev/null +++ b/internal/command/trace/collector.go @@ -0,0 +1,105 @@ +package trace + +import ( + "context" + "errors" + "fmt" + "io" + "time" + + tea "github.com/charmbracelet/bubbletea" + "github.com/skpr/api/pb" + "github.com/skpr/compass/pkg/app/events" +) + +const ( + traceRetryInitial = time.Second + traceRetryMaximum = 30 * time.Second +) + +type messageSender interface { + Send(tea.Msg) +} + +type errorLogger interface { + Error(string, ...any) +} + +func collectTraces(ctx context.Context, api commandAPI, environment string, sender messageSender, logger errorLogger) error { + retryDelay := traceRetryInitial + + for { + if ctx.Err() != nil { + return nil + } + + sender.Send(events.Connection{State: events.ConnectionStateConnecting}) + + stream, err := api.StreamTraces(ctx, &pb.StreamTracesRequest{Environment: environment}) + if err == nil && stream == nil { + err = errors.New("trace stream was not created") + } + if err == nil { + sender.Send(events.Connection{State: events.ConnectionStateConnected}) + retryDelay = traceRetryInitial + err = receiveTraces(ctx, stream, sender) + } + + if ctx.Err() != nil { + return nil + } + + if errors.Is(err, io.EOF) { + err = errors.New("trace stream closed") + } else { + err = fmt.Errorf("trace stream failed: %w", err) + } + + logger.Error(err.Error()) + sender.Send(events.Connection{State: events.ConnectionStateRetrying, Err: err}) + + timer := time.NewTimer(retryDelay) + select { + case <-ctx.Done(): + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + return nil + case <-timer.C: + } + + if retryDelay < traceRetryMaximum { + retryDelay *= 2 + if retryDelay > traceRetryMaximum { + retryDelay = traceRetryMaximum + } + } + } +} + +func receiveTraces(ctx context.Context, stream traceStream, sender messageSender) error { + for { + response, err := stream.Recv() + if err != nil { + return err + } + if response == nil { + return errors.New("received an empty trace stream response") + } + + for _, item := range response.GetTraces() { + converted := traceFromProto(item) + sender.Send(events.Trace{ + IngestionTime: converted.Metadata.StartTime, + Trace: converted, + }) + } + + if ctx.Err() != nil { + return ctx.Err() + } + } +} diff --git a/internal/command/trace/collector_test.go b/internal/command/trace/collector_test.go new file mode 100644 index 0000000..2ef0c97 --- /dev/null +++ b/internal/command/trace/collector_test.go @@ -0,0 +1,157 @@ +package trace + +import ( + "context" + "errors" + "io" + "testing" + + tea "github.com/charmbracelet/bubbletea" + "github.com/skpr/api/pb" + "github.com/skpr/compass/pkg/app/events" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type fakeTraceStream struct { + recv func() (*pb.StreamTracesResponse, error) +} + +func (stream *fakeTraceStream) Recv() (*pb.StreamTracesResponse, error) { + return stream.recv() +} + +type recordingSender struct { + messages []tea.Msg + onSend func(tea.Msg) +} + +func (sender *recordingSender) Send(message tea.Msg) { + sender.messages = append(sender.messages, message) + if sender.onSend != nil { + sender.onSend(message) + } +} + +type recordingLogger struct { + errors []string +} + +func (logger *recordingLogger) Error(message string, _ ...any) { + logger.errors = append(logger.errors, message) +} + +func TestCollectTracesSurfacesStreamErrors(t *testing.T) { + tests := []struct { + name string + streamTraces func(context.Context, *pb.StreamTracesRequest) (traceStream, error) + expectedError string + expectedStates []events.ConnectionState + }{ + { + name: "stream setup", + streamTraces: func(_ context.Context, request *pb.StreamTracesRequest) (traceStream, error) { + assert.Equal(t, "staging", request.GetEnvironment()) + return nil, errors.New("unavailable") + }, + expectedError: "trace stream failed: unavailable", + expectedStates: []events.ConnectionState{ + events.ConnectionStateConnecting, + events.ConnectionStateRetrying, + }, + }, + { + name: "stream receive", + streamTraces: func(context.Context, *pb.StreamTracesRequest) (traceStream, error) { + return &fakeTraceStream{recv: func() (*pb.StreamTracesResponse, error) { + return nil, errors.New("connection lost") + }}, nil + }, + expectedError: "trace stream failed: connection lost", + expectedStates: []events.ConnectionState{ + events.ConnectionStateConnecting, + events.ConnectionStateConnected, + events.ConnectionStateRetrying, + }, + }, + { + name: "stream EOF", + streamTraces: func(context.Context, *pb.StreamTracesRequest) (traceStream, error) { + return &fakeTraceStream{recv: func() (*pb.StreamTracesResponse, error) { + return nil, io.EOF + }}, nil + }, + expectedError: "trace stream closed", + expectedStates: []events.ConnectionState{ + events.ConnectionStateConnecting, + events.ConnectionStateConnected, + events.ConnectionStateRetrying, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + api := &fakeCommandAPI{streamTraces: tt.streamTraces} + logger := &recordingLogger{} + sender := &recordingSender{} + sender.onSend = func(message tea.Msg) { + if connection, ok := message.(events.Connection); ok && connection.State == events.ConnectionStateRetrying { + cancel() + } + } + + err := collectTraces(ctx, api, "staging", sender, logger) + require.NoError(t, err) + assert.Equal(t, []string{tt.expectedError}, logger.errors) + + var states []events.ConnectionState + for _, message := range sender.messages { + if connection, ok := message.(events.Connection); ok { + states = append(states, connection.State) + if connection.State == events.ConnectionStateRetrying { + require.EqualError(t, connection.Err, tt.expectedError) + } + } + } + assert.Equal(t, tt.expectedStates, states) + }) + } +} + +func TestCollectTracesCancellationIsClean(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + sender := &recordingSender{} + logger := &recordingLogger{} + err := collectTraces(ctx, &fakeCommandAPI{}, "staging", sender, logger) + + require.NoError(t, err) + assert.Empty(t, sender.messages) + assert.Empty(t, logger.errors) +} + +func TestCollectTracesActiveStreamCancellationIsClean(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + api := &fakeCommandAPI{ + streamTraces: func(context.Context, *pb.StreamTracesRequest) (traceStream, error) { + return &fakeTraceStream{recv: func() (*pb.StreamTracesResponse, error) { + cancel() + return nil, context.Canceled + }}, nil + }, + } + + sender := &recordingSender{} + logger := &recordingLogger{} + err := collectTraces(ctx, api, "staging", sender, logger) + + require.NoError(t, err) + assert.Empty(t, logger.errors) + assert.Equal(t, []tea.Msg{ + events.Connection{State: events.ConnectionStateConnecting}, + events.Connection{State: events.ConnectionStateConnected}, + }, sender.messages) +} diff --git a/internal/command/trace/command.go b/internal/command/trace/command.go index e80fa47..17347a9 100644 --- a/internal/command/trace/command.go +++ b/internal/command/trace/command.go @@ -3,17 +3,13 @@ package trace import ( "context" "fmt" - "io" tea "github.com/charmbracelet/bubbletea" "github.com/skpr/api/pb" - "github.com/skpr/compass/tracing/cli/app" - "github.com/skpr/compass/tracing/cli/app/events" - applogger "github.com/skpr/compass/tracing/cli/app/logger" - "github.com/skpr/compass/tracing/trace" + "github.com/skpr/compass/pkg/app" + applogger "github.com/skpr/compass/pkg/app/logger" + compasstrace "github.com/skpr/compass/pkg/trace" "golang.org/x/sync/errgroup" - - "github.com/skpr/cli/internal/client" ) // Command to trace environments. @@ -23,7 +19,12 @@ type Command struct { // Run the command. func (cmd *Command) Run(ctx context.Context) error { - p := tea.NewProgram(app.NewModel(""), tea.WithAltScreen()) + ctx, api, err := cmd.preflight(ctx, connectAPI) + if err != nil { + return err + } + + p := tea.NewProgram(app.NewModel("", app.DefaultMaxTraces, app.DefaultMaxLogs), tea.WithAltScreen()) logger, err := applogger.New(p) if err != nil { @@ -31,78 +32,142 @@ func (cmd *Command) Run(ctx context.Context) error { } ctx, cancel := context.WithCancel(ctx) + defer cancel() eg := errgroup.Group{} // Start the collector. eg.Go(func() error { - logger.Info("Connecting to Skpr API...") - - ctx, client, err := client.New(ctx) - if err != nil { - return err - } - - stream, err := client.Trace().StreamTraces(ctx, &pb.StreamTracesRequest{ - Environment: cmd.Environment, - }) - if err != nil { - return err - } - - for { - select { - case <-ctx.Done(): - return nil - default: - resp, err := stream.Recv() - if err == io.EOF { - break - } - if err != nil { - return fmt.Errorf("streaming trace failed: %w", err) - } - - for _, t := range resp.Traces { - var fcalls []trace.FunctionCall - - for _, f := range t.FunctionCalls { - fcalls = append(fcalls, trace.FunctionCall{ - Name: f.Name, - StartTime: int64(f.StartTime.Nanos), - Elapsed: int64(f.Elapsed.Nanos), - }) - } - - p.Send(events.Trace{ - IngestionTime: t.Metadata.StartTime.AsTime(), - Trace: trace.Trace{ - Metadata: trace.Metadata{ - RequestID: t.Metadata.RequestId, - URI: t.Metadata.Uri, - Method: t.Metadata.Method, - StartTime: int64(t.Metadata.StartTime.Nanos), - EndTime: int64(t.Metadata.EndTime.Nanos), - }, - FunctionCalls: fcalls, - }, - }) - } - } - } + return collectTraces(ctx, api, cmd.Environment, p, logger) }) // Start the application. eg.Go(func() error { _, err := p.Run() + cancel() if err != nil { return fmt.Errorf("failed to run program: %w", err) } - cancel() - return nil }) return eg.Wait() } + +func traceFromProto(src *pb.Trace) compasstrace.Trace { + var dst compasstrace.Trace + if src == nil { + return dst + } + + if metadata := src.GetMetadata(); metadata != nil { + dst.Metadata.ID = metadata.GetRequestId() + dst.Metadata.Source = traceSourceFromProto(metadata.GetSource()) + dst.Metadata.Runtime = traceRuntimeFromProto(metadata.GetRuntime()) + + if startTime := metadata.GetStartTime(); startTime != nil { + dst.Metadata.StartTime = startTime.AsTime() + } + if endTime := metadata.GetEndTime(); endTime != nil { + dst.Metadata.EndTime = endTime.AsTime() + } + if httpMetadata := metadata.GetHttp(); httpMetadata != nil { + dst.Metadata.HTTP = compasstrace.MetadataHTTP{ + Method: httpMetadata.GetMethod(), + URI: httpMetadata.GetUri(), + } + } + if cliMetadata := metadata.GetCli(); cliMetadata != nil { + dst.Metadata.CLI.Command = cliMetadata.GetCommand() + } + } + + if resources := src.GetResourceUtilisation(); resources != nil { + dst.ResourceUtilisation.MaxMemory = resources.GetMaxMemory() + } + + dst.FunctionCalls = make([]compasstrace.FunctionCall, 0, len(src.GetFunctionCalls())) + for _, functionCall := range src.GetFunctionCalls() { + if functionCall == nil { + continue + } + + converted := compasstrace.FunctionCall{ + Name: functionCall.GetName(), + Memory: functionCall.GetMemory(), + } + if offset := functionCall.GetOffset(); offset != nil { + converted.Offset = offset.AsDuration() + } + if elapsed := functionCall.GetElapsed(); elapsed != nil { + converted.Elapsed = elapsed.AsDuration() + } + + dst.FunctionCalls = append(dst.FunctionCalls, converted) + } + dst.FunctionCallsDropped = int(src.GetFunctionCallsDropped()) + + if drupal := src.GetDrupal(); drupal != nil { + dst.Drupal = &compasstrace.Drupal{ + CacheEvents: make([]compasstrace.CacheEvent, 0, len(drupal.GetCacheEvents())), + CacheEventsDropped: int(drupal.GetCacheEventsDropped()), + } + + for _, cacheEvent := range drupal.GetCacheEvents() { + if cacheEvent == nil { + continue + } + + converted := compasstrace.CacheEvent{ + Origin: traceDrupalCacheOriginFromProto(cacheEvent.GetOrigin()), + Caller: cacheEvent.GetCaller(), + ObjectType: cacheEvent.GetObjectType(), + MaxAge: cacheEvent.GetMaxAge(), + Tags: append([]string(nil), cacheEvent.GetTags()...), + Contexts: append([]string(nil), cacheEvent.GetContexts()...), + Calls: cacheEvent.GetCalls(), + } + if offset := cacheEvent.GetOffset(); offset != nil { + converted.Offset = offset.AsDuration() + } + + dst.Drupal.CacheEvents = append(dst.Drupal.CacheEvents, converted) + } + } + + return dst +} + +func traceSourceFromProto(src pb.TraceSource) compasstrace.Source { + switch src { + case pb.TraceSource_TRACE_SOURCE_HTTP: + return compasstrace.SourceHTTP + case pb.TraceSource_TRACE_SOURCE_CLI: + return compasstrace.SourceCLI + default: + return "" + } +} + +func traceRuntimeFromProto(src pb.TraceRuntime) compasstrace.Runtime { + switch src { + case pb.TraceRuntime_TRACE_RUNTIME_PHP: + return compasstrace.RuntimePHP + case pb.TraceRuntime_TRACE_RUNTIME_NODE: + return compasstrace.RuntimeNode + default: + return "" + } +} + +func traceDrupalCacheOriginFromProto(src pb.TraceDrupalCacheOrigin) compasstrace.CacheOrigin { + switch src { + case pb.TraceDrupalCacheOrigin_TRACE_DRUPAL_CACHE_ORIGIN_RENDER_ARRAY: + return compasstrace.CacheOriginRenderArray + case pb.TraceDrupalCacheOrigin_TRACE_DRUPAL_CACHE_ORIGIN_OBJECT: + return compasstrace.CacheOriginObject + default: + return "" + } +} diff --git a/internal/command/trace/command_test.go b/internal/command/trace/command_test.go new file mode 100644 index 0000000..ed10c95 --- /dev/null +++ b/internal/command/trace/command_test.go @@ -0,0 +1,114 @@ +package trace + +import ( + "testing" + "time" + + "github.com/skpr/api/pb" + compasstrace "github.com/skpr/compass/pkg/trace" + "github.com/stretchr/testify/assert" + "google.golang.org/protobuf/types/known/durationpb" + "google.golang.org/protobuf/types/known/timestamppb" +) + +func TestTraceFromProto(t *testing.T) { + startTime := time.Date(2026, time.September, 4, 12, 0, 0, 0, time.UTC) + endTime := startTime.Add(1500 * time.Millisecond) + + actual := traceFromProto(&pb.Trace{ + Metadata: &pb.TraceMetadata{ + RequestId: "request-123", + StartTime: timestamppb.New(startTime), + EndTime: timestamppb.New(endTime), + Source: pb.TraceSource_TRACE_SOURCE_HTTP, + Runtime: pb.TraceRuntime_TRACE_RUNTIME_PHP, + Http: &pb.TraceMetadataHTTP{ + Method: "GET", + Uri: "/articles/123", + }, + }, + ResourceUtilisation: &pb.TraceResourceUtilisation{MaxMemory: 4096}, + FunctionCalls: []*pb.TraceFunctionCall{ + { + Name: "Drupal\\Core\\Kernel::handle", + Offset: durationpb.New(25 * time.Millisecond), + Elapsed: durationpb.New(750 * time.Millisecond), + Memory: 1024, + }, + nil, + }, + FunctionCallsDropped: 2, + Drupal: &pb.TraceDrupal{ + CacheEvents: []*pb.TraceDrupalCacheEvent{ + { + Origin: pb.TraceDrupalCacheOrigin_TRACE_DRUPAL_CACHE_ORIGIN_OBJECT, + Caller: "Drupal\\node\\Entity\\Node::getCacheTags", + ObjectType: "node", + MaxAge: 3600, + Tags: []string{"node:123"}, + Contexts: []string{"url.path"}, + Offset: durationpb.New(100 * time.Millisecond), + Calls: 3, + }, + nil, + }, + CacheEventsDropped: 1, + }, + }) + + expected := compasstrace.Trace{ + Metadata: compasstrace.Metadata{ + Source: compasstrace.SourceHTTP, + Runtime: compasstrace.RuntimePHP, + ID: "request-123", + HTTP: compasstrace.MetadataHTTP{Method: "GET", URI: "/articles/123"}, + StartTime: startTime, + EndTime: endTime, + }, + ResourceUtilisation: compasstrace.ResourceUtilisation{MaxMemory: 4096}, + FunctionCalls: []compasstrace.FunctionCall{ + { + Name: "Drupal\\Core\\Kernel::handle", + Offset: 25 * time.Millisecond, + Elapsed: 750 * time.Millisecond, + Memory: 1024, + }, + }, + FunctionCallsDropped: 2, + Drupal: &compasstrace.Drupal{ + CacheEvents: []compasstrace.CacheEvent{ + { + Origin: compasstrace.CacheOriginObject, + Caller: "Drupal\\node\\Entity\\Node::getCacheTags", + ObjectType: "node", + MaxAge: 3600, + Tags: []string{"node:123"}, + Contexts: []string{"url.path"}, + Offset: 100 * time.Millisecond, + Calls: 3, + }, + }, + CacheEventsDropped: 1, + }, + } + + assert.Equal(t, expected, actual) +} + +func TestTraceFromProtoCLI(t *testing.T) { + actual := traceFromProto(&pb.Trace{ + Metadata: &pb.TraceMetadata{ + Source: pb.TraceSource_TRACE_SOURCE_CLI, + Runtime: pb.TraceRuntime_TRACE_RUNTIME_NODE, + Cli: &pb.TraceMetadataCLI{Command: "drush cr"}, + }, + }) + + assert.Equal(t, compasstrace.SourceCLI, actual.Metadata.Source) + assert.Equal(t, compasstrace.RuntimeNode, actual.Metadata.Runtime) + assert.Equal(t, "drush cr", actual.Metadata.CLI.Command) +} + +func TestTraceFromProtoNil(t *testing.T) { + assert.Equal(t, compasstrace.Trace{}, traceFromProto(nil)) +}