andreacavalli/scylla-jmx
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge "Adding the messaging service counters" from Amnon

Browse Source 
"This series adds the dropped, timeout and their recently version counter to the
 MessagingService."
This commit is contained in:
Pekka Enberg 2015-11-18 08:59:32 +02:00
commit d06e6fdde1
6 changed files with 267 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
src/main/java/com/cloudius/urchin/metrics/APIMetrics.java
View File

@ -288,6 +288,23 @@ public class APIMetrics {
return DEFAULT_REGISTRY.newMeter(url, metricName, eventType, unit);
}
/**
* Creates a new {@link com.yammer.metrics.core.Meter} and registers it
* under the given metric name.
*
* @param metricName
* the name of the metric
* @param eventType
* the plural name of the type of events the meter is measuring
* (e.g., {@code "requests"})
* @param unit
* the rate unit of the new meter
* @return a new {@link com.yammer.metrics.core.Meter}
*/
public static Meter newSettableMeter(MetricName metricName,
String eventType, TimeUnit unit) {
return DEFAULT_REGISTRY.newSettableMeter(metricName, eventType, unit);
}
/**
* Creates a new {@link com.yammer.metrics.core.APITimer} and registers it
* under the given class and name.

15
src/main/java/com/yammer/metrics/core/APIMeter.java
View File

@ -10,7 +10,7 @@ import java.util.concurrent.TimeUnit;
import com.cloudius.urchin.api.APIClient;
public class APIMeter extends Meter {
public class APIMeter extends APISettableMeter {
String url;
private APIClient c = new APIClient();
@ -25,19 +25,8 @@ public class APIMeter extends Meter {
return c.getLongValue(url);
}
// Meter doesn't have a set value method.
// to mimic it, we clear the old value and set it to a new one.
// This is safe because the only this method would be used
// to update the values
public long set(long new_value) {
long res = super.count();
mark(-res);
mark(new_value);
return res;
}
@Override
void tick() {
public void tick() {
set(get_value());
super.tick();
}

22
src/main/java/com/yammer/metrics/core/APIMetricsRegistry.java
View File

@ -181,6 +181,28 @@ public class APIMetricsRegistry extends MetricsRegistry {
eventType, unit, getClock()));
}
/**
* Creates a new {@link APISettableMeter} and registers it under the given metric name.
*
* @param metricName
* the name of the metric
* @param eventType
* the plural name of the type of events the meter is measuring
* (e.g., {@code "requests"})
* @param unit
* the rate unit of the new meter
* @return a new {@link Meter}
*/
public Meter newSettableMeter(MetricName metricName, String eventType,
TimeUnit unit) {
final Metric existingMetric = getMetrics().get(metricName);
if (existingMetric != null) {
return (Meter) existingMetric;
}
return getOrAdd(metricName, new APISettableMeter(newMeterTickThreadPool(),
eventType, unit, getClock()));
}
/**
* Creates a new {@link Histogram} and registers it under the given class
* and name.

49
src/main/java/com/yammer/metrics/core/APISettableMeter.java Normal file
View File

@ -0,0 +1,49 @@
package com.yammer.metrics.core;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/*
* Copyright 2015 ScyllaDB
*
*/
/*
* This file is part of Scylla.
*
* Scylla is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* Scylla is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with Scylla. If not, see <http://www.gnu.org/licenses/>.
*/
public class APISettableMeter extends Meter {
public APISettableMeter(ScheduledExecutorService tickThread,
String eventType, TimeUnit rateUnit, Clock clock) {
super(tickThread, eventType, rateUnit, clock);
}
// Meter doesn't have a set value method.
// to mimic it, we clear the old value and set it to a new one.
// This is safe because the only this method would be used
// to update the values
public long set(long new_value) {
long res = super.count();
mark(-res);
mark(new_value);
return res;
}
@Override
public void tick() {
super.tick();
}
}

65
src/main/java/org/apache/cassandra/metrics/DroppedMessageMetrics.java Normal file
View File

