natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove compaction Severity from DynamicEndpointSnitch 

patch by jbellis; reviewed by Jeremiah Jordan for CASSANDRA-11738
This commit is contained in:
Jonathan Ellis 2016-07-22 17:20:29 -05:00
parent a8f6a69452
commit 286f6a1435
7 changed files with 46 additions and 221 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGES.txt
View File

@ -12,6 +12,8 @@
* Remove DatabaseDescriptor dependency from SegmentedFile (CASSANDRA-11580)
* Add supplied username to authentication error messages (CASSANDRA-12076)
* Remove pre-startup check for open JMX port (CASSANDRA-12074)
* Remove compaction Severity from DynamicEndpointSnitch (CASSANDRA-11738)
3.9

20
src/java/org/apache/cassandra/db/compaction/CompactionInfo.java
View File

@ -138,8 +138,6 @@ public final class CompactionInfo implements Serializable
{
private volatile boolean stopRequested = false;
public abstract CompactionInfo getCompactionInfo();
double load = StorageMetrics.load.getCount();
double reportedSeverity = 0d;
public void stop()
{
@ -150,23 +148,5 @@ public final class CompactionInfo implements Serializable
{
return stopRequested;
}
/**
* report event on the size of the compaction.
*/
public void started()
{
reportedSeverity = getCompactionInfo().getTotal() / load;
StorageService.instance.reportSeverity(reportedSeverity);
}
/**
* remove the event complete
*/
public void finished()
{
if (reportedSeverity != 0d)
StorageService.instance.reportSeverity(-(reportedSeverity));
reportedSeverity = 0d;
}
}
}

25
src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java
View File

@ -23,6 +23,7 @@ import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import com.codahale.metrics.ExponentiallyDecayingReservoir;
@ -31,9 +32,14 @@ import javax.management.ObjectName;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.gms.ApplicationState;
import org.apache.cassandra.gms.EndpointState;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.gms.VersionedValue;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.ftpserver.command.impl.STOR;
/**
@ -283,7 +289,7 @@ public class DynamicEndpointSnitch extends AbstractEndpointSnitch implements ILa
// finally, add the severity without any weighting, since hosts scale this relative to their own load and the size of the task causing the severity.
// "Severity" is basically a measure of compaction activity (CASSANDRA-3722).
if (USE_SEVERITY)
score += StorageService.instance.getSeverity(entry.getKey());
score += getSeverity(entry.getKey());
// lowest score (least amount of badness) wins.
newScores.put(entry.getKey(), score);
}
@ -333,12 +339,25 @@ public class DynamicEndpointSnitch extends AbstractEndpointSnitch implements ILa
public void setSeverity(double severity)
{
StorageService.instance.reportManualSeverity(severity);
Gossiper.instance.addLocalApplicationState(ApplicationState.SEVERITY, StorageService.instance.valueFactory.severity(severity));
}
private double getSeverity(InetAddress endpoint)
{
EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
if (state == null)
return 0.0;
VersionedValue event = state.getApplicationState(ApplicationState.SEVERITY);
if (event == null)
return 0.0;
return Double.parseDouble(event.value);
}
public double getSeverity()
{
return StorageService.instance.getSeverity(FBUtilities.getBroadcastAddress());
return getSeverity(FBUtilities.getBroadcastAddress());
}
public boolean isWorthMergingForRangeQuery(List<InetAddress> merged, List<InetAddress> l1, List<InetAddress> l2)

25
src/java/org/apache/cassandra/locator/DynamicEndpointSnitchMBean.java
View File

@ -29,11 +29,30 @@ public interface DynamicEndpointSnitchMBean {
public double getBadnessThreshold();
public String getSubsnitchClassName();
public List<Double> dumpTimings(String hostname) throws UnknownHostException;
/**
* Use this if you want to specify a severity; it can be negative
* Example: Page cache is cold and you want data to be sent
* though it is not preferred one.
* Setting a Severity allows operators to inject preference information into the Dynamic Snitch
* replica selection.
*
* When choosing which replicas to participate in a read request, the DSnitch sorts replicas
* by response latency, and selects the fastest replicas. Latencies are normalized to a score
* from 0 to 1, with lower scores being faster.
*
* The Severity injected here will be added to the normalized score.
*
* Thus, adding a Severity greater than 1 will mean the replica will never be contacted
* (unless needed for ALL or if it is added later for rapid read protection).
*
* Conversely, adding a negative Severity means the replica will *always* be contacted.
*
* (The "Severity" term is historical and dates to when this was used to represent how
* badly background tasks like compaction were affecting a replica's performance.
* See CASSANDRA-3722 for when this was introduced and CASSANDRA-11738 for why it was removed.)
*/
public void setSeverity(double severity);
/**
* @return the current manually injected Severity.
*/
public double getSeverity();
}

4
src/java/org/apache/cassandra/metrics/CompactionMetrics.java
View File

