Merge pull request #4116 from esl/add-status-to-cets-api

Add status to cets api

big_tests/tests/graphql_cets_SUITE.erl

@@ -3,7 +3,7 @@
 
 -compile([export_all, nowarn_export_all]).
 
--import(distributed_helper, [mim/0]).
+-import(distributed_helper, [mim/0, mim2/0, rpc/4]).
 -import(domain_helper, [host_type/1]).
 -import(mongooseimctl_helper, [rpc_call/3]).
 -import(graphql_helper, [execute_command/4, get_unauthorized/1, get_ok_value/2]).
@@ -14,21 +14,43 @@ all() ->
      {group, domain_admin_cets}].
 
 groups() ->
-    [{admin_cets_http, [sequence], admin_cets_tests()},
-     {admin_cets_cli, [sequence], admin_cets_tests()},
+    [{admin_cets_http, [parallel], admin_cets_tests()},
+     {admin_cets_cli, [parallel], admin_cets_tests()},
      {domain_admin_cets, [], domain_admin_tests()}].
 
 admin_cets_tests() ->
-    [has_sm_table_in_info].
+    [has_sm_table_in_info,
+     available_nodes,
+     unavailable_nodes,
+     joined_nodes,
+     discovered_nodes,
+     remote_nodes_without_disco,
+     remote_nodes_with_unknown_tables,
+     remote_unknown_tables,
+     remote_nodes_with_missing_tables,
+     remote_missing_tables,
+     conflict_nodes,
+     conflict_tables,
+     discovery_works].
 
 domain_admin_tests() ->
-    [domain_admin_get_info_test].
+    [domain_admin_get_table_info_test,
+     domain_admin_get_system_info_test].
 
 init_per_suite(Config) ->
-    Config1 = escalus:init_per_suite(Config),
-    ejabberd_node_utils:init(mim(), Config1).
+    case rpc_call(mongoose_config, get_opt, [[internal_databases, cets, backend], undefined]) of
+        rdbms ->
+            Config1 = escalus:init_per_suite(Config),
+            Config2 = ejabberd_node_utils:init(mim(), Config1),
+            add_bad_node(),
+            ok = rpc_call(cets_discovery, wait_for_ready, [mongoose_cets_discovery, 5000]),
+            Config2 ++ distributed_helper:require_rpc_nodes([mim, mim2]);
+        _ ->
+            {skip, "CETS is not configured with RDBMS"}
+    end.
 
 end_per_suite(Config) ->
+    ensure_bad_node_unregistered(),
     escalus:end_per_suite(Config).
 
 init_per_group(admin_cets_http, Config) ->
@@ -55,23 +77,163 @@ init_per_testcase(_, Config) ->
 % Admin tests
 
 has_sm_table_in_info(Config) ->
-    Res = get_info(Config),
-    Tables = get_ok_value([data, cets, systemInfo], Res),
+    Res = get_table_info(Config),
+    Tables = get_ok_value([data, cets, tableInfo], Res),
     [T] = [T || T = #{<<"tableName">> := <<"cets_session">>} <- Tables],
     #{<<"memory">> := Mem, <<"nodes">> := Nodes, <<"size">> := Size} = T,
-    true = is_integer(Mem),
-    true = is_integer(Size),
+    ?assert(is_integer(Mem), T),
+    ?assert(is_integer(Size), T),
     #{node := Node1} = mim(),
-    lists:member(Node1, Nodes).
+    assert_member(atom_to_binary(Node1), Nodes).
+
+available_nodes(Config) ->
+    #{node := Node1} = mim(),
+    #{node := Node2} = mim2(),
+    Res = get_system_info(Config),
+    Info = get_ok_value([data, cets, systemInfo], Res),
+    #{<<"availableNodes">> := Nodes} = Info,
+    assert_member(atom_to_binary(Node1), Nodes),
+    assert_member(atom_to_binary(Node2), Nodes),
+    assert_not_member(<<"badnode@localhost">>, Nodes).
+
+unavailable_nodes(Config) ->
+    #{node := Node1} = mim(),
+    #{node := Node2} = mim2(),
+    Res = get_system_info(Config),
+    Info = get_ok_value([data, cets, systemInfo], Res),
+    #{<<"unavailableNodes">> := Nodes} = Info,
+    assert_member(<<"badnode@localhost">>, Nodes),
+    assert_not_member(atom_to_binary(Node1), Nodes),
+    assert_not_member(atom_to_binary(Node2), Nodes).
+
+joined_nodes(Config) ->
+    #{node := Node1} = mim(),
+    #{node := Node2} = mim2(),
+    Res = get_system_info(Config),
+    Info = get_ok_value([data, cets, systemInfo], Res),
+    #{<<"joinedNodes">> := Nodes} = Info,
+    assert_member(atom_to_binary(Node1), Nodes),
+    assert_member(atom_to_binary(Node2), Nodes),
+    assert_not_member(<<"badnode@localhost">>, Nodes).
+
+remote_nodes_without_disco(Config) ->
+    Res = get_system_info(Config),
+    Info = get_ok_value([data, cets, systemInfo], Res),
+    ?assert(is_list(maps:get(<<"remoteNodesWithoutDisco">>, Info)), Info).
+
+remote_nodes_with_unknown_tables(Config) ->
+    Res = get_system_info(Config),
+    Info = get_ok_value([data, cets, systemInfo], Res),
+    ?assert(is_list(maps:get(<<"remoteNodesWithUnknownTables">>, Info)), Info).
+
+remote_unknown_tables(Config) ->
+    Res = get_system_info(Config),
+    Info = get_ok_value([data, cets, systemInfo], Res),
+    ?assert(is_list(maps:get(<<"remoteUnknownTables">>, Info)), Info).
+
+remote_nodes_with_missing_tables(Config) ->
+    Res = get_system_info(Config),
+    Info = get_ok_value([data, cets, systemInfo], Res),
+    ?assert(is_list(maps:get(<<"remoteNodesWithMissingTables">>, Info)), Info).
+
+remote_missing_tables(Config) ->
+    Res = get_system_info(Config),
+    Info = get_ok_value([data, cets, systemInfo], Res),
+    ?assert(is_list(maps:get(<<"remoteMissingTables">>, Info)), Info).
+
+conflict_nodes(Config) ->
+    Res = get_system_info(Config),
+    Info = get_ok_value([data, cets, systemInfo], Res),
+    ?assertMatch(#{<<"conflictNodes">> := []}, Info).
+
+conflict_tables(Config) ->
+    Res = get_system_info(Config),
+    Info = get_ok_value([data, cets, systemInfo], Res),
+    ?assertMatch(#{<<"conflictTables">> := []}, Info).
+
+conflict_nodes_count(Config) ->
+    Res = get_system_info(Config),
+    Info = get_ok_value([data, cets, systemInfo], Res),
+    ?assertMatch(#{<<"conflictNodesCount">> := 0}, Info).
+
+discovered_nodes(Config) ->
+    #{node := Node1} = mim(),
+    #{node := Node2} = mim2(),
+    Res = get_system_info(Config),
+    Info = get_ok_value([data, cets, systemInfo], Res),
+    #{<<"discoveredNodes">> := Nodes} = Info,
+    assert_member(atom_to_binary(Node1), Nodes),
+    assert_member(atom_to_binary(Node2), Nodes),
+    assert_member(<<"badnode@localhost">>, Nodes).
+
+discovered_nodes_count(Config) ->
+    Res = get_system_info(Config),
+    Info = get_ok_value([data, cets, systemInfo], Res),
+    #{<<"discoveredNodesCount">> := Count} = Info,
+    ?assert(is_integer(Count), Info),
+    ?assert(Count > 2, Info).
+
+discovery_works(Config) ->
+    Res = get_system_info(Config),
+    Info = get_ok_value([data, cets, systemInfo], Res),
+    ?assertMatch(#{<<"discoveryWorks">> := true}, Info).
 
 % Domain admin tests
 
-domain_admin_get_info_test(Config) ->
-    get_unauthorized(get_info(Config)).
+domain_admin_get_table_info_test(Config) ->
+    get_unauthorized(get_table_info(Config)).
+
+domain_admin_get_system_info_test(Config) ->
+    get_unauthorized(get_system_info(Config)).
 
 %--------------------------------------------------------------------------------------------------
 %                                         Helpers
 %--------------------------------------------------------------------------------------------------
 
-get_info(Config) ->
+get_table_info(Config) ->
+    execute_command(<<"cets">>, <<"tableInfo">>, #{}, Config).
+
+get_system_info(Config) ->
     execute_command(<<"cets">>, <<"systemInfo">>, #{}, Config).
+
+add_bad_node() ->
+    ensure_bad_node_unregistered(),
+    register_bad_node(),
+    force_check(),
+    wait_for_has_bad_node().
+
+register_bad_node() ->
+    ClusterName = <<"mim">>,
+    Node = <<"badnode@localhost">>,
+    Num = 100,
+    Timestamp = rpc(mim(), mongoose_rdbms_timestamp, select, []),
+    InsertArgs = [ClusterName, Node, Num, Timestamp],
+    {updated, 1} = rpc(mim(), mongoose_rdbms, execute, [global, cets_disco_insert_new, InsertArgs]).
+
+ensure_bad_node_unregistered() ->
+    ClusterName = <<"mim">>,
+    Node = <<"badnode@localhost">>,
+    DeleteArgs = [ClusterName, Node],
+    %% Ensure the node is removed
+    {updated, _} = rpc(mim(), mongoose_rdbms, execute, [global, cets_delete_node_from_db, DeleteArgs]).
+
+force_check() ->
+    Pid = rpc(mim(), erlang, whereis, [mongoose_cets_discovery]),
+    true = is_pid(Pid),
+    Pid ! check.
+
+has_bad_node() ->
+    #{unavailable_nodes := UnNodes} =
+        rpc(mim(), cets_discovery, system_info, [mongoose_cets_discovery]),
+    lists:member('badnode@localhost', UnNodes).
+
+wait_for_has_bad_node() ->
+    mongoose_helper:wait_until(fun() -> has_bad_node() end, true).
+
+assert_member(Elem, List) ->
+    lists:member(Elem, List)
+        orelse ct:fail({assert_member_failed, Elem, List}).
+
+assert_not_member(Elem, List) ->
+    lists:member(Elem, List)
+        andalso ct:fail({assert_member_failed, Elem, List}).

big_tests/tests/graphql_metric_SUITE.erl

@@ -35,6 +35,7 @@ metrics_tests() ->
      get_process_queue_length,
      get_inet_stats,
      get_vm_stats_memory,
+     get_cets_system,
      get_all_metrics_as_dicts,
      get_by_name_metrics_as_dicts,
      get_metrics_as_dicts_by_nonexistent_name,
@@ -78,6 +79,13 @@ init_per_group(domain_admin_metrics, Config) ->
 end_per_group(_GroupName, _Config) ->
     graphql_helper:clean().
 
+init_per_testcase(get_cets_system = CaseName, Config) ->
+     case is_cets_enabled() of
+         true ->
+             escalus:init_per_testcase(CaseName, Config);
+         false ->
+             {skip, cets_not_enabled}
+     end;
 init_per_testcase(CaseName, Config) ->
      escalus:init_per_testcase(CaseName, Config).
 
@@ -125,7 +133,14 @@ type_to_keys(<<"vm_system_info">>) ->
     [<<"ets_limit">>, <<"port_count">>, <<"port_limit">>,
      <<"process_count">>, <<"process_limit">>];
 type_to_keys(<<"probe_queues">>) ->
-    [<<"fsm">>, <<"regular">>, <<"total">>].
+    [<<"fsm">>, <<"regular">>, <<"total">>];
+type_to_keys(<<"cets_system">>) ->
+    [<<"available_nodes">>, <<"unavailable_nodes">>,
+    <<"remote_nodes_without_disco">>, <<"joined_nodes">>,
+    <<"remote_nodes_with_unknown_tables">>, <<"remote_unknown_tables">>,
+    <<"remote_nodes_with_missing_tables">>, <<"remote_missing_tables">>,
+    <<"conflict_nodes">>, <<"conflict_tables">>,
+    <<"discovered_nodes">>, <<"discovery_works">>].
 
 get_by_name_global_erlang_metrics(Config) ->
     %% Filter by name works
@@ -189,6 +204,12 @@ get_vm_stats_memory(Config) ->
     #{<<"type">> := <<"vm_stats_memory">>} = Mem,
     check_metric_by_type(Mem).
 
+get_cets_system(Config) ->
+    Result = get_metrics([<<"global">>, <<"cets">>, <<"system">>], Config),
+    ParsedResult = get_ok_value([data, metric, getMetrics], Result),
+    [#{<<"type">> := <<"cets_system">>} = Sys] = ParsedResult,
+    check_metric_by_type(Sys).
+
 get_all_metrics_as_dicts(Config) ->
     Result = get_metrics_as_dicts(Config),
     ParsedResult = get_ok_value([data, metric, getMetricsAsDicts], Result),
@@ -454,3 +475,11 @@ values_are_integers(Map, Keys) ->
 
 metric_host_type() ->
     binary:replace(domain_helper:host_type(), <<" ">>, <<"_">>, [global]).
+
+is_cets_enabled() ->
+    case rpc(mim(), mongoose_config, lookup_opt, [[internal_databases, cets]]) of
+        {ok, _} ->
+            true;
+        _ ->
+            false
+    end.

big_tests/tests/graphql_muc_SUITE.erl

@@ -227,7 +227,7 @@ init_per_suite(Config) ->
 end_per_suite(Config) ->
     escalus_fresh:clean(),
     mongoose_helper:ensure_muc_clean(),
-    muc_helper:unload_muc(),
+    ensure_muc_stopped(),
     dynamic_modules:restore_modules(Config),
     escalus:end_per_suite(Config).
 

big_tests/tests/mongooseimctl_helper.erl

@@ -59,4 +59,6 @@ loop(Cmd, Args, Port, Data, Timeout) ->
 log_command(Cmd, Args, Data, ExitStatus) ->
     Pattern = lists:flatten(lists:duplicate(length(Args), " ~s")),
     ct:log("Execute ~s " ++ Pattern ++ "~nResult ~p~nExitStatus ~p",
-           [Cmd] ++ Args  ++ [Data, ExitStatus]).
+           [Cmd] ++ Args  ++ [Data, ExitStatus]),
+    %% For easy read of multiline outputs:
+    ct:log("ResultString:~n~ts", [Data]).

priv/graphql/schemas/admin/cets.gql

@@ -1,11 +1,14 @@
 "Allow admin to get information about CETS status"
 type CETSAdminQuery @protected{
-    "Get from the local node. Only for global admin"
-    systemInfo: [CETSInfo]
+    "Get a list of tables from the local node"
+    tableInfo: [CETSTableInfo]
+      @protected(type: GLOBAL)
+    "Get status of CETS"
+    systemInfo: CETSSystemInfo
       @protected(type: GLOBAL)
 }
 
-type CETSInfo {
+type CETSTableInfo {
   "ETS table name"
   tableName: String
   "Memory (in words)"
@@ -15,3 +18,30 @@ type CETSInfo {
   "A list of clustered nodes"
   nodes: [String]
 }
+
+type CETSSystemInfo {
+  "Available nodes (nodes that are connected to us and have the CETS disco process started)"
+  availableNodes: [String]
+  "Unavailable nodes (nodes that do not respond to our pings)"
+  unavailableNodes: [String]
+  "Joined nodes (nodes that have our local tables running)"
+  joinedNodes: [String]
+  "Discovered nodes (nodes that are extracted from the discovery backend)."
+  discoveredNodes: [String]
+  "Nodes with stopped CETS discovery"
+  remoteNodesWithoutDisco: [String]
+  "Nodes that have more tables registered than the local node"
+  remoteNodesWithUnknownTables: [String]
+  "Unknown remote tables"
+  remoteUnknownTables: [String]
+  "Nodes that are available, but do not host some of our local tables"
+  remoteNodesWithMissingTables: [String]
+  "Missing remote tables"
+  remoteMissingTables: [String]
+  "Nodes that replicate at least one of our local tables to a different list of nodes"
+  conflictNodes: [String]
+  "Tables that have conflicting replication destinations"
+  conflictTables: [String]
+  "Returns true if the last discovery attempt is successful"
+  discoveryWorks: Boolean
+}