management/python/lib/qlslibs/efp.py

#
# 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.
#

"""
Module: qlslibs.efp

Contains empty file pool (EFP) classes.
"""

import os
import os.path
import qlslibs.err
import shutil
import uuid

class EfpManager(object):
    """
    Top level class to analyze the Qpid Linear Store (QLS) directory for the partitions that make up the
    Empty File Pool (EFP).
    """
    def __init__(self, directory, disk_space_required_kb):
        if not os.path.exists(directory):
            raise qlslibs.err.InvalidQlsDirectoryNameError(directory)
        self.directory = directory
        self.disk_space_required_kb = disk_space_required_kb
        self.efp_partitions = []
        self.efp_pools = {}
        self.total_num_files = 0
        self.total_cum_file_size_kb = 0
        self.current_efp_partition = None
    def add_file_pool(self, file_size_kb, num_files):
        """ Add an EFP in the specified partition of the specified size containing the specified number of files """
        dir_name = EmptyFilePool.get_directory_name(file_size_kb)
        print 'Adding pool \'%s\' to partition %s' % (dir_name, self.current_efp_partition.partition_number)
        self.total_cum_file_size_kb += self.current_efp_partition.create_new_efp(file_size_kb, num_files)
        self.total_num_files += num_files
    def freshen_file_pool(self, file_size_kb, num_files):
        """ Freshen an EFP in the specified partition and of the specified size to the specified number of files """
        if self.current_efp_partition is None:
            partition_list = self.efp_partitions
            partition_str = 'all partitions'
        else:
            partition_list = [self.current_efp_partition]
            partition_str = 'partition %d' % self.current_efp_partition.partition_number
        if file_size_kb is None:
            pool_str = 'all pools'
        else:
            pool_str = 'pool \'%s\'' % EmptyFilePool.get_directory_name(int(file_size_kb))
        print 'Freshening %s in %s to %d files' % (pool_str, partition_str, num_files)
        for self.current_efp_partition in partition_list: # Partition objects
            if file_size_kb is None:
                file_size_list = self.current_efp_partition.efp_pools.keys()
            else:
                file_size_list = ['%sk' % file_size_kb]
            for file_size in file_size_list:
                efp = self.current_efp_partition.efp_pools[file_size]
                num_files_needed = num_files - efp.get_tot_file_count()
                if num_files_needed > 0:
                    self.current_efp_partition.create_new_efp_files(qlslibs.utils.efp_directory_size(file_size),
                                                                    num_files_needed)
                else:
                    print '  WARNING: Pool %s in partition %s already contains %d files: no action taken' % \
                          (self.current_efp_partition.efp_pools[file_size].size_str,
                           self.current_efp_partition.partition_number, efp.get_num_files())
    def remove_file_pool(self, file_size_kb):
        """ Remove an existing EFP from the specified partition and of the specified size """
        dir_name = EmptyFilePool.get_directory_name(file_size_kb)
        print 'Removing pool \'%s\' from partition %s' % (dir_name, self.current_efp_partition.partition_number)
        self.efp_partitions.remove(self.current_efp_partition)
        shutil.rmtree(os.path.join(self.current_efp_partition.efp_directory, dir_name))
    def report(self):
        print 'Empty File Pool (EFP) report'
        print '============================'
        print 'Found', len(self.efp_partitions), 'partition(s)'
        if (len(self.efp_partitions)) > 0:
            sorted_efp_partitions = sorted(self.efp_partitions, key=lambda x: x.partition_number)
            EfpPartition.print_report_table_header()
            for ptn in sorted_efp_partitions:
                ptn.print_report_table_line()
            print
            for ptn in sorted_efp_partitions:
                ptn.report()
    def run(self, arg_tup):
        self._analyze_efp()
        if arg_tup is not None:
            _, arg_file_size, arg_num_files, arg_add, arg_remove, arg_freshen, arg_list = arg_tup
            self._check_args(arg_tup)
            if arg_add:
                self.add_file_pool(int(arg_file_size), int(arg_num_files))
            if arg_remove:
                self.remove_file_pool(int(arg_file_size))
            if arg_freshen:
                self.freshen_file_pool(arg_file_size, int(arg_num_files))
            if arg_list:
                self.report()
    def _analyze_efp(self):
        for dir_entry in os.listdir(self.directory):
            try:
                efp_partition = EfpPartition(os.path.join(self.directory, dir_entry), self.disk_space_required_kb)
                efp_partition.scan()
                self.efp_partitions.append(efp_partition)
                for efpl in efp_partition.efp_pools.iterkeys():
                    if efpl not in self.efp_pools:
                        self.efp_pools[efpl] = []
                    self.efp_pools[efpl].append(efp_partition.efp_pools[efpl])
                self.total_num_files += efp_partition.tot_file_count
                self.total_cum_file_size_kb += efp_partition.tot_file_size_kb
            except qlslibs.err.InvalidPartitionDirectoryNameError:
                pass
    def _check_args(self, arg_tup):
        """ Value check of args. The names of partitions and pools are validated against the discovered instances """
        arg_partition, arg_file_size, _, arg_add, arg_remove, arg_freshen, _ = arg_tup
        if arg_partition is not None:
            try:
                if arg_partition[0] == 'p': # string partition name, eg 'p001'
                    partition_num = int(arg_partition[1:])
                else: # numeric partition, eg '1'
                    partition_num = int(arg_partition)
                found = False
                for partition in self.efp_partitions:
                    if partition.partition_number == partition_num:
                        self.current_efp_partition = partition
                        found = True
                        break
                if not found:
                    raise qlslibs.err.PartitionDoesNotExistError(arg_partition)
            except ValueError:
                raise qlslibs.err.InvalidPartitionDirectoryNameError(arg_partition)
        if self.current_efp_partition is not None:
            pool_list = self.current_efp_partition.efp_pools.keys()
            efp_directory_name = EmptyFilePool.get_directory_name(int(arg_file_size))
            if arg_add and efp_directory_name in pool_list:
                raise qlslibs.err.PoolDirectoryAlreadyExistsError(efp_directory_name)
            if (arg_remove or arg_freshen) and efp_directory_name not in pool_list:
                raise qlslibs.err.PoolDirectoryDoesNotExistError(efp_directory_name)

