You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by st...@apache.org on 2013/05/03 05:58:35 UTC
svn commit: r1478639 [7/10] - in /hbase/branches/0.95:
hbase-client/src/main/java/org/apache/hadoop/hbase/
hbase-client/src/main/java/org/apache/hadoop/hbase/catalog/
hbase-client/src/main/java/org/apache/hadoop/hbase/client/
hbase-client/src/main/java...
Modified: hbase/branches/0.95/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcServer.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.95/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcServer.java?rev=1478639&r1=1478638&r2=1478639&view=diff
==============================================================================
--- hbase/branches/0.95/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcServer.java (original)
+++ hbase/branches/0.95/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcServer.java Fri May 3 03:58:33 2013
@@ -1,4 +1,4 @@
-/*
+/**
*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@@ -8,7 +8,7 @@
* "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
+ * 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,
@@ -19,58 +19,2474 @@
package org.apache.hadoop.hbase.ipc;
+import static org.apache.hadoop.fs.CommonConfigurationKeys.HADOOP_SECURITY_AUTHORIZATION;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
import java.io.IOException;
-import java.lang.reflect.Method;
+import java.net.BindException;
+import java.net.InetAddress;
import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketException;
+import java.net.UnknownHostException;
+import java.nio.ByteBuffer;
+import java.nio.channels.CancelledKeyException;
+import java.nio.channels.Channels;
+import java.nio.channels.ClosedChannelException;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.nio.channels.WritableByteChannel;
+import java.security.PrivilegedExceptionAction;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.security.sasl.Sasl;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+import com.google.common.collect.Lists;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CellScanner;
-import org.apache.hadoop.hbase.IpcProtocol;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.client.Operation;
+import org.apache.hadoop.hbase.codec.Codec;
+import org.apache.hadoop.hbase.exceptions.CallerDisconnectedException;
+import org.apache.hadoop.hbase.exceptions.DoNotRetryIOException;
+import org.apache.hadoop.hbase.exceptions.RegionMovedException;
+import org.apache.hadoop.hbase.exceptions.ServerNotRunningYetException;
+import org.apache.hadoop.hbase.io.ByteBufferOutputStream;
import org.apache.hadoop.hbase.monitoring.MonitoredRPCHandler;
+import org.apache.hadoop.hbase.monitoring.TaskMonitor;
+import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.CellBlockMeta;
+import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.ConnectionHeader;
+import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.ExceptionResponse;
import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader;
+import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.ResponseHeader;
+import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.UserInformation;
+import org.apache.hadoop.hbase.regionserver.HRegionServer;
+import org.apache.hadoop.hbase.security.AuthMethod;
+import org.apache.hadoop.hbase.security.HBasePolicyProvider;
+import org.apache.hadoop.hbase.security.HBaseSaslRpcServer;
+import org.apache.hadoop.hbase.security.HBaseSaslRpcServer.SaslDigestCallbackHandler;
+import org.apache.hadoop.hbase.security.HBaseSaslRpcServer.SaslGssCallbackHandler;
+import org.apache.hadoop.hbase.security.SaslStatus;
+import org.apache.hadoop.hbase.security.SaslUtil;
+import org.apache.hadoop.hbase.security.User;
+import org.apache.hadoop.hbase.security.token.AuthenticationTokenSecretManager;
+import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableUtils;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.hbase.Server;
+import org.apache.hadoop.security.AccessControlException;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod;
+import org.apache.hadoop.security.authorize.AuthorizationException;
+import org.apache.hadoop.security.authorize.PolicyProvider;
+import org.apache.hadoop.security.authorize.ProxyUsers;
+import org.apache.hadoop.security.authorize.ServiceAuthorizationManager;
+import org.apache.hadoop.security.token.SecretManager;
+import org.apache.hadoop.security.token.SecretManager.InvalidToken;
+import org.apache.hadoop.security.token.TokenIdentifier;
+import org.apache.hadoop.util.StringUtils;
+import org.cliffc.high_scale_lib.Counter;
+import org.cloudera.htrace.Sampler;
+import org.cloudera.htrace.Span;
+import org.cloudera.htrace.Trace;
+import org.cloudera.htrace.TraceInfo;
+import org.cloudera.htrace.impl.NullSpan;
+import org.codehaus.jackson.map.ObjectMapper;
import com.google.common.base.Function;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.google.protobuf.BlockingService;
+import com.google.protobuf.CodedInputStream;
+import com.google.protobuf.Descriptors.MethodDescriptor;
import com.google.protobuf.Message;
+import com.google.protobuf.Message.Builder;
+import com.google.protobuf.ServiceException;
+import com.google.protobuf.TextFormat;
+// Uses Writables doing sasl
+/**
+ * An RPC server that hosts protobuf described Services.
+ * <p>Once was copied from Hadoop to local to fix HBASE-900 but deviated long ago.
+ *
+ * @see RpcClient
+ */
@InterfaceAudience.Private
-public interface RpcServer {
- // TODO: Needs cleanup. Why a 'start', and then a 'startThreads' and an 'openServer'?
- // Also, the call takes a RpcRequestBody, an already composed combination of
- // rpc Request and metadata. Should disentangle metadata and rpc Request Message.
+public class RpcServer implements RpcServerInterface {
+ // The logging package is deliberately outside of standard o.a.h.h package so it is not on
+ // by default.
+ public static final Log LOG = LogFactory.getLog("org.apache.hadoop.ipc.RpcServer");
- void setSocketSendBufSize(int size);
+ private final boolean authorize;
+ private boolean isSecurityEnabled;
- void start();
+ public static final byte CURRENT_VERSION = 0;
- void stop();
+ /**
+ * How many calls/handler are allowed in the queue.
+ */
+ private static final int DEFAULT_MAX_CALLQUEUE_LENGTH_PER_HANDLER = 10;
- void join() throws InterruptedException;
+ /**
+ * The maximum size that we can hold in the IPC queue
+ */
+ private static final int DEFAULT_MAX_CALLQUEUE_SIZE = 1024 * 1024 * 1024;
- InetSocketAddress getListenerAddress();
+ static final int BUFFER_INITIAL_SIZE = 1024;
- /** Called for each call.
- * @param method Method to invoke.
- * @param param parameter
- * @param receiveTime time
- * @param status
- * @return Message Protobuf response Message and optionally the Cells that make up the response.
- * @throws java.io.IOException e
+ private static final String WARN_DELAYED_CALLS = "hbase.ipc.warn.delayedrpc.number";
+
+ private static final int DEFAULT_WARN_DELAYED_CALLS = 1000;
+
+ private final int warnDelayedCalls;
+
+ private AtomicInteger delayedCalls;
+ private final IPCUtil ipcUtil;
+
+ private static final String AUTH_FAILED_FOR = "Auth failed for ";
+ private static final String AUTH_SUCCESSFUL_FOR = "Auth successful for ";
+ private static final Log AUDITLOG = LogFactory.getLog("SecurityLogger." +
+ Server.class.getName());
+ protected SecretManager<TokenIdentifier> secretManager;
+ protected ServiceAuthorizationManager authManager;
+
+ protected static final ThreadLocal<RpcServerInterface> SERVER = new ThreadLocal<RpcServerInterface>();
+ private volatile boolean started = false;
+
+ /** This is set to Call object before Handler invokes an RPC and reset
+ * after the call returns.
+ */
+ protected static final ThreadLocal<Call> CurCall = new ThreadLocal<Call>();
+
+ protected final InetSocketAddress isa;
+ protected int port; // port we listen on
+ private int handlerCount; // number of handler threads
+ private int priorityHandlerCount;
+ private int readThreads; // number of read threads
+ protected int maxIdleTime; // the maximum idle time after
+ // which a client may be
+ // disconnected
+ protected int thresholdIdleConnections; // the number of idle
+ // connections after which we
+ // will start cleaning up idle
+ // connections
+ int maxConnectionsToNuke; // the max number of
+ // connections to nuke
+ // during a cleanup
+
+ protected MetricsHBaseServer metrics;
+
+ protected final Configuration conf;
+
+ private int maxQueueLength;
+ private int maxQueueSize;
+ protected int socketSendBufferSize;
+ protected final boolean tcpNoDelay; // if T then disable Nagle's Algorithm
+ protected final boolean tcpKeepAlive; // if T then use keepalives
+ protected final long purgeTimeout; // in milliseconds
+
+ volatile protected boolean running = true; // true while server runs
+ protected BlockingQueue<Call> callQueue; // queued calls
+ protected final Counter callQueueSize = new Counter();
+ protected BlockingQueue<Call> priorityCallQueue;
+
+ protected int highPriorityLevel; // what level a high priority call is at
+
+ protected final List<Connection> connectionList =
+ Collections.synchronizedList(new LinkedList<Connection>());
+ //maintain a list
+ //of client connections
+ private Listener listener = null;
+ protected Responder responder = null;
+ protected int numConnections = 0;
+ private Handler[] handlers = null;
+ private Handler[] priorityHandlers = null;
+ /** replication related queue; */
+ protected BlockingQueue<Call> replicationQueue;
+ private int numOfReplicationHandlers = 0;
+ private Handler[] replicationHandlers = null;
+
+ protected HBaseRPCErrorHandler errorHandler = null;
+
+ private static final String WARN_RESPONSE_TIME = "hbase.ipc.warn.response.time";
+ private static final String WARN_RESPONSE_SIZE = "hbase.ipc.warn.response.size";
+
+ /** Default value for above params */
+ private static final int DEFAULT_WARN_RESPONSE_TIME = 10000; // milliseconds
+ private static final int DEFAULT_WARN_RESPONSE_SIZE = 100 * 1024 * 1024;
+
+ private final int warnResponseTime;
+ private final int warnResponseSize;
+ private final Object serverInstance;
+ private final List<BlockingServiceAndInterface> services;
+
+ /**
+ * Datastructure that holds all necessary to a method invocation and then afterward, carries
+ * the result.
+ */
+ class Call implements RpcCallContext {
+ protected int id; // the client's call id
+ protected BlockingService service;
+ protected MethodDescriptor md;
+ protected Message param; // the parameter passed
+ // Optional cell data passed outside of protobufs.
+ protected CellScanner cellScanner;
+ protected Connection connection; // connection to client
+ protected long timestamp; // the time received when response is null
+ // the time served when response is not null
+ protected ByteBuffer response; // the response for this call
+ protected boolean delayResponse;
+ protected Responder responder;
+ protected boolean delayReturnValue; // if the return value should be
+ // set at call completion
+ protected long size; // size of current call
+ protected boolean isError;
+ protected TraceInfo tinfo;
+
+ Call(int id, final BlockingService service, final MethodDescriptor md, Message param,
+ CellScanner cellScanner, Connection connection, Responder responder, long size,
+ TraceInfo tinfo) {
+ this.id = id;
+ this.service = service;
+ this.md = md;
+ this.param = param;
+ this.cellScanner = cellScanner;
+ this.connection = connection;
+ this.timestamp = System.currentTimeMillis();
+ this.response = null;
+ this.delayResponse = false;
+ this.responder = responder;
+ this.isError = false;
+ this.size = size;
+ this.tinfo = tinfo;
+ }
+
+ @Override
+ public String toString() {
+ String serviceName = this.connection.service != null?
+ this.connection.service.getDescriptorForType().getName(): "null";
+ return "callId: " + this.id + " service: " + serviceName + " methodName: " +
+ ((this.md != null)? this.md.getName(): null) + " param: " +
+ (this.param != null? IPCUtil.getRequestShortTextFormat(this.param): "") +
+ " connection: " + connection.toString();
+ }
+
+ protected synchronized void setSaslTokenResponse(ByteBuffer response) {
+ this.response = response;
+ }
+
+ protected synchronized void setResponse(Object m, final CellScanner cells,
+ Throwable t, String errorMsg) {
+ if (this.isError) return;
+ if (t != null) this.isError = true;
+ ByteBufferOutputStream bbos = null;
+ try {
+ ResponseHeader.Builder headerBuilder = ResponseHeader.newBuilder();
+ // Presume it a pb Message. Could be null.
+ Message result = (Message)m;
+ // Call id.
+ headerBuilder.setCallId(this.id);
+ if (t != null) {
+ ExceptionResponse.Builder exceptionBuilder = ExceptionResponse.newBuilder();
+ exceptionBuilder.setExceptionClassName(t.getClass().getName());
+ exceptionBuilder.setStackTrace(errorMsg);
+ exceptionBuilder.setDoNotRetry(t instanceof DoNotRetryIOException);
+ if (t instanceof RegionMovedException) {
+ // Special casing for this exception. This is only one carrying a payload.
+ // Do this instead of build a generic system for allowing exceptions carry
+ // any kind of payload.
+ RegionMovedException rme = (RegionMovedException)t;
+ exceptionBuilder.setHostname(rme.getHostname());
+ exceptionBuilder.setPort(rme.getPort());
+ }
+ // Set the exception as the result of the method invocation.
+ headerBuilder.setException(exceptionBuilder.build());
+ }
+ ByteBuffer cellBlock =
+ ipcUtil.buildCellBlock(this.connection.codec, this.connection.compressionCodec, cells);
+ if (cellBlock != null) {
+ CellBlockMeta.Builder cellBlockBuilder = CellBlockMeta.newBuilder();
+ // Presumes the cellBlock bytebuffer has been flipped so limit has total size in it.
+ cellBlockBuilder.setLength(cellBlock.limit());
+ headerBuilder.setCellBlockMeta(cellBlockBuilder.build());
+ }
+ Message header = headerBuilder.build();
+ bbos = IPCUtil.write(header, result, cellBlock);
+ if (connection.useWrap) {
+ wrapWithSasl(bbos);
+ }
+ } catch (IOException e) {
+ LOG.warn("Exception while creating response " + e);
+ }
+ ByteBuffer bb = null;
+ if (bbos != null) {
+ // TODO: If SASL, maybe buffer already been flipped and written?
+ bb = bbos.getByteBuffer();
+ bb.position(0);
+ }
+ this.response = bb;
+ }
+
+ private void wrapWithSasl(ByteBufferOutputStream response)
+ throws IOException {
+ if (connection.useSasl) {
+ // getByteBuffer calls flip()
+ ByteBuffer buf = response.getByteBuffer();
+ byte[] token;
+ // synchronization may be needed since there can be multiple Handler
+ // threads using saslServer to wrap responses.
+ synchronized (connection.saslServer) {
+ token = connection.saslServer.wrap(buf.array(),
+ buf.arrayOffset(), buf.remaining());
+ }
+ if (LOG.isDebugEnabled())
+ LOG.debug("Adding saslServer wrapped token of size " + token.length
+ + " as call response.");
+ buf.clear();
+ DataOutputStream saslOut = new DataOutputStream(response);
+ saslOut.writeInt(token.length);
+ saslOut.write(token, 0, token.length);
+ }
+ }
+
+ @Override
+ public synchronized void endDelay(Object result) throws IOException {
+ assert this.delayResponse;
+ assert this.delayReturnValue || result == null;
+ this.delayResponse = false;
+ delayedCalls.decrementAndGet();
+ if (this.delayReturnValue) {
+ this.setResponse(result, null, null, null);
+ }
+ this.responder.doRespond(this);
+ }
+
+ @Override
+ public synchronized void endDelay() throws IOException {
+ this.endDelay(null);
+ }
+
+ @Override
+ public synchronized void startDelay(boolean delayReturnValue) {
+ assert !this.delayResponse;
+ this.delayResponse = true;
+ this.delayReturnValue = delayReturnValue;
+ int numDelayed = delayedCalls.incrementAndGet();
+ if (numDelayed > warnDelayedCalls) {
+ LOG.warn("Too many delayed calls: limit " + warnDelayedCalls + " current " + numDelayed);
+ }
+ }
+
+ @Override
+ public synchronized void endDelayThrowing(Throwable t) throws IOException {
+ this.setResponse(null, null, t, StringUtils.stringifyException(t));
+ this.delayResponse = false;
+ this.sendResponseIfReady();
+ }
+
+ @Override
+ public synchronized boolean isDelayed() {
+ return this.delayResponse;
+ }
+
+ @Override
+ public synchronized boolean isReturnValueDelayed() {
+ return this.delayReturnValue;
+ }
+
+ @Override
+ public void throwExceptionIfCallerDisconnected() throws CallerDisconnectedException {
+ if (!connection.channel.isOpen()) {
+ long afterTime = System.currentTimeMillis() - timestamp;
+ throw new CallerDisconnectedException(
+ "Aborting call " + this + " after " + afterTime + " ms, since " +
+ "caller disconnected");
+ }
+ }
+
+ public long getSize() {
+ return this.size;
+ }
+
+ /**
+ * If we have a response, and delay is not set, then respond
+ * immediately. Otherwise, do not respond to client. This is
+ * called the by the RPC code in the context of the Handler thread.
+ */
+ public synchronized void sendResponseIfReady() throws IOException {
+ if (!this.delayResponse) {
+ this.responder.doRespond(this);
+ }
+ }
+ }
+
+ /** Listens on the socket. Creates jobs for the handler threads*/
+ private class Listener extends Thread {
+
+ private ServerSocketChannel acceptChannel = null; //the accept channel
+ private Selector selector = null; //the selector that we use for the server
+ private Reader[] readers = null;
+ private int currentReader = 0;
+ private Random rand = new Random();
+ private long lastCleanupRunTime = 0; //the last time when a cleanup connec-
+ //-tion (for idle connections) ran
+ private long cleanupInterval = 10000; //the minimum interval between
+ //two cleanup runs
+ private int backlogLength = conf.getInt("ipc.server.listen.queue.size", 128);
+
+ private ExecutorService readPool;
+
+ public Listener(final String name) throws IOException {
+ super(name);
+ // Create a new server socket and set to non blocking mode
+ acceptChannel = ServerSocketChannel.open();
+ acceptChannel.configureBlocking(false);
+
+ // Bind the server socket to the local host and port
+ bind(acceptChannel.socket(), isa, backlogLength);
+ port = acceptChannel.socket().getLocalPort(); //Could be an ephemeral port
+ // create a selector;
+ selector= Selector.open();
+
+ readers = new Reader[readThreads];
+ readPool = Executors.newFixedThreadPool(readThreads,
+ new ThreadFactoryBuilder().setNameFormat(
+ "IPC Reader %d on port " + port).setDaemon(true).build());
+ for (int i = 0; i < readThreads; ++i) {
+ Reader reader = new Reader();
+ readers[i] = reader;
+ readPool.execute(reader);
+ }
+ LOG.info(getName() + ": started " + readThreads + " reader(s).");
+
+ // Register accepts on the server socket with the selector.
+ acceptChannel.register(selector, SelectionKey.OP_ACCEPT);
+ this.setName("IPC Server listener on " + port);
+ this.setDaemon(true);
+ }
+
+
+ private class Reader implements Runnable {
+ private volatile boolean adding = false;
+ private final Selector readSelector;
+
+ Reader() throws IOException {
+ this.readSelector = Selector.open();
+ }
+ public void run() {
+ try {
+ doRunLoop();
+ } finally {
+ try {
+ readSelector.close();
+ } catch (IOException ioe) {
+ LOG.error(getName() + ": error closing read selector in " + getName(), ioe);
+ }
+ }
+ }
+
+ private synchronized void doRunLoop() {
+ while (running) {
+ SelectionKey key = null;
+ try {
+ readSelector.select();
+ while (adding) {
+ this.wait(1000);
+ }
+
+ Iterator<SelectionKey> iter = readSelector.selectedKeys().iterator();
+ while (iter.hasNext()) {
+ key = iter.next();
+ iter.remove();
+ if (key.isValid()) {
+ if (key.isReadable()) {
+ doRead(key);
+ }
+ }
+ key = null;
+ }
+ } catch (InterruptedException e) {
+ if (running) { // unexpected -- log it
+ LOG.info(getName() + ": unexpectedly interrupted: " +
+ StringUtils.stringifyException(e));
+ }
+ } catch (IOException ex) {
+ LOG.error(getName() + ": error in Reader", ex);
+ }
+ }
+ }
+
+ /**
+ * This gets reader into the state that waits for the new channel
+ * to be registered with readSelector. If it was waiting in select()
+ * the thread will be woken up, otherwise whenever select() is called
+ * it will return even if there is nothing to read and wait
+ * in while(adding) for finishAdd call
+ */
+ public void startAdd() {
+ adding = true;
+ readSelector.wakeup();
+ }
+
+ public synchronized SelectionKey registerChannel(SocketChannel channel)
+ throws IOException {
+ return channel.register(readSelector, SelectionKey.OP_READ);
+ }
+
+ public synchronized void finishAdd() {
+ adding = false;
+ this.notify();
+ }
+ }
+
+ /** cleanup connections from connectionList. Choose a random range
+ * to scan and also have a limit on the number of the connections
+ * that will be cleanedup per run. The criteria for cleanup is the time
+ * for which the connection was idle. If 'force' is true then all
+ * connections will be looked at for the cleanup.
+ * @param force all connections will be looked at for cleanup
+ */
+ private void cleanupConnections(boolean force) {
+ if (force || numConnections > thresholdIdleConnections) {
+ long currentTime = System.currentTimeMillis();
+ if (!force && (currentTime - lastCleanupRunTime) < cleanupInterval) {
+ return;
+ }
+ int start = 0;
+ int end = numConnections - 1;
+ if (!force) {
+ start = rand.nextInt() % numConnections;
+ end = rand.nextInt() % numConnections;
+ int temp;
+ if (end < start) {
+ temp = start;
+ start = end;
+ end = temp;
+ }
+ }
+ int i = start;
+ int numNuked = 0;
+ while (i <= end) {
+ Connection c;
+ synchronized (connectionList) {
+ try {
+ c = connectionList.get(i);
+ } catch (Exception e) {return;}
+ }
+ if (c.timedOut(currentTime)) {
+ if (LOG.isDebugEnabled())
+ LOG.debug(getName() + ": disconnecting client " + c.getHostAddress());
+ closeConnection(c);
+ numNuked++;
+ end--;
+ //noinspection UnusedAssignment
+ c = null;
+ if (!force && numNuked == maxConnectionsToNuke) break;
+ }
+ else i++;
+ }
+ lastCleanupRunTime = System.currentTimeMillis();
+ }
+ }
+
+ @Override
+ public void run() {
+ LOG.info(getName() + ": starting");
+ SERVER.set(RpcServer.this);
+
+ while (running) {
+ SelectionKey key = null;
+ try {
+ selector.select(); // FindBugs IS2_INCONSISTENT_SYNC
+ Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
+ while (iter.hasNext()) {
+ key = iter.next();
+ iter.remove();
+ try {
+ if (key.isValid()) {
+ if (key.isAcceptable())
+ doAccept(key);
+ }
+ } catch (IOException ignored) {
+ }
+ key = null;
+ }
+ } catch (OutOfMemoryError e) {
+ if (errorHandler != null) {
+ if (errorHandler.checkOOME(e)) {
+ LOG.info(getName() + ": exiting on OutOfMemoryError");
+ closeCurrentConnection(key, e);
+ cleanupConnections(true);
+ return;
+ }
+ } else {
+ // we can run out of memory if we have too many threads
+ // log the event and sleep for a minute and give
+ // some thread(s) a chance to finish
+ LOG.warn(getName() + ": OutOfMemoryError in server select", e);
+ closeCurrentConnection(key, e);
+ cleanupConnections(true);
+ try { Thread.sleep(60000); } catch (Exception ignored) {}
+ }
+ } catch (Exception e) {
+ closeCurrentConnection(key, e);
+ }
+ cleanupConnections(false);
+ }
+ LOG.info(getName() + ": stopping");
+
+ synchronized (this) {
+ try {
+ acceptChannel.close();
+ selector.close();
+ } catch (IOException ignored) { }
+
+ selector= null;
+ acceptChannel= null;
+
+ // clean up all connections
+ while (!connectionList.isEmpty()) {
+ closeConnection(connectionList.remove(0));
+ }
+ }
+ }
+
+ private void closeCurrentConnection(SelectionKey key, Throwable e) {
+ if (key != null) {
+ Connection c = (Connection)key.attachment();
+ if (c != null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(getName() + ": disconnecting client " + c.getHostAddress() +
+ (e != null ? " on error " + e.getMessage() : ""));
+ }
+ closeConnection(c);
+ key.attach(null);
+ }
+ }
+ }
+
+ InetSocketAddress getAddress() {
+ return (InetSocketAddress)acceptChannel.socket().getLocalSocketAddress();
+ }
+
+ void doAccept(SelectionKey key) throws IOException, OutOfMemoryError {
+ Connection c;
+ ServerSocketChannel server = (ServerSocketChannel) key.channel();
+
+ SocketChannel channel;
+ while ((channel = server.accept()) != null) {
+ channel.configureBlocking(false);
+ channel.socket().setTcpNoDelay(tcpNoDelay);
+ channel.socket().setKeepAlive(tcpKeepAlive);
+
+ Reader reader = getReader();
+ try {
+ reader.startAdd();
+ SelectionKey readKey = reader.registerChannel(channel);
+ c = getConnection(channel, System.currentTimeMillis());
+ readKey.attach(c);
+ synchronized (connectionList) {
+ connectionList.add(numConnections, c);
+ numConnections++;
+ }
+ if (LOG.isDebugEnabled())
+ LOG.debug(getName() + ": connection from " + c.toString() +
+ "; # active connections: " + numConnections +
+ "; # queued calls: " + callQueue.size());
+ } finally {
+ reader.finishAdd();
+ }
+ }
+ }
+
+ void doRead(SelectionKey key) throws InterruptedException {
+ int count = 0;
+ Connection c = (Connection)key.attachment();
+ if (c == null) {
+ return;
+ }
+ c.setLastContact(System.currentTimeMillis());
+ try {
+ count = c.readAndProcess();
+ } catch (InterruptedException ieo) {
+ throw ieo;
+ } catch (Exception e) {
+ LOG.warn(getName() + ": count of bytes read: " + count, e);
+ count = -1; //so that the (count < 0) block is executed
+ }
+ if (count < 0) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(getName() + ": DISCONNECTING client " + c.toString() +
+ " because read count=" + count +
+ ". Number of active connections: " + numConnections);
+ }
+ closeConnection(c);
+ // c = null;
+ } else {
+ c.setLastContact(System.currentTimeMillis());
+ }
+ }
+
+ synchronized void doStop() {
+ if (selector != null) {
+ selector.wakeup();
+ Thread.yield();
+ }
+ if (acceptChannel != null) {
+ try {
+ acceptChannel.socket().close();
+ } catch (IOException e) {
+ LOG.info(getName() + ": exception in closing listener socket. " + e);
+ }
+ }
+ readPool.shutdownNow();
+ }
+
+ // The method that will return the next reader to work with
+ // Simplistic implementation of round robin for now
+ Reader getReader() {
+ currentReader = (currentReader + 1) % readers.length;
+ return readers[currentReader];
+ }
+ }
+
+ // Sends responses of RPC back to clients.
+ protected class Responder extends Thread {
+ private final Selector writeSelector;
+ private int pending; // connections waiting to register
+
+ Responder() throws IOException {
+ this.setName("IPC Server Responder");
+ this.setDaemon(true);
+ writeSelector = Selector.open(); // create a selector
+ pending = 0;
+ }
+
+ @Override
+ public void run() {
+ LOG.info(getName() + ": starting");
+ SERVER.set(RpcServer.this);
+ try {
+ doRunLoop();
+ } finally {
+ LOG.info(getName() + ": stopping");
+ try {
+ writeSelector.close();
+ } catch (IOException ioe) {
+ LOG.error(getName() + ": couldn't close write selector", ioe);
+ }
+ }
+ }
+
+ private void doRunLoop() {
+ long lastPurgeTime = 0; // last check for old calls.
+
+ while (running) {
+ try {
+ waitPending(); // If a channel is being registered, wait.
+ writeSelector.select(purgeTimeout);
+ Iterator<SelectionKey> iter = writeSelector.selectedKeys().iterator();
+ while (iter.hasNext()) {
+ SelectionKey key = iter.next();
+ iter.remove();
+ try {
+ if (key.isValid() && key.isWritable()) {
+ doAsyncWrite(key);
+ }
+ } catch (IOException e) {
+ LOG.info(getName() + ": asyncWrite", e);
+ }
+ }
+ long now = System.currentTimeMillis();
+ if (now < lastPurgeTime + purgeTimeout) {
+ continue;
+ }
+ lastPurgeTime = now;
+ //
+ // If there were some calls that have not been sent out for a
+ // long time, discard them.
+ //
+ if (LOG.isDebugEnabled()) LOG.debug(getName() + ": checking for old call responses.");
+ ArrayList<Call> calls;
+
+ // get the list of channels from list of keys.
+ synchronized (writeSelector.keys()) {
+ calls = new ArrayList<Call>(writeSelector.keys().size());
+ iter = writeSelector.keys().iterator();
+ while (iter.hasNext()) {
+ SelectionKey key = iter.next();
+ Call call = (Call)key.attachment();
+ if (call != null && key.channel() == call.connection.channel) {
+ calls.add(call);
+ }
+ }
+ }
+
+ for(Call call : calls) {
+ try {
+ doPurge(call, now);
+ } catch (IOException e) {
+ LOG.warn(getName() + ": error in purging old calls " + e);
+ }
+ }
+ } catch (OutOfMemoryError e) {
+ if (errorHandler != null) {
+ if (errorHandler.checkOOME(e)) {
+ LOG.info(getName() + ": exiting on OutOfMemoryError");
+ return;
+ }
+ } else {
+ //
+ // we can run out of memory if we have too many threads
+ // log the event and sleep for a minute and give
+ // some thread(s) a chance to finish
+ //
+ LOG.warn(getName() + ": OutOfMemoryError in server select", e);
+ try { Thread.sleep(60000); } catch (Exception ignored) {}
+ }
+ } catch (Exception e) {
+ LOG.warn(getName() + ": exception in Responder " +
+ StringUtils.stringifyException(e));
+ }
+ }
+ LOG.info(getName() + ": stopped");
+ }
+
+ private void doAsyncWrite(SelectionKey key) throws IOException {
+ Call call = (Call)key.attachment();
+ if (call == null) {
+ return;
+ }
+ if (key.channel() != call.connection.channel) {
+ throw new IOException("doAsyncWrite: bad channel");
+ }
+
+ synchronized(call.connection.responseQueue) {
+ if (processResponse(call.connection.responseQueue, false)) {
+ try {
+ key.interestOps(0);
+ } catch (CancelledKeyException e) {
+ /* The Listener/reader might have closed the socket.
+ * We don't explicitly cancel the key, so not sure if this will
+ * ever fire.
+ * This warning could be removed.
+ */
+ LOG.warn("Exception while changing ops : " + e);
+ }
+ }
+ }
+ }
+
+ //
+ // Remove calls that have been pending in the responseQueue
+ // for a long time.
+ //
+ private void doPurge(Call call, long now) throws IOException {
+ synchronized (call.connection.responseQueue) {
+ Iterator<Call> iter = call.connection.responseQueue.listIterator(0);
+ while (iter.hasNext()) {
+ Call nextCall = iter.next();
+ if (now > nextCall.timestamp + purgeTimeout) {
+ closeConnection(nextCall.connection);
+ break;
+ }
+ }
+ }
+ }
+
+ // Processes one response. Returns true if there are no more pending
+ // data for this channel.
+ //
+ private boolean processResponse(final LinkedList<Call> responseQueue, boolean inHandler)
+ throws IOException {
+ boolean error = true;
+ boolean done = false; // there is more data for this channel.
+ int numElements;
+ Call call = null;
+ try {
+ //noinspection SynchronizationOnLocalVariableOrMethodParameter
+ synchronized (responseQueue) {
+ //
+ // If there are no items for this channel, then we are done
+ //
+ numElements = responseQueue.size();
+ if (numElements == 0) {
+ error = false;
+ return true; // no more data for this channel.
+ }
+ //
+ // Extract the first call
+ //
+ call = responseQueue.removeFirst();
+ SocketChannel channel = call.connection.channel;
+ //
+ // Send as much data as we can in the non-blocking fashion
+ //
+ int numBytes = channelWrite(channel, call.response);
+ if (numBytes < 0) {
+ return true;
+ }
+ if (!call.response.hasRemaining()) {
+ call.connection.decRpcCount();
+ //noinspection RedundantIfStatement
+ if (numElements == 1) { // last call fully processes.
+ done = true; // no more data for this channel.
+ } else {
+ done = false; // more calls pending to be sent.
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(getName() + ": callId: " + call.id + " wrote " + numBytes + " bytes.");
+ }
+ } else {
+ //
+ // If we were unable to write the entire response out, then
+ // insert in Selector queue.
+ //
+ call.connection.responseQueue.addFirst(call);
+
+ if (inHandler) {
+ // set the serve time when the response has to be sent later
+ call.timestamp = System.currentTimeMillis();
+ if (enqueueInSelector(call))
+ done = true;
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(getName() + call.toString() + " partially sent, wrote " +
+ numBytes + " bytes.");
+ }
+ }
+ error = false; // everything went off well
+ }
+ } finally {
+ if (error && call != null) {
+ LOG.warn(getName() + call.toString() + ": output error");
+ done = true; // error. no more data for this channel.
+ closeConnection(call.connection);
+ }
+ }
+ return done;
+ }
+
+ //
+ // Enqueue for background thread to send responses out later.
+ //
+ private boolean enqueueInSelector(Call call) throws IOException {
+ boolean done = false;
+ incPending();
+ try {
+ // Wake up the thread blocked on select, only then can the call
+ // to channel.register() complete.
+ SocketChannel channel = call.connection.channel;
+ writeSelector.wakeup();
+ channel.register(writeSelector, SelectionKey.OP_WRITE, call);
+ } catch (ClosedChannelException e) {
+ //It's OK. Channel might be closed else where.
+ done = true;
+ } finally {
+ decPending();
+ }
+ return done;
+ }
+
+ //
+ // Enqueue a response from the application.
+ //
+ void doRespond(Call call) throws IOException {
+ // set the serve time when the response has to be sent later
+ call.timestamp = System.currentTimeMillis();
+
+ boolean doRegister = false;
+ synchronized (call.connection.responseQueue) {
+ call.connection.responseQueue.addLast(call);
+ if (call.connection.responseQueue.size() == 1) {
+ doRegister = !processResponse(call.connection.responseQueue, false);
+ }
+ }
+ if (doRegister) {
+ enqueueInSelector(call);
+ }
+ }
+
+ private synchronized void incPending() { // call waiting to be enqueued.
+ pending++;
+ }
+
+ private synchronized void decPending() { // call done enqueueing.
+ pending--;
+ notify();
+ }
+
+ private synchronized void waitPending() throws InterruptedException {
+ while (pending > 0) {
+ wait();
+ }
+ }
+ }
+
+ @SuppressWarnings("serial")
+ public static class CallQueueTooBigException extends IOException {
+ CallQueueTooBigException() {
+ super();
+ }
+ }
+
+ private Function<Pair<RequestHeader, Message>, Integer> qosFunction = null;
+
+ /**
+ * Gets the QOS level for this call. If it is higher than the highPriorityLevel and there
+ * are priorityHandlers available it will be processed in it's own thread set.
+ *
+ * @param newFunc
+ */
+ @Override
+ public void setQosFunction(Function<Pair<RequestHeader, Message>, Integer> newFunc) {
+ qosFunction = newFunc;
+ }
+
+ protected int getQosLevel(Pair<RequestHeader, Message> headerAndParam) {
+ if (qosFunction == null) return 0;
+ Integer res = qosFunction.apply(headerAndParam);
+ return res == null? 0: res;
+ }
+
+ /** Reads calls from a connection and queues them for handling. */
+ @edu.umd.cs.findbugs.annotations.SuppressWarnings(
+ value="VO_VOLATILE_INCREMENT",
+ justification="False positive according to http://sourceforge.net/p/findbugs/bugs/1032/")
+ public class Connection {
+ // If initial preamble with version and magic has been read or not.
+ private boolean connectionPreambleRead = false;
+ // If the connection header has been read or not.
+ private boolean connectionHeaderRead = false;
+ protected SocketChannel channel;
+ private ByteBuffer data;
+ private ByteBuffer dataLengthBuffer;
+ protected final LinkedList<Call> responseQueue;
+ private volatile int rpcCount = 0; // number of outstanding rpcs
+ private long lastContact;
+ private InetAddress addr;
+ protected Socket socket;
+ // Cache the remote host & port info so that even if the socket is
+ // disconnected, we can say where it used to connect to.
+ protected String hostAddress;
+ protected int remotePort;
+ ConnectionHeader connectionHeader;
+ /**
+ * Codec the client asked use.
+ */
+ private Codec codec;
+ /**
+ * Compression codec the client asked us use.
+ */
+ private CompressionCodec compressionCodec;
+ BlockingService service;
+ protected UserGroupInformation user = null;
+ private AuthMethod authMethod;
+ private boolean saslContextEstablished;
+ private boolean skipInitialSaslHandshake;
+ private ByteBuffer unwrappedData;
+ // When is this set? FindBugs wants to know! Says NP
+ private ByteBuffer unwrappedDataLengthBuffer = ByteBuffer.allocate(4);
+ boolean useSasl;
+ SaslServer saslServer;
+ private boolean useWrap = false;
+ // Fake 'call' for failed authorization response
+ private static final int AUTHROIZATION_FAILED_CALLID = -1;
+ private final Call authFailedCall =
+ new Call(AUTHROIZATION_FAILED_CALLID, this.service, null, null, null, this, null, 0, null);
+ private ByteArrayOutputStream authFailedResponse =
+ new ByteArrayOutputStream();
+ // Fake 'call' for SASL context setup
+ private static final int SASL_CALLID = -33;
+ private final Call saslCall =
+ new Call(SASL_CALLID, this.service, null, null, null, this, null, 0, null);
+
+ public UserGroupInformation attemptingUser = null; // user name before auth
+
+ public Connection(SocketChannel channel, long lastContact) {
+ this.channel = channel;
+ this.lastContact = lastContact;
+ this.data = null;
+ this.dataLengthBuffer = ByteBuffer.allocate(4);
+ this.socket = channel.socket();
+ this.addr = socket.getInetAddress();
+ if (addr == null) {
+ this.hostAddress = "*Unknown*";
+ } else {
+ this.hostAddress = addr.getHostAddress();
+ }
+ this.remotePort = socket.getPort();
+ this.responseQueue = new LinkedList<Call>();
+ if (socketSendBufferSize != 0) {
+ try {
+ socket.setSendBufferSize(socketSendBufferSize);
+ } catch (IOException e) {
+ LOG.warn("Connection: unable to set socket send buffer size to " +
+ socketSendBufferSize);
+ }
+ }
+ }
+
+ @Override
+ public String toString() {
+ return getHostAddress() + ":" + remotePort;
+ }
+
+ public String getHostAddress() {
+ return hostAddress;
+ }
+
+ public InetAddress getHostInetAddress() {
+ return addr;
+ }
+
+ public int getRemotePort() {
+ return remotePort;
+ }
+
+ public void setLastContact(long lastContact) {
+ this.lastContact = lastContact;
+ }
+
+ public long getLastContact() {
+ return lastContact;
+ }
+
+ /* Return true if the connection has no outstanding rpc */
+ private boolean isIdle() {
+ return rpcCount == 0;
+ }
+
+ /* Decrement the outstanding RPC count */
+ protected void decRpcCount() {
+ rpcCount--;
+ }
+
+ /* Increment the outstanding RPC count */
+ protected void incRpcCount() {
+ rpcCount++;
+ }
+
+ protected boolean timedOut(long currentTime) {
+ return isIdle() && currentTime - lastContact > maxIdleTime;
+ }
+
+ private UserGroupInformation getAuthorizedUgi(String authorizedId)
+ throws IOException {
+ if (authMethod == AuthMethod.DIGEST) {
+ TokenIdentifier tokenId = HBaseSaslRpcServer.getIdentifier(authorizedId,
+ secretManager);
+ UserGroupInformation ugi = tokenId.getUser();
+ if (ugi == null) {
+ throw new AccessControlException(
+ "Can't retrieve username from tokenIdentifier.");
+ }
+ ugi.addTokenIdentifier(tokenId);
+ return ugi;
+ } else {
+ return UserGroupInformation.createRemoteUser(authorizedId);
+ }
+ }
+
+ private void saslReadAndProcess(byte[] saslToken) throws IOException,
+ InterruptedException {
+ if (saslContextEstablished) {
+ if (LOG.isDebugEnabled())
+ LOG.debug("Have read input token of size " + saslToken.length
+ + " for processing by saslServer.unwrap()");
+
+ if (!useWrap) {
+ processOneRpc(saslToken);
+ } else {
+ byte [] plaintextData = saslServer.unwrap(saslToken, 0, saslToken.length);
+ processUnwrappedData(plaintextData);
+ }
+ } else {
+ byte[] replyToken = null;
+ try {
+ if (saslServer == null) {
+ switch (authMethod) {
+ case DIGEST:
+ if (secretManager == null) {
+ throw new AccessControlException(
+ "Server is not configured to do DIGEST authentication.");
+ }
+ saslServer = Sasl.createSaslServer(AuthMethod.DIGEST
+ .getMechanismName(), null, SaslUtil.SASL_DEFAULT_REALM,
+ SaslUtil.SASL_PROPS, new SaslDigestCallbackHandler(
+ secretManager, this));
+ break;
+ default:
+ UserGroupInformation current = UserGroupInformation
+ .getCurrentUser();
+ String fullName = current.getUserName();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Kerberos principal name is " + fullName);
+ }
+ final String names[] = SaslUtil.splitKerberosName(fullName);
+ if (names.length != 3) {
+ throw new AccessControlException(
+ "Kerberos principal name does NOT have the expected "
+ + "hostname part: " + fullName);
+ }
+ current.doAs(new PrivilegedExceptionAction<Object>() {
+ @Override
+ public Object run() throws SaslException {
+ saslServer = Sasl.createSaslServer(AuthMethod.KERBEROS
+ .getMechanismName(), names[0], names[1],
+ SaslUtil.SASL_PROPS, new SaslGssCallbackHandler());
+ return null;
+ }
+ });
+ }
+ if (saslServer == null)
+ throw new AccessControlException(
+ "Unable to find SASL server implementation for "
+ + authMethod.getMechanismName());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Created SASL server with mechanism = " + authMethod.getMechanismName());
+ }
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Have read input token of size " + saslToken.length
+ + " for processing by saslServer.evaluateResponse()");
+ }
+ replyToken = saslServer.evaluateResponse(saslToken);
+ } catch (IOException e) {
+ IOException sendToClient = e;
+ Throwable cause = e;
+ while (cause != null) {
+ if (cause instanceof InvalidToken) {
+ sendToClient = (InvalidToken) cause;
+ break;
+ }
+ cause = cause.getCause();
+ }
+ doRawSaslReply(SaslStatus.ERROR, null, sendToClient.getClass().getName(),
+ sendToClient.getLocalizedMessage());
+ metrics.authenticationFailure();
+ String clientIP = this.toString();
+ // attempting user could be null
+ AUDITLOG.warn(AUTH_FAILED_FOR + clientIP + ":" + attemptingUser);
+ throw e;
+ }
+ if (replyToken != null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Will send token of size " + replyToken.length
+ + " from saslServer.");
+ }
+ doRawSaslReply(SaslStatus.SUCCESS, new BytesWritable(replyToken), null,
+ null);
+ }
+ if (saslServer.isComplete()) {
+ String qop = (String) saslServer.getNegotiatedProperty(Sasl.QOP);
+ useWrap = qop != null && !"auth".equalsIgnoreCase(qop);
+ user = getAuthorizedUgi(saslServer.getAuthorizationID());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("SASL server context established. Authenticated client: "
+ + user + ". Negotiated QoP is "
+ + saslServer.getNegotiatedProperty(Sasl.QOP));
+ }
+ metrics.authenticationSuccess();
+ AUDITLOG.info(AUTH_SUCCESSFUL_FOR + user);
+ saslContextEstablished = true;
+ }
+ }
+ }
+ /**
+ * No protobuf encoding of raw sasl messages
+ */
+ private void doRawSaslReply(SaslStatus status, Writable rv,
+ String errorClass, String error) throws IOException {
+ //In my testing, have noticed that sasl messages are usually
+ //in the ballpark of 100-200. That's why the initialcapacity is 256.
+ ByteBufferOutputStream saslResponse = new ByteBufferOutputStream(256);
+ DataOutputStream out = new DataOutputStream(saslResponse);
+ out.writeInt(status.state); // write status
+ if (status == SaslStatus.SUCCESS) {
+ rv.write(out);
+ } else {
+ WritableUtils.writeString(out, errorClass);
+ WritableUtils.writeString(out, error);
+ }
+ saslCall.setSaslTokenResponse(saslResponse.getByteBuffer());
+ saslCall.responder = responder;
+ saslCall.sendResponseIfReady();
+ }
+
+ private void disposeSasl() {
+ if (saslServer != null) {
+ try {
+ saslServer.dispose();
+ saslServer = null;
+ } catch (SaslException ignored) {
+ }
+ }
+ }
+
+ /**
+ * Read off the wire.
+ * @return Returns -1 if failure (and caller will close connection) else return how many
+ * bytes were read and processed
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ public int readAndProcess() throws IOException, InterruptedException {
+ while (true) {
+ // Try and read in an int. If new connection, the int will hold the 'HBas' HEADER. If it
+ // does, read in the rest of the connection preamble, the version and the auth method.
+ // Else it will be length of the data to read (or -1 if a ping). We catch the integer
+ // length into the 4-byte this.dataLengthBuffer.
+ int count;
+ if (this.dataLengthBuffer.remaining() > 0) {
+ count = channelRead(channel, this.dataLengthBuffer);
+ if (count < 0 || this.dataLengthBuffer.remaining() > 0) {
+ return count;
+ }
+ }
+ // If we have not read the connection setup preamble, look to see if that is on the wire.
+ if (!connectionPreambleRead) {
+ // Check for 'HBas' magic.
+ this.dataLengthBuffer.flip();
+ if (!HConstants.RPC_HEADER.equals(dataLengthBuffer)) {
+ return doBadPreambleHandling("Expected HEADER=" +
+ Bytes.toStringBinary(HConstants.RPC_HEADER.array()) +
+ " but received HEADER=" + Bytes.toStringBinary(dataLengthBuffer.array()) +
+ " from " + toString());
+ }
+ // Now read the next two bytes, the version and the auth to use.
+ ByteBuffer versionAndAuthBytes = ByteBuffer.allocate(2);
+ count = channelRead(channel, versionAndAuthBytes);
+ if (count < 0 || versionAndAuthBytes.remaining() > 0) {
+ return count;
+ }
+ int version = versionAndAuthBytes.get(0);
+ byte authbyte = versionAndAuthBytes.get(1);
+ this.authMethod = AuthMethod.valueOf(authbyte);
+ if (version != CURRENT_VERSION) {
+ String msg = getFatalConnectionString(version, authbyte);
+ return doBadPreambleHandling(msg, new WrongVersionException(msg));
+ }
+ if (authMethod == null) {
+ String msg = getFatalConnectionString(version, authbyte);
+ return doBadPreambleHandling(msg, new BadAuthException(msg));
+ }
+ if (isSecurityEnabled && authMethod == AuthMethod.SIMPLE) {
+ AccessControlException ae = new AccessControlException("Authentication is required");
+ setupResponse(authFailedResponse, authFailedCall, ae, ae.getMessage());
+ responder.doRespond(authFailedCall);
+ throw ae;
+ }
+ if (!isSecurityEnabled && authMethod != AuthMethod.SIMPLE) {
+ doRawSaslReply(SaslStatus.SUCCESS, new IntWritable(
+ SaslUtil.SWITCH_TO_SIMPLE_AUTH), null, null);
+ authMethod = AuthMethod.SIMPLE;
+ // client has already sent the initial Sasl message and we
+ // should ignore it. Both client and server should fall back
+ // to simple auth from now on.
+ skipInitialSaslHandshake = true;
+ }
+ if (authMethod != AuthMethod.SIMPLE) {
+ useSasl = true;
+ }
+ connectionPreambleRead = true;
+ // Preamble checks out. Go around again to read actual connection header.
+ dataLengthBuffer.clear();
+ continue;
+ }
+ // We have read a length and we have read the preamble. It is either the connection header
+ // or it is a request.
+ if (data == null) {
+ dataLengthBuffer.flip();
+ int dataLength = dataLengthBuffer.getInt();
+ if (dataLength == RpcClient.PING_CALL_ID) {
+ if (!useWrap) { //covers the !useSasl too
+ dataLengthBuffer.clear();
+ return 0; //ping message
+ }
+ }
+ if (dataLength < 0) {
+ throw new IllegalArgumentException("Unexpected data length "
+ + dataLength + "!! from " + getHostAddress());
+ }
+ data = ByteBuffer.allocate(dataLength);
+ incRpcCount(); // Increment the rpc count
+ }
+ count = channelRead(channel, data);
+ if (data.remaining() == 0) {
+ dataLengthBuffer.clear();
+ data.flip();
+ if (skipInitialSaslHandshake) {
+ data = null;
+ skipInitialSaslHandshake = false;
+ continue;
+ }
+ boolean headerRead = connectionHeaderRead;
+ if (useSasl) {
+ saslReadAndProcess(data.array());
+ } else {
+ processOneRpc(data.array());
+ }
+ this.data = null;
+ if (!headerRead) {
+ continue;
+ }
+ } else {
+ // More to read still; go around again.
+ if (LOG.isTraceEnabled()) LOG.trace("Continue to read rest of data " + data.remaining());
+ continue;
+ }
+ return count;
+ }
+ }
+
+ private String getFatalConnectionString(final int version, final byte authByte) {
+ return "serverVersion=" + CURRENT_VERSION +
+ ", clientVersion=" + version + ", authMethod=" + authByte +
+ ", authSupported=" + (authMethod != null) + " from " + toString();
+ }
+
+ private int doBadPreambleHandling(final String msg) throws IOException {
+ return doBadPreambleHandling(msg, new FatalConnectionException(msg));
+ }
+
+ private int doBadPreambleHandling(final String msg, final Exception e) throws IOException {
+ LOG.warn(msg);
+ Call fakeCall = new Call(-1, null, null, null, null, this, responder, -1, null);
+ setupResponse(null, fakeCall, e, msg);
+ responder.doRespond(fakeCall);
+ // Returning -1 closes out the connection.
+ return -1;
+ }
+
+ // Reads the connection header following version
+ private void processConnectionHeader(byte[] buf) throws IOException {
+ this.connectionHeader = ConnectionHeader.parseFrom(buf);
+ String serviceName = connectionHeader.getServiceName();
+ if (serviceName == null) throw new EmptyServiceNameException();
+ this.service = getService(services, serviceName);
+ if (this.service == null) throw new UnknownServiceException(serviceName);
+ setupCellBlockCodecs(this.connectionHeader);
+ UserGroupInformation protocolUser = createUser(connectionHeader);
+ if (!useSasl) {
+ user = protocolUser;
+ if (user != null) {
+ user.setAuthenticationMethod(AuthMethod.SIMPLE.authenticationMethod);
+ }
+ } else {
+ // user is authenticated
+ user.setAuthenticationMethod(authMethod.authenticationMethod);
+ //Now we check if this is a proxy user case. If the protocol user is
+ //different from the 'user', it is a proxy user scenario. However,
+ //this is not allowed if user authenticated with DIGEST.
+ if ((protocolUser != null)
+ && (!protocolUser.getUserName().equals(user.getUserName()))) {
+ if (authMethod == AuthMethod.DIGEST) {
+ // Not allowed to doAs if token authentication is used
+ throw new AccessControlException("Authenticated user (" + user
+ + ") doesn't match what the client claims to be ("
+ + protocolUser + ")");
+ } else {
+ // Effective user can be different from authenticated user
+ // for simple auth or kerberos auth
+ // The user is the real user. Now we create a proxy user
+ UserGroupInformation realUser = user;
+ user = UserGroupInformation.createProxyUser(protocolUser
+ .getUserName(), realUser);
+ // Now the user is a proxy user, set Authentication method Proxy.
+ user.setAuthenticationMethod(AuthenticationMethod.PROXY);
+ }
+ }
+ }
+ }
+
+ /**
+ * Set up cell block codecs
+ * @param header
+ * @throws FatalConnectionException
+ */
+ private void setupCellBlockCodecs(final ConnectionHeader header)
+ throws FatalConnectionException {
+ // TODO: Plug in other supported decoders.
+ if (!header.hasCellBlockCodecClass()) throw new FatalConnectionException("No codec");
+ String className = header.getCellBlockCodecClass();
+ try {
+ this.codec = (Codec)Class.forName(className).newInstance();
+ } catch (Exception e) {
+ throw new UnsupportedCellCodecException(className, e);
+ }
+ if (!header.hasCellBlockCompressorClass()) return;
+ className = header.getCellBlockCompressorClass();
+ try {
+ this.compressionCodec = (CompressionCodec)Class.forName(className).newInstance();
+ } catch (Exception e) {
+ throw new UnsupportedCompressionCodecException(className, e);
+ }
+ }
+
+ private void processUnwrappedData(byte[] inBuf) throws IOException,
+ InterruptedException {
+ ReadableByteChannel ch = Channels.newChannel(new ByteArrayInputStream(inBuf));
+ // Read all RPCs contained in the inBuf, even partial ones
+ while (true) {
+ int count = -1;
+ if (unwrappedDataLengthBuffer.remaining() > 0) {
+ count = channelRead(ch, unwrappedDataLengthBuffer);
+ if (count <= 0 || unwrappedDataLengthBuffer.remaining() > 0)
+ return;
+ }
+
+ if (unwrappedData == null) {
+ unwrappedDataLengthBuffer.flip();
+ int unwrappedDataLength = unwrappedDataLengthBuffer.getInt();
+
+ if (unwrappedDataLength == RpcClient.PING_CALL_ID) {
+ if (LOG.isDebugEnabled())
+ LOG.debug("Received ping message");
+ unwrappedDataLengthBuffer.clear();
+ continue; // ping message
+ }
+ unwrappedData = ByteBuffer.allocate(unwrappedDataLength);
+ }
+
+ count = channelRead(ch, unwrappedData);
+ if (count <= 0 || unwrappedData.remaining() > 0)
+ return;
+
+ if (unwrappedData.remaining() == 0) {
+ unwrappedDataLengthBuffer.clear();
+ unwrappedData.flip();
+ processOneRpc(unwrappedData.array());
+ unwrappedData = null;
+ }
+ }
+ }
+
+ private void processOneRpc(byte[] buf) throws IOException, InterruptedException {
+ if (connectionHeaderRead) {
+ processRequest(buf);
+ } else {
+ processConnectionHeader(buf);
+ this.connectionHeaderRead = true;
+ if (!authorizeConnection()) {
+ // Throw FatalConnectionException wrapping ACE so client does right thing and closes
+ // down the connection instead of trying to read non-existent retun.
+ throw new AccessControlException("Connection from " + this + " for service " +
+ connectionHeader.getServiceName() + " is unauthorized for user: " + user);
+ }
+ }
+ }
+
+ /**
+ * @param buf Has the request header and the request param and optionally encoded data buffer
+ * all in this one array.
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ protected void processRequest(byte[] buf) throws IOException, InterruptedException {
+ long totalRequestSize = buf.length;
+ int offset = 0;
+ // Here we read in the header. We avoid having pb
+ // do its default 4k allocation for CodedInputStream. We force it to use backing array.
+ CodedInputStream cis = CodedInputStream.newInstance(buf, offset, buf.length);
+ int headerSize = cis.readRawVarint32();
+ offset = cis.getTotalBytesRead();
+ RequestHeader header = RequestHeader.newBuilder().mergeFrom(buf, offset, headerSize).build();
+ offset += headerSize;
+ int id = header.getCallId();
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("RequestHeader " + TextFormat.shortDebugString(header) +
+ " totalRequestSize: " + totalRequestSize + " bytes");
+ }
+ // Enforcing the call queue size, this triggers a retry in the client
+ // This is a bit late to be doing this check - we have already read in the total request.
+ if ((totalRequestSize + callQueueSize.get()) > maxQueueSize) {
+ final Call callTooBig =
+ new Call(id, this.service, null, null, null, this, responder, totalRequestSize, null);
+ ByteArrayOutputStream responseBuffer = new ByteArrayOutputStream();
+ setupResponse(responseBuffer, callTooBig, new CallQueueTooBigException(),
+ "Call queue is full, is ipc.server.max.callqueue.size too small?");
+ responder.doRespond(callTooBig);
+ return;
+ }
+ MethodDescriptor md = null;
+ Message param = null;
+ CellScanner cellScanner = null;
+ try {
+ if (header.hasRequestParam() && header.getRequestParam()) {
+ md = this.service.getDescriptorForType().findMethodByName(header.getMethodName());
+ Builder builder = this.service.getRequestPrototype(md).newBuilderForType();
+ // To read the varint, I need an inputstream; might as well be a CIS.
+ cis = CodedInputStream.newInstance(buf, offset, buf.length);
+ int paramSize = cis.readRawVarint32();
+ offset += cis.getTotalBytesRead();
+ if (builder != null) {
+ param = builder.mergeFrom(buf, offset, paramSize).build();
+ }
+ offset += paramSize;
+ }
+ if (header.hasCellBlockMeta()) {
+ cellScanner = ipcUtil.createCellScanner(this.codec, this.compressionCodec,
+ buf, offset, buf.length);
+ }
+ } catch (Throwable t) {
+ String msg = "Unable to read call parameter from client " + getHostAddress();
+ LOG.warn(msg, t);
+ final Call readParamsFailedCall =
+ new Call(id, this.service, null, null, null, this, responder, totalRequestSize, null);
+ ByteArrayOutputStream responseBuffer = new ByteArrayOutputStream();
+ setupResponse(responseBuffer, readParamsFailedCall, t,
+ msg + "; " + t.getMessage());
+ responder.doRespond(readParamsFailedCall);
+ return;
+ }
+
+ Call call = null;
+ if (header.hasTraceInfo()) {
+ call = new Call(id, this.service, md, param, cellScanner, this, responder, totalRequestSize,
+ new TraceInfo(header.getTraceInfo().getTraceId(), header.getTraceInfo().getParentId()));
+ } else {
+ call = new Call(id, this.service, md, param, cellScanner, this, responder,
+ totalRequestSize, null);
+ }
+ callQueueSize.add(totalRequestSize);
+ Pair<RequestHeader, Message> headerAndParam = new Pair<RequestHeader, Message>(header, param);
+ if (priorityCallQueue != null && getQosLevel(headerAndParam) > highPriorityLevel) {
+ priorityCallQueue.put(call);
+ } else if (replicationQueue != null &&
+ getQosLevel(headerAndParam) == HConstants.REPLICATION_QOS) {
+ replicationQueue.put(call);
+ } else {
+ callQueue.put(call); // queue the call; maybe blocked here
+ }
+ }
+
+ private boolean authorizeConnection() throws IOException {
+ try {
+ // If auth method is DIGEST, the token was obtained by the
+ // real user for the effective user, therefore not required to
+ // authorize real user. doAs is allowed only for simple or kerberos
+ // authentication
+ if (user != null && user.getRealUser() != null
+ && (authMethod != AuthMethod.DIGEST)) {
+ ProxyUsers.authorize(user, this.getHostAddress(), conf);
+ }
+ authorize(user, connectionHeader, getHostInetAddress());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Authorized " + TextFormat.shortDebugString(connectionHeader));
+ }
+ metrics.authorizationSuccess();
+ } catch (AuthorizationException ae) {
+ LOG.debug("Connection authorization failed: " + ae.getMessage(), ae);
+ metrics.authorizationFailure();
+ setupResponse(authFailedResponse, authFailedCall, ae, ae.getMessage());
+ responder.doRespond(authFailedCall);
+ return false;
+ }
+ return true;
+ }
+
+ protected synchronized void close() {
+ disposeSasl();
+ data = null;
+ this.dataLengthBuffer = null;
+ if (!channel.isOpen())
+ return;
+ try {socket.shutdownOutput();} catch(Exception ignored) {} // FindBugs DE_MIGHT_IGNORE
+ if (channel.isOpen()) {
+ try {channel.close();} catch(Exception ignored) {}
+ }
+ try {socket.close();} catch(Exception ignored) {}
+ }
+
+ private UserGroupInformation createUser(ConnectionHeader head) {
+ UserGroupInformation ugi = null;
+
+ if (!head.hasUserInfo()) {
+ return null;
+ }
+ UserInformation userInfoProto = head.getUserInfo();
+ String effectiveUser = null;
+ if (userInfoProto.hasEffectiveUser()) {
+ effectiveUser = userInfoProto.getEffectiveUser();
+ }
+ String realUser = null;
+ if (userInfoProto.hasRealUser()) {
+ realUser = userInfoProto.getRealUser();
+ }
+ if (effectiveUser != null) {
+ if (realUser != null) {
+ UserGroupInformation realUserUgi =
+ UserGroupInformation.createRemoteUser(realUser);
+ ugi = UserGroupInformation.createProxyUser(effectiveUser, realUserUgi);
+ } else {
+ ugi = UserGroupInformation.createRemoteUser(effectiveUser);
+ }
+ }
+ return ugi;
+ }
+ }
+
+ /** Handles queued calls . */
+ private class Handler extends Thread {
+ private final BlockingQueue<Call> myCallQueue;
+ private MonitoredRPCHandler status;
+
+ public Handler(final BlockingQueue<Call> cq, int instanceNumber) {
+ this.myCallQueue = cq;
+ this.setDaemon(true);
+
+ String threadName = "IPC Server handler " + instanceNumber + " on " + port;
+ if (cq == priorityCallQueue) {
+ // this is just an amazing hack, but it works.
+ threadName = "PRI " + threadName;
+ } else if (cq == replicationQueue) {
+ threadName = "REPL " + threadName;
+ }
+ this.setName(threadName);
+ this.status = TaskMonitor.get().createRPCStatus(threadName);
+ }
+
+ @Override
+ public void run() {
+ LOG.info(getName() + ": starting");
+ status.setStatus("starting");
+ SERVER.set(RpcServer.this);
+ while (running) {
+ try {
+ status.pause("Waiting for a call");
+ Call call = myCallQueue.take(); // pop the queue; maybe blocked here
+ status.setStatus("Setting up call");
+ status.setConnection(call.connection.getHostAddress(), call.connection.getRemotePort());
+ if (LOG.isDebugEnabled()) {
+ UserGroupInformation remoteUser = call.connection.user;
+ LOG.debug(call.toString() + " executing as " +
+ ((remoteUser == null)? "NULL principal": remoteUser.getUserName()));
+ }
+ Throwable errorThrowable = null;
+ String error = null;
+ Pair<Message, CellScanner> resultPair = null;
+ CurCall.set(call);
+ Span currentRequestSpan = NullSpan.getInstance();
+ try {
+ if (!started) {
+ throw new ServerNotRunningYetException("Server is not running yet");
+ }
+ if (call.tinfo != null) {
+ currentRequestSpan = Trace.startSpan(
+ "handling " + call.toString(), call.tinfo, Sampler.ALWAYS);
+ }
+ RequestContext.set(User.create(call.connection.user), getRemoteIp(),
+ call.connection.service);
+
+ // make the call
+ resultPair = call(call.service, call.md, call.param, call.cellScanner, call.timestamp,
+ status);
+ } catch (Throwable e) {
+ LOG.debug(getName() + ": " + call.toString(), e);
+ errorThrowable = e;
+ error = StringUtils.stringifyException(e);
+ } finally {
+ currentRequestSpan.stop();
+ // Must always clear the request context to avoid leaking
+ // credentials between requests.
+ RequestContext.clear();
+ }
+ CurCall.set(null);
+ callQueueSize.add(call.getSize() * -1);
+ // Set the response for undelayed calls and delayed calls with
+ // undelayed responses.
+ if (!call.isDelayed() || !call.isReturnValueDelayed()) {
+ Message param = resultPair != null? resultPair.getFirst(): null;
+ CellScanner cells = resultPair != null? resultPair.getSecond(): null;
+ call.setResponse(param, cells, errorThrowable, error);
+ }
+ call.sendResponseIfReady();
+ status.markComplete("Sent response");
+ } catch (InterruptedException e) {
+ if (running) { // unexpected -- log it
+ LOG.info(getName() + ": caught: " + StringUtils.stringifyException(e));
+ }
+ } catch (OutOfMemoryError e) {
+ if (errorHandler != null) {
+ if (errorHandler.checkOOME(e)) {
+ LOG.info(getName() + ": exiting on OutOfMemoryError");
+ return;
+ }
+ } else {
+ // rethrow if no handler
+ throw e;
+ }
+ } catch (ClosedChannelException cce) {
+ LOG.warn(getName() + ": caught a ClosedChannelException, " +
+ "this means that the server was processing a " +
+ "request but the client went away. The error message was: " +
+ cce.getMessage());
+ } catch (Exception e) {
+ LOG.warn(getName() + ": caught: " + StringUtils.stringifyException(e));
+ }
+ }
+ LOG.info(getName() + ": exiting");
+ }
+ }
+
+ /**
+ * Datastructure for passing a {@link BlockingService} and its associated class of
+ * protobuf service interface. For example, a server that fielded what is defined
+ * in the client protobuf service would pass in an implementation of the client blocking service
+ * and then its ClientService.BlockingInterface.class. Used checking connection setup.
*/
- Pair<Message, CellScanner> call(Class<? extends IpcProtocol> protocol, Method method,
- Message param, CellScanner cellScanner, long receiveTime, MonitoredRPCHandler status)
- throws IOException;
+ public static class BlockingServiceAndInterface {
+ private final BlockingService service;
+ private final Class<?> serviceInterface;
+ public BlockingServiceAndInterface(final BlockingService service,
+ final Class<?> serviceInterface) {
+ this.service = service;
+ this.serviceInterface = serviceInterface;
+ }
+ public Class<?> getServiceInterface() {
+ return this.serviceInterface;
+ }
+ public BlockingService getBlockingService() {
+ return this.service;
+ }
+ }
+
+
+ /**
+ * Minimal setup. Used by tests mostly.
+ * @param service
+ * @param isa
+ * @param conf
+ * @throws IOException
+ */
+ public RpcServer(final BlockingService service, final InetSocketAddress isa,
+ final Configuration conf)
+ throws IOException {
+ this(null, "generic", Lists.newArrayList(new BlockingServiceAndInterface(service, null)),
+ isa, 3, 3, conf,
+ HConstants.QOS_THRESHOLD);
+ }
+
+ /**
+ * Constructs a server listening on the named port and address.
+ * @param serverInstance hosting instance of {@link Server}. We will do authentications if an
+ * instance else pass null for no authentication check.
+ * @param name Used keying this rpc servers' metrics and for naming the Listener thread.
+ * @param services A list of services.
+ * @param isa Where to listen
+ * @param handlerCount the number of handler threads that will be used to process calls
+ * @param priorityHandlerCount How many threads for priority handling.
+ * @param conf
+ * @param highPriorityLevel
+ * @throws IOException
+ */
+ public RpcServer(final Server serverInstance, final String name,
+ final List<BlockingServiceAndInterface> services,
+ final InetSocketAddress isa, int handlerCount, int priorityHandlerCount, Configuration conf,
+ int highPriorityLevel)
+ throws IOException {
+ this.serverInstance = serverInstance;
+ this.services = services;
+ this.isa = isa;
+ this.conf = conf;
+ this.handlerCount = handlerCount;
+ this.priorityHandlerCount = priorityHandlerCount;
+ this.socketSendBufferSize = 0;
+ this.maxQueueLength = this.conf.getInt("ipc.server.max.callqueue.length",
+ handlerCount * DEFAULT_MAX_CALLQUEUE_LENGTH_PER_HANDLER);
+ this.maxQueueSize =
+ this.conf.getInt("ipc.server.max.callqueue.size", DEFAULT_MAX_CALLQUEUE_SIZE);
+ this.readThreads = conf.getInt("ipc.server.read.threadpool.size", 10);
+ this.callQueue = new LinkedBlockingQueue<Call>(maxQueueLength);
+ if (priorityHandlerCount > 0) {
+ this.priorityCallQueue = new LinkedBlockingQueue<Call>(maxQueueLength); // TODO hack on size
+ } else {
+ this.priorityCallQueue = null;
+ }
+ this.highPriorityLevel = highPriorityLevel;
+ this.maxIdleTime = 2*conf.getInt("ipc.client.connection.maxidletime", 1000);
+ this.maxConnectionsToNuke = conf.getInt("ipc.client.kill.max", 10);
+ this.thresholdIdleConnections = conf.getInt("ipc.client.idlethreshold", 4000);
+ this.purgeTimeout = conf.getLong("ipc.client.call.purge.timeout",
+ 2 * HConstants.DEFAULT_HBASE_RPC_TIMEOUT);
+ this.numOfReplicationHandlers = conf.getInt("hbase.regionserver.replication.handler.count", 3);
+ if (numOfReplicationHandlers > 0) {
+ this.replicationQueue = new LinkedBlockingQueue<Call>(maxQueueSize);
+ }
+
+ this.warnResponseTime = conf.getInt(WARN_RESPONSE_TIME, DEFAULT_WARN_RESPONSE_TIME);
+ this.warnResponseSize = conf.getInt(WARN_RESPONSE_SIZE, DEFAULT_WARN_RESPONSE_SIZE);
- void setErrorHandler(HBaseRPCErrorHandler handler);
+ // Start the listener here and let it bind to the port
+ listener = new Listener(name);
+ this.port = listener.getAddress().getPort();
- void openServer();
+ this.metrics = new MetricsHBaseServer(name, new MetricsHBaseServerWrapperImpl(this));
+ this.tcpNoDelay = conf.getBoolean("ipc.server.tcpnodelay", true);
+ this.tcpKeepAlive = conf.getBoolean("ipc.server.tcpkeepalive", true);
- void startThreads();
+ this.warnDelayedCalls = conf.getInt(WARN_DELAYED_CALLS, DEFAULT_WARN_DELAYED_CALLS);
+ this.delayedCalls = new AtomicInteger(0);
+ this.ipcUtil = new IPCUtil(conf);
+
+
+ // Create the responder here
+ responder = new Responder();
+ this.authorize = conf.getBoolean(HADOOP_SECURITY_AUTHORIZATION, false);
+ this.isSecurityEnabled = User.isHBaseSecurityEnabled(this.conf);
+ if (isSecurityEnabled) {
+ HBaseSaslRpcServer.init(conf);
+ }
+ }
+
+ /**
+ * Subclasses of HBaseServer can override this to provide their own
+ * Connection implementations.
+ */
+ protected Connection getConnection(SocketChannel channel, long time) {
+ return new Connection(channel, time);
+ }
+
+ /**
+ * Setup response for the IPC Call.
+ *
+ * @param response buffer to serialize the response into
+ * @param call {@link Call} to which we are setting up the response
+ * @param status {@link Status} of the IPC call
+ * @param errorClass error class, if the the call failed
+ * @param error error message, if the call failed
+ * @throws IOException
+ */
+ private void setupResponse(ByteArrayOutputStream response, Call call, Throwable t, String error)
+ throws IOException {
+ if (response != null) response.reset();
+ call.setResponse(null, null, t, error);
+ }
+
+ protected void closeConnection(Connection connection) {
+ synchronized (connectionList) {
+ if (connectionList.remove(connection)) {
+ numConnections--;
+ }
+ }
+ connection.close();
+ }
+
+ Configuration getConf() {
+ return conf;
+ }
+
+ /** Sets the socket buffer size used for responding to RPCs.
+ * @param size send size
+ */
+ @Override
+ public void setSocketSendBufSize(int size) { this.socketSendBufferSize = size; }
+
+ /** Starts the service. Must be called before any calls will be handled. */
+ @Override
+ public void start() {
+ startThreads();
+ openServer();
+ }
+
+ /**
+ * Open a previously started server.
+ */
+ @Override
+ public void openServer() {
+ started = true;
+ }
+
+ /**
+ * Starts the service threads but does not allow requests to be responded yet.
+ * Client will get {@link ServerNotRunningYetException} instead.
+ */
+ @Override
+ public synchronized void startThreads() {
+ AuthenticationTokenSecretManager mgr = createSecretManager();
+ if (mgr != null) {
+ setSecretManager(mgr);
+ mgr.start();
+ }
+ this.authManager = new ServiceAuthorizationManager();
+ HBasePolicyProvider.init(conf, authManager);
+ responder.start();
+ listener.start();
+ handlers = startHandlers(callQueue, handlerCount);
+ priorityHandlers = startHandlers(priorityCallQueue, priorityHandlerCount);
+ replicationHandlers = startHandlers(replicationQueue, numOfReplicationHandlers);
+ }
+
+ @Override
+ public void refreshAuthManager(PolicyProvider pp) {
+ this.authManager.refresh(this.conf, pp);
+ }
+
+ private Handler[] startHandlers(BlockingQueue<Call> queue, int numOfHandlers) {
+ if (numOfHandlers <= 0) {
+ return null;
+ }
+ Handler[] handlers = new Handler[numOfHandlers];
+ for (int i = 0; i < numOfHandlers; i++) {
+ handlers[i] = new Handler(queue, i);
+ handlers[i].start();
+ }
+ return handlers;
+ }
+
+ private AuthenticationTokenSecretManager createSecretManager() {
+ if (!isSecurityEnabled) return null;
+ if (serverInstance == null) return null;
+ if (!(serverInstance instanceof org.apache.hadoop.hbase.Server)) return null;
+ org.apache.hadoop.hbase.Server server = (org.apache.hadoop.hbase.Server)serverInstance;
+ Configuration conf = server.getConfiguration();
+ long keyUpdateInterval =
+ conf.getLong("hbase.auth.key.update.interval", 24*60*60*1000);
+ long maxAge =
+ conf.getLong("hbase.auth.token.max.lifetime", 7*24*60*60*1000);
+ return new AuthenticationTokenSecretManager(conf, server.getZooKeeper(),
+ server.getServerName().toString(), keyUpdateInterval, maxAge);
+ }
+
+ public SecretManager<? extends TokenIdentifier> getSecretManager() {
+ return this.secretManager;
+ }
+
+ @SuppressWarnings("unchecked")
+ public void setSecretManager(SecretManager<? extends TokenIdentifier> secretManager) {
+ this.secretManager = (SecretManager<TokenIdentifier>) secretManager;
+ }
+
+ /**
+ * This is a server side method, which is invoked over RPC. On success
+ * the return response has protobuf response payload. On failure, the
+ * exception name and the stack trace are returned in the protobuf response.
+ */
+ public Pair<Message, CellScanner> call(BlockingService service, MethodDescriptor md,
+ Message param, CellScanner cellScanner, long receiveTime, MonitoredRPCHandler status)
+ throws IOException {
+ try {
+ status.setRPC(md.getName(), new Object[]{param}, receiveTime);
+ // TODO: Review after we add in encoded data blocks.
+ status.setRPCPacket(param);
+ status.resume("Servicing call");
+ //get an instance of the method arg type
+ long startTime = System.currentTimeMillis();
+ PayloadCarryingRpcController controller = new PayloadCarryingRpcController(cellScanner);
+ Message result = service.callBlockingMethod(md, controller, param);
+ int processingTime = (int) (System.currentTimeMillis() - startTime);
+ int qTime = (int) (startTime - receiveTime);
+ if (LOG.isTraceEnabled()) {
+ LOG.trace(CurCall.get().toString() +
+ ", response " + TextFormat.shortDebugString(result) +
+ " queueTime: " + qTime +
+ " processingTime: " + processingTime);
+ }
+ metrics.dequeuedCall(qTime);
+ metrics.processedCall(processingTime);
+ long responseSize = result.getSerializedSize();
+ // log any RPC responses that are slower than the configured warn
+ // response time or larger than configured warning size
+ boolean tooSlow = (processingTime > warnResponseTime && warnResponseTime > -1);
+ boolean tooLarge = (responseSize > warnResponseSize && warnResponseSize > -1);
+ if (tooSlow || tooLarge) {
+ // when tagging, we let TooLarge trump TooSmall to keep output simple
+ // note that large responses will often also be slow.
+ StringBuilder buffer = new StringBuilder(256);
+ buffer.append(md.getName());
+ buffer.append("(");
+ buffer.append(param.getClass().getName());
+ buffer.append(")");
+ logResponse(new Object[]{param},
+ md.getName(), buffer.toString(), (tooLarge ? "TooLarge" : "TooSlow"),
+ status.getClient(), startTime, processingTime, qTime,
+ responseSize);
+ }
+ return new Pair<Message, CellScanner>(result,
+ controller != null? controller.cellScanner(): null);
+ } catch (Throwable e) {
+ // The above callBlockingMethod will always return a SE. Strip the SE wrapper before
+ // putting it on the wire. Its needed to adhere to the pb Service Interface but we don't
+ // need to pass it over the wire.
+ if (e instanceof ServiceException) e = e.getCause();
+ if (e instanceof IOException) throw (IOException)e;
+ LOG.error("Unexpected throwable object ", e);
+ throw new IOException(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Logs an RPC response to the LOG file, producing valid JSON objects for
+ * client Operations.
+ * @param params The parameters received in the call.
+ * @param methodName The name of the method invoked
+ * @param call The string representation of the call
+ * @param tag The tag that will be used to indicate this event in the log.
+ * @param clientAddress The address of the client who made this call.
+ * @param startTime The time that the call was initiated, in ms.
+ * @param processingTime The duration that the call took to run, in ms.
+ * @param qTime The duration that the call spent on the queue
+ * prior to being initiated, in ms.
+ * @param responseSize The size in bytes of the response buffer.
+ */
+ void logResponse(Object[] params, String methodName, String call, String tag,
+ String clientAddress, long startTime, int processingTime, int qTime,
+ long responseSize)
+ throws IOException {
+ // for JSON encoding
+ ObjectMapper mapper = new ObjectMapper();
+ // base information that is reported regardless of type of call
+ Map<String, Object> responseInfo = new HashMap<String, Object>();
+ responseInfo.put("starttimems", startTime);
+ responseInfo.put("processingtimems", processingTime);
+ responseInfo.put("queuetimems", qTime);
+ responseInfo.put("responsesize", responseSize);
+ responseInfo.put("client", clientAddress);
+ responseInfo.put("class", serverInstance == null? "": serverInstance.getClass().getSimpleName());
+ responseInfo.put("method", methodName);
+ if (params.length == 2 && serverInstance instanceof HRegionServer &&
+ params[0] instanceof byte[] &&
+ params[1] instanceof Operation) {
+ // if the slow process is a query, we want to log its table as well
+ // as its own fingerprint
+ byte [] tableName =
+ HRegionInfo.parseRegionName((byte[]) params[0])[0];
+ responseInfo.put("table", Bytes.toStringBinary(tableName));
+ // annotate the response map with operation details
+ responseInfo.putAll(((Operation) params[1]).toMap());
+ // report to the log file
+ LOG.warn("(operation" + tag + "): " +
+ mapper.writeValueAsString(responseInfo));
+ } else if (params.length == 1 && serverInstance instanceof HRegionServer &&
+ params[0] instanceof Operation) {
+ // annotate the response map with operation details
+ responseInfo.putAll(((Operation) params[0]).toMap());
+ // report to the log file
+ LOG.warn("(operation" + tag + "): " +
+ mapper.writeValueAsString(responseInfo));
+ } else {
+ // can't get JSON details, so just report call.toString() along with
+ // a more generic tag.
+ responseInfo.put("call", call);
+ LOG.warn("(response" + tag + "): " + mapper.writeValueAsString(responseInfo));
+ }
+ }
+
+ /** Stops the service. No new calls will be handled after this is called. */
+ @Override
+ public synchronized void stop() {
+ LOG.info("Stopping server on " + port);
+ running = false;
+ stopHandlers(handlers);
+ stopHandlers(priorityHandlers);
+ stopHandlers(replicationHandlers);
+ listener.interrupt();
+ listener.doStop();
+ responder.interrupt();
+ notifyAll();
+ }
+
+ private void stopHandlers(Handler[] handlers) {
+ if (handlers != null) {
+ for (Handler handler : handlers) {
+ if (handler != null) {
+ handler.interrupt();
+ }
+ }
+ }
+ }
+
+ /** Wait for the server to be stopped.
+ * Does not wait for all subthreads to finish.
+ * See {@link #stop()}.
[... 268 lines stripped ...]