From 99b1b73967c81a5399be98262250dc9415d1bca9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BA=8C=E4=B8=AB=E8=AE=B2=E6=A2=B5?= Date: Fri, 31 Jul 2026 01:06:02 +0800 Subject: [PATCH] feat(etcd): add cluster operations and access control --- .../java/com/dbx/agent/AgentProtocol.java | 79 +- .../dbx/agent/MultiSessionJsonRpcServer.java | 92 +- .../java/com/dbx/agent/SessionRpcHandler.java | 23 + .../src/main/resources/agent-protocol-v1.json | 4 +- .../src/main/resources/agent-protocol-v2.json | 4 +- .../agent/CommonJavaCompatibilityTest.java | 19 + .../java/com/dbx/agent/etcd/EtcdAgent.java | 888 ++++++++++++++++- .../com/dbx/agent/etcd/EtcdAgentTest.java | 150 ++- .../src/components/etcd/EtcdAccessControl.vue | 768 +++++++++++++++ .../src/components/etcd/EtcdAdminConsole.vue | 915 ++++++++++++++++++ .../src/components/etcd/EtcdDashboard.vue | 681 ++++++------- .../src/components/etcd/EtcdKeyBrowser.vue | 869 +++++++++++++---- .../etcd/__tests__/EtcdKeyBrowser.spec.ts | 49 +- .../src/components/kv/KvKeyBrowser.vue | 200 +++- .../kv/__tests__/KvKeyBrowserExport.spec.ts | 10 +- .../src/components/layout/AppTabBar.vue | 5 +- .../src/components/layout/ContentArea.vue | 7 + .../src/components/sidebar/ConnectionTree.vue | 2 +- .../sidebar/SidebarTreeRuntimeHost.vue | 6 +- .../src/components/sidebar/TreeItem.vue | 3 + apps/desktop/src/i18n/locales/en.ts | 204 ++++ apps/desktop/src/i18n/locales/es.ts | 204 ++++ apps/desktop/src/i18n/locales/it.ts | 204 ++++ apps/desktop/src/i18n/locales/ja.ts | 204 ++++ apps/desktop/src/i18n/locales/pt-BR.ts | 204 ++++ apps/desktop/src/i18n/locales/zh-CN.ts | 204 ++++ apps/desktop/src/i18n/locales/zh-TW.ts | 204 ++++ .../lib/__tests__/etcd/etcdKeyTree.spec.ts | 12 +- .../lib/__tests__/etcd/watchLifecycle.spec.ts | 76 ++ .../src/lib/__tests__/kv/kvKeyTree.spec.ts | 18 +- .../sidebar/sidebarTreeItemLayout.spec.ts | 3 +- apps/desktop/src/lib/backend/api.ts | 23 + apps/desktop/src/lib/backend/http.ts | 32 + apps/desktop/src/lib/backend/tauri.ts | 96 ++ .../__tests__/connectionHealth.spec.ts | 6 + .../src/lib/connection/connectionHealth.ts | 4 + apps/desktop/src/lib/etcd/watchLifecycle.ts | 19 + apps/desktop/src/lib/kv/kvKeyTree.ts | 51 +- .../src/lib/sidebar/sidebarActiveTabTarget.ts | 12 + .../src/lib/sidebar/sidebarTreeItemLayout.ts | 1 + apps/desktop/src/lib/sidebar/treeNodeClick.ts | 2 +- apps/desktop/src/lib/sidebar/treeNodeIcon.ts | 4 +- apps/desktop/src/lib/tabs/tabPresentation.ts | 5 + apps/desktop/src/stores/connectionStore.ts | 11 +- apps/desktop/src/types/database.ts | 2 + crates/dbx-core/assets/agent-protocol-v1.json | 4 +- crates/dbx-core/assets/agent-protocol-v2.json | 4 +- crates/dbx-core/src/agent_kv.rs | 855 ++++++++++++++-- crates/dbx-core/src/connection.rs | 141 ++- crates/dbx-core/src/db/agent_driver.rs | 123 ++- crates/dbx-web/src/main.rs | 9 + crates/dbx-web/src/routes/etcd.rs | 254 +++++ examples/etcd/dbx-etcd-test-bundle.json | 84 ++ src-tauri/src/commands/etcd_cmd.rs | 190 +++- src-tauri/src/lib.rs | 9 + 55 files changed, 7502 insertions(+), 750 deletions(-) create mode 100644 agents/common/src/main/java/com/dbx/agent/SessionRpcHandler.java create mode 100644 apps/desktop/src/components/etcd/EtcdAccessControl.vue create mode 100644 apps/desktop/src/components/etcd/EtcdAdminConsole.vue create mode 100644 apps/desktop/src/lib/__tests__/etcd/watchLifecycle.spec.ts create mode 100644 apps/desktop/src/lib/etcd/watchLifecycle.ts create mode 100644 examples/etcd/dbx-etcd-test-bundle.json diff --git a/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java b/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java index 868050d89..3323881d3 100644 --- a/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java +++ b/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java @@ -70,6 +70,29 @@ public final class AgentProtocol { public static final String KV_METHOD_RENAME = "kv_rename"; public static final String KV_METHOD_HISTORY = "kv_history"; public static final String KV_METHOD_STATUS = "kv_status"; + public static final String ETCD_METHOD_COMPACT = "etcd_compact"; + public static final String ETCD_METHOD_DEFRAG = "etcd_defrag"; + public static final String ETCD_METHOD_WATCH_START = "etcd_watch_start"; + public static final String ETCD_METHOD_WATCH_POLL = "etcd_watch_poll"; + public static final String ETCD_METHOD_WATCH_STOP = "etcd_watch_stop"; + public static final String ETCD_METHOD_LEASE_LIST = "etcd_lease_list"; + public static final String ETCD_METHOD_LEASE_GET = "etcd_lease_get"; + public static final String ETCD_METHOD_LEASE_GRANT = "etcd_lease_grant"; + public static final String ETCD_METHOD_LEASE_KEEPALIVE = "etcd_lease_keepalive_once"; + public static final String ETCD_METHOD_LEASE_REVOKE = "etcd_lease_revoke"; + public static final String ETCD_METHOD_AUTH_USER_LIST = "etcd_auth_user_list"; + public static final String ETCD_METHOD_AUTH_USER_GET = "etcd_auth_user_get"; + public static final String ETCD_METHOD_AUTH_USER_ADD = "etcd_auth_user_add"; + public static final String ETCD_METHOD_AUTH_USER_DELETE = "etcd_auth_user_delete"; + public static final String ETCD_METHOD_AUTH_USER_CHANGE_PASSWORD = "etcd_auth_user_change_password"; + public static final String ETCD_METHOD_AUTH_USER_GRANT_ROLE = "etcd_auth_user_grant_role"; + public static final String ETCD_METHOD_AUTH_USER_REVOKE_ROLE = "etcd_auth_user_revoke_role"; + public static final String ETCD_METHOD_AUTH_ROLE_LIST = "etcd_auth_role_list"; + public static final String ETCD_METHOD_AUTH_ROLE_GET = "etcd_auth_role_get"; + public static final String ETCD_METHOD_AUTH_ROLE_ADD = "etcd_auth_role_add"; + public static final String ETCD_METHOD_AUTH_ROLE_DELETE = "etcd_auth_role_delete"; + public static final String ETCD_METHOD_AUTH_ROLE_GRANT_PERMISSION = "etcd_auth_role_grant_permission"; + public static final String ETCD_METHOD_AUTH_ROLE_REVOKE_PERMISSION = "etcd_auth_role_revoke_permission"; public static final String CAPABILITY_CONNECT = "connect"; public static final String CAPABILITY_TEST_CONNECTION = "test_connection"; @@ -84,6 +107,11 @@ public final class AgentProtocol { public static final String CAPABILITY_KV_LIST_VALUES = "kv_list_values"; public static final String CAPABILITY_KV_STATUS = "kv_status"; public static final String CAPABILITY_KV_HISTORY = "kv_history"; + public static final String CAPABILITY_ETCD_COMPACTION = "etcd_compaction"; + public static final String CAPABILITY_ETCD_DEFRAG = "etcd_defrag"; + public static final String CAPABILITY_ETCD_WATCH = "etcd_watch"; + public static final String CAPABILITY_ETCD_LEASE = "etcd_lease"; + public static final String CAPABILITY_ETCD_AUTH = "etcd_auth"; public static final String CAPABILITY_MULTI_SESSION = "multi_session"; public static final List CAPABILITIES = Collections.unmodifiableList(Arrays.asList( @@ -109,9 +137,17 @@ public final class AgentProtocol { CAPABILITY_KV_CAS, CAPABILITY_KV_LIST_VALUES, CAPABILITY_KV_STATUS, - CAPABILITY_KV_HISTORY + CAPABILITY_KV_HISTORY, + CAPABILITY_ETCD_COMPACTION, + CAPABILITY_ETCD_DEFRAG, + CAPABILITY_ETCD_WATCH, + CAPABILITY_ETCD_LEASE, + CAPABILITY_ETCD_AUTH )); + public static final List MULTI_SESSION_CAPABILITIES; + public static final List MULTI_SESSION_ALL_CAPABILITIES; + public static final List COMMON_METHODS = Collections.unmodifiableList(Arrays.asList( METHOD_HANDSHAKE, METHOD_CONNECT, @@ -150,6 +186,14 @@ public final class AgentProtocol { public static final List MULTI_SESSION_METHODS; static { + List capabilities = new java.util.ArrayList<>(CAPABILITIES); + capabilities.add(CAPABILITY_MULTI_SESSION); + MULTI_SESSION_CAPABILITIES = Collections.unmodifiableList(capabilities); + + List allCapabilities = new java.util.ArrayList<>(ALL_CAPABILITIES); + allCapabilities.add(CAPABILITY_MULTI_SESSION); + MULTI_SESSION_ALL_CAPABILITIES = Collections.unmodifiableList(allCapabilities); + List methods = new java.util.ArrayList<>(COMMON_METHODS); int insertAt = methods.indexOf(METHOD_CONNECT) + 1; methods.addAll(insertAt, Arrays.asList( @@ -185,7 +229,30 @@ public final class AgentProtocol { KV_METHOD_DELETE, KV_METHOD_RENAME, KV_METHOD_HISTORY, - KV_METHOD_STATUS + KV_METHOD_STATUS, + ETCD_METHOD_COMPACT, + ETCD_METHOD_DEFRAG, + ETCD_METHOD_WATCH_START, + ETCD_METHOD_WATCH_POLL, + ETCD_METHOD_WATCH_STOP, + ETCD_METHOD_LEASE_LIST, + ETCD_METHOD_LEASE_GET, + ETCD_METHOD_LEASE_GRANT, + ETCD_METHOD_LEASE_KEEPALIVE, + ETCD_METHOD_LEASE_REVOKE, + ETCD_METHOD_AUTH_USER_LIST, + ETCD_METHOD_AUTH_USER_GET, + ETCD_METHOD_AUTH_USER_ADD, + ETCD_METHOD_AUTH_USER_DELETE, + ETCD_METHOD_AUTH_USER_CHANGE_PASSWORD, + ETCD_METHOD_AUTH_USER_GRANT_ROLE, + ETCD_METHOD_AUTH_USER_REVOKE_ROLE, + ETCD_METHOD_AUTH_ROLE_LIST, + ETCD_METHOD_AUTH_ROLE_GET, + ETCD_METHOD_AUTH_ROLE_ADD, + ETCD_METHOD_AUTH_ROLE_DELETE, + ETCD_METHOD_AUTH_ROLE_GRANT_PERMISSION, + ETCD_METHOD_AUTH_ROLE_REVOKE_PERMISSION )); private AgentProtocol() { @@ -196,9 +263,11 @@ public final class AgentProtocol { } public static HandshakeResult multiSessionHandshakeResult() { - List capabilities = new java.util.ArrayList<>(CAPABILITIES); - capabilities.add(CAPABILITY_MULTI_SESSION); - return new HandshakeResult(MULTI_SESSION_PROTOCOL_VERSION, MULTI_SESSION_PROTOCOL_VERSION, capabilities); + return new HandshakeResult( + MULTI_SESSION_PROTOCOL_VERSION, + MULTI_SESSION_PROTOCOL_VERSION, + MULTI_SESSION_CAPABILITIES + ); } public static final class HandshakeResult { diff --git a/agents/common/src/main/java/com/dbx/agent/MultiSessionJsonRpcServer.java b/agents/common/src/main/java/com/dbx/agent/MultiSessionJsonRpcServer.java index f1155ed8b..a8f236096 100644 --- a/agents/common/src/main/java/com/dbx/agent/MultiSessionJsonRpcServer.java +++ b/agents/common/src/main/java/com/dbx/agent/MultiSessionJsonRpcServer.java @@ -26,6 +26,7 @@ public final class MultiSessionJsonRpcServer implements AutoCloseable { private static final long MAINTENANCE_INTERVAL_MILLIS = 60_000L; private final Supplier agentFactory; + private final Supplier sessionHandlerFactory; private final Map sessions = new ConcurrentHashMap<>(); private final ExecutorService requests = Executors.newCachedThreadPool(); private final JdbcConnectionPoolRegistry poolRegistry; @@ -52,9 +53,21 @@ public final class MultiSessionJsonRpcServer implements AutoCloseable { JdbcConnectionPoolRegistry poolRegistry ) { this.agentFactory = agentFactory; + this.sessionHandlerFactory = null; this.poolRegistry = poolRegistry; } + private MultiSessionJsonRpcServer(Supplier sessionHandlerFactory, boolean customHandler) { + this.agentFactory = null; + this.sessionHandlerFactory = sessionHandlerFactory; + this.poolRegistry = new JdbcConnectionPoolRegistry(); + } + + /** Creates a protocol v2 server for a non-JDBC, session-scoped agent. */ + public static MultiSessionJsonRpcServer forSessionHandlers(Supplier sessionHandlerFactory) { + return new MultiSessionJsonRpcServer(sessionHandlerFactory, true); + } + public void run() { synchronized (outputLock) { protocolOutput.println("{\"ready\":true}"); @@ -94,7 +107,7 @@ public final class MultiSessionJsonRpcServer implements AutoCloseable { try { Object result; if (AgentProtocol.METHOD_HANDSHAKE.equals(method)) { - result = AgentProtocol.multiSessionHandshakeResult(); + result = sessionHandlerFactory == null ? AgentProtocol.multiSessionHandshakeResult() : customHandshake(); } else if (AgentProtocol.METHOD_OPEN_SESSION.equals(method)) { result = openSession(requiredSessionId(params), params); } else if (AgentProtocol.METHOD_CLOSE_SESSION.equals(method)) { @@ -105,7 +118,9 @@ public final class MultiSessionJsonRpcServer implements AutoCloseable { session(requiredSessionId(params)).cancel(); result = Collections.singletonMap("ok", true); } else if (AgentProtocol.METHOD_TEST_CONNECTION.equals(method)) { - result = new JsonRpcServer(agentFactory.get()).dispatchForRuntime(method, params); + result = sessionHandlerFactory == null + ? new JsonRpcServer(agentFactory.get()).dispatchForRuntime(method, params) + : testSession(params); } else if (AgentProtocol.METHOD_CONNECT.equals(method)) { closeSession(LEGACY_SESSION_ID); result = openSession(LEGACY_SESSION_ID, params); @@ -131,18 +146,23 @@ public final class MultiSessionJsonRpcServer implements AutoCloseable { if (sessions.size() >= MAX_SESSIONS && !sessions.containsKey(sessionId)) { throw new IllegalStateException("Agent session limit reached: " + MAX_SESSIONS); } - DatabaseAgent agent = agentFactory.get(); - if (poolRegistry.isEnabled() && agent instanceof AbstractJdbcAgent jdbcAgent) { - jdbcAgent.attachConnectionPoolRegistry(poolRegistry); - ensureMaintenanceStarted(); + Session session; + if (sessionHandlerFactory != null) { + session = new Session(sessionHandlerFactory.get()); + } else { + DatabaseAgent agent = agentFactory.get(); + if (poolRegistry.isEnabled() && agent instanceof AbstractJdbcAgent jdbcAgent) { + jdbcAgent.attachConnectionPoolRegistry(poolRegistry); + ensureMaintenanceStarted(); + } + session = new Session(new JsonRpcServer(agent)); } - Session session = new Session(new JsonRpcServer(agent)); Session existing = sessions.putIfAbsent(sessionId, session); if (existing != null) { throw new IllegalStateException("Agent session already exists: " + sessionId); } try { - return session.handle(AgentProtocol.METHOD_CONNECT, params); + return session.connect(params); } catch (Exception error) { sessions.remove(sessionId, session); session.close(); @@ -158,6 +178,24 @@ public final class MultiSessionJsonRpcServer implements AutoCloseable { return Collections.singletonMap("ok", true); } + private Object testSession(JsonObject params) throws Exception { + Session session = new Session(sessionHandlerFactory.get()); + try { + return session.connect(params); + } finally { + session.close(); + } + } + + private Object customHandshake() { + SessionRpcHandler handler = sessionHandlerFactory.get(); + try { + return handler.handshake(); + } finally { + handler.close(); + } + } + private Session session(String sessionId) { Session session = sessions.get(sessionId); if (session == null) { @@ -253,16 +291,34 @@ public final class MultiSessionJsonRpcServer implements AutoCloseable { private static final class Session { private final JsonRpcServer server; + private final SessionRpcHandler handler; private final ReentrantLock lock = new ReentrantLock(); private Session(JsonRpcServer server) { this.server = server; + this.handler = null; + } + + private Session(SessionRpcHandler handler) { + this.server = null; + this.handler = handler; } private Object handle(String method, JsonObject params) throws Exception { lock.lock(); try { - return server.dispatchForRuntime(method, params); + return handler == null ? server.dispatchForRuntime(method, params) : handler.handle(method, params); + } finally { + lock.unlock(); + } + } + + private Object connect(JsonObject params) throws Exception { + lock.lock(); + try { + return handler == null + ? server.dispatchForRuntime(AgentProtocol.METHOD_CONNECT, params) + : handler.connect(params); } finally { lock.unlock(); } @@ -271,7 +327,11 @@ public final class MultiSessionJsonRpcServer implements AutoCloseable { private void close() { lock.lock(); try { - server.dispatchForRuntime(AgentProtocol.METHOD_DISCONNECT, new JsonObject()); + if (handler != null) { + handler.close(); + } else { + server.dispatchForRuntime(AgentProtocol.METHOD_DISCONNECT, new JsonObject()); + } } catch (Exception ignored) { } finally { lock.unlock(); @@ -279,10 +339,17 @@ public final class MultiSessionJsonRpcServer implements AutoCloseable { } private void cancel() { - server.cancelActiveStatements(); + if (handler == null) { + server.cancelActiveStatements(); + } else { + handler.cancel(); + } } private void expireIdleResources() { + if (handler != null) { + return; + } if (!lock.tryLock()) { return; } @@ -294,6 +361,9 @@ public final class MultiSessionJsonRpcServer implements AutoCloseable { } private void expireIdleResources(long nowMillis, long idleTimeoutMillis) { + if (handler != null) { + return; + } if (!lock.tryLock()) { return; } diff --git a/agents/common/src/main/java/com/dbx/agent/SessionRpcHandler.java b/agents/common/src/main/java/com/dbx/agent/SessionRpcHandler.java new file mode 100644 index 000000000..568b54238 --- /dev/null +++ b/agents/common/src/main/java/com/dbx/agent/SessionRpcHandler.java @@ -0,0 +1,23 @@ +package com.dbx.agent; + +import com.google.gson.JsonObject; + +/** + * Per-session handler for protocol v2 agents that do not implement the JDBC + * {@link DatabaseAgent} contract. Implementations own every resource created + * for a logical DBX connection and must release it from {@link #close()}. + */ +public interface SessionRpcHandler { + default Object handshake() { + return AgentProtocol.multiSessionHandshakeResult(); + } + + Object connect(JsonObject params) throws Exception; + + Object handle(String method, JsonObject params) throws Exception; + + default void cancel() { + } + + void close(); +} diff --git a/agents/common/src/main/resources/agent-protocol-v1.json b/agents/common/src/main/resources/agent-protocol-v1.json index cbc33d0f4..b2c77b6f8 100644 --- a/agents/common/src/main/resources/agent-protocol-v1.json +++ b/agents/common/src/main/resources/agent-protocol-v1.json @@ -2,7 +2,7 @@ "protocolVersion": 1, "handshakeMethod": "handshake", "handshakeResponseFields": ["protocolVersion", "agentProtocolVersion", "capabilities"], - "allCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "kv", "kv_ttl", "kv_cas", "kv_list_values", "kv_status", "kv_history"], + "allCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "kv", "kv_ttl", "kv_cas", "kv_list_values", "kv_status", "kv_history", "etcd_compaction", "etcd_defrag", "etcd_watch", "etcd_lease", "etcd_auth"], "capabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl"], "defaultSqlCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl"], "commonMethods": [ @@ -40,5 +40,5 @@ "shutdown" ], "mongoLegacyMethods": ["list_databases", "list_collections", "find_documents", "find_documents_extended_json", "count_documents", "server_version", "create_index", "drop_indexes", "drop_collection", "insert_document", "update_document", "update_documents", "delete_document", "delete_documents"], - "kvMethods": ["kv_list_prefix", "kv_get", "kv_put", "kv_delete", "kv_rename", "kv_history", "kv_status"] + "kvMethods": ["kv_list_prefix", "kv_get", "kv_put", "kv_delete", "kv_rename", "kv_history", "kv_status", "etcd_compact", "etcd_defrag", "etcd_watch_start", "etcd_watch_poll", "etcd_watch_stop", "etcd_lease_list", "etcd_lease_get", "etcd_lease_grant", "etcd_lease_keepalive_once", "etcd_lease_revoke", "etcd_auth_user_list", "etcd_auth_user_get", "etcd_auth_user_add", "etcd_auth_user_delete", "etcd_auth_user_change_password", "etcd_auth_user_grant_role", "etcd_auth_user_revoke_role", "etcd_auth_role_list", "etcd_auth_role_get", "etcd_auth_role_add", "etcd_auth_role_delete", "etcd_auth_role_grant_permission", "etcd_auth_role_revoke_permission"] } diff --git a/agents/common/src/main/resources/agent-protocol-v2.json b/agents/common/src/main/resources/agent-protocol-v2.json index 33fbd99a0..8764572ba 100644 --- a/agents/common/src/main/resources/agent-protocol-v2.json +++ b/agents/common/src/main/resources/agent-protocol-v2.json @@ -2,7 +2,7 @@ "protocolVersion": 2, "handshakeMethod": "handshake", "handshakeResponseFields": ["protocolVersion", "agentProtocolVersion", "capabilities"], - "allCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "kv", "kv_ttl", "kv_cas", "kv_list_values", "kv_status", "kv_history", "multi_session"], + "allCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "kv", "kv_ttl", "kv_cas", "kv_list_values", "kv_status", "kv_history", "etcd_compaction", "etcd_defrag", "etcd_watch", "etcd_lease", "etcd_auth", "multi_session"], "capabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "multi_session"], "defaultSqlCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "multi_session"], "commonMethods": [ @@ -44,7 +44,7 @@ "shutdown" ], "mongoLegacyMethods": ["list_databases", "list_collections", "find_documents", "find_documents_extended_json", "count_documents", "server_version", "create_index", "drop_indexes", "drop_collection", "insert_document", "update_document", "update_documents", "delete_document", "delete_documents"], - "kvMethods": ["kv_list_prefix", "kv_get", "kv_put", "kv_delete", "kv_rename", "kv_history", "kv_status"], + "kvMethods": ["kv_list_prefix", "kv_get", "kv_put", "kv_delete", "kv_rename", "kv_history", "kv_status", "etcd_compact", "etcd_defrag", "etcd_watch_start", "etcd_watch_poll", "etcd_watch_stop", "etcd_lease_list", "etcd_lease_get", "etcd_lease_grant", "etcd_lease_keepalive_once", "etcd_lease_revoke", "etcd_auth_user_list", "etcd_auth_user_get", "etcd_auth_user_add", "etcd_auth_user_delete", "etcd_auth_user_change_password", "etcd_auth_user_grant_role", "etcd_auth_user_revoke_role", "etcd_auth_role_list", "etcd_auth_role_get", "etcd_auth_role_add", "etcd_auth_role_delete", "etcd_auth_role_grant_permission", "etcd_auth_role_revoke_permission"], "sessionField": "agentSessionId", "cursorSessionField": "sessionId" } diff --git a/agents/common/src/test/java/com/dbx/agent/CommonJavaCompatibilityTest.java b/agents/common/src/test/java/com/dbx/agent/CommonJavaCompatibilityTest.java index 54d2c7406..469273a76 100644 --- a/agents/common/src/test/java/com/dbx/agent/CommonJavaCompatibilityTest.java +++ b/agents/common/src/test/java/com/dbx/agent/CommonJavaCompatibilityTest.java @@ -66,7 +66,26 @@ class CommonJavaCompatibilityTest { JsonObject contract = protocolContract("/agent-protocol-v2.json"); assertEquals(AgentProtocol.MULTI_SESSION_PROTOCOL_VERSION, contract.get("protocolVersion").getAsInt()); + assertEquals(AgentProtocol.METHOD_HANDSHAKE, contract.get("handshakeMethod").getAsString()); + assertEquals( + Arrays.asList("protocolVersion", "agentProtocolVersion", "capabilities"), + strings(contract.getAsJsonArray("handshakeResponseFields")) + ); + assertEquals( + AgentProtocol.MULTI_SESSION_ALL_CAPABILITIES, + strings(contract.getAsJsonArray("allCapabilities")) + ); + assertEquals( + AgentProtocol.MULTI_SESSION_CAPABILITIES, + strings(contract.getAsJsonArray("capabilities")) + ); + assertEquals( + AgentProtocol.MULTI_SESSION_CAPABILITIES, + strings(contract.getAsJsonArray("defaultSqlCapabilities")) + ); assertEquals(AgentProtocol.MULTI_SESSION_METHODS, strings(contract.getAsJsonArray("commonMethods"))); + assertEquals(AgentProtocol.MONGO_LEGACY_METHODS, strings(contract.getAsJsonArray("mongoLegacyMethods"))); + assertEquals(AgentProtocol.KV_METHODS, strings(contract.getAsJsonArray("kvMethods"))); } @Test diff --git a/agents/drivers/etcd/src/main/java/com/dbx/agent/etcd/EtcdAgent.java b/agents/drivers/etcd/src/main/java/com/dbx/agent/etcd/EtcdAgent.java index fb596d3fe..9605ecd06 100644 --- a/agents/drivers/etcd/src/main/java/com/dbx/agent/etcd/EtcdAgent.java +++ b/agents/drivers/etcd/src/main/java/com/dbx/agent/etcd/EtcdAgent.java @@ -1,12 +1,21 @@ package com.dbx.agent.etcd; import com.dbx.agent.AgentProtocol; +import com.dbx.agent.MultiSessionJsonRpcServer; +import com.dbx.agent.SessionRpcHandler; +import com.google.protobuf.CodedInputStream; +import com.google.protobuf.WireFormat; import com.google.gson.Gson; import com.google.gson.JsonElement; import com.google.gson.JsonObject; import com.google.gson.JsonParser; +import io.grpc.MethodDescriptor; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; import io.grpc.netty.GrpcSslContexts; +import io.grpc.stub.ClientCalls; import io.etcd.jetcd.ByteSequence; +import io.etcd.jetcd.Auth; import io.etcd.jetcd.Client; import io.etcd.jetcd.ClientBuilder; import io.etcd.jetcd.Cluster; @@ -20,6 +29,7 @@ import io.etcd.jetcd.cluster.MemberListResponse; import io.etcd.jetcd.kv.TxnResponse; import io.etcd.jetcd.lease.LeaseGrantResponse; import io.etcd.jetcd.lease.LeaseTimeToLiveResponse; +import io.etcd.jetcd.api.VertxLeaseGrpc; import io.etcd.jetcd.maintenance.AlarmMember; import io.etcd.jetcd.maintenance.AlarmResponse; import io.etcd.jetcd.maintenance.StatusResponse; @@ -36,15 +46,21 @@ import io.etcd.jetcd.options.PutOption; import io.etcd.jetcd.options.WatchOption; import io.etcd.jetcd.watch.WatchEvent; import io.etcd.jetcd.watch.WatchResponse; +import io.etcd.jetcd.auth.Permission; import io.netty.handler.ssl.SslContext; import io.netty.handler.ssl.SslContextBuilder; +import java.io.ByteArrayInputStream; import java.io.BufferedReader; import java.io.File; +import java.io.IOException; +import java.io.InputStream; import java.io.InputStreamReader; +import java.lang.reflect.Field; import java.nio.ByteBuffer; import java.nio.charset.CharacterCodingException; import java.nio.charset.CharsetDecoder; import java.nio.charset.CodingErrorAction; +import java.time.Duration; import java.util.Arrays; import java.util.ArrayList; import java.util.ArrayDeque; @@ -64,6 +80,8 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.ConcurrentHashMap; +import java.util.UUID; public final class EtcdAgent { private static final Gson GSON = new Gson(); @@ -79,33 +97,77 @@ public final class EtcdAgent { AgentProtocol.CAPABILITY_KV_CAS, AgentProtocol.CAPABILITY_KV_LIST_VALUES, AgentProtocol.CAPABILITY_KV_STATUS, - AgentProtocol.CAPABILITY_KV_HISTORY + AgentProtocol.CAPABILITY_KV_HISTORY, + AgentProtocol.CAPABILITY_ETCD_COMPACTION, + AgentProtocol.CAPABILITY_ETCD_DEFRAG, + AgentProtocol.CAPABILITY_ETCD_WATCH, + AgentProtocol.CAPABILITY_ETCD_LEASE, + AgentProtocol.CAPABILITY_ETCD_AUTH, + AgentProtocol.CAPABILITY_MULTI_SESSION )); - private static Client client; - private static KV kv; - private static List connectedEndpoints = Collections.emptyList(); + private static final int MAX_WATCHES = 4; + private static final int MAX_WATCH_BATCHES = 256; + private static final int MAX_WATCH_EVENTS = 10_000; + static final long MAX_WATCH_BUFFER_BYTES = 8L * 1024 * 1024; + static final long MAX_SESSION_WATCH_BUFFER_BYTES = 16L * 1024 * 1024; + private static final int MAX_LEASE_ATTACHED_KEYS = 256; + private static final int DEFAULT_LEASE_LIST_LIMIT = 100; + private static final int MAX_LEASE_LIST_LIMIT = 200; + private static final int LEASE_LIST_CONCURRENCY = 8; + private static final int LEASE_LIST_DEADLINE_SECONDS = 5; + private static final MethodDescriptor.Marshaller BYTE_ARRAY_MARSHALLER = new MethodDescriptor.Marshaller<>() { + @Override + public InputStream stream(byte[] value) { + return new ByteArrayInputStream(value); + } + + @Override + public byte[] parse(InputStream stream) { + try { + return stream.readAllBytes(); + } catch (IOException error) { + throw new IllegalStateException("Failed to read etcd gRPC response", error); + } + } + }; + private static final MethodDescriptor LEASES_METHOD = MethodDescriptor.newBuilder() + .setType(MethodDescriptor.MethodType.UNARY) + .setFullMethodName(MethodDescriptor.generateFullMethodName("etcdserverpb.Lease", "LeaseLeases")) + .setRequestMarshaller(BYTE_ARRAY_MARSHALLER) + .setResponseMarshaller(BYTE_ARRAY_MARSHALLER) + .build(); + private static final ThreadLocal CURRENT_SESSION = new ThreadLocal<>(); + private static final EtcdSessionState LEGACY_SESSION = new EtcdSessionState(); private EtcdAgent() { } private static Object handshakeResult() { - return new HandshakeResult(AgentProtocol.PROTOCOL_VERSION, AgentProtocol.PROTOCOL_VERSION, CAPABILITIES); + return new HandshakeResult(AgentProtocol.MULTI_SESSION_PROTOCOL_VERSION, AgentProtocol.MULTI_SESSION_PROTOCOL_VERSION, CAPABILITIES); } private static Object connect(JsonObject params) throws Exception { + EtcdSessionState state = sessionState(); JsonObject connection = connectionObject(params); Client nextClient = buildClient(connection); - nextClient.getKVClient().get(byteSequence("\0")).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + try { + probeClient(nextClient, endpoints(connection)); + } catch (Exception error) { + nextClient.close(); + throw error; + } closeClient(); - client = nextClient; - kv = client.getKVClient(); - connectedEndpoints = endpoints(connection); + state.client = nextClient; + state.kv = nextClient.getKVClient(); + state.connectedEndpoints = endpoints(connection); return Collections.singletonMap("ok", true); } static Client buildClient(JsonObject connection) throws Exception { List endpoints = endpoints(connection); - ClientBuilder builder = Client.builder().endpoints(endpoints.toArray(String[]::new)); + ClientBuilder builder = Client.builder() + .endpoints(endpoints.toArray(String[]::new)) + .connectTimeout(Duration.ofSeconds(connectTimeoutSeconds(connection))); String username = stringOrEmpty(connection, "username"); String password = stringOrEmpty(connection, "password"); if (!username.isBlank()) { @@ -118,6 +180,46 @@ public final class EtcdAgent { return builder.build(); } + static int connectTimeoutSeconds(JsonObject connection) { + return Math.min(300, Math.max(1, intOrDefault(connection, "connect_timeout_secs", RPC_TIMEOUT_SECONDS))); + } + + private static Map validateConnectedClient() throws Exception { + EtcdSessionState state = sessionState(); + Client active = requireClient(); + return probeClient(active, state.connectedEndpoints); + } + + private static Map probeClient(Client candidate, List endpoints) throws Exception { + Maintenance maintenance = candidate.getMaintenanceClient(); + Exception lastFailure = null; + for (String endpoint : endpoints) { + try { + maintenance.statusMember(endpoint).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + Map result = new LinkedHashMap<>(); + result.put("ok", true); + result.put("endpoint", endpoint); + return result; + } catch (Exception error) { + Throwable cause = rootCause(error); + // A restricted etcd user may not be allowed to call Maintenance.Status. + // PERMISSION_DENIED still proves that the channel reached an etcd server. + if (Status.fromThrowable(cause).getCode() == Status.Code.PERMISSION_DENIED) { + Map result = new LinkedHashMap<>(); + result.put("ok", true); + result.put("endpoint", endpoint); + result.put("limited", true); + return result; + } + lastFailure = error; + } + } + if (lastFailure != null) { + throw lastFailure; + } + throw new IllegalStateException("No etcd endpoint configured"); + } + static List endpoints(JsonObject connection) { String configured = firstNonBlank( stringOrNull(connection, "etcd_endpoints"), @@ -408,7 +510,7 @@ public final class EtcdAgent { if (requestedEnd != null && requestedEnd > 0) { latestOption.withRevision(requestedEnd); } - GetResponse latest = kv.get(key, latestOption.build()).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + GetResponse latest = sessionState().kv.get(key, latestOption.build()).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); long endRevision = requestedEnd == null ? latest.getHeader().getRevision() : requestedEnd; long targetKeyRevision = latest.getKvs().isEmpty() ? endRevision @@ -428,7 +530,7 @@ public final class EtcdAgent { AtomicReference failure = new AtomicReference<>(); CountDownLatch created = new CountDownLatch(1); CountDownLatch completed = new CountDownLatch(1); - Watch watch = client.getWatchClient(); + Watch watch = sessionState().client.getWatchClient(); WatchOption option = WatchOption.newBuilder() .withRevision(startRevision) .withPrevKV(true) @@ -552,10 +654,10 @@ public final class EtcdAgent { private static Object status() throws Exception { requireKv(); - Maintenance maintenance = client.getMaintenanceClient(); - Cluster cluster = client.getClusterClient(); + Maintenance maintenance = sessionState().client.getMaintenanceClient(); + Cluster cluster = sessionState().client.getClusterClient(); Map membersById = new HashMap<>(); - List endpoints = new ArrayList<>(connectedEndpoints); + List endpoints = new ArrayList<>(sessionState().connectedEndpoints); try { MemberListResponse memberList = cluster.listMember().get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); for (Member member : memberList.getMembers()) { @@ -588,7 +690,7 @@ public final class EtcdAgent { .withRange(ByteSequence.from(new byte[] {0})) .withCountOnly(true) .build(); - GetResponse countResponse = kv.get(ByteSequence.from(new byte[] {0}), countOption) + GetResponse countResponse = sessionState().kv.get(ByteSequence.from(new byte[] {0}), countOption) .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); List> statusMembers = new ArrayList<>(); @@ -661,10 +763,628 @@ public final class EtcdAgent { return result; } + private static Object compact(JsonObject params) throws Exception { + long revision = requiredPositiveLong(params, "revision"); + try { + requireKv().get(ByteSequence.from(new byte[] {0}), GetOption.newBuilder() + .withRange(ByteSequence.from(new byte[] {0})) + .withCountOnly(true) + .withRevision(revision) + .build()) + .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } catch (Exception error) { + Throwable cause = rootCause(error); + if (cause instanceof io.etcd.jetcd.common.exception.CompactedException compacted) { + throw new IllegalArgumentException( + "ETCD_INVALID_REVISION: revision was already compacted at " + compacted.getCompactedRevision() + ); + } + throw error; + } + GetResponse current = requireKv().get(ByteSequence.from(new byte[] {0}), GetOption.newBuilder() + .withRange(ByteSequence.from(new byte[] {0})).withCountOnly(true).build()) + .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + if (revision > current.getHeader().getRevision()) { + throw new IllegalArgumentException("ETCD_INVALID_REVISION: revision is newer than the current revision"); + } + requireKv().compact(revision).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + return Map.of("revision", longString(revision)); + } + + private static Object defrag(JsonObject params) throws Exception { + JsonElement endpointsValue = params.get("endpoints"); + if (endpointsValue == null || !endpointsValue.isJsonArray() || endpointsValue.getAsJsonArray().isEmpty()) { + throw new IllegalArgumentException("ETCD_DEFRAG_TARGET_REQUIRED: at least one endpoint is required"); + } + Maintenance maintenance = requireClient().getMaintenanceClient(); + List> members = new ArrayList<>(); + List remaining = new ArrayList<>(); + for (JsonElement value : endpointsValue.getAsJsonArray()) { + String endpoint = value.getAsString(); + if (!endpoint.isBlank() && !remaining.contains(endpoint)) remaining.add(endpoint); + } + while (!remaining.isEmpty()) { + String endpoint = nextDefragEndpoint(maintenance, remaining); + long started = System.nanoTime(); + Map row = new LinkedHashMap<>(); + row.put("endpoint", endpoint); + try { + maintenance.defragmentMember(endpoint).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + row.put("status", "succeeded"); + row.put("durationMs", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started)); + } catch (Exception error) { + row.put("status", "failed"); + row.put("durationMs", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started)); + row.put("error", safeMessage(rootCause(error))); + members.add(row); + appendUnexecutedDefragMembers(members, remaining, endpoint); + break; + } + members.add(row); + remaining.remove(endpoint); + } + return Map.of("members", members); + } + + static void appendUnexecutedDefragMembers( + List> members, + List remaining, + String failedEndpoint + ) { + for (String endpoint : remaining) { + if (endpoint.equals(failedEndpoint)) continue; + Map row = new LinkedHashMap<>(); + row.put("endpoint", endpoint); + row.put("status", "not_executed"); + row.put("durationMs", null); + row.put("error", null); + members.add(row); + } + } + + /** Re-evaluate leadership before each member so a leader change is not defragmented early. */ + private static String nextDefragEndpoint(Maintenance maintenance, List remaining) throws Exception { + String leaderEndpoint = null; + for (String endpoint : remaining) { + try { + StatusResponse status = maintenance.statusMember(endpoint).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + if (status.getHeader().getMemberId() == status.getLeader()) { + leaderEndpoint = endpoint; + break; + } + } catch (Exception ignored) { + // The following defragment call reports this member as the failed step. + } + } + for (String endpoint : remaining) { + if (!endpoint.equals(leaderEndpoint)) return endpoint; + } + return leaderEndpoint == null ? remaining.get(0) : leaderEndpoint; + } + + private static Object watchStart(JsonObject params) throws Exception { + EtcdSessionState session = sessionState(); + if (session.watches.size() >= MAX_WATCHES) { + throw new IllegalStateException("ETCD_WATCH_LIMIT: at most " + MAX_WATCHES + " watches are allowed per connection"); + } + ByteSequence key = keyBytes(params); + String scope = stringOrDefault(params, "scope", "key"); + if (!"key".equals(scope) && !"prefix".equals(scope)) { + throw new IllegalArgumentException("ETCD_WATCH_SCOPE_INVALID: scope must be key or prefix"); + } + Long requestedRevision = longOrNull(params, "startRevision"); + long startedRevision; + if (requestedRevision != null && requestedRevision > 0) { + startedRevision = requestedRevision; + } else { + GetResponse response = requireKv().get(ByteSequence.from(new byte[] {0}), GetOption.newBuilder() + .withRange(ByteSequence.from(new byte[] {0})).withCountOnly(true).build()) + .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + startedRevision = response.getHeader().getRevision() + 1; + } + String watchId = UUID.randomUUID().toString(); + EtcdWatchState state = new EtcdWatchState(watchId, session); + WatchOption.Builder option = WatchOption.newBuilder().withRevision(startedRevision) + .withPrevKV(boolOrDefault(params, "includePrevKv", false)); + if ("prefix".equals(scope)) option.withRange(prefixEnd(key)); + try { + Watch.Watcher watcher = requireClient().getWatchClient().watch(key, option.build(), new Watch.Listener() { + @Override + public void onNext(WatchResponse response) { + if (response.getEvents().isEmpty()) return; + List> events = new ArrayList<>(); + long bufferedBytes = 128; + for (WatchEvent event : response.getEvents()) { + KeyValue item = event.getKeyValue(); + KeyValue previous = event.getPrevKV(); + bufferedBytes += watchEventBufferBytes(item, previous); + if (bufferedBytes > MAX_WATCH_BUFFER_BYTES) { + state.overflow(); + return; + } + Map row = new LinkedHashMap<>(); + row.put("eventType", event.getEventType() == WatchEvent.EventType.DELETE ? "delete" : "put"); + row.put("revision", longString(item.getModRevision())); + row.put("key", displayBytes(item.getKey().getBytes())); + row.put("keyBytes", bytesObject(item.getKey().getBytes())); + row.put("value", event.getEventType() == WatchEvent.EventType.DELETE ? null : valueObject(item.getValue().getBytes())); + row.put("previousValue", previous != null && previous.getVersion() > 0 ? valueObject(previous.getValue().getBytes()) : null); + row.put("metadata", event.getEventType() == WatchEvent.EventType.DELETE && previous != null ? metadata(previous) : metadata(item)); + events.add(row); + } + state.append(response.getHeader().getRevision(), events, bufferedBytes); + } + + @Override + public void onError(Throwable error) { + Throwable cause = rootCause(error); + if (cause instanceof io.etcd.jetcd.common.exception.CompactedException compacted) { + state.fail("compacted", "ETCD_COMPACTED", compacted.getCompactedRevision()); + } else { + state.fail("error", safeMessage(cause), null); + } + } + + @Override + public void onCompleted() { + state.fail("closed", "watch closed", null); + } + }); + state.setWatcher(watcher); + session.watches.put(watchId, state); + } catch (Exception error) { + state.close(); + throw error; + } + return Map.of("watchId", watchId, "startedRevision", longString(startedRevision)); + } + + private static Object watchPoll(JsonObject params) throws Exception { + return pollWatchState(sessionState(), stringOrEmpty(params, "watchId")); + } + + static Map pollWatchState(EtcdSessionState session, String watchId) { + EtcdWatchState state = session.watches.get(watchId); + if (state == null) throw new IllegalStateException("ETCD_WATCH_NOT_FOUND: watch does not exist"); + Map result = state.poll(); + if (result.containsKey("terminal") && session.watches.remove(watchId, state)) { + state.close(); + } + return result; + } + + private static Object watchStop(JsonObject params) { + EtcdWatchState state = sessionState().watches.remove(stringOrEmpty(params, "watchId")); + if (state != null) state.close(); + return Map.of("stopped", true); + } + + private static Object leaseList(JsonObject params) throws Exception { + long deadlineNanos = System.nanoTime() + TimeUnit.SECONDS.toNanos(LEASE_LIST_DEADLINE_SECONDS); + int limit = Math.min(Math.max(1, intOrDefault(params, "limit", DEFAULT_LEASE_LIST_LIMIT)), MAX_LEASE_LIST_LIMIT); + String continuation = stringOrNull(params, "continuation"); + Long afterLeaseId = continuation == null || continuation.isBlank() + ? null + : Long.parseUnsignedLong(continuation); + boolean partial = false; + List leaseIds; + try { + leaseIds = clusterLeaseIds(remainingMillis(deadlineNanos)); + sessionState().knownLeases.addAll(leaseIds); + } catch (ReflectiveOperationException error) { + leaseIds = new ArrayList<>(sessionState().knownLeases); + partial = true; + } catch (java.util.concurrent.TimeoutException error) { + leaseIds = new ArrayList<>(sessionState().knownLeases); + partial = true; + } catch (StatusRuntimeException error) { + if (error.getStatus().getCode() != Status.Code.UNIMPLEMENTED + && error.getStatus().getCode() != Status.Code.DEADLINE_EXCEEDED) { + throw error; + } + leaseIds = new ArrayList<>(sessionState().knownLeases); + partial = true; + } + leaseIds = leasePageIds(leaseIds, afterLeaseId, limit + 1); + boolean hasMore = leaseIds.size() > limit; + List pageIds = new ArrayList<>(leaseIds.subList(0, Math.min(limit, leaseIds.size()))); + + List> leases = new ArrayList<>(); + Long lastProcessedId = null; + boolean deadlineReached = false; + outer: + for (int offset = 0; offset < pageIds.size(); offset += LEASE_LIST_CONCURRENCY) { + List chunkIds = pageIds.subList(offset, Math.min(offset + LEASE_LIST_CONCURRENCY, pageIds.size())); + Map> requests = new LinkedHashMap<>(); + for (Long id : chunkIds) { + requests.put(id, requireClient().getLeaseClient().timeToLive(id, LeaseOption.DEFAULT)); + } + for (Map.Entry> request : requests.entrySet()) { + try { + LeaseTimeToLiveResponse response = request.getValue().get( + remainingMillis(deadlineNanos), + TimeUnit.MILLISECONDS + ); + Map row = new LinkedHashMap<>(); + row.put("id", longString(response.getID())); + row.put("ttl", response.getTTL()); + row.put("grantedTtl", response.getGrantedTTL()); + leases.add(row); + lastProcessedId = request.getKey(); + } catch (java.util.concurrent.TimeoutException error) { + requests.values().forEach(future -> future.cancel(true)); + partial = true; + deadlineReached = true; + break outer; + } catch (Exception error) { + Throwable cause = rootCause(error); + if (Status.fromThrowable(cause).getCode() == Status.Code.NOT_FOUND) { + sessionState().knownLeases.remove(request.getKey()); + } else { + partial = true; + } + lastProcessedId = request.getKey(); + } + } + } + String nextContinuation = null; + if (deadlineReached && !pageIds.isEmpty()) { + nextContinuation = unsignedLongString(lastProcessedId != null ? lastProcessedId : afterLeaseId != null ? afterLeaseId : 0L); + } else if (hasMore && !pageIds.isEmpty()) { + nextContinuation = unsignedLongString(pageIds.get(pageIds.size() - 1)); + } + Map result = new LinkedHashMap<>(); + result.put("leases", leases); + result.put("partial", partial); + result.put("nextContinuation", nextContinuation); + return result; + } + + static List leasePageIds(List leaseIds, Long afterLeaseId, int fetchLimit) { + return leaseIds.stream() + .sorted(Long::compareUnsigned) + .filter(id -> afterLeaseId == null || Long.compareUnsigned(id, afterLeaseId) > 0) + .limit(Math.max(0, fetchLimit)) + .toList(); + } + + private static long remainingMillis(long deadlineNanos) throws java.util.concurrent.TimeoutException { + long remaining = TimeUnit.NANOSECONDS.toMillis(deadlineNanos - System.nanoTime()); + if (remaining <= 0) throw new java.util.concurrent.TimeoutException("ETCD_LEASE_LIST_TIMEOUT"); + return remaining; + } + + private static List clusterLeaseIds(long timeoutMillis) throws Exception { + VertxLeaseGrpc.LeaseVertxStub stub = leaseStub(requireClient().getLeaseClient()); + byte[] response = ClientCalls.blockingUnaryCall( + stub.getChannel(), + LEASES_METHOD, + stub.getCallOptions().withDeadlineAfter(timeoutMillis, TimeUnit.MILLISECONDS), + new byte[0] + ); + return leaseIdsFromResponse(response); + } + + static List leaseIdsFromResponse(byte[] response) throws IOException { + List ids = new ArrayList<>(); + CodedInputStream input = CodedInputStream.newInstance(response); + while (!input.isAtEnd()) { + int tag = input.readTag(); + if (tag == 0) break; + if (WireFormat.getTagFieldNumber(tag) != 2 || WireFormat.getTagWireType(tag) != WireFormat.WIRETYPE_LENGTH_DELIMITED) { + input.skipField(tag); + continue; + } + CodedInputStream lease = CodedInputStream.newInstance(input.readByteArray()); + while (!lease.isAtEnd()) { + int leaseTag = lease.readTag(); + if (leaseTag == 0) break; + if (WireFormat.getTagFieldNumber(leaseTag) == 1 && WireFormat.getTagWireType(leaseTag) == WireFormat.WIRETYPE_VARINT) { + ids.add(lease.readInt64()); + } else { + lease.skipField(leaseTag); + } + } + } + return ids; + } + + private static Object leaseGet(JsonObject params) throws Exception { + long id = requiredPositiveLong(params, "id"); + LeaseOption option = boolOrDefault(params, "includeKeys", false) ? LeaseOption.newBuilder().withAttachedKeys().build() : LeaseOption.DEFAULT; + LeaseTimeToLiveResponse response = requireClient().getLeaseClient().timeToLive(id, option) + .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + sessionState().knownLeases.add(id); + List> keys = new ArrayList<>(); + List attachedKeys = response.getKeys(); + for (int index = 0; index < attachedKeys.size() && index < MAX_LEASE_ATTACHED_KEYS; index++) { + keys.add(bytesObject(attachedKeys.get(index).getBytes())); + } + Map result = new LinkedHashMap<>(); + result.put("id", longString(response.getID())); + result.put("ttl", response.getTTL()); + result.put("grantedTtl", response.getGrantedTTL()); + result.put("keys", keys); + result.put("truncated", attachedKeys.size() > MAX_LEASE_ATTACHED_KEYS); + return result; + } + + private static Object leaseGrant(JsonObject params) throws Exception { + long ttl = requiredPositiveLong(params, "ttl"); + Long requestedId = longOrNull(params, "id"); + if (requestedId != null && requestedId < 0) throw new IllegalArgumentException("id must be a positive integer or 0"); + LeaseGrantResponse response = requestedId == null || requestedId == 0 + ? requireClient().getLeaseClient().grant(ttl).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS) + : grantLeaseWithRequestedId(ttl, requestedId); + sessionState().knownLeases.add(response.getID()); + return Map.of("id", longString(response.getID()), "ttl", response.getTTL()); + } + + private static LeaseGrantResponse grantLeaseWithRequestedId(long ttl, long requestedId) throws Exception { + Lease lease = requireClient().getLeaseClient(); + try { + VertxLeaseGrpc.LeaseVertxStub stub = leaseStub(lease); + io.etcd.jetcd.api.LeaseGrantResponse response = stub + .withDeadlineAfter(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .leaseGrant(io.etcd.jetcd.api.LeaseGrantRequest.newBuilder().setTTL(ttl).setID(requestedId).build()) + .toCompletionStage() + .toCompletableFuture() + .get(RPC_TIMEOUT_SECONDS + 1L, TimeUnit.SECONDS); + return new LeaseGrantResponse(response); + } catch (ReflectiveOperationException error) { + throw new IllegalStateException("Custom Lease ID is unavailable in the installed Jetcd version", error); + } + } + + private static VertxLeaseGrpc.LeaseVertxStub leaseStub(Lease lease) throws ReflectiveOperationException { + Field stubField = lease.getClass().getDeclaredField("stub"); + stubField.setAccessible(true); + return (VertxLeaseGrpc.LeaseVertxStub) stubField.get(lease); + } + + private static Object leaseKeepAlive(JsonObject params) throws Exception { + long id = requiredPositiveLong(params, "id"); + long ttl = requireClient().getLeaseClient().keepAliveOnce(id).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS).getTTL(); + sessionState().knownLeases.add(id); + return Map.of("id", longString(id), "ttl", ttl); + } + + private static Object leaseRevoke(JsonObject params) throws Exception { + long id = requiredPositiveLong(params, "id"); + requireClient().getLeaseClient().revoke(id).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + sessionState().knownLeases.remove(id); + return Map.of("id", longString(id), "revoked", true); + } + + private static Object authUserList() throws Exception { + return Map.of("users", requireClient().getAuthClient().userList().get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS).getUsers()); + } + + private static Object authUserGet(JsonObject params) throws Exception { + String user = requiredString(params, "user"); + return Map.of("user", user, "roles", requireClient().getAuthClient().userGet(byteSequence(user)) + .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS).getRoles()); + } + + private static Object authUserAdd(JsonObject params) throws Exception { + requireClient().getAuthClient().userAdd(byteSequence(requiredString(params, "user")), byteSequence(requiredString(params, "password"))) + .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + return Map.of("created", true); + } + + private static Object authUserDelete(JsonObject params) throws Exception { + requireClient().getAuthClient().userDelete(byteSequence(requiredString(params, "user"))).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + return Map.of("deleted", true); + } + + private static Object authUserChangePassword(JsonObject params) throws Exception { + requireClient().getAuthClient().userChangePassword(byteSequence(requiredString(params, "user")), byteSequence(requiredString(params, "password"))) + .get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + return Map.of("changed", true); + } + + private static Object authUserRole(JsonObject params, boolean grant) throws Exception { + Auth auth = requireClient().getAuthClient(); + ByteSequence user = byteSequence(requiredString(params, "user")); + ByteSequence role = byteSequence(requiredString(params, "role")); + if (grant) auth.userGrantRole(user, role).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + else auth.userRevokeRole(user, role).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + return Map.of("updated", true); + } + + private static Object authRoleList() throws Exception { + return Map.of("roles", requireClient().getAuthClient().roleList().get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS).getRoles()); + } + + private static Object authRoleGet(JsonObject params) throws Exception { + String role = requiredString(params, "role"); + List> permissions = new ArrayList<>(); + for (Permission permission : requireClient().getAuthClient().roleGet(byteSequence(role)).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS).getPermissions()) { + Map item = new LinkedHashMap<>(); + item.put("access", permission.getPermType().name().toLowerCase()); + item.put("key", bytesObject(permission.getKey().getBytes())); + item.put("rangeEnd", bytesObject(permission.getRangeEnd().getBytes())); + byte[] keyBytes = permission.getKey().getBytes(); + byte[] rangeEndBytes = permission.getRangeEnd().getBytes(); + String resource = keyBytes.length == 1 && keyBytes[0] == 0 + && rangeEndBytes.length == 1 && rangeEndBytes[0] == 0 + ? "all" + : rangeEndBytes.length == 0 ? "key" : "prefix"; + item.put("resource", resource); + permissions.add(item); + } + return Map.of("role", role, "permissions", permissions); + } + + private static Object authRoleAdd(JsonObject params) throws Exception { + requireClient().getAuthClient().roleAdd(byteSequence(requiredString(params, "role"))).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + return Map.of("created", true); + } + + private static Object authRoleDelete(JsonObject params) throws Exception { + requireClient().getAuthClient().roleDelete(byteSequence(requiredString(params, "role"))).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + return Map.of("deleted", true); + } + + private static Object authRolePermission(JsonObject params, boolean grant) throws Exception { + Auth auth = requireClient().getAuthClient(); + ByteSequence role = byteSequence(requiredString(params, "role")); + String resource = stringOrDefault(params, "resource", "key"); + boolean all = "all".equals(resource); + boolean prefix = "prefix".equals(resource); + ByteSequence key = all ? ByteSequence.from(new byte[] {0}) : keyBytes(params); + ByteSequence rangeEnd = all + ? ByteSequence.from(new byte[] {0}) + : prefix ? prefixEnd(key) : ByteSequence.from(new byte[0]); + if (grant) { + Permission.Type access = Permission.Type.valueOf(requiredString(params, "access").toUpperCase()); + auth.roleGrantPermission(role, key, rangeEnd, access).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } else { + auth.roleRevokePermission(role, key, rangeEnd).get(RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } + return Map.of("updated", true); + } + + private static long requiredPositiveLong(JsonObject params, String field) { + Long value = longOrNull(params, field); + if (value == null || value <= 0) throw new IllegalArgumentException("ETCD_INVALID_" + field.toUpperCase() + ": a positive integer is required"); + return value; + } + + private static String requiredString(JsonObject params, String field) { + String value = stringOrNull(params, field); + if (value == null || value.isBlank()) throw new IllegalArgumentException("ETCD_" + field.toUpperCase() + "_REQUIRED"); + return value; + } + + static long watchEventBufferBytes(KeyValue item, KeyValue previous) { + long bytes = 512L + estimatedBufferedBytes(item.getKey().size()); + bytes += estimatedBufferedBytes(item.getValue().size()); + if (previous != null && previous.getVersion() > 0) { + bytes += estimatedBufferedBytes(previous.getValue().size()); + } + return bytes; + } + + static long estimatedBufferedBytes(int sourceBytes) { + return Math.max(0L, sourceBytes) * 4L; + } + + static final class BufferedWatchBatch { + private final Map payload; + private final long bufferedBytes; + + private BufferedWatchBatch(Map payload, long bufferedBytes) { + this.payload = payload; + this.bufferedBytes = bufferedBytes; + } + } + + static final class EtcdWatchState { + private final String watchId; + private final EtcdSessionState session; + private final Deque batches = new ArrayDeque<>(); + private int eventCount; + private long bufferedBytes; + private String terminalReason; + private String terminalMessage; + private Long compactedRevision; + private Watch.Watcher watcher; + + EtcdWatchState(String watchId, EtcdSessionState session) { + this.watchId = watchId; + this.session = session; + } + + synchronized void append(long revision, List> events, long batchBytes) { + if (terminalReason != null) return; + if (batches.size() >= MAX_WATCH_BATCHES + || eventCount + events.size() > MAX_WATCH_EVENTS + || batchBytes > MAX_WATCH_BUFFER_BYTES + || bufferedBytes + batchBytes > MAX_WATCH_BUFFER_BYTES + || !session.reserveWatchBuffer(batchBytes)) { + overflow(); + return; + } + Map batch = new LinkedHashMap<>(); + batch.put("revision", longString(revision)); + batch.put("events", events); + batches.addLast(new BufferedWatchBatch(batch, batchBytes)); + eventCount += events.size(); + bufferedBytes += batchBytes; + } + + synchronized void overflow() { + if (terminalReason != null) return; + terminalReason = "overflow"; + terminalMessage = "ETCD_WATCH_OVERFLOW: the event buffer reached its byte or event limit"; + closeWatcher(); + } + + private synchronized void fail(String reason, String message, Long compacted) { + if (terminalReason == null) { + terminalReason = reason; + terminalMessage = message; + compactedRevision = compacted; + } + } + + synchronized Map poll() { + List> page = new ArrayList<>(); + while (!batches.isEmpty() && page.size() < 64) { + BufferedWatchBatch batch = batches.removeFirst(); + eventCount -= ((List) batch.payload.get("events")).size(); + bufferedBytes -= batch.bufferedBytes; + session.releaseWatchBuffer(batch.bufferedBytes); + page.add(batch.payload); + } + Map result = new LinkedHashMap<>(); + result.put("watchId", watchId); + result.put("batches", page); + if (terminalReason != null && batches.isEmpty()) { + Map terminal = new LinkedHashMap<>(); + terminal.put("reason", terminalReason); + terminal.put("message", terminalMessage); + terminal.put("compactedRevision", compactedRevision == null ? null : longString(compactedRevision)); + result.put("terminal", terminal); + } + return result; + } + + private synchronized void close() { + fail("stopped", "watch stopped", null); + clearBuffered(); + closeWatcher(); + } + + private void clearBuffered() { + if (bufferedBytes > 0) session.releaseWatchBuffer(bufferedBytes); + batches.clear(); + eventCount = 0; + bufferedBytes = 0; + } + + private synchronized void setWatcher(Watch.Watcher createdWatcher) { + if (terminalReason != null) { + createdWatcher.close(); + } else { + watcher = createdWatcher; + } + } + + private void closeWatcher() { + if (watcher != null) { + watcher.close(); + watcher = null; + } + } + } + private static Object dispatch(String method, JsonObject params) throws Exception { return switch (method) { case AgentProtocol.METHOD_HANDSHAKE -> handshakeResult(); case AgentProtocol.METHOD_CONNECT, AgentProtocol.METHOD_TEST_CONNECTION -> connect(params); + case AgentProtocol.METHOD_VALIDATE_CONNECTION -> validateConnectedClient(); case AgentProtocol.KV_METHOD_LIST_PREFIX -> listPrefix(params); case AgentProtocol.KV_METHOD_GET -> get(params); case AgentProtocol.KV_METHOD_PUT -> put(params); @@ -672,6 +1392,29 @@ public final class EtcdAgent { case AgentProtocol.KV_METHOD_RENAME -> rename(params); case AgentProtocol.KV_METHOD_HISTORY -> history(params); case AgentProtocol.KV_METHOD_STATUS -> status(); + case AgentProtocol.ETCD_METHOD_COMPACT -> compact(params); + case AgentProtocol.ETCD_METHOD_DEFRAG -> defrag(params); + case AgentProtocol.ETCD_METHOD_WATCH_START -> watchStart(params); + case AgentProtocol.ETCD_METHOD_WATCH_POLL -> watchPoll(params); + case AgentProtocol.ETCD_METHOD_WATCH_STOP -> watchStop(params); + case AgentProtocol.ETCD_METHOD_LEASE_LIST -> leaseList(params); + case AgentProtocol.ETCD_METHOD_LEASE_GET -> leaseGet(params); + case AgentProtocol.ETCD_METHOD_LEASE_GRANT -> leaseGrant(params); + case AgentProtocol.ETCD_METHOD_LEASE_KEEPALIVE -> leaseKeepAlive(params); + case AgentProtocol.ETCD_METHOD_LEASE_REVOKE -> leaseRevoke(params); + case AgentProtocol.ETCD_METHOD_AUTH_USER_LIST -> authUserList(); + case AgentProtocol.ETCD_METHOD_AUTH_USER_GET -> authUserGet(params); + case AgentProtocol.ETCD_METHOD_AUTH_USER_ADD -> authUserAdd(params); + case AgentProtocol.ETCD_METHOD_AUTH_USER_DELETE -> authUserDelete(params); + case AgentProtocol.ETCD_METHOD_AUTH_USER_CHANGE_PASSWORD -> authUserChangePassword(params); + case AgentProtocol.ETCD_METHOD_AUTH_USER_GRANT_ROLE -> authUserRole(params, true); + case AgentProtocol.ETCD_METHOD_AUTH_USER_REVOKE_ROLE -> authUserRole(params, false); + case AgentProtocol.ETCD_METHOD_AUTH_ROLE_LIST -> authRoleList(); + case AgentProtocol.ETCD_METHOD_AUTH_ROLE_GET -> authRoleGet(params); + case AgentProtocol.ETCD_METHOD_AUTH_ROLE_ADD -> authRoleAdd(params); + case AgentProtocol.ETCD_METHOD_AUTH_ROLE_DELETE -> authRoleDelete(params); + case AgentProtocol.ETCD_METHOD_AUTH_ROLE_GRANT_PERMISSION -> authRolePermission(params, true); + case AgentProtocol.ETCD_METHOD_AUTH_ROLE_REVOKE_PERMISSION -> authRolePermission(params, false); case AgentProtocol.METHOD_DISCONNECT -> { closeClient(); yield Collections.singletonMap("ok", true); @@ -697,6 +1440,7 @@ public final class EtcdAgent { response.addProperty("jsonrpc", "2.0"); response.add("id", id); + CURRENT_SESSION.set(LEGACY_SESSION); try { Object result = dispatch(method, params); response.add("result", GSON.toJsonTree(result)); @@ -705,6 +1449,8 @@ public final class EtcdAgent { error.addProperty("code", -1); error.addProperty("message", e.getMessage() == null ? "Unknown error" : e.getMessage()); response.add("error", error); + } finally { + CURRENT_SESSION.remove(); } return GSON.toJson(response); @@ -716,27 +1462,39 @@ public final class EtcdAgent { } private static KV requireKv() { + KV kv = sessionState().kv; if (kv == null) { throw new IllegalStateException("Not connected"); } return kv; } + private static EtcdSessionState sessionState() { + EtcdSessionState state = CURRENT_SESSION.get(); + if (state == null) throw new IllegalStateException("No active etcd Agent session"); + return state; + } + private static Client requireClient() { + Client client = sessionState().client; if (client == null) throw new IllegalStateException("Not connected"); return client; } private static void closeClient() { - if (kv != null) { - kv.close(); - kv = null; + EtcdSessionState state = sessionState(); + for (EtcdWatchState watch : state.watches.values()) watch.close(); + state.watches.clear(); + state.knownLeases.clear(); + if (state.kv != null) { + state.kv.close(); + state.kv = null; } - if (client != null) { - client.close(); - client = null; + if (state.client != null) { + state.client.close(); + state.client = null; } - connectedEndpoints = Collections.emptyList(); + state.connectedEndpoints = Collections.emptyList(); } private static ByteSequence byteSequence(String value) { @@ -913,19 +1671,7 @@ public final class EtcdAgent { } public static void main(String[] args) throws Exception { - System.out.println("{\"ready\":true}"); - System.out.flush(); - - BufferedReader reader = new BufferedReader(new InputStreamReader(System.in)); - while (true) { - String line = reader.readLine(); - if (line == null) { - break; - } - - System.out.println(handleRequest(line)); - System.out.flush(); - } + MultiSessionJsonRpcServer.forSessionHandlers(EtcdSessionHandler::new).run(); } private static final class HandshakeResult { @@ -939,4 +1685,74 @@ public final class EtcdAgent { this.capabilities = capabilities; } } + + static final class EtcdSessionState { + private Client client; + private KV kv; + private List connectedEndpoints = Collections.emptyList(); + private final Map watches = new ConcurrentHashMap<>(); + private final Set knownLeases = ConcurrentHashMap.newKeySet(); + private long watchBufferedBytes; + + private synchronized boolean reserveWatchBuffer(long bytes) { + if (bytes < 0 || watchBufferedBytes + bytes > MAX_SESSION_WATCH_BUFFER_BYTES) return false; + watchBufferedBytes += bytes; + return true; + } + + private synchronized void releaseWatchBuffer(long bytes) { + watchBufferedBytes = Math.max(0, watchBufferedBytes - Math.max(0, bytes)); + } + + synchronized long watchBufferedBytes() { + return watchBufferedBytes; + } + + void addWatch(String watchId, EtcdWatchState watch) { + watches.put(watchId, watch); + } + + int watchCount() { + return watches.size(); + } + } + + private static final class EtcdSessionHandler implements SessionRpcHandler { + private final EtcdSessionState state = new EtcdSessionState(); + + @Override + public Object handshake() { + return handshakeResult(); + } + + @Override + public Object connect(JsonObject params) throws Exception { + return withSession(() -> EtcdAgent.connect(params)); + } + + @Override + public Object handle(String method, JsonObject params) throws Exception { + return withSession(() -> dispatch(method, params)); + } + + @Override + public void close() { + try { + withSession(() -> { + closeClient(); + return null; + }); + } catch (Exception ignored) { + } + } + + private T withSession(Callable task) throws Exception { + CURRENT_SESSION.set(state); + try { + return task.call(); + } finally { + CURRENT_SESSION.remove(); + } + } + } } diff --git a/agents/drivers/etcd/src/test/java/com/dbx/agent/etcd/EtcdAgentTest.java b/agents/drivers/etcd/src/test/java/com/dbx/agent/etcd/EtcdAgentTest.java index 6b41fd846..d8cec636b 100644 --- a/agents/drivers/etcd/src/test/java/com/dbx/agent/etcd/EtcdAgentTest.java +++ b/agents/drivers/etcd/src/test/java/com/dbx/agent/etcd/EtcdAgentTest.java @@ -15,9 +15,11 @@ import io.etcd.jetcd.kv.TxnResponse; import java.lang.reflect.Proxy; import java.nio.charset.StandardCharsets; import java.util.ArrayDeque; +import java.util.ArrayList; import java.util.Arrays; import java.util.Deque; import java.util.List; +import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; @@ -34,13 +36,19 @@ final class EtcdAgentTest { JsonObject result = JsonParser.parseString(response).getAsJsonObject().getAsJsonObject("result"); - Assertions.assertEquals(1, result.get("protocolVersion").getAsInt()); + Assertions.assertEquals(2, result.get("protocolVersion").getAsInt()); Assertions.assertTrue(result.getAsJsonArray("capabilities").contains(JsonParser.parseString("\"kv\""))); Assertions.assertTrue(result.getAsJsonArray("capabilities").contains(JsonParser.parseString("\"kv_ttl\""))); Assertions.assertTrue(result.getAsJsonArray("capabilities").contains(JsonParser.parseString("\"kv_cas\""))); Assertions.assertTrue(result.getAsJsonArray("capabilities").contains(JsonParser.parseString("\"kv_list_values\""))); Assertions.assertTrue(result.getAsJsonArray("capabilities").contains(JsonParser.parseString("\"kv_status\""))); Assertions.assertTrue(result.getAsJsonArray("capabilities").contains(JsonParser.parseString("\"kv_history\""))); + Assertions.assertTrue(result.getAsJsonArray("capabilities").contains(JsonParser.parseString("\"etcd_compaction\""))); + Assertions.assertTrue(result.getAsJsonArray("capabilities").contains(JsonParser.parseString("\"etcd_defrag\""))); + Assertions.assertTrue(result.getAsJsonArray("capabilities").contains(JsonParser.parseString("\"etcd_watch\""))); + Assertions.assertTrue(result.getAsJsonArray("capabilities").contains(JsonParser.parseString("\"etcd_lease\""))); + Assertions.assertTrue(result.getAsJsonArray("capabilities").contains(JsonParser.parseString("\"etcd_auth\""))); + Assertions.assertTrue(result.getAsJsonArray("capabilities").contains(JsonParser.parseString("\"multi_session\""))); Assertions.assertTrue(result.getAsJsonArray("capabilities").contains(JsonParser.parseString("\"connect\""))); } @@ -65,6 +73,34 @@ final class EtcdAgentTest { Assertions.assertEquals(List.of("http://127.0.0.1:2379"), EtcdAgent.endpoints(connection)); } + @Test + void connectTimeoutUsesConfiguredValueAndSafeBounds() { + Assertions.assertEquals( + 45, + EtcdAgent.connectTimeoutSeconds(JsonParser.parseString("{\"connect_timeout_secs\":45}").getAsJsonObject()) + ); + Assertions.assertEquals( + 1, + EtcdAgent.connectTimeoutSeconds(JsonParser.parseString("{\"connect_timeout_secs\":0}").getAsJsonObject()) + ); + Assertions.assertEquals( + 300, + EtcdAgent.connectTimeoutSeconds(JsonParser.parseString("{\"connect_timeout_secs\":999}").getAsJsonObject()) + ); + } + + @Test + void validateConnectionRequiresAnActiveSession() { + String response = EtcdAgent.handleRequest( + "{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"validate_connection\",\"params\":{}}" + ); + + JsonObject error = JsonParser.parseString(response).getAsJsonObject().getAsJsonObject("error"); + + Assertions.assertEquals(-1, error.get("code").getAsInt()); + Assertions.assertEquals("Not connected", error.get("message").getAsString()); + } + @Test void kvMethodDispatchReturnsJsonRpcErrorWhenDisconnected() { String response = EtcdAgent.handleRequest( @@ -210,6 +246,118 @@ final class EtcdAgentTest { Assertions.assertEquals(123, revokedLeaseId.get()); } + @Test + void defragFailureMarksEveryRemainingEndpointAsNotExecuted() { + List> members = new ArrayList<>(); + + EtcdAgent.appendUnexecutedDefragMembers( + members, + List.of("http://follower-2:2379", "http://follower-3:2379", "http://leader:2379"), + "http://follower-2:2379" + ); + + Assertions.assertEquals( + List.of("http://follower-3:2379", "http://leader:2379"), + members.stream().map(member -> member.get("endpoint")).toList() + ); + Assertions.assertTrue(members.stream().allMatch(member -> "not_executed".equals(member.get("status")))); + } + + @Test + void leaseListResponseParserReadsEveryLeaseId() throws Exception { + byte[] response = new byte[] { + 0x12, 0x02, 0x08, 0x7b, + 0x12, 0x03, 0x08, (byte) 0xc8, 0x03 + }; + + Assertions.assertEquals(List.of(123L, 456L), EtcdAgent.leaseIdsFromResponse(response)); + } + + @Test + void leaseListPageUsesAStableCursorAndBoundsTheResult() { + List leaseIds = new ArrayList<>(); + for (long id = 1000; id >= 1; id--) leaseIds.add(id); + + List first = EtcdAgent.leasePageIds(leaseIds, null, 101); + List second = EtcdAgent.leasePageIds(leaseIds, first.get(99), 101); + + Assertions.assertEquals(101, first.size()); + Assertions.assertEquals(1L, first.get(0)); + Assertions.assertEquals(101L, first.get(100)); + Assertions.assertEquals(101L, second.get(0)); + Assertions.assertEquals(201L, second.get(100)); + } + + @Test + void watchBufferRejectsLargeValueAndPreviousValuePayloads() { + long payloadBytes = EtcdAgent.estimatedBufferedBytes(2 * 1024 * 1024) + + EtcdAgent.estimatedBufferedBytes(2 * 1024 * 1024); + + Assertions.assertTrue(payloadBytes > EtcdAgent.MAX_WATCH_BUFFER_BYTES); + } + + @Test + void watchOverflowPreservesBufferedPayloadAndReportsTerminalState() { + EtcdAgent.EtcdSessionState session = new EtcdAgent.EtcdSessionState(); + EtcdAgent.EtcdWatchState watch = new EtcdAgent.EtcdWatchState("watch-1", session); + List> events = List.of(Map.of("eventType", "put")); + + watch.append(1, events, 4L * 1024 * 1024); + Assertions.assertEquals(4L * 1024 * 1024, session.watchBufferedBytes()); + + watch.append(2, events, 5L * 1024 * 1024); + Map result = watch.poll(); + + Assertions.assertEquals(0, session.watchBufferedBytes()); + List batches = (List) result.get("batches"); + Assertions.assertEquals(1, batches.size()); + Assertions.assertEquals("1", ((Map) batches.get(0)).get("revision")); + Assertions.assertEquals("overflow", ((Map) result.get("terminal")).get("reason")); + } + + @Test + void watchPollReleasesTheSessionByteBudget() { + EtcdAgent.EtcdSessionState session = new EtcdAgent.EtcdSessionState(); + EtcdAgent.EtcdWatchState watch = new EtcdAgent.EtcdWatchState("watch-1", session); + + watch.append(1, List.of(Map.of("eventType", "put")), 1024); + Assertions.assertEquals(1024, session.watchBufferedBytes()); + + watch.poll(); + + Assertions.assertEquals(0, session.watchBufferedBytes()); + } + + @Test + void aggregateSessionBudgetTerminatesOnlyTheWatchThatExceedsIt() { + EtcdAgent.EtcdSessionState session = new EtcdAgent.EtcdSessionState(); + List> events = List.of(Map.of("eventType", "put")); + EtcdAgent.EtcdWatchState first = new EtcdAgent.EtcdWatchState("watch-1", session); + EtcdAgent.EtcdWatchState second = new EtcdAgent.EtcdWatchState("watch-2", session); + EtcdAgent.EtcdWatchState third = new EtcdAgent.EtcdWatchState("watch-3", session); + + first.append(1, events, EtcdAgent.MAX_WATCH_BUFFER_BYTES); + second.append(1, events, EtcdAgent.MAX_WATCH_BUFFER_BYTES); + third.append(1, events, 1); + + Assertions.assertEquals(EtcdAgent.MAX_SESSION_WATCH_BUFFER_BYTES, session.watchBufferedBytes()); + Assertions.assertEquals("overflow", ((Map) third.poll().get("terminal")).get("reason")); + Assertions.assertEquals(EtcdAgent.MAX_SESSION_WATCH_BUFFER_BYTES, session.watchBufferedBytes()); + } + + @Test + void terminalPollRemovesTheWatchSlot() { + EtcdAgent.EtcdSessionState session = new EtcdAgent.EtcdSessionState(); + EtcdAgent.EtcdWatchState watch = new EtcdAgent.EtcdWatchState("watch-1", session); + session.addWatch("watch-1", watch); + watch.overflow(); + + Map result = EtcdAgent.pollWatchState(session, "watch-1"); + + Assertions.assertEquals("overflow", ((Map) result.get("terminal")).get("reason")); + Assertions.assertEquals(0, session.watchCount()); + } + private static KV scriptedKv( AtomicInteger getCalls, AtomicInteger txnCalls, diff --git a/apps/desktop/src/components/etcd/EtcdAccessControl.vue b/apps/desktop/src/components/etcd/EtcdAccessControl.vue new file mode 100644 index 000000000..80a77415b --- /dev/null +++ b/apps/desktop/src/components/etcd/EtcdAccessControl.vue @@ -0,0 +1,768 @@ + + + diff --git a/apps/desktop/src/components/etcd/EtcdAdminConsole.vue b/apps/desktop/src/components/etcd/EtcdAdminConsole.vue new file mode 100644 index 000000000..2c2703d5f --- /dev/null +++ b/apps/desktop/src/components/etcd/EtcdAdminConsole.vue @@ -0,0 +1,915 @@ + + + diff --git a/apps/desktop/src/components/etcd/EtcdDashboard.vue b/apps/desktop/src/components/etcd/EtcdDashboard.vue index 9a6856e2b..df3c22d54 100644 --- a/apps/desktop/src/components/etcd/EtcdDashboard.vue +++ b/apps/desktop/src/components/etcd/EtcdDashboard.vue @@ -216,12 +216,12 @@ defineExpose({ refresh }); diff --git a/apps/desktop/src/components/etcd/EtcdKeyBrowser.vue b/apps/desktop/src/components/etcd/EtcdKeyBrowser.vue index 61fed498c..2c8b1204e 100644 --- a/apps/desktop/src/components/etcd/EtcdKeyBrowser.vue +++ b/apps/desktop/src/components/etcd/EtcdKeyBrowser.vue @@ -1,22 +1,28 @@