@cryptotaxi247 / netdata-1 / commits / 3aa7be1f7

Add HTTP access log messages for ACLK-NG (#11318)

* aclk: link libcap after libwebsockets for legacy build libwebsockets requires libcap. without this, we get errors about undefined libcap functions that libwebsockets uses. This is needed for building aclk-legacy. * aclk: create log for HTTP requests similar to legacy version.

Uman Shahzad committed Jul 8, 2021 at 22:05 UTC 3aa7be1f7859d715171bc81401c73abfe61a6fdf

4 files changed +52 -27

Makefile.am

+1 -1

@@ -826,8 +826,8 @@ endif

826 if ACLK_LEGACY

827 netdata_LDADD += \

828 $(abs_top_srcdir)/externaldeps/mosquitto/libmosquitto.a \

829 - $(OPTIONAL_LIBCAP_LIBS) \

829 $(OPTIONAL_LWS_LIBS) \

830 + $(OPTIONAL_LIBCAP_LIBS) \

831 $(NULL)

832 endif #ACLK_LEGACY

833

aclk/aclk_query.c

+47 -25

@@ -17,20 +17,20 @@ pthread_mutex_t query_lock_wait = PTHREAD_MUTEX_INITIALIZER;

17 typedef struct aclk_query_handler {

18 aclk_query_type_t type;

19 char *name; // for logging purposes

20 - int(*fnc)(mqtt_wss_client client, aclk_query_t query);

20 + int(*fnc)(struct aclk_query_thread *query_thr, aclk_query_t query);

21 } aclk_query_handler;

23 -static int info_metadata(mqtt_wss_client client, aclk_query_t query)

23 +static int info_metadata(struct aclk_query_thread *query_thr, aclk_query_t query)

24 {

25 - aclk_send_info_metadata(client,

25 + aclk_send_info_metadata(query_thr->client,

26 !query->data.metadata_info.initial_on_connect,

27 query->data.metadata_info.host);

28 return 0;

29 }

31 -static int alarms_metadata(mqtt_wss_client client, aclk_query_t query)

31 +static int alarms_metadata(struct aclk_query_thread *query_thr, aclk_query_t query)

32 {

33 - aclk_send_alarm_metadata(client,

33 + aclk_send_alarm_metadata(query_thr->client,

34 !query->data.metadata_info.initial_on_connect);

35 return 0;

36 }

@@ -76,7 +76,7 @@ static RRDHOST *node_id_2_rrdhost(const char *node_id)

76 // TODO this function should be quarantied and written nicely

77 // lots of skeletons from initial ACLK Legacy impl.

78 // quick and dirty from the start

79 -static int http_api_v2(mqtt_wss_client client, aclk_query_t query)

79 +static int http_api_v2(struct aclk_query_thread *query_thr, aclk_query_t query)

80 {

81 int retval = 0;

82 usec_t t;

@@ -126,7 +126,11 @@ static int http_api_v2(mqtt_wss_client client, aclk_query_t query)

126 mysep = strrchr(query->data.http_api_v2.query, '/');

127

128 // execute the query

129 + w->tv_in = query->created_tv;

130 + now_realtime_timeval(&w->tv_ready);

131 t = aclk_web_api_v1_request(query_host, w, mysep ? mysep + 1 : "noop");

132 + size_t size = (w->mode == WEB_CLIENT_MODE_FILECOPY) ? w->response.rlen : w->response.data->len;

133 + size_t sent = size;

134

135 #ifdef NETDATA_WITH_ZLIB

136 // check if gzip encoding can and should be used

@@ -175,7 +179,6 @@ static int http_api_v2(mqtt_wss_client client, aclk_query_t query)

179 }

180 #endif

181

178 - now_realtime_timeval(&w->tv_ready);

182 w->response.data->date = w->tv_ready.tv_sec;

183 web_client_build_http_header(w);

184 local_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);

@@ -189,6 +192,7 @@ static int http_api_v2(mqtt_wss_client client, aclk_query_t query)

192 buffer_need_bytes(local_buffer, w->response.data->len);

193 memcpy(&local_buffer->buffer[local_buffer->len], w->response.data->buffer, w->response.data->len);

194 local_buffer->len += w->response.data->len;

195 + sent = sent - size + w->response.data->len;

196 } else {

197 #endif

198 buffer_strcat(local_buffer, w->response.data->buffer);

@@ -197,7 +201,26 @@ static int http_api_v2(mqtt_wss_client client, aclk_query_t query)

201 #endif

202 }

203

200 - aclk_http_msg_v2(client, query->callback_topic, query->msg_id, t, query->created, w->response.code, local_buffer->buffer, local_buffer->len);

204 + // send msg.

205 + aclk_http_msg_v2(query_thr->client, query->callback_topic, query->msg_id, t, query->created, w->response.code, local_buffer->buffer, local_buffer->len);

206 +

207 + // log.

208 + struct timeval tv;

209 + now_realtime_timeval(&tv);

210 + log_access("%llu: %d '[ACLK]:%d' '%s' (sent/all = %zu/%zu bytes %0.0f%%, prep/sent/total = %0.2f/%0.2f/%0.2f ms) %d '%s'",

211 + w->id

212 + , gettid()

213 + , query_thr->idx

214 + , "DATA"

215 + , sent

216 + , size

217 + , size > sent ? -(((size - sent) / (double)size) * 100.0) : ((size > 0) ? (((sent - size ) / (double)size) * 100.0) : 0.0)

218 + , dt_usec(&w->tv_ready, &w->tv_in) / 1000.0

219 + , dt_usec(&tv, &w->tv_ready) / 1000.0

220 + , dt_usec(&tv, &w->tv_in) / 1000.0

221 + , w->response.code

222 + , strip_control_characters(query->data.http_api_v2.query)

223 + );

224

225 cleanup:

226 #ifdef NETDATA_WITH_ZLIB

@@ -213,33 +236,33 @@ cleanup:

236 return retval;

237 }

