@cryptotaxi247 / netdata-1 / commits / f6d9792fe

logs-management: Add function cancellability (#16484)

* Add function cancellability * Do not comment out removal of persistent dirs in run_stress_test.sh

Dimitris P committed Nov 28, 2023 at 15:55 UTC f6d9792fe8a985b96e097cc7c0dbd7d9d876fb07

5 files changed +41 -13

logsmanagement/circular_buffer.c

+1 -1

@@ -91,7 +91,7 @@ void circ_buff_search(logs_query_params_t *const p_query_params, struct File_inf

92 /* If exceeding quota or timeout is reached and new timestamp is different than previous,

93 * terminate query but inform caller about act_to_ts to continue from (its next value) in next call. */

94 - if((res_buff->len >= p_query_params->quota || now_monotonic_usec() > p_query_params->stop_monotonic_ut) &&

94 + if( (res_buff->len >= p_query_params->quota || terminate_logs_manag_query(p_query_params)) &&

95 items[i].cbi->timestamp != res_hdr.timestamp){

96 p_query_params->act_to_ts = res_hdr.timestamp;

97 break;

logsmanagement/db_api.c

+1 -1

@@ -1288,7 +1288,7 @@ void db_search(logs_query_params_t *const p_query_params, struct File_info *cons

1288

1289 /* If exceeding quota or timeout is reached and new timestamp

1290 * is different than previous, terminate query. */

1291 - if((res_buff->len >= p_query_params->quota || now_monotonic_usec() > p_query_params->stop_monotonic_ut) &&

1291 + if((res_buff->len >= p_query_params->quota || terminate_logs_manag_query(p_query_params)) &&

1292 tmp_itm.timestamp != res_hdr.timestamp){

1293 p_query_params->act_to_ts = res_hdr.timestamp;

1294 break;

logsmanagement/functions.c

+4 -5

@@ -121,7 +121,6 @@ typedef struct function_query_status {

121 usec_t started_monotonic_ut;

122

123 // request

124 - // SD_JOURNAL_FILE_SOURCE_TYPE source_type;

124 STRING *source;

125 usec_t after_ut;

126 usec_t before_ut;

@@ -249,7 +248,6 @@ static void logsmanagement_function_facets(const char *transaction, char *functi

248 const char *chart = NULL;

249 const char *source = NULL;

250 const char *progress_id = NULL;

252 - // SD_JOURNAL_FILE_SOURCE_TYPE source_type = SDJF_ALL;

251 // size_t filters = 0;

252

253 buffer_json_member_add_object(wb, "_request");

@@ -444,7 +442,6 @@ static void logsmanagement_function_facets(const char *transaction, char *functi

442 fqs->delta = (fqs->data_only) ? delta : false;

443 fqs->tail = (fqs->data_only && fqs->if_modified_since) ? tail : false;

444 fqs->source = string_strdupz(source);

447 - // fqs->source_type = source_type;

445 fqs->entries = last;

446 fqs->last_modified = 0;

447 // fqs->filters = filters;

@@ -516,7 +513,6 @@ static void logsmanagement_function_facets(const char *transaction, char *functi

513 buffer_json_member_add_boolean(wb, LOGS_MANAG_FUNC_PARAM_TAIL, fqs->tail);

514 buffer_json_member_add_string(wb, LOGS_MANAG_FUNC_PARAM_ID, progress_id);

515 buffer_json_member_add_string(wb, LOGS_MANAG_FUNC_PARAM_SOURCE, string2str(fqs->source));

519 - // buffer_json_member_add_uint64(wb, "source_type", fqs->source_type);

516 buffer_json_member_add_uint64(wb, LOGS_MANAG_FUNC_PARAM_AFTER, fqs->after_ut / USEC_PER_SEC);

517 buffer_json_member_add_uint64(wb, LOGS_MANAG_FUNC_PARAM_BEFORE, fqs->before_ut / USEC_PER_SEC);

518 buffer_json_member_add_uint64(wb, LOGS_MANAG_FUNC_PARAM_IF_MODIFIED_SINCE, fqs->if_modified_since);

@@ -593,6 +589,7 @@ static void logsmanagement_function_facets(const char *transaction, char *functi

589 (fqs->data_only && fqs->anchor.start_ut) ? fqs->anchor.start_ut / USEC_PER_MS : after_s * MSEC_PER_SEC;

590 }

591

592 + query_params.cancelled = cancelled;

593 query_params.stop_monotonic_ut = now_monotonic_usec() + (timeout - 1) * USEC_PER_SEC;

594 query_params.results_buff = buffer_create(query_params.quota, NULL);

595

@@ -604,6 +601,7 @@ static void logsmanagement_function_facets(const char *transaction, char *functi

601

602 ret = execute_logs_manag_query(&query_params);

603

604 +

605 size_t res_off = 0;

606 logs_query_res_hdr_t *p_res_hdr;

607 while(query_params.results_buff->len - res_off > 0){

@@ -680,7 +678,8 @@ static void logsmanagement_function_facets(const char *transaction, char *functi

678 buffer_json_object_close(wb); // logs_management_meta

679

680 buffer_json_member_add_uint64(wb, "status", ret->http_code);

683 - buffer_json_member_add_boolean(wb, "partial", ret->http_code != HTTP_RESP_OK);

681 + buffer_json_member_add_boolean(wb, "partial", ret->http_code != HTTP_RESP_OK ||

682 + ret->err_code == LOGS_QRY_RES_ERR_CODE_TIMEOUT);

683 buffer_json_member_add_string(wb, "type", "table");

684

685

logsmanagement/query.c

+20 -2

@@ -102,6 +102,17 @@ const logs_qry_res_err_t *fetch_log_sources(BUFFER *wb){

102 return &logs_qry_res_err[LOGS_QRY_RES_ERR_CODE_OK];

103 }

104

105 +bool terminate_logs_manag_query(logs_query_params_t *const p_query_params){

106 + if(p_query_params->cancelled && __atomic_load_n(p_query_params->cancelled, __ATOMIC_RELAXED)) {

107 + return true;

108 + }

109 +

110 + if(now_monotonic_usec() > p_query_params->stop_monotonic_ut)

111 + return true;

112 +

113 + return false;

114 +}

115 +

116 const logs_qry_res_err_t *execute_logs_manag_query(logs_query_params_t *p_query_params) {

117 struct File_info *p_file_infos[LOGS_MANAG_MAX_COMPOUND_QUERY_SOURCES] = {NULL};

118

@@ -182,12 +193,12 @@ const logs_qry_res_err_t *execute_logs_manag_query(logs_query_params_t *p_query_

193 db_search(p_query_params, p_file_infos);

194

195 if( p_query_params->results_buff->len < p_query_params->quota &&

185 - now_monotonic_usec() <= p_query_params->stop_monotonic_ut)

196 + !terminate_logs_manag_query(p_query_params))

197 circ_buff_search(p_query_params, p_file_infos);

198

199 if(!p_query_params->order_by_asc &&

200 p_query_params->results_buff->len < p_query_params->quota &&

190 - now_monotonic_usec() <= p_query_params->stop_monotonic_ut)

201 + !terminate_logs_manag_query(p_query_params))

202 db_search(p_query_params, p_file_infos);

203

204 for(int pfi_off = 0; p_file_infos[pfi_off]; pfi_off++)

@@ -214,6 +225,13 @@ const logs_qry_res_err_t *execute_logs_manag_query(logs_query_params_t *p_query_

225 freez(p_query_params->keyword);

226 }

227

228 + if(terminate_logs_manag_query(p_query_params)){

229 + return (p_query_params->cancelled &&

230 + __atomic_load_n(p_query_params->cancelled, __ATOMIC_RELAXED)) ?

231 + &logs_qry_res_err[LOGS_QRY_RES_ERR_CODE_CANCELLED] /* cancelled */ :

232 + &logs_qry_res_err[LOGS_QRY_RES_ERR_CODE_TIMEOUT] /* timed out */ ;

233 + }

234 +

235 if(!p_query_params->results_buff->len)

236 return &logs_qry_res_err[LOGS_QRY_RES_ERR_CODE_NOT_FOUND_ERR];

237

logsmanagement/query.h

+15 -4

@@ -29,7 +29,9 @@ typedef struct {

29 LOGS_QRY_RES_ERR_CODE_NOT_FOUND_ERR,

30 LOGS_QRY_RES_ERR_CODE_NOT_INIT_ERR,

31 LOGS_QRY_RES_ERR_CODE_SERVER_ERR,

32 - LOGS_QRY_RES_ERR_CODE_UNMODIFIED } err_code;

32 + LOGS_QRY_RES_ERR_CODE_UNMODIFIED,

33 + LOGS_QRY_RES_ERR_CODE_CANCELLED,

34 + LOGS_QRY_RES_ERR_CODE_TIMEOUT } err_code;

35 char const *const err_str;

36 const int http_code;

37 } logs_qry_res_err_t;

@@ -40,7 +42,9 @@ static const logs_qry_res_err_t logs_qry_res_err[] = {

42 { LOGS_QRY_RES_ERR_CODE_NOT_FOUND_ERR, "no results found", HTTP_RESP_OK },

43 { LOGS_QRY_RES_ERR_CODE_NOT_INIT_ERR, "logs management engine not running", HTTP_RESP_SERVICE_UNAVAILABLE },

44 { LOGS_QRY_RES_ERR_CODE_SERVER_ERR, "server error", HTTP_RESP_INTERNAL_SERVER_ERROR },

43 - { LOGS_QRY_RES_ERR_CODE_UNMODIFIED, "not modified", HTTP_RESP_NOT_MODIFIED }

45 + { LOGS_QRY_RES_ERR_CODE_UNMODIFIED, "not modified", HTTP_RESP_NOT_MODIFIED },

46 + { LOGS_QRY_RES_ERR_CODE_CANCELLED, "cancelled", HTTP_RESP_CLIENT_CLOSED_REQUEST },

47 + { LOGS_QRY_RES_ERR_CODE_TIMEOUT, "query timed out", HTTP_RESP_OK }

48 };

50 const logs_qry_res_err_t *fetch_log_sources(BUFFER *wb);

@@ -108,6 +112,7 @@ typedef struct logs_query_params {

112 msec_t act_to_ts;

113 int order_by_asc;

114 unsigned long quota;

115 + bool *cancelled;

116 usec_t stop_monotonic_ut;

117 char *chartname[LOGS_MANAG_MAX_COMPOUND_QUERY_SOURCES];

118 char *filename[LOGS_MANAG_MAX_COMPOUND_QUERY_SOURCES];

@@ -129,10 +134,16 @@ typedef struct logs_query_res_hdr {

134 char chartname[20];

135 } logs_query_res_hdr_t;

136

137 +/**

138 + * @brief Check if query should be terminated.

139 + * @param p_query_params See documentation of logs_query_params_t struct.

140 + * @return true if query should be terminated of false otherwise.

141 +*/

142 +bool terminate_logs_manag_query(logs_query_params_t *p_query_params);

143 +

144 /**

145 * @brief Primary query API.

134 - * @param p_query_params See documentation of logs_query_params_t struct on how

135 - * to use argument.

146 + * @param p_query_params See documentation of logs_query_params_t struct.

147 * @return enum of LOGS_QRY_RES_ERR_CODE with result of query

148 * @todo Cornercase if filename not found in DB? Return specific message?

149 */