grpc plugin: Always populate *instance fields in a value-list.

[collectd.git] / src / grpc.cc

diff --git a/src/grpc.cc b/src/grpc.cc
index 9523fc2..0330aa0 100644 (file)
--- a/src/grpc.cc
+++ b/src/grpc.cc
@@ -38,6 +38,8 @@ extern "C" {
 #include "configfile.h"
 #include "plugin.h"
 
+#include "daemon/utils_cache.h"
+
        typedef struct {
                char *addr;
                char *port;
@@ -47,6 +49,13 @@ extern "C" {
        static size_t listeners_num;
 }
 
+using collectd::Collectd;
+
+using collectd::DispatchValuesRequest;
+using collectd::DispatchValuesReply;
+using collectd::ListValuesRequest;
+using collectd::ListValuesReply;
+
 using google::protobuf::util::TimeUtil;
 
 /*
@@ -79,12 +88,10 @@ static grpc::Status unmarshal_value_list(const collectd::types::ValueList &msg,
        sstrncpy(vl->type, s.c_str(), sizeof(vl->type));
 
        s = msg.plugin_instance();
-       if (s.length())
-               sstrncpy(vl->plugin_instance, s.c_str(), sizeof(vl->plugin_instance));
+       sstrncpy(vl->plugin_instance, s.c_str(), sizeof(vl->plugin_instance));
 
        s = msg.type_instance();
-       if (s.length())
-               sstrncpy(vl->type_instance, s.c_str(), sizeof(vl->type_instance));
+       sstrncpy(vl->type_instance, s.c_str(), sizeof(vl->type_instance));
 
        value_t *values = NULL;
        size_t values_len = 0;
@@ -117,7 +124,7 @@ static grpc::Status unmarshal_value_list(const collectd::types::ValueList &msg,
                        break;
                default:
                        status = grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
-                                       grpc::string("unkown value type"));
+                                       grpc::string("unknown value type"));
                        break;
                }
 
@@ -136,13 +143,51 @@ static grpc::Status unmarshal_value_list(const collectd::types::ValueList &msg,
 } /* unmarshal_value_list() */
 
 /*
- * request call objects
+ * request call-backs and call objects
  */
 
+static grpc::Status Process(grpc::ServerContext *ctx,
+               DispatchValuesRequest request, DispatchValuesReply *reply)
+{
+       value_list_t vl = VALUE_LIST_INIT;
+       auto status = unmarshal_value_list(request.values(), &vl);
+       if (!status.ok())
+               return status;
+
+       if (plugin_dispatch_values(&vl))
+               status = grpc::Status(grpc::StatusCode::INTERNAL,
+                               grpc::string("failed to enqueue values for writing"));
+       return status;
+} /* Process(): DispatchValues */
+
+static grpc::Status Process(grpc::ServerContext *ctx,
+               ListValuesRequest request, ListValuesReply *reply)
+{
+       char **names = NULL;
+       cdtime_t *times = NULL;
+       size_t i, n = 0;
+
+       if (uc_get_names(&names, &times, &n))
+               return grpc::Status(grpc::StatusCode::INTERNAL,
+                               grpc::string("failed to retrieve values"));
+
+       for (i = 0; i < n; i++) {
+               auto v = reply->add_value();
+               auto t = TimeUtil::NanosecondsToTimestamp(CDTIME_T_TO_NS(times[i]));
+               v->set_name(names[i]);
+               v->set_allocated_time(new google::protobuf::Timestamp(t));
+               sfree(names[i]);
+       }
+       sfree(names);
+       sfree(times);
+
+       return grpc::Status::OK;
+} /* Process(): ListValues */
+
 class Call
 {
 public:
-       Call(collectd::Collectd::AsyncService *service, grpc::ServerCompletionQueue *cq)
+       Call(Collectd::AsyncService *service, grpc::ServerCompletionQueue *cq)
                : service_(service), cq_(cq), status_(CREATE)
        { }
 
@@ -170,7 +215,7 @@ protected:
        virtual void Process() = 0;
        virtual void Finish() = 0;
 
-       collectd::Collectd::AsyncService *service_;
+       Collectd::AsyncService *service_;
        grpc::ServerCompletionQueue *cq_;
        grpc::ServerContext ctx_;
 
@@ -179,40 +224,35 @@ private:
        CallStatus status_;
 }; /* class Call */
 
-class DispatchValuesCall : public Call
+template<typename RequestT, typename ReplyT>
+class RpcCall final : public Call
 {
+       typedef void (Collectd::AsyncService::*CreatorT)(grpc::ServerContext *,
+                       RequestT *, grpc::ServerAsyncResponseWriter<ReplyT> *,
+                       grpc::CompletionQueue *, grpc::ServerCompletionQueue *, void *);
+
 public:
-       DispatchValuesCall(collectd::Collectd::AsyncService *service, grpc::ServerCompletionQueue *cq)
-               : Call(service, cq), responder_(&ctx_)
+       RpcCall(Collectd::AsyncService *service,
+                       CreatorT creator, grpc::ServerCompletionQueue *cq)
+               : Call(service, cq), creator_(creator), responder_(&ctx_)
        {
                Handle();
-       } /* DispatchValuesCall() */
+       } /* RpcCall() */
 
-       virtual ~DispatchValuesCall()
+       virtual ~RpcCall()
        { }
 
 private:
        void Create()
        {
-               service_->RequestDispatchValues(&ctx_, &request_, &responder_, cq_, cq_, this);
+               (service_->*creator_)(&ctx_, &request_, &responder_, cq_, cq_, this);
        } /* Create() */
 
        void Process()
        {
                // Add a new request object to the queue.
-               new DispatchValuesCall(service_, cq_);
-
-               value_list_t vl = VALUE_LIST_INIT;
-               auto status = unmarshal_value_list(request_.values(), &vl);
-               if (!status.ok()) {
-                       responder_.Finish(reply_, status, this);
-                       return;
-               }
-
-               if (plugin_dispatch_values(&vl))
-                       status = grpc::Status(grpc::StatusCode::INTERNAL,
-                                       grpc::string("failed to enqueue values for writing"));
-
+               new RpcCall<RequestT, ReplyT>(service_, creator_, cq_);
+               grpc::Status status = ::Process(&ctx_, request_, &reply_);
                responder_.Finish(reply_, status, this);
        } /* Process() */
 
@@ -221,11 +261,13 @@ private:
                delete this;
        } /* Finish() */
 
-       collectd::DispatchValuesRequest request_;
-       collectd::DispatchValuesReply reply_;
+       CreatorT creator_;
+
+       RequestT request_;
+       ReplyT reply_;
 
-       grpc::ServerAsyncResponseWriter<collectd::DispatchValuesReply> responder_;
-};
+       grpc::ServerAsyncResponseWriter<ReplyT> responder_;
+}; /* class RpcCall */
 
 /*
  * gRPC server implementation
@@ -257,7 +299,7 @@ public:
                        }
                }
 
-               builder.RegisterAsyncService(&service_);
+               builder.RegisterService(&service_);
                cq_ = builder.AddCompletionQueue();
                server_ = builder.BuildAndStart();
        } /* Start() */
@@ -271,7 +313,10 @@ public:
        void Mainloop()
        {
                // Register request types.
-               new DispatchValuesCall(&service_, cq_.get());
+               new RpcCall<DispatchValuesRequest, DispatchValuesReply>(&service_,
+                               &Collectd::AsyncService::RequestDispatchValues, cq_.get());
+               new RpcCall<ListValuesRequest, ListValuesReply>(&service_,
+                               &Collectd::AsyncService::RequestListValues, cq_.get());
 
                while (true) {
                        void *req = NULL;
@@ -289,7 +334,7 @@ public:
        } /* Mainloop() */
 
 private:
-       collectd::Collectd::AsyncService service_;
+       Collectd::AsyncService service_;
 
        std::unique_ptr<grpc::Server> server_;
        std::unique_ptr<grpc::ServerCompletionQueue> cq_;