IGNITE-28714 Add data center ID to client connections system view

docs/_docs/clustering/multi-data-center.adoc

@@ -249,6 +249,10 @@ This behavior applies to both full and historical rebalance.
 
 == Monitoring
 
-The data center ID is exposed in the `NODES` system view as part of node information.
-For client nodes, the `io.discovery.ClientRouterNodeId` metric exposes the ID of the server node used as
-the discovery router.
+Use the `NODES` system view to check the data center IDs of cluster nodes, and the `CLIENT_CONNECTIONS` view to check
+the data center IDs of thin-client connections. This makes it easy to spot clients with a wrong or missing ID.
+
+For Java thin clients, the `IGNITE_DATA_CENTER_ID` system property takes precedence over the user attribute of the same
+name.
+
+For a client node, the `io.discovery.ClientRouterNodeId` metric identifies the server node used as the discovery router.

docs/_docs/monitoring-metrics/system-views.adoc

@@ -439,6 +439,7 @@ This view exposes information about currently opened client connections: JDBC, O
 |TYPE | string |  Type of the connection
 |USER | string |  User name
 |VERSION | string |  Protocol version
+|DATA_CENTER_ID | string | Effective data center ID reported by the client
 |===
 
 == CLIENT_CONNECTION_ATTRIBUTES

modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinMetadataSelfTest.java

@@ -950,6 +950,7 @@ public void testGetAllColumns() throws Exception {
                 "SYS.CLIENT_CONNECTIONS.TYPE.null",
                 "SYS.CLIENT_CONNECTIONS.USER.null",
                 "SYS.CLIENT_CONNECTIONS.VERSION.null",
+                "SYS.CLIENT_CONNECTIONS.DATA_CENTER_ID.null",
                 "SYS.CLIENT_CONNECTION_ATTRIBUTES.CONNECTION_ID.null",
                 "SYS.CLIENT_CONNECTION_ATTRIBUTES.NAME.null",
                 "SYS.CLIENT_CONNECTION_ATTRIBUTES.VALUE.null",

modules/control-utility/src/test/java/org/apache/ignite/util/SystemViewCommandTest.java

@@ -601,7 +601,9 @@ public void testViewColumns() {
                 InetSocketAddress.class.getName()),
             asList("TYPE", "CLIENT_CONNECTIONS", SCHEMA_SYS, "null", "true", "-1", "-1", String.class.getName()),
             asList("USER", "CLIENT_CONNECTIONS", SCHEMA_SYS, "null", "true", "-1", "-1", String.class.getName()),
-            asList("VERSION", "CLIENT_CONNECTIONS", SCHEMA_SYS, "null", "true", "-1", "-1", String.class.getName())
+            asList("VERSION", "CLIENT_CONNECTIONS", SCHEMA_SYS, "null", "true", "-1", "-1", String.class.getName()),
+            asList("DATA_CENTER_ID", "CLIENT_CONNECTIONS", SCHEMA_SYS, "null", "true", "-1", "-1",
+                String.class.getName())
         ));
 
         Set<List<String>> sqlViewColumnsView = systemView(ignite0, SQL_VIEW_COLS_VIEW).stream()

modules/core/src/main/java/org/apache/ignite/spi/systemview/view/ClientConnectionView.java

@@ -25,10 +25,12 @@
 import org.apache.ignite.internal.processors.odbc.odbc.OdbcConnectionContext;
 import org.apache.ignite.internal.processors.platform.client.ClientConnectionContext;
 import org.apache.ignite.internal.processors.security.SecurityContext;
+import org.apache.ignite.internal.systemview.Order;
 import org.apache.ignite.internal.systemview.SystemViewDescriptor;
 import org.apache.ignite.internal.util.nio.GridNioSession;
 import org.jetbrains.annotations.Nullable;
 
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_DATA_CENTER_ID;
 import static org.apache.ignite.internal.processors.odbc.ClientListenerNioListener.CONN_CTX_META_KEY;
 
 /**
@@ -49,6 +51,7 @@ public ClientConnectionView(GridNioSession ses) {
     }
 
     /** @return Connection id. */