@ -0,0 +1,65 @@
/*
* 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.
*/
/*
* Copyright (C) 2015 ScyllaDB
*/
/*
* Moddified by ScyllaDB
*/
package org.apache.cassandra.metrics;
import java.util.concurrent.TimeUnit;
import org.apache.cassandra.net.MessagingService;
import com.cloudius.urchin.metrics.APIMetrics;
import com.cloudius.urchin.metrics.DefaultNameFactory;
import com.cloudius.urchin.metrics.MetricNameFactory;
import com.yammer.metrics.core.APISettableMeter;
/**
* Metrics for dropped messages by verb.
*/
public class DroppedMessageMetrics {
/** Number of dropped messages */
public final APISettableMeter dropped;
private long lastDropped = 0;
public DroppedMessageMetrics(MessagingService.Verb verb) {
MetricNameFactory factory = new DefaultNameFactory("DroppedMessage",
verb.toString());
dropped = (APISettableMeter) APIMetrics.newSettableMeter(
factory.createMetricName("Dropped"), "dropped",
TimeUnit.SECONDS);
dropped.stop();
}
@Deprecated
public int getRecentlyDropped() {
long currentDropped = dropped.count();
long recentlyDropped = currentDropped - lastDropped;
lastDropped = currentDropped;
return (int) recentlyDropped;
}
public APISettableMeter getMeter() {
return dropped;
}
}

122
src/main/java/org/apache/cassandra/net/MessagingService.java
View File

