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