[collectd 6] New plugin (sort of): OpenTelemetry receiver

Makefile.am

@@ -484,7 +484,7 @@ libformat_influxdb_la_SOURCES = \
 	src/utils/format_influxdb/format_influxdb.c \
 	src/utils/format_influxdb/format_influxdb.h
 
-if BUILD_PLUGIN_WRITE_OPEN_TELEMETRY
+if BUILD_PLUGIN_OPEN_TELEMETRY
 noinst_LTLIBRARIES += libformat_open_telemetry.la
 libformat_open_telemetry_la_SOURCES = \
 	src/utils/format_open_telemetry/format_open_telemetry.cc \
@@ -1744,6 +1744,19 @@ openvpn_la_SOURCES = src/openvpn.c
 openvpn_la_LDFLAGS = $(PLUGIN_LDFLAGS)
 endif
 
+if BUILD_PLUGIN_OPEN_TELEMETRY
+pkglib_LTLIBRARIES += open_telemetry.la
+open_telemetry_la_SOURCES = \
+	src/open_telemetry.cc src/open_telemetry.h \
+	src/open_telemetry_exporter.cc \
+	src/open_telemetry_receiver.cc \
+	opentelemetry/proto/collector/metrics/v1/metrics_service.grpc.pb.cc \
+	opentelemetry/proto/collector/metrics/v1/metrics_service.grpc.pb.h
+open_telemetry_la_CPPFLAGS = $(AM_CPPFLAGS) $(BUILD_WITH_LIBGRPCPP_CPPFLAGS) $(BUILD_WITH_LIBPROTOBUF_CPPFLAGS)
+open_telemetry_la_LDFLAGS = $(PLUGIN_LDFLAGS) $(BUILD_WITH_LIBGRPCPP_LDFLAGS) $(BUILD_WITH_LIBPROTOBUF_LDFLAGS)
+open_telemetry_la_LIBADD = $(BUILD_WITH_LIBGRPCPP_LIBS) $(BUILD_WITH_LIBPROTOBUF_LIBS) libformat_open_telemetry.la
+endif
+
 if BUILD_PLUGIN_ORACLE
 pkglib_LTLIBRARIES += oracle.la
 oracle_la_SOURCES = \
@@ -2359,16 +2372,6 @@ write_mongodb_la_LDFLAGS = $(PLUGIN_LDFLAGS) $(BUILD_WITH_LIBMONGOC_LDFLAGS)
 write_mongodb_la_LIBADD = $(BUILD_WITH_LIBMONGOC_LIBS)
 endif
 
-if BUILD_PLUGIN_WRITE_OPEN_TELEMETRY
-pkglib_LTLIBRARIES += write_open_telemetry.la
-write_open_telemetry_la_SOURCES = src/write_open_telemetry.cc \
-				  opentelemetry/proto/collector/metrics/v1/metrics_service.grpc.pb.cc \
-				  opentelemetry/proto/collector/metrics/v1/metrics_service.grpc.pb.h
-write_open_telemetry_la_CPPFLAGS = $(AM_CPPFLAGS) $(BUILD_WITH_LIBGRPCPP_CPPFLAGS)
-write_open_telemetry_la_LDFLAGS = $(PLUGIN_LDFLAGS) $(BUILD_WITH_LIBGRPCPP_LDFLAGS)
-write_open_telemetry_la_LIBADD = $(BUILD_WITH_LIBGRPCPP_LIBS) libformat_open_telemetry.la
-endif
-
 if BUILD_PLUGIN_WRITE_PROMETHEUS
 pkglib_LTLIBRARIES += write_prometheus.la
 write_prometheus_la_SOURCES = src/write_prometheus.c
@@ -2517,7 +2520,7 @@ types.pb.cc: $(srcdir)/proto/types.proto
 endif
 endif
 
-if BUILD_PLUGIN_WRITE_OPEN_TELEMETRY
+if BUILD_PLUGIN_OPEN_TELEMETRY
 BUILT_SOURCES += \
 	opentelemetry/proto/collector/metrics/v1/metrics_service.grpc.pb.cc \
 	opentelemetry/proto/collector/metrics/v1/metrics_service.grpc.pb.h \