@ -25,26 +25,82 @@ package org.apache.cassandra.net;
import java.lang.management.ManagementFactory;
import java.net.*;
import java.util.*;
import java.util.Map.Entry;
import javax.json.JsonArray;
import javax.json.JsonObject;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import org.apache.cassandra.metrics.DroppedMessageMetrics;
import com.cloudius.urchin.api.APIClient;
import com.yammer.metrics.core.APISettableMeter;
public final class MessagingService implements MessagingServiceMBean {
static final int INTERVAL = 1000; // update every 1second
public static final String MBEAN_NAME = "org.apache.cassandra.net:type=MessagingService";
private static final java.util.logging.Logger logger = java.util.logging.Logger
.getLogger(MessagingService.class.getName());
Map<String, DroppedMessageMetrics> dropped;
private APIClient c = new APIClient();
Map<String, Long> resent_timeout = new HashMap<String, Long>();
private final ObjectName jmxObjectName;
private long recentTimeoutCount;
/* All verb handler identifiers */
public enum Verb
{
MUTATION,
@Deprecated BINARY,
READ_REPAIR,
READ,
REQUEST_RESPONSE, // client-initiated reads and writes
@Deprecated STREAM_INITIATE,
@Deprecated STREAM_INITIATE_DONE,
@Deprecated STREAM_REPLY,
@Deprecated STREAM_REQUEST,
RANGE_SLICE,
@Deprecated BOOTSTRAP_TOKEN,
@Deprecated TREE_REQUEST,
@Deprecated TREE_RESPONSE,
@Deprecated JOIN,
GOSSIP_DIGEST_SYN,
GOSSIP_DIGEST_ACK,
GOSSIP_DIGEST_ACK2,
@Deprecated DEFINITIONS_ANNOUNCE,
DEFINITIONS_UPDATE,
TRUNCATE,
SCHEMA_CHECK,
@Deprecated INDEX_SCAN,
REPLICATION_FINISHED,
INTERNAL_RESPONSE, // responses to internal calls
COUNTER_MUTATION,
@Deprecated STREAMING_REPAIR_REQUEST,
@Deprecated STREAMING_REPAIR_RESPONSE,
SNAPSHOT, // Similar to nt snapshot
MIGRATION_REQUEST,
GOSSIP_SHUTDOWN,
_TRACE, // dummy verb so we can use MS.droppedMessages
ECHO,
REPAIR_MESSAGE,
// use as padding for backwards compatability where a previous version needs to validate a verb from the future.
PAXOS_PREPARE,
PAXOS_PROPOSE,
PAXOS_COMMIT,
PAGED_RANGE,
// remember to add new verbs at the end, since we serialize by ordinal
UNUSED_1,
UNUSED_2,
UNUSED_3,
;
}
public void log(String str) {
System.out.println(str);
logger.info(str);
}
private static Timer timer = new Timer("Dropped messages");
public MessagingService() {
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
try {
@ -55,10 +111,30 @@ public final class MessagingService implements MessagingServiceMBean {
} catch (Exception e) {
throw new RuntimeException(e);
}
timer.schedule(new CheckDroppedMessages(), INTERVAL, INTERVAL);
}
static MessagingService instance = new MessagingService();
private static final class CheckDroppedMessages extends TimerTask {
@Override
public void run() {
if (instance.dropped == null) {
instance.dropped = new HashMap<String, DroppedMessageMetrics>();
for (Verb v : Verb.values()) {
instance.dropped.put(v.name(), new DroppedMessageMetrics(v));
}
}
Map<String, Integer> val = instance.getDroppedMessages();
for (String k : val.keySet()) {
APISettableMeter meter = instance.dropped.get(k).getMeter();
meter.set(val.get(k));
System.out.println("tick " + k + " " + meter.count());
meter.tick();
}
}
}
public static MessagingService getInstance() {
return instance;
}
@ -86,7 +162,7 @@ public final class MessagingService implements MessagingServiceMBean {
*/
public Map<String, Long> getCommandDroppedTasks() {
log(" getCommandDroppedTasks()");
return c.getMapStringLongValue("");
return c.getMapStringLongValue("/messaging_service/messages/dropped");
}
/**
@ -110,7 +186,13 @@ public final class MessagingService implements MessagingServiceMBean {
*/
public Map<String, Integer> getDroppedMessages() {
log(" getDroppedMessages()");
return c.getMapStringIntegerValue("/messaging_service/messages/dropped");
Map<String, Integer> res = new HashMap<String, Integer>();
JsonArray arr = c.getJsonArray("/messaging_service/messages/dropped_by_ver");
for (int i = 0; i < arr.size(); i++) {
JsonObject obj = arr.getJsonObject(i);
res.put(obj.getString("verb"), obj.getInt("count"));
}
return res;
}
/**
@ -118,7 +200,10 @@ public final class MessagingService implements MessagingServiceMBean {
*/
public Map<String, Integer> getRecentlyDroppedMessages() {
log(" getRecentlyDroppedMessages()");
return c.getMapStringIntegerValue("");
Map<String, Integer> map = new HashMap<String, Integer>();
for (Map.Entry<String, DroppedMessageMetrics> entry : dropped.entrySet())
map.put(entry.getKey(), entry.getValue().getRecentlyDropped());
return map;
}
/**
@ -126,7 +211,12 @@ public final class MessagingService implements MessagingServiceMBean {
*/
public long getTotalTimeouts() {
log(" getTotalTimeouts()");
return c.getLongValue("");
Map<String, Long> timeouts = getTimeoutsPerHost();
long res = 0;
for (Entry<String, Long> t : timeouts.entrySet()) {
res += t.getValue();
}
return res;
}
/**
@ -134,7 +224,7 @@ public final class MessagingService implements MessagingServiceMBean {
*/
public Map<String, Long> getTimeoutsPerHost() {
log(" getTimeoutsPerHost()");
return c.getMapStringLongValue("");
return c.getMapStringLongValue("/messaging_service/messages/timeout");
}
/**
@ -142,7 +232,10 @@ public final class MessagingService implements MessagingServiceMBean {
*/
public long getRecentTotalTimouts() {
log(" getRecentTotalTimouts()");
return c.getLongValue("");
long timeoutCount = getTotalTimeouts();
long recent = timeoutCount - recentTimeoutCount;
recentTimeoutCount = timeoutCount;
return recent;
}
/**
@ -150,7 +243,16 @@ public final class MessagingService implements MessagingServiceMBean {
*/
public Map<String, Long> getRecentTimeoutsPerHost() {
log(" getRecentTimeoutsPerHost()");
return c.getMapStringLongValue("");
Map<String, Long> timeouts = getTimeoutsPerHost();
Map<String, Long> result = new HashMap<String, Long>();
for ( Entry<String, Long> e : timeouts.entrySet()) {
long res = e.getValue().longValue() -
((resent_timeout.containsKey(e.getKey()))? (resent_timeout.get(e.getKey())).longValue()
: 0);
resent_timeout.put(e.getKey(), e.getValue());
result.put(e.getKey(),res);
}
return result;
}
public int getVersion(String address) throws UnknownHostException {