diff --git a/proxyapi/grpc_complex_search.go b/proxyapi/grpc_complex_search.go index 850c4ab0..ce8aa03d 100644 --- a/proxyapi/grpc_complex_search.go +++ b/proxyapi/grpc_complex_search.go @@ -99,7 +99,7 @@ func (g *grpcV1) useStreamSearch( streamSearchReq, err := buildStreamSearchReqFromComplexSearchReq(req) if err != nil { - return nil, status.Error(codes.InvalidArgument, `can't build stream search request`) + return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("can't build stream search request: %s", err.Error())) } storesStream, err := g.openStoreStreams(ctx, streamSearchReq, tr) @@ -206,6 +206,16 @@ func shouldUseStreamSearch(useStreamSearch bool, req *seqproxyapi.ComplexSearchR func buildStreamSearchReqFromComplexSearchReq( req *seqproxyapi.ComplexSearchRequest, ) (*search.StreamSearchRequest, error) { + if req.Query == nil { + return nil, status.Error(codes.InvalidArgument, "search query must be provided") + } + if req.Query.From == nil || req.Query.To == nil { + return nil, status.Error(codes.InvalidArgument, `search query "from" and "to" fields must be provided`) + } + if req.Query.From.AsTime().After(req.Query.To.AsTime()) { + return nil, status.Error(codes.InvalidArgument, `"from" timestamp must not be after "to" timestamp`) + } + seqql, err := parser.ParseSeqQL(req.Query.Query, nil) if err != nil { return nil, fmt.Errorf("parse query: %w", err) diff --git a/proxyapi/grpc_complex_search_stream_test.go b/proxyapi/grpc_complex_search_stream_test.go index 95501487..062671d8 100644 --- a/proxyapi/grpc_complex_search_stream_test.go +++ b/proxyapi/grpc_complex_search_stream_test.go @@ -119,6 +119,42 @@ func TestBuildStreamSearchReqFromComplexSearchReq(t *testing.T) { }, }, }, + { + name: "missing_from", + req: &seqproxyapi.ComplexSearchRequest{ + Query: &seqproxyapi.SearchQuery{ + Query: "message:ok", + To: timestamppb.New(now.Add(time.Hour)), + }, + }, + wantErr: true, + }, + { + name: "missing_to", + req: &seqproxyapi.ComplexSearchRequest{ + Query: &seqproxyapi.SearchQuery{ + Query: "message:ok", + From: timestamppb.New(now), + }, + }, + wantErr: true, + }, + { + name: "nil_query", + req: &seqproxyapi.ComplexSearchRequest{}, + wantErr: true, + }, + { + name: "from_after_to", + req: &seqproxyapi.ComplexSearchRequest{ + Query: &seqproxyapi.SearchQuery{ + Query: "message:ok", + From: timestamppb.New(now.Add(time.Hour)), + To: timestamppb.New(now), + }, + }, + wantErr: true, + }, { name: "invalid_query", req: &seqproxyapi.ComplexSearchRequest{ diff --git a/proxyapi/grpc_stream_search.go b/proxyapi/grpc_stream_search.go index 20c80cb2..919e7a52 100644 --- a/proxyapi/grpc_stream_search.go +++ b/proxyapi/grpc_stream_search.go @@ -42,7 +42,7 @@ func (g *grpcV1) StreamSearch(stream seqproxyapi.SeqProxyApi_StreamSearchServer) // The first message must carry the search query. req, err := stream.Recv() if err == io.EOF { - return nil + return status.Error(codes.InvalidArgument, "no search query provided") } if err != nil { return err @@ -385,6 +385,13 @@ func (m aggRecordMapper) toRecord(r *query.Record) *seqproxyapi.Record { } func buildSearchReq(q *seqproxyapi.StreamSearchQuery) (*search.StreamSearchRequest, error) { + if q.From == nil || q.To == nil { + return nil, status.Error(codes.InvalidArgument, `search query "from" and "to" fields must be provided`) + } + if q.From.AsTime().After(q.To.AsTime()) { + return nil, status.Error(codes.InvalidArgument, `"from" timestamp must not be after "to" timestamp`) + } + seqql, err := parser.ParseSeqQL(q.Query, nil) if err != nil { return nil, err diff --git a/proxyapi/grpc_stream_search_test.go b/proxyapi/grpc_stream_search_test.go index eabfe8c8..fa6e9490 100644 --- a/proxyapi/grpc_stream_search_test.go +++ b/proxyapi/grpc_stream_search_test.go @@ -4,12 +4,14 @@ import ( "context" "errors" "fmt" + "io" "strconv" "testing" "github.com/golang/mock/gomock" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" "google.golang.org/grpc/status" @@ -236,3 +238,24 @@ func failPartialCtx(fail bool) context.Context { return metadata.NewIncomingContext(context.Background(), metadata.Pairs("fail-partial-response", strconv.FormatBool(fail))) } + +type emptyStream struct { + grpc.ServerStream +} + +func (s emptyStream) Context() context.Context { return context.Background() } + +func (s emptyStream) Send(*seqproxyapi.StreamSearchResponse) error { return nil } + +func (s emptyStream) Recv() (*seqproxyapi.StreamSearchRequest, error) { + return nil, io.EOF +} + +func TestGrpcV1_StreamSearchEmptyRequest(t *testing.T) { + g := newGrpcV1(APIConfig{}, nil, nil, nil, nil) + + err := g.StreamSearch(emptyStream{}) + + require.Error(t, err) + assert.Equal(t, codes.InvalidArgument, status.Code(err)) +}