configure.ac

@@ -2880,6 +2880,37 @@ if test "x$with_libgrpcpp" = "xyes"; then
   AC_LANG_POP(C++)
 fi
 
+if test "x$with_libgrpcpp" = "xyes"; then
+  AC_MSG_CHECKING([for grpc++_reflection])
+  AC_LANG_PUSH(C++)
+  SAVE_CPPFLAGS="$CPPFLAGS"
+  SAVE_LDFLAGS="$LDFLAGS"
+  SAVE_LIBS="$LIBS"
+  CPPFLAGS="-std=c++11 $with_libgrpcpp_cppflags $GRPCPP_CFLAGS $CPPFLAGS"
+  LDFLAGS="$with_libgrpcpp_ldflags"
+  LIBS="-lgrpc++_reflection $GRPCPP_LIBS"
+  AC_LINK_IFELSE(
+    [
+      AC_LANG_PROGRAM(
+        [[#include <grpc++/ext/proto_server_reflection_plugin.h>]],
+        [[grpc::reflection::InitProtoReflectionServerBuilderPlugin();]]
+      )
+    ],
+    [
+      AC_MSG_RESULT([yes])
+      GRPCPP_LIBS="$LIBS"
+      AC_DEFINE([HAVE_GRPCPP_REFLECTION], [1], [Define if the grpc++_reflection library is available.])
+    ],
+    [
+      AC_MSG_RESULT([no])
+    ]
+  )
+  CPPFLAGS="$SAVE_CPPFLAGS"
+  LDFLAGS="$SAVE_LDFLAGS"
+  LIBS="$SAVE_LIBS"
+  AC_LANG_POP(C++)
+fi
+
 BUILD_WITH_LIBGRPCPP_CPPFLAGS="-std=c++11 $with_libgrpcpp_cppflags $GRPCPP_CFLAGS"
 BUILD_WITH_LIBGRPCPP_LDFLAGS="$with_libgrpcpp_ldflags"
 BUILD_WITH_LIBGRPCPP_LIBS="$GRPCPP_LIBS"
@@ -6810,6 +6841,7 @@ plugin_nut="$with_libupsclient"
 plugin_olsrd="yes"
 plugin_onewire="$with_libowcapi"
 plugin_openldap="$with_libldap"
+plugin_open_telemetry="yes"
 plugin_openvpn="yes"
 plugin_oracle="$with_oracle"
 plugin_ovs_events="no"
@@ -6874,7 +6906,6 @@ plugin_write_influxdb_udp="yes"
 plugin_write_kafka="$with_librdkafka"
 plugin_write_log="no"
 plugin_write_mongodb="$with_libmongoc"
-plugin_write_open_telemetry="yes"
 plugin_write_prometheus="no"
 plugin_write_redis="yes"
 plugin_write_riemann="$with_libriemann_client"
@@ -7145,22 +7176,22 @@ fi
 
 if test "x$GRPC_CPP_PLUGIN" = "x"; then
   plugin_grpc="no (grpc_cpp_plugin not found)"
-  plugin_write_open_telemetry="no (grpc_cpp_plugin not found)"
+  plugin_open_telemetry="no (grpc_cpp_plugin not found)"
 fi
 if test "x$have_protoc3" != "xyes"; then
   plugin_grpc="no (protoc3 not found)"
-  plugin_write_open_telemetry="no (protoc3 not found)"
+  plugin_open_telemetry="no (protoc3 not found)"
 fi
 if test "x$with_libprotobuf" != "xyes"; then
   plugin_grpc="no (libprotobuf not found)"
-  plugin_write_open_telemetry="no (libprotobuf not found)"
+  plugin_open_telemetry="no (libprotobuf not found)"
 fi
 if test "x$with_libgrpcpp" != "xyes"; then
   plugin_grpc="no (libgrpc++ not found)"
-  plugin_write_open_telemetry="no (libgrpc++ not found)"
+  plugin_open_telemetry="no (libgrpc++ not found)"
 fi
 if test "x$protoc3_optional" = "xno"; then
-  plugin_write_open_telemetry="no (protoc does not support optional fields)"
+  plugin_open_telemetry="no (protoc does not support optional fields)"
 fi
 
 if test "x$have_getifaddrs" = "xyes"; then
@@ -7517,6 +7548,7 @@ AC_PLUGIN([olsrd],               [$plugin_olsrd],             [olsrd statistics]
 AC_PLUGIN([onewire],             [$plugin_onewire],           [OneWire sensor statistics])
 AC_PLUGIN([openldap],            [$plugin_openldap],          [OpenLDAP statistics])
 AC_PLUGIN([openvpn],             [$plugin_openvpn],           [OpenVPN client statistics])
+AC_PLUGIN([open_telemetry],      [$plugin_open_telemetry],    [OpenTelemetry exporter/receiver])
 AC_PLUGIN([oracle],              [$plugin_oracle],            [Oracle plugin])
 AC_PLUGIN([ovs_events],          [$plugin_ovs_events],        [OVS events plugin])
 AC_PLUGIN([ovs_stats],           [$plugin_ovs_stats],         [OVS statistics plugin])
@@ -7581,7 +7613,6 @@ AC_PLUGIN([write_influxdb_udp],  [$plugin_write_influxdb_udp],[Influxdb udp outp
 AC_PLUGIN([write_kafka],         [$plugin_write_kafka],       [Kafka output plugin])
 AC_PLUGIN([write_log],           [$plugin_write_log],         [Log output plugin])
 AC_PLUGIN([write_mongodb],       [$plugin_write_mongodb],     [MongoDB output plugin])
-AC_PLUGIN([write_open_telemetry],[$plugin_write_open_telemetry],[Write OpenTelemetry plugin])
 AC_PLUGIN([write_prometheus],    [$plugin_write_prometheus],  [Prometheus write plugin])
 AC_PLUGIN([write_redis],         [$plugin_write_redis],       [Redis output plugin])
 AC_PLUGIN([write_riemann],       [$plugin_write_riemann],     [Riemann output plugin])
@@ -7964,6 +7995,7 @@ AC_MSG_RESULT([    nut . . . . . . . . . $enable_nut])
 AC_MSG_RESULT([    olsrd . . . . . . . . $enable_olsrd])
 AC_MSG_RESULT([    onewire . . . . . . . $enable_onewire])
 AC_MSG_RESULT([    openldap  . . . . . . $enable_openldap])
+AC_MSG_RESULT([    open_telemetry  . . . $enable_open_telemetry])
 AC_MSG_RESULT([    openvpn . . . . . . . $enable_openvpn])
 AC_MSG_RESULT([    oracle  . . . . . . . $enable_oracle])
 AC_MSG_RESULT([    ovs_events  . . . . . $enable_ovs_events])
@@ -8028,7 +8060,6 @@ AC_MSG_RESULT([    write_influxdb_udp. . $enable_write_influxdb_udp])
 AC_MSG_RESULT([    write_kafka . . . . . $enable_write_kafka])
 AC_MSG_RESULT([    write_log . . . . . . $enable_write_log])
 AC_MSG_RESULT([    write_mongodb . . . . $enable_write_mongodb])
-AC_MSG_RESULT([    write_open_telemetry  $enable_write_open_telemetry])
 AC_MSG_RESULT([    write_prometheus. . . $enable_write_prometheus])
 AC_MSG_RESULT([    write_redis . . . . . $enable_write_redis])
 AC_MSG_RESULT([    write_riemann . . . . $enable_write_riemann])

src/collectd.conf.in

@@ -182,6 +182,7 @@
 #@BUILD_PLUGIN_OLSRD_TRUE@LoadPlugin olsrd
 #@BUILD_PLUGIN_ONEWIRE_TRUE@LoadPlugin onewire
 #@BUILD_PLUGIN_OPENLDAP_TRUE@LoadPlugin openldap
+#@BUILD_PLUGIN_OPEN_TELEMETRY_TRUE@LoadPlugin open_telemetry
 #@BUILD_PLUGIN_OPENVPN_TRUE@LoadPlugin openvpn
 #@BUILD_PLUGIN_ORACLE_TRUE@LoadPlugin oracle
 #@BUILD_PLUGIN_OVS_EVENTS_TRUE@LoadPlugin ovs_events
@@ -236,7 +237,6 @@
 #@BUILD_PLUGIN_WRITE_KAFKA_TRUE@LoadPlugin write_kafka
 #@BUILD_PLUGIN_WRITE_LOG_TRUE@LoadPlugin write_log
 #@BUILD_PLUGIN_WRITE_MONGODB_TRUE@LoadPlugin write_mongodb
-#@BUILD_PLUGIN_WRITE_OPEN_TELEMETRY_TRUE@LoadPlugin write_open_telemetry
 #@BUILD_PLUGIN_WRITE_PROMETHEUS_TRUE@LoadPlugin write_prometheus
 #@BUILD_PLUGIN_WRITE_REDIS_TRUE@LoadPlugin write_redis
 #@BUILD_PLUGIN_WRITE_RIEMANN_TRUE@LoadPlugin write_riemann
@@ -1331,6 +1331,22 @@
 #	CollectUserCount false
 #</Plugin>
 
+#<Plugin open_telemetry>
+#  <Receiver "0.0.0.0" "4317">
+#    EnableSSL false
+#    SSLCACertificateFile "/path/to/root.pem"
+#    SSLCertificateFile "/path/to/client.pem"
+#    SSLCertificateKeyFile "/path/to/client.key"
+#    VerifyPeer true
+#  </Receiver>
+#  <Exporter "localhost" "4317">
+#    EnableSSL false
+#    SSLCACertificateFile "/path/to/root.pem"
+#    SSLCertificateFile "/path/to/client.pem"
+#    SSLCertificateKeyFile "/path/to/client.key"
+#  </Exporter>
+#</Plugin>
+
 #<Plugin oracle>
 #  <Query "out_of_stock">
 #    Statement "SELECT category, COUNT(*) AS value FROM products WHERE in_stock = 0 GROUP BY category"
@@ -2070,13 +2086,6 @@
 #	</Node>
 #</Plugin>
 
-#<Plugin write_open_telemetry>
-#  <Node "example">
-#    Host "localhost"
-#    Port "4317"
-#  </Node>
-#</Plugin>
-
 #<Plugin write_prometheus>
 #	Port "9103"
 #</Plugin>

src/collectd.conf.pod

@@ -7153,6 +7153,101 @@ can be configured independently from that option. Defaults to B<false>.
 
 =back
 
+=head2 Plugin C<open_telemetry>
+
+The I<open_telemetry plugin> implements an OpenTelemetry exporter and receiver
+using OTLP. Specifically, it implements a gRPC C<MetricsService> client and
+server.
+
+B<Synopsis:>
+
+  <Plugin open_telemetry>
+    <Receiver "0.0.0.0" "4317">
+      EnableSSL false
+      SSLCACertificateFile "/path/to/root.pem"
+      SSLCertificateFile "/path/to/client.pem"
+      SSLCertificateKeyFile "/path/to/client.key"
+      VerifyPeer true
+    </Receiver>
+    <Exporter "localhost" "4317">
+      EnableSSL false
+      SSLCACertificateFile "/path/to/root.pem"
+      SSLCertificateFile "/path/to/client.pem"
+      SSLCertificateKeyFile "/path/to/client.key"
+    </Exporter>
+  </Plugin>
+
+=over 4
+
+=item B<Receiver> I<Host> [I<Port>]
+
+The B<Receiver> statement sets the network address and port to which to bind.
+Multiple B<Receiver> blocks can be specified to listen on multiple
+interfaces/ports. When no B<Receiver> blocks are specified, the plugin will not
+receive any metrics.
+
+The argument I<Host> may be a hostname, an IPv4 address, or an IPv6 address.
+
+The I<Port> argument may be omitted. In that case the default C<"4317"> is
+used.
+
+Optionally, B<Receiver> blocks support the following options:
+
+=over 4
+
+=item B<EnableSSL> I<false>|I<true>
+
+Whether to enable SSL for incoming connections. Default: I<false>.
+
+=item B<SSLCACertificateFile> I<Filename>
+
+=item B<SSLCertificateFile> I<Filename>
+
+=item B<SSLCertificateKeyFile> I<Filename>
+
+Filenames specifying SSL certificate and key material to be used with SSL
+connections.
+
+=item B<VerifyPeer> B<true>|B<false>
+
+When enabled, a valid client certificate is required to connect to the server.
+When disabled, a client certificate is not requested and any unsolicited client
+certificate is accepted.
+Enabled by default.
+
+=back
+
+=item B<Exporter> I<Host> [I<Port>]
+
+The B<Exporter> option configures an OTLP export to the given collector
+address. Multiple collectors can be specified using multiple B<Exporter> lines.
+
+The argument I<Host> may be a hostname, an IPv4 address, or an IPv6 address.
+
+The I<Port> argument may be omitted. In that case the default C<"4317"> is
+used.
+
+Optionally, B<Exporter> blocks support the following options:
+
+=over 4
+
+=item B<EnableSSL> I<false>|I<true>
+
+Whether to require SSL when connecting. Default: I<false>.
+
+=item B<SSLCACertificateFile> I<Filename>
+
+=item B<SSLCertificateFile> I<Filename>
+
+=item B<SSLCertificateKeyFile> I<Filename>
+
+Filenames specifying SSL certificate and key material to be used with SSL
+connections.
+
+=back
+
+=back
+
 =head2 Plugin C<oracle>
 
 The "oracle" plugin uses the Oracle® Call Interface I<(OCI)> to connect to an
@@ -11151,35 +11246,6 @@ want to use authentication all three fields must be set.
 
 =back
 
-=head2 Plugin C<write_open_telemetry>
-
-The I<write_open_telemetry plugin> will export metrics to an I<OpenTelemetry
-collector> using the gRPC based OTLP protocol.
-
-B<Synopsis:>
-
- <Plugin "write_open_telemetry">
-   <Node "default">
-     Host "localhost"
-     Port "4317"
-   </Node>
- </Plugin>
-
-The plugin can export metrics to multiple collectors by specifying multiple
-B<Node> blocks. Within the B<Node> blocks, the following options are available:
-
-=over 4
-
-=item B<Host> I<Address>
-
-Hostname or address to connect to. Defaults to C<localhost>.
-
-=item B<Port> I<Service>
-
-Service name or port number to connect to. Defaults to C<4317>.
-
-=back
-
 =head2 Plugin C<write_prometheus>
 
 The I<write_prometheus plugin> implements a tiny webserver that can be scraped

src/daemon/utils_cache.c

@@ -523,13 +523,13 @@ int uc_get_value_by_name(const char *name, value_t *ret_values) {
 
     /* remove missing values from getval */
     if (ce->state == STATE_MISSING) {
-      status = -1;
+      status = EAGAIN;
     } else {
       *ret_values = ce->last_value;
     }
   } else {
     DEBUG("utils_cache: uc_get_value_by_name: No such value: %s", name);
-    status = -1;
+    status = ENOENT;
   }
 
   pthread_mutex_unlock(&cache_lock);
@@ -739,7 +739,7 @@ int uc_get_history_by_name(const char *name, gauge_t *ret_history,
   status = c_avl_get(cache_tree, name, (void *)&ce);
   if (status != 0) {
     pthread_mutex_unlock(&cache_lock);
-    return -ENOENT;
+    return ENOENT;
   }
   /* Check if there are enough values available. If not, increase the buffer
    * size. */
@@ -749,7 +749,7 @@ int uc_get_history_by_name(const char *name, gauge_t *ret_history,
     tmp = realloc(ce->history, sizeof(*ce->history) * num_steps);
     if (tmp == NULL) {
       pthread_mutex_unlock(&cache_lock);
-      return -ENOMEM;
+      return ENOMEM;
     }
 
     for (size_t i = ce->history_length; i < num_steps; i++)