View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.util;
20  
21  import java.io.IOException;
22  import java.util.Map;
23  import java.util.concurrent.ConcurrentHashMap;
24  import java.util.concurrent.locks.Lock;
25  
26  import org.apache.hadoop.conf.Configuration;
27  import org.apache.hadoop.hbase.ChoreService;
28  import org.apache.hadoop.hbase.ScheduledChore;
29  import org.apache.hadoop.hbase.Stoppable;
30  import org.apache.hadoop.hbase.classification.InterfaceAudience;
31  import org.apache.hadoop.hbase.client.HBaseAdmin;
32  import org.apache.hadoop.hbase.client.HConnection;
33  import org.apache.hadoop.hbase.client.HConnectionManager;
34  import org.apache.hadoop.hbase.client.HTableInterface;
35  import org.apache.hadoop.hbase.security.User;
36  import org.apache.hadoop.hbase.security.UserProvider;
37  import org.apache.hadoop.security.UserGroupInformation;
38  import org.apache.log4j.Logger;
39  
40  /**
41   * A utility to store user specific HConnections in memory.
42   * There is a chore to clean up connections idle for too long.
43   * This class is used by REST server and Thrift server to
44   * support authentication and impersonation.
45   */
46  @InterfaceAudience.Private
47  public class ConnectionCache {
48    private static Logger LOG = Logger.getLogger(ConnectionCache.class);
49  
50    private final Map<String, ConnectionInfo>
51     connections = new ConcurrentHashMap<String, ConnectionInfo>();
52    private final KeyLocker<String> locker = new KeyLocker<String>();
53    private final String realUserName;
54    private final UserGroupInformation realUser;
55    private final UserProvider userProvider;
56    private final Configuration conf;
57    private final ChoreService choreService;
58  
59    private final ThreadLocal<String> effectiveUserNames =
60        new ThreadLocal<String>() {
61      protected String initialValue() {
62        return realUserName;
63      }
64    };
65  
66    public ConnectionCache(final Configuration conf,
67        final UserProvider userProvider,
68        final int cleanInterval, final int maxIdleTime) throws IOException {
69      Stoppable stoppable = new Stoppable() {
70        private volatile boolean isStopped = false;
71        @Override public void stop(String why) { isStopped = true;}
72        @Override public boolean isStopped() {return isStopped;}
73      };
74      this.choreService = new ChoreService("ConnectionCache");
75      ScheduledChore cleaner = new ScheduledChore("ConnectionCleaner", stoppable, cleanInterval) {
76        @Override
77        protected void chore() {
78          for (Map.Entry<String, ConnectionInfo> entry: connections.entrySet()) {
79            ConnectionInfo connInfo = entry.getValue();
80            if (connInfo.timedOut(maxIdleTime)) {
81              if (connInfo.admin != null) {
82                try {
83                  connInfo.admin.close();
84                } catch (Throwable t) {
85                  LOG.info("Got exception in closing idle admin", t);
86                }
87              }
88              try {
89                connInfo.connection.close();
90              } catch (Throwable t) {
91                LOG.info("Got exception in closing idle connection", t);
92              }
93            }
94          }
95        }
96      };
97      // Start the daemon cleaner chore
98      choreService.scheduleChore(cleaner);
99      this.realUser = userProvider.getCurrent().getUGI();
100     this.realUserName = realUser.getShortUserName();
101     this.userProvider = userProvider;
102     this.conf = conf;
103   }
104 
105   /**
106    * Set the current thread local effective user
107    */
108   public void setEffectiveUser(String user) {
109     effectiveUserNames.set(user);
110   }
111 
112   /**
113    * Get the current thread local effective user
114    */
115   public String getEffectiveUser() {
116     return effectiveUserNames.get();
117   }
118 
119   /**
120    * Called when cache is no longer needed so that it can perform cleanup operations
121    */
122   public void shutdown() {
123     if (choreService != null) choreService.shutdown();
124   }
125 
126   /**
127    * Caller doesn't close the admin afterwards.
128    * We need to manage it and close it properly.
129    */
130   @SuppressWarnings("deprecation")
131   public HBaseAdmin getAdmin() throws IOException {
132     ConnectionInfo connInfo = getCurrentConnection();
133     if (connInfo.admin == null) {
134       Lock lock = locker.acquireLock(getEffectiveUser());
135       try {
136         if (connInfo.admin == null) {
137           connInfo.admin = new HBaseAdmin(connInfo.connection);
138         }
139       } finally {
140         lock.unlock();
141       }
142     }
143     return connInfo.admin;
144   }
145 
146   /**
147    * Caller closes the table afterwards.
148    */
149   public HTableInterface getTable(String tableName) throws IOException {
150     ConnectionInfo connInfo = getCurrentConnection();
151     return connInfo.connection.getTable(tableName);
152   }
153 
154   /**
155    * Get the cached connection for the current user.
156    * If none or timed out, create a new one.
157    */
158   ConnectionInfo getCurrentConnection() throws IOException {
159     String userName = getEffectiveUser();
160     ConnectionInfo connInfo = connections.get(userName);
161     if (connInfo == null || !connInfo.updateAccessTime()) {
162       Lock lock = locker.acquireLock(userName);
163       try {
164         connInfo = connections.get(userName);
165         if (connInfo == null) {
166           UserGroupInformation ugi = realUser;
167           if (!userName.equals(realUserName)) {
168             ugi = UserGroupInformation.createProxyUser(userName, realUser);
169           }
170           User user = userProvider.create(ugi);
171           HConnection conn = HConnectionManager.createConnection(conf, user);
172           connInfo = new ConnectionInfo(conn, userName);
173           connections.put(userName, connInfo);
174         }
175       } finally {
176         lock.unlock();
177       }
178     }
179     return connInfo;
180   }
181 
182   class ConnectionInfo {
183     final HConnection connection;
184     final String userName;
185 
186     volatile HBaseAdmin admin;
187     private long lastAccessTime;
188     private boolean closed;
189 
190     ConnectionInfo(HConnection conn, String user) {
191       lastAccessTime = EnvironmentEdgeManager.currentTime();
192       connection = conn;
193       closed = false;
194       userName = user;
195     }
196 
197     synchronized boolean updateAccessTime() {
198       if (closed) {
199         return false;
200       }
201       if (connection.isAborted() || connection.isClosed()) {
202         LOG.info("Unexpected: cached HConnection is aborted/closed, removed from cache");
203         connections.remove(userName);
204         return false;
205       }
206       lastAccessTime = EnvironmentEdgeManager.currentTime();
207       return true;
208     }
209 
210     synchronized boolean timedOut(int maxIdleTime) {
211       long timeoutTime = lastAccessTime + maxIdleTime;
212       if (EnvironmentEdgeManager.currentTime() > timeoutTime) {
213         connections.remove(userName);
214         closed = true;
215       }
216       return false;
217     }
218   }
219 }