Allow configuration of the size of large partitions that get logged by Cassandra during compaction.

priam/src/main/java/com/netflix/priam/IConfiguration.java

@@ -501,6 +501,12 @@ public interface IConfiguration {
 
     public int getIndexInterval();
 
+    /*
+     * @return the warning threshold in MB's for large partitions encountered during compaction.
+     * Default value of 100 is used (default from cassandra.yaml)
+     */
+    public int getCompactionLargePartitionWarnThresholdInMB();
+
     public String getExtraConfigParams();
 
     public String getCassYamlVal(String priamKey);

priam/src/main/java/com/netflix/priam/defaultimpl/PriamConfiguration.java

@@ -1074,7 +1074,12 @@ public int getRpcMaxThreads() {
     public int getIndexInterval() {
     	return config.get(CONFIG_INDEX_INTERVAL, DEFAULT_INDEX_INTERVAL);
     }
-
+
+    @Override
+    public int getCompactionLargePartitionWarnThresholdInMB() {
+        return config.get(PRIAM_PRE + ".compaction.large.partition.warn.threshold", 100);
+    }
+
     public String getExtraConfigParams() {
     	return config.get(CONFIG_EXTRA_PARAMS);
     }

priam/src/main/java/com/netflix/priam/defaultimpl/StandardTuner.java

@@ -113,6 +113,7 @@ public void writeAllProperties(String yamlLocation, String hostname, String seed
         map.put("streaming_socket_timeout_in_ms", config.getStreamingSocketTimeoutInMS());
 
         map.put("memtable_cleanup_threshold", config.getMemtableCleanupThreshold());
+        map.put("compaction_large_partition_warning_threshold_mb", config.getCompactionLargePartitionWarnThresholdInMB());
 
         List<?> seedp = (List) map.get("seed_provider");
         Map<String, String> m = (Map<String, String>) seedp.get(0);

priam/src/main/java/com/netflix/priam/dse/AuditLogTunerLog4J.java

@@ -0,0 +1,149 @@
+/*
+ * Copyright 2016 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package com.netflix.priam.dse;
+
+import com.google.common.base.Joiner;
+import com.google.common.io.Files;
+import com.google.inject.Inject;
+import com.netflix.priam.IConfiguration;
+import org.apache.cassandra.io.util.FileUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.nio.charset.Charset;
+import java.util.List;
+
+/**
+ * Dse tuner for audit log via log4j.
+ * Use this instead of AuditLogTunerYaml if you are on DSE version 3.x.
+ * Created by aagrawal on 8/8/17.
+ */
+public class AuditLogTunerLog4J implements IAuditLogTuner {
+
+    private IConfiguration config;
+    private IDseConfiguration dseConfig;
+    protected static final String AUDIT_LOG_ADDITIVE_ENTRY = "log4j.additivity.DataAudit";
+    protected static final String AUDIT_LOG_FILE = "/conf/log4j-server.properties";
+    protected static final String PRIMARY_AUDIT_LOG_ENTRY = "log4j.logger.DataAudit";
+    private static final Logger logger = LoggerFactory.getLogger(AuditLogTunerLog4J.class);
+
+    @Inject
+    public AuditLogTunerLog4J(IConfiguration config, IDseConfiguration dseConfig)
+    {
+        this.config = config;
+        this.dseConfig = dseConfig;
+    }
+
+    /**
+     * Note: supporting the direct hacking of a log4j props file is far from elegant,
+     * but seems less odious than other solutions I've come up with.
+     * Operates under the assumption that the only people mucking with the audit log
+     * entries in the value are DataStax themselves and this program, and that the original
+     * property names are somehow still preserved. Otherwise, YMMV.
+     */
+    public void tuneAuditLog()
+    {
+        BufferedWriter writer = null;
+        try
+        {
+            final File srcFile = new File(config.getCassHome() + AUDIT_LOG_FILE);
+            final List<String> lines = Files.readLines(srcFile, Charset.defaultCharset());
+            final File backupFile = new File(config.getCassHome() + AUDIT_LOG_FILE + "." + System.currentTimeMillis());
+            Files.move(srcFile, backupFile);
+            writer = Files.newWriter(srcFile, Charset.defaultCharset());
+
+            String loggerPrefix = "log4j.appender.";
+            try
+            {
+                loggerPrefix += findAuditLoggerName(lines);
+            }
+            catch (IllegalStateException ise)
+            {
+                logger.warn(String.format("cannot locate %s property, will ignore any audit log updating", PRIMARY_AUDIT_LOG_ENTRY));
+                return;
+            }
+
+            for(String line : lines)
+            {
+                if(line.contains(loggerPrefix) || line.contains(PRIMARY_AUDIT_LOG_ENTRY) || line.contains(AUDIT_LOG_ADDITIVE_ENTRY))
+                {
+                    if(dseConfig.isAuditLogEnabled())
+                    {
+                        //first, check to see if we need to uncomment the line
+                        while(line.startsWith("#"))
+                        {
+                            line = line.substring(1);
+                        }
+
+                        //next, check if we need to change the prop's value
+                        if(line.contains("ActiveCategories"))
+                        {
+                            final String cats = Joiner.on(",").join(dseConfig.getAuditLogCategories());
+                            line = line.substring(0, line.indexOf("=") + 1).concat(cats);
+                        }
+                        else if(line.contains("ExemptKeyspaces"))
+                        {
+                            line = line.substring(0, line.indexOf("=") + 1).concat(dseConfig.getAuditLogExemptKeyspaces());
+                        }
+                    }
+                    else
+                    {
+                        if(line.startsWith("#"))
+                        {
+                            //make sure there's only one # at the beginning of the line
+                            while(line.charAt(1) == '#')
+                                line = line.substring(1);
+                        }
+                        else
+                        {
+                            line = "#" + line;
+                        }
+                    }
+                }
+                writer.append(line);
+                writer.newLine();
+            }
+        }
+        catch (Exception e)
+        {
+            e.printStackTrace();
+            throw new RuntimeException("Unable to read " + AUDIT_LOG_FILE, e);
+
+        }
+        finally
+        {
+            FileUtils.closeQuietly(writer);
+        }
+    }
+
+
+    private final String findAuditLoggerName(List<String> lines) throws IllegalStateException
+    {
+        for(final String l : lines)
+        {
+            if(l.contains(PRIMARY_AUDIT_LOG_ENTRY))
+            {
+                final String[] valTokens = l.split(",");
+                return valTokens[valTokens.length -1].trim();
+            }
+        }
+        throw new IllegalStateException();
+    }
+}

priam/src/main/java/com/netflix/priam/dse/AuditLogTunerYaml.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2016 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package com.netflix.priam.dse;
+
+import com.google.inject.Inject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.yaml.snakeyaml.DumperOptions;
+import org.yaml.snakeyaml.Yaml;
+
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * Dse tuner for audit log via YAML. Use this for DSE version 4.x
+ * Created by aagrawal on 8/8/17.
+ */
+public class AuditLogTunerYaml implements IAuditLogTuner {
+
+    private IDseConfiguration dseConfig;
+    protected static final String AUDIT_LOG_DSE_ENTRY = "audit_logging_options";
+    private static final Logger logger = LoggerFactory.getLogger(AuditLogTunerYaml.class);
+
+    @Inject
+    public AuditLogTunerYaml(IDseConfiguration dseConfig)
+    {
+        this.dseConfig = dseConfig;
+    }
+
+    public void tuneAuditLog()
+    {
+        DumperOptions options = new DumperOptions();
+        options.setDefaultFlowStyle(DumperOptions.FlowStyle.BLOCK);
+        Yaml yaml = new Yaml(options);
+        String dseYaml = dseConfig.getDseYamlLocation();
+        try {
+            Map<String, Object> map = (Map<String, Object>) yaml.load(new FileInputStream(dseYaml));
+
+            if (map.containsKey(AUDIT_LOG_DSE_ENTRY)) {
+                Boolean isEnabled = (Boolean) ((Map<String, Object>) map.get(AUDIT_LOG_DSE_ENTRY)).get("enabled");
+
+                // Enable/disable audit logging (need this in addition to log4j-server.properties settings)
+                if (dseConfig.isAuditLogEnabled()) {
+                    if (!isEnabled) {
+                        ((Map<String, Object>) map.get(AUDIT_LOG_DSE_ENTRY)).put("enabled", true);
+                    }
+                } else if (isEnabled) {
+                    ((Map<String, Object>) map.get(AUDIT_LOG_DSE_ENTRY)).put("enabled", false);
+                }
+            }
+
+            logger.info("Updating dse-yaml:\n" + yaml.dump(map));
+            yaml.dump(map, new FileWriter(dseYaml));
+        }catch (FileNotFoundException fileNotFound)
+        {
+            logger.error(String.format("FileNotFound while trying to read yaml audit log for tuning: {}", dseYaml));
+        }
+        catch (IOException e)
+        {
+            logger.error(String.format("IOException while trying to write yaml file for audit log tuning: {}", dseYaml));
+        }
+    }
+}