Skip to content

feat(async search): add export async search docs method #58

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
forshev wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(async search): add export async search docs method #58

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions api/seqapi/v1/seq_api.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -213,6 +213,14 @@ message ExportRequest {
uint32 downsample = 8;
}

message ExportAsyncSearchRequest {
string search_id = 1;
int32 limit = 4;
int32 offset = 5;
ExportFormat format = 6;
repeated string fields = 7;
}

message GetLimitsRequest {}

message GetLimitsResponse {
Expand Down
1 change: 1 addition & 0 deletions internal/api/seqapi/v1/http/api.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -160,6 +160,7 @@ func (a *API) Router() chi.Router {
mux.Post("/async_search/list", a.serveGetAsyncSearchesList)
mux.Post("/async_search/{id}/cancel", a.serveCancelAsyncSearch)
mux.Delete("/async_search/{id}", a.serveDeleteAsyncSearch)
mux.Post("/async_search/export", a.serveExportAsyncSearch)

return mux
}
Expand Down
150 changes: 150 additions & 0 deletions internal/api/seqapi/v1/http/export_async_search.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,150 @@
package http

import (
"encoding/json"
"errors"
"fmt"
"net/http"

"go.opentelemetry.io/otel/attribute"

"github.com/ozontech/seq-ui/internal/api/httputil"
"github.com/ozontech/seq-ui/internal/app/types"
"github.com/ozontech/seq-ui/metric"
"github.com/ozontech/seq-ui/pkg/seqapi/v1"
"github.com/ozontech/seq-ui/tracing"
)

// serveExportAsyncSearch go doc.
//
// @Router /seqapi/v1/async_search/export [post]
// @ID seqapi_v1_export_async_search
// @Tags seqapi_v1
// @Param env query string false "Environment"
// @Param body body exportAsyncSearchRequest true "Request body"
// @Success 200 {object} exportResponse "A successful streaming responses"
// @Failure default {object} httputil.Error "An unexpected error response"
// @Security bearer
func (a *API) serveExportAsyncSearch(w http.ResponseWriter, r *http.Request) {
wr := httputil.NewWriter(w)

if a.asyncSearches == nil {
wr.Error(types.ErrAsyncSearchesDisabled, http.StatusBadRequest)
return
}

ctx, span := tracing.StartSpan(r.Context(), "seqapi_v1_export_async_search_http")
defer span.End()

env := getEnvFromContext(ctx)

userStr := "_"
if userName, err := types.GetUserKey(ctx); err == nil {
userStr = userName
}

var httpReq exportAsyncSearchRequest
if err := json.NewDecoder(r.Body).Decode(&httpReq); err != nil {
wr.Error(fmt.Errorf("failed to parse export async search request: %w", err), http.StatusBadRequest)
return
}

if httpReq.Format == "" {
httpReq.Format = efJSONL
}

attributes := []attribute.KeyValue{
{
Key: "search_id",
Value: attribute.StringValue(httpReq.SearchID),
},
{
Key: "limit",
Value: attribute.IntValue(int(httpReq.Limit)),
},
{
Key: "offset",
Value: attribute.IntValue(int(httpReq.Offset)),
},
{
Key: "format",
Value: attribute.StringValue(string(httpReq.Format)),
},
{
Key: "fields",
Value: attribute.StringSliceValue(httpReq.Fields),
},
}

if env != "" {
attributes = append(attributes, attribute.String("env", env))
}

span.SetAttributes(attributes...)

if err := checkUUID(httpReq.SearchID); err != nil {
wr.Error(err, http.StatusBadRequest)
return
}

if err := checkLimitOffset(httpReq.Limit, httpReq.Offset); err != nil {
wr.Error(err, http.StatusBadRequest)
return
}

params, err := a.GetEnvParams(env)
if err != nil {
wr.Error(err, http.StatusBadRequest)
return
}

if params.exportLimiter.Limited(userStr) {
metric.ServerExportRequestLimits.Inc()
wr.Error(errors.New("parallel export limit exceeded"), http.StatusTooManyRequests)
return
}
defer params.exportLimiter.Fill(userStr)

if httpReq.Limit > params.options.MaxExportLimit {
wr.Error(fmt.Errorf("too many events are requested: count=%d, max=%d",
httpReq.Limit, params.options.MaxExportLimit),
http.StatusBadRequest)
return
}

if httpReq.Format == efCSV && len(httpReq.Fields) == 0 {
wr.Error(errors.New("csv export required 'fields'"), http.StatusBadRequest)
return
}

cw, err := httputil.NewChunkedWriter(wr.ResponseWriter)
if err != nil {
wr.Error(err, http.StatusBadRequest)
return
}

if err := a.asyncSearches.ExportAsyncSearch(ctx, httpReq.toProto(), cw); err != nil {
wr.Error(err, http.StatusInternalServerError)
return
}

wr.WriteHeader(http.StatusOK)
}

type exportAsyncSearchRequest struct {
SearchID string `json:"search_id" format:"uuid"`
Limit int32 `json:"limit" format:"int32"`
Offset int32 `json:"offset" format:"int32"`
Format exportFormat `json:"format" default:"jsonl"`
Fields []string `json:"fields,omitempty"`
} // @name seqapi.v1.ExportAsyncSearchRequest

func (r exportAsyncSearchRequest) toProto() *seqapi.ExportAsyncSearchRequest {
return &seqapi.ExportAsyncSearchRequest{
SearchId: r.SearchID,
Limit: r.Limit,
Offset: r.Offset,
Format: r.Format.toProto(),
Fields: r.Fields,
}
}
207 changes: 207 additions & 0 deletions internal/api/seqapi/v1/http/export_async_search_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,207 @@
package http

import (
"net/http"
"testing"

"go.uber.org/mock/gomock"

"github.com/ozontech/seq-ui/internal/api/httputil"
"github.com/ozontech/seq-ui/internal/api/seqapi/v1/test"
"github.com/ozontech/seq-ui/internal/app/config"
mock_asyncsearches "github.com/ozontech/seq-ui/internal/pkg/service/async_searches/mock"
"github.com/ozontech/seq-ui/pkg/seqapi/v1"
)

func TestServeExportAsyncSearch(t *testing.T) {
type mockArgs struct {
req *seqapi.ExportAsyncSearchRequest
err error
}

tests := []struct {
name string

req exportAsyncSearchRequest
cfg config.SeqAPI
wantErr bool

mockArgs *mockArgs
}{
{
name: "ok_jsonl",
req: exportAsyncSearchRequest{
SearchID: testSearchID,
Limit: 50,
Offset: 0,
},
cfg: config.SeqAPI{
SeqAPIOptions: &config.SeqAPIOptions{
MaxExportLimit: 100,
MaxParallelExportRequests: 1,
},
},
mockArgs: &mockArgs{
req: &seqapi.ExportAsyncSearchRequest{
SearchId: testSearchID,
Limit: 50,
Offset: 0,
Format: seqapi.ExportFormat_EXPORT_FORMAT_JSONL,
},
},
},
{
name: "ok_csv",
req: exportAsyncSearchRequest{
SearchID: testSearchID,
Limit: 50,
Offset: 0,
Format: efCSV,
Fields: []string{"field1", "field2"},
},
cfg: config.SeqAPI{
SeqAPIOptions: &config.SeqAPIOptions{
MaxExportLimit: 100,
MaxParallelExportRequests: 1,
},
},
mockArgs: &mockArgs{
req: &seqapi.ExportAsyncSearchRequest{
SearchId: testSearchID,
Limit: 50,
Offset: 0,
Format: seqapi.ExportFormat_EXPORT_FORMAT_CSV,
Fields: []string{"field1", "field2"},
},
},
},
{
name: "err_parallel_limited",
req: exportAsyncSearchRequest{
SearchID: testSearchID,
Limit: 0,
Offset: 0,
},
cfg: config.SeqAPI{
SeqAPIOptions: &config.SeqAPIOptions{
MaxParallelExportRequests: 0,
},
},
wantErr: true,
},
{
name: "err_export_limit_max",
req: exportAsyncSearchRequest{
SearchID: testSearchID,
Limit: 10,
Offset: 0,
},
cfg: config.SeqAPI{
SeqAPIOptions: &config.SeqAPIOptions{
MaxExportLimit: 5,
MaxParallelExportRequests: 1,
},
},
wantErr: true,
},
{
name: "err_csv_empty_fields",
req: exportAsyncSearchRequest{
SearchID: testSearchID,
Limit: 10,
Offset: 0,
Format: efCSV,
},
cfg: config.SeqAPI{
SeqAPIOptions: &config.SeqAPIOptions{
MaxExportLimit: 100,
MaxParallelExportRequests: 1,
},
},
wantErr: true,
},
{
name: "err_invalid_id",
req: exportAsyncSearchRequest{
SearchID: "some invalid id",
Limit: 10,
Offset: 0,
},
cfg: config.SeqAPI{
SeqAPIOptions: &config.SeqAPIOptions{
MaxExportLimit: 100,
MaxParallelExportRequests: 1,
},
},
wantErr: true,
},
{
name: "err_client",
req: exportAsyncSearchRequest{
SearchID: testSearchID,
Limit: 50,
Offset: 0,
},
cfg: config.SeqAPI{
SeqAPIOptions: &config.SeqAPIOptions{
MaxExportLimit: 100,
MaxParallelExportRequests: 1,
},
},
wantErr: true,
mockArgs: &mockArgs{
req: &seqapi.ExportAsyncSearchRequest{
SearchId: testSearchID,
Limit: 50,
Offset: 0,
Format: seqapi.ExportFormat_EXPORT_FORMAT_JSONL,
},
err: errSomethingWrong,
},
},
}

for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
t.Parallel()

ctrl := gomock.NewController(t)
svcMock := mock_asyncsearches.NewMockService(ctrl)

seqData := test.APITestData{
Cfg: tt.cfg,
}
seqData.Mocks.AsyncSearchesSvc = svcMock

if tt.mockArgs != nil {
svcMock.EXPECT().
ExportAsyncSearch(gomock.Any(), tt.mockArgs.req, gomock.Any()).
Return(tt.mockArgs.err).
Times(1)
}

api := setupTestAPI(seqData)

httputil.DoTestHTTPEx(t, httputil.TestDataHTTPEx[exportAsyncSearchRequest, struct{}]{
Method: http.MethodPost,
Target: "/seqapi/v1/async_search/export",
Req: tt.req,
Handler: api.serveExportAsyncSearch,
WantErr: tt.wantErr,
NoResp: true,
})
})
}
}

