write_redis_plugin: Add document for MaxSetDuration
[collectd.git] / src / grpc.cc
1 /**
2  * collectd - src/grpc.cc
3  * Copyright (C) 2015-2016 Sebastian Harl
4  * Copyright (C) 2016      Florian octo Forster
5  *
6  * Permission is hereby granted, free of charge, to any person obtaining a
7  * copy of this software and associated documentation files (the "Software"),
8  * to deal in the Software without restriction, including without limitation
9  * the rights to use, copy, modify, merge, publish, distribute, sublicense,
10  * and/or sell copies of the Software, and to permit persons to whom the
11  * Software is furnished to do so, subject to the following conditions:
12  *
13  * The above copyright notice and this permission notice shall be included in
14  * all copies or substantial portions of the Software.
15  *
16  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
17  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
18  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
19  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
20  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
21  * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
22  * DEALINGS IN THE SOFTWARE.
23  *
24  * Authors:
25  *   Sebastian Harl <sh at tokkee.org>
26  *   Florian octo Forster <octo at collectd.org>
27  **/
28
29 #include <google/protobuf/util/time_util.h>
30 #include <grpc++/grpc++.h>
31
32 #include <fstream>
33 #include <iostream>
34 #include <queue>
35 #include <vector>
36
37 #include "collectd.grpc.pb.h"
38
39 extern "C" {
40 #include <fnmatch.h>
41 #include <stdbool.h>
42
43 #include "collectd.h"
44 #include "common.h"
45 #include "plugin.h"
46
47 #include "daemon/utils_cache.h"
48 }
49
50 using collectd::Collectd;
51
52 using collectd::PutValuesRequest;
53 using collectd::PutValuesResponse;
54 using collectd::QueryValuesRequest;
55 using collectd::QueryValuesResponse;
56
57 using google::protobuf::util::TimeUtil;
58
59 typedef google::protobuf::Map<grpc::string, collectd::types::MetadataValue> grpcMetadata;
60
61 /*
62  * private types
63  */
64
65 struct Listener {
66   grpc::string addr;
67   grpc::string port;
68
69   grpc::SslServerCredentialsOptions *ssl;
70 };
71 static std::vector<Listener> listeners;
72 static grpc::string default_addr("0.0.0.0:50051");
73
74 /*
75  * helper functions
76  */
77
78 static bool ident_matches(const value_list_t *vl, const value_list_t *matcher) {
79   if (fnmatch(matcher->host, vl->host, 0))
80     return false;
81
82   if (fnmatch(matcher->plugin, vl->plugin, 0))
83     return false;
84   if (fnmatch(matcher->plugin_instance, vl->plugin_instance, 0))
85     return false;
86
87   if (fnmatch(matcher->type, vl->type, 0))
88     return false;
89   if (fnmatch(matcher->type_instance, vl->type_instance, 0))
90     return false;
91
92   return true;
93 } /* ident_matches */
94
95 static grpc::string read_file(const char *filename) {
96   std::ifstream f;
97   grpc::string s, content;
98
99   f.open(filename);
100   if (!f.is_open()) {
101     ERROR("grpc: Failed to open '%s'", filename);
102     return "";
103   }
104
105   while (std::getline(f, s)) {
106     content += s;
107     content.push_back('\n');
108   }
109   f.close();
110   return content;
111 } /* read_file */
112
113 /*
114  * proto conversion
115  */
116
117 static void marshal_ident(const value_list_t *vl,
118                           collectd::types::Identifier *msg) {
119   msg->set_host(vl->host);
120   msg->set_plugin(vl->plugin);
121   if (vl->plugin_instance[0] != '\0')
122     msg->set_plugin_instance(vl->plugin_instance);
123   msg->set_type(vl->type);
124   if (vl->type_instance[0] != '\0')
125     msg->set_type_instance(vl->type_instance);
126 } /* marshal_ident */
127
128 static grpc::Status unmarshal_ident(const collectd::types::Identifier &msg,
129                                     value_list_t *vl, bool require_fields) {
130   std::string s;
131
132   s = msg.host();
133   if (!s.length() && require_fields)
134     return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
135                         grpc::string("missing host name"));
136   sstrncpy(vl->host, s.c_str(), sizeof(vl->host));
137
138   s = msg.plugin();
139   if (!s.length() && require_fields)
140     return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
141                         grpc::string("missing plugin name"));
142   sstrncpy(vl->plugin, s.c_str(), sizeof(vl->plugin));
143
144   s = msg.type();
145   if (!s.length() && require_fields)
146     return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
147                         grpc::string("missing type name"));
148   sstrncpy(vl->type, s.c_str(), sizeof(vl->type));
149
150   s = msg.plugin_instance();
151   sstrncpy(vl->plugin_instance, s.c_str(), sizeof(vl->plugin_instance));
152
153   s = msg.type_instance();
154   sstrncpy(vl->type_instance, s.c_str(), sizeof(vl->type_instance));
155
156   return grpc::Status::OK;
157 } /* unmarshal_ident() */
158
159 static grpc::Status marshal_meta_data(meta_data_t *meta,
160                                       grpcMetadata *mutable_meta_data) {
161   char **meta_data_keys = nullptr;
162   int meta_data_keys_len = meta_data_toc(meta, &meta_data_keys);
163   if (meta_data_keys_len < 0) {
164     return grpc::Status(grpc::StatusCode::INTERNAL,
165                         grpc::string("error getting metadata keys"));
166   }
167
168   for (int i = 0; i < meta_data_keys_len; i++) {
169     char *key = meta_data_keys[i];
170     int md_type = meta_data_type(meta, key);
171
172     collectd::types::MetadataValue md_value;
173     md_value.Clear();
174
175     switch (md_type) {
176     case MD_TYPE_STRING:
177       char *md_string;
178       if (meta_data_get_string(meta, key, &md_string) != 0 || md_string == nullptr) {
179         strarray_free(meta_data_keys, meta_data_keys_len);
180         return grpc::Status(grpc::StatusCode::INTERNAL,
181                           grpc::string("missing metadata"));
182       }
183       md_value.set_string_value(md_string);
184       free(md_string);
185       break;
186     case MD_TYPE_SIGNED_INT:
187       int64_t int64_value;
188       if (meta_data_get_signed_int(meta, key, &int64_value) != 0) {
189         strarray_free(meta_data_keys, meta_data_keys_len);
190         return grpc::Status(grpc::StatusCode::INTERNAL,
191                           grpc::string("missing metadata"));
192       }
193       md_value.set_int64_value(int64_value);
194       break;
195     case MD_TYPE_UNSIGNED_INT:
196       uint64_t uint64_value;
197       if (meta_data_get_unsigned_int(meta, key, &uint64_value) != 0) {
198         strarray_free(meta_data_keys, meta_data_keys_len);
199         return grpc::Status(grpc::StatusCode::INTERNAL,
200                           grpc::string("missing metadata"));
201       }
202       md_value.set_uint64_value(uint64_value);
203       break;
204     case MD_TYPE_DOUBLE:
205       double double_value;
206       if (meta_data_get_double(meta, key, &double_value) != 0) {
207         strarray_free(meta_data_keys, meta_data_keys_len);
208         return grpc::Status(grpc::StatusCode::INTERNAL,
209                           grpc::string("missing metadata"));
210       }
211       md_value.set_double_value(double_value);
212       break;
213     case MD_TYPE_BOOLEAN:
214       bool bool_value;
215       if (meta_data_get_boolean(meta, key, &bool_value) != 0) {
216         strarray_free(meta_data_keys, meta_data_keys_len);
217         return grpc::Status(grpc::StatusCode::INTERNAL,
218                           grpc::string("missing metadata"));
219       }
220       md_value.set_bool_value(bool_value);
221       break;
222     default:
223       strarray_free(meta_data_keys, meta_data_keys_len);
224       ERROR("grpc: invalid metadata type (%d)", md_type);
225       return grpc::Status(grpc::StatusCode::INTERNAL,
226                           grpc::string("unknown metadata type"));
227     }
228
229     (*mutable_meta_data)[grpc::string(key)] = md_value;
230
231     strarray_free(meta_data_keys, meta_data_keys_len);
232   }
233
234   return grpc::Status::OK;
235 }
236
237 static grpc::Status unmarshal_meta_data(const grpcMetadata &rpc_metadata,
238                                         meta_data_t **md_out) {
239   *md_out = meta_data_create();
240   if (*md_out == nullptr) {
241     return grpc::Status(grpc::StatusCode::RESOURCE_EXHAUSTED,
242                         grpc::string("failed to metadata list"));
243   }
244   for (auto kv: rpc_metadata) {
245     auto k = kv.first.c_str();
246     auto v = kv.second;
247
248     // The meta_data collection individually allocates copies of the keys and
249     // string values for each entry, so it's safe for us to pass a reference
250     // to our short-lived strings.
251
252     switch (v.value_case()) {
253     case collectd::types::MetadataValue::ValueCase::kStringValue:
254       meta_data_add_string(*md_out, k, v.string_value().c_str());
255       break;
256     case collectd::types::MetadataValue::ValueCase::kInt64Value:
257       meta_data_add_signed_int(*md_out, k, v.int64_value());
258       break;
259     case collectd::types::MetadataValue::ValueCase::kUint64Value:
260       meta_data_add_unsigned_int(*md_out, k, v.uint64_value());
261       break;
262     case collectd::types::MetadataValue::ValueCase::kDoubleValue:
263       meta_data_add_double(*md_out, k, v.double_value());
264       break;
265     case collectd::types::MetadataValue::ValueCase::kBoolValue:
266       meta_data_add_boolean(*md_out, k, v.bool_value());
267       break;
268     default:
269       meta_data_destroy(*md_out);
270       return  grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
271                            grpc::string("Metadata of unknown type"));
272     }
273   }
274   return grpc::Status::OK;
275 }
276
277 static grpc::Status marshal_value_list(const value_list_t *vl,
278                                        collectd::types::ValueList *msg) {
279   auto id = msg->mutable_identifier();
280   marshal_ident(vl, id);
281
282   auto ds = plugin_get_ds(vl->type);
283   if ((ds == NULL) || (ds->ds_num != vl->values_len)) {
284     return grpc::Status(grpc::StatusCode::INTERNAL,
285                         grpc::string("failed to retrieve data-set for values"));
286   }
287
288   auto t = TimeUtil::NanosecondsToTimestamp(CDTIME_T_TO_NS(vl->time));
289   auto d = TimeUtil::NanosecondsToDuration(CDTIME_T_TO_NS(vl->interval));
290   msg->set_allocated_time(new google::protobuf::Timestamp(t));
291   msg->set_allocated_interval(new google::protobuf::Duration(d));
292
293   msg->clear_meta_data();
294   if (vl->meta != nullptr) {
295     grpc::Status status = marshal_meta_data(vl->meta, msg->mutable_meta_data());
296     if (!status.ok()) {
297       return status;
298     }
299   }
300
301   for (size_t i = 0; i < vl->values_len; ++i) {
302     auto v = msg->add_values();
303     int value_type = ds->ds[i].type;
304     switch (value_type) {
305     case DS_TYPE_COUNTER:
306       v->set_counter(vl->values[i].counter);
307       break;
308     case DS_TYPE_GAUGE:
309       v->set_gauge(vl->values[i].gauge);
310       break;
311     case DS_TYPE_DERIVE:
312       v->set_derive(vl->values[i].derive);
313       break;
314     case DS_TYPE_ABSOLUTE:
315       v->set_absolute(vl->values[i].absolute);
316       break;
317     default:
318       ERROR("grpc: invalid value type (%d)", value_type);
319       return grpc::Status(grpc::StatusCode::INTERNAL,
320                           grpc::string("unknown value type"));
321     }
322
323     auto name = msg->add_ds_names();
324     name->assign(ds->ds[i].name);
325   }
326
327   return grpc::Status::OK;
328 } /* marshal_value_list */
329
330 static grpc::Status unmarshal_value_list(const collectd::types::ValueList &msg,
331                                          value_list_t *vl) {
332   vl->time = NS_TO_CDTIME_T(TimeUtil::TimestampToNanoseconds(msg.time()));
333   vl->interval =
334       NS_TO_CDTIME_T(TimeUtil::DurationToNanoseconds(msg.interval()));
335
336   auto status = unmarshal_ident(msg.identifier(), vl, true);
337   if (!status.ok())
338     return status;
339
340   status = unmarshal_meta_data(msg.meta_data(), &vl->meta);
341   if (!status.ok())
342     return status;
343
344   value_t *values = NULL;
345   size_t values_len = 0;
346
347   status = grpc::Status::OK;
348   for (auto v : msg.values()) {
349     value_t *val =
350         (value_t *)realloc(values, (values_len + 1) * sizeof(*values));
351     if (!val) {
352       status = grpc::Status(grpc::StatusCode::RESOURCE_EXHAUSTED,
353                             grpc::string("failed to allocate values array"));
354       break;
355     }
356
357     values = val;
358     val = values + values_len;
359     values_len++;
360
361     switch (v.value_case()) {
362     case collectd::types::Value::ValueCase::kCounter:
363       val->counter = counter_t(v.counter());
364       break;
365     case collectd::types::Value::ValueCase::kGauge:
366       val->gauge = gauge_t(v.gauge());
367       break;
368     case collectd::types::Value::ValueCase::kDerive:
369       val->derive = derive_t(v.derive());
370       break;
371     case collectd::types::Value::ValueCase::kAbsolute:
372       val->absolute = absolute_t(v.absolute());
373       break;
374     default:
375       status = grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
376                             grpc::string("unknown value type"));
377       break;
378     }
379
380     if (!status.ok())
381       break;
382   }
383   if (status.ok()) {
384     vl->values = values;
385     vl->values_len = values_len;
386   } else {
387     meta_data_destroy(vl->meta);
388     free(values);
389   }
390
391   return status;
392 } /* unmarshal_value_list() */
393
394 /*
395  * Collectd service
396  */
397 class CollectdImpl : public collectd::Collectd::Service {
398 public:
399   grpc::Status
400   QueryValues(grpc::ServerContext *ctx, QueryValuesRequest const *req,
401               grpc::ServerWriter<QueryValuesResponse> *writer) override {
402     value_list_t match;
403     auto status = unmarshal_ident(req->identifier(), &match, false);
404     if (!status.ok()) {
405       return status;
406     }
407
408     std::queue<value_list_t> value_lists;
409     status = this->queryValuesRead(&match, &value_lists);
410     if (status.ok()) {
411       status = this->queryValuesWrite(ctx, writer, &value_lists);
412     }
413
414     while (!value_lists.empty()) {
415       auto vl = value_lists.front();
416       value_lists.pop();
417       sfree(vl.values);
418       meta_data_destroy(vl.meta);
419     }
420
421     return status;
422   }
423
424   grpc::Status PutValues(grpc::ServerContext *ctx,
425                          grpc::ServerReader<PutValuesRequest> *reader,
426                          PutValuesResponse *res) override {
427     PutValuesRequest req;
428
429     while (reader->Read(&req)) {
430       value_list_t vl = {0};
431       auto status = unmarshal_value_list(req.value_list(), &vl);
432       if (!status.ok())
433         return status;
434
435       if (plugin_dispatch_values(&vl))
436         return grpc::Status(
437             grpc::StatusCode::INTERNAL,
438             grpc::string("failed to enqueue values for writing"));
439     }
440
441     res->Clear();
442     return grpc::Status::OK;
443   }
444
445 private:
446   grpc::Status queryValuesRead(value_list_t const *match,
447                                std::queue<value_list_t> *value_lists) {
448     uc_iter_t *iter;
449     if ((iter = uc_get_iterator()) == NULL) {
450       return grpc::Status(
451           grpc::StatusCode::INTERNAL,
452           grpc::string("failed to query values: cannot create iterator"));
453     }
454
455     grpc::Status status = grpc::Status::OK;
456     char *name = NULL;
457     while (uc_iterator_next(iter, &name) == 0) {
458       value_list_t vl;
459       if (parse_identifier_vl(name, &vl) != 0) {
460         status = grpc::Status(grpc::StatusCode::INTERNAL,
461                               grpc::string("failed to parse identifier"));
462         break;
463       }
464
465       if (!ident_matches(&vl, match))
466         continue;
467       if (uc_iterator_get_time(iter, &vl.time) < 0) {
468         status =
469             grpc::Status(grpc::StatusCode::INTERNAL,
470                          grpc::string("failed to retrieve value timestamp"));
471         break;
472       }
473       if (uc_iterator_get_interval(iter, &vl.interval) < 0) {
474         status =
475             grpc::Status(grpc::StatusCode::INTERNAL,
476                          grpc::string("failed to retrieve value interval"));
477         break;
478       }
479       if (uc_iterator_get_values(iter, &vl.values, &vl.values_len) < 0) {
480         status = grpc::Status(grpc::StatusCode::INTERNAL,
481                               grpc::string("failed to retrieve values"));
482         break;
483       }
484       if (uc_iterator_get_meta(iter, &vl.meta) < 0) {
485         status = grpc::Status(grpc::StatusCode::INTERNAL,
486                               grpc::string("failed to retrieve value metadata"));
487       }
488
489       value_lists->push(vl);
490     } // while (uc_iterator_next(iter, &name) == 0)
491
492     uc_iterator_destroy(iter);
493     return status;
494   }
495
496   grpc::Status queryValuesWrite(grpc::ServerContext *ctx,
497                                 grpc::ServerWriter<QueryValuesResponse> *writer,
498                                 std::queue<value_list_t> *value_lists) {
499     while (!value_lists->empty()) {
500       auto vl = value_lists->front();
501       QueryValuesResponse res;
502       res.Clear();
503
504       auto status = marshal_value_list(&vl, res.mutable_value_list());
505       if (!status.ok()) {
506         return status;
507       }
508
509       if (!writer->Write(res)) {
510         return grpc::Status::CANCELLED;
511       }
512
513       value_lists->pop();
514       sfree(vl.values);
515     }
516
517     return grpc::Status::OK;
518   }
519 };
520
521 /*
522  * gRPC server implementation
523  */
524 class CollectdServer final {
525 public:
526   void Start() {
527     auto auth = grpc::InsecureServerCredentials();
528
529     grpc::ServerBuilder builder;
530
531     if (listeners.empty()) {
532       builder.AddListeningPort(default_addr, auth);
533       INFO("grpc: Listening on %s", default_addr.c_str());
534     } else {
535       for (auto l : listeners) {
536         grpc::string addr = l.addr + ":" + l.port;
537
538         auto use_ssl = grpc::string("");
539         auto a = auth;
540         if (l.ssl != nullptr) {
541           use_ssl = grpc::string(" (SSL enabled)");
542           a = grpc::SslServerCredentials(*l.ssl);
543         }
544
545         builder.AddListeningPort(addr, a);
546         INFO("grpc: Listening on %s%s", addr.c_str(), use_ssl.c_str());
547       }
548     }
549
550     builder.RegisterService(&collectd_service_);
551
552     server_ = builder.BuildAndStart();
553   } /* Start() */
554
555   void Shutdown() { server_->Shutdown(); } /* Shutdown() */
556
557 private:
558   CollectdImpl collectd_service_;
559
560   std::unique_ptr<grpc::Server> server_;
561 }; /* class CollectdServer */
562
563 class CollectdClient final {
564 public:
565   CollectdClient(std::shared_ptr<grpc::ChannelInterface> channel)
566       : stub_(Collectd::NewStub(channel)) {}
567
568   int PutValues(value_list_t const *vl) {
569     grpc::ClientContext ctx;
570
571     PutValuesRequest req;
572     auto status = marshal_value_list(vl, req.mutable_value_list());
573     if (!status.ok()) {
574       ERROR("grpc: Marshalling value_list_t failed.");
575       return -1;
576     }
577
578     PutValuesResponse res;
579     auto stream = stub_->PutValues(&ctx, &res);
580     if (!stream->Write(req)) {
581       NOTICE("grpc: Broken stream.");
582       /* intentionally not returning. */
583     }
584
585     stream->WritesDone();
586     status = stream->Finish();
587     if (!status.ok()) {
588       ERROR("grpc: Error while closing stream.");
589       return -1;
590     }
591
592     return 0;
593   } /* int PutValues */
594
595 private:
596   std::unique_ptr<Collectd::Stub> stub_;
597 };
598
599 static CollectdServer *server = nullptr;
600
601 /*
602  * collectd plugin interface
603  */
604 extern "C" {
605 static void c_grpc_destroy_write_callback(void *ptr) {
606   delete (CollectdClient *)ptr;
607 }
608
609 static int c_grpc_write(__attribute__((unused)) data_set_t const *ds,
610                         value_list_t const *vl, user_data_t *ud) {
611   CollectdClient *c = (CollectdClient *)ud->data;
612   return c->PutValues(vl);
613 }
614
615 static int c_grpc_config_listen(oconfig_item_t *ci) {
616   if ((ci->values_num != 2) || (ci->values[0].type != OCONFIG_TYPE_STRING) ||
617       (ci->values[1].type != OCONFIG_TYPE_STRING)) {
618     ERROR("grpc: The `%s` config option needs exactly "
619           "two string argument (address and port).",
620           ci->key);
621     return -1;
622   }
623
624   auto listener = Listener();
625   listener.addr = grpc::string(ci->values[0].value.string);
626   listener.port = grpc::string(ci->values[1].value.string);
627   listener.ssl = nullptr;
628
629   auto ssl_opts = new (grpc::SslServerCredentialsOptions);
630   grpc::SslServerCredentialsOptions::PemKeyCertPair pkcp = {};
631   bool use_ssl = false;
632
633   for (int i = 0; i < ci->children_num; i++) {
634     oconfig_item_t *child = ci->children + i;
635
636     if (!strcasecmp("EnableSSL", child->key)) {
637       if (cf_util_get_boolean(child, &use_ssl)) {
638         ERROR("grpc: Option `%s` expects a boolean value", child->key);
639         return -1;
640       }
641     } else if (!strcasecmp("SSLCACertificateFile", child->key)) {
642       char *certs = NULL;
643       if (cf_util_get_string(child, &certs)) {
644         ERROR("grpc: Option `%s` expects a string value", child->key);
645         return -1;
646       }
647       ssl_opts->pem_root_certs = read_file(certs);
648     } else if (!strcasecmp("SSLCertificateKeyFile", child->key)) {
649       char *key = NULL;
650       if (cf_util_get_string(child, &key)) {
651         ERROR("grpc: Option `%s` expects a string value", child->key);
652         return -1;
653       }
654       pkcp.private_key = read_file(key);
655     } else if (!strcasecmp("SSLCertificateFile", child->key)) {
656       char *cert = NULL;
657       if (cf_util_get_string(child, &cert)) {
658         ERROR("grpc: Option `%s` expects a string value", child->key);
659         return -1;
660       }
661       pkcp.cert_chain = read_file(cert);
662     } else {
663       WARNING("grpc: Option `%s` not allowed in <%s> block.", child->key,
664               ci->key);
665     }
666   }
667
668   ssl_opts->pem_key_cert_pairs.push_back(pkcp);
669   if (use_ssl)
670     listener.ssl = ssl_opts;
671   else
672     delete (ssl_opts);
673
674   listeners.push_back(listener);
675   return 0;
676 } /* c_grpc_config_listen() */
677
678 static int c_grpc_config_server(oconfig_item_t *ci) {
679   if ((ci->values_num != 2) || (ci->values[0].type != OCONFIG_TYPE_STRING) ||
680       (ci->values[1].type != OCONFIG_TYPE_STRING)) {
681     ERROR("grpc: The `%s` config option needs exactly "
682           "two string argument (address and port).",
683           ci->key);
684     return -1;
685   }
686
687   grpc::SslCredentialsOptions ssl_opts;
688   bool use_ssl = false;
689
690   for (int i = 0; i < ci->children_num; i++) {
691     oconfig_item_t *child = ci->children + i;
692
693     if (!strcasecmp("EnableSSL", child->key)) {
694       if (cf_util_get_boolean(child, &use_ssl)) {
695         return -1;
696       }
697     } else if (!strcasecmp("SSLCACertificateFile", child->key)) {
698       char *certs = NULL;
699       if (cf_util_get_string(child, &certs)) {
700         return -1;
701       }
702       ssl_opts.pem_root_certs = read_file(certs);
703     } else if (!strcasecmp("SSLCertificateKeyFile", child->key)) {
704       char *key = NULL;
705       if (cf_util_get_string(child, &key)) {
706         return -1;
707       }
708       ssl_opts.pem_private_key = read_file(key);
709     } else if (!strcasecmp("SSLCertificateFile", child->key)) {
710       char *cert = NULL;
711       if (cf_util_get_string(child, &cert)) {
712         return -1;
713       }
714       ssl_opts.pem_cert_chain = read_file(cert);
715     } else {
716       WARNING("grpc: Option `%s` not allowed in <%s> block.", child->key,
717               ci->key);
718     }
719   }
720
721   auto node = grpc::string(ci->values[0].value.string);
722   auto service = grpc::string(ci->values[1].value.string);
723   auto addr = node + ":" + service;
724
725   CollectdClient *client;
726   if (use_ssl) {
727     auto channel_creds = grpc::SslCredentials(ssl_opts);
728     auto channel = grpc::CreateChannel(addr, channel_creds);
729     client = new CollectdClient(channel);
730   } else {
731     auto channel =
732         grpc::CreateChannel(addr, grpc::InsecureChannelCredentials());
733     client = new CollectdClient(channel);
734   }
735
736   auto callback_name = grpc::string("grpc/") + addr;
737   user_data_t ud = {
738       .data = client, .free_func = c_grpc_destroy_write_callback,
739   };
740
741   plugin_register_write(callback_name.c_str(), c_grpc_write, &ud);
742   return 0;
743 } /* c_grpc_config_server() */
744
745 static int c_grpc_config(oconfig_item_t *ci) {
746   int i;
747
748   for (i = 0; i < ci->children_num; i++) {
749     oconfig_item_t *child = ci->children + i;
750
751     if (!strcasecmp("Listen", child->key)) {
752       if (c_grpc_config_listen(child))
753         return -1;
754     } else if (!strcasecmp("Server", child->key)) {
755       if (c_grpc_config_server(child))
756         return -1;
757     }
758
759     else {
760       WARNING("grpc: Option `%s` not allowed here.", child->key);
761     }
762   }
763
764   return 0;
765 } /* c_grpc_config() */
766
767 static int c_grpc_init(void) {
768   server = new CollectdServer();
769   if (!server) {
770     ERROR("grpc: Failed to create server");
771     return -1;
772   }
773
774   server->Start();
775   return 0;
776 } /* c_grpc_init() */
777
778 static int c_grpc_shutdown(void) {
779   if (!server)
780     return 0;
781
782   server->Shutdown();
783
784   delete server;
785   server = nullptr;
786
787   return 0;
788 } /* c_grpc_shutdown() */
789
790 void module_register(void) {
791   plugin_register_complex_config("grpc", c_grpc_config);
792   plugin_register_init("grpc", c_grpc_init);
793   plugin_register_shutdown("grpc", c_grpc_shutdown);
794 } /* module_register() */
795 } /* extern "C" */