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