hbase-shell/src/main/ruby/hbase/quotas.rb

   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
  20 include Java
  21 java_import java.util.concurrent.TimeUnit
  22 java_import org.apache.hadoop.hbase.TableName
  23 java_import org.apache.hadoop.hbase.ServerName
  24 java_import org.apache.hadoop.hbase.quotas.ThrottleType
  25 java_import org.apache.hadoop.hbase.quotas.QuotaFilter
  26 java_import org.apache.hadoop.hbase.quotas.QuotaRetriever
  27 java_import org.apache.hadoop.hbase.quotas.QuotaScope
  28 java_import org.apache.hadoop.hbase.quotas.QuotaSettingsFactory
  29 java_import org.apache.hadoop.hbase.quotas.QuotaTableUtil
  30 java_import org.apache.hadoop.hbase.quotas.SpaceViolationPolicy
  31
  32 module HBaseQuotasConstants
  33   # RPC Quota constants
  34   GLOBAL_BYPASS = 'GLOBAL_BYPASS'.freeze
  35   THROTTLE_TYPE = 'THROTTLE_TYPE'.freeze
  36   THROTTLE = 'THROTTLE'.freeze
  37   REQUEST = 'REQUEST'.freeze
  38   WRITE = 'WRITE'.freeze
  39   READ = 'READ'.freeze
  40   SCOPE = 'SCOPE'.freeze
  41   CLUSTER = 'CLUSTER'.freeze
  42   MACHINE = 'MACHINE'.freeze
  43   # Space quota constants
  44   SPACE = 'SPACE'.freeze
  45   NO_INSERTS = 'NO_INSERTS'.freeze
  46   NO_WRITES = 'NO_WRITES'.freeze
  47   NO_WRITES_COMPACTIONS = 'NO_WRITES_COMPACTIONS'.freeze
  48   DISABLE = 'DISABLE'.freeze
  49 end
  50
  51 module Hbase
  52   # rubocop:disable Metrics/ClassLength
  53   class QuotasAdmin
  54     def initialize(admin)
  55       @admin = admin
  56     end
  57
  58     def close
  59       @admin.close
  60     end
  61
  62     def throttle(args)
  63       raise(ArgumentError, 'Arguments should be a Hash') unless args.is_a?(Hash)
  64       type = args.fetch(THROTTLE_TYPE, REQUEST)
  65       args.delete(THROTTLE_TYPE)
  66       type, limit, time_unit = _parse_limit(args.delete(LIMIT), ThrottleType, type)
  67       scope = _parse_scope(args.fetch(SCOPE, MACHINE))
  68       args.delete(SCOPE)
  69       if args.key?(USER)
  70         user = args.delete(USER)
  71         if args.key?(TABLE)
  72           table = TableName.valueOf(args.delete(TABLE))
  73           raise(ArgumentError, 'Unexpected arguments: ' + args.inspect) unless args.empty?
  74           settings = QuotaSettingsFactory.throttleUser(user, table, type, limit, time_unit, scope)
  75         elsif args.key?(NAMESPACE)
  76           namespace = args.delete(NAMESPACE)
  77           raise(ArgumentError, 'Unexpected arguments: ' + args.inspect) unless args.empty?
  78           settings = QuotaSettingsFactory.throttleUser(user, namespace, type, limit, time_unit, scope)
  79         else
  80           raise(ArgumentError, 'Unexpected arguments: ' + args.inspect) unless args.empty?
  81           settings = QuotaSettingsFactory.throttleUser(user, type, limit, time_unit, scope)
  82         end
  83       elsif args.key?(TABLE)
  84         table = TableName.valueOf(args.delete(TABLE))
  85         raise(ArgumentError, 'Unexpected arguments: ' + args.inspect) unless args.empty?
  86         settings = QuotaSettingsFactory.throttleTable(table, type, limit, time_unit, scope)
  87       elsif args.key?(NAMESPACE)
  88         namespace = args.delete(NAMESPACE)
  89         raise(ArgumentError, 'Unexpected arguments: ' + args.inspect) unless args.empty?
  90         settings = QuotaSettingsFactory.throttleNamespace(namespace, type, limit, time_unit, scope)
  91       elsif args.key?(REGIONSERVER)
  92         # TODO: Setting specified region server quota isn't supported currently and using 'all' for all RS
  93         if scope == QuotaScope.valueOf(CLUSTER)
  94           raise(ArgumentError, 'Invalid region server throttle scope, must be MACHINE')
  95         end
  96         settings = QuotaSettingsFactory.throttleRegionServer('all', type, limit, time_unit)
  97       else
  98         raise 'One of USER, TABLE, NAMESPACE or REGIONSERVER must be specified'
  99       end
 100       @admin.setQuota(settings)
 101     end
 102
 103     def unthrottle(args)
 104       raise(ArgumentError, 'Arguments should be a Hash') unless args.is_a?(Hash)
 105       if args.key?(USER)
 106         user = args.delete(USER)
 107         if args.key?(TABLE)
 108           table = TableName.valueOf(args.delete(TABLE))
 109           raise(ArgumentError, 'Unexpected arguments: ' + args.inspect) unless args.empty?
 110           settings = QuotaSettingsFactory.unthrottleUser(user, table)
 111         elsif args.key?(NAMESPACE)
 112           namespace = args.delete(NAMESPACE)
 113           raise(ArgumentError, 'Unexpected arguments: ' + args.inspect) unless args.empty?
 114           settings = QuotaSettingsFactory.unthrottleUser(user, namespace)
 115         else
 116           raise(ArgumentError, 'Unexpected arguments: ' + args.inspect) unless args.empty?
 117           settings = QuotaSettingsFactory.unthrottleUser(user)
 118         end
 119       elsif args.key?(TABLE)
 120         table = TableName.valueOf(args.delete(TABLE))
 121         raise(ArgumentError, 'Unexpected arguments: ' + args.inspect) unless args.empty?
 122         settings = QuotaSettingsFactory.unthrottleTable(table)
 123       elsif args.key?(NAMESPACE)
 124         namespace = args.delete(NAMESPACE)
 125         raise(ArgumentError, 'Unexpected arguments: ' + args.inspect) unless args.empty?
 126         settings = QuotaSettingsFactory.unthrottleNamespace(namespace)
 127       elsif args.key?(REGIONSERVER)
 128         regionServer = args.delete(REGIONSERVER)
 129         raise(ArgumentError, 'Unexpected arguments: ' + args.inspect) unless args.empty?
 130         # TODO: Setting specified region server quota isn't supported currently and using 'all' for all RS
 131         settings = QuotaSettingsFactory.unthrottleRegionServer('all')
 132       else
 133         raise 'One of USER, TABLE, NAMESPACE or REGIONSERVER must be specified'
 134       end
 135       @admin.setQuota(settings)
 136     end
 137
 138     # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity
 139     # rubocop:disable Metrics/MethodLength, Metrics/PerceivedComplexity
 140     def limit_space(args)
 141       raise(ArgumentError, 'Argument should be a Hash') unless !args.nil? && args.is_a?(Hash)
 142       # Let the user provide a raw number
 143       limit = if args[LIMIT].is_a?(Numeric)
 144                 args[LIMIT]
 145               else
 146                 # Parse a string a 1K, 2G, etc.
 147                 _parse_size(args[LIMIT])
 148               end
 149       if limit <= 0
 150         raise(ArgumentError, 'Invalid space limit, must be greater than 0')
 151       end
 152
 153       # Extract the policy, failing if something bogus was provided
 154       policy = SpaceViolationPolicy.valueOf(args[POLICY])
 155       # Create a table or namespace quota
 156       if args.key?(TABLE)
 157         if args.key?(NAMESPACE)
 158           raise(ArgumentError, 'Only one of TABLE or NAMESPACE can be specified.')
 159         end
 160         settings = QuotaSettingsFactory.limitTableSpace(TableName.valueOf(args.delete(TABLE)), limit, policy)
 161       elsif args.key?(NAMESPACE)
 162         if args.key?(TABLE)
 163           raise(ArgumentError, 'Only one of TABLE or NAMESPACE can be specified.')
 164         end
 165         settings = QuotaSettingsFactory.limitNamespaceSpace(args.delete(NAMESPACE), limit, policy)
 166       else
 167         raise(ArgumentError, 'One of TABLE or NAMESPACE must be specified.')
 168       end
 169       # Apply the quota
 170       @admin.setQuota(settings)
 171     end
 172     # rubocop:enable Metrics/AbcSize, Metrics/CyclomaticComplexity
 173     # rubocop:enable Metrics/MethodLength, Metrics/PerceivedComplexity
 174
 175     def remove_space_limit(args)
 176       raise(ArgumentError, 'Argument should be a Hash') unless !args.nil? && args.is_a?(Hash)
 177       if args.key?(TABLE)
 178         if args.key?(NAMESPACE)
 179           raise(ArgumentError, 'Only one of TABLE or NAMESPACE can be specified.')
 180         end
 181         table = TableName.valueOf(args.delete(TABLE))
 182         settings = QuotaSettingsFactory.removeTableSpaceLimit(table)
 183       elsif args.key?(NAMESPACE)
 184         if args.key?(TABLE)
 185           raise(ArgumentError, 'Only one of TABLE or NAMESPACE can be specified.')
 186         end
 187         settings = QuotaSettingsFactory.removeNamespaceSpaceLimit(args.delete(NAMESPACE))
 188       else
 189         raise(ArgumentError, 'One of TABLE or NAMESPACE must be specified.')
 190       end
 191       @admin.setQuota(settings)
 192     end
 193
 194     def get_master_table_sizes
 195       @admin.getSpaceQuotaTableSizes
 196     end
 197
 198     def get_quota_snapshots(regionserver = nil)
 199       # Ask a regionserver if we were given one
 200       return get_rs_quota_snapshots(regionserver) if regionserver
 201       # Otherwise, read from the quota table
 202       get_quota_snapshots_from_table
 203     end
 204
 205     def get_quota_snapshots_from_table
 206       # Reads the snapshots from the hbase:quota table
 207       QuotaTableUtil.getSnapshots(@admin.getConnection)
 208     end
 209
 210     def get_rs_quota_snapshots(rs)
 211       # Reads the snapshots from a specific regionserver
 212       @admin.getRegionServerSpaceQuotaSnapshots(ServerName.valueOf(rs))
 213     end
 214
 215     def set_global_bypass(bypass, args)
 216       raise(ArgumentError, 'Arguments should be a Hash') unless args.is_a?(Hash)
 217
 218       if args.key?(USER)
 219         user = args.delete(USER)
 220         raise(ArgumentError, 'Unexpected arguments: ' + args.inspect) unless args.empty?
 221         settings = QuotaSettingsFactory.bypassGlobals(user, bypass)
 222       else
 223         raise 'Expected USER'
 224       end
 225       @admin.setQuota(settings)
 226     end
 227
 228     def list_quotas(args = {})
 229       raise(ArgumentError, 'Arguments should be a Hash') unless args.is_a?(Hash)
 230
 231       limit = args.delete('LIMIT') || -1
 232       count = 0
 233
 234       filter = QuotaFilter.new
 235       filter.setUserFilter(args.delete(USER)) if args.key?(USER)
 236       filter.setTableFilter(args.delete(TABLE)) if args.key?(TABLE)
 237       filter.setNamespaceFilter(args.delete(NAMESPACE)) if args.key?(NAMESPACE)
 238       raise(ArgumentError, 'Unexpected arguments: ' + args.inspect) unless args.empty?
 239
 240       # Start the scanner
 241       quotas = @admin.getQuota(filter)
 242       iter = quotas.iterator
 243
 244       # Iterate results
 245       while iter.hasNext
 246         break if limit > 0 && count >= limit
 247
 248         settings = iter.next
 249         owner = {
 250           USER => settings.getUserName,
 251           TABLE => settings.getTableName,
 252           NAMESPACE => settings.getNamespace,
 253           REGIONSERVER => settings.getRegionServer
 254         }.delete_if { |_k, v| v.nil? }.map { |k, v| k.to_s + ' => ' + v.to_s } * ', '
 255
 256         yield owner, settings.to_s
 257
 258         count += 1
 259       end
 260       count
 261     end
 262
 263     def list_snapshot_sizes
 264       QuotaTableUtil.getObservedSnapshotSizes(@admin.getConnection)
 265     end
 266
 267     def switch_rpc_throttle(enabled)
 268       @admin.switchRpcThrottle(java.lang.Boolean.valueOf(enabled))
 269     end
 270
 271     def switch_exceed_throttle_quota(enabled)
 272       @admin.exceedThrottleQuotaSwitch(java.lang.Boolean.valueOf(enabled))
 273     end
 274
 275     def _parse_size(str_limit)
 276       str_limit = str_limit.downcase
 277       match = /^(\d+)([bkmgtp%]?)$/.match(str_limit)
 278       if match
 279         if match[2] == '%'
 280           return match[1].to_i
 281         else
 282           return _size_from_str(match[1].to_i, match[2])
 283         end
 284       else
 285         raise(ArgumentError, 'Invalid size limit syntax')
 286       end
 287     end
 288
 289     # rubocop:disable Metrics/AbcSize, Metrics/MethodLength
 290     def _parse_limit(str_limit, type_cls, type)
 291       str_limit = str_limit.downcase
 292       match = /^(\d+)(req|cu|[bkmgtp])\/(sec|min|hour|day)$/.match(str_limit)
 293       if match
 294         limit = match[1].to_i
 295         if match[2] == 'req'
 296           type = type_cls.valueOf(type + '_NUMBER')
 297         elsif match[2] == 'cu'
 298           type = type_cls.valueOf(type + '_CAPACITY_UNIT')
 299         else
 300           limit = _size_from_str(limit, match[2])
 301           type = type_cls.valueOf(type + '_SIZE')
 302         end
 303
 304         if limit <= 0
 305           raise(ArgumentError, 'Invalid throttle limit, must be greater than 0')
 306         end
 307
 308         case match[3]
 309         when 'sec'  then time_unit = TimeUnit::SECONDS
 310         when 'min'  then time_unit = TimeUnit::MINUTES
 311         when 'hour' then time_unit = TimeUnit::HOURS
 312         when 'day'  then time_unit = TimeUnit::DAYS
 313         end
 314
 315         return type, limit, time_unit
 316       else
 317         raise(ArgumentError, 'Invalid throttle limit syntax')
 318       end
 319     end
 320     # rubocop:enable Metrics/AbcSize, Metrics/MethodLength
 321
 322     def _size_from_str(value, suffix)
 323       case suffix
 324       when 'k' then value <<= 10
 325       when 'm' then value <<= 20
 326       when 'g' then value <<= 30
 327       when 't' then value <<= 40
 328       when 'p' then value <<= 50
 329       end
 330       value
 331     end
 332
 333     def _parse_scope(scope_str)
 334       scope_str = scope_str.upcase
 335       return QuotaScope.valueOf(scope_str) if [CLUSTER, MACHINE].include?(scope_str)
 336       unless raise(ArgumentError, 'Invalid throttle scope, must be either CLUSTER or MACHINE')
 337       end
 338     end
 339   end
 340   # rubocop:enable Metrics/ClassLength
 341 end