Improve nodetool enable{audit,fullquery}log

Patch by marcuse; reviewed by Dinesh Joshi and Mick Semb Wever for CASSANDRA-18550

CHANGES.txt

@@ -1,4 +1,5 @@
 4.0.10
+ * Improve nodetool enable{audit,fullquery}log (CASSANDRA-18550)
  * Report network cache info in nodetool (CASSANDRa-18400)
  * Partial compaction can resurrect deleted data (CASSANDRA-18507)
  * Allow internal address to change with reconnecting snitches (CASSANDRA-16718)

conf/cassandra.yaml

@@ -1360,6 +1360,8 @@ audit_logging_options:
     # max_log_size: 17179869184 # 16 GiB
     ## archive command is "/path/to/script.sh %path" where %path is replaced with the file being rolled:
     # archive_command:
+    ## note that enabling this allows anyone with JMX/nodetool access to run local shell commands as the user running cassandra
+    # allow_nodetool_archive_command: false
     # max_archive_retries: 10
 
 # validate tombstones on reads and compaction

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -3140,6 +3140,11 @@ public static FullQueryLoggerOptions getFullQueryLogOptions()
         return  conf.full_query_logging_options;
     }
 
+    public static void setFullQueryLogOptions(FullQueryLoggerOptions options)
+    {
+        conf.full_query_logging_options = options;
+    }
+
     public static boolean getBlockForPeersInRemoteDatacenters()
     {
         return conf.block_for_peers_in_remote_dcs;

src/java/org/apache/cassandra/service/StorageService.java

@@ -6081,6 +6081,8 @@ public void enableFullQueryLogger(String path, String rollCycle, Boolean blockin
         blocking = blocking != null ? blocking : fqlOptions.block;
         maxQueueWeight = maxQueueWeight != Integer.MIN_VALUE ? maxQueueWeight : fqlOptions.max_queue_weight;
         maxLogSize = maxLogSize != Long.MIN_VALUE ? maxLogSize : fqlOptions.max_log_size;
+        if (archiveCommand != null && !fqlOptions.allow_nodetool_archive_command)
+            throw new ConfigurationException("Can't enable full query log archiving via nodetool unless full_query_logging_options.allow_nodetool_archive_command is set to true");
         archiveCommand = archiveCommand != null ? archiveCommand : fqlOptions.archive_command;
         maxArchiveRetries = maxArchiveRetries != Integer.MIN_VALUE ? maxArchiveRetries : fqlOptions.max_archive_retries;
 

src/java/org/apache/cassandra/tools/nodetool/EnableFullQueryLog.java

@@ -42,7 +42,8 @@ public class EnableFullQueryLog extends NodeToolCmd
     private String path = null;
 
     @Option(title = "archive_command", name = {"--archive-command"}, description = "Command that will handle archiving rolled full query log files." +
-                                                                                   " Format is \"/path/to/script.sh %path\" where %path will be replaced with the file to archive")
+                                                                                   " Format is \"/path/to/script.sh %path\" where %path will be replaced with the file to archive" +
+                                                                                   " Enable this by setting the full_query_logging_options.allow_nodetool_archive_command: true in the config.")
     private String archiveCommand = null;
 
     @Option(title = "archive_retries", name = {"--max-archive-retries"}, description = "Max number of archive retries.")

src/java/org/apache/cassandra/utils/binlog/BinLogOptions.java

@@ -23,6 +23,13 @@
 public class BinLogOptions
 {
     public String archive_command = StringUtils.EMPTY;
+
+    /**
+     * enable if a user should be able to set the archive command via nodetool/jmx
+     *
+     * do not make this a hotprop.
+     */
+    public boolean allow_nodetool_archive_command = false;
     /**
      * How often to roll BinLog segments so they can potentially be reclaimed. Available options are:
      * MINUTELY, HOURLY, DAILY, LARGE_DAILY, XLARGE_DAILY, HUGE_DAILY.

test/unit/org/apache/cassandra/fql/FullQueryLoggerTest.java

@@ -32,7 +32,6 @@
 
 import org.apache.commons.lang3.StringUtils;
 import org.junit.After;
-import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
@@ -44,15 +43,18 @@
 import net.openhft.chronicle.wire.ValueIn;
 import net.openhft.chronicle.wire.WireOut;
 import org.apache.cassandra.Util;
+import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.cql3.CQLTester;
 import org.apache.cassandra.cql3.QueryOptions;
 import org.apache.cassandra.cql3.statements.BatchStatement;
+import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.fql.FullQueryLogger.Query;
 import org.apache.cassandra.fql.FullQueryLogger.Batch;
 import org.apache.cassandra.cql3.statements.BatchStatement.Type;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.QueryState;
+import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.transport.ProtocolVersion;
 import org.apache.cassandra.utils.ObjectSizes;
 import org.apache.cassandra.utils.binlog.BinLogTest;
@@ -74,6 +76,7 @@
 import static org.apache.cassandra.fql.FullQueryLogger.TYPE;
 import static org.apache.cassandra.fql.FullQueryLogger.VALUES;
 import static org.apache.cassandra.fql.FullQueryLogger.VERSION;
+import static org.junit.Assert.fail;
 
 public class FullQueryLoggerTest extends CQLTester
 {
@@ -670,6 +673,29 @@ public void testLogQueryNegativeTime() throws Exception
         logQuery("", QueryOptions.DEFAULT, queryState(), -1);
     }
 
+    @Test
+    public void testJMXArchiveCommand()
+    {
+        FullQueryLoggerOptions options = new FullQueryLoggerOptions();
+
+        try
+        {
+            StorageService.instance.enableFullQueryLogger(options.log_dir, options.roll_cycle, false, 1000, 1000, "/xyz/not/null", 0);
+            fail("not allowed");
+        }
+        catch (ConfigurationException e)
+        {
+            assertTrue(e.getMessage().contains("Can't enable full query log archiving via nodetool"));
+        }
+
+        options.archive_command = "/xyz/not/null";
+        options.log_dir = "/tmp/abc";
+        DatabaseDescriptor.setFullQueryLogOptions(options);
+        StorageService.instance.enableFullQueryLogger(options.log_dir, options.roll_cycle, false, 1000, 1000, null, 0);
+        assertTrue(FullQueryLogger.instance.isEnabled());
+        assertEquals("/xyz/not/null", FullQueryLogger.instance.getFullQueryLoggerOptions().archive_command);
+    }
+
     private static void compareQueryOptions(QueryOptions a, QueryOptions b)
     {
         assertEquals(a.getClass(), b.getClass());