CEP-38: Management API MVP Implementation [CASSANDRA-19476]

bin/nodetool

@@ -43,30 +43,51 @@ if [ -z "$CASSANDRA_CONF" -o -z "$CLASSPATH" ]; then
     exit 1
 fi
 
-JMX_PORT=""
-
-# Try to parse port from configure_jmx method call, when not commented out
-if [ -f "$CASSANDRA_CONF/cassandra-env.sh" ]; then
-    jmx_method_call=$(grep "^configure_jmx \+[0-9]\+$" "$CASSANDRA_CONF/cassandra-env.sh")
-    if [ ! "x${jmx_method_call}" = "x" ]; then
-        JMX_PORT=$(echo "${jmx_method_call}" | tr -s " " | cut -d " " -f2)
+# Check if protocol is cql (from environment variable)
+PROTOCOL_IS_CQL=""
+if [ "x$CASSANDRA_CLI_EXECUTION_PROTOCOL" != "x" ]; then
+    protocol_lower=$(echo "$CASSANDRA_CLI_EXECUTION_PROTOCOL" | tr '[:upper:]' '[:lower:]')
+    if [ "$protocol_lower" = "cql" ]; then
+        PROTOCOL_IS_CQL="true"
     fi
 fi
 
-# In case JMX_PORT is not set (when configure_jmx in cassandra-env.sh is commented out),
-# try to parse it from cassandra.yaml.
-if [ "x$JMX_PORT" = "x" ]; then
+# Check for cql protocol in system properties to determine if we should parse CQL or JMX port
+for arg in "$@"; do
+    case "$arg" in
+        -Dcassandra.cli.execution.protocol=cql|-Dcassandra.cli.execution.protocol=CQL)
+            PROTOCOL_IS_CQL="true"
+            break
+            ;;
+    esac
+done
+
+# Parse port from config files based on protocol
+CONNECTION_PORT=""
+if [ "$PROTOCOL_IS_CQL" = "true" ]; then
+    # For CQL, parse native_transport_management_port from cassandra.yaml
     if [ -f "$CASSANDRA_CONF/cassandra.yaml" ]; then
-        JMX_PORT=$(grep jmx_port "$CASSANDRA_CONF/cassandra.yaml" | cut -d ':' -f 2 | tr -d '[[:space:]]')
+        CONNECTION_PORT=$(grep native_transport_management_port "$CASSANDRA_CONF/cassandra.yaml" | cut -d ':' -f 2 | tr -d '[[:space:]]')
+    fi
+else
+    # For JMX, parse port from configure_jmx method call, when not commented out
+    if [ -f "$CASSANDRA_CONF/cassandra-env.sh" ]; then
+        jmx_method_call=$(grep "^configure_jmx \+[0-9]\+$" "$CASSANDRA_CONF/cassandra-env.sh")
+        if [ ! "x${jmx_method_call}" = "x" ]; then
+            CONNECTION_PORT=$(echo "${jmx_method_call}" | tr -s " " | cut -d " " -f2)
+        fi
     fi
-fi
 
-# If, by any chance, it is not there either, set it to default.
-if [ "x$JMX_PORT" = "x" ]; then
-    JMX_PORT=7199
+    # In case CONNECTION_PORT is not set (when configure_jmx in cassandra-env.sh is commented out),
+    # try to parse it from cassandra.yaml.
+    if [ "x$CONNECTION_PORT" = "x" ]; then
+        if [ -f "$CASSANDRA_CONF/cassandra.yaml" ]; then
+            CONNECTION_PORT=$(grep jmx_port "$CASSANDRA_CONF/cassandra.yaml" | cut -d ':' -f 2 | tr -d '[[:space:]]')
+        fi
+    fi
 fi
 
-# JMX Port passed via cmd line args (-p 9999 / --port 9999 / --port=9999)
+# Connection Port passed via cmd line args (-p 9999 / --port 9999 / --port=9999)
 # should override the value from cassandra-env.sh
 ARGS=""
 JVM_ARGS=""
@@ -76,19 +97,19 @@ do
   if [ ! $1 ]; then break; fi
   case $1 in
     -p)
-      JMX_PORT=$2
+      CONNECTION_PORT=$2
       shift
       ;;
     --port=*)
-      JMX_PORT=$(echo $1 | cut -d '=' -f 2)
+      CONNECTION_PORT=$(echo $1 | cut -d '=' -f 2)
       ;;
     --port)
-      JMX_PORT=$2
+      CONNECTION_PORT=$2
       shift
       ;;
     --ssl)
       if [ -f $SSL_FILE ]
