| /* |
| * 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. |
| */ |
| package org.apache.cassandra.metrics; |
| |
| import java.util.EnumMap; |
| import java.util.HashMap; |
| import java.util.Map; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.function.Consumer; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import org.apache.cassandra.concurrent.ScheduledExecutors; |
| import org.apache.cassandra.config.DatabaseDescriptor; |
| |
| import com.codahale.metrics.Timer; |
| import org.apache.cassandra.locator.InetAddressAndPort; |
| import org.apache.cassandra.net.Message; |
| import org.apache.cassandra.net.Verb; |
| import org.apache.cassandra.net.InboundMessageHandlers; |
| import org.apache.cassandra.net.LatencyConsumer; |
| import org.apache.cassandra.utils.StatusLogger; |
| |
| import static java.util.concurrent.TimeUnit.MILLISECONDS; |
| import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics; |
| |
| /** |
| * Metrics for messages |
| */ |
| public class MessagingMetrics implements InboundMessageHandlers.GlobalMetricCallbacks |
| { |
| private static final MetricNameFactory factory = new DefaultNameFactory("Messaging"); |
| private static final Logger logger = LoggerFactory.getLogger(MessagingMetrics.class); |
| private static final int LOG_DROPPED_INTERVAL_IN_MS = 5000; |
| |
| public static class DCLatencyRecorder implements LatencyConsumer |
| { |
| public final Timer dcLatency; |
| public final Timer allLatency; |
| |
| DCLatencyRecorder(Timer dcLatency, Timer allLatency) |
| { |
| this.dcLatency = dcLatency; |
| this.allLatency = allLatency; |
| } |
| |
| public void accept(long timeTaken, TimeUnit units) |
| { |
| if (timeTaken > 0) |
| { |
| dcLatency.update(timeTaken, units); |
| allLatency.update(timeTaken, units); |
| } |
| } |
| } |
| |
| private static final class DroppedForVerb |
| { |
| final DroppedMessageMetrics metrics; |
| final AtomicInteger droppedFromSelf; |
| final AtomicInteger droppedFromPeer; |
| |
| DroppedForVerb(Verb verb) |
| { |
| this(new DroppedMessageMetrics(verb)); |
| } |
| |
| DroppedForVerb(DroppedMessageMetrics metrics) |
| { |
| this.metrics = metrics; |
| this.droppedFromSelf = new AtomicInteger(0); |
| this.droppedFromPeer = new AtomicInteger(0); |
| } |
| } |
| |
| private final Timer allLatency; |
| public final ConcurrentHashMap<String, DCLatencyRecorder> dcLatency; |
| public final EnumMap<Verb, Timer> internalLatency; |
| |
| // total dropped message counts for server lifetime |
| private final Map<Verb, DroppedForVerb> droppedMessages = new EnumMap<>(Verb.class); |
| |
| public MessagingMetrics() |
| { |
| allLatency = Metrics.timer(factory.createMetricName("CrossNodeLatency")); |
| dcLatency = new ConcurrentHashMap<>(); |
| internalLatency = new EnumMap<>(Verb.class); |
| for (Verb verb : Verb.VERBS) |
| internalLatency.put(verb, Metrics.timer(factory.createMetricName(verb + "-WaitLatency"))); |
| for (Verb verb : Verb.values()) |
| droppedMessages.put(verb, new DroppedForVerb(verb)); |
| } |
| |
| public DCLatencyRecorder internodeLatencyRecorder(InetAddressAndPort from) |
| { |
| String dcName = DatabaseDescriptor.getEndpointSnitch().getDatacenter(from); |
| DCLatencyRecorder dcUpdater = dcLatency.get(dcName); |
| if (dcUpdater == null) |
| dcUpdater = dcLatency.computeIfAbsent(dcName, k -> new DCLatencyRecorder(Metrics.timer(factory.createMetricName(dcName + "-Latency")), allLatency)); |
| return dcUpdater; |
| } |
| |
| public void recordInternalLatency(Verb verb, long timeTaken, TimeUnit units) |
| { |
| if (timeTaken > 0) |
| internalLatency.get(verb).update(timeTaken, units); |
| } |
| |
| public void recordSelfDroppedMessage(Verb verb) |
| { |
| recordDroppedMessage(droppedMessages.get(verb), false); |
| } |
| |
| public void recordSelfDroppedMessage(Verb verb, long timeElapsed, TimeUnit timeUnit) |
| { |
| recordDroppedMessage(verb, timeElapsed, timeUnit, false); |
| } |
| |
| public void recordInternodeDroppedMessage(Verb verb, long timeElapsed, TimeUnit timeUnit) |
| { |
| recordDroppedMessage(verb, timeElapsed, timeUnit, true); |
| } |
| |
| public void recordDroppedMessage(Message<?> message, long timeElapsed, TimeUnit timeUnit) |
| { |
| recordDroppedMessage(message.verb(), timeElapsed, timeUnit, message.isCrossNode()); |
| } |
| |
| public void recordDroppedMessage(Verb verb, long timeElapsed, TimeUnit timeUnit, boolean isCrossNode) |
| { |
| recordDroppedMessage(droppedMessages.get(verb), timeElapsed, timeUnit, isCrossNode); |
| } |
| |
| private static void recordDroppedMessage(DroppedForVerb droppedMessages, long timeTaken, TimeUnit units, boolean isCrossNode) |
| { |
| if (isCrossNode) |
| droppedMessages.metrics.crossNodeDroppedLatency.update(timeTaken, units); |
| else |
| droppedMessages.metrics.internalDroppedLatency.update(timeTaken, units); |
| recordDroppedMessage(droppedMessages, isCrossNode); |
| } |
| |
| private static void recordDroppedMessage(DroppedForVerb droppedMessages, boolean isCrossNode) |
| { |
| droppedMessages.metrics.dropped.mark(); |
| if (isCrossNode) |
| droppedMessages.droppedFromPeer.incrementAndGet(); |
| else |
| droppedMessages.droppedFromSelf.incrementAndGet(); |
| } |
| |
| public void scheduleLogging() |
| { |
| ScheduledExecutors.scheduledTasks.scheduleWithFixedDelay(this::logDroppedMessages, |
| LOG_DROPPED_INTERVAL_IN_MS, |
| LOG_DROPPED_INTERVAL_IN_MS, |
| MILLISECONDS); |
| } |
| |
| public Map<String, Integer> getDroppedMessages() |
| { |
| Map<String, Integer> map = new HashMap<>(droppedMessages.size()); |
| for (Map.Entry<Verb, DroppedForVerb> entry : droppedMessages.entrySet()) |
| map.put(entry.getKey().toString(), (int) entry.getValue().metrics.dropped.getCount()); |
| return map; |
| } |
| |
| private void logDroppedMessages() |
| { |
| if (resetAndConsumeDroppedErrors(logger::info) > 0) |
| StatusLogger.log(); |
| } |
| |
| @VisibleForTesting |
| public int resetAndConsumeDroppedErrors(Consumer<String> messageConsumer) |
| { |
| int count = 0; |
| for (Map.Entry<Verb, DroppedForVerb> entry : droppedMessages.entrySet()) |
| { |
| Verb verb = entry.getKey(); |
| DroppedForVerb droppedForVerb = entry.getValue(); |
| |
| int droppedInternal = droppedForVerb.droppedFromSelf.getAndSet(0); |
| int droppedCrossNode = droppedForVerb.droppedFromPeer.getAndSet(0); |
| if (droppedInternal > 0 || droppedCrossNode > 0) |
| { |
| messageConsumer.accept(String.format("%s messages were dropped in last %d ms: %d internal and %d cross node." |
| + " Mean internal dropped latency: %d ms and Mean cross-node dropped latency: %d ms", |
| verb, |
| LOG_DROPPED_INTERVAL_IN_MS, |
| droppedInternal, |
| droppedCrossNode, |
| TimeUnit.NANOSECONDS.toMillis((long) droppedForVerb.metrics.internalDroppedLatency.getSnapshot().getMean()), |
| TimeUnit.NANOSECONDS.toMillis((long) droppedForVerb.metrics.crossNodeDroppedLatency.getSnapshot().getMean()))); |
| ++count; |
| } |
| } |
| return count; |
| } |
| |
| @VisibleForTesting |
| public void resetDroppedMessages() |
| { |
| droppedMessages.replaceAll((u, v) -> new DroppedForVerb(new DroppedMessageMetrics(u))); |
| } |
| |
| } |