Implements new capability fields in aclk_schemas (#12602)

use new capability fields

CMakeLists.txt

@@ -825,6 +825,8 @@ set(ACLK_FILES
         aclk/schema-wrappers/alarm_config.h
         aclk/schema-wrappers/node_info.cc
         aclk/schema-wrappers/node_info.h
+        aclk/schema-wrappers/capability.cc
+        aclk/schema-wrappers/capability.h
         aclk/schema-wrappers/schema_wrappers.h
         aclk/schema-wrappers/schema_wrapper_utils.cc
         aclk/schema-wrappers/schema_wrapper_utils.h

Makefile.am

@@ -657,6 +657,8 @@ ACLK_FILES += \
     aclk/schema-wrappers/alarm_config.h \
     aclk/schema-wrappers/node_info.cc \
     aclk/schema-wrappers/node_info.h \
+    aclk/schema-wrappers/capability.cc \
+    aclk/schema-wrappers/capability.h \
     aclk/schema-wrappers/schema_wrappers.h \
     aclk/schema-wrappers/schema_wrapper_utils.cc \
     aclk/schema-wrappers/schema_wrapper_utils.h \

aclk/aclk-schemas

@@ -1 +1 @@
-Subproject commit b23f6a671ccf6d2766d6a208fc1e48b0fbf2fdad
+Subproject commit d8342ee6d932c152a78c9fe886281fe28170a6c4

aclk/aclk_tx_msgs.c

@@ -452,10 +452,21 @@ int aclk_send_app_layer_disconnect(mqtt_wss_client client, const char *message)
 uint16_t aclk_send_agent_connection_update(mqtt_wss_client client, int reachable) {
     size_t len;
     uint16_t pid;
+
+    struct capability agent_capabilities[] = {
+        { .name = "json",  .version = 2, .enabled = 0 },
+        { .name = "proto", .version = 1, .enabled = 1 },
+#ifdef ENABLE_ML
+        { .name = "ml",    .version = 1, .enabled = ml_enabled(localhost) },
+#endif
+        { .name = NULL,    .version = 0, .enabled = 0 }
+    };
+
     update_agent_connection_t conn = {
         .reachable = (reachable ? 1 : 0),
         .lwt = 0,
-        .session_id = aclk_session_newarch
+        .session_id = aclk_session_newarch,
+        .capabilities = agent_capabilities
     };
 
     rrdhost_aclk_state_lock(localhost);
@@ -490,7 +501,8 @@ char *aclk_generate_lwt(size_t *size) {
     update_agent_connection_t conn = {
         .reachable = 0,
         .lwt = 1,
-        .session_id = aclk_session_newarch
+        .session_id = aclk_session_newarch,
+        .capabilities = NULL
     };
 
     rrdhost_aclk_state_lock(localhost);

aclk/schema-wrappers/capability.cc

@@ -0,0 +1,11 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include "proto/aclk/v1/lib.pb.h"
+
+#include "capability.h"
+
+void capability_set(aclk_lib::v1::Capability *proto_capa, struct capability *c_capa) {
+    proto_capa->set_name(c_capa->name);
+    proto_capa->set_enabled(c_capa->enabled);
+    proto_capa->set_version(c_capa->version);
+}

aclk/schema-wrappers/capability.h

@@ -0,0 +1,24 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#ifndef ACLK_SCHEMA_CAPABILITY_H
+#define ACLK_SCHEMA_CAPABILITY_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+struct capability {
+    const char *name;
+    uint32_t version;
+    int enabled;
+};
+
+#ifdef __cplusplus
+}
+
+#include "proto/aclk/v1/lib.pb.h"
+
+void capability_set(aclk_lib::v1::Capability *proto_capa, struct capability *c_capa);
+#endif
+
+#endif /* ACLK_SCHEMA_CAPABILITY_H */

aclk/schema-wrappers/connection.cc

@@ -28,6 +28,15 @@ char *generate_update_agent_connection(size_t *len, const update_agent_connectio
     timestamp->set_seconds(tv.tv_sec);
     timestamp->set_nanos(tv.tv_usec * 1000);
 
+    if (data->capabilities) {
+        struct capability *capa = data->capabilities;
+        while (capa->name) {
+            aclk_lib::v1::Capability *proto_capa = connupd.add_capabilities();
+            capability_set(proto_capa, capa);
+            capa++;
+        }
+    }
+
     *len = PROTO_COMPAT_MSG_SIZE(connupd);
     char *msg = (char*)malloc(*len);
     if (msg)

aclk/schema-wrappers/connection.h

@@ -3,6 +3,8 @@
 #ifndef ACLK_SCHEMA_WRAPPER_CONNECTION_H
 #define ACLK_SCHEMA_WRAPPER_CONNECTION_H
 
+#include "capability.h"
+
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -15,6 +17,8 @@ typedef struct {
 
     unsigned int lwt:1;
 
+    struct capability *capabilities;
+
 // TODO in future optional fields
 // > 15 optional fields:
 // How long the system was running until connection (only applicable when reachable=true)

aclk/schema-wrappers/node_info.cc

@@ -94,6 +94,24 @@ char *generate_update_node_info_message(size_t *len, struct update_node_info *in
     ml_info->set_ml_capable(info->ml_info.ml_capable);
     ml_info->set_ml_enabled(info->ml_info.ml_enabled);
 
+    struct capability *capa;
+    if (info->node_capabilities) {
+        capa = info->node_capabilities;
+        while (capa->name) {
+            aclk_lib::v1::Capability *proto_capa = msg.mutable_node_info()->add_capabilities();
+            capability_set(proto_capa, capa);
+            capa++;
+        }
+    }
+    if (info->node_instance_capabilities) {
+        capa = info->node_instance_capabilities;
+        while (capa->name) {
+            aclk_lib::v1::Capability *proto_capa = msg.mutable_node_instance_info()->add_capabilities();
+            capability_set(proto_capa, capa);
+            capa++;
+        }
+    }
+
     *len = PROTO_COMPAT_MSG_SIZE(msg);
     char *bin = (char*)malloc(*len);
     if (bin)

aclk/schema-wrappers/node_info.h

@@ -6,6 +6,7 @@
 #include <stdlib.h>
 
 #include "database/rrd.h"
+#include "capability.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -67,6 +68,9 @@ struct update_node_info {
     int child;
 
     struct machine_learning_info ml_info;
+
+    struct capability *node_capabilities;
+    struct capability *node_instance_capabilities;
 };
 
 char *generate_update_node_info_message(size_t *len, struct update_node_info *info);

aclk/schema-wrappers/schema_wrappers.h

@@ -13,5 +13,6 @@
 #include "alarm_config.h"
 #include "alarm_stream.h"
 #include "node_info.h"
+#include "capability.h"
 
 #endif /* SCHEMA_WRAPPERS_H */

database/sqlite/sqlite_aclk_node.c

@@ -24,6 +24,14 @@ void sql_build_node_info(struct aclk_database_worker_config *wc, struct aclk_dat
     node_info.child = (wc->host != localhost);
     node_info.ml_info.ml_capable = ml_capable(localhost);
     node_info.ml_info.ml_enabled = ml_enabled(wc->host);
+
+    struct capability instance_caps[] = {
+        { .name = "proto", .version = 1,                     .enabled = 1 },
+        { .name = "ml",    .version = ml_capable(localhost), .enabled = ml_enabled(wc->host) },
+        { .name = NULL,    .version = 0,                     .enabled = 0 }
+    };
+    node_info.node_instance_capabilities = instance_caps;
+
     now_realtime_timeval(&node_info.updated_at);
 
     RRDHOST *host = wc->host;
@@ -55,6 +63,13 @@ void sql_build_node_info(struct aclk_database_worker_config *wc, struct aclk_dat
     node_info.data.services = NULL;   // char **
     node_info.data.service_count = 0;
     node_info.data.machine_guid = wc->host_guid;
+
+    struct capability node_caps[] = {
+        { .name = "ml", .version = host->system_info->ml_capable, .enabled = host->system_info->ml_enabled },
+        { .name = NULL, .version = 0, .enabled = 0 }
+    };
+    node_info.node_capabilities = node_caps;
+
     node_info.data.ml_info.ml_capable = host->system_info->ml_capable;
     node_info.data.ml_info.ml_enabled = host->system_info->ml_enabled;