@cryptotaxi247 / netdata-1
Code Commits Issues Pulls Sessions CI/CD
master
master
cc 258 lines 9.81 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include <google/pubsub/v1/pubsub.grpc.pb.h>
4 #include <grpcpp/grpcpp.h>
5 #include <stdexcept>
6 #include "pubsub_publish.h"
7
8 #define EVENT_CHECK_TIMEOUT 50
9
10 struct response {
11 grpc::ClientContext *context;
12 google::pubsub::v1::PublishResponse *publish_response;
13 size_t tag;
14 grpc::Status *status;
15
16 size_t published_metrics;
17 size_t published_bytes;
18 };
19
20 static inline void copy_error_message(char *error_message_dst, const char *error_message_src)
21 {
22 std::strncpy(error_message_dst, error_message_src, ERROR_LINE_MAX);
23 error_message_dst[ERROR_LINE_MAX] = '\0';
24 }
25
26 /**
27 * Initialize a Pub/Sub client and a data structure for responses.
28 *
29 * @param pubsub_specific_data_p a pointer to a structure with instance-wide data.
30 * @param error_message report error message to a caller.
31 * @param destination a Pub/Sub service endpoint.
32 * @param credentials_file a full path for a file with google application credentials.
33 * @param project_id a project ID.
34 * @param topic_id a topic ID.
35 * @return Returns 0 on success, 1 on failure.
36 */
37 int pubsub_init(
38 void *pubsub_specific_data_p, char *error_message, const char *destination, const char *credentials_file,
39 const char *project_id, const char *topic_id)
40 {
41 struct pubsub_specific_data *connector_specific_data = (struct pubsub_specific_data *)pubsub_specific_data_p;
42
43 try {
44 setenv("GOOGLE_APPLICATION_CREDENTIALS", credentials_file, 0);
45
46 std::shared_ptr<grpc::ChannelCredentials> credentials = grpc::GoogleDefaultCredentials();
47 if (credentials == nullptr) {
48 copy_error_message(error_message, "Can't load credentials");
49 return 1;
50 }
51
52 std::shared_ptr<grpc::Channel> channel = grpc::CreateChannel(destination, credentials);
53
54 google::pubsub::v1::Publisher::Stub *stub = new google::pubsub::v1::Publisher::Stub(channel);
55 if (!stub) {
56 copy_error_message(error_message, "Can't create a publisher stub");
57 return 1;
58 }
59
60 connector_specific_data->stub = stub;
61
62 google::pubsub::v1::PublishRequest *request = new google::pubsub::v1::PublishRequest;
63 connector_specific_data->request = request;
64 ((google::pubsub::v1::PublishRequest *)(connector_specific_data->request))
65 ->set_topic(std::string("projects/") + project_id + "/topics/" + topic_id);
66
67 grpc::CompletionQueue *cq = new grpc::CompletionQueue;
68 connector_specific_data->completion_queue = cq;
69
70 connector_specific_data->responses = new std::list<struct response>;
71
72 return 0;
73 } catch (std::exception const &ex) {
74 std::string em(std::string("Standard exception raised: ") + ex.what());
75 copy_error_message(error_message, em.c_str());
76 return 1;
77 }
78
79 return 0;
80 }
81
82 /**
83 * Clean the PubSub connector instance specific data
84 */
85 void pubsub_cleanup(void *pubsub_specific_data_p)
86 {
87 struct pubsub_specific_data *connector_specific_data = (struct pubsub_specific_data *)pubsub_specific_data_p;
88
89 std::list<struct response> *responses = (std::list<struct response> *)connector_specific_data->responses;
90 std::list<struct response>::iterator response;
91 for (response = responses->begin(); response != responses->end(); ++response) {
92 // TODO: If we do this, there are a huge amount of possibly lost records. We need to find a right way of
93 // cleaning up contexts
94 // delete response->context;
95 delete response->publish_response;
96 delete response->status;
97 }
98 delete responses;
99
100 ((grpc::CompletionQueue *)connector_specific_data->completion_queue)->Shutdown();
101 delete (grpc::CompletionQueue *)connector_specific_data->completion_queue;
102 delete (google::pubsub::v1::PublishRequest *)connector_specific_data->request;
103 delete (google::pubsub::v1::Publisher::Stub *)connector_specific_data->stub;
104
105 // TODO: Find how to shutdown grpc gracefully. grpc_shutdown() doesn't seem to work.
106 // grpc_shutdown();
107
108 return;
109 }
110
111 /**
112 * Add data to a Pub/Sub request message.
113 *
114 * @param pubsub_specific_data_p a pointer to a structure with instance-wide data.
115 * @param data a text buffer with metrics.
116 * @return Returns 0 on success, 1 on failure.
117 */
118 int pubsub_add_message(void *pubsub_specific_data_p, char *data)
119 {
120 struct pubsub_specific_data *connector_specific_data = (struct pubsub_specific_data *)pubsub_specific_data_p;
121
122 try {
123 google::pubsub::v1::PubsubMessage *message =
124 ((google::pubsub::v1::PublishRequest *)(connector_specific_data->request))->add_messages();
125 if (!message)
126 return 1;
127
128 message->set_data(data);
129 } catch (std::exception const &ex) {
130 return 1;
131 }
132
133 return 0;
134 }
135
136 /**
137 * Send data to the Pub/Sub service
138 *
139 * @param pubsub_specific_data_p a pointer to a structure with client and request outcome information.
140 * @param error_message report error message to a caller.
141 * @param buffered_metrics the number of metrics we are going to send.
142 * @param buffered_bytes the number of bytes we are going to send.
143 * @return Returns 0 on success, 1 on failure.
144 */
145 int pubsub_publish(void *pubsub_specific_data_p, char *error_message, size_t buffered_metrics, size_t buffered_bytes)
146 {
147 struct pubsub_specific_data *connector_specific_data = (struct pubsub_specific_data *)pubsub_specific_data_p;
148
149 try {
150 grpc::ClientContext *context = new grpc::ClientContext;
151
152 std::unique_ptr<grpc::ClientAsyncResponseReader<google::pubsub::v1::PublishResponse> > rpc(
153 ((google::pubsub::v1::Publisher::Stub *)(connector_specific_data->stub))
154 ->AsyncPublish(
155 context, (*(google::pubsub::v1::PublishRequest *)(connector_specific_data->request)),
156 ((grpc::CompletionQueue *)(connector_specific_data->completion_queue))));
157
158 struct response response;
159 response.context = context;
160 response.publish_response = new google::pubsub::v1::PublishResponse;
161 response.tag = connector_specific_data->last_tag++;
162 response.status = new grpc::Status;
163 response.published_metrics = buffered_metrics;
164 response.published_bytes = buffered_bytes;
165
166 rpc->Finish(response.publish_response, response.status, (void *)response.tag);
167
168 ((google::pubsub::v1::PublishRequest *)(connector_specific_data->request))->clear_messages();
169
170 ((std::list<struct response> *)(connector_specific_data->responses))->push_back(response);
171 } catch (std::exception const &ex) {
172 std::string em(std::string("Standard exception raised: ") + ex.what());
173 copy_error_message(error_message, em.c_str());
174 return 1;
175 }
176
177 return 0;
178 }
179
180 /**
181 * Get results from service responses
182 *
183 * @param pubsub_specific_data_p a pointer to a structure with instance-wide data.
184 * @param error_message report error message to a caller.
185 * @param sent_metrics report to a caller how many metrics was successfully sent.
186 * @param sent_bytes report to a caller how many bytes was successfully sent.
187 * @param lost_metrics report to a caller how many metrics was lost during transmission.
188 * @param lost_bytes report to a caller how many bytes was lost during transmission.
189 * @return Returns 0 if all data was sent successfully, 1 when data was lost on transmission.
190 */
191 int pubsub_get_result(
192 void *pubsub_specific_data_p, char *error_message,
193 size_t *sent_metrics, size_t *sent_bytes, size_t *lost_metrics, size_t *lost_bytes)
194 {
195 struct pubsub_specific_data *connector_specific_data = (struct pubsub_specific_data *)pubsub_specific_data_p;
196 std::list<struct response> *responses = (std::list<struct response> *)connector_specific_data->responses;
197 grpc::CompletionQueue::NextStatus next_status;
198
199 *sent_metrics = 0;
200 *sent_bytes = 0;
201 *lost_metrics = 0;
202 *lost_bytes = 0;
203
204 try {
205 do {
206 std::list<struct response>::iterator response;
207 void *got_tag;
208 bool ok = false;
209
210 auto deadline = std::chrono::system_clock::now() + std::chrono::milliseconds(50);
211 next_status = (*(grpc::CompletionQueue *)(connector_specific_data->completion_queue))
212 .AsyncNext(&got_tag, &ok, deadline);
213
214 if (next_status == grpc::CompletionQueue::GOT_EVENT) {
215 for (response = responses->begin(); response != responses->end(); ++response) {
216 if ((void *)response->tag == got_tag)
217 break;
218 }
219
220 if (response == responses->end()) {
221 copy_error_message(error_message, "Cannot get Pub/Sub response");
222 return 1;
223 }
224
225 if (ok && response->publish_response->message_ids_size()) {
226 *sent_metrics += response->published_metrics;
227 *sent_bytes += response->published_bytes;
228 } else {
229 *lost_metrics += response->published_metrics;
230 *lost_bytes += response->published_bytes;
231 response->status->error_message().copy(error_message, ERROR_LINE_MAX);
232 error_message[ERROR_LINE_MAX] = '\0';
233 }
234
235 delete response->context;
236 delete response->publish_response;
237 delete response->status;
238 responses->erase(response);
239 }
240
241 if (next_status == grpc::CompletionQueue::SHUTDOWN) {
242 copy_error_message(error_message, "Completion queue shutdown");
243 return 1;
244 }
245
246 } while (next_status == grpc::CompletionQueue::GOT_EVENT);
247 } catch (std::exception const &ex) {
248 std::string em(std::string("Standard exception raised: ") + ex.what());
249 copy_error_message(error_message, em.c_str());
250 return 1;
251 }
252
253 if (*lost_metrics) {
254 return 1;
255 }
256
257 return 0;
258 }