Skip to content

Commit e14daca

Browse files
adameatadameat
authored
viewer 270325 - merge of improvements to query http streaming handler - stable-25-1 (#16307)
1 parent bc3026c commit e14daca

File tree

3 files changed

+75
-24
lines changed

3 files changed

+75
-24
lines changed

‎ydb/core/viewer/json_handlers_viewer.cpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -230,7 +230,7 @@ void InitViewerWhoAmIJsonHandler(TJsonHandlers& handlers) {
230230
}
231231

232232
void InitViewerQueryJsonHandler(TJsonHandlers& handlers) {
233-
handlers.AddHandler("/viewer/query", new THttpHandler<TJsonQuery>(TJsonQuery::GetSwagger()), 7);
233+
handlers.AddHandler("/viewer/query", new THttpHandler<TJsonQuery>(TJsonQuery::GetSwagger()), 8);
234234
}
235235

236236
void InitViewerNetInfoJsonHandler(TJsonHandlers& handlers) {

‎ydb/core/viewer/tests/canondata/result.json

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -3014,7 +3014,7 @@
30143014
]
30153015
}
30163016
],
3017-
"version": 7
3017+
"version": 8
30183018
},
30193019
"/Root/dedicated_db": {
30203020
"result": [
@@ -3032,7 +3032,7 @@
30323032
]
30333033
}
30343034
],
3035-
"version": 7
3035+
"version": 8
30363036
},
30373037
"/Root/serverless_db": {
30383038
"result": [
@@ -3050,7 +3050,7 @@
30503050
]
30513051
}
30523052
],
3053-
"version": 7
3053+
"version": 8
30543054
},
30553055
"/Root/shared_db": {
30563056
"result": [
@@ -3068,7 +3068,7 @@
30683068
]
30693069
}
30703070
],
3071-
"version": 7
3071+
"version": 8
30723072
}
30733073
},
30743074
"test.test_viewer_query_issue_13757": {
@@ -3091,7 +3091,7 @@
30913091
]
30923092
}
30933093
],
3094-
"version": 7
3094+
"version": 8
30953095
},
30963096
"/Root/dedicated_db": {
30973097
"result": [
@@ -3112,7 +3112,7 @@
31123112
]
31133113
}
31143114
],
3115-
"version": 7
3115+
"version": 8
31163116
},
31173117
"/Root/serverless_db": {
31183118
"result": [
@@ -3133,7 +3133,7 @@
31333133
]
31343134
}
31353135
],
3136-
"version": 7
3136+
"version": 8
31373137
},
31383138
"/Root/shared_db": {
31393139
"result": [
@@ -3154,7 +3154,7 @@
31543154
]
31553155
}
31563156
],
3157-
"version": 7
3157+
"version": 8
31583158
}
31593159
},
31603160
"test.test_viewer_query_issue_13945": {
@@ -3174,7 +3174,7 @@
31743174
]
31753175
}
31763176
],
3177-
"version": 7
3177+
"version": 8
31783178
},
31793179
"/Root/dedicated_db": {
31803180
"result": [
@@ -3192,7 +3192,7 @@
31923192
]
31933193
}
31943194
],
3195-
"version": 7
3195+
"version": 8
31963196
},
31973197
"/Root/serverless_db": {
31983198
"result": [
@@ -3210,7 +3210,7 @@
32103210
]
32113211
}
32123212
],
3213-
"version": 7
3213+
"version": 8
32143214
},
32153215
"/Root/shared_db": {
32163216
"result": [
@@ -3228,7 +3228,7 @@
32283228
]
32293229
}
32303230
],
3231-
"version": 7
3231+
"version": 8
32323232
}
32333233
},
32343234
"test.test_viewer_sysinfo": {

‎ydb/core/viewer/viewer_query.h

Lines changed: 62 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -17,8 +17,6 @@ using namespace NNodeWhiteboard;
1717
class TJsonQuery : public TViewerPipeClient {
1818
using TThis = TJsonQuery;
1919
using TBase = TViewerPipeClient;
20-
TJsonSettings JsonSettings;
21-
ui32 Timeout = 60000;
2220
std::vector<std::vector<Ydb::ResultSet>> ResultSets;
2321
TString Query;
2422
TString Action;
@@ -30,6 +28,11 @@ class TJsonQuery : public TViewerPipeClient {
3028
bool IsBase64Encode = true;
3129
int LimitRows = 10000;
3230
int TotalRows = 0;
31+
bool CollectDiagnostics = true;
32+
TDuration StatsPeriod;
33+
TDuration KeepAlive = TDuration::MilliSeconds(10000);
34+
TInstant LastSendTime;
35+
static constexpr TDuration WakeupPeriod = TDuration::Seconds(1);
3336

3437
enum ESchemaType {
3538
Classic,
@@ -66,10 +69,8 @@ class TJsonQuery : public TViewerPipeClient {
6669
}
6770
}
6871

69-
void ParseCgiParameters(const TCgiParameters& params) {
70-
JsonSettings.EnumAsNumbers = !FromStringWithDefault<bool>(params.Get("enums"), false);
71-
JsonSettings.UI64AsString = !FromStringWithDefault<bool>(params.Get("ui64"), false);
72-
Timeout = FromStringWithDefault<ui32>(params.Get("timeout"), Timeout);
72+
void InitConfig(const TCgiParameters& params) {
73+
Timeout = TDuration::MilliSeconds(FromStringWithDefault<ui32>(params.Get("timeout"), 60000)); // override default timeout to 60 seconds
7374
if (params.Has("query")) {
7475
Query = params.Get("query");
7576
}
@@ -89,6 +90,9 @@ class TJsonQuery : public TViewerPipeClient {
8990
if (params.Has("concurrent_results")) {
9091
ConcurrentResults = FromStringWithDefault<bool>(params.Get("concurrent_results"), ConcurrentResults);
9192
}
93+
if (params.Has("keep_alive")) {
94+
KeepAlive = TDuration::MilliSeconds(FromStringWithDefault<ui32>(params.Get("keep_alive"), KeepAlive.MilliSeconds()));
95+
}
9296
}
9397
}
9498
if (params.Has("syntax")) {
@@ -112,18 +116,22 @@ class TJsonQuery : public TViewerPipeClient {
112116
if (params.Has("output_chunk_max_size")) {
113117
OutputChunkMaxSize = FromStringWithDefault<ui64>(params.Get("output_chunk_max_size"), OutputChunkMaxSize);
114118
}
119+
CollectDiagnostics = FromStringWithDefault<bool>(params.Get("collect_diagnostics"), CollectDiagnostics);
120+
if (params.Has("stats_period")) {
121+
StatsPeriod = TDuration::MilliSeconds(std::clamp<ui64>(FromStringWithDefault<ui64>(params.Get("stats_period"), StatsPeriod.MilliSeconds()), 1000, 600000));
122+
}
115123
}
116124

117125
TJsonQuery(IViewer* viewer, NHttp::TEvHttpProxy::TEvHttpIncomingRequest::TPtr& ev)
118126
: TBase(viewer, ev)
119127
{
128+
InitConfig(Params);
120129
}
121130

122131
void Bootstrap() override {
123132
if (NeedToRedirect()) {
124133
return;
125134
}
126-
ParseCgiParameters(Params);
127135
if (Query.empty() && Action != "cancel-query") {
128136
return TBase::ReplyAndPassAway(GetHTTPBADREQUEST("text/plain", "Query is empty"), "EmptyQuery");
129137
}
@@ -147,7 +155,11 @@ class TJsonQuery : public TViewerPipeClient {
147155
}
148156
Send(HttpEvent->Sender, new NHttp::TEvHttpProxy::TEvSubscribeForCancel(), IEventHandle::FlagTrackDelivery);
149157
SendKpqProxyRequest();
150-
Become(&TThis::StateWork, TDuration::MilliSeconds(Timeout), new TEvents::TEvWakeup());
158+
Become(&TThis::StateWork);
159+
if (Timeout || KeepAlive) {
160+
Schedule(WakeupPeriod, new TEvents::TEvWakeup());
161+
}
162+
LastSendTime = TActivationContext::Now();
151163
}
152164

153165
void CancelQuery() {
@@ -203,7 +215,7 @@ class TJsonQuery : public TViewerPipeClient {
203215
hFunc(NKqp::TEvKqpExecuter::TEvStreamData, HandleReply);
204216
cFunc(NHttp::TEvHttpProxy::EvRequestCancelled, Cancelled);
205217
hFunc(TEvents::TEvUndelivered, Undelivered);
206-
cFunc(TEvents::TSystem::Wakeup, HandleTimeout);
218+
cFunc(TEvents::TSystem::Wakeup, HandleWakeup);
207219
}
208220
}
209221

@@ -374,6 +386,10 @@ class TJsonQuery : public TViewerPipeClient {
374386
if (OutputChunkMaxSize) {
375387
request.SetOutputChunkMaxSize(OutputChunkMaxSize);
376388
}
389+
request.SetCollectDiagnostics(CollectDiagnostics);
390+
if (StatsPeriod) {
391+
event->SetProgressStatsPeriod(StatsPeriod);
392+
}
377393
ActorIdToProto(SelfId(), event->Record.MutableRequestActorId());
378394
QueryResponse = MakeRequest<NKqp::TEvKqp::TEvQueryResponse>(NKqp::MakeKqpProxyID(SelfId().NodeId()), event.Release());
379395

@@ -656,8 +672,15 @@ class TJsonQuery : public TViewerPipeClient {
656672
ReplyWithJsonAndPassAway(json);
657673
}
658674

659-
void HandleTimeout() {
660-
ReplyWithError("Timeout executing query");
675+
void HandleWakeup() {
676+
auto now = TActivationContext::Now();
677+
if (Timeout && (now - LastSendTime > Timeout)) {
678+
return ReplyWithError("Timeout executing query");
679+
}
680+
if (KeepAlive && (now - LastSendTime > KeepAlive)) {
681+
SendKeepAlive();
682+
}
683+
Schedule(WakeupPeriod, new TEvents::TEvWakeup());
661684
}
662685

663686
private:
@@ -853,6 +876,7 @@ class TJsonQuery : public TViewerPipeClient {
853876
data << "--boundary\r\nContent-Type: application/json\r\nContent-Length: " << content.Size() << "\r\n\r\n" << content.Str() << "\r\n";
854877
auto dataChunk = HttpResponse->CreateDataChunk(data);
855878
Send(HttpEvent->Sender, new NHttp::TEvHttpProxy::TEvHttpOutgoingDataChunk(dataChunk));
879+
LastSendTime = TActivationContext::Now();
856880
}
857881

858882
void StreamJsonResponse(const NKikimrKqp::TEvExecuterStreamData& data) {
@@ -876,6 +900,19 @@ class TJsonQuery : public TViewerPipeClient {
876900
auto dataChunk = HttpResponse->CreateDataChunk("--boundary--\r\n");
877901
dataChunk->SetEndOfData();
878902
Send(HttpEvent->Sender, new NHttp::TEvHttpProxy::TEvHttpOutgoingDataChunk(dataChunk));
903+
LastSendTime = TActivationContext::Now();
904+
}
905+
906+
void SendKeepAlive() {
907+
if (Streaming) {
908+
NJson::TJsonValue json;
909+
NJson::TJsonValue& jsonMeta = json["meta"];
910+
jsonMeta["event"] = "KeepAlive";
911+
StreamJsonResponse(json);
912+
}
913+
if (SessionId) {
914+
PingSession();
915+
}
879916
}
880917

881918
void ReplyWithJsonAndPassAway(const NJson::TJsonValue& json, const TString& error = {}) {
@@ -1011,6 +1048,20 @@ class TJsonQuery : public TViewerPipeClient {
10111048
description: resource pool in which the query will be executed
10121049
type: string
10131050
required: false
1051+
- name: keep_alive
1052+
in: query
1053+
description: time of inactivity to send keep-alive in stream (multipart) queries
1054+
type: integer
1055+
default: 10000
1056+
- name: collect_diagnostics
1057+
in: query
1058+
description: collect query diagnostics
1059+
type: boolean
1060+
default: true
1061+
- name: stats_period
1062+
in: query
1063+
description: time interval for sending periodical query statistics in ms
1064+
type: integer
10141065
requestBody:
10151066
description: Executes SQL query
10161067
required: false

0 commit comments

Comments
 (0)