+    @Order(0)
     public long connectionId() {
         if (ctx == null)
             return -1;
@@ -57,6 +60,7 @@ public long connectionId() {
     }
 
     /** @return Connection type. */
+    @Order(3)
     public String type() {
         if (ctx == null)
             return null;
@@ -72,16 +76,19 @@ else if (ctx instanceof ClientConnectionContext)
     }
 
     /** @return Connection local address. */
+    @Order(1)
     public InetSocketAddress localAddress() {
         return ses.localAddress();
     }
 
     /** @return Connection remote address. */
+    @Order(2)
     public InetSocketAddress remoteAddress() {
         return ses.remoteAddress();
     }
 
     /** @return User name. */
+    @Order(4)
     public String user() {
         SecurityContext secCtx = ctx == null ? null : ctx.securityContext();
 
@@ -92,6 +99,7 @@ public String user() {
     }
 
     /** @return Protocol version. */
+    @Order(5)
     public String version() {
         if (ctx == null)
             return null;
@@ -105,4 +113,10 @@ public String version() {
 
         return ver == null ? null : ver.asString();
     }
+
+    /** @return Client data center ID. */
+    @Order(6)
+    @Nullable public String dataCenterId() {
+        return ctx == null ? null : ctx.attributes().get(IGNITE_DATA_CENTER_ID);
+    }
 }

modules/core/src/test/java/org/apache/ignite/internal/metric/SystemViewClientTest.java

@@ -18,7 +18,9 @@
 package org.apache.ignite.internal.metric;
 
 import java.sql.Connection;
+import java.util.Collections;
 import java.util.Iterator;
+import java.util.Map;
 import java.util.Properties;
 import org.apache.ignite.IgniteJdbcThinDriver;
 import org.apache.ignite.Ignition;
@@ -35,8 +37,10 @@
 import org.apache.ignite.spi.systemview.view.FiltrableSystemView;
 import org.apache.ignite.spi.systemview.view.SystemView;
 import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.WithSystemProperty;
 import org.junit.Test;
 
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_DATA_CENTER_ID;
 import static org.apache.ignite.internal.processors.odbc.ClientListenerProcessor.CLI_CONN_ATTR_VIEW;
 import static org.apache.ignite.internal.processors.odbc.ClientListenerProcessor.CLI_CONN_VIEW;
 import static org.apache.ignite.internal.util.lang.GridFunc.identity;
@@ -88,6 +92,43 @@ public void testClientsConnections() throws Exception {
         }
     }
 
+    /** */
+    @Test
+    @WithSystemProperty(key = IGNITE_DATA_CENTER_ID, value = "server-dc")
+    public void testClientConnectionDataCenterId() throws Exception {
+        try (IgniteEx g0 = startGrid(0)) {
+            assertEquals("server-dc", g0.localNode().dataCenterId());
+
+            System.clearProperty(IGNITE_DATA_CENTER_ID);
+
+            SystemView<ClientConnectionView> conns = g0.context().systemView().view(CLI_CONN_VIEW);
+
+            Map<String, String> userAttrs = Collections.singletonMap(IGNITE_DATA_CENTER_ID, "user-dc");
+
+            try (IgniteClient ignored = Ignition.startClient(
+                new ClientConfiguration()
+                    .setAddresses(Config.SERVER)
+                    .setUserAttributes(userAttrs))
+            ) {
+                assertEquals("user-dc", conns.iterator().next().dataCenterId());
+            }
+
+            assertTrue(GridTestUtils.waitForCondition(() -> conns.size() == 0, 5_000));
+
+            System.setProperty(IGNITE_DATA_CENTER_ID, "property-dc");
+
+            try (
+                IgniteClient ignored = Ignition.startClient(new ClientConfiguration()
+                    .setAddresses(Config.SERVER)
+                    .setUserAttributes(userAttrs))
+            ) {
+
+                assertEquals("property-dc", conns.iterator().next().dataCenterId());
+                assertEquals("user-dc", userAttrs.get(IGNITE_DATA_CENTER_ID));
+            }
+        }
+    }
+
     /** */
     @Test
     public void testClientConnectionAttributes() throws Exception {

modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/metric/SqlViewExporterSpiTest.java

@@ -86,6 +86,7 @@
 import org.junit.Test;
 
 import static java.util.Arrays.asList;
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_DATA_CENTER_ID;
 import static org.apache.ignite.internal.metric.SystemViewQueriesTest.TEST_PREDICATE;
 import static org.apache.ignite.internal.metric.SystemViewQueriesTest.TEST_TRANSFORMER;
 import static org.apache.ignite.internal.processors.cache.GridCacheUtils.cacheGroupId;
@@ -323,11 +324,16 @@ public void testClientsConnections() throws Exception {
 
         int port = ignite0.configuration().getClientConnectorConfiguration().getPort();
 
-        try (IgniteClient client = Ignition.startClient(new ClientConfiguration().setAddresses(host + ":" + port))) {
+        String dcId = "dc0";
+
+        try (IgniteClient client = Ignition.startClient(new ClientConfiguration().setAddresses(host + ":" + port)
+            .setUserAttributes(F.asMap(IGNITE_DATA_CENTER_ID, dcId)))) {
             try (Connection conn = new IgniteJdbcThinDriver().connect("jdbc:ignite:thin://" + host, new Properties())) {
-                List<List<?>> conns = execute(ignite0, "SELECT * FROM SYS.CLIENT_CONNECTIONS");
+                List<List<?>> conns = execute(ignite0, "SELECT TYPE, DATA_CENTER_ID FROM SYS.CLIENT_CONNECTIONS");
 
                 assertEquals(2, conns.size());
+                assertTrue(conns.stream().anyMatch(row -> "THIN".equals(row.get(0)) && dcId.equals(row.get(1))));
+                assertTrue(conns.stream().anyMatch(row -> "JDBC".equals(row.get(0)) && row.get(1) == null));
             }
         }
     }
@@ -569,7 +575,8 @@ public void testViewColumns() {
             asList("REMOTE_ADDRESS", "CLIENT_CONNECTIONS", SCHEMA_SYS, null, true, -1, -1, InetSocketAddress.class.getName()),
             asList("TYPE", "CLIENT_CONNECTIONS", SCHEMA_SYS, null, true, -1, -1, String.class.getName()),
             asList("USER", "CLIENT_CONNECTIONS", SCHEMA_SYS, null, true, -1, -1, String.class.getName()),
-            asList("VERSION", "CLIENT_CONNECTIONS", SCHEMA_SYS, null, true, -1, -1, String.class.getName())
+            asList("VERSION", "CLIENT_CONNECTIONS", SCHEMA_SYS, null, true, -1, -1, String.class.getName()),
+            asList("DATA_CENTER_ID", "CLIENT_CONNECTIONS", SCHEMA_SYS, null, true, -1, -1, String.class.getName())
         );
 
         List<List<?>> res = execute(ignite0, "SELECT * FROM SYS.VIEW_COLUMNS WHERE VIEW_NAME = 'CLIENT_CONNECTIONS'");

modules/thin-client/impl/src/main/java/org/apache/ignite/internal/client/thin/ClientChannelConfiguration.java

@@ -24,7 +24,6 @@
 import java.util.concurrent.Executor;
 import javax.cache.configuration.Factory;
 import javax.net.ssl.SSLContext;
-
 import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.client.SslMode;
 import org.apache.ignite.client.SslProtocol;
@@ -123,7 +122,11 @@ final class ClientChannelConfiguration {
      * Constructor.
      */
     @SuppressWarnings("UnnecessaryThis")
-    ClientChannelConfiguration(ClientConfiguration cfg, List<InetSocketAddress> addrs) {
+    ClientChannelConfiguration(
+        ClientConfiguration cfg,
+        List<InetSocketAddress> addrs,
+        Map<String, String> userAttrs
+    ) {
         this.sslMode = cfg.getSslMode();
         this.tcpNoDelay = cfg.isTcpNoDelay();
         this.handshakeTimeout = cfg.getHandshakeTimeout();
@@ -145,7 +148,7 @@ final class ClientChannelConfiguration {
         this.reconnectThrottlingPeriod = cfg.getReconnectThrottlingPeriod();
         this.reconnectThrottlingRetries = cfg.getReconnectThrottlingRetries();
         this.addrs = Collections.unmodifiableList(addrs);
-        this.userAttrs = cfg.getUserAttributes();
+        this.userAttrs = userAttrs;
         this.asyncContinuationExecutor = cfg.getAsyncContinuationExecutor();
         this.heartbeatEnabled = cfg.isHeartbeatEnabled();
         this.heartbeatInterval = cfg.getHeartbeatInterval();

modules/thin-client/impl/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannelImpl.java

@@ -20,6 +20,7 @@
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
@@ -88,6 +89,9 @@ final class ReliableChannelImpl implements ReliableChannelEx {
     /** Client configuration. */
     private final ClientConfiguration clientCfg;
 
+    /** Effective user attributes. */
+    private final Map<String, String> effectiveUserAttrs;
+
     /** Logger. */
     private final IgniteLogger log;
 
@@ -141,10 +145,14 @@ final class ReliableChannelImpl implements ReliableChannelEx {
 
         partitionAwarenessEnabled = clientCfg.isPartitionAwarenessEnabled();
 
+        Map<String, String> cfgUserAttrs = clientCfg.getUserAttributes();
+
         String dcId = IgniteCommonsSystemProperties.getString(IgniteCommonsSystemProperties.IGNITE_DATA_CENTER_ID);
 
-        if (dcId == null && !F.isEmpty(clientCfg.getUserAttributes()))
-            dcId = clientCfg.getUserAttributes().get(IgniteCommonsSystemProperties.IGNITE_DATA_CENTER_ID);
+        if (dcId == null && !F.isEmpty(cfgUserAttrs))
+            dcId = cfgUserAttrs.get(IgniteCommonsSystemProperties.IGNITE_DATA_CENTER_ID);
+
+        effectiveUserAttrs = effectiveUserAttributes(cfgUserAttrs, dcId);
 
         affinityCtx = new ClientCacheAffinityContext(
             binary,
@@ -701,14 +709,14 @@ synchronized void initChannelHolders() {
 
                 if (hld != null) {
                     if (!hld.getAddresses().equals(addrs)) // Enrich holder addresses.
-                        hld.setConfiguration(new ClientChannelConfiguration(clientCfg, addrs));
+                        hld.setConfiguration(new ClientChannelConfiguration(clientCfg, addrs, effectiveUserAttrs));
 
                     break;
                 }
             }
 
             if (hld == null) { // If not found, create the new one.
-                hld = new ClientChannelHolder(new ClientChannelConfiguration(clientCfg, addrs));
+                hld = new ClientChannelHolder(new ClientChannelConfiguration(clientCfg, addrs, effectiveUserAttrs));
 
                 reinitHolders.add(hld);
 
@@ -749,6 +757,27 @@ synchronized void initChannelHolders() {
         finishChannelsReInit = System.currentTimeMillis();
     }
 
+    /**
+     * Adds the effective data center ID to user attributes.
+     *
+     * @param cfgUserAttrs Configured user attributes.
+     * @param dcId Effective data center ID.
+     * @return Effective user attributes.
+     */
+    private static Map<String, String> effectiveUserAttributes(
+        @Nullable Map<String, String> cfgUserAttrs,
+        @Nullable String dcId
+    ) {
+        if (dcId == null)
+            return cfgUserAttrs;
+
+        Map<String, String> attrs = cfgUserAttrs == null ? new HashMap<>() : new HashMap<>(cfgUserAttrs);
+
+        attrs.put(IgniteCommonsSystemProperties.IGNITE_DATA_CENTER_ID, dcId);
+
+        return Collections.unmodifiableMap(attrs);
+    }
+
     /**
      * Establishing connections to servers. If partition awareness feature is enabled connections are created
      * for every configured server. Otherwise only default channel is connected.