| # |
| # |
| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| # |
| |
| include Java |
| java_import java.util.Arrays |
| java_import java.util.regex.Pattern |
| java_import org.apache.hadoop.hbase.util.Pair |
| java_import org.apache.hadoop.hbase.util.RegionSplitter |
| java_import org.apache.hadoop.hbase.util.Bytes |
| java_import org.apache.hadoop.hbase.ServerName |
| java_import org.apache.hadoop.hbase.TableName |
| java_import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder |
| java_import org.apache.hadoop.hbase.client.CoprocessorDescriptorBuilder |
| java_import org.apache.hadoop.hbase.client.TableDescriptorBuilder |
| java_import org.apache.hadoop.hbase.HConstants |
| |
| # Wrapper for org.apache.hadoop.hbase.client.HBaseAdmin |
| |
| module Hbase |
| # rubocop:disable Metrics/ClassLength |
| class Admin |
| include HBaseConstants |
| |
| def initialize(connection) |
| @connection = connection |
| # Java Admin instance |
| @admin = @connection.getAdmin |
| @hbck = @connection.getHbck |
| @conf = @connection.getConfiguration |
| end |
| |
| def close |
| @admin.close |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Returns a list of tables in hbase |
| def list(regex = '.*') |
| @admin.listTableNames(Pattern.compile(regex)).map(&:getNameAsString) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Requests a table or region or region server flush |
| def flush(name, family = nil) |
| family_bytes = nil |
| family_bytes = family.to_java_bytes unless family.nil? |
| if family_bytes.nil? |
| @admin.flushRegion(name.to_java_bytes) |
| else |
| @admin.flushRegion(name.to_java_bytes, family_bytes) |
| end |
| rescue java.lang.IllegalArgumentException, org.apache.hadoop.hbase.UnknownRegionException |
| # Unknown region. Try table. |
| begin |
| if family_bytes.nil? |
| @admin.flush(TableName.valueOf(name)) |
| else |
| @admin.flush(TableName.valueOf(name), family_bytes) |
| end |
| rescue java.lang.IllegalArgumentException |
| # Unknown table. Try region server. |
| @admin.flushRegionServer(ServerName.valueOf(name)) |
| end |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Requests a table or region or column family compaction |
| def compact(table_or_region_name, family = nil, type = 'NORMAL') |
| family_bytes = nil |
| family_bytes = family.to_java_bytes unless family.nil? |
| compact_type = nil |
| if type == 'NORMAL' |
| compact_type = org.apache.hadoop.hbase.client.CompactType::NORMAL |
| elsif type == 'MOB' |
| compact_type = org.apache.hadoop.hbase.client.CompactType::MOB |
| else |
| raise ArgumentError, 'only NORMAL or MOB accepted for type!' |
| end |
| |
| begin |
| if family_bytes.nil? |
| @admin.compactRegion(table_or_region_name.to_java_bytes) |
| else |
| @admin.compactRegion(table_or_region_name.to_java_bytes, family_bytes) |
| end |
| rescue java.lang.IllegalArgumentException, org.apache.hadoop.hbase.UnknownRegionException |
| if family_bytes.nil? |
| @admin.compact(TableName.valueOf(table_or_region_name), compact_type) |
| else |
| @admin.compact(TableName.valueOf(table_or_region_name), family_bytes, compact_type) |
| end |
| end |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Switch compaction on/off at runtime on a region server |
| def compaction_switch(on_or_off, regionserver_names) |
| region_servers = regionserver_names.flatten.compact |
| servers = java.util.ArrayList.new |
| if region_servers.any? |
| region_servers.each do |s| |
| servers.add(s) |
| end |
| end |
| @admin.compactionSwitch(java.lang.Boolean.valueOf(on_or_off), servers) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Gets compaction state for specified table |
| def getCompactionState(table_name) |
| @admin.getCompactionState(TableName.valueOf(table_name)).name |
| end |
| |
| # Requests to compact all regions on the regionserver |
| def compact_regionserver(servername, major = false) |
| if major |
| @admin.majorCompactRegionServer(ServerName.valueOf(servername)) |
| else |
| @admin.compactRegionServer(ServerName.valueOf(servername)) |
| end |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Requests a table or region or column family major compaction |
| def major_compact(table_or_region_name, family = nil, type = 'NORMAL') |
| family_bytes = nil |
| family_bytes = family.to_java_bytes unless family.nil? |
| compact_type = nil |
| if type == 'NORMAL' |
| compact_type = org.apache.hadoop.hbase.client.CompactType::NORMAL |
| elsif type == 'MOB' |
| compact_type = org.apache.hadoop.hbase.client.CompactType::MOB |
| else |
| raise ArgumentError, 'only NORMAL or MOB accepted for type!' |
| end |
| |
| begin |
| if family_bytes.nil? |
| @admin.majorCompactRegion(table_or_region_name.to_java_bytes) |
| else |
| @admin.majorCompactRegion(table_or_region_name.to_java_bytes, family_bytes) |
| end |
| rescue java.lang.IllegalArgumentException, org.apache.hadoop.hbase.UnknownRegionException |
| if family_bytes.nil? |
| @admin.majorCompact(TableName.valueOf(table_or_region_name), compact_type) |
| else |
| @admin.majorCompact(TableName.valueOf(table_or_region_name), family_bytes, compact_type) |
| end |
| end |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Requests a regionserver's WAL roll |
| def wal_roll(server_name) |
| @admin.rollWALWriter(ServerName.valueOf(server_name)) |
| end |
| # TODO: remove older hlog_roll version |
| alias hlog_roll wal_roll |
| |
| #---------------------------------------------------------------------------------------------- |
| # Requests a table or region split |
| def split(table_or_region_name, split_point = nil) |
| split_point_bytes = nil |
| split_point_bytes = split_point.to_java_bytes unless split_point.nil? |
| begin |
| if split_point_bytes.nil? |
| org.apache.hadoop.hbase.util.FutureUtils.get(@admin.splitRegionAsync(table_or_region_name.to_java_bytes)) |
| else |
| org.apache.hadoop.hbase.util.FutureUtils.get(@admin.splitRegionAsync(table_or_region_name.to_java_bytes, split_point_bytes)) |
| end |
| rescue java.lang.IllegalArgumentException, org.apache.hadoop.hbase.UnknownRegionException |
| if split_point_bytes.nil? |
| @admin.split(TableName.valueOf(table_or_region_name)) |
| else |
| @admin.split(TableName.valueOf(table_or_region_name), split_point_bytes) |
| end |
| end |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Enable/disable one split or merge switch |
| # Returns previous switch setting. |
| def splitormerge_switch(type, enabled) |
| if type == 'SPLIT' |
| @admin.splitSwitch(java.lang.Boolean.valueOf(enabled), java.lang.Boolean.valueOf(false)) |
| elsif type == 'MERGE' |
| @admin.mergeSwitch(java.lang.Boolean.valueOf(enabled), java.lang.Boolean.valueOf(false)) |
| else |
| raise ArgumentError, 'only SPLIT or MERGE accepted for type!' |
| end |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Query the current state of the split or merge switch. |
| # Returns the switch's state (true is enabled). |
| def splitormerge_enabled(type) |
| if type == 'SPLIT' |
| @admin.isSplitEnabled |
| elsif type == 'MERGE' |
| @admin.isMergeEnabled |
| else |
| raise ArgumentError, 'only SPLIT or MERGE accepted for type!' |
| end |
| end |
| |
| def locate_region(table_name, row_key) |
| locator = @connection.getRegionLocator(TableName.valueOf(table_name)) |
| begin |
| return locator.getRegionLocation(Bytes.toBytesBinary(row_key)) |
| ensure |
| locator.close |
| end |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Requests a cluster balance |
| # Returns true if balancer ran |
| def balancer(force) |
| @admin.balance(java.lang.Boolean.valueOf(force)) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Enable/disable balancer |
| # Returns previous balancer switch setting. |
| def balance_switch(enableDisable) |
| @admin.balancerSwitch( |
| java.lang.Boolean.valueOf(enableDisable), java.lang.Boolean.valueOf(false) |
| ) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Query the current state of the LoadBalancer. |
| # Returns the balancer's state (true is enabled). |
| def balancer_enabled? |
| @admin.isBalancerEnabled |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Requests clear block cache for table |
| def clear_block_cache(table_name) |
| @admin.clearBlockCache(org.apache.hadoop.hbase.TableName.valueOf(table_name)).toString |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Requests region normalization for all configured tables in the cluster |
| # Returns true if normalize request was successfully submitted |
| def normalize(*args) |
| builder = org.apache.hadoop.hbase.client.NormalizeTableFilterParams::Builder.new |
| args.each do |arg| |
| unless arg.is_a?(String) || arg.is_a?(Hash) |
| raise(ArgumentError, "#{arg.class} of #{arg.inspect} is not of Hash or String type") |
| end |
| |
| if arg.key?(TABLE_NAME) |
| table_name = arg.delete(TABLE_NAME) |
| unless table_name.is_a?(String) |
| raise(ArgumentError, "#{TABLE_NAME} must be of type String") |
| end |
| |
| builder.tableNames(java.util.Collections.singletonList(TableName.valueOf(table_name))) |
| elsif arg.key?(TABLE_NAMES) |
| table_names = arg.delete(TABLE_NAMES) |
| unless table_names.is_a?(Array) |
| raise(ArgumentError, "#{TABLE_NAMES} must be of type Array") |
| end |
| |
| table_name_list = java.util.LinkedList.new |
| table_names.each do |tn| |
| unless tn.is_a?(String) |
| raise(ArgumentError, "#{TABLE_NAMES} value #{tn} must be of type String") |
| end |
| |
| table_name_list.add(TableName.valueOf(tn)) |
| end |
| builder.tableNames(table_name_list) |
| elsif arg.key?(REGEX) |
| regex = arg.delete(REGEX) |
| raise(ArgumentError, "#{REGEX} must be of type String") unless regex.is_a?(String) |
| |
| builder.regex(regex) |
| elsif arg.key?(NAMESPACE) |
| namespace = arg.delete(NAMESPACE) |
| unless namespace.is_a?(String) |
| raise(ArgumentError, "#{NAMESPACE} must be of type String") |
| end |
| |
| builder.namespace(namespace) |
| else |
| raise(ArgumentError, "Unrecognized argument #{arg}") |
| end |
| end |
| ntfp = builder.build |
| @admin.normalize(ntfp) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Enable/disable region normalizer |
| # Returns previous normalizer switch setting. |
| def normalizer_switch(enableDisable) |
| @admin.normalizerSwitch(java.lang.Boolean.valueOf(enableDisable)) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Query the current state of region normalizer. |
| # Returns the state of region normalizer (true is enabled). |
| def normalizer_enabled? |
| @admin.isNormalizerEnabled |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Query the current state of master in maintenance mode. |
| # Returns the state of maintenance mode (true is on). |
| def in_maintenance_mode? |
| @admin.isMasterInMaintenanceMode |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Request HBCK chore to run |
| def hbck_chore_run |
| @hbck.runHbckChore |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Request a scan of the catalog table (for garbage collection) |
| # Returns an int signifying the number of entries cleaned |
| def catalogjanitor_run |
| @admin.runCatalogJanitor |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Enable/disable the catalog janitor |
| # Returns previous catalog janitor switch setting. |
| def catalogjanitor_switch(enableDisable) |
| @admin.catalogJanitorSwitch(java.lang.Boolean.valueOf(enableDisable)) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Query on the catalog janitor state (enabled/disabled?) |
| # Returns catalog janitor state (true signifies enabled). |
| def catalogjanitor_enabled |
| @admin.isCatalogJanitorEnabled |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Request cleaner chore to run (for garbage collection of HFiles and WAL files) |
| def cleaner_chore_run |
| @admin.runCleanerChore |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Enable/disable the cleaner chore |
| # Returns previous cleaner switch setting. |
| def cleaner_chore_switch(enableDisable) |
| @admin.cleanerChoreSwitch(java.lang.Boolean.valueOf(enableDisable)) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Query on the cleaner chore state (enabled/disabled?) |
| # Returns cleaner state (true signifies enabled). |
| def cleaner_chore_enabled |
| @admin.isCleanerChoreEnabled |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Enables a table |
| def enable(table_name) |
| tableExists(table_name) |
| return if enabled?(table_name) |
| @admin.enableTable(TableName.valueOf(table_name)) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Enables all tables matching the given regex |
| def enable_all(regex) |
| pattern = Pattern.compile(regex.to_s) |
| failed = java.util.ArrayList.new |
| @admin.listTableNames(pattern).each do |table_name| |
| begin |
| @admin.enableTable(table_name) |
| rescue java.io.IOException => e |
| puts "table:#{table_name}, error:#{e.toString}" |
| failed.add(table_name) |
| end |
| end |
| failed |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Disables a table |
| def disable(table_name) |
| tableExists(table_name) |
| return if disabled?(table_name) |
| @admin.disableTable(TableName.valueOf(table_name)) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Disables all tables matching the given regex |
| def disable_all(regex) |
| pattern = Pattern.compile(regex.to_s) |
| failed = java.util.ArrayList.new |
| @admin.listTableNames(pattern).each do |table_name| |
| begin |
| @admin.disableTable(table_name) |
| rescue java.io.IOException => e |
| puts "table:#{table_name}, error:#{e.toString}" |
| failed.add(table_name) |
| end |
| end |
| failed |
| end |
| |
| #--------------------------------------------------------------------------------------------- |
| # Throw exception if table doesn't exist |
| def tableExists(table_name) |
| raise ArgumentError, "Table #{table_name} does not exist." unless exists?(table_name) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Is table disabled? |
| def disabled?(table_name) |
| @admin.isTableDisabled(TableName.valueOf(table_name)) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Drops a table |
| def drop(table_name) |
| tableExists(table_name) |
| raise ArgumentError, "Table #{table_name} is enabled. Disable it first." if enabled?( |
| table_name |
| ) |
| |
| @admin.deleteTable(org.apache.hadoop.hbase.TableName.valueOf(table_name)) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Drops a table |
| def drop_all(regex) |
| pattern = Pattern.compile(regex.to_s) |
| failed = java.util.ArrayList.new |
| @admin.listTableNames(pattern).each do |table_name| |
| begin |
| @admin.deleteTable(table_name) |
| rescue java.io.IOException => e |
| puts puts "table:#{table_name}, error:#{e.toString}" |
| failed.add(table_name) |
| end |
| end |
| failed |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Returns ZooKeeper status dump |
| def zk_dump |
| @zk_wrapper = org.apache.hadoop.hbase.zookeeper.ZKWatcher.new( |
| @admin.getConfiguration, |
| 'admin', |
| nil |
| ) |
| zk = @zk_wrapper.getRecoverableZooKeeper.getZooKeeper |
| @zk_main = org.apache.zookeeper.ZooKeeperMain.new(zk) |
| org.apache.hadoop.hbase.zookeeper.ZKUtil.dump(@zk_wrapper) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Creates a table |
| def create(table_name, *args) |
| # Fail if table name is not a string |
| raise(ArgumentError, 'Table name must be of type String') unless table_name.is_a?(String) |
| |
| # Flatten params array |
| args = args.flatten.compact |
| has_columns = false |
| |
| # Start defining the table |
| tdb = TableDescriptorBuilder.newBuilder(TableName.valueOf(table_name)) |
| splits = nil |
| # Args are either columns or splits, add them to the table definition |
| # TODO: add table options support |
| args.each do |arg| |
| unless arg.is_a?(String) || arg.is_a?(Hash) |
| raise(ArgumentError, "#{arg.class} of #{arg.inspect} is not of Hash or String type") |
| end |
| |
| # First, handle all the cases where arg is a column family. |
| if arg.is_a?(String) || arg.key?(NAME) |
| # If the arg is a string, default action is to add a column to the table. |
| # If arg has a name, it must also be a column descriptor. |
| descriptor = cfd(arg, tdb) |
| # Warn if duplicate columns are added |
| if tdb.build.hasColumnFamily(descriptor.getName) |
| puts "Family '" + descriptor.getNameAsString + "' already exists, the old one will be replaced" |
| tdb.modifyColumnFamily(descriptor) |
| else |
| tdb.setColumnFamily(descriptor) |
| end |
| has_columns = true |
| next |
| end |
| if arg.key?(REGION_REPLICATION) |
| region_replication = JInteger.valueOf(arg.delete(REGION_REPLICATION)) |
| tdb.setRegionReplication(region_replication) |
| end |
| |
| # Get rid of the "METHOD", which is deprecated for create. |
| # We'll do whatever it used to do below if it's table_att. |
| if (method = arg.delete(METHOD)) |
| raise(ArgumentError, 'table_att is currently the only supported method') unless method == 'table_att' |
| end |
| |
| # The hash is not a column family. Figure out what's in it. |
| # First, handle splits. |
| if arg.key?(SPLITS_FILE) |
| splits_file = arg.delete(SPLITS_FILE) |
| unless File.exist?(splits_file) |
| raise(ArgumentError, "Splits file #{splits_file} doesn't exist") |
| end |
| arg[SPLITS] = [] |
| File.foreach(splits_file) do |line| |
| arg[SPLITS].push(line.chomp) |
| end |
| tdb.setValue(SPLITS_FILE, splits_file) |
| end |
| |
| if arg.key?(SPLITS) |
| splits = Java::byte[][arg[SPLITS].size].new |
| idx = 0 |
| arg.delete(SPLITS).each do |split| |
| splits[idx] = org.apache.hadoop.hbase.util.Bytes.toBytesBinary(split) |
| idx += 1 |
| end |
| elsif arg.key?(NUMREGIONS) || arg.key?(SPLITALGO) |
| # deprecated region pre-split API; if one of the above is specified, will be ignored. |
| raise(ArgumentError, 'Number of regions must be specified') unless arg.key?(NUMREGIONS) |
| raise(ArgumentError, 'Split algorithm must be specified') unless arg.key?(SPLITALGO) |
| raise(ArgumentError, 'Number of regions must be greater than 1') unless arg[NUMREGIONS] > 1 |
| num_regions = arg.delete(NUMREGIONS) |
| split_algo = RegionSplitter.newSplitAlgoInstance(@conf, arg.delete(SPLITALGO)) |
| splits = split_algo.split(JInteger.valueOf(num_regions)) |
| end |
| |
| # Done with splits; apply formerly-table_att parameters. |
| update_tdb_from_arg(tdb, arg) |
| |
| arg.each_key do |ignored_key| |
| puts(format('An argument ignored (unknown or overridden): %s', ignored_key)) |
| end |
| end |
| |
| # Fail if no column families defined |
| raise(ArgumentError, 'Table must have at least one column family') unless has_columns |
| |
| if splits.nil? |
| # Perform the create table call |
| @admin.createTable(tdb.build) |
| else |
| # Perform the create table call |
| @admin.createTable(tdb.build, splits) |
| end |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| #---------------------------------------------------------------------------------------------- |
| # Assign a region |
| def assign(region_name) |
| @admin.assign(region_name.to_java_bytes) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Unassign a region |
| # the force parameter is deprecated, if it is specified, will be ignored. |
| def unassign(region_name, force = nil) |
| @admin.unassign(region_name.to_java_bytes) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Move a region |
| def move(encoded_region_name, server = nil) |
| @admin.move(encoded_region_name.to_java_bytes, server ? server.to_java_bytes : nil) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Merge multiple regions |
| def merge_region(regions, force) |
| unless regions.is_a?(Array) |
| raise(ArgumentError, "Type of #{regions.inspect} is #{regions.class}, but expected Array") |
| end |
| region_array = Java::byte[][regions.length].new |
| i = 0 |
| while i < regions.length |
| unless regions[i].is_a?(String) |
| raise( |
| ArgumentError, |
| "Type of #{regions[i].inspect} is #{regions[i].class}, but expected String" |
| ) |
| end |
| region_array[i] = regions[i].to_java_bytes |
| i += 1 |
| end |
| org.apache.hadoop.hbase.util.FutureUtils.get( |
| @admin.mergeRegionsAsync( |
| region_array, |
| java.lang.Boolean.valueOf(force) |
| ) |
| ) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Returns table's structure description |
| def describe(table_name) |
| tableExists(table_name) |
| @admin.getDescriptor(TableName.valueOf(table_name)).to_s |
| end |
| |
| def get_column_families(table_name) |
| tableExists(table_name) |
| @admin.getDescriptor(TableName.valueOf(table_name)).getColumnFamilies |
| end |
| |
| def get_table_attributes(table_name) |
| tableExists(table_name) |
| td = @admin.getDescriptor TableName.valueOf(table_name) |
| # toStringTableAttributes is a public method, but it is defined on the private class |
| # ModifiableTableDescriptor, so we need reflection to access it in JDK 11+. |
| # TODO Maybe move this to a utility class in the future? |
| method = td.java_class.declared_method :toStringTableAttributes |
| method.accessible = true |
| method.invoke td |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Enable/disable snapshot auto-cleanup based on TTL expiration |
| # Returns previous snapshot auto-cleanup switch setting. |
| def snapshot_cleanup_switch(enable_disable) |
| @admin.snapshotCleanupSwitch( |
| java.lang.Boolean.valueOf(enable_disable), java.lang.Boolean.valueOf(false) |
| ) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Query the current state of the snapshot auto-cleanup based on TTL |
| # Returns the snapshot auto-cleanup state (true if enabled) |
| def snapshot_cleanup_enabled? |
| @admin.isSnapshotCleanupEnabled |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Truncates table (deletes all records by recreating the table) |
| def truncate(table_name_str) |
| puts "Truncating '#{table_name_str}' table (it may take a while):" |
| table_name = TableName.valueOf(table_name_str) |
| |
| if enabled?(table_name_str) |
| puts 'Disabling table...' |
| disable(table_name_str) |
| end |
| |
| puts 'Truncating table...' |
| @admin.truncateTable(table_name, false) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Truncates table while maintaining region boundaries |
| # (deletes all records by recreating the table) |
| def truncate_preserve(table_name_str) |
| puts "Truncating '#{table_name_str}' table (it may take a while):" |
| table_name = TableName.valueOf(table_name_str) |
| |
| if enabled?(table_name_str) |
| puts 'Disabling table...' |
| disable(table_name_str) |
| end |
| |
| puts 'Truncating table...' |
| @admin.truncateTable(table_name, true) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Check the status of alter command (number of regions reopened) |
| def alter_status(table_name) |
| # Table name should be a string |
| raise(ArgumentError, 'Table name must be of type String') unless table_name.is_a?(String) |
| |
| # Table should exist |
| raise(ArgumentError, "Can't find a table: #{table_name}") unless exists?(table_name) |
| |
| begin |
| cluster_metrics = @admin.getClusterMetrics |
| table_region_status = cluster_metrics |
| .getTableRegionStatesCount |
| .get(org.apache.hadoop.hbase.TableName.valueOf(table_name)) |
| if table_region_status.getTotalRegions != 0 |
| updated_regions = table_region_status.getTotalRegions - |
| table_region_status.getRegionsInTransition - |
| table_region_status.getClosedRegions |
| puts "#{updated_regions}/#{table_region_status.getTotalRegions} regions updated." |
| else |
| puts 'All regions updated.' |
| end |
| sleep 1 |
| end while !table_region_status.nil? && table_region_status.getRegionsInTransition != 0 |
| puts 'Done.' |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Use our internal logic to convert from "spec string" format to a coprocessor descriptor |
| # |
| # Provided for backwards shell compatibility |
| # |
| # @param [String] spec_str |
| # @return [ColumnDescriptor] |
| def coprocessor_descriptor_from_spec_str(spec_str) |
| method = TableDescriptorBuilder.java_class.declared_method_smart :toCoprocessorDescriptor |
| method.accessible = true |
| result = method.invoke(nil, spec_str).to_java |
| # unpack java's Optional to be more rubonic |
| return result.isPresent ? result.get : nil |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Use CoprocessorDescriptorBuilder to convert a Hash to CoprocessorDescriptor |
| # |
| # @param [Hash] spec column descriptor specification |
| # @return [ColumnDescriptor] |
| def coprocessor_descriptor_from_hash(spec) |
| classname = spec[CLASSNAME] |
| raise ArgumentError.new "CLASSNAME must be provided in spec" if classname.nil? |
| jar_path = spec[JAR_PATH] |
| priority = spec[PRIORITY] |
| properties = spec[PROPERTIES] |
| |
| builder = CoprocessorDescriptorBuilder.newBuilder classname |
| builder.setJarPath jar_path unless jar_path.nil? |
| builder.setPriority priority unless priority.nil? |
| properties&.each { |k, v| builder.setProperty(k, v.to_s) } |
| |
| builder.build |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Change table structure or table options |
| def alter(table_name_str, wait = true, *args) |
| # Table name should be a string |
| raise(ArgumentError, 'Table name must be of type String') unless |
| table_name_str.is_a?(String) |
| |
| # Table should exist |
| raise(ArgumentError, "Can't find a table: #{table_name_str}") unless exists?(table_name_str) |
| |
| # There should be at least one argument |
| raise(ArgumentError, 'There should be at least one argument but the table name') if args.empty? |
| |
| table_name = TableName.valueOf(table_name_str) |
| |
| # Get table descriptor |
| tdb = TableDescriptorBuilder.newBuilder(@admin.getDescriptor(table_name)) |
| hasTableUpdate = false |
| |
| # Process all args |
| args.each do |arg| |
| # Normalize args to support column name only alter specs |
| arg = { NAME => arg } if arg.is_a?(String) |
| |
| # Normalize args to support shortcut delete syntax |
| arg = { METHOD => 'delete', NAME => arg['delete'] } if arg['delete'] |
| |
| # There are 3 possible options. |
| # 1) Column family spec. Distinguished by having a NAME and no METHOD. |
| method = arg.delete(METHOD) |
| if method.nil? && arg.key?(NAME) |
| descriptor = cfd(arg, tdb) |
| column_name = descriptor.getNameAsString |
| |
| # If column already exist, then try to alter it. Create otherwise. |
| if tdb.build.hasColumnFamily(column_name.to_java_bytes) |
| tdb.modifyColumnFamily(descriptor) |
| else |
| tdb.setColumnFamily(descriptor) |
| end |
| hasTableUpdate = true |
| next |
| end |
| |
| # 2) Method other than table_att, with some args. |
| name = arg.delete(NAME) |
| if !method.nil? && method != 'table_att' |
| # Delete column family |
| if method == 'delete' |
| raise(ArgumentError, 'NAME parameter missing for delete method') unless name |
| tdb.removeColumnFamily(name.to_java_bytes) |
| hasTableUpdate = true |
| # Unset table attributes |
| elsif method == 'table_att_unset' |
| raise(ArgumentError, 'NAME parameter missing for table_att_unset method') unless name |
| if name.is_a?(Array) |
| name.each do |key| |
| if tdb.build.getValue(key).nil? |
| raise ArgumentError, "Could not find attribute: #{key}" |
| end |
| tdb.removeValue(key) |
| end |
| else |
| if tdb.build.getValue(name).nil? |
| raise ArgumentError, "Could not find attribute: #{name}" |
| end |
| tdb.removeValue(name) |
| end |
| hasTableUpdate = true |
| # Unset table configuration |
| elsif method == 'table_conf_unset' |
| raise(ArgumentError, 'NAME parameter missing for table_conf_unset method') unless name |
| if name.is_a?(Array) |
| name.each do |key| |
| if tdb.build.getValue(key).nil? |
| raise ArgumentError, "Could not find configuration: #{key}" |
| end |
| tdb.removeValue(key) |
| end |
| else |
| if tdb.build.getValue(name).nil? |
| raise ArgumentError, "Could not find configuration: #{name}" |
| end |
| tdb.removeValue(name) |
| end |
| hasTableUpdate = true |
| # Unknown method |
| else |
| raise ArgumentError, "Unknown method: #{method}" |
| end |
| |
| arg.each_key do |unknown_key| |
| puts(format('Unknown argument ignored: %s', unknown_key)) |
| end |
| |
| next |
| end |
| |
| # 3) Some args for the table, optionally with METHOD => table_att (deprecated) |
| update_tdb_from_arg(tdb, arg) |
| |
| # set a coprocessor attribute |
| valid_coproc_keys = [] |
| next unless arg.is_a?(Hash) |
| arg.each do |key, value| |
| k = String.new(key) # prepare to strip |
| k.strip! |
| |
| # Uses insensitive matching so we can accept lowercase 'coprocessor' for compatibility |
| next unless k =~ /#{COPROCESSOR}/i |
| if value.is_a? String |
| # Specifying a coprocessor by this "spec string" is here for backwards compatibility |
| v = String.new value |
| v.strip! |
| cp = coprocessor_descriptor_from_spec_str v |
| elsif value.is_a? Hash |
| cp = coprocessor_descriptor_from_hash value |
| else |
| raise ArgumentError.new 'coprocessor must be provided as a String or Hash' |
| end |
| tdb.setCoprocessor cp |
| valid_coproc_keys << key |
| end |
| |
| valid_coproc_keys.each do |key| |
| arg.delete(key) |
| end |
| |
| hasTableUpdate = true |
| |
| arg.each_key do |unknown_key| |
| puts(format('Unknown argument ignored: %s', unknown_key)) |
| end |
| |
| next |
| end |
| |
| # Bulk apply all table modifications. |
| if hasTableUpdate |
| future = @admin.modifyTableAsync(tdb.build) |
| |
| if wait == true |
| puts 'Updating all regions with the new schema...' |
| future.get |
| end |
| end |
| end |
| |
| def status(format, type) |
| cluster_metrics = @admin.getClusterMetrics |
| if format == 'detailed' |
| puts(format('version %s', cluster_metrics.getHBaseVersion)) |
| # Put regions in transition first because usually empty |
| puts(format('%d regionsInTransition', cluster_metrics.getRegionStatesInTransition.size)) |
| for v in cluster_metrics.getRegionStatesInTransition |
| puts(format(' %s', v)) |
| end |
| master = cluster_metrics.getMasterName |
| puts(format('active master: %s:%d %d', master.getHostname, master.getPort, |
| master.getStartcode)) |
| puts(format('%d backup masters', cluster_metrics.getBackupMasterNames.size)) |
| for server in cluster_metrics.getBackupMasterNames |
| puts(format(' %s:%d %d', server.getHostname, server.getPort, server.getStartcode)) |
| end |
| |
| master_coprocs = @admin.getMasterCoprocessorNames.toString |
| unless master_coprocs.nil? |
| puts(format('master coprocessors: %s', master_coprocs)) |
| end |
| puts(format('%d live servers', cluster_metrics.getLiveServerMetrics.size)) |
| for server in cluster_metrics.getLiveServerMetrics.keySet |
| puts(format(' %s:%d %d', server.getHostname, server.getPort, server.getStartcode)) |
| puts(format(' %s', cluster_metrics.getLiveServerMetrics.get(server).toString)) |
| for name, region in cluster_metrics.getLiveServerMetrics.get(server).getRegionMetrics |
| puts(format(' %s', region.getNameAsString.dump)) |
| puts(format(' %s', region.toString)) |
| end |
| end |
| puts(format('%d dead servers', cluster_metrics.getDeadServerNames.size)) |
| for server in cluster_metrics.getDeadServerNames |
| puts(format(' %s', server)) |
| end |
| elsif format == 'replication' |
| puts(format('version %<version>s', version: cluster_metrics.getHBaseVersion)) |
| puts(format('%<servers>d live servers', |
| servers: cluster_metrics.getLiveServerMetrics.size)) |
| cluster_metrics.getLiveServerMetrics.keySet.each do |server_name| |
| sl = cluster_metrics.getLiveServerMetrics.get(server_name) |
| r_sink_string = ' SINK:' |
| r_source_string = ' SOURCE:' |
| r_load_sink = sl.getReplicationLoadSink |
| next if r_load_sink.nil? |
| if r_load_sink.getTimestampsOfLastAppliedOp() == r_load_sink.getTimestampStarted() |
| # If we have applied no operations since we've started replication, |
| # assume that we're not acting as a sink and don't print the normal information |
| r_sink_string << " TimeStampStarted=" + r_load_sink.getTimestampStarted().to_s |
| r_sink_string << ", Waiting for OPs... " |
| else |
| r_sink_string << " TimeStampStarted=" + r_load_sink.getTimestampStarted().to_s |
| r_sink_string << ", AgeOfLastAppliedOp=" + r_load_sink.getAgeOfLastAppliedOp().to_s |
| r_sink_string << ", TimeStampsOfLastAppliedOp=" + |
| (java.util.Date.new(r_load_sink.getTimestampsOfLastAppliedOp())).toString() |
| end |
| |
| r_load_source_map = sl.getReplicationLoadSourceMap |
| build_source_string(r_load_source_map, r_source_string) |
| puts(format(' %<host>s:', host: server_name.getHostname)) |
| if type.casecmp('SOURCE').zero? |
| puts(format('%<source>s', source: r_source_string)) |
| elsif type.casecmp('SINK').zero? |
| puts(format('%<sink>s', sink: r_sink_string)) |
| else |
| puts(format('%<source>s', source: r_source_string)) |
| puts(format('%<sink>s', sink: r_sink_string)) |
| end |
| end |
| elsif format == 'simple' |
| load = 0 |
| regions = 0 |
| master = cluster_metrics.getMasterName |
| puts(format('active master: %s:%d %d', master.getHostname, master.getPort, |
| master.getStartcode)) |
| puts(format('%d backup masters', cluster_metrics.getBackupMasterNames.size)) |
| for server in cluster_metrics.getBackupMasterNames |
| puts(format(' %s:%d %d', server.getHostname, server.getPort, server.getStartcode)) |
| end |
| puts(format('%d live servers', cluster_metrics.getLiveServerMetrics.size)) |
| for server in cluster_metrics.getLiveServerMetrics.keySet |
| puts(format(' %s:%d %d', server.getHostname, server.getPort, server.getStartcode)) |
| puts(format(' %s', cluster_metrics.getLiveServerMetrics.get(server).toString)) |
| load += cluster_metrics.getLiveServerMetrics.get(server).getRequestCountPerSecond |
| regions += cluster_metrics.getLiveServerMetrics.get(server).getRegionMetrics.size |
| end |
| puts(format('%d dead servers', cluster_metrics.getDeadServerNames.size)) |
| for server in cluster_metrics.getDeadServerNames |
| puts(format(' %s', server)) |
| end |
| puts(format('Aggregate load: %d, regions: %d', load, regions)) |
| else |
| puts "1 active master, #{cluster_metrics.getBackupMasterNames.size} backup masters, |
| #{cluster_metrics.getLiveServerMetrics.size} servers, |
| #{cluster_metrics.getDeadServerNames.size} dead, |
| #{format('%.4f', cluster_metrics.getAverageLoad)} average load" |
| end |
| end |
| |
| def build_source_string(r_load_source_map, r_source_string) |
| r_load_source_map.each do |peer, sources| |
| r_source_string << ' PeerID=' + peer |
| sources.each do |source_load| |
| build_queue_title(source_load, r_source_string) |
| build_running_source_stats(source_load, r_source_string) |
| end |
| end |
| end |
| |
| def build_queue_title(source_load, r_source_string) |
| r_source_string << if source_load.isRecovered |
| "\n Recovered Queue: " |
| else |
| "\n Normal Queue: " |
| end |
| r_source_string << source_load.getQueueId |
| end |
| |
| def build_running_source_stats(source_load, r_source_string) |
| if source_load.isRunning |
| build_shipped_stats(source_load, r_source_string) |
| build_load_general_stats(source_load, r_source_string) |
| r_source_string << ', Replication Lag=' + |
| source_load.getReplicationLag.to_s |
| else |
| r_source_string << "\n " |
| r_source_string << 'No Reader/Shipper threads runnning yet.' |
| end |
| end |
| |
| def build_shipped_stats(source_load, r_source_string) |
| r_source_string << if source_load.getTimestampOfLastShippedOp.zero? |
| "\n " \ |
| 'No Ops shipped since last restart' |
| else |
| "\n AgeOfLastShippedOp=" + |
| source_load.getAgeOfLastShippedOp.to_s + |
| ', TimeStampOfLastShippedOp=' + |
| java.util.Date.new(source_load |
| .getTimestampOfLastShippedOp).toString |
| end |
| end |
| |
| def build_load_general_stats(source_load, r_source_string) |
| r_source_string << ', SizeOfLogQueue=' + |
| source_load.getSizeOfLogQueue.to_s |
| r_source_string << ', EditsReadFromLogQueue=' + |
| source_load.getEditsRead.to_s |
| r_source_string << ', OpsShippedToTarget=' + |
| source_load.getOPsShipped.to_s |
| build_edits_for_source(source_load, r_source_string) |
| end |
| |
| def build_edits_for_source(source_load, r_source_string) |
| if source_load.hasEditsSinceRestart |
| r_source_string << ', TimeStampOfNextToReplicate=' + |
| java.util.Date.new(source_load |
| .getTimeStampOfNextToReplicate).toString |
| else |
| r_source_string << ', No edits for this source' |
| r_source_string << ' since it started' |
| end |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # |
| # Helper methods |
| # |
| |
| # Does table exist? |
| def exists?(table_name) |
| @admin.tableExists(TableName.valueOf(table_name)) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Is table enabled |
| def enabled?(table_name) |
| @admin.isTableEnabled(TableName.valueOf(table_name)) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Return a new ColumnFamilyDescriptor made of passed args |
| def cfd(arg, tdb) |
| # String arg, single parameter constructor |
| return ColumnFamilyDescriptorBuilder.of(arg) if arg.is_a?(String) |
| |
| raise(ArgumentError, "Column family #{arg} must have a name") unless name = arg.delete(NAME) |
| |
| descriptor = tdb.build.getColumnFamily(name.to_java_bytes) |
| unless descriptor.nil? |
| cfdb = ColumnFamilyDescriptorBuilder.newBuilder(descriptor) |
| end |
| # create it if it's a new family |
| cfdb ||= ColumnFamilyDescriptorBuilder.newBuilder(name.to_java_bytes) |
| |
| cfdb.setBlockCacheEnabled(JBoolean.valueOf(arg.delete(ColumnFamilyDescriptorBuilder::BLOCKCACHE))) if arg.include?(ColumnFamilyDescriptorBuilder::BLOCKCACHE) |
| cfdb.setScope(JInteger.valueOf(arg.delete(ColumnFamilyDescriptorBuilder::REPLICATION_SCOPE))) if arg.include?(ColumnFamilyDescriptorBuilder::REPLICATION_SCOPE) |
| cfdb.setCacheDataOnWrite(JBoolean.valueOf(arg.delete(ColumnFamilyDescriptorBuilder::CACHE_DATA_ON_WRITE))) if arg.include?(ColumnFamilyDescriptorBuilder::CACHE_DATA_ON_WRITE) |
| cfdb.setCacheIndexesOnWrite(JBoolean.valueOf(arg.delete(ColumnFamilyDescriptorBuilder::CACHE_INDEX_ON_WRITE))) if arg.include?(ColumnFamilyDescriptorBuilder::CACHE_INDEX_ON_WRITE) |
| cfdb.setCacheBloomsOnWrite(JBoolean.valueOf(arg.delete(ColumnFamilyDescriptorBuilder::CACHE_BLOOMS_ON_WRITE))) if arg.include?(ColumnFamilyDescriptorBuilder::CACHE_BLOOMS_ON_WRITE) |
| cfdb.setEvictBlocksOnClose(JBoolean.valueOf(arg.delete(ColumnFamilyDescriptorBuilder::EVICT_BLOCKS_ON_CLOSE))) if arg.include?(ColumnFamilyDescriptorBuilder::EVICT_BLOCKS_ON_CLOSE) |
| cfdb.setInMemory(JBoolean.valueOf(arg.delete(ColumnFamilyDescriptorBuilder::IN_MEMORY))) if arg.include?(ColumnFamilyDescriptorBuilder::IN_MEMORY) |
| if arg.include?(ColumnFamilyDescriptorBuilder::IN_MEMORY_COMPACTION) |
| cfdb.setInMemoryCompaction( |
| org.apache.hadoop.hbase.MemoryCompactionPolicy.valueOf(arg.delete(ColumnFamilyDescriptorBuilder::IN_MEMORY_COMPACTION)) |
| ) |
| end |
| cfdb.setTimeToLive(arg.delete(ColumnFamilyDescriptorBuilder::TTL)) if arg.include?(ColumnFamilyDescriptorBuilder::TTL) |
| cfdb.setDataBlockEncoding(org.apache.hadoop.hbase.io.encoding.DataBlockEncoding.valueOf(arg.delete(ColumnFamilyDescriptorBuilder::DATA_BLOCK_ENCODING))) if arg.include?(ColumnFamilyDescriptorBuilder::DATA_BLOCK_ENCODING) |
| cfdb.setBlocksize(arg.delete(ColumnFamilyDescriptorBuilder::BLOCKSIZE)) if arg.include?(ColumnFamilyDescriptorBuilder::BLOCKSIZE) |
| cfdb.setMaxVersions(JInteger.valueOf(arg.delete(HConstants::VERSIONS))) if arg.include?(HConstants::VERSIONS) |
| cfdb.setMinVersions(JInteger.valueOf(arg.delete(ColumnFamilyDescriptorBuilder::MIN_VERSIONS))) if arg.include?(ColumnFamilyDescriptorBuilder::MIN_VERSIONS) |
| cfdb.setKeepDeletedCells(org.apache.hadoop.hbase.KeepDeletedCells.valueOf(arg.delete(ColumnFamilyDescriptorBuilder::KEEP_DELETED_CELLS).to_s.upcase)) if arg.include?(ColumnFamilyDescriptorBuilder::KEEP_DELETED_CELLS) |
| cfdb.setCompressTags(JBoolean.valueOf(arg.delete(ColumnFamilyDescriptorBuilder::COMPRESS_TAGS))) if arg.include?(ColumnFamilyDescriptorBuilder::COMPRESS_TAGS) |
| cfdb.setPrefetchBlocksOnOpen(JBoolean.valueOf(arg.delete(ColumnFamilyDescriptorBuilder::PREFETCH_BLOCKS_ON_OPEN))) if arg.include?(ColumnFamilyDescriptorBuilder::PREFETCH_BLOCKS_ON_OPEN) |
| cfdb.setMobEnabled(JBoolean.valueOf(arg.delete(ColumnFamilyDescriptorBuilder::IS_MOB))) if arg.include?(ColumnFamilyDescriptorBuilder::IS_MOB) |
| cfdb.setMobThreshold(JLong.valueOf(arg.delete(ColumnFamilyDescriptorBuilder::MOB_THRESHOLD))) if arg.include?(ColumnFamilyDescriptorBuilder::MOB_THRESHOLD) |
| cfdb.setNewVersionBehavior(JBoolean.valueOf(arg.delete(ColumnFamilyDescriptorBuilder::NEW_VERSION_BEHAVIOR))) if arg.include?(ColumnFamilyDescriptorBuilder::NEW_VERSION_BEHAVIOR) |
| if arg.include?(ColumnFamilyDescriptorBuilder::BLOOMFILTER) |
| bloomtype = arg.delete(ColumnFamilyDescriptorBuilder::BLOOMFILTER).upcase.to_sym |
| if org.apache.hadoop.hbase.regionserver.BloomType.constants.include?(bloomtype) |
| cfdb.setBloomFilterType(org.apache.hadoop.hbase.regionserver.BloomType.valueOf(bloomtype)) |
| else |
| raise(ArgumentError, "BloomFilter type #{bloomtype} is not supported. Use one of " + org.apache.hadoop.hbase.regionserver.StoreFile::BloomType.constants.join(' ')) |
| end |
| end |
| if arg.include?(ColumnFamilyDescriptorBuilder::COMPRESSION) |
| compression = arg.delete(ColumnFamilyDescriptorBuilder::COMPRESSION).upcase.to_sym |
| if org.apache.hadoop.hbase.io.compress.Compression::Algorithm.constants.include?(compression) |
| cfdb.setCompressionType(org.apache.hadoop.hbase.io.compress.Compression::Algorithm.valueOf(compression)) |
| else |
| raise(ArgumentError, "Compression #{compression} is not supported. Use one of " + org.apache.hadoop.hbase.io.compress.Compression::Algorithm.constants.join(' ')) |
| end |
| end |
| if arg.include?(ColumnFamilyDescriptorBuilder::ENCRYPTION) |
| algorithm = arg.delete(ColumnFamilyDescriptorBuilder::ENCRYPTION).upcase |
| cfdb.setEncryptionType(algorithm) |
| if arg.include?(ColumnFamilyDescriptorBuilder::ENCRYPTION_KEY) |
| key = org.apache.hadoop.hbase.io.crypto.Encryption.generateSecretKey( |
| @conf, algorithm, arg.delete(ColumnFamilyDescriptorBuilder::ENCRYPTION_KEY) |
| ) |
| cfdb.setEncryptionKey(org.apache.hadoop.hbase.security.EncryptionUtil.wrapKey(@conf, key, |
| algorithm)) |
| end |
| end |
| if arg.include?(ColumnFamilyDescriptorBuilder::COMPRESSION_COMPACT) |
| compression = arg.delete(ColumnFamilyDescriptorBuilder::COMPRESSION_COMPACT).upcase.to_sym |
| if org.apache.hadoop.hbase.io.compress.Compression::Algorithm.constants.include?(compression) |
| cfdb.setCompactionCompressionType(org.apache.hadoop.hbase.io.compress.Compression::Algorithm.valueOf(compression)) |
| else |
| raise(ArgumentError, "Compression #{compression} is not supported. Use one of " + org.apache.hadoop.hbase.io.compress.Compression::Algorithm.constants.join(' ')) |
| end |
| end |
| if arg.include?(ColumnFamilyDescriptorBuilder::COMPRESSION_COMPACT_MAJOR) |
| compression = arg.delete(ColumnFamilyDescriptorBuilder::COMPRESSION_COMPACT_MAJOR).upcase.to_sym |
| if org.apache.hadoop.hbase.io.compress.Compression::Algorithm.constants.include?(compression) |
| cfdb.setMajorCompactionCompressionType(org.apache.hadoop.hbase.io.compress.Compression::Algorithm.valueOf(compression)) |
| else |
| raise(ArgumentError, "Compression #{compression} is not supported. Use one of " + org.apache.hadoop.hbase.io.compress.Compression::Algorithm.constants.join(' ')) |
| end |
| end |
| if arg.include?(ColumnFamilyDescriptorBuilder::COMPRESSION_COMPACT_MINOR) |
| compression = arg.delete(ColumnFamilyDescriptorBuilder::COMPRESSION_COMPACT_MINOR).upcase.to_sym |
| if org.apache.hadoop.hbase.io.compress.Compression::Algorithm.constants.include?(compression) |
| cfdb.setMinorCompactionCompressionType(org.apache.hadoop.hbase.io.compress.Compression::Algorithm.valueOf(compression)) |
| else |
| raise(ArgumentError, "Compression #{compression} is not supported. Use one of " + org.apache.hadoop.hbase.io.compress.Compression::Algorithm.constants.join(' ')) |
| end |
| end |
| if arg.include?(ColumnFamilyDescriptorBuilder::STORAGE_POLICY) |
| storage_policy = arg.delete(ColumnFamilyDescriptorBuilder::STORAGE_POLICY).upcase |
| cfdb.setStoragePolicy(storage_policy) |
| end |
| if arg.include?(ColumnFamilyDescriptorBuilder::MOB_COMPACT_PARTITION_POLICY) |
| mob_partition_policy = arg.delete(ColumnFamilyDescriptorBuilder::MOB_COMPACT_PARTITION_POLICY).upcase.to_sym |
| if MobCompactPartitionPolicy.constants.include?(mob_partition_policy) |
| cfdb.setMobCompactPartitionPolicy(MobCompactPartitionPolicy.valueOf(mob_partition_policy)) |
| else |
| raise(ArgumentError, "MOB_COMPACT_PARTITION_POLICY #{mob_partition_policy} is not supported. Use one of " + MobCompactPartitionPolicy.constants.join(' ')) |
| end |
| end |
| |
| set_user_metadata(cfdb, arg.delete(METADATA)) if arg[METADATA] |
| set_descriptor_config(cfdb, arg.delete(CONFIGURATION)) if arg[CONFIGURATION] |
| if arg.include?(ColumnFamilyDescriptorBuilder::DFS_REPLICATION) |
| cfdb.setDFSReplication(JInteger.valueOf(arg.delete(ColumnFamilyDescriptorBuilder::DFS_REPLICATION))) |
| end |
| |
| arg.each_key do |unknown_key| |
| puts(format('Unknown argument ignored for column family %s: %s', name, unknown_key)) |
| end |
| |
| cfdb.build |
| end |
| |
| # Apply user metadata to table/column descriptor |
| def set_user_metadata(descriptor, metadata) |
| raise(ArgumentError, "#{METADATA} must be a Hash type") unless metadata.is_a?(Hash) |
| for k, v in metadata |
| v = v.to_s unless v.nil? |
| descriptor.setValue(k, v) |
| end |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Take a snapshot of specified table |
| def snapshot(table, snapshot_name, *args) |
| # Table name should be a string |
| raise(ArgumentError, 'Table name must be of type String') unless table.is_a?(String) |
| |
| # Snapshot name should be a string |
| raise(ArgumentError, 'Snapshot name must be of type String') unless |
| snapshot_name.is_a?(String) |
| |
| table_name = TableName.valueOf(table) |
| if args.empty? |
| @admin.snapshot(snapshot_name, table_name) |
| else |
| args.each do |arg| |
| ttl = arg[TTL] |
| ttl = ttl ? ttl.to_java(:long) : -1 |
| snapshot_props = java.util.HashMap.new |
| snapshot_props.put("TTL", ttl) |
| max_filesize = arg[MAX_FILESIZE] |
| max_filesize = max_filesize ? max_filesize.to_java(:long) : -1 |
| snapshot_props.put("MAX_FILESIZE", max_filesize) |
| if arg[SKIP_FLUSH] == true |
| @admin.snapshot(snapshot_name, table_name, |
| org.apache.hadoop.hbase.client.SnapshotType::SKIPFLUSH, snapshot_props) |
| else |
| @admin.snapshot(snapshot_name, table_name, snapshot_props) |
| end |
| end |
| end |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Restore specified snapshot |
| def restore_snapshot(snapshot_name, restore_acl = false) |
| conf = @connection.getConfiguration |
| take_fail_safe_snapshot = conf.getBoolean('hbase.snapshot.restore.take.failsafe.snapshot', false) |
| @admin.restoreSnapshot(snapshot_name, take_fail_safe_snapshot, restore_acl) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Create a new table by cloning the snapshot content |
| def clone_snapshot(snapshot_name, table, restore_acl = false) |
| @admin.cloneSnapshot(snapshot_name, TableName.valueOf(table), restore_acl) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Delete specified snapshot |
| def delete_snapshot(snapshot_name) |
| @admin.deleteSnapshot(snapshot_name) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Deletes the snapshots matching the given regex |
| def delete_all_snapshot(regex) |
| @admin.deleteSnapshots(Pattern.compile(regex)).to_a |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Deletes the table snapshots matching the given regex |
| def delete_table_snapshots(tableNameRegex, snapshotNameRegex = '.*') |
| @admin.deleteTableSnapshots(Pattern.compile(tableNameRegex), |
| Pattern.compile(snapshotNameRegex)).to_a |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Returns a list of snapshots |
| def list_snapshot(regex = '.*') |
| @admin.listSnapshots(Pattern.compile(regex)).to_a |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Returns a list of table snapshots |
| def list_table_snapshots(tableNameRegex, snapshotNameRegex = '.*') |
| @admin.listTableSnapshots(Pattern.compile(tableNameRegex), |
| Pattern.compile(snapshotNameRegex)).to_a |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Returns the whole ClusterMetrics containing details: |
| # |
| # hbase version |
| # cluster id |
| # primary/backup master(s) |
| # master's coprocessors |
| # live/dead regionservers |
| # balancer |
| # regions in transition |
| def getClusterMetrics |
| @admin.getClusterMetrics |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Returns a list of regionservers |
| def getRegionServers |
| @admin.getClusterMetrics.getLiveServerMetrics.keySet.map { |server_name| server_name } |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Returns servername corresponding to passed server_name_string |
| def getServerName(server_name_string) |
| regionservers = getRegionServers |
| |
| if ServerName.isFullServerName(server_name_string) |
| return ServerName.valueOf(server_name_string) |
| else |
| name_list = server_name_string.split(',') |
| |
| regionservers.each do|sn| |
| if name_list[0] == sn.hostname && (name_list[1].nil? ? true : (name_list[1] == sn.port.to_s)) |
| return sn |
| end |
| end |
| end |
| |
| return nil |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Returns a list of servernames |
| def getServerNames(servers, should_return_all_if_servers_empty) |
| regionservers = getRegionServers |
| servernames = [] |
| |
| if servers.empty? |
| # if no servers were specified as arguments, get a list of all servers |
| if should_return_all_if_servers_empty |
| servernames = regionservers |
| end |
| else |
| # Strings replace with ServerName objects in servers array |
| i = 0 |
| while i < servers.length |
| server = servers[i] |
| |
| if ServerName.isFullServerName(server) |
| servernames.push(ServerName.valueOf(server)) |
| else |
| name_list = server.split(',') |
| j = 0 |
| while j < regionservers.length |
| sn = regionservers[j] |
| if name_list[0] == sn.hostname && (name_list[1].nil? ? true : (name_list[1] == sn.port.to_s)) |
| servernames.push(sn) |
| end |
| j += 1 |
| end |
| end |
| i += 1 |
| end |
| end |
| |
| servernames |
| end |
| |
| # Apply config specific to a table/column to its descriptor |
| def set_descriptor_config(descriptor, config) |
| raise(ArgumentError, "#{CONFIGURATION} must be a Hash type") unless config.is_a?(Hash) |
| for k, v in config |
| v = v.to_s unless v.nil? |
| descriptor.setValue(k, v) |
| end |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Updates the configuration of one regionserver. |
| def update_config(serverName) |
| @admin.updateConfiguration(ServerName.valueOf(serverName)) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Updates the configuration of all the regionservers. |
| def update_all_config |
| @admin.updateConfiguration |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Returns namespace's structure description |
| def describe_namespace(namespace_name) |
| namespace = @admin.getNamespaceDescriptor(namespace_name) |
| |
| return namespace.to_s unless namespace.nil? |
| |
| raise(ArgumentError, "Failed to find namespace named #{namespace_name}") |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Returns a list of namespaces in hbase |
| def list_namespace(regex = '.*') |
| pattern = java.util.regex.Pattern.compile(regex) |
| list = @admin.listNamespaces |
| list.select { |s| pattern.match(s) } |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Returns a list of tables in namespace |
| def list_namespace_tables(namespace_name) |
| unless namespace_name.nil? |
| return @admin.listTableNamesByNamespace(namespace_name).map(&:getQualifierAsString) |
| end |
| |
| raise(ArgumentError, "Failed to find namespace named #{namespace_name}") |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Creates a namespace |
| def create_namespace(namespace_name, *args) |
| # Fail if table name is not a string |
| raise(ArgumentError, 'Namespace name must be of type String') unless namespace_name.is_a?(String) |
| |
| # Flatten params array |
| args = args.flatten.compact |
| |
| # Start defining the table |
| nsb = org.apache.hadoop.hbase.NamespaceDescriptor.create(namespace_name) |
| args.each do |arg| |
| unless arg.is_a?(Hash) |
| raise(ArgumentError, "#{arg.class} of #{arg.inspect} is not of Hash or String type") |
| end |
| for k, v in arg |
| v = v.to_s unless v.nil? |
| nsb.addConfiguration(k, v) |
| end |
| end |
| @admin.createNamespace(nsb.build) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # modify a namespace |
| def alter_namespace(namespace_name, *args) |
| # Fail if namespace name is not a string |
| raise(ArgumentError, 'Namespace name must be of type String') unless namespace_name.is_a?(String) |
| |
| nsd = @admin.getNamespaceDescriptor(namespace_name) |
| |
| raise(ArgumentError, 'Namespace does not exist') unless nsd |
| nsb = org.apache.hadoop.hbase.NamespaceDescriptor.create(nsd) |
| |
| # Flatten params array |
| args = args.flatten.compact |
| |
| # Start defining the table |
| args.each do |arg| |
| unless arg.is_a?(Hash) |
| raise(ArgumentError, "#{arg.class} of #{arg.inspect} is not of Hash type") |
| end |
| method = arg[METHOD] |
| if method == 'unset' |
| nsb.removeConfiguration(arg[NAME]) |
| elsif method == 'set' |
| arg.delete(METHOD) |
| for k, v in arg |
| v = v.to_s unless v.nil? |
| |
| nsb.addConfiguration(k, v) |
| end |
| else |
| raise(ArgumentError, "Unknown method #{method}") |
| end |
| end |
| @admin.modifyNamespace(nsb.build) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Get namespace's rsgroup |
| def get_namespace_rsgroup(namespace_name) |
| # Fail if namespace name is not a string |
| raise(ArgumentError, 'Namespace name must be of type String') unless namespace_name.is_a?(String) |
| nsd = @admin.getNamespaceDescriptor(namespace_name) |
| raise(ArgumentError, 'Namespace does not exist') unless nsd |
| nsd.getConfigurationValue("hbase.rsgroup.name") |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Drops a table |
| def drop_namespace(namespace_name) |
| @admin.deleteNamespace(namespace_name) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Get security capabilities |
| def get_security_capabilities |
| @admin.getSecurityCapabilities |
| end |
| |
| # List all procedures |
| def list_procedures |
| @admin.getProcedures |
| end |
| |
| # List all locks |
| def list_locks |
| @admin.getLocks |
| end |
| |
| # Parse arguments and update TableDescriptorBuilder accordingly |
| # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity |
| def update_tdb_from_arg(tdb, arg) |
| tdb.setMaxFileSize(arg.delete(TableDescriptorBuilder::MAX_FILESIZE)) if arg.include?(TableDescriptorBuilder::MAX_FILESIZE) |
| tdb.setReadOnly(JBoolean.valueOf(arg.delete(TableDescriptorBuilder::READONLY))) if arg.include?(TableDescriptorBuilder::READONLY) |
| tdb.setCompactionEnabled(JBoolean.valueOf(arg.delete(TableDescriptorBuilder::COMPACTION_ENABLED))) if arg.include?(TableDescriptorBuilder::COMPACTION_ENABLED) |
| tdb.setSplitEnabled(JBoolean.valueOf(arg.delete(TableDescriptorBuilder::SPLIT_ENABLED))) if arg.include?(TableDescriptorBuilder::SPLIT_ENABLED) |
| tdb.setMergeEnabled(JBoolean.valueOf(arg.delete(TableDescriptorBuilder::MERGE_ENABLED))) if arg.include?(TableDescriptorBuilder::MERGE_ENABLED) |
| tdb.setNormalizationEnabled(JBoolean.valueOf(arg.delete(TableDescriptorBuilder::NORMALIZATION_ENABLED))) if arg.include?(TableDescriptorBuilder::NORMALIZATION_ENABLED) |
| tdb.setNormalizerTargetRegionCount(JInteger.valueOf(arg.delete(TableDescriptorBuilder::NORMALIZER_TARGET_REGION_COUNT))) if arg.include?(TableDescriptorBuilder::NORMALIZER_TARGET_REGION_COUNT) |
| tdb.setNormalizerTargetRegionSize(JLong.valueOf(arg.delete(TableDescriptorBuilder::NORMALIZER_TARGET_REGION_SIZE))) if arg.include?(TableDescriptorBuilder::NORMALIZER_TARGET_REGION_SIZE) |
| tdb.setMemStoreFlushSize(arg.delete(TableDescriptorBuilder::MEMSTORE_FLUSHSIZE)) if arg.include?(TableDescriptorBuilder::MEMSTORE_FLUSHSIZE) |
| tdb.setDurability(org.apache.hadoop.hbase.client.Durability.valueOf(arg.delete(TableDescriptorBuilder::DURABILITY))) if arg.include?(TableDescriptorBuilder::DURABILITY) |
| tdb.setPriority(JInteger.valueOf(arg.delete(TableDescriptorBuilder::PRIORITY))) if arg.include?(TableDescriptorBuilder::PRIORITY) |
| tdb.setFlushPolicyClassName(arg.delete(TableDescriptorBuilder::FLUSH_POLICY)) if arg.include?(TableDescriptorBuilder::FLUSH_POLICY) |
| tdb.setRegionMemStoreReplication(JBoolean.valueOf(arg.delete(TableDescriptorBuilder::REGION_MEMSTORE_REPLICATION))) if arg.include?(TableDescriptorBuilder::REGION_MEMSTORE_REPLICATION) |
| tdb.setRegionSplitPolicyClassName(arg.delete(TableDescriptorBuilder::SPLIT_POLICY)) if arg.include?(TableDescriptorBuilder::SPLIT_POLICY) |
| tdb.setRegionReplication(JInteger.valueOf(arg.delete(TableDescriptorBuilder::REGION_REPLICATION))) if arg.include?(TableDescriptorBuilder::REGION_REPLICATION) |
| set_user_metadata(tdb, arg.delete(METADATA)) if arg[METADATA] |
| set_descriptor_config(tdb, arg.delete(CONFIGURATION)) if arg[CONFIGURATION] |
| end |
| # rubocop:enable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity |
| |
| #---------------------------------------------------------------------------------------------- |
| # clear compaction queues |
| def clear_compaction_queues(server_name, queue_name = nil) |
| names = %w[long short] |
| queues = java.util.HashSet.new |
| if queue_name.nil? |
| queues.add('long') |
| queues.add('short') |
| elsif queue_name.is_a?(String) |
| queues.add(queue_name) |
| unless names.include?(queue_name) |
| raise(ArgumentError, "Unknown queue name #{queue_name}") |
| end |
| elsif queue_name.is_a?(Array) |
| queue_name.each do |s| |
| queues.add(s) |
| unless names.include?(s) |
| raise(ArgumentError, "Unknown queue name #{s}") |
| end |
| end |
| else |
| raise(ArgumentError, "Unknown queue name #{queue_name}") |
| end |
| @admin.clearCompactionQueues(ServerName.valueOf(server_name), queues) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # clear dead region servers |
| def list_deadservers |
| @admin.listDeadServers.to_a |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # clear dead region servers |
| def clear_deadservers(dead_servers) |
| # Flatten params array |
| dead_servers = dead_servers.flatten.compact |
| if dead_servers.empty? |
| servers = list_deadservers |
| else |
| servers = java.util.ArrayList.new |
| dead_servers.each do |s| |
| servers.add(ServerName.valueOf(s)) |
| end |
| end |
| @admin.clearDeadServers(servers).to_a |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # List live region servers |
| def list_liveservers |
| @admin.getClusterMetrics.getLiveServerMetrics.keySet.to_a |
| end |
| |
| #--------------------------------------------------------------------------- |
| # create a new table by cloning the existent table schema. |
| def clone_table_schema(table_name, new_table_name, preserve_splits = true) |
| @admin.cloneTableSchema(TableName.valueOf(table_name), |
| TableName.valueOf(new_table_name), |
| preserve_splits) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # List decommissioned RegionServers |
| def list_decommissioned_regionservers |
| @admin.listDecommissionedRegionServers |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Retrieve SlowLog Responses from RegionServers |
| def get_slowlog_responses(server_names, args, is_large_log = false) |
| unless server_names.is_a?(Array) || server_names.is_a?(String) |
| raise(ArgumentError, |
| "#{server_names.class} of #{server_names.inspect} is not of Array/String type") |
| end |
| if server_names == '*' |
| server_names = getServerNames([], true) |
| else |
| server_names_list = to_server_names(server_names) |
| server_names = getServerNames(server_names_list, false) |
| end |
| filter_params = get_filter_params(args) |
| if args.key? 'LIMIT' |
| limit = args['LIMIT'] |
| else |
| limit = 10 |
| end |
| if is_large_log |
| log_type = 'LARGE_LOG' |
| else |
| log_type = 'SLOW_LOG' |
| end |
| log_dest = org.apache.hadoop.hbase.client.ServerType::REGION_SERVER |
| server_names_set = java.util.HashSet.new(server_names) |
| slow_log_responses = @admin.getLogEntries(server_names_set, log_type, log_dest, limit, |
| filter_params) |
| slow_log_responses_arr = [] |
| slow_log_responses.each { |slow_log_response| |
| slow_log_responses_arr << slow_log_response.toJsonPrettyPrint |
| } |
| slow_log_responses_arr |
| end |
| |
| def get_filter_params(args) |
| filter_params = java.util.HashMap.new |
| if args.key? 'REGION_NAME' |
| region_name = args['REGION_NAME'] |
| filter_params.put('regionName', region_name) |
| end |
| if args.key? 'TABLE_NAME' |
| table_name = args['TABLE_NAME'] |
| filter_params.put('tableName', table_name) |
| end |
| if args.key? 'CLIENT_IP' |
| client_address = args['CLIENT_IP'] |
| filter_params.put('clientAddress', client_address) |
| end |
| if args.key? 'USER' |
| user = args['USER'] |
| filter_params.put('userName', user) |
| end |
| if args.key? 'FILTER_BY_OP' |
| filter_by_op = args['FILTER_BY_OP'] |
| if filter_by_op != 'OR' && filter_by_op != 'AND' |
| raise(ArgumentError, "FILTER_BY_OP should be either OR / AND") |
| end |
| if filter_by_op == 'AND' |
| filter_params.put('filterByOperator', 'AND') |
| end |
| end |
| filter_params |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Clears SlowLog Responses from RegionServers |
| def clear_slowlog_responses(server_names) |
| unless server_names.nil? || server_names.is_a?(Array) || server_names.is_a?(String) |
| raise(ArgumentError, |
| "#{server_names.class} of #{server_names.inspect} is not of correct type") |
| end |
| if server_names.nil? |
| server_names = getServerNames([], true) |
| else |
| server_names_list = to_server_names(server_names) |
| server_names = getServerNames(server_names_list, false) |
| end |
| clear_log_responses = @admin.clearSlowLogResponses(java.util.HashSet.new(server_names)) |
| clear_log_success_count = 0 |
| clear_log_responses.each do |response| |
| if response |
| clear_log_success_count += 1 |
| end |
| end |
| puts 'Cleared Slowlog responses from ' \ |
| "#{clear_log_success_count}/#{clear_log_responses.size} RegionServers" |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Decommission a list of region servers, optionally offload corresponding regions |
| def decommission_regionservers(host_or_servers, should_offload) |
| # Fail if host_or_servers is neither a string nor an array |
| unless host_or_servers.is_a?(Array) || host_or_servers.is_a?(String) |
| raise(ArgumentError, |
| "#{host_or_servers.class} of #{host_or_servers.inspect} is not of Array/String type") |
| end |
| |
| # Fail if should_offload is neither a TrueClass/FalseClass nor a string |
| unless (!!should_offload == should_offload) || should_offload.is_a?(String) |
| raise(ArgumentError, "#{should_offload} is not a boolean value") |
| end |
| |
| # If a string is passed, convert it to an array |
| _host_or_servers = host_or_servers.is_a?(Array) ? |
| host_or_servers : |
| java.util.Arrays.asList(host_or_servers) |
| |
| # Retrieve the server names corresponding to passed _host_or_servers list |
| server_names = getServerNames(_host_or_servers, false) |
| |
| # Fail, if we can not find any server(s) corresponding to the passed host_or_servers |
| if server_names.empty? |
| raise(ArgumentError, |
| "Could not find any server(s) with specified name(s): #{host_or_servers}") |
| end |
| |
| @admin.decommissionRegionServers(server_names, |
| java.lang.Boolean.valueOf(should_offload)) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Recommission a region server, optionally load a list of passed regions |
| def recommission_regionserver(server_name_string, encoded_region_names) |
| # Fail if server_name_string is not a string |
| unless server_name_string.is_a?(String) |
| raise(ArgumentError, |
| "#{server_name_string.class} of #{server_name_string.inspect} is not of String type") |
| end |
| |
| # Fail if encoded_region_names is not an array |
| unless encoded_region_names.is_a?(Array) |
| raise(ArgumentError, |
| "#{encoded_region_names.class} of #{encoded_region_names.inspect} is not of Array type") |
| end |
| |
| # Convert encoded_region_names from string to bytes (element-wise) |
| region_names_in_bytes = encoded_region_names |
| .map {|region_name| region_name.to_java_bytes} |
| .compact |
| |
| # Retrieve the server name corresponding to the passed server_name_string |
| server_name = getServerName(server_name_string) |
| |
| # Fail if we can not find a server corresponding to the passed server_name_string |
| if server_name.nil? |
| raise(ArgumentError, |
| "Could not find any server with name #{server_name_string}") |
| end |
| |
| @admin.recommissionRegionServer(server_name, region_names_in_bytes) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Retrieve latest balancer decisions made by LoadBalancers |
| def get_balancer_decisions(args) |
| if args.key? 'LIMIT' |
| limit = args['LIMIT'] |
| else |
| limit = 250 |
| end |
| log_type = 'BALANCER_DECISION' |
| log_dest = org.apache.hadoop.hbase.client.ServerType::MASTER |
| balancer_decisions_responses = @admin.getLogEntries(nil, log_type, log_dest, limit, nil) |
| balancer_decisions_resp_arr = [] |
| balancer_decisions_responses.each { |balancer_dec_resp| |
| balancer_decisions_resp_arr << balancer_dec_resp.toJsonPrettyPrint |
| } |
| balancer_decisions_resp_arr |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Retrieve latest balancer rejections made by LoadBalancers |
| def get_balancer_rejections(args) |
| if args.key? 'LIMIT' |
| limit = args['LIMIT'] |
| else |
| limit = 250 |
| end |
| |
| log_type = 'BALANCER_REJECTION' |
| log_dest = org.apache.hadoop.hbase.client.ServerType::MASTER |
| balancer_rejections_responses = @admin.getLogEntries(nil, log_type, log_dest, limit, nil) |
| balancer_rejections_resp_arr = [] |
| balancer_rejections_responses.each { |balancer_dec_resp| |
| balancer_rejections_resp_arr << balancer_dec_resp.toJsonPrettyPrint |
| } |
| balancer_rejections_resp_arr |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Stop the active Master |
| def stop_master |
| @admin.stopMaster |
| end |
| |
| # Stop the given RegionServer |
| def stop_regionserver(hostport) |
| @admin.stopRegionServer(hostport) |
| end |
| |
| #---------------------------------------------------------------------------------------------- |
| # Get list of server names |
| def to_server_names(server_names) |
| if server_names.is_a?(Array) |
| server_names |
| else |
| java.util.Arrays.asList(server_names) |
| end |
| end |
| end |
| # rubocop:enable Metrics/ClassLength |
| end |