class EfpPartition(object):
    """
    Class that represents a EFP partition. Each partition contains one or more Empty File Pools (EFPs).
    """
    PTN_DIR_PREFIX = 'p'
    EFP_DIR_NAME = 'efp'
    def __init__(self, directory, disk_space_required_kb):
        self.directory = directory
        self.partition_number = None
        self.efp_pools = {}
        self.tot_file_count = 0
        self.tot_file_size_kb = 0
        self._validate_partition_directory(disk_space_required_kb)
    def create_new_efp_files(self, file_size_kb, num_files):
        """ Create new EFP files in this partition """
        dir_name = EmptyFilePool.get_directory_name(file_size_kb)
        if dir_name in self.efp_pools.keys():
            efp = self.efp_pools[dir_name]
        else:
            efp = EmptyFilePool(os.path.join(self.directory, EfpPartition.EFP_DIR_NAME), dir_name)
        this_tot_file_size_kb = efp.create_new_efp_files(num_files)
        self.tot_file_size_kb += this_tot_file_size_kb
        self.tot_file_count += num_files
        return this_tot_file_size_kb
    @staticmethod
    def print_report_table_header():
        print 'p_no no_efp tot_files tot_size_kb directory'
        print '---- ------ --------- ----------- ---------'
    def print_report_table_line(self):
        print '%4d %6d %9d %11d %s' % (self.partition_number, len(self.efp_pools), self.tot_file_count,
                                       self.tot_file_size_kb, self.directory)
    def report(self):
        print 'Partition %s:' % os.path.basename(self.directory)
        if len(self.efp_pools) > 0:
            EmptyFilePool.print_report_table_header()
            for dir_name in self.efp_pools.keys():
                self.efp_pools[dir_name].print_report_table_line()
        else:
            print '<empty - no EFPs found in this partition>'
        print
    def scan(self):
        if os.path.exists(self.directory):
            efp_dir = os.path.join(self.directory, EfpPartition.EFP_DIR_NAME)
            for dir_entry in os.listdir(efp_dir):
                efp = EmptyFilePool(os.path.join(efp_dir, dir_entry), self.partition_number)
                efp.scan()
                self.tot_file_count += efp.get_tot_file_count()
                self.tot_file_size_kb += efp.get_tot_file_size_kb()
                self.efp_pools[dir_entry] = efp
    def _validate_partition_directory(self, disk_space_required_kb):
        if os.path.basename(self.directory)[0] is not EfpPartition.PTN_DIR_PREFIX:
            raise qlslibs.err.InvalidPartitionDirectoryNameError(self.directory)
        try:
            self.partition_number = int(os.path.basename(self.directory)[1:])
        except ValueError:
            raise qlslibs.err.InvalidPartitionDirectoryNameError(self.directory)
        if not qlslibs.utils.has_write_permission(self.directory):
            raise qlslibs.err.WritePermissionError(self.directory)
        if disk_space_required_kb is not None:
            space_avail = qlslibs.utils.get_avail_disk_space(self.directory)
            if space_avail < (disk_space_required_kb * 1024):
                raise qlslibs.err.InsufficientSpaceOnDiskError(self.directory, space_avail,
                                                               disk_space_required_kb * 1024)