@ -140,15 +140,11 @@ public class CompactionMetrics implements CompactionManager.CompactionExecutorSt
public void beginCompaction(CompactionInfo.Holder ci)
{
// notify
ci.started();
compactions.add(ci);
}
public void finishCompaction(CompactionInfo.Holder ci)
{
// notify
ci.finished();
compactions.remove(ci);
bytesCompacted.inc(ci.getCompactionInfo().getTotal());
totalCompactionsCompleted.mark();

20
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -189,8 +189,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
private final List<IEndpointLifecycleSubscriber> lifecycleSubscribers = new CopyOnWriteArrayList<>();
private static final BackgroundActivityMonitor bgMonitor = new BackgroundActivityMonitor();
private final ObjectName jmxObjectName;
private Collection<Token> bootstrapTokens = null;
@ -1467,24 +1465,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
return tokenMetadata;
}
/**
* Increment about the known Compaction severity of the events in this node
*/
public void reportSeverity(double incr)
{
bgMonitor.incrCompactionSeverity(incr);
}
public void reportManualSeverity(double incr)
{
bgMonitor.incrManualSeverity(incr);
}
public double getSeverity(InetAddress endpoint)
{
return bgMonitor.getSeverity(endpoint);
}
/**
* for a keyspace, return the ranges and corresponding listen addresses.
* @param keyspace

171
src/java/org/apache/cassandra/utils/BackgroundActivityMonitor.java
View File

@ -1,171 +0,0 @@
package org.apache.cassandra.utils;
/*
*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you 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.
*
*/
import java.io.IOException;
import java.io.RandomAccessFile;
import java.lang.management.ManagementFactory;
import java.net.InetAddress;
import java.util.StringTokenizer;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.cassandra.concurrent.DebuggableScheduledThreadPoolExecutor;
import org.apache.cassandra.gms.ApplicationState;
import org.apache.cassandra.gms.EndpointState;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.gms.VersionedValue;
import org.apache.cassandra.service.StorageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.util.concurrent.AtomicDouble;
public class BackgroundActivityMonitor
{
private static final Logger logger = LoggerFactory.getLogger(BackgroundActivityMonitor.class);
public static final int USER_INDEX = 0;
public static final int NICE_INDEX = 1;
public static final int SYS_INDEX = 2;
public static final int IDLE_INDEX = 3;
public static final int IOWAIT_INDEX = 4;
public static final int IRQ_INDEX = 5;
public static final int SOFTIRQ_INDEX = 6;
private static final int NUM_CPUS = Runtime.getRuntime().availableProcessors();
private static final String PROC_STAT_PATH = "/proc/stat";
private final AtomicDouble compaction_severity = new AtomicDouble();
private final AtomicDouble manual_severity = new AtomicDouble();
private final ScheduledExecutorService reportThread = new DebuggableScheduledThreadPoolExecutor("Background_Reporter");
private RandomAccessFile statsFile;
private long[] lastReading;
public BackgroundActivityMonitor()
{
try
{
statsFile = new RandomAccessFile(PROC_STAT_PATH, "r");
lastReading = readAndCompute();
}
catch (IOException ex)
{
if (FBUtilities.hasProcFS())
logger.warn("Couldn't open /proc/stats");
statsFile = null;
}
reportThread.scheduleAtFixedRate(new BackgroundActivityReporter(), 1, 1, TimeUnit.SECONDS);
}
private long[] readAndCompute() throws IOException
{
statsFile.seek(0);
StringTokenizer tokenizer = new StringTokenizer(statsFile.readLine());
String name = tokenizer.nextToken();
assert name.equalsIgnoreCase("cpu");
long[] returned = new long[tokenizer.countTokens()];
for (int i = 0; i < returned.length; i++)
returned[i] = Long.parseLong(tokenizer.nextToken());
return returned;
}
private float compareAtIndex(long[] reading1, long[] reading2, int index)
{
long total1 = 0, total2 = 0;
for (int i = 0; i <= SOFTIRQ_INDEX; i++)
{
total1 += reading1[i];
total2 += reading2[i];
}
float totalDiff = total2 - total1;
long intrested1 = reading1[index], intrested2 = reading2[index];
float diff = intrested2 - intrested1;
if (diff == 0)
return 0f;
return (diff / totalDiff) * 100; // yes it is hard coded to 100 [update
// unit?]
}
public void incrCompactionSeverity(double sev)
{
compaction_severity.addAndGet(sev);
}
public void incrManualSeverity(double sev)
{
manual_severity.addAndGet(sev);
}
public double getIOWait() throws IOException
{
if (statsFile == null)
return -1d;
long[] newComp = readAndCompute();
double value = compareAtIndex(lastReading, newComp, IOWAIT_INDEX);
lastReading = newComp;
return value;
}
public double getNormalizedLoadAvg()
{
double avg = ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage();
return avg / NUM_CPUS;
}
public double getSeverity(InetAddress endpoint)
{
VersionedValue event;
EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
if (state != null && (event = state.getApplicationState(ApplicationState.SEVERITY)) != null)
return Double.parseDouble(event.value);
return 0.0;
}
public class BackgroundActivityReporter implements Runnable
{
public void run()
{
double report = -1;
try
{
report = getIOWait();
}
catch (IOException e)
{
// ignore;
if (FBUtilities.hasProcFS())
logger.warn("Couldn't read /proc/stats");
}
if (report == -1d)
report = compaction_severity.get();
if (!Gossiper.instance.isEnabled())
return;
report += manual_severity.get(); // add manual severity setting.
VersionedValue updated = StorageService.instance.valueFactory.severity(report);
Gossiper.instance.addLocalApplicationState(ApplicationState.SEVERITY, updated);
}
}
}