CassandraJmxHttpServer.java

package com.smeet.cassandra;

import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpHandler;
import com.sun.net.httpserver.HttpServer;
import org.apache.cassandra.db.CompactionManager;
import org.apache.cassandra.db.CompactionManagerMBean;
import org.apache.cassandra.service.StorageProxyMBean;
import org.apache.cassandra.service.StorageServiceMBean;

import javax.management.JMX;
import javax.management.MBeanServerConnection;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
import javax.management.remote.JMXConnector;
import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXServiceURL;
import java.io.IOException;
import java.io.OutputStream;
import java.lang.management.ManagementFactory;
import java.lang.management.MemoryMXBean;
import java.lang.management.MemoryUsage;
import java.net.InetSocketAddress;
import java.text.NumberFormat;
import java.util.Collections;

/**
 * @author smeet
 */
public class CassandraJmxHttpServer {

    private static final String FMT_URL = "service:jmx:rmi:///jndi/rmi://%s:%d/jmxrmi";
    private static final String SS_OBJECT_NAME = "org.apache.cassandra.service:type=StorageService";
    private static final String SP_OBJECT_NAME = "org.apache.cassandra.service:type=StorageProxy";
    
    private static final int DEFAULT_PORT = 8080;

    private int port = DEFAULT_PORT;

    private String host;

    private StorageServiceMBean _ssProxy;
    private MemoryMXBean _memProxy;
    private CompactionManagerMBean _mcmProxy;
    private StorageProxyMBean _spProxy;

    public CassandraJmxHttpServer(String host) {
        this.host = host;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public void setPort(int port) {
        this.port = port;
    }

    /**
     * Create a connection to the JMX agent and setup the M[X]Bean proxies.
     *
     * @throws IOException on connection failures
     */
    public void start() throws IOException {
        JMXServiceURL jmxUrl = new JMXServiceURL(String.format(FMT_URL, host, port));
        JMXConnector jmxc = JMXConnectorFactory.connect(jmxUrl, null);
        MBeanServerConnection mbeanServerConn = jmxc.getMBeanServerConnection();

        try {
            _ssProxy = JMX.newMBeanProxy(mbeanServerConn, new ObjectName(SS_OBJECT_NAME), StorageServiceMBean.class);
            _spProxy = JMX.newMBeanProxy(mbeanServerConn, new ObjectName(SP_OBJECT_NAME), StorageProxyMBean.class);
            _mcmProxy = JMX.newMBeanProxy(mbeanServerConn, new ObjectName(CompactionManager.MBEAN_OBJECT_NAME), CompactionManagerMBean.class);
        
        } catch (MalformedObjectNameException e) {
            throw new RuntimeException("Cannot connect.", e);
        }

        _memProxy = ManagementFactory.newPlatformMXBeanProxy(mbeanServerConn,
                ManagementFactory.MEMORY_MXBEAN_NAME, MemoryMXBean.class);

        HttpServer httpServer = HttpServer.create();
        
        httpServer.createContext("/jmxexport", new HttpHandler() {
            @Override
            public void handle(HttpExchange httpExchange) throws IOException {
                httpExchange.getResponseHeaders().put("Content-Type", Collections.singletonList("text/plain; charset=UTF8"));
                StringBuilder body = new StringBuilder();

                // Storage Service
                append(body, "OperationMode", _ssProxy.getOperationMode());
                double load = _ssProxy.getLoad() / (1024 * 1024 * 1024);
                NumberFormat formatter = NumberFormat.getInstance();
                formatter.setGroupingUsed(false);                
                append(body, "Load", formatter.format(load));

                // Storage Proxy
                append(body, "ReadOperations", _spProxy.getReadOperations());
                append(body, "WriteOperations", _spProxy.getWriteOperations());
                append(body, "TotalReadLatencyMicros", _spProxy.getTotalReadLatencyMicros());
                append(body, "TotalWriteLatencyMicros", _spProxy.getTotalWriteLatencyMicros());
                
                // Compactions
                Long bytesCompacted = _mcmProxy.getBytesCompacted();
                append(body, "BytesCompacted", bytesCompacted != null ? bytesCompacted : 0);
                Long bytesTotalInProgress = _mcmProxy.getBytesTotalInProgress();
                append(body, "BytesTotalInProgress", bytesTotalInProgress != null ? bytesTotalInProgress : 0);
                append(body, "PendingTasks", _mcmProxy.getPendingTasks());

                MemoryUsage heapMemoryUsage = _memProxy.getHeapMemoryUsage();
                append(body, "HeapUsed", heapMemoryUsage.getUsed());

                OutputStream out = httpExchange.getResponseBody();

                byte[] bytes = body.toString().getBytes("UTF8");
                httpExchange.sendResponseHeaders(200, bytes.length);
                out.write(bytes);
                httpExchange.close();
            }

            private void append(StringBuilder body, String name, Object value) {
                body.append(name).append('=').append(value).append('\n');
            }
        });
        httpServer.bind(new InetSocketAddress(9090), 0);
        httpServer.start();
    }

    public static void main(String[] args) throws IOException, InterruptedException {
        String host = args.length > 0 ? args[0] : "localhost";
        CassandraJmxHttpServer server = new CassandraJmxHttpServer(host);
        server.start();
    }
}