func TestServeExportAsyncSearch_Disabled(t *testing.T) {
seqData := test.APITestData{}
api := setupTestAPI(seqData)

httputil.DoTestHTTPEx(t, httputil.TestDataHTTPEx[exportAsyncSearchRequest, struct{}]{
Method: http.MethodPost,
Target: "/seqapi/v1/async_search/export",
Handler: api.serveExportAsyncSearch,
WantErr: true,
})
}
1 change: 1 addition & 0 deletions internal/pkg/client/seqdb/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ type Client interface {
Search(context.Context, *seqapi.SearchRequest) (*seqapi.SearchResponse, error)
Status(context.Context, *seqapi.StatusRequest) (*seqapi.StatusResponse, error)
Export(context.Context, *seqapi.ExportRequest, *httputil.ChunkedWriter) error
ExportAsyncSearch(context.Context, *seqapi.ExportAsyncSearchRequest, *httputil.ChunkedWriter) error
StartAsyncSearch(context.Context, *seqapi.StartAsyncSearchRequest) (*seqapi.StartAsyncSearchResponse, error)
FetchAsyncSearchResult(context.Context, *seqapi.FetchAsyncSearchResultRequest) (*seqapi.FetchAsyncSearchResultResponse, error)
GetAsyncSearchesList(context.Context, *seqapi.GetAsyncSearchesListRequest, []string) (*seqapi.GetAsyncSearchesListResponse, error)
Expand Down
Loading
Loading