238

216 -static int chart_query(mqtt_wss_client client, aclk_query_t query)

239 +static int chart_query(struct aclk_query_thread *query_thr, aclk_query_t query)

240 {

218 - aclk_chart_msg(client, query->data.chart_add_del.host, query->data.chart_add_del.chart_name);

241 + aclk_chart_msg(query_thr->client, query->data.chart_add_del.host, query->data.chart_add_del.chart_name);

242 return 0;

243 }

244

222 -static int alarm_state_update_query(mqtt_wss_client client, aclk_query_t query)

245 +static int alarm_state_update_query(struct aclk_query_thread *query_thr, aclk_query_t query)

246 {

224 - aclk_alarm_state_msg(client, query->data.alarm_update);

247 + aclk_alarm_state_msg(query_thr->client, query->data.alarm_update);

248 // aclk_alarm_state_msg frees the json object including the header it generates

249 query->data.alarm_update = NULL;

250 return 0;

251 }

252

230 -static int register_node(mqtt_wss_client client, aclk_query_t query) {

253 +static int register_node(struct aclk_query_thread *query_thr, aclk_query_t query) {

254 // TODO create a pending registrations list

255 // with some timeouts to detect registration requests that

256 // go unanswered from the cloud

234 - aclk_generate_node_registration(client, &query->data.node_creation);

257 + aclk_generate_node_registration(query_thr->client, &query->data.node_creation);

258 return 0;

259 }

260

238 -static int node_state_update(mqtt_wss_client client, aclk_query_t query) {

261 +static int node_state_update(struct aclk_query_thread *query_thr, aclk_query_t query) {

262 // TODO create a pending registrations list

263 // with some timeouts to detect registration requests that

264 // go unanswered from the cloud

242 - aclk_generate_node_state_update(client, &query->data.node_update);

265 + aclk_generate_node_state_update(query_thr->client, &query->data.node_update);

266 return 0;

267 }

268

@@ -256,17 +279,17 @@ aclk_query_handler aclk_query_handlers[] = {

279 };

280

281

259 -static void aclk_query_process_msg(struct aclk_query_thread *info, aclk_query_t query)

282 +static void aclk_query_process_msg(struct aclk_query_thread *query_thr, aclk_query_t query)

283 {

284 for (int i = 0; aclk_query_handlers[i].type != UNKNOWN; i++) {

285 if (aclk_query_handlers[i].type == query->type) {

286 debug(D_ACLK, "Processing Queued Message of type: \"%s\"", aclk_query_handlers[i].name);

264 - aclk_query_handlers[i].fnc(info->client, query);

287 + aclk_query_handlers[i].fnc(query_thr, query);

288 aclk_query_free(query);

289 if (aclk_stats_enabled) {

290 ACLK_STATS_LOCK;

291 aclk_metrics_per_sample.queries_dispatched++;

269 - aclk_queries_per_thread[info->idx]++;

292 + aclk_queries_per_thread[query_thr->idx]++;

293 ACLK_STATS_UNLOCK;

294 }

295 return;

@@ -277,11 +300,11 @@ static void aclk_query_process_msg(struct aclk_query_thread *info, aclk_query_t

300

301 /* Processes messages from queue. Compete for work with other threads

302 */

280 -int aclk_query_process_msgs(struct aclk_query_thread *info)

303 +int aclk_query_process_msgs(struct aclk_query_thread *query_thr)

304 {

305 aclk_query_t query;

306 while ((query = aclk_queue_pop()))

284 - aclk_query_process_msg(info, query);

307 + aclk_query_process_msg(query_thr, query);

308

309 return 0;

310 }

@@ -291,15 +314,14 @@ int aclk_query_process_msgs(struct aclk_query_thread *info)

314 */

315 void *aclk_query_main_thread(void *ptr)

316 {

294 - struct aclk_query_thread *info = ptr;

317 + struct aclk_query_thread *query_thr = ptr;

318 +

319 while (!netdata_exit) {

296 - aclk_query_process_msgs(info);

320 + aclk_query_process_msgs(query_thr);

321

322 QUERY_THREAD_LOCK;

299 -

323 if (unlikely(pthread_cond_wait(&query_cond_wait, &query_lock_wait)))

324 sleep_usec(USEC_PER_SEC * 1);

302 -

325 QUERY_THREAD_UNLOCK;

326 }

327 return NULL;

aclk/aclk_query_queue.c

+3 -1

@@ -20,7 +20,9 @@ static struct aclk_query_queue {

21 static inline int _aclk_queue_query(aclk_query_t query)

22 {

23 + now_realtime_timeval(&query->created_tv);

24 query->created = now_realtime_usec();

25 +

26 ACLK_QUEUE_LOCK;

27 if (aclk_query_queue.block_push) {

28 ACLK_QUEUE_UNLOCK;

@@ -110,7 +112,7 @@ void aclk_query_free(aclk_query_t query)

112

113 if (query->type == CHART_NEW)

114 freez(query->data.chart_add_del.chart_name);

113 -

115 +

116 if (query->type == ALARM_STATE_UPDATE && query->data.alarm_update)

117 json_object_put(query->data.alarm_update);

118

aclk/aclk_query_queue.h

@@ -47,6 +47,7 @@ struct aclk_query {

47 char *callback_topic;

48 char *msg_id;

50 + struct timeval created_tv;

51 usec_t created;

53 aclk_query_t next;