Source code

001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.thrift;
019
020import static org.apache.hadoop.hbase.http.HttpServerUtil.PATH_SPEC_ANY;
021import static org.apache.hadoop.hbase.thrift.Constants.BACKLOG_CONF_DEAFULT;
022import static org.apache.hadoop.hbase.thrift.Constants.BACKLOG_CONF_KEY;
023import static org.apache.hadoop.hbase.thrift.Constants.BIND_CONF_KEY;
024import static org.apache.hadoop.hbase.thrift.Constants.BIND_OPTION;
025import static org.apache.hadoop.hbase.thrift.Constants.COMPACT_CONF_DEFAULT;
026import static org.apache.hadoop.hbase.thrift.Constants.COMPACT_CONF_KEY;
027import static org.apache.hadoop.hbase.thrift.Constants.COMPACT_OPTION;
028import static org.apache.hadoop.hbase.thrift.Constants.DEFAULT_BIND_ADDR;
029import static org.apache.hadoop.hbase.thrift.Constants.DEFAULT_HTTP_MAX_HEADER_SIZE;
030import static org.apache.hadoop.hbase.thrift.Constants.DEFAULT_LISTEN_PORT;
031import static org.apache.hadoop.hbase.thrift.Constants.FRAMED_CONF_DEFAULT;
032import static org.apache.hadoop.hbase.thrift.Constants.FRAMED_CONF_KEY;
033import static org.apache.hadoop.hbase.thrift.Constants.FRAMED_OPTION;
034import static org.apache.hadoop.hbase.thrift.Constants.HTTP_MAX_THREADS_KEY;
035import static org.apache.hadoop.hbase.thrift.Constants.HTTP_MAX_THREADS_KEY_DEFAULT;
036import static org.apache.hadoop.hbase.thrift.Constants.HTTP_MIN_THREADS_KEY;
037import static org.apache.hadoop.hbase.thrift.Constants.HTTP_MIN_THREADS_KEY_DEFAULT;
038import static org.apache.hadoop.hbase.thrift.Constants.INFOPORT_OPTION;
039import static org.apache.hadoop.hbase.thrift.Constants.KEEP_ALIVE_SEC_OPTION;
040import static org.apache.hadoop.hbase.thrift.Constants.MAX_FRAME_SIZE_CONF_DEFAULT;
041import static org.apache.hadoop.hbase.thrift.Constants.MAX_FRAME_SIZE_CONF_KEY;
042import static org.apache.hadoop.hbase.thrift.Constants.MAX_QUEUE_SIZE_OPTION;
043import static org.apache.hadoop.hbase.thrift.Constants.MAX_WORKERS_OPTION;
044import static org.apache.hadoop.hbase.thrift.Constants.MIN_WORKERS_OPTION;
045import static org.apache.hadoop.hbase.thrift.Constants.PORT_CONF_KEY;
046import static org.apache.hadoop.hbase.thrift.Constants.PORT_OPTION;
047import static org.apache.hadoop.hbase.thrift.Constants.READ_TIMEOUT_OPTION;
048import static org.apache.hadoop.hbase.thrift.Constants.SELECTOR_NUM_OPTION;
049import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_FILTERS;
050import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_HTTP_ALLOW_OPTIONS_METHOD;
051import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_HTTP_ALLOW_OPTIONS_METHOD_DEFAULT;
052import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_INFO_SERVER_BINDING_ADDRESS;
053import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_INFO_SERVER_BINDING_ADDRESS_DEFAULT;
054import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_INFO_SERVER_PORT;
055import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_INFO_SERVER_PORT_DEFAULT;
056import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_KERBEROS_PRINCIPAL_KEY;
057import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_KEYTAB_FILE_KEY;
058import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_QOP_KEY;
059import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SELECTOR_NUM;
060import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SERVER_SOCKET_READ_TIMEOUT_DEFAULT;
061import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SERVER_SOCKET_READ_TIMEOUT_KEY;
062import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SPNEGO_KEYTAB_FILE_KEY;
063import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SPNEGO_PRINCIPAL_KEY;
064import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_ENABLED_KEY;
065import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_EXCLUDE_CIPHER_SUITES_KEY;
066import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_EXCLUDE_PROTOCOLS_KEY;
067import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_INCLUDE_CIPHER_SUITES_KEY;
068import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_INCLUDE_PROTOCOLS_KEY;
069import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_KEYSTORE_KEYPASSWORD_KEY;
070import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_KEYSTORE_PASSWORD_KEY;
071import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_KEYSTORE_STORE_KEY;
072import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_KEYSTORE_TYPE_DEFAULT;
073import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_KEYSTORE_TYPE_KEY;
074import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SUPPORT_PROXYUSER_KEY;
075import static org.apache.hadoop.hbase.thrift.Constants.USE_HTTP_CONF_KEY;
076
077import java.io.IOException;
078import java.net.InetAddress;
079import java.net.InetSocketAddress;
080import java.net.UnknownHostException;
081import java.security.PrivilegedAction;
082import java.util.List;
083import java.util.Map;
084import java.util.concurrent.BlockingQueue;
085import java.util.concurrent.ExecutorService;
086import java.util.concurrent.LinkedBlockingQueue;
087import java.util.concurrent.ThreadPoolExecutor;
088import java.util.concurrent.TimeUnit;
089import javax.security.auth.callback.Callback;
090import javax.security.auth.callback.UnsupportedCallbackException;
091import javax.security.sasl.AuthorizeCallback;
092import javax.security.sasl.SaslServer;
093import org.apache.commons.lang3.ArrayUtils;
094import org.apache.hadoop.conf.Configuration;
095import org.apache.hadoop.conf.Configured;
096import org.apache.hadoop.hbase.HBaseConfiguration;
097import org.apache.hadoop.hbase.HBaseInterfaceAudience;
098import org.apache.hadoop.hbase.filter.ParseFilter;
099import org.apache.hadoop.hbase.http.HttpServerUtil;
100import org.apache.hadoop.hbase.http.InfoServer;
101import org.apache.hadoop.hbase.log.HBaseMarkers;
102import org.apache.hadoop.hbase.security.SaslUtil;
103import org.apache.hadoop.hbase.security.SecurityUtil;
104import org.apache.hadoop.hbase.security.UserProvider;
105import org.apache.hadoop.hbase.thrift.generated.Hbase;
106import org.apache.hadoop.hbase.util.DNS;
107import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
108import org.apache.hadoop.hbase.util.JvmPauseMonitor;
109import org.apache.hadoop.hbase.util.VersionInfo;
110import org.apache.hadoop.security.SaslRpcServer;
111import org.apache.hadoop.security.UserGroupInformation;
112import org.apache.hadoop.security.authorize.ProxyUsers;
113import org.apache.hadoop.util.Shell.ExitCodeException;
114import org.apache.hadoop.util.Tool;
115import org.apache.hadoop.util.ToolRunner;
116import org.apache.yetus.audience.InterfaceAudience;
117import org.slf4j.Logger;
118import org.slf4j.LoggerFactory;
119
120import org.apache.hbase.thirdparty.com.google.common.base.Joiner;
121import org.apache.hbase.thirdparty.com.google.common.base.Splitter;
122import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
123import org.apache.hbase.thirdparty.org.apache.commons.cli.CommandLine;
124import org.apache.hbase.thirdparty.org.apache.commons.cli.CommandLineParser;
125import org.apache.hbase.thirdparty.org.apache.commons.cli.DefaultParser;
126import org.apache.hbase.thirdparty.org.apache.commons.cli.HelpFormatter;
127import org.apache.hbase.thirdparty.org.apache.commons.cli.Options;
128import org.apache.hbase.thirdparty.org.apache.thrift.TProcessor;
129import org.apache.hbase.thirdparty.org.apache.thrift.protocol.TBinaryProtocol;
130import org.apache.hbase.thirdparty.org.apache.thrift.protocol.TCompactProtocol;
131import org.apache.hbase.thirdparty.org.apache.thrift.protocol.TProtocolFactory;
132import org.apache.hbase.thirdparty.org.apache.thrift.server.THsHaServer;
133import org.apache.hbase.thirdparty.org.apache.thrift.server.TNonblockingServer;
134import org.apache.hbase.thirdparty.org.apache.thrift.server.TServer;
135import org.apache.hbase.thirdparty.org.apache.thrift.server.TServlet;
136import org.apache.hbase.thirdparty.org.apache.thrift.server.TThreadedSelectorServer;
137import org.apache.hbase.thirdparty.org.apache.thrift.transport.TNonblockingServerSocket;
138import org.apache.hbase.thirdparty.org.apache.thrift.transport.TNonblockingServerTransport;
139import org.apache.hbase.thirdparty.org.apache.thrift.transport.TSaslServerTransport;
140import org.apache.hbase.thirdparty.org.apache.thrift.transport.TServerSocket;
141import org.apache.hbase.thirdparty.org.apache.thrift.transport.TServerTransport;
142import org.apache.hbase.thirdparty.org.apache.thrift.transport.TTransportFactory;
143import org.apache.hbase.thirdparty.org.apache.thrift.transport.layered.TFramedTransport;
144import org.apache.hbase.thirdparty.org.eclipse.jetty.ee8.servlet.ServletContextHandler;
145import org.apache.hbase.thirdparty.org.eclipse.jetty.ee8.servlet.ServletHolder;
146import org.apache.hbase.thirdparty.org.eclipse.jetty.http.HttpVersion;
147import org.apache.hbase.thirdparty.org.eclipse.jetty.server.HttpConfiguration;
148import org.apache.hbase.thirdparty.org.eclipse.jetty.server.HttpConnectionFactory;
149import org.apache.hbase.thirdparty.org.eclipse.jetty.server.SecureRequestCustomizer;
150import org.apache.hbase.thirdparty.org.eclipse.jetty.server.Server;
151import org.apache.hbase.thirdparty.org.eclipse.jetty.server.ServerConnector;
152import org.apache.hbase.thirdparty.org.eclipse.jetty.server.SslConnectionFactory;
153import org.apache.hbase.thirdparty.org.eclipse.jetty.util.ssl.SslContextFactory;
154import org.apache.hbase.thirdparty.org.eclipse.jetty.util.thread.QueuedThreadPool;
155
156/**
157 * ThriftServer- this class starts up a Thrift server which implements the Hbase API specified in
158 * the Hbase.thrift IDL file. The server runs in an independent process.
159 */
160@InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.TOOLS)
161public class ThriftServer extends Configured implements Tool {
162
163  private static final Logger LOG = LoggerFactory.getLogger(ThriftServer.class);
164
165  protected Configuration conf;
166
167  protected InfoServer infoServer;
168
169  protected TProcessor processor;
170
171  protected ThriftMetrics metrics;
172  protected HBaseServiceHandler hbaseServiceHandler;
173  protected UserGroupInformation serviceUGI;
174  protected UserGroupInformation httpUGI;
175  protected boolean httpEnabled;
176
177  protected SaslUtil.QualityOfProtection qop;
178  protected String host;
179  protected int listenPort;
180
181  protected boolean securityEnabled;
182  protected boolean doAsEnabled;
183
184  protected JvmPauseMonitor pauseMonitor;
185
186  protected volatile TServer tserver;
187  protected volatile Server httpServer;
188
189  //
190  // Main program and support routines
191  //
192
193  public ThriftServer(Configuration conf) {
194    this.conf = HBaseConfiguration.create(conf);
195  }
196
197  protected ThriftMetrics createThriftMetrics(Configuration conf) {
198    return new ThriftMetrics(conf, ThriftMetrics.ThriftServerType.ONE);
199  }
200
201  protected void setupParamters() throws IOException {
202    // login the server principal (if using secure Hadoop)
203    UserProvider userProvider = UserProvider.instantiate(conf);
204    securityEnabled =
205      userProvider.isHadoopSecurityEnabled() && userProvider.isHBaseSecurityEnabled();
206    if (securityEnabled) {
207      host = DNS.getHostname(conf, DNS.ServerType.THRIFT);
208      userProvider.login(THRIFT_KEYTAB_FILE_KEY, THRIFT_KERBEROS_PRINCIPAL_KEY, host);
209
210      // Setup the SPNEGO user for HTTP if configured
211      String spnegoPrincipal = getSpengoPrincipal(conf, host);
212      String spnegoKeytab = getSpnegoKeytab(conf);
213      UserGroupInformation.setConfiguration(conf);
214      // login the SPNEGO principal using UGI to avoid polluting the login user
215      this.httpUGI =
216        UserGroupInformation.loginUserFromKeytabAndReturnUGI(spnegoPrincipal, spnegoKeytab);
217    }
218    this.serviceUGI = userProvider.getCurrent().getUGI();
219    if (httpUGI == null) {
220      this.httpUGI = serviceUGI;
221    }
222
223    this.listenPort = conf.getInt(PORT_CONF_KEY, DEFAULT_LISTEN_PORT);
224    this.metrics = createThriftMetrics(conf);
225    this.pauseMonitor = new JvmPauseMonitor(conf, this.metrics.getSource());
226    this.hbaseServiceHandler = createHandler(conf, userProvider);
227    this.hbaseServiceHandler.initMetrics(metrics);
228    this.processor = createProcessor();
229
230    httpEnabled = conf.getBoolean(USE_HTTP_CONF_KEY, false);
231    doAsEnabled = conf.getBoolean(THRIFT_SUPPORT_PROXYUSER_KEY, false);
232    if (doAsEnabled && !httpEnabled) {
233      LOG.warn("Fail to enable the doAs feature. " + USE_HTTP_CONF_KEY + " is not configured");
234    }
235
236    String strQop = conf.get(THRIFT_QOP_KEY);
237    if (strQop != null) {
238      this.qop = SaslUtil.getQop(strQop);
239    }
240    if (qop != null) {
241      if (
242        qop != SaslUtil.QualityOfProtection.AUTHENTICATION
243          && qop != SaslUtil.QualityOfProtection.INTEGRITY
244          && qop != SaslUtil.QualityOfProtection.PRIVACY
245      ) {
246        throw new IOException(String.format("Invalid %s: It must be one of %s, %s, or %s.",
247          THRIFT_QOP_KEY, SaslUtil.QualityOfProtection.AUTHENTICATION.name(),
248          SaslUtil.QualityOfProtection.INTEGRITY.name(),
249          SaslUtil.QualityOfProtection.PRIVACY.name()));
250      }
251      checkHttpSecurity(qop, conf);
252      if (!securityEnabled) {
253        throw new IOException("Thrift server must run in secure mode to support authentication");
254      }
255    }
256    registerFilters(conf);
257    pauseMonitor.start();
258  }
259
260  private String getSpengoPrincipal(Configuration conf, String host) throws IOException {
261    String principal = conf.get(THRIFT_SPNEGO_PRINCIPAL_KEY);
262    if (principal == null) {
263      // We cannot use the Hadoop configuration deprecation handling here since
264      // the THRIFT_KERBEROS_PRINCIPAL_KEY config is still valid for regular Kerberos
265      // communication. The preference should be to use the THRIFT_SPNEGO_PRINCIPAL_KEY
266      // config so that THRIFT_KERBEROS_PRINCIPAL_KEY doesn't control both backend
267      // Kerberos principal and SPNEGO principal.
268      LOG.info("Using deprecated {} config for SPNEGO principal. Use {} instead.",
269        THRIFT_KERBEROS_PRINCIPAL_KEY, THRIFT_SPNEGO_PRINCIPAL_KEY);
270      principal = conf.get(THRIFT_KERBEROS_PRINCIPAL_KEY);
271    }
272    // Handle _HOST in principal value
273    return org.apache.hadoop.security.SecurityUtil.getServerPrincipal(principal, host);
274  }
275
276  private String getSpnegoKeytab(Configuration conf) {
277    String keytab = conf.get(THRIFT_SPNEGO_KEYTAB_FILE_KEY);
278    if (keytab == null) {
279      // We cannot use the Hadoop configuration deprecation handling here since
280      // the THRIFT_KEYTAB_FILE_KEY config is still valid for regular Kerberos
281      // communication. The preference should be to use the THRIFT_SPNEGO_KEYTAB_FILE_KEY
282      // config so that THRIFT_KEYTAB_FILE_KEY doesn't control both backend
283      // Kerberos keytab and SPNEGO keytab.
284      LOG.info("Using deprecated {} config for SPNEGO keytab. Use {} instead.",
285        THRIFT_KEYTAB_FILE_KEY, THRIFT_SPNEGO_KEYTAB_FILE_KEY);
286      keytab = conf.get(THRIFT_KEYTAB_FILE_KEY);
287    }
288    return keytab;
289  }
290
291  protected void startInfoServer() throws IOException {
292    // Put up info server.
293    int port = conf.getInt(THRIFT_INFO_SERVER_PORT, THRIFT_INFO_SERVER_PORT_DEFAULT);
294
295    if (port >= 0) {
296      conf.setLong("startcode", EnvironmentEdgeManager.currentTime());
297      String a =
298        conf.get(THRIFT_INFO_SERVER_BINDING_ADDRESS, THRIFT_INFO_SERVER_BINDING_ADDRESS_DEFAULT);
299      infoServer = new InfoServer("thrift", a, port, false, conf);
300      infoServer.setAttribute("hbase.conf", conf);
301      infoServer.setAttribute("hbase.thrift.server.type", metrics.getThriftServerType().name());
302      infoServer.start();
303    }
304  }
305
306  protected void checkHttpSecurity(SaslUtil.QualityOfProtection qop, Configuration conf) {
307    if (
308      qop == SaslUtil.QualityOfProtection.PRIVACY && conf.getBoolean(USE_HTTP_CONF_KEY, false)
309        && !conf.getBoolean(THRIFT_SSL_ENABLED_KEY, false)
310    ) {
311      throw new IllegalArgumentException(
312        "Thrift HTTP Server's QoP is privacy, but " + THRIFT_SSL_ENABLED_KEY + " is false");
313    }
314  }
315
316  protected HBaseServiceHandler createHandler(Configuration conf, UserProvider userProvider)
317    throws IOException {
318    return new ThriftHBaseServiceHandler(conf, userProvider);
319  }
320
321  protected TProcessor createProcessor() {
322    return new Hbase.Processor<>(
323      HbaseHandlerMetricsProxy.newInstance((Hbase.Iface) hbaseServiceHandler, metrics, conf));
324  }
325
326  /**
327   * the thrift server, not null means the server is started, for test only
328   * @return the tServer
329   */
330  @InterfaceAudience.Private
331  public TServer getTserver() {
332    return tserver;
333  }
334
335  /**
336   * the Jetty server, not null means the HTTP server is started, for test only
337   * @return the http server
338   */
339  @InterfaceAudience.Private
340  public Server getHttpServer() {
341    return httpServer;
342  }
343
344  protected void printUsageAndExit(Options options, int exitCode) throws ExitCodeException {
345    HelpFormatter formatter = new HelpFormatter();
346    formatter.printHelp("Thrift", null, options,
347      "To start the Thrift server run 'hbase-daemon.sh start thrift' or " + "'hbase thrift'\n"
348        + "To shutdown the thrift server run 'hbase-daemon.sh stop "
349        + "thrift' or send a kill signal to the thrift server pid",
350      true);
351    System.exit(exitCode);
352  }
353
354  /**
355   * Create a Servlet for the http server
356   * @param protocolFactory protocolFactory
357   * @return the servlet
358   */
359  protected TServlet createTServlet(TProtocolFactory protocolFactory) {
360    return new ThriftHttpServlet(processor, protocolFactory, serviceUGI, httpUGI,
361      hbaseServiceHandler, securityEnabled, doAsEnabled);
362  }
363
364  /**
365   * Setup an HTTP Server using Jetty to serve calls from THttpClient
366   * @throws IOException IOException
367   */
368  protected void setupHTTPServer() throws IOException {
369    TProtocolFactory protocolFactory = new TBinaryProtocol.Factory();
370    TServlet thriftHttpServlet = createTServlet(protocolFactory);
371
372    // Set the default max thread number to 100 to limit
373    // the number of concurrent requests so that Thrfit HTTP server doesn't OOM easily.
374    // Jetty set the default max thread number to 250, if we don't set it.
375    //
376    // Our default min thread number 2 is the same as that used by Jetty.
377    int minThreads = conf.getInt(HTTP_MIN_THREADS_KEY, conf
378      .getInt(TBoundedThreadPoolServer.MIN_WORKER_THREADS_CONF_KEY, HTTP_MIN_THREADS_KEY_DEFAULT));
379    int maxThreads = conf.getInt(HTTP_MAX_THREADS_KEY, conf
380      .getInt(TBoundedThreadPoolServer.MAX_WORKER_THREADS_CONF_KEY, HTTP_MAX_THREADS_KEY_DEFAULT));
381    QueuedThreadPool threadPool = new QueuedThreadPool(maxThreads);
382    threadPool.setMinThreads(minThreads);
383    httpServer = new Server(threadPool);
384
385    // Context handler
386    boolean isSecure = conf.getBoolean(THRIFT_SSL_ENABLED_KEY, false);
387    ServletContextHandler ctxHandler =
388      new ServletContextHandler(httpServer, "/", ServletContextHandler.SESSIONS);
389    HttpServerUtil.addClickjackingPreventionFilter(ctxHandler, conf, PATH_SPEC_ANY);
390    HttpServerUtil.addSecurityHeadersFilter(ctxHandler, conf, isSecure, PATH_SPEC_ANY);
391    ctxHandler.addServlet(new ServletHolder(thriftHttpServlet), PATH_SPEC_ANY);
392    HttpServerUtil.constrainHttpMethods(ctxHandler,
393      conf.getBoolean(THRIFT_HTTP_ALLOW_OPTIONS_METHOD, THRIFT_HTTP_ALLOW_OPTIONS_METHOD_DEFAULT));
394
395    // set up Jetty and run the embedded server
396    HttpConfiguration httpConfig = new HttpConfiguration();
397    httpConfig.setSecureScheme("https");
398    httpConfig.setSecurePort(listenPort);
399    httpConfig.setHeaderCacheSize(DEFAULT_HTTP_MAX_HEADER_SIZE);
400    httpConfig.setRequestHeaderSize(DEFAULT_HTTP_MAX_HEADER_SIZE);
401    httpConfig.setResponseHeaderSize(DEFAULT_HTTP_MAX_HEADER_SIZE);
402    httpConfig.setSendServerVersion(false);
403    httpConfig.setSendDateHeader(false);
404
405    ServerConnector serverConnector;
406    if (isSecure) {
407      HttpConfiguration httpsConfig = new HttpConfiguration(httpConfig);
408      httpsConfig.addCustomizer(new SecureRequestCustomizer());
409
410      SslContextFactory.Server sslCtxFactory = new SslContextFactory.Server();
411      String keystore = conf.get(THRIFT_SSL_KEYSTORE_STORE_KEY);
412      String password =
413        HBaseConfiguration.getPassword(conf, THRIFT_SSL_KEYSTORE_PASSWORD_KEY, null);
414      String keyPassword =
415        HBaseConfiguration.getPassword(conf, THRIFT_SSL_KEYSTORE_KEYPASSWORD_KEY, password);
416      sslCtxFactory.setKeyStorePath(keystore);
417      sslCtxFactory.setKeyStorePassword(password);
418      sslCtxFactory.setKeyManagerPassword(keyPassword);
419      sslCtxFactory
420        .setKeyStoreType(conf.get(THRIFT_SSL_KEYSTORE_TYPE_KEY, THRIFT_SSL_KEYSTORE_TYPE_DEFAULT));
421
422      String[] excludeCiphers =
423        conf.getStrings(THRIFT_SSL_EXCLUDE_CIPHER_SUITES_KEY, ArrayUtils.EMPTY_STRING_ARRAY);
424      if (excludeCiphers.length != 0) {
425        sslCtxFactory.setExcludeCipherSuites(excludeCiphers);
426      }
427      String[] includeCiphers =
428        conf.getStrings(THRIFT_SSL_INCLUDE_CIPHER_SUITES_KEY, ArrayUtils.EMPTY_STRING_ARRAY);
429      if (includeCiphers.length != 0) {
430        sslCtxFactory.setIncludeCipherSuites(includeCiphers);
431      }
432
433      // Disable SSLv3 by default due to "Poodle" Vulnerability - CVE-2014-3566
434      String[] excludeProtocols = conf.getStrings(THRIFT_SSL_EXCLUDE_PROTOCOLS_KEY, "SSLv3");
435      if (excludeProtocols.length != 0) {
436        sslCtxFactory.setExcludeProtocols(excludeProtocols);
437      }
438      String[] includeProtocols =
439        conf.getStrings(THRIFT_SSL_INCLUDE_PROTOCOLS_KEY, ArrayUtils.EMPTY_STRING_ARRAY);
440      if (includeProtocols.length != 0) {
441        sslCtxFactory.setIncludeProtocols(includeProtocols);
442      }
443
444      serverConnector = new ServerConnector(httpServer,
445        new SslConnectionFactory(sslCtxFactory, HttpVersion.HTTP_1_1.toString()),
446        new HttpConnectionFactory(httpsConfig));
447    } else {
448      serverConnector = new ServerConnector(httpServer, new HttpConnectionFactory(httpConfig));
449    }
450    serverConnector.setPort(listenPort);
451    serverConnector.setHost(getBindAddress(conf).getHostAddress());
452    httpServer.addConnector(serverConnector);
453    httpServer.setStopAtShutdown(true);
454
455    if (doAsEnabled) {
456      ProxyUsers.refreshSuperUserGroupsConfiguration(conf);
457    }
458    LOG.info("Starting Thrift HTTP Server on {}", Integer.toString(listenPort));
459  }
460
461  /**
462   * Setting up the thrift TServer
463   */
464  protected void setupServer() throws Exception {
465    // Construct correct ProtocolFactory
466    TProtocolFactory protocolFactory = getProtocolFactory();
467
468    ImplType implType = ImplType.getServerImpl(conf);
469    TProcessor processorToUse = processor;
470
471    // Construct correct TransportFactory
472    TTransportFactory transportFactory;
473    if (conf.getBoolean(FRAMED_CONF_KEY, FRAMED_CONF_DEFAULT) || implType.isAlwaysFramed) {
474      if (qop != null) {
475        throw new RuntimeException(
476          "Thrift server authentication" + " doesn't work with framed transport yet");
477      }
478      transportFactory = new TFramedTransport.Factory(
479        conf.getInt(MAX_FRAME_SIZE_CONF_KEY, MAX_FRAME_SIZE_CONF_DEFAULT) * 1024 * 1024);
480      LOG.debug("Using framed transport");
481    } else if (qop == null) {
482      transportFactory = new TTransportFactory();
483    } else {
484      // Extract the name from the principal
485      String thriftKerberosPrincipal = conf.get(THRIFT_KERBEROS_PRINCIPAL_KEY);
486      if (thriftKerberosPrincipal == null) {
487        throw new IllegalArgumentException(THRIFT_KERBEROS_PRINCIPAL_KEY + " cannot be null");
488      }
489      String name = SecurityUtil.getUserFromPrincipal(thriftKerberosPrincipal);
490      Map<String, String> saslProperties = SaslUtil.initSaslProperties(qop.name());
491      TSaslServerTransport.Factory saslFactory = new TSaslServerTransport.Factory();
492      saslFactory.addServerDefinition("GSSAPI", name, host, saslProperties,
493        new SaslRpcServer.SaslGssCallbackHandler() {
494          @Override
495          public void handle(Callback[] callbacks) throws UnsupportedCallbackException {
496            AuthorizeCallback ac = null;
497            for (Callback callback : callbacks) {
498              if (callback instanceof AuthorizeCallback) {
499                ac = (AuthorizeCallback) callback;
500              } else {
501                throw new UnsupportedCallbackException(callback,
502                  "Unrecognized SASL GSSAPI Callback");
503              }
504            }
505            if (ac != null) {
506              String authid = ac.getAuthenticationID();
507              String authzid = ac.getAuthorizationID();
508              if (!authid.equals(authzid)) {
509                ac.setAuthorized(false);
510              } else {
511                ac.setAuthorized(true);
512                String userName = SecurityUtil.getUserFromPrincipal(authzid);
513                LOG.info("Effective user: {}", userName);
514                ac.setAuthorizedID(userName);
515              }
516            }
517          }
518        });
519      transportFactory = saslFactory;
520
521      // Create a processor wrapper, to get the caller
522      processorToUse = (inProt, outProt) -> {
523        TSaslServerTransport saslServerTransport = (TSaslServerTransport) inProt.getTransport();
524        SaslServer saslServer = saslServerTransport.getSaslServer();
525        String principal = saslServer.getAuthorizationID();
526        hbaseServiceHandler.setEffectiveUser(principal);
527        processor.process(inProt, outProt);
528      };
529    }
530
531    if (conf.get(BIND_CONF_KEY) != null && !implType.canSpecifyBindIP) {
532      LOG.error(
533        "Server types {} don't support IP address binding at the moment. See "
534          + "https://issues.apache.org/jira/browse/HBASE-2155 for details.",
535        Joiner.on(", ").join(ImplType.serversThatCannotSpecifyBindIP()));
536      throw new RuntimeException("-" + BIND_CONF_KEY + " not supported with " + implType);
537    }
538
539    InetSocketAddress inetSocketAddress = new InetSocketAddress(getBindAddress(conf), listenPort);
540    if (
541      implType == ImplType.HS_HA || implType == ImplType.NONBLOCKING
542        || implType == ImplType.THREADED_SELECTOR
543    ) {
544      TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(inetSocketAddress);
545      if (implType == ImplType.NONBLOCKING) {
546        tserver = getTNonBlockingServer(serverTransport, protocolFactory, processorToUse,
547          transportFactory, inetSocketAddress);
548      } else if (implType == ImplType.HS_HA) {
549        tserver = getTHsHaServer(serverTransport, protocolFactory, processorToUse, transportFactory,
550          inetSocketAddress);
551      } else { // THREADED_SELECTOR
552        tserver = getTThreadedSelectorServer(serverTransport, protocolFactory, processorToUse,
553          transportFactory, inetSocketAddress);
554      }
555      LOG.info("starting HBase {} server on {}", implType.simpleClassName(),
556        Integer.toString(listenPort));
557    } else if (implType == ImplType.THREAD_POOL) {
558      this.tserver =
559        getTThreadPoolServer(protocolFactory, processorToUse, transportFactory, inetSocketAddress);
560    } else {
561      throw new AssertionError(
562        "Unsupported Thrift server implementation: " + implType.simpleClassName());
563    }
564
565    // A sanity check that we instantiated the right type of server.
566    if (tserver.getClass() != implType.serverClass) {
567      throw new AssertionError("Expected to create Thrift server class "
568        + implType.serverClass.getName() + " but got " + tserver.getClass().getName());
569    }
570  }
571
572  protected TServer getTNonBlockingServer(TNonblockingServerTransport serverTransport,
573    TProtocolFactory protocolFactory, TProcessor processor, TTransportFactory transportFactory,
574    InetSocketAddress inetSocketAddress) {
575    LOG.info("starting HBase Nonblocking Thrift server on " + inetSocketAddress.toString());
576    TNonblockingServer.Args serverArgs = new TNonblockingServer.Args(serverTransport);
577    serverArgs.processor(processor);
578    serverArgs.transportFactory(transportFactory);
579    serverArgs.protocolFactory(protocolFactory);
580    return new TNonblockingServer(serverArgs);
581  }
582
583  protected TServer getTHsHaServer(TNonblockingServerTransport serverTransport,
584    TProtocolFactory protocolFactory, TProcessor processor, TTransportFactory transportFactory,
585    InetSocketAddress inetSocketAddress) {
586    LOG.info("starting HBase HsHA Thrift server on " + inetSocketAddress.toString());
587    THsHaServer.Args serverArgs = new THsHaServer.Args(serverTransport);
588    int queueSize = conf.getInt(TBoundedThreadPoolServer.MAX_QUEUED_REQUESTS_CONF_KEY,
589      TBoundedThreadPoolServer.DEFAULT_MAX_QUEUED_REQUESTS);
590    CallQueue callQueue = new CallQueue(new LinkedBlockingQueue<>(queueSize), metrics);
591    int workerThread = conf.getInt(TBoundedThreadPoolServer.MAX_WORKER_THREADS_CONF_KEY,
592      serverArgs.getMaxWorkerThreads());
593    ExecutorService executorService = createExecutor(callQueue, workerThread, workerThread);
594    serverArgs.executorService(executorService).processor(processor)
595      .transportFactory(transportFactory).protocolFactory(protocolFactory);
596    return new THsHaServer(serverArgs);
597  }
598
599  protected TServer getTThreadedSelectorServer(TNonblockingServerTransport serverTransport,
600    TProtocolFactory protocolFactory, TProcessor processor, TTransportFactory transportFactory,
601    InetSocketAddress inetSocketAddress) {
602    LOG.info("starting HBase ThreadedSelector Thrift server on " + inetSocketAddress.toString());
603    TThreadedSelectorServer.Args serverArgs =
604      new HThreadedSelectorServerArgs(serverTransport, conf);
605    int queueSize = conf.getInt(TBoundedThreadPoolServer.MAX_QUEUED_REQUESTS_CONF_KEY,
606      TBoundedThreadPoolServer.DEFAULT_MAX_QUEUED_REQUESTS);
607    CallQueue callQueue = new CallQueue(new LinkedBlockingQueue<>(queueSize), metrics);
608    int workerThreads = conf.getInt(TBoundedThreadPoolServer.MAX_WORKER_THREADS_CONF_KEY,
609      serverArgs.getWorkerThreads());
610    int selectorThreads = conf.getInt(THRIFT_SELECTOR_NUM, serverArgs.getSelectorThreads());
611    serverArgs.selectorThreads(selectorThreads);
612    ExecutorService executorService = createExecutor(callQueue, workerThreads, workerThreads);
613    serverArgs.executorService(executorService).processor(processor)
614      .transportFactory(transportFactory).protocolFactory(protocolFactory);
615    return new TThreadedSelectorServer(serverArgs);
616  }
617
618  protected TServer getTThreadPoolServer(TProtocolFactory protocolFactory, TProcessor processor,
619    TTransportFactory transportFactory, InetSocketAddress inetSocketAddress) throws Exception {
620    LOG.info("starting HBase ThreadPool Thrift server on " + inetSocketAddress.toString());
621    // Thrift's implementation uses '0' as a placeholder for 'use the default.'
622    int backlog = conf.getInt(BACKLOG_CONF_KEY, BACKLOG_CONF_DEAFULT);
623    int readTimeout =
624      conf.getInt(THRIFT_SERVER_SOCKET_READ_TIMEOUT_KEY, THRIFT_SERVER_SOCKET_READ_TIMEOUT_DEFAULT);
625    TServerTransport serverTransport =
626      new TServerSocket(new TServerSocket.ServerSocketTransportArgs().bindAddr(inetSocketAddress)
627        .backlog(backlog).clientTimeout(readTimeout));
628
629    TBoundedThreadPoolServer.Args serverArgs =
630      new TBoundedThreadPoolServer.Args(serverTransport, conf);
631    serverArgs.processor(processor).transportFactory(transportFactory)
632      .protocolFactory(protocolFactory);
633    return new TBoundedThreadPoolServer(serverArgs, metrics);
634  }
635
636  protected TProtocolFactory getProtocolFactory() {
637    TProtocolFactory protocolFactory;
638
639    if (conf.getBoolean(COMPACT_CONF_KEY, COMPACT_CONF_DEFAULT)) {
640      LOG.debug("Using compact protocol");
641      protocolFactory = new TCompactProtocol.Factory();
642    } else {
643      LOG.debug("Using binary protocol");
644      protocolFactory = new TBinaryProtocol.Factory();
645    }
646
647    return protocolFactory;
648  }
649
650  protected ExecutorService createExecutor(BlockingQueue<Runnable> callQueue, int minWorkers,
651    int maxWorkers) {
652    ThreadFactoryBuilder tfb = new ThreadFactoryBuilder();
653    tfb.setDaemon(true);
654    tfb.setNameFormat("thrift-worker-%d");
655    ThreadPoolExecutor threadPool = new THBaseThreadPoolExecutor(minWorkers, maxWorkers,
656      Long.MAX_VALUE, TimeUnit.SECONDS, callQueue, tfb.build(), metrics);
657    threadPool.allowCoreThreadTimeOut(true);
658    return threadPool;
659  }
660
661  protected InetAddress getBindAddress(Configuration conf) throws UnknownHostException {
662    String bindAddressStr = conf.get(BIND_CONF_KEY, DEFAULT_BIND_ADDR);
663    return InetAddress.getByName(bindAddressStr);
664  }
665
666  public static void registerFilters(Configuration conf) {
667    String[] filters = conf.getStrings(THRIFT_FILTERS);
668    Splitter splitter = Splitter.on(':');
669    if (filters != null) {
670      for (String filterClass : filters) {
671        List<String> filterPart = splitter.splitToList(filterClass);
672        if (filterPart.size() != 2) {
673          LOG.warn("Invalid filter specification " + filterClass + " - skipping");
674        } else {
675          ParseFilter.registerFilter(filterPart.get(0), filterPart.get(1));
676        }
677      }
678    }
679  }
680
681  /**
682   * Add options to command lines
683   * @param options options
684   */
685  protected void addOptions(Options options) {
686    options.addOption("b", BIND_OPTION, true,
687      "Address to bind " + "the Thrift server to. [default: " + DEFAULT_BIND_ADDR + "]");
688    options.addOption("p", PORT_OPTION, true,
689      "Port to bind to [default: " + DEFAULT_LISTEN_PORT + "]");
690    options.addOption("f", FRAMED_OPTION, false, "Use framed transport");
691    options.addOption("c", COMPACT_OPTION, false, "Use the compact protocol");
692    options.addOption("h", "help", false, "Print help information");
693    options.addOption("s", SELECTOR_NUM_OPTION, true, "How many selector threads to use.");
694    options.addOption(null, INFOPORT_OPTION, true, "Port for web UI");
695
696    options.addOption("m", MIN_WORKERS_OPTION, true,
697      "The minimum number of worker threads for " + ImplType.THREAD_POOL.simpleClassName());
698
699    options.addOption("w", MAX_WORKERS_OPTION, true,
700      "The maximum number of worker threads for " + ImplType.THREAD_POOL.simpleClassName());
701
702    options.addOption("q", MAX_QUEUE_SIZE_OPTION, true,
703      "The maximum number of queued requests in " + ImplType.THREAD_POOL.simpleClassName());
704
705    options.addOption("k", KEEP_ALIVE_SEC_OPTION, true,
706      "The amount of time in secods to keep a thread alive when idle in "
707        + ImplType.THREAD_POOL.simpleClassName());
708
709    options.addOption("t", READ_TIMEOUT_OPTION, true,
710      "Amount of time in milliseconds before a server thread will timeout "
711        + "waiting for client to send data on a connected socket. Currently, "
712        + "only applies to TBoundedThreadPoolServer");
713
714    options.addOptionGroup(ImplType.createOptionGroup());
715  }
716
717  protected void parseCommandLine(CommandLine cmd, Options options) throws ExitCodeException {
718    // Get port to bind to
719    try {
720      if (cmd.hasOption(PORT_OPTION)) {
721        int listenPort = Integer.parseInt(cmd.getOptionValue(PORT_OPTION));
722        conf.setInt(PORT_CONF_KEY, listenPort);
723      }
724    } catch (NumberFormatException e) {
725      LOG.error("Could not parse the value provided for the port option", e);
726      printUsageAndExit(options, -1);
727    }
728    // check for user-defined info server port setting, if so override the conf
729    try {
730      if (cmd.hasOption(INFOPORT_OPTION)) {
731        String val = cmd.getOptionValue(INFOPORT_OPTION);
732        conf.setInt(THRIFT_INFO_SERVER_PORT, Integer.parseInt(val));
733        LOG.debug("Web UI port set to " + val);
734      }
735    } catch (NumberFormatException e) {
736      LOG.error("Could not parse the value provided for the " + INFOPORT_OPTION + " option", e);
737      printUsageAndExit(options, -1);
738    }
739    // Make optional changes to the configuration based on command-line options
740    optionToConf(cmd, MIN_WORKERS_OPTION, conf,
741      TBoundedThreadPoolServer.MIN_WORKER_THREADS_CONF_KEY);
742    optionToConf(cmd, MAX_WORKERS_OPTION, conf,
743      TBoundedThreadPoolServer.MAX_WORKER_THREADS_CONF_KEY);
744    optionToConf(cmd, MAX_QUEUE_SIZE_OPTION, conf,
745      TBoundedThreadPoolServer.MAX_QUEUED_REQUESTS_CONF_KEY);
746    optionToConf(cmd, KEEP_ALIVE_SEC_OPTION, conf,
747      TBoundedThreadPoolServer.THREAD_KEEP_ALIVE_TIME_SEC_CONF_KEY);
748    optionToConf(cmd, READ_TIMEOUT_OPTION, conf, THRIFT_SERVER_SOCKET_READ_TIMEOUT_KEY);
749    optionToConf(cmd, SELECTOR_NUM_OPTION, conf, THRIFT_SELECTOR_NUM);
750
751    // Set general thrift server options
752    boolean compact = cmd.hasOption(COMPACT_OPTION) || conf.getBoolean(COMPACT_CONF_KEY, false);
753    conf.setBoolean(COMPACT_CONF_KEY, compact);
754    boolean framed = cmd.hasOption(FRAMED_OPTION) || conf.getBoolean(FRAMED_CONF_KEY, false);
755    conf.setBoolean(FRAMED_CONF_KEY, framed);
756
757    optionToConf(cmd, BIND_OPTION, conf, BIND_CONF_KEY);
758
759    ImplType.setServerImpl(cmd, conf);
760  }
761
762  /**
763   * Parse the command line options to set parameters the conf.
764   */
765  protected void processOptions(final String[] args) throws Exception {
766    if (args == null || args.length == 0) {
767      return;
768    }
769    Options options = new Options();
770    addOptions(options);
771
772    CommandLineParser parser = new DefaultParser();
773    CommandLine cmd = parser.parse(options, args);
774
775    if (cmd.hasOption("help")) {
776      printUsageAndExit(options, 0);
777    }
778    parseCommandLine(cmd, options);
779  }
780
781  public void stop() {
782    if (this.infoServer != null) {
783      LOG.info("Stopping infoServer");
784      try {
785        this.infoServer.stop();
786      } catch (Exception ex) {
787        LOG.error("Failed to stop infoServer", ex);
788      }
789    }
790    if (pauseMonitor != null) {
791      pauseMonitor.stop();
792    }
793    if (tserver != null) {
794      tserver.stop();
795      tserver = null;
796    }
797    if (httpServer != null) {
798      try {
799        httpServer.stop();
800        httpServer = null;
801      } catch (Exception e) {
802        LOG.error("Problem encountered in shutting down HTTP server", e);
803      }
804      httpServer = null;
805    }
806  }
807
808  protected static void optionToConf(CommandLine cmd, String option, Configuration conf,
809    String destConfKey) {
810    if (cmd.hasOption(option)) {
811      String value = cmd.getOptionValue(option);
812      LOG.info("Set configuration key:" + destConfKey + " value:" + value);
813      conf.set(destConfKey, value);
814    }
815  }
816
817  /**
818   * Run without any command line arguments
819   * @return exit code
820   * @throws Exception exception
821   */
822  public int run() throws Exception {
823    return run(null);
824  }
825
826  @Override
827  public int run(String[] strings) throws Exception {
828    processOptions(strings);
829    setupParamters();
830    if (httpEnabled) {
831      setupHTTPServer();
832    } else {
833      setupServer();
834    }
835    return serviceUGI.doAs(new PrivilegedAction<Integer>() {
836      @Override
837      public Integer run() {
838        try {
839          startInfoServer();
840          if (httpEnabled) {
841            httpServer.start();
842            httpServer.join();
843          } else {
844            tserver.serve();
845          }
846          return 0;
847        } catch (Exception e) {
848          LOG.error(HBaseMarkers.FATAL, "Cannot run ThriftServer", e);
849          return -1;
850        }
851      }
852    });
853  }
854
855  public static void main(String[] args) throws Exception {
856    LOG.info("***** STARTING service '" + ThriftServer.class.getSimpleName() + "' *****");
857    VersionInfo.logVersion();
858    final Configuration conf = HBaseConfiguration.create();
859    // for now, only time we return is on an argument error.
860    final int status = ToolRunner.run(conf, new ThriftServer(conf), args);
861    LOG.info("***** STOPPING service '" + ThriftServer.class.getSimpleName() + "' *****");
862    System.exit(status);
863  }
864}