class EmptyFilePool(object):
    """
    Class that represents a single Empty File Pool within a partition. Each EFP contains pre-formatted linear store
    journal files (but it may also be empty).
    """
    EFP_DIR_SUFFIX = 'k'
    EFP_JRNL_EXTENTION = '.jrnl'
    EFP_INUSE_DIRNAME = 'in_use'
    EFP_RETURNED_DIRNAME = 'returned'
    def __init__(self, directory, partition_number):
        self.base_dir_name = os.path.basename(directory)
        self.directory = directory
        self.partition_number = partition_number
        self.data_size_kb = None
        self.efp_files = []
        self.in_use_files = []
        self.returned_files = []
        self._validate_efp_directory()
    def create_new_efp_files(self, num_files):
        """ Create one or more new empty journal files of the prescribed size for this EFP """
        this_total_file_size = 0
        for _ in range(num_files):
            this_total_file_size += self._create_new_efp_file()
        return this_total_file_size
    def get_directory(self):
        return self.directory
    @staticmethod
    def get_directory_name(file_size_kb):
        """ Static function to create an EFP directory name from the size of the files it contains """
        return '%dk' % file_size_kb
    def get_tot_file_count(self):
        return len(self.efp_files)
    def get_tot_file_size_kb(self):
        return self.data_size_kb * len(self.efp_files)
    @staticmethod
    def print_report_table_header():
        print '             ---------- efp ------------ --------- in_use ---------- -------- returned ---------'
        print 'data_size_kb file_count tot_file_size_kb file_count tot_file_size_kb file_count tot_file_size_kb efp_directory'
        print '------------ ---------- ---------------- ---------- ---------------- ---------- ---------------- -------------'
    def print_report_table_line(self):
        print '%12d %10d %16d %10d %16d %10d %16d %s' % (self.data_size_kb, len(self.efp_files),
                                                         self.data_size_kb * len(self.efp_files),
                                                         len(self.in_use_files),
                                                         self.data_size_kb * len(self.in_use_files),
                                                         len(self.returned_files),
                                                         self.data_size_kb * len(self.returned_files),
                                                         self.get_directory())
    def scan(self):
        for efp_file in os.listdir(self.directory):
            if efp_file == self.EFP_INUSE_DIRNAME:
                for in_use_file in os.listdir(os.path.join(self.directory, self.EFP_INUSE_DIRNAME)):
                    self.in_use_files.append(in_use_file)
                continue
            if efp_file == self.EFP_RETURNED_DIRNAME:
                for returned_file in os.listdir(os.path.join(self.directory, self.EFP_RETURNED_DIRNAME)):
                    self.returned_files.append(returned_file)
                continue
            if self._validate_efp_file(os.path.join(self.directory, efp_file)):
                self.efp_files.append(efp_file)
    def _add_efp_file(self, efp_file_name):
        """ Add a single journal file of the appropriate size to this EFP. No file size check is made here. """
        self.efp_files.append(efp_file_name)
    def _create_new_efp_file(self):
        """ Create a single new empty journal file of the prescribed size for this EFP """
        file_name = str(uuid.uuid4()) + EmptyFilePool.EFP_JRNL_EXTENTION
        file_header = qlslibs.jrnl.FileHeader(0, qlslibs.jrnl.FileHeader.MAGIC, qlslibs.utils.DEFAULT_RECORD_VERSION,
                                              0, 0, 0)
        file_header.init(None, None, qlslibs.utils.DEFAULT_HEADER_SIZE_SBLKS, self.partition_number, self.data_size_kb,
                         0, 0, 0, 0, 0)
        efh = file_header.encode()
        efh_bytes = len(efh)
        file_handle = open(os.path.join(self.directory, file_name), 'wb')
        file_handle.write(efh)
        file_handle.write('\xff' * (qlslibs.utils.DEFAULT_SBLK_SIZE - efh_bytes))
        file_handle.write('\x00' * (int(self.data_size_kb) * 1024))
        file_handle.close()
        fqfn = os.path.join(self.directory, file_name)
        self._add_efp_file(fqfn)
        return os.path.getsize(fqfn)
    def _validate_efp_directory(self):
        if self.base_dir_name[-1] is not EmptyFilePool.EFP_DIR_SUFFIX:
            raise qlslibs.err.InvalidEfpDirectoryNameError(self.directory)
        try:
            self.data_size_kb = int(os.path.basename(self.base_dir_name)[:-1])
        except ValueError:
            raise qlslibs.err.InvalidEfpDirectoryNameError(self.directory)
    def _validate_efp_file(self, efp_file):
        file_size = os.path.getsize(efp_file)
        expected_file_size = (self.data_size_kb * 1024) + qlslibs.utils.DEFAULT_SBLK_SIZE
        if file_size != expected_file_size:
            print 'WARNING: File %s not of correct size (size=%d, expected=%d): Ignoring' % (efp_file, file_size,
                                                                                             expected_file_size)
            return False
        file_handle = open(efp_file)
        args = qlslibs.utils.load_args(file_handle, qlslibs.jrnl.RecordHeader)
        file_hdr = qlslibs.jrnl.FileHeader(*args)
        file_hdr.init(file_handle, *qlslibs.utils.load_args(file_handle, qlslibs.jrnl.FileHeader))
        if not file_hdr.is_header_valid(file_hdr):
            file_handle.close()
            return False
        file_hdr.load(file_handle)
        file_handle.close()
        if not file_hdr.is_valid(True):
            return False
        return True


# =============================================================================

if __name__ == "__main__":
    print "This is a library, and cannot be executed."