-      then 
+      then
           SSL_ARGS=$(cat $SSL_FILE | tr '\n' ' ')
       fi
       JVM_ARGS="$JVM_ARGS -Dssl.enable=true $SSL_ARGS"
@@ -108,6 +129,14 @@ do
   shift
 done
 
+if [ "x$CONNECTION_PORT" = "x" ]; then
+    if [ "$PROTOCOL_IS_CQL" = "true" ]; then
+        CONNECTION_PORT=11211
+    else
+        CONNECTION_PORT=7199
+    fi
+fi
+
 if [ "x$MAX_HEAP_SIZE" = "x" ]; then
     MAX_HEAP_SIZE="128m"
 fi
@@ -123,7 +152,7 @@ CMD=$(echo "$JAVA" $JAVA_AGENT -ea -cp "$CLASSPATH" $JVM_OPTS -Xmx$MAX_HEAP_SIZE
             -Dcassandra.logdir="$CASSANDRA_LOG_DIR" \
             -Dlogback.configurationFile=logback-tools.xml \
             $JVM_ARGS \
-            org.apache.cassandra.tools.NodeTool -p $JMX_PORT $ARGS)
+            org.apache.cassandra.tools.NodeTool -p $CONNECTION_PORT $ARGS)
 
 if [ "x$ARCHIVE_COMMAND" != "x" ]
 then

conf/cassandra.yaml

@@ -1110,6 +1110,56 @@ rpc_address: localhost
 # enable or disable keepalive on rpc/native connections
 rpc_keepalive: true
 
+# Whether to start the native management transport server.
+# The address on which the native management transport is bound can be configured
+# using rpc_management_address or rpc_management_interface. If neither is set,
+# it defaults to rpc_address (or rpc_interface if that was used).
+start_native_transport_management: false
+
+# The address or interface to bind the native management transport server to.
+# This allows you to bind management operations to a different network interface
+# than regular client connections, providing better security isolation.
+#
+# Set rpc_management_address OR rpc_management_interface, not both.
+#
+# If neither rpc_management_address nor rpc_management_interface is set,
+# the management transport will use the same address as the regular native
+# transport (rpc_address or rpc_interface).
+#
+# For security reasons, you should not expose this port to the internet.
+# Firewall it if needed. Consider binding to a management network interface
+# that is not accessible from public networks.
+# rpc_management_address: localhost
+
+# Set rpc_management_address OR rpc_management_interface, not both. Interfaces
+# must correspond to a single address, IP aliasing is not supported.
+#
+# This allows you to bind the management transport to a specific network
+# interface by name (e.g., eth1, eth0). The IP address will be automatically
+# resolved from the interface. This is useful when the interface IP may change
+# (e.g., DHCP) or when you want to bind to a specific interface without
+# hardcoding the IP address.
+# rpc_management_interface: eth1
+
+# If you choose to specify the management interface by name and the interface
+# has both an ipv4 and an ipv6 address, you can specify which should be chosen
+# using rpc_management_interface_prefer_ipv6. If false the first ipv4 address
+# will be used. If true the first ipv6 address will be used. Defaults to false
+# preferring ipv4. If there is only one address it will be selected regardless
+# of ipv4/ipv6.
+# rpc_management_interface_prefer_ipv6: false
+
+# Port for the management CQL native transport to listen for clients on.
+# For security reasons, you should not expose this port to the internet.
+# Firewall it if needed.
+native_transport_management_port: 11211
+# The maximum threads for handling management requests are set
+# separately from regular requests to allow better isolation and
+# prioritization of management operations. This is important to
+# ensure that management operations can proceed even under a high
+# load of regular client requests. Defaults to 2.
+# native_transport_management_max_threads: 2
+
 # Uncomment to set socket buffer size for internode communication
 # Note that when setting this, the buffer size is limited by net.core.wmem_max
 # and when not setting it it is defined by net.ipv4.tcp_wmem

conf/cassandra_latest.yaml

@@ -1099,6 +1099,56 @@ rpc_address: localhost
 # enable or disable keepalive on rpc/native connections
 rpc_keepalive: true
 
+# Whether to start the native management transport server.
+# The address on which the native management transport is bound can be configured
+# using rpc_management_address or rpc_management_interface. If neither is set,
+# it defaults to rpc_address (or rpc_interface if that was used).
+start_native_transport_management: false
+
+# The address or interface to bind the native management transport server to.
+# This allows you to bind management operations to a different network interface
+# than regular client connections, providing better security isolation.
+#
+# Set rpc_management_address OR rpc_management_interface, not both.
+#
+# If neither rpc_management_address nor rpc_management_interface is set,
+# the management transport will use the same address as the regular native
+# transport (rpc_address or rpc_interface).
+#
+# For security reasons, you should not expose this port to the internet.
+# Firewall it if needed. Consider binding to a management network interface
+# that is not accessible from public networks.
+# rpc_management_address: localhost
+
+# Set rpc_management_address OR rpc_management_interface, not both. Interfaces
+# must correspond to a single address, IP aliasing is not supported.
+#
+# This allows you to bind the management transport to a specific network
+# interface by name (e.g., eth1, eth0). The IP address will be automatically
+# resolved from the interface. This is useful when the interface IP may change
+# (e.g., DHCP) or when you want to bind to a specific interface without
+# hardcoding the IP address.
+# rpc_management_interface: eth1
+
+# If you choose to specify the management interface by name and the interface
+# has both an ipv4 and an ipv6 address, you can specify which should be chosen
+# using rpc_management_interface_prefer_ipv6. If false the first ipv4 address
+# will be used. If true the first ipv6 address will be used. Defaults to false
+# preferring ipv4. If there is only one address it will be selected regardless
+# of ipv4/ipv6.
+# rpc_management_interface_prefer_ipv6: false
+
+# Port for the management CQL native transport to listen for clients on.
+# For security reasons, you should not expose this port to the internet.
+# Firewall it if needed.
+native_transport_management_port: 11211
+# The maximum threads for handling management requests are set
+# separately from regular requests to allow better isolation and
+# prioritization of management operations. This is important to
+# ensure that management operations can proceed even under a high
+# load of regular client requests. Defaults to 2.
+# native_transport_management_max_threads: 2
+
 # Uncomment to set socket buffer size for internode communication
 # Note that when setting this, the buffer size is limited by net.core.wmem_max
 # and when not setting it it is defined by net.ipv4.tcp_wmem

doc/native_protocol_v5.spec

@@ -1419,6 +1419,15 @@ Table of Contents
                            acknowledged the request.
                 <blockfor> is an [int] representing the number of replicas whose
                            acknowledgement is required to achieve <cl>.
+    0x1800    COMMAND_FAILED: An exception occurred during command execution. This error is returned
+              when a command executed via the native management interface fails during execution.
+              The exception may or may not include more detail in the accompanying error message.
+              The rest of the ERROR message body will be
+                <execution_id>
+              where:
+                <execution_id> is a [uuid] representing the unique identifier for the command
+                              execution that failed. This identifier can be used to correlate the
+                              error with command execution logs on the server.
 
     0x2000    Syntax_error: The submitted query has a syntax error.
     0x2100    Unauthorized: The logged user doesn't have the right to perform

src/antlr/Lexer.g

@@ -148,6 +148,7 @@ K_MODIFY:      M O D I F Y;
 K_AUTHORIZE:   A U T H O R I Z E;
 K_DESCRIBE:    D E S C R I B E;
 K_EXECUTE:     E X E C U T E;
+K_COMMAND:     C O M M A N D;
 K_NORECURSIVE: N O R E C U R S I V E;
 K_MBEAN:       M B E A N;
 K_MBEANS:      M B E A N S;

src/antlr/Parser.g

@@ -294,6 +294,7 @@ cqlStatement returns [CQLStatement.Raw stmt]
     | st55=securityLabelOnUserTypeStatement    { $stmt = st55; }
     | st56=commentOnUserTypeFieldStatement     { $stmt = st56; }
     | st57=securityLabelOnUserTypeFieldStatement   { $stmt = st57; }
+    | st58=executeCommandStatement             { $stmt = st58; }
     ;
 
 /*
@@ -1370,6 +1371,44 @@ securityLabelOnUserTypeFieldStatement returns [SecurityLabelOnUserTypeFieldState
       { $stmt = new SecurityLabelOnUserTypeFieldStatement.Raw($typeFieldRef.typeName, $typeFieldRef.field, label != null ? $label.text : null, provider); }
     ;
 
+/**
+ * COMMAND <commandName> [WITH key1 = value1 AND key2 = value2];
+ */
+executeCommandStatement returns [ExecuteCommandStatement.Raw stmt]
+    @init {
+        stmtBegins();
+        java.util.Map<String, Object> args = new java.util.LinkedHashMap<>();
+    }
+    : K_COMMAND cmdName=noncol_ident 
+      (K_WITH commandProperties[args])?
+      {
+          $stmt = new ExecuteCommandStatement.Raw(cmdName.toString(), args);
+      }
+    ;
+
+commandProperties[java.util.Map<String, Object> args]
+    : commandProperty[args] (K_AND commandProperty[args])*
+    ;
+
+commandProperty[java.util.Map<String, Object> args]
+    : k=noncol_ident '=' v=commandPropertyValue
+      {
+          String key = k.toString();
+          Object value = v;
+          if (args.put(key, value) != null)
+          {
+              addRecognitionError("Duplicate argument: " + key);
+          }
+      }
+    ;
+
+commandPropertyValue returns [Object value]
+    : s=STRING_LITERAL { $value = $s.text; }
+    | i=INTEGER { $value = $i.text; }
+    | b=BOOLEAN { $value = $b.text; }
+    | l=listLiteral { $value = $l.value; }
+    ;
+
 /**
  * DROP TABLE [IF EXISTS] <table>;
  */
@@ -1980,7 +2019,7 @@ collectionLiteral returns [Term.Raw value]
 
 listLiteral returns [Term.Raw value]
     @init {List<Term.Raw> l = new ArrayList<Term.Raw>();}
-    @after {$value = new ArrayLiteral(l);}
+@after {$value = new ArrayLiteral(l);}
     : '[' ( t1=term { l.add(t1); } ( ',' tn=term { l.add(tn); } )* )? ']' { $value = new ArrayLiteral(l); }
     ;
 
@@ -2481,5 +2520,6 @@ basic_unreserved_keyword returns [String str]
         | K_LABELS
         | K_FIELD
         | K_COLUMN
+        | K_COMMAND
         ) { $str = $k.text; }
     ;

src/java/org/apache/cassandra/audit/AuditLogEntryType.java

@@ -82,7 +82,8 @@ public enum AuditLogEntryType
     UNAUTHORIZED_ATTEMPT(AuditLogEntryCategory.AUTH),
     LOGIN_SUCCESS(AuditLogEntryCategory.AUTH),
     LIST_SUPERUSERS(AuditLogEntryCategory.DCL),
-    JMX(AuditLogEntryCategory.JMX);
+    JMX(AuditLogEntryCategory.JMX),
+    EXECUTE_COMMAND(AuditLogEntryCategory.OTHER);
 
     private final AuditLogEntryCategory category;
 

src/java/org/apache/cassandra/auth/AbstractCIDRAuthorizer.java

@@ -28,8 +28,8 @@
  */
 public abstract class AbstractCIDRAuthorizer implements ICIDRAuthorizer
 {
-    protected static CIDRPermissionsManager cidrPermissionsManager;
-    protected static CIDRGroupsMappingManager cidrGroupsMappingManager;
+    public static CIDRPermissionsManager cidrPermissionsManager;
+    public static CIDRGroupsMappingManager cidrGroupsMappingManager;
 
     protected static CIDRAuthorizerMetrics cidrAuthorizerMetrics;
 

src/java/org/apache/cassandra/auth/AuthCache.java

@@ -45,6 +45,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.cassandra.auth.jmx.AuthorizationProxy;
 import org.apache.cassandra.cache.UnweightedCacheSize;
 import org.apache.cassandra.concurrent.ExecutorPlus;
 import org.apache.cassandra.concurrent.ScheduledExecutors;
@@ -468,6 +469,36 @@ public int entries()
         return Ints.checkedCast(getEstimatedSize());
     }
 
+    /**
+     * Accepts a visitor for this cache instance. Subclasses should override this method to dispatch
+     * to the appropriate visitor method based on the specific MBean interface they implement.
+     * @param visitor the visitor to accept
+     */
+    public void accept(MBeanVisitor visitor)
+    {
+        visitor.visit(this);
+    }
+
+    /**
+     * Visitor interface for processing auth cache MBeans.
+     * Allows type-safe iteration over different cache types without instanceof checks.
+     */
+    public interface MBeanVisitor
+    {
+        /** Visits a credentials cache MBean. */
+        default void visitCredentials(PasswordAuthenticator.CredentialsCacheMBean cache) {}
+        /** Visits a JMX permissions cache MBean. */
+        default void visitJmxPermissions(AuthorizationProxy.JmxPermissionsCacheMBean cache) {}
+        /** Visits a permissions cache MBean. */
+        default void visitPermissions(PermissionsCacheMBean cache) {}
+        /** Visits a network permissions cache MBean. */
+        default void visitNetwork(NetworkPermissionsCacheMBean cache) {}
+        /** Visits a roles cache MBean. */
+        default void visitRoles(RolesCacheMBean cache) {}
+        /** Visits a generic auth cache (fallback for caches that don't implement specific MBean interfaces).*/
+        default void visit(AuthCacheMBean cache) {}
+    }
+
     private class MetricsUpdater implements StatsCounter
     {
         @Override