@cryptotaxi247 / netdata-1 / commits / 2a491f793

Add Cancel Pending Request Message (#14953)

Timotej S committed May 2, 2023 at 17:43 UTC 2a491f7932b21768149a787c8ef94c5b817471aa
10 files changed +185
CMakeLists.txt
+3
@@ -936,6 +936,8 @@ set(ACLK_FILES
936 aclk/schema-wrappers/schema_wrappers.h
937 aclk/schema-wrappers/schema_wrapper_utils.cc
938 aclk/schema-wrappers/schema_wrapper_utils.h
939 + aclk/schema-wrappers/agent_cmds.cc \
940 + aclk/schema-wrappers/agent_cmds.h \
941 aclk/helpers/mqtt_wss_pal.h
942 aclk/helpers/ringbuffer_pal.h
943 )
@@ -1259,6 +1261,7 @@ set(ACLK_PROTO_DEFS
1261 aclk/aclk-schemas/proto/nodeinstance/info/v1/info.proto
1262 aclk/aclk-schemas/proto/context/v1/context.proto
1263 aclk/aclk-schemas/proto/context/v1/stream.proto
1264 + aclk/aclk-schemas/proto/agent/v1/cmds.proto
1265 )
1266 PROTOBUF_ACLK_GENERATE_CPP(ACLK_PROTO_BUILT_SRCS ACLK_PROTO_BUILT_HDRS ${ACLK_PROTO_DEFS})
1267
Makefile.am
+9
@@ -726,6 +726,8 @@ ACLK_FILES = \
726 aclk/schema-wrappers/context_stream.h \
727 aclk/schema-wrappers/context.cc \
728 aclk/schema-wrappers/context.h \
729 + aclk/schema-wrappers/agent_cmds.cc \
730 + aclk/schema-wrappers/agent_cmds.h \
731 $(NULL)
732
733 noinst_LIBRARIES += libmqttwebsockets.a
@@ -768,6 +770,7 @@ ACLK_PROTO_DEFINITIONS = \
770 aclk/aclk-schemas/proto/nodeinstance/info/v1/info.proto \
771 aclk/aclk-schemas/proto/context/v1/context.proto \
772 aclk/aclk-schemas/proto/context/v1/stream.proto \
773 + aclk/aclk-schemas/proto/agent/v1/cmds.proto \
774 $(NULL)
775
776 dist_noinst_DATA += $(ACLK_PROTO_DEFINITIONS)
@@ -792,6 +795,8 @@ ACLK_PROTO_BUILT_FILES = aclk/aclk-schemas/proto/agent/v1/connection.pb.cc \
795 aclk/aclk-schemas/proto/context/v1/context.pb.h \
796 aclk/aclk-schemas/proto/context/v1/stream.pb.cc \
797 aclk/aclk-schemas/proto/context/v1/stream.pb.h \
798 + aclk/aclk-schemas/proto/agent/v1/cmds.pb.cc \
799 + aclk/aclk-schemas/proto/agent/v1/cmds.pb.h \
800 $(NULL)
801
802 BUILT_SOURCES += $(ACLK_PROTO_BUILT_FILES)
@@ -838,6 +843,10 @@ aclk/aclk-schemas/proto/context/v1/stream.pb.cc \
843 aclk/aclk-schemas/proto/context/v1/stream.pb.h: aclk/aclk-schemas/proto/context/v1/stream.proto
844 $(PROTOC) -I=aclk/aclk-schemas --cpp_out=$(builddir)/aclk/aclk-schemas $^
845
846 +aclk/aclk-schemas/proto/agent/v1/cmds.pb.cc \
847 +aclk/aclk-schemas/proto/agent/v1/cmds.pb.h: aclk/aclk-schemas/proto/agent/v1/cmds.proto
848 + $(PROTOC) -I=aclk/aclk-schemas --cpp_out=$(builddir)/aclk/aclk-schemas $^
849 +
850 endif #ENABLE_ACLK
851
852 ACLK_ALWAYS_BUILD_FILES = \
aclk/aclk_capas.c
+2
@@ -15,6 +15,7 @@ const struct capability *aclk_get_agent_capas()
15 { .name = "funcs", .version = 1, .enabled = 1 },
16 { .name = "http_api_v2", .version = 1, .enabled = 1 },
17 { .name = "health", .version = 1, .enabled = 0 },
18 + { .name = "req_cancel", .version = 1, .enabled = 1 },
19 { .name = NULL, .version = 0, .enabled = 0 }
20 };
21 agent_capabilities[2].version = ml_capable() ? 1 : 0;
@@ -40,6 +41,7 @@ struct capability *aclk_get_node_instance_capas(RRDHOST *host)
41 { .name = "funcs", .version = 0, .enabled = 0 },
42 { .name = "http_api_v2", .version = 2, .enabled = 1 },
43 { .name = "health", .version = 1, .enabled = host->health.health_enabled },
44 + { .name = "req_cancel", .version = 1, .enabled = 1 },
45 { .name = NULL, .version = 0, .enabled = 0 }
46 };
47
aclk/aclk_query.c
+81
@@ -13,6 +13,82 @@ pthread_mutex_t query_lock_wait = PTHREAD_MUTEX_INITIALIZER;
13 #define QUERY_THREAD_LOCK pthread_mutex_lock(&query_lock_wait)
14 #define QUERY_THREAD_UNLOCK pthread_mutex_unlock(&query_lock_wait)
15
16 +struct pending_req_list {
17 + const char *msg_id;
18 + uint32_t hash;
19 +
20 + int canceled;
21 +
22 + struct pending_req_list *next;
23 +};
24 +
25 +static struct pending_req_list *pending_req_list_head = NULL;
26 +static pthread_mutex_t pending_req_list_lock = PTHREAD_MUTEX_INITIALIZER;
27 +
28 +static struct pending_req_list *pending_req_list_add(const char *msg_id)
29 +{
30 + struct pending_req_list *new = callocz(1, sizeof(struct pending_req_list));
31 + new->msg_id = msg_id;
32 + new->hash = simple_hash(msg_id);
33 +
34 + pthread_mutex_lock(&pending_req_list_lock);
35 + new->next = pending_req_list_head;
36 + pending_req_list_head = new;
37 + pthread_mutex_unlock(&pending_req_list_lock);
38 + return new;
39 +}
40 +
41 +void pending_req_list_rm(const char *msg_id)
42 +{
43 + uint32_t hash = simple_hash(msg_id);
44 + struct pending_req_list *prev = NULL;
45 +
46 + pthread_mutex_lock(&pending_req_list_lock);
47 + struct pending_req_list *curr = pending_req_list_head;
48 +
49 + while (curr) {
50 + if (curr->hash == hash && strcmp(curr->msg_id, msg_id) == 0) {
51 + if (prev)
52 + prev->next = curr->next;
53 + else
54 + pending_req_list_head = curr->next;
55 +
56 + freez(curr);
57 + break;
58 + }
59 +
60 + prev = curr;
61 + curr = curr->next;
62 + }
63 + pthread_mutex_unlock(&pending_req_list_lock);
64 +}
65 +
66 +int mark_pending_req_cancelled(const char *msg_id)
67 +{
68 + uint32_t hash = simple_hash(msg_id);
69 +
70 + pthread_mutex_lock(&pending_req_list_lock);
71 + struct pending_req_list *curr = pending_req_list_head;
72 +
73 + while (curr) {
74 + if (curr->hash == hash && strcmp(curr->msg_id, msg_id) == 0) {
75 + curr->canceled = 1;
76 + pthread_mutex_unlock(&pending_req_list_lock);
77 + return 0;
78 + }
79 +
80 + curr = curr->next;
81 + }
82 + pthread_mutex_unlock(&pending_req_list_lock);
83 + return 1;
84 +}
85 +
86 +static bool aclk_web_client_interrupt_cb(struct web_client *w __maybe_unused, void *data)
87 +{
88 + struct pending_req_list *req = (struct pending_req_list *)data;
89 + return req->canceled;
90 +}
91 +
92 static int http_api_v2(struct aclk_query_thread *query_thr, aclk_query_t query) {
93 int retval = 0;
94 BUFFER *local_buffer = NULL;
@@ -30,6 +106,9 @@ static int http_api_v2(struct aclk_query_thread *query_thr, aclk_query_t query)
106 w->mode = WEB_CLIENT_MODE_GET;
107 w->timings.tv_in = query->created_tv;
108
109 + w->interrupt.callback = aclk_web_client_interrupt_cb;
110 + w->interrupt.callback_data = pending_req_list_add(query->msg_id);
111 +
112 usec_t t;
113 web_client_timeout_checkpoint_set(w, query->timeout);
114 if(web_client_timeout_checkpoint_and_check(w, &t)) {
@@ -168,6 +247,8 @@ cleanup:
247
248 web_client_release_to_cache(w);
249
250 + pending_req_list_rm(query->msg_id);
251 +
252 #ifdef NETDATA_WITH_ZLIB
253 buffer_free(z_buffer);
254 #endif
aclk/aclk_query.h
+2
@@ -33,4 +33,6 @@ void aclk_query_threads_cleanup(struct aclk_query_threads *query_threads);
33
34 const char *aclk_query_get_name(aclk_query_type_t qt, int unknown_ok);
35
36 +int mark_pending_req_cancelled(const char *msg_id);
37 +
38 #endif //NETDATA_AGENT_CLOUD_LINK_H
aclk/aclk_rx_msgs.c
+19
@@ -6,6 +6,7 @@
6 #include "aclk_query_queue.h"
7 #include "aclk.h"
8 #include "aclk_capas.h"
9 +#include "aclk_query.h"
10
11 #include "schema-wrappers/proto_2_json.h"
12
@@ -446,6 +447,23 @@ int stop_streaming_contexts(const char *msg, size_t msg_len)
447 return 0;
448 }
449
450 +int cancel_pending_req(const char *msg, size_t msg_len)
451 +{
452 + struct aclk_cancel_pending_req cmd;
453 + if(parse_cancel_pending_req(msg, msg_len, &cmd)) {
454 + error_report("Error parsing CancelPendingReq");
455 + return 1;
456 + }
457 +
458 + log_access("ACLK CancelPendingRequest REQ: %s, cloud trace-id: %s", cmd.request_id, cmd.trace_id);
459 +
460 + if (mark_pending_req_cancelled(cmd.request_id))
461 + error_report("CancelPending Request for %s failed. No such pending request.", cmd.request_id);
462 +
463 + free_cancel_pending_req(&cmd);
464 + return 0;
465 +}
466 +
467 typedef struct {
468 const char *name;
469 simple_hash_t name_hash;
@@ -466,6 +484,7 @@ new_cloud_rx_msg_t rx_msgs[] = {
484 { .name = "DisconnectReq", .name_hash = 0, .fnc = handle_disconnect_req },
485 { .name = "ContextsCheckpoint", .name_hash = 0, .fnc = contexts_checkpoint },
486 { .name = "StopStreamingContexts", .name_hash = 0, .fnc = stop_streaming_contexts },
487 + { .name = "CancelPendingRequest", .name_hash = 0, .fnc = cancel_pending_req },
488 { .name = NULL, .name_hash = 0, .fnc = NULL },
489 };
490
aclk/schema-wrappers/agent_cmds.cc new
+38
@@ -0,0 +1,38 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +#include "proto/agent/v1/cmds.pb.h"
4 +
5 +#include "agent_cmds.h"
6 +
7 +#include "schema_wrapper_utils.h"
8 +
9 +using namespace agent::v1;
10 +
11 +int parse_cancel_pending_req(const char *msg, size_t msg_len, struct aclk_cancel_pending_req *req)
12 +{
13 + CancelPendingRequest msg_parsed;
14 +
15 + if (!msg_parsed.ParseFromArray(msg, msg_len)) {
16 + error_report("Failed to parse CancelPendingRequest message");
17 + return 1;
18 + }
19 +
20 + if (msg_parsed.request_id().c_str() == NULL) {
21 + error_report("CancelPendingRequest message missing request_id");
22 + return 1;
23 + }
24 + req->request_id = strdupz(msg_parsed.request_id().c_str());
25 +
26 + if (msg_parsed.trace_id().c_str())
27 + req->trace_id = strdupz(msg_parsed.trace_id().c_str());
28 +
29 + set_timeval_from_google_timestamp(msg_parsed.timestamp(), &req->timestamp);
30 +
31 + return 0;
32 +}
33 +
34 +void free_cancel_pending_req(struct aclk_cancel_pending_req *req)
35 +{
36 + freez(req->request_id);
37 + freez(req->trace_id);
38 +}
aclk/schema-wrappers/agent_cmds.h new
+27
@@ -0,0 +1,27 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +#ifndef ACLK_SCHEMA_WRAPPERS_AGENT_CMDS_H
4 +#define ACLK_SCHEMA_WRAPPERS_AGENT_CMDS_H
5 +
6 +#include "libnetdata/libnetdata.h"
7 +
8 +#ifdef __cplusplus
9 +extern "C" {
10 +#endif
11 +
12 +struct aclk_cancel_pending_req {
13 + char *request_id;
14 +
15 + struct timeval timestamp;
16 +
17 + char *trace_id;
18 +};
19 +
20 +int parse_cancel_pending_req(const char *msg, size_t msg_len, struct aclk_cancel_pending_req *req);
21 +void free_cancel_pending_req(struct aclk_cancel_pending_req *req);
22 +
23 +#ifdef __cplusplus
24 +}
25 +#endif
26 +
27 +#endif /* ACLK_SCHEMA_WRAPPERS_AGENT_CMDS_H */
aclk/schema-wrappers/proto_2_json.cc
+3
@@ -11,6 +11,7 @@
11 #include "proto/nodeinstance/info/v1/info.pb.h"
12 #include "proto/context/v1/stream.pb.h"
13 #include "proto/context/v1/context.pb.h"
14 +#include "proto/agent/v1/cmds.pb.h"
15
16 #include "libnetdata/libnetdata.h"
17
@@ -63,6 +64,8 @@ static google::protobuf::Message *msg_name_to_protomsg(const char *msgname)
64 return new context::v1::ContextsCheckpoint;
65 if (!strcmp(msgname, "StopStreamingContexts"))
66 return new context::v1::StopStreamingContexts;
67 + if (!strcmp(msgname, "CancelPendingRequest"))
68 + return new agent::v1::CancelPendingRequest;
69
70 return NULL;
71 }
aclk/schema-wrappers/schema_wrappers.h
+1
@@ -14,5 +14,6 @@
14 #include "capability.h"
15 #include "context_stream.h"
16 #include "context.h"
17 +#include "agent_cmds.h"
18
19 #endif /* SCHEMA_WRAPPERS_H */