1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.thrift2;
20
21 import java.io.IOException;
22 import java.net.InetAddress;
23 import java.net.InetSocketAddress;
24 import java.net.UnknownHostException;
25 import java.security.PrivilegedAction;
26 import java.util.List;
27 import java.util.Map;
28 import java.util.concurrent.ExecutorService;
29 import java.util.concurrent.LinkedBlockingQueue;
30 import java.util.concurrent.SynchronousQueue;
31 import java.util.concurrent.ThreadPoolExecutor;
32 import java.util.concurrent.TimeUnit;
33
34 import javax.security.auth.callback.Callback;
35 import javax.security.auth.callback.UnsupportedCallbackException;
36 import javax.security.sasl.AuthorizeCallback;
37 import javax.security.sasl.SaslServer;
38
39 import org.apache.commons.cli.CommandLine;
40 import org.apache.commons.cli.CommandLineParser;
41 import org.apache.commons.cli.HelpFormatter;
42 import org.apache.commons.cli.Option;
43 import org.apache.commons.cli.OptionGroup;
44 import org.apache.commons.cli.Options;
45 import org.apache.commons.cli.ParseException;
46 import org.apache.commons.cli.PosixParser;
47 import org.apache.commons.logging.Log;
48 import org.apache.commons.logging.LogFactory;
49 import org.apache.hadoop.conf.Configuration;
50 import org.apache.hadoop.hbase.HBaseConfiguration;
51 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
52 import org.apache.hadoop.hbase.classification.InterfaceAudience;
53 import org.apache.hadoop.hbase.filter.ParseFilter;
54 import org.apache.hadoop.hbase.http.InfoServer;
55 import org.apache.hadoop.hbase.security.SaslUtil;
56 import org.apache.hadoop.hbase.security.SecurityUtil;
57 import org.apache.hadoop.hbase.security.UserProvider;
58 import org.apache.hadoop.hbase.thrift.CallQueue;
59 import org.apache.hadoop.hbase.thrift.CallQueue.Call;
60 import org.apache.hadoop.hbase.thrift.THBaseThreadPoolExecutor;
61 import org.apache.hadoop.hbase.thrift.ThriftMetrics;
62 import org.apache.hadoop.hbase.thrift2.generated.THBaseService;
63 import org.apache.hadoop.hbase.util.DNS;
64 import org.apache.hadoop.hbase.util.JvmPauseMonitor;
65 import org.apache.hadoop.hbase.util.Strings;
66 import org.apache.hadoop.security.SaslRpcServer.SaslGssCallbackHandler;
67 import org.apache.hadoop.security.UserGroupInformation;
68 import org.apache.hadoop.util.GenericOptionsParser;
69 import org.apache.thrift.TException;
70 import org.apache.thrift.TProcessor;
71 import org.apache.thrift.protocol.TBinaryProtocol;
72 import org.apache.thrift.protocol.TCompactProtocol;
73 import org.apache.thrift.protocol.TProtocol;
74 import org.apache.thrift.protocol.TProtocolFactory;
75 import org.apache.thrift.server.THsHaServer;
76 import org.apache.thrift.server.TNonblockingServer;
77 import org.apache.thrift.server.TServer;
78 import org.apache.thrift.server.TThreadPoolServer;
79 import org.apache.thrift.server.TThreadedSelectorServer;
80 import org.apache.thrift.transport.TFramedTransport;
81 import org.apache.thrift.transport.TNonblockingServerSocket;
82 import org.apache.thrift.transport.TNonblockingServerTransport;
83 import org.apache.thrift.transport.TSaslServerTransport;
84 import org.apache.thrift.transport.TServerSocket;
85 import org.apache.thrift.transport.TServerTransport;
86 import org.apache.thrift.transport.TTransportException;
87 import org.apache.thrift.transport.TTransportFactory;
88
89 import com.google.common.util.concurrent.ThreadFactoryBuilder;
90
91
92
93
94
95 @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.TOOLS)
96 @SuppressWarnings({ "rawtypes", "unchecked" })
97 public class ThriftServer {
98 private static final Log log = LogFactory.getLog(ThriftServer.class);
99
100
101
102
103
104
105
106
107
108
109 static final String THRIFT_QOP_KEY = "hbase.thrift.security.qop";
110
111 static final String BACKLOG_CONF_KEY = "hbase.regionserver.thrift.backlog";
112
113 public static final int DEFAULT_LISTEN_PORT = 9090;
114
115 private static final String READ_TIMEOUT_OPTION = "readTimeout";
116
117
118
119
120
121
122 public static final String THRIFT_SERVER_SOCKET_READ_TIMEOUT_KEY =
123 "hbase.thrift.server.socket.read.timeout";
124 public static final int THRIFT_SERVER_SOCKET_READ_TIMEOUT_DEFAULT = 60000;
125
126 public ThriftServer() {
127 }
128
129 private static void printUsage() {
130 HelpFormatter formatter = new HelpFormatter();
131 formatter.printHelp("Thrift", null, getOptions(),
132 "To start the Thrift server run 'bin/hbase-daemon.sh start thrift2'\n" +
133 "To shutdown the thrift server run 'bin/hbase-daemon.sh stop thrift2' or" +
134 " send a kill signal to the thrift server pid",
135 true);
136 }
137
138 private static Options getOptions() {
139 Options options = new Options();
140 options.addOption("b", "bind", true,
141 "Address to bind the Thrift server to. [default: 0.0.0.0]");
142 options.addOption("p", "port", true, "Port to bind to [default: " + DEFAULT_LISTEN_PORT + "]");
143 options.addOption("f", "framed", false, "Use framed transport");
144 options.addOption("c", "compact", false, "Use the compact protocol");
145 options.addOption("w", "workers", true, "How many worker threads to use.");
146 options.addOption("s", "selectors", true, "How many selector threads to use.");
147 options.addOption("q", "callQueueSize", true,
148 "Max size of request queue (unbounded by default)");
149 options.addOption("h", "help", false, "Print help information");
150 options.addOption(null, "infoport", true, "Port for web UI");
151 options.addOption("t", READ_TIMEOUT_OPTION, true,
152 "Amount of time in milliseconds before a server thread will timeout " +
153 "waiting for client to send data on a connected socket. Currently, " +
154 "only applies to TBoundedThreadPoolServer");
155 options.addOption("ro", "readonly", false,
156 "Respond only to read method requests [default: false]");
157 OptionGroup servers = new OptionGroup();
158 servers.addOption(
159 new Option("nonblocking", false, "Use the TNonblockingServer. This implies the framed transport."));
160 servers.addOption(new Option("hsha", false, "Use the THsHaServer. This implies the framed transport."));
161 servers.addOption(new Option("selector", false, "Use the TThreadedSelectorServer. This implies the framed transport."));
162 servers.addOption(new Option("threadpool", false, "Use the TThreadPoolServer. This is the default."));
163 options.addOptionGroup(servers);
164 return options;
165 }
166
167 private static CommandLine parseArguments(Configuration conf, Options options, String[] args)
168 throws ParseException, IOException {
169 GenericOptionsParser genParser = new GenericOptionsParser(conf, args);
170 String[] remainingArgs = genParser.getRemainingArgs();
171 CommandLineParser parser = new PosixParser();
172 return parser.parse(options, remainingArgs);
173 }
174
175 private static TProtocolFactory getTProtocolFactory(boolean isCompact) {
176 if (isCompact) {
177 log.debug("Using compact protocol");
178 return new TCompactProtocol.Factory();
179 } else {
180 log.debug("Using binary protocol");
181 return new TBinaryProtocol.Factory();
182 }
183 }
184
185 private static TTransportFactory getTTransportFactory(
186 SaslUtil.QualityOfProtection qop, String name, String host,
187 boolean framed, int frameSize) {
188 if (framed) {
189 if (qop != null) {
190 throw new RuntimeException("Thrift server authentication"
191 + " doesn't work with framed transport yet");
192 }
193 log.debug("Using framed transport");
194 return new TFramedTransport.Factory(frameSize);
195 } else if (qop == null) {
196 return new TTransportFactory();
197 } else {
198 Map<String, String> saslProperties = SaslUtil.initSaslProperties(qop.name());
199 TSaslServerTransport.Factory saslFactory = new TSaslServerTransport.Factory();
200 saslFactory.addServerDefinition("GSSAPI", name, host, saslProperties,
201 new SaslGssCallbackHandler() {
202 @Override
203 public void handle(Callback[] callbacks)
204 throws UnsupportedCallbackException {
205 AuthorizeCallback ac = null;
206 for (Callback callback : callbacks) {
207 if (callback instanceof AuthorizeCallback) {
208 ac = (AuthorizeCallback) callback;
209 } else {
210 throw new UnsupportedCallbackException(callback,
211 "Unrecognized SASL GSSAPI Callback");
212 }
213 }
214 if (ac != null) {
215 String authid = ac.getAuthenticationID();
216 String authzid = ac.getAuthorizationID();
217 if (!authid.equals(authzid)) {
218 ac.setAuthorized(false);
219 } else {
220 ac.setAuthorized(true);
221 String userName = SecurityUtil.getUserFromPrincipal(authzid);
222 log.info("Effective user: " + userName);
223 ac.setAuthorizedID(userName);
224 }
225 }
226 }
227 });
228 return saslFactory;
229 }
230 }
231
232
233
234
235 private static InetSocketAddress bindToPort(String bindValue, int listenPort)
236 throws UnknownHostException {
237 try {
238 if (bindValue == null) {
239 return new InetSocketAddress(listenPort);
240 } else {
241 return new InetSocketAddress(InetAddress.getByName(bindValue), listenPort);
242 }
243 } catch (UnknownHostException e) {
244 throw new RuntimeException("Could not bind to provided ip address", e);
245 }
246 }
247
248 private static TServer getTNonBlockingServer(TProtocolFactory protocolFactory, TProcessor processor,
249 TTransportFactory transportFactory, InetSocketAddress inetSocketAddress) throws TTransportException {
250 TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(inetSocketAddress);
251 log.info("starting HBase Nonblocking Thrift server on " + inetSocketAddress.toString());
252 TNonblockingServer.Args serverArgs = new TNonblockingServer.Args(serverTransport);
253 serverArgs.processor(processor);
254 serverArgs.transportFactory(transportFactory);
255 serverArgs.protocolFactory(protocolFactory);
256 return new TNonblockingServer(serverArgs);
257 }
258
259 private static TServer getTHsHaServer(TProtocolFactory protocolFactory,
260 TProcessor processor, TTransportFactory transportFactory,
261 int workerThreads, int maxCallQueueSize,
262 InetSocketAddress inetSocketAddress, ThriftMetrics metrics)
263 throws TTransportException {
264 TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(inetSocketAddress);
265 log.info("starting HBase HsHA Thrift server on " + inetSocketAddress.toString());
266 THsHaServer.Args serverArgs = new THsHaServer.Args(serverTransport);
267 if (workerThreads > 0) {
268
269 serverArgs.minWorkerThreads(workerThreads).maxWorkerThreads(workerThreads);
270 }
271 ExecutorService executorService = createExecutor(
272 workerThreads, maxCallQueueSize, metrics);
273 serverArgs.executorService(executorService);
274 serverArgs.processor(processor);
275 serverArgs.transportFactory(transportFactory);
276 serverArgs.protocolFactory(protocolFactory);
277 return new THsHaServer(serverArgs);
278 }
279
280 private static TServer getTThreadedSelectorServer(TProtocolFactory protocolFactory,
281 TProcessor processor, TTransportFactory transportFactory,
282 int workerThreads, int selectorThreads, int maxCallQueueSize,
283 InetSocketAddress inetSocketAddress, ThriftMetrics metrics)
284 throws TTransportException {
285 TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(inetSocketAddress);
286 log.info("starting HBase ThreadedSelector Thrift server on " + inetSocketAddress.toString());
287 TThreadedSelectorServer.Args serverArgs = new TThreadedSelectorServer.Args(serverTransport);
288 if (workerThreads > 0) {
289 serverArgs.workerThreads(workerThreads);
290 }
291 if (selectorThreads > 0) {
292 serverArgs.selectorThreads(selectorThreads);
293 }
294
295 ExecutorService executorService = createExecutor(
296 workerThreads, maxCallQueueSize, metrics);
297 serverArgs.executorService(executorService);
298 serverArgs.processor(processor);
299 serverArgs.transportFactory(transportFactory);
300 serverArgs.protocolFactory(protocolFactory);
301 return new TThreadedSelectorServer(serverArgs);
302 }
303
304 private static ExecutorService createExecutor(
305 int workerThreads, int maxCallQueueSize, ThriftMetrics metrics) {
306 CallQueue callQueue;
307 if (maxCallQueueSize > 0) {
308 callQueue = new CallQueue(new LinkedBlockingQueue<Call>(maxCallQueueSize), metrics);
309 } else {
310 callQueue = new CallQueue(new LinkedBlockingQueue<Call>(), metrics);
311 }
312
313 ThreadFactoryBuilder tfb = new ThreadFactoryBuilder();
314 tfb.setDaemon(true);
315 tfb.setNameFormat("thrift2-worker-%d");
316 ThreadPoolExecutor pool = new THBaseThreadPoolExecutor(workerThreads, workerThreads,
317 Long.MAX_VALUE, TimeUnit.SECONDS, callQueue, tfb.build(), metrics);
318 pool.prestartAllCoreThreads();
319 return pool;
320 }
321
322 private static TServer getTThreadPoolServer(TProtocolFactory protocolFactory,
323 TProcessor processor,
324 TTransportFactory transportFactory,
325 int workerThreads,
326 InetSocketAddress inetSocketAddress,
327 int backlog,
328 int clientTimeout,
329 ThriftMetrics metrics)
330 throws TTransportException {
331 TServerTransport serverTransport = new TServerSocket(
332 new TServerSocket.ServerSocketTransportArgs().
333 bindAddr(inetSocketAddress).backlog(backlog).
334 clientTimeout(clientTimeout));
335 log.info("starting HBase ThreadPool Thrift server on " + inetSocketAddress.toString());
336 TThreadPoolServer.Args serverArgs = new TThreadPoolServer.Args(serverTransport);
337 serverArgs.processor(processor);
338 serverArgs.transportFactory(transportFactory);
339 serverArgs.protocolFactory(protocolFactory);
340 if (workerThreads > 0) {
341 serverArgs.maxWorkerThreads(workerThreads);
342 }
343 ThreadPoolExecutor executor = new THBaseThreadPoolExecutor(serverArgs.minWorkerThreads,
344 serverArgs.maxWorkerThreads, serverArgs.stopTimeoutVal, TimeUnit.SECONDS,
345 new SynchronousQueue<Runnable>(), metrics);
346 serverArgs.executorService(executor);
347
348 return new TThreadPoolServer(serverArgs);
349 }
350
351
352
353
354
355
356 protected static void registerFilters(Configuration conf) {
357 String[] filters = conf.getStrings("hbase.thrift.filters");
358 if(filters != null) {
359 for(String filterClass: filters) {
360 String[] filterPart = filterClass.split(":");
361 if(filterPart.length != 2) {
362 log.warn("Invalid filter specification " + filterClass + " - skipping");
363 } else {
364 ParseFilter.registerFilter(filterPart[0], filterPart[1]);
365 }
366 }
367 }
368 }
369
370
371
372
373
374
375 public static void main(String[] args) throws Exception {
376 TServer server = null;
377 Options options = getOptions();
378 Configuration conf = HBaseConfiguration.create();
379 CommandLine cmd = parseArguments(conf, options, args);
380 int workerThreads = 0;
381 int selectorThreads = 0;
382 int maxCallQueueSize = -1;
383
384
385
386
387
388 List<?> argList = cmd.getArgList();
389 if (cmd.hasOption("help") || !argList.contains("start") || argList.contains("stop")) {
390 printUsage();
391 System.exit(1);
392 }
393
394
395 String bindAddress;
396 if (cmd.hasOption("bind")) {
397 bindAddress = cmd.getOptionValue("bind");
398 conf.set("hbase.thrift.info.bindAddress", bindAddress);
399 } else {
400 bindAddress = conf.get("hbase.thrift.info.bindAddress");
401 }
402
403
404 if (cmd.hasOption("readonly")) {
405 conf.setBoolean("hbase.thrift.readonly", true);
406 if (log.isDebugEnabled()) {
407 log.debug("readonly set to true");
408 }
409 }
410
411
412 int readTimeout = THRIFT_SERVER_SOCKET_READ_TIMEOUT_DEFAULT;
413 if (cmd.hasOption(READ_TIMEOUT_OPTION)) {
414 try {
415 readTimeout = Integer.parseInt(cmd.getOptionValue(READ_TIMEOUT_OPTION));
416 } catch (NumberFormatException e) {
417 throw new RuntimeException("Could not parse the value provided for the timeout option", e);
418 }
419 } else {
420 readTimeout = conf.getInt(THRIFT_SERVER_SOCKET_READ_TIMEOUT_KEY,
421 THRIFT_SERVER_SOCKET_READ_TIMEOUT_DEFAULT);
422 }
423
424
425 int listenPort = 0;
426 try {
427 if (cmd.hasOption("port")) {
428 listenPort = Integer.parseInt(cmd.getOptionValue("port"));
429 } else {
430 listenPort = conf.getInt("hbase.regionserver.thrift.port", DEFAULT_LISTEN_PORT);
431 }
432 } catch (NumberFormatException e) {
433 throw new RuntimeException("Could not parse the value provided for the port option", e);
434 }
435
436
437 int backlog = conf.getInt(BACKLOG_CONF_KEY, 0);
438
439
440
441 String host = null;
442 String name = null;
443
444 UserProvider userProvider = UserProvider.instantiate(conf);
445
446 boolean securityEnabled = userProvider.isHadoopSecurityEnabled()
447 && userProvider.isHBaseSecurityEnabled();
448 if (securityEnabled) {
449 host = Strings.domainNamePointerToHostName(DNS.getDefaultHost(
450 conf.get("hbase.thrift.dns.interface", "default"),
451 conf.get("hbase.thrift.dns.nameserver", "default")));
452 userProvider.login("hbase.thrift.keytab.file",
453 "hbase.thrift.kerberos.principal", host);
454 }
455
456 UserGroupInformation realUser = userProvider.getCurrent().getUGI();
457 String stringQop = conf.get(THRIFT_QOP_KEY);
458 SaslUtil.QualityOfProtection qop = null;
459 if (stringQop != null) {
460 qop = SaslUtil.getQop(stringQop);
461 if (!securityEnabled) {
462 throw new IOException("Thrift server must"
463 + " run in secure mode to support authentication");
464 }
465
466 name = SecurityUtil.getUserFromPrincipal(
467 conf.get("hbase.thrift.kerberos.principal"));
468 }
469
470 boolean nonblocking = cmd.hasOption("nonblocking");
471 boolean hsha = cmd.hasOption("hsha");
472 boolean selector = cmd.hasOption("selector");
473
474 ThriftMetrics metrics = new ThriftMetrics(conf, ThriftMetrics.ThriftServerType.TWO);
475 final JvmPauseMonitor pauseMonitor = new JvmPauseMonitor(conf, metrics.getSource());
476
477 String implType = "threadpool";
478 if (nonblocking) {
479 implType = "nonblocking";
480 } else if (hsha) {
481 implType = "hsha";
482 } else if (selector) {
483 implType = "selector";
484 }
485
486 conf.set("hbase.regionserver.thrift.server.type", implType);
487 conf.setInt("hbase.regionserver.thrift.port", listenPort);
488 registerFilters(conf);
489
490
491 boolean compact = cmd.hasOption("compact") ||
492 conf.getBoolean("hbase.regionserver.thrift.compact", false);
493 TProtocolFactory protocolFactory = getTProtocolFactory(compact);
494 final ThriftHBaseServiceHandler hbaseHandler =
495 new ThriftHBaseServiceHandler(conf, userProvider);
496 THBaseService.Iface handler =
497 ThriftHBaseServiceHandler.newInstance(hbaseHandler, metrics);
498 final THBaseService.Processor p = new THBaseService.Processor(handler);
499 conf.setBoolean("hbase.regionserver.thrift.compact", compact);
500 TProcessor processor = p;
501
502 boolean framed = cmd.hasOption("framed") ||
503 conf.getBoolean("hbase.regionserver.thrift.framed", false) || nonblocking || hsha;
504 TTransportFactory transportFactory = getTTransportFactory(qop, name, host, framed,
505 conf.getInt("hbase.regionserver.thrift.framed.max_frame_size_in_mb", 2) * 1024 * 1024);
506 InetSocketAddress inetSocketAddress = bindToPort(bindAddress, listenPort);
507 conf.setBoolean("hbase.regionserver.thrift.framed", framed);
508 if (qop != null) {
509
510 processor = new TProcessor() {
511 @Override
512 public boolean process(TProtocol inProt,
513 TProtocol outProt) throws TException {
514 TSaslServerTransport saslServerTransport =
515 (TSaslServerTransport)inProt.getTransport();
516 SaslServer saslServer = saslServerTransport.getSaslServer();
517 String principal = saslServer.getAuthorizationID();
518 hbaseHandler.setEffectiveUser(principal);
519 return p.process(inProt, outProt);
520 }
521 };
522 }
523
524 if (cmd.hasOption("w")) {
525 workerThreads = Integer.parseInt(cmd.getOptionValue("w"));
526 }
527 if (cmd.hasOption("s")) {
528 selectorThreads = Integer.parseInt(cmd.getOptionValue("s"));
529 }
530 if (cmd.hasOption("q")) {
531 maxCallQueueSize = Integer.parseInt(cmd.getOptionValue("q"));
532 }
533
534
535 try {
536 if (cmd.hasOption("infoport")) {
537 String val = cmd.getOptionValue("infoport");
538 conf.setInt("hbase.thrift.info.port", Integer.parseInt(val));
539 log.debug("Web UI port set to " + val);
540 }
541 } catch (NumberFormatException e) {
542 log.error("Could not parse the value provided for the infoport option", e);
543 printUsage();
544 System.exit(1);
545 }
546
547
548 int port = conf.getInt("hbase.thrift.info.port", 9095);
549 if (port >= 0) {
550 conf.setLong("startcode", System.currentTimeMillis());
551 String a = conf.get("hbase.thrift.info.bindAddress", "0.0.0.0");
552 InfoServer infoServer = new InfoServer("thrift", a, port, false, conf);
553 infoServer.setAttribute("hbase.conf", conf);
554 infoServer.start();
555 }
556
557 if (nonblocking) {
558 server = getTNonBlockingServer(protocolFactory,
559 processor,
560 transportFactory,
561 inetSocketAddress);
562 } else if (hsha) {
563 server = getTHsHaServer(protocolFactory,
564 processor,
565 transportFactory,
566 workerThreads,
567 maxCallQueueSize,
568 inetSocketAddress,
569 metrics);
570 } else if (selector) {
571 server = getTThreadedSelectorServer(protocolFactory,
572 processor,
573 transportFactory,
574 workerThreads,
575 selectorThreads,
576 maxCallQueueSize,
577 inetSocketAddress,
578 metrics);
579 } else {
580 server = getTThreadPoolServer(protocolFactory,
581 processor,
582 transportFactory,
583 workerThreads,
584 inetSocketAddress,
585 backlog,
586 readTimeout,
587 metrics);
588 }
589
590 final TServer tserver = server;
591 realUser.doAs(
592 new PrivilegedAction<Object>() {
593 @Override
594 public Object run() {
595 pauseMonitor.start();
596 try {
597 tserver.serve();
598 return null;
599 } finally {
600 pauseMonitor.stop();
601 }
602 }
603 });
604 }
605 }