| # 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. |
| # Line too long - pylint: disable=C0301 |
| |
| import sys |
| import os |
| import errno |
| import shutil |
| import hashlib |
| import signal |
| import urlparse |
| |
| from gppylib.db import dbconn, catalog |
| from gppylib.db.dbconn import UnexpectedRowsError, executeUpdateOrInsert |
| from gppylib import gplog |
| from gppylib.commands.gp import GpStop, GpStart, get_local_db_mode, get_masterdatadir |
| from gppylib.commands.unix import Ping, RemoveDirectory, RemoveFiles, MakeDirectory |
| from gppylib.operations import Operation |
| from gppylib.operations.utils import RemoteOperation, ParallelOperation |
| from gppylib.operations.unix import CheckFile, CheckDir, ListFiles |
| from gppylib.gparray import GpArray |
| from hawqpylib.hawqarray import HAWQArray |
| from hawqpylib.hawqlib import local_ssh, HawqXMLParser, parse_hosts_file |
| |
| logger = gplog.get_default_logger() |
| |
| GP_TRANSACTION_FILES_FILESPACE = 'gp_transaction_files_filespace' |
| GP_TEMPORARY_FILES_FILESPACE = 'gp_temporary_files_filespace' |
| PG_SYSTEM_FILESPACE = 'pg_system' |
| MASTER_DATA_DIR = get_masterdatadir() |
| NUM_WORKERS = 16 #Number of simultaneous parallel process that are started by ParallelOperation |
| |
| class FileType: |
| """ |
| This class is to emulate an enum |
| """ |
| |
| TRANSACTION_FILES, TEMPORARY_FILES = range(2) |
| lookup = ['TRANSACTION_FILES', 'TEMPORARY_FILES'] |
| |
| class MoveFilespaceError(Exception): |
| def __init__(self, msg): |
| Exception.__init__(self, msg) |
| |
| class CheckFilespaceIsSame(Operation): |
| """ |
| Checks that the given filespace is different |
| from the one currently being used. |
| |
| @return: True if Filespace is different False otherwise |
| """ |
| |
| def __init__(self, gparray, filespace_name, file_type): |
| self.gparray = gparray |
| self.filespace_name = filespace_name |
| self.file_type = file_type |
| |
| def execute(self): |
| logger.info('Checking if filespace is same as current filespace') |
| |
| try: |
| if self.file_type == FileType.TRANSACTION_FILES: |
| filename = GP_TRANSACTION_FILES_FILESPACE |
| else: |
| filename = GP_TEMPORARY_FILES_FILESPACE |
| |
| pg_system_filespace_entries = GetFilespaceEntries(self.gparray, PG_SYSTEM_FILESPACE).run() |
| |
| file_path = os.path.join(pg_system_filespace_entries[0][2], filename) |
| |
| if self.filespace_name == 'pg_system' and not CheckFile(file_path).run(): |
| return True |
| |
| if CheckFile(file_path).run(): |
| with open(file_path) as file: |
| oid = int(file.readline().strip()) |
| if self.gparray.getFileSpaceName(oid) == self.filespace_name: |
| return True |
| except Exception, e: |
| raise MoveFilespaceError('Failed to check if filespace is same.') |
| |
| return False |
| |
| class CheckFilespace(Operation): |
| """ |
| Checks that the given filespace exists. |
| |
| @return: True if Filespace exists False otherwise |
| """ |
| SELECT_FILESPACENAME_QUERY = """ |
| select * from pg_filespace where fsname = \'%s\' |
| """ |
| |
| def __init__(self, filespace): |
| self.filespace = filespace |
| |
| def execute(self): |
| logger.info('Checking if filespace %s exists' % self.filespace) |
| |
| dburl = dbconn.DbURL() |
| query = self.SELECT_FILESPACENAME_QUERY % self.filespace |
| |
| logger.debug('Connecting to database') |
| with dbconn.connect(dburl, utility=True) as conn: |
| try: |
| logger.debug('Executing query %s' % query) |
| tuple = dbconn.execSQLForSingletonRow(conn, query) |
| except UnexpectedRowsError, e: |
| if e.actual == 0: |
| return False |
| raise |
| |
| return True |
| |
| class CheckSuperUser(Operation): |
| """ |
| Checks if the user running the gpfilespace utility |
| is a super user or not. |
| """ |
| def __init__(self, dburl): |
| self.dburl = dburl |
| |
| def execute(self): |
| is_super = None |
| conn = dbconn.connect(self.dburl, utility=True) |
| try: |
| q = "SELECT usesuper FROM pg_user where usename = user" |
| rows = catalog.basicSQLExec(conn, q) |
| is_super = rows[0][0] |
| except Exception, e: |
| raise MoveFilespaceError(str(e)) |
| finally: |
| conn.close() |
| |
| if not is_super: |
| return False |
| |
| return True |
| |
| class CheckConnectivity(Operation): |
| """ |
| Check if all hosts are reachable |
| in the cluster. |
| """ |
| def __init__(self, gparray): |
| self.gparray = gparray |
| |
| def execute(self): |
| logger.info('Checking for connectivity') |
| |
| host_set = self.gparray.getHostList() |
| |
| for host in host_set: |
| try: |
| Ping.local("pinging host %s" % host, host) |
| except Exception, e: |
| logger.error('Cannot connect to host %s: %s' % (host, e)) |
| return False |
| |
| return True |
| |
| class CheckFilespaceOidLocally(Operation): |
| """ |
| Returns the oid present in the flat files |
| """ |
| def __init__(self, flat_file_location): |
| self.flat_file_location = flat_file_location |
| |
| def execute(self): |
| logger.info('flat file location = %s' % self.flat_file_location) |
| if not CheckFile(self.flat_file_location).run(): |
| return -1 |
| |
| with open(self.flat_file_location, 'r') as file: |
| oid = file.readline().strip() |
| return oid |
| |
| class CheckFilespaceEntriesLocally(Operation): |
| """ |
| Check if filespace entries are valid |
| """ |
| def __init__(self, cur_filespace_entry, peer_filespace_entry, filespace_location, file_type): |
| self.cur_filespace_entry = cur_filespace_entry |
| self.peer_filespace_entry = peer_filespace_entry |
| self.filespace_location = filespace_location |
| self.file_type = file_type |
| |
| def execute(self): |
| if self.file_type == FileType.TRANSACTION_FILES: |
| flat_file = GP_TRANSACTION_FILES_FILESPACE |
| elif self.file_type == FileType.TEMPORARY_FILES: |
| flat_file = GP_TEMPORARY_FILES_FILESPACE |
| |
| is_consistent = True |
| flat_file_location = os.path.join(self.filespace_location, flat_file) |
| |
| #Default filespace pg_system |
| if not os.path.exists(flat_file_location): |
| return is_consistent |
| |
| if not os.path.exists(self.cur_filespace_entry[2]): |
| logger.info('path %s does not exist' % self.cur_filespace_entry[2]) |
| is_consistent = False |
| |
| #Read flat file entries and check if the directories exist |
| entries_set = set() |
| with open(flat_file_location) as file: |
| for line in file: |
| entries = line.split() |
| if len(entries) != 2: |
| continue |
| else: |
| logger.info('path = %s' % entries[1].strip()) |
| logger.info('dbid = %s' % entries[0].strip()) |
| entries_set.add(entries[0].strip() + ' ' + entries[1].strip()) |
| |
| required_set = set() |
| required_set.add(str(self.cur_filespace_entry[1]) + ' ' + self.cur_filespace_entry[2]) |
| |
| if self.peer_filespace_entry is not None: |
| required_set.add(str(self.peer_filespace_entry[1]) + ' ' + self.peer_filespace_entry[2]) |
| |
| if required_set != entries_set: |
| is_consistent = False |
| |
| return is_consistent |
| |
| class CheckFilespacePermissions(Operation): |
| """ |
| Checks if the filespace has the correct permissions |
| """ |
| def __init__(self, location): |
| self.location = location |
| |
| def execute(self): |
| |
| path = os.path.dirname(self.location) |
| |
| #In case of recoverseg, the segment directories will be |
| #created by the backend code. |
| if not os.path.exists(path): |
| path = os.path.dirname(path) |
| |
| self.location = os.path.join(path, os.path.basename(self.location)) |
| |
| try: |
| with open(self.location, 'w') as file: |
| pass |
| os.remove(self.location) |
| except IOError, e: |
| if e.errno == errno.EACCES: |
| return False |
| else: |
| raise |
| return True |
| |
| class CheckFilespaceConsistency(Operation): |
| """ |
| Checks that the filespaces for transaction and temporary |
| files are consistent on all segments |
| """ |
| def __init__(self, gparray, file_type): |
| self.gparray = gparray |
| self.file_type = file_type |
| |
| def execute(self): |
| logger.info('Checking for filespace consistency') |
| if self.file_type == FileType.TRANSACTION_FILES: |
| flat_file = GP_TRANSACTION_FILES_FILESPACE |
| elif self.file_type == FileType.TEMPORARY_FILES: |
| flat_file = GP_TEMPORARY_FILES_FILESPACE |
| |
| operations = [] |
| pg_system_fs_entries = GetFilespaceEntriesDict(GetFilespaceEntries(self.gparray, |
| PG_SYSTEM_FILESPACE).run()).run() |
| cur_filespace_entries = GetFilespaceEntriesDict(GetCurrentFilespaceEntries(self.gparray, |
| self.file_type).run()).run() |
| for seg in self.gparray.getDbList(): |
| flat_file_location = os.path.join(pg_system_fs_entries[seg.getSegmentDbId()][2], |
| flat_file) |
| logger.debug('flat file location = %s' % flat_file_location) |
| operations.append(RemoteOperation(CheckFilespaceOidLocally(flat_file_location), |
| seg.getSegmentHostName() |
| ) |
| ) |
| ParallelOperation(operations, NUM_WORKERS).run() |
| |
| try: |
| oid_set = set([int(op.get_ret()) for op in operations]) |
| except Exception, e: |
| logger.error('Invalid OID in flat file on host %s' % op.host) |
| return False |
| |
| logger.debug('%s oid set = %s' % (FileType.lookup[self.file_type], oid_set)) |
| |
| if len(oid_set) != 1: |
| logger.error('Multiple OIDs found in flat files') |
| return False |
| |
| #Verify that we have a valid oid and that |
| #the filespace exists in the catalog. |
| try: |
| oid = int(oid_set.pop()) |
| if oid == -1: |
| fs_name = 'pg_system' |
| else: |
| fs_name = self.gparray.getFileSpaceName(oid) |
| except Exception, e: |
| logger.error('%s OIDs are inconsistent for %s filespace' % (FileType.lookup[self.file_type], fs_name)) |
| return False |
| logger.info('%s OIDs are consistent for %s filespace' % (FileType.lookup[self.file_type], fs_name)) |
| |
| #Now check for the filespace entries |
| operation_list = [] |
| for seg in self.gparray.getDbList(): |
| cur_filespace_entry = cur_filespace_entries[seg.getSegmentDbId()] |
| peer_filespace_entry = get_peer_filespace_entry(cur_filespace_entries, seg.getSegmentDbId(), seg.getSegmentContentId(), self.gparray.getDbList()) |
| logger.debug('current_filespace_entry = %s' % str(cur_filespace_entry)) |
| logger.debug('peer_filespace_entry = %s' % str(peer_filespace_entry)) |
| operation_list.append(RemoteOperation(CheckFilespaceEntriesLocally(cur_filespace_entry, peer_filespace_entry, pg_system_fs_entries[seg.getSegmentDbId()][2], self.file_type), |
| seg.getSegmentHostName() |
| ) |
| ) |
| |
| ParallelOperation(operation_list, NUM_WORKERS).run() |
| |
| for operation in operation_list: |
| try: |
| if not operation.get_ret(): |
| logger.error('%s entries are inconsistent for %s filespace on host %s' % (FileType.lookup[self.file_type], fs_name, operation.host)) |
| return False |
| except Exception, e: |
| logger.error('%s entries are inconsistent for %s filespace on host %s' % (FileType.lookup[self.file_type], fs_name, operation.host)) |
| return False |
| |
| logger.info('%s entries are consistent for %s filespace' % (FileType.lookup[self.file_type], fs_name)) |
| |
| return True |
| |
| def get_peer_filespace_entry(filespace_entries, dbid, content_id, segments): |
| """ |
| Returns the filespace entry for a given |
| DbId corresponding to its peer. |
| i.e given primary, it returns the filespace |
| entry of the mirror, and given mirror, it |
| returns the filespace entry of the primary. |
| |
| @param: DbId of the primary segment |
| content id of the segment |
| @return: A tuple of the form |
| filespace oid, dbid, filespace directory |
| """ |
| logger.debug('Getting peer filespace entry for %s' % dbid) |
| for seg in segments: |
| if seg.getSegmentContentId() == content_id and seg.getSegmentDbId() != dbid: |
| return filespace_entries[seg.getSegmentDbId()] |
| |
| |
| class UpdateFlatFilesLocally(Operation): |
| """ |
| Update the flat files on primaries/mirrors locally |
| """ |
| def __init__(self, flat_file, filespace_oid, cur_filespace_entry, peer_filespace_entry): |
| self.flat_file = flat_file |
| self.filespace_oid = filespace_oid |
| self.cur_filespace_entry = cur_filespace_entry |
| self.peer_filespace_entry = peer_filespace_entry |
| |
| def execute(self): |
| logger.info('Updating flat files') |
| |
| #If flat file directory does not exist, we create it |
| if not os.path.exists(os.path.dirname(self.flat_file)): |
| os.mkdir(os.path.dirname(self.flat_file)) |
| |
| temp_flat_file = self.flat_file + '.tmp' |
| lines_to_write = str(self.filespace_oid) + '\n' +\ |
| str(self.cur_filespace_entry[1]) + ' ' + self.cur_filespace_entry[2] + '\n' |
| if self.peer_filespace_entry is not None: |
| lines_to_write += str(self.peer_filespace_entry[1]) + ' ' + self.peer_filespace_entry[2] + '\n' |
| |
| with open(temp_flat_file, 'w') as file: |
| file.write(lines_to_write) |
| |
| logger.debug('Wrote contents = %s' % lines_to_write) |
| |
| #Read back and check if what we've written is the same |
| with open(temp_flat_file) as file: |
| contents = file.read() |
| if contents != lines_to_write: |
| raise MoveFilespaceError('Failed to write contents to flat file %s' % lines_to_write) |
| |
| shutil.move(temp_flat_file, self.flat_file) |
| |
| if not os.path.exists(self.cur_filespace_entry[2]): |
| os.makedirs(self.cur_filespace_entry[2]) |
| |
| return self.flat_file + ' ' + contents |
| |
| class UpdateFlatFiles(Operation): |
| """ |
| Update the flat files on primaries separately |
| in case of addmirrors. |
| Update the flat files on the mirrors separately |
| in case of gpexpand. |
| """ |
| def __init__(self, gparray, primaries=True, expansion=False): |
| self.gparray = gparray |
| self.primaries = primaries |
| self.expansion = expansion |
| |
| def execute(self): |
| |
| #Obtain list of segments from gparray |
| if self.expansion: |
| db_list = self.gparray.getExpansionSegDbList() |
| else: |
| db_list = self.gparray.getDbList() |
| |
| if self.primaries: |
| segments = [seg for seg in db_list if seg.isSegmentPrimary()] |
| else: |
| segments = [seg for seg in db_list if seg.isSegmentMirror()] |
| |
| logger.debug('segment_list = %s' % self.gparray.getDbList()) |
| logger.debug('segments on which flat files will be updated = %s' % segments) |
| pg_system_filespace_entries = GetFilespaceEntriesDict(GetFilespaceEntries(self.gparray, |
| PG_SYSTEM_FILESPACE).run() |
| ).run() |
| transaction_flat_file = os.path.join(pg_system_filespace_entries[1][2], GP_TRANSACTION_FILES_FILESPACE) |
| if os.path.exists(transaction_flat_file): |
| logger.debug('Updating transaction flat files') |
| cur_filespace_entries = GetFilespaceEntriesDict(GetCurrentFilespaceEntries(self.gparray, FileType.TRANSACTION_FILES |
| ).run() |
| ).run() |
| operation_list = [] |
| for seg in segments: |
| filespace_oid = cur_filespace_entries[seg.getSegmentDbId()][0] |
| cur_filespace_entry = cur_filespace_entries[seg.getSegmentDbId()] |
| peer_filespace_entry = get_peer_filespace_entry(cur_filespace_entries, seg.getSegmentDbId(), seg.getSegmentContentId(), db_list) |
| logger.debug('cur_filespace_entry = %s' % str(cur_filespace_entry)) |
| logger.debug('peer_filespace_entry = %s' % str(peer_filespace_entry)) |
| flat_file = os.path.join(pg_system_filespace_entries[seg.getSegmentDbId()][2], |
| GP_TRANSACTION_FILES_FILESPACE) |
| operation_list.append(RemoteOperation(UpdateFlatFilesLocally(flat_file, |
| filespace_oid, |
| cur_filespace_entry, |
| peer_filespace_entry |
| ), |
| seg.getSegmentHostName()) |
| ) |
| |
| ParallelOperation(operation_list, NUM_WORKERS).run() |
| |
| try: |
| for operation in operation_list: |
| operation.get_ret() |
| except Exception, e: |
| raise MoveFilespaceError('Failed to update transaction flat file.') |
| |
| temporary_flat_file = os.path.join(pg_system_filespace_entries[1][2], GP_TEMPORARY_FILES_FILESPACE) |
| if os.path.exists(temporary_flat_file): |
| logger.debug('Updating temporary flat files') |
| cur_filespace_entries = GetFilespaceEntriesDict(GetCurrentFilespaceEntries(self.gparray, FileType.TEMPORARY_FILES |
| ).run() |
| ).run() |
| operation_list = [] |
| for seg in segments: |
| filespace_oid = cur_filespace_entries[seg.getSegmentDbId()][0] |
| cur_filespace_entry = cur_filespace_entries[seg.getSegmentDbId()] |
| peer_filespace_entry = get_peer_filespace_entry(cur_filespace_entries, seg.getSegmentDbId(), seg.getSegmentContentId(), db_list) |
| logger.debug('cur_filespace_entry = %s' % str(cur_filespace_entry)) |
| logger.debug('peer_filespace_entry = %s' % str(peer_filespace_entry)) |
| flat_file = os.path.join(pg_system_filespace_entries[seg.getSegmentDbId()][2], |
| GP_TEMPORARY_FILES_FILESPACE) |
| operation_list.append(RemoteOperation(UpdateFlatFilesLocally(flat_file, |
| filespace_oid, |
| cur_filespace_entry, |
| peer_filespace_entry |
| ), |
| seg.getSegmentHostName()) |
| ) |
| |
| ParallelOperation(operation_list, NUM_WORKERS).run() |
| |
| try: |
| for operation in operation_list: |
| operation.get_ret() |
| except Exception, e: |
| raise MoveFilespaceError('Failed to update temporary flat file.') |
| |
| class MoveTransFilespaceLocally(Operation): |
| """ |
| Move the Filespace to the new location for |
| transaction files. |
| This requires the pg_system filespace entry for a particular dbid |
| because that information is needed to write the gp_transaction_files_filespace |
| flat file. |
| |
| @param: filespace entries from the pg_filespace_entry table. |
| @return: None |
| """ |
| TRANSACTION_FILES_DIRS = ['pg_xlog', 'pg_multixact', 'pg_subtrans', 'pg_clog', |
| 'pg_distributedlog', 'pg_distributedxidmap'] |
| |
| def __init__(self, current_filespace_entry, new_filespace_name, new_filespace_entry, peer_filespace_entry, pg_system_filespace_entry, rollback=False): |
| self.new_filespace_name = new_filespace_name |
| self.new_filespace_entry = new_filespace_entry |
| self.peer_filespace_entry = peer_filespace_entry |
| self.current_filespace_entry = current_filespace_entry |
| self.pg_system_filespace_entry = pg_system_filespace_entry |
| self.rollback = rollback |
| |
| def get_md5(self, directory): |
| m = hashlib.md5() |
| |
| files_to_hash = [] |
| for root, dirs, filenames in os.walk(directory): |
| for filename in filenames: |
| file = os.path.join(root, filename) |
| files_to_hash.append(file) |
| |
| files_to_hash.sort() |
| |
| for file in files_to_hash: |
| with open(file, 'rb') as f: |
| while True: |
| data = f.read(128) |
| if not data: |
| break |
| m.update(data) |
| return m.hexdigest() |
| |
| def md5_check(self, src_dir, dst_dir): |
| src_md5 = self.get_md5(src_dir) |
| dst_md5 = self.get_md5(dst_dir) |
| |
| logger.info('md5 checksum for %s = %s' % (src_dir, src_md5)) |
| logger.info('md5 checksum for %s = %s' % (dst_dir, dst_md5)) |
| |
| if src_md5 != dst_md5: |
| raise MoveFilespaceError('MD5 Checksum failed for %s' % src_dir) |
| |
| def execute(self): |
| |
| #if the filespace directory is not present on the segment, |
| #We simply return. |
| if not CheckDir(self.current_filespace_entry[2]).run(): |
| return |
| |
| tmp_file = os.path.join(self.new_filespace_entry[2], 'tmp_file') |
| if not CheckFilespacePermissions(tmp_file).run(): |
| raise MoveFilespaceError('Invalid permissions for %s' % tmp_file) |
| |
| stats = os.statvfs(os.path.dirname(self.new_filespace_entry[2])) |
| free_bytes = stats.f_bfree * stats.f_frsize |
| logger.info('free_bytes for %s = %s' % (self.new_filespace_entry[2], free_bytes)) |
| if not free_bytes: |
| raise MoveFilespaceError('Insufficient diskspace for %s' % self.new_filespace_entry[2]) |
| |
| #1. Copy the contents of the filespace dirs from old filespace to new filespace |
| for directory in self.TRANSACTION_FILES_DIRS: |
| src_dir = os.path.join(self.current_filespace_entry[2], directory) |
| dst_dir = os.path.join(self.new_filespace_entry[2], directory) |
| |
| logger.info('Copying directories from %s to %s' % (src_dir, dst_dir)) |
| try: |
| #In case of a rollback, we need to retain the directories that |
| #already exist on the source segment. If it doesn't then we do a |
| #copy. |
| if self.rollback: |
| if CheckDir(src_dir).run() and not CheckDir(dst_dir).run(): |
| logger.info('copying %s' % src_dir) |
| shutil.copytree(src_dir, dst_dir) |
| try: |
| self.md5_check(src_dir, dst_dir) |
| except (IOError, OSError), e: |
| raise MoveFilespaceError('Failed to calculate md5 checksums !') |
| elif CheckDir(src_dir).run(): |
| #We use the src segment as the source of truth and copy everything |
| if CheckDir(dst_dir).run(): |
| shutil.rmtree(dst_dir) |
| |
| logger.info('copying %s' % src_dir) |
| shutil.copytree(src_dir, dst_dir) |
| |
| try: |
| self.md5_check(src_dir, dst_dir) |
| except (IOError, OSError), e: |
| raise MoveFilespaceError('Failed to calculate md5 checksums !') |
| |
| except (IOError, OSError), e: |
| logger.error('Failed to copy transaction files to new Filespace location.') |
| raise |
| |
| #2. Drop the directories in current filespace |
| for directory in self.TRANSACTION_FILES_DIRS: |
| src_dir = os.path.join(self.current_filespace_entry[2], directory) |
| try: |
| if CheckDir(src_dir).run(): |
| logger.info('Dropping dir %s' % src_dir) |
| shutil.rmtree(src_dir) |
| except (IOError, OSError), e: |
| logger.error('Failed to drop transaction files directories from current filespace.') |
| raise |
| |
| gp_transaction_files_filespace_path = os.path.join(self.pg_system_filespace_entry[2], GP_TRANSACTION_FILES_FILESPACE) |
| #If we are moving to default Filespace, then we need to delete the flat file |
| if self.new_filespace_name == PG_SYSTEM_FILESPACE: |
| if CheckFile(gp_transaction_files_filespace_path).run(): |
| os.remove(gp_transaction_files_filespace_path) |
| return |
| |
| #3. Write the dbid and directories into tmp directory |
| temp_gp_transaction_files_filespace_path = gp_transaction_files_filespace_path + '.tmp' |
| |
| logger.info('Writing filespace information into flat file') |
| with open(temp_gp_transaction_files_filespace_path, 'w') as file: |
| lines_to_write = str(self.new_filespace_entry[0]) + '\n' +\ |
| str(self.new_filespace_entry[1]) + ' ' + self.new_filespace_entry[2] + '\n' |
| #In case of system without mirrors |
| if self.peer_filespace_entry is not None: |
| lines_to_write += str(self.peer_filespace_entry[1]) + ' ' + self.peer_filespace_entry[2] + '\n' |
| |
| file.write(lines_to_write) |
| |
| #Read back and check the file contents |
| with open(temp_gp_transaction_files_filespace_path) as file: |
| contents = file.read() |
| if contents != lines_to_write: |
| raise MoveFilespaceError('Failed to write contents to flat file %s' % lines_to_write) |
| |
| #4. After the file copy has been completed successfully, |
| # copy over the tmp directory file |
| shutil.move(temp_gp_transaction_files_filespace_path, gp_transaction_files_filespace_path) |
| |
| class MoveTempFilespaceLocally(Operation): |
| """ |
| Move the Filespace to the new location for |
| temporary files. |
| |
| @return: None |
| """ |
| TEMPORARY_FILES_DIR = 'pgsql_tmp' |
| |
| def __init__(self, current_filespace_entry, new_filespace_name, new_filespace_entry, peer_filespace_entry, pg_system_filespace_entry, rollback=None): |
| self.new_filespace_name = new_filespace_name |
| self.new_filespace_entry = new_filespace_entry |
| self.peer_filespace_entry = peer_filespace_entry |
| self.current_filespace_entry = current_filespace_entry |
| self.pg_system_filespace_entry = pg_system_filespace_entry |
| self.rollback = rollback |
| |
| def execute(self): |
| |
| if not CheckDir(self.current_filespace_entry[2]).run(): |
| return |
| |
| tmp_file = os.path.join(self.new_filespace_entry[2], 'tmp_file') |
| if not CheckFilespacePermissions(tmp_file).run(): |
| raise MoveFilespaceError('Invalid permissions for %s' % tmp_file) |
| |
| stats = os.statvfs(os.path.dirname(self.new_filespace_entry[2])) |
| free_bytes = stats.f_bfree * stats.f_frsize |
| logger.info('free_bytes for %s = %s' % (self.new_filespace_entry[2], free_bytes)) |
| if not free_bytes: |
| raise MoveFilespaceError('Insufficient diskspace for %s' % self.new_filespace_entry[2]) |
| |
| #1. Drop the directories from old filespace location |
| #If the filespace is being moved from pg_system filespace |
| #for temp files will be slightly different as they will be stored under |
| #<filespace_dir>/base/<database_oid>/ |
| gp_temporary_files_filespace_path = os.path.join(self.pg_system_filespace_entry[2], GP_TEMPORARY_FILES_FILESPACE) |
| |
| if not CheckFile(gp_temporary_files_filespace_path).run(): |
| base_dir = os.path.join(self.current_filespace_entry[2], 'base') |
| |
| directories = [] |
| try: |
| directories = ListFiles(base_dir).run() |
| except Exception, e: |
| if self.rollback: |
| pass |
| else: |
| raise |
| |
| for directory in directories: |
| try: |
| dst_dir = os.path.join(base_dir, directory, self.TEMPORARY_FILES_DIR) |
| if CheckDir(dst_dir).run(): |
| logger.info('Dropping dir %s' % dst_dir) |
| shutil.rmtree(dst_dir) |
| except (IOError, OSError), e: |
| logger.error('Failed to delete temporary files') |
| raise |
| else: |
| #else if filespace has been moved before, the temp files directory will always be under |
| #<filespace_dir>/pgsql_tmp |
| dst_dir = os.path.join(self.current_filespace_entry[2], self.TEMPORARY_FILES_DIR) |
| try: |
| if CheckDir(dst_dir).run(): |
| logger.info('Dropping dir %s' %dst_dir) |
| shutil.rmtree(dst_dir) |
| except (IOError, OSError), e: |
| logger.error('Failed to delete temporary filespace directories') |
| raise |
| |
| #If we are moving to default Filespace, then we need to delete the flat file |
| if self.new_filespace_name == PG_SYSTEM_FILESPACE: |
| if CheckFile(gp_temporary_files_filespace_path).run(): |
| os.remove(gp_temporary_files_filespace_path) |
| return |
| |
| #2. Write the dbid and directories into tmp directory |
| temp_gp_temporary_files_filespace_path = gp_temporary_files_filespace_path + '.tmp' |
| |
| logger.info('Writing filespace information into flat file') |
| with open(temp_gp_temporary_files_filespace_path, 'w') as file: |
| lines_to_write = str(self.new_filespace_entry[0]) + '\n' +\ |
| str(self.new_filespace_entry[1]) + ' ' + self.new_filespace_entry[2] + '\n' |
| #In case of master, when there is no standby, peer will be None |
| if self.peer_filespace_entry is not None: |
| lines_to_write += str(self.peer_filespace_entry[1]) + ' ' + self.peer_filespace_entry[2] + '\n' |
| |
| file.write(lines_to_write) |
| |
| #Read back and check the contents |
| with open(temp_gp_temporary_files_filespace_path) as file: |
| contents = file.read() |
| if contents != lines_to_write: |
| raise MoveFilespaceError('Failed to write contents to flat file %s' % lines_to_write) |
| |
| #3. After the file copy has been completed successfully, |
| # copy over the tmp directory file |
| shutil.move(temp_gp_temporary_files_filespace_path, gp_temporary_files_filespace_path) |
| |
| class GetFilespaceEntries(Operation): |
| """ |
| Returns the pg_filespace_entry table contents for a given filespace. |
| Each entry will be of the form oid, dbid, filespace_directory |
| |
| @return: List containing the directories used by the filespace |
| """ |
| def __init__(self, gparray, filespace_name): |
| self.filespace_name = filespace_name |
| self.gparray = gparray |
| |
| def execute(self): |
| logger.debug('Obtaining filespace information for filespace %s' % self.filespace_name) |
| filespace_entries = [] |
| |
| filespaces = self.gparray.getFilespaces(includeSystemFilespace=True) |
| filespace_oid = None |
| for fs in filespaces: |
| if fs.getName() == self.filespace_name: |
| filespace_oid = fs.getOid() |
| |
| if filespace_oid is None: |
| raise MoveFilespaceError('Invalid filespace name.') |
| |
| for seg in self.gparray.getDbList(includeExpansionSegs=True): |
| filespace_entries.append((filespace_oid, seg.getSegmentDbId(), seg.getSegmentFilespaces()[filespace_oid])) |
| |
| return filespace_entries |
| |
| class GetCurrentFilespaceEntries(Operation): |
| """ |
| The results returned will contain the entries from |
| pg_filespace_entry table for the current filespace used |
| by transaction/temporary files. |
| """ |
| |
| def __init__(self, gparray, file_type): |
| self.file_type = file_type |
| self.gparray = gparray |
| |
| def execute(self): |
| logger.info('Obtaining current filespace entries used by %s' % FileType.lookup[self.file_type]) |
| filespace_entries = [] |
| default_filespace_entries = GetFilespaceEntries(self.gparray, PG_SYSTEM_FILESPACE).run() |
| flat_files_dir = default_filespace_entries[0][2] |
| flat_file = None |
| |
| if self.file_type == FileType.TRANSACTION_FILES: |
| flat_file = GP_TRANSACTION_FILES_FILESPACE |
| elif self.file_type == FileType.TEMPORARY_FILES: |
| flat_file = GP_TEMPORARY_FILES_FILESPACE |
| |
| gp_filespace_path = os.path.join(flat_files_dir, flat_file) |
| |
| try: |
| if CheckFile(gp_filespace_path).run(): |
| with open(gp_filespace_path) as file: |
| fs_oid = file.readline().strip() |
| filespace_name = self.gparray.getFileSpaceName(int(fs_oid)) |
| filespace_entries = GetFilespaceEntries(self.gparray, filespace_name).run() |
| except TypeError, e: |
| raise MoveFilespaceError('Invalid oid in flat file. %s' % str(e)) |
| except (IOError, OSError), e: |
| raise MoveFilespaceError('Failed to read flat file.%s' % str(e)) |
| |
| if len(filespace_entries) == 0: |
| filespace_entries = default_filespace_entries |
| |
| return filespace_entries |
| |
| class GetFilespaceEntriesDict(Operation): |
| """ |
| @param: A list of filespace entries |
| @return: A dict containing the directories for each |
| corresponding database id. |
| """ |
| |
| def __init__(self, filespace_entries): |
| self.filespace_entries = filespace_entries |
| |
| def execute(self): |
| logger.debug('Converting filespace entries into dict') |
| db_directories_dict = {} |
| |
| for entry in self.filespace_entries: |
| dbid = entry[1] |
| db_directories_dict[dbid] = entry |
| |
| return db_directories_dict |
| |
| class RollBackFilespaceChanges(Operation): |
| """ |
| This does the reverse operation of Move(). |
| Since the Move from cur -> new Filespace failed, |
| we do the reverse and move files back from new -> cur. |
| """ |
| def __init__(self, segments, file_type, cur_filespace_name, cur_filespace_entries, new_filespace_entries, pg_system_filespace_entries): |
| self.segments = segments |
| self.file_type = file_type |
| self.cur_filespace_name = cur_filespace_name |
| self.cur_filespace_entries = cur_filespace_entries |
| self.new_filespace_entries = new_filespace_entries |
| self.pg_system_filespace_entries = pg_system_filespace_entries |
| |
| def execute(self): |
| logger.info('Rolling back filespace changes ...') |
| operations = [] |
| for seg in self.segments: |
| logger.debug('Creating RemoteOperation for segment %s' % seg) |
| peer_filespace_entry = get_peer_filespace_entry(self.cur_filespace_entries, seg.getSegmentDbId(), seg.getSegmentContentId(), self.segments) |
| if self.file_type == FileType.TRANSACTION_FILES: |
| #Move from new -> cur |
| operations.append(RemoteOperation(MoveTransFilespaceLocally(self.new_filespace_entries[seg.getSegmentDbId()], |
| self.cur_filespace_name, |
| self.cur_filespace_entries[seg.getSegmentDbId()], |
| peer_filespace_entry, |
| self.pg_system_filespace_entries[seg.getSegmentDbId()], |
| rollback=True |
| ), |
| seg.getSegmentHostName() |
| ), |
| ) |
| elif self.file_type == FileType.TEMPORARY_FILES: |
| operations.append(RemoteOperation(MoveTempFilespaceLocally(self.new_filespace_entries[seg.getSegmentDbId()], |
| self.cur_filespace_name, |
| self.cur_filespace_entries[seg.getSegmentDbId()], |
| peer_filespace_entry, |
| self.pg_system_filespace_entries[seg.getSegmentDbId()], |
| rollback=True |
| ), |
| seg.getSegmentHostName() |
| ), |
| ) |
| |
| logger.debug('Running remote operations in parallel') |
| ParallelOperation(operations, NUM_WORKERS).run() |
| |
| logger.debug('Checking results of parallel operations') |
| for operation in operations: |
| operation.get_ret() |
| |
| class GetMoveOperationList(Operation): |
| """ |
| @return: A list containing a set of RemoteOperations |
| that are executed on remote hosts via ParallelOperation. |
| """ |
| def __init__(self, segments, file_type, new_filespace_name, new_filespace_entries, cur_filespace_entries, pg_system_filespace_entries): |
| self.segments = segments |
| self.file_type = file_type |
| self.new_filespace_name = new_filespace_name |
| self.new_filespace_entries = new_filespace_entries |
| self.cur_filespace_entries = cur_filespace_entries |
| self.pg_system_filespace_entries = pg_system_filespace_entries |
| |
| def execute(self): |
| logger.info('Creating RemoteOperations list') |
| operations = [] |
| for seg in self.segments: |
| logger.debug('segment_dbid = %s' % seg.getSegmentDbId()) |
| logger.debug('segmenthostname = %s' % seg.getSegmentHostName()) |
| logger.debug(self.new_filespace_entries[seg.getSegmentDbId()]) |
| logger.debug(self.cur_filespace_entries[seg.getSegmentDbId()]) |
| peer_filespace_entry = get_peer_filespace_entry(self.new_filespace_entries, seg.getSegmentDbId(), seg.getSegmentContentId(), self.segments) |
| if self.file_type == FileType.TRANSACTION_FILES: |
| operations.append(RemoteOperation(MoveTransFilespaceLocally(self.cur_filespace_entries[seg.getSegmentDbId()], |
| self.new_filespace_name, |
| self.new_filespace_entries[seg.getSegmentDbId()], |
| peer_filespace_entry, |
| self.pg_system_filespace_entries[seg.getSegmentDbId()] |
| ), |
| seg.getSegmentHostName())) |
| elif self.file_type == FileType.TEMPORARY_FILES: |
| operations.append(RemoteOperation(MoveTempFilespaceLocally(self.cur_filespace_entries[seg.getSegmentDbId()], |
| self.new_filespace_name, |
| self.new_filespace_entries[seg.getSegmentDbId()], |
| peer_filespace_entry, |
| self.pg_system_filespace_entries[seg.getSegmentDbId()] |
| ), |
| seg.getSegmentHostName())) |
| return operations |
| |
| class MoveFileSpaceLocation(Operation): |
| """ |
| Move filespace location to a new location on dfs |
| """ |
| |
| def __init__(self, fsname, location, user, pswd): |
| self.user = user |
| self.pswd = pswd |
| self.fsname = fsname |
| self.location = location |
| self.dburl = dbconn.DbURL(username=self.user, password=self.pswd) |
| |
| def start_master_only(self): |
| logger.info('Starting Database in upgrade mode') |
| cmd = 'hawq start master -U upgrade' |
| result = local_ssh(cmd, logger) |
| if result != 0: |
| logger.error("Failed to start HAWQ Database in upgrade mode") |
| return result |
| |
| def stop_master_only(self): |
| logger.info('Stopping HAWQ master') |
| cmd = 'hawq stop master -a' |
| result = local_ssh(cmd, logger) |
| if result != 0: |
| logger.error("Failed to stop HAWQ master") |
| return result |
| |
| def stop_database(self): |
| logger.info('Stopping HAWQ cluster') |
| cmd = 'hawq stop cluster -a' |
| result = local_ssh(cmd, logger) |
| if result != 0: |
| logger.error("Failed to stop HAWQ master") |
| return result |
| |
| def check_database_stopped(self): |
| try: |
| mode = get_local_db_mode(MASTER_DATA_DIR) |
| logger.info('Database was started in %s mode' % mode) |
| except Exception, e: |
| logger.info('Database might already be stopped.') |
| return True |
| |
| return False |
| |
| def prepare_tasks(self): |
| conn = dbconn.connect(dbconn.DbURL(username=self.user, password=self.pswd), utility=True) |
| |
| try: |
| # Check if the filespace exist and can be changed, get its oid |
| filespaceRow = dbconn.execSQL(conn, ''' |
| SELECT oid, fsfsys |
| FROM pg_filespace |
| WHERE fsname = '%s'; |
| ''' % self.fsname) |
| |
| if filespaceRow.rowcount != 1: |
| logger.error('Filespace %s does not exist' % self.fsname) |
| return None |
| |
| filespace = filespaceRow.fetchall()[0] |
| |
| if filespace[1] == 0: |
| logger.error('Filespace "%s" is not created on distributed file system' % self.fsname) |
| return None |
| |
| self.fsoid = filespace[0] |
| filespaceRow.close() |
| |
| #remove last slash |
| if self.location[-1] is '/': |
| self.location = self.location[:-1] |
| |
| queries = [] |
| #generate sql for pg_filespace_entry |
| uri = self.location |
| queries.append(''' |
| UPDATE pg_filespace_entry |
| SET fselocation = '%s' |
| WHERE fsefsoid = %d; |
| ''' %(uri, self.fsoid)) |
| |
| #get gp_persistent_filespace |
| rows = dbconn.execSQL(conn, ''' |
| SELECT location |
| FROM gp_persistent_filespace_node |
| WHERE filespace_oid = %d; |
| ''' % self.fsoid) |
| |
| #generate sql for gp_persistent_filespace |
| for row in rows: |
| uri = '%s' % (self.location) |
| |
| paddinglen = len(row[0]) |
| if len(uri) > paddinglen: |
| logger.error('target location "%s" is too long' % self.location) |
| return None |
| |
| queries.append(''' |
| UPDATE gp_persistent_filespace_node |
| SET location = '%s' |
| WHERE filespace_oid = %d; |
| ''' %(uri.ljust(paddinglen), self.fsoid)) |
| |
| return queries |
| |
| finally: |
| conn.close() |
| |
| |
| def update_filespace_locations(self, queries): |
| conn = dbconn.connect(dbconn.DbURL(username=self.user, password=self.pswd), utility=True, allowSystemTableMods='all') |
| |
| try: |
| logger.info('Updating catalog...') |
| |
| for query in queries: |
| logger.debug('executing ...%s' % query) |
| executeUpdateOrInsert(conn, query, 1) |
| |
| dbconn.execSQL(conn, "COMMIT") |
| logger.info('catalog updated successfully') |
| |
| finally: |
| conn.close() |
| |
| def execute(self): |
| try: |
| # Disable Ctrl+C |
| signal.signal(signal.SIGINT, signal.SIG_IGN) |
| |
| # StopGPDB - Check for GPDB connections first and fail if connections exist |
| if not self.check_database_stopped(): |
| logger.error('stop Database first') |
| sys.exit(1) |
| |
| # StartGPDB in Master only mode |
| self.start_master_only() |
| |
| try: |
| if not CheckSuperUser(self.dburl).run(): |
| raise MoveFilespaceError('gpfilespace requires database superuser privileges.') |
| |
| queries = self.prepare_tasks(); |
| |
| if queries is None: |
| self.stop_master_only() |
| sys.exit(1) |
| |
| except Exception, e: |
| self.stop_master_only() |
| raise |
| |
| try: |
| self.update_filespace_locations(queries); |
| except Exception, e: |
| logger.fatal("Fatal error happened when updating catalog, PLEASE RESTORE MASTER DATA DIRECTORY!") |
| raise |
| finally: |
| # Bring GPDB Offline |
| self.stop_master_only() |
| |
| finally: |
| # Enable Ctrl+C |
| signal.signal(signal.SIGINT, signal.default_int_handler) |
| |
| class MoveFilespace(Operation): |
| """ |
| Main class which configures the database to use a custom filespace |
| for temporary and transaction files. |
| """ |
| |
| def __init__(self, new_filespace_name, file_type, user, pswd): |
| self.user = user |
| self.pswd = pswd |
| self.new_filespace_name = new_filespace_name |
| self.file_type = file_type |
| |
| def start_master_only(self): |
| logger.info('Starting Greenplum Database in master only mode') |
| cmd = GpStart('Start Greenplum Database in master only mode', masterOnly=True) |
| cmd.run() |
| if cmd.get_results().rc != 0: |
| logger.error('Failed to start Greenplum Database in master only mode.') |
| cmd.validate() |
| |
| def stop_master_only(self): |
| logger.info('Stopping Greenplum Database in master only mode') |
| cmd = GpStop('Stop Greenplum Database in master only mode', masterOnly=True) |
| cmd.run() |
| if cmd.get_results().rc != 0: |
| logger.error('Failed to stop Greenplum Database in master only mode.') |
| cmd.validate() |
| |
| def start_database(self): |
| logger.info('Starting Greenplum Database') |
| cmd = GpStart('Start Greenplum Database') |
| cmd.run() |
| if cmd.get_results().rc != 0: |
| logger.error('Failed to start Greenplum Database.') |
| cmd.validate() |
| |
| def stop_database(self): |
| logger.info('Stopping Greenplum Database') |
| cmd = GpStop('Stop Greenplum Databse') |
| cmd.run() |
| if cmd.get_results().rc != 0: |
| logger.error('Failed to stop Greenplum Database.') |
| cmd.validate() |
| |
| def check_database_stopped(self): |
| try: |
| mode = get_local_db_mode(MASTER_DATA_DIR) |
| logger.info('Database was started in %s mode' % mode) |
| except Exception, e: |
| logger.info('Database might already be stopped.') |
| return True |
| |
| return False |
| |
| def execute(self): |
| |
| try: |
| #Disable Ctrl+C |
| signal.signal(signal.SIGINT,signal.SIG_IGN) |
| |
| #StopGPDB - Check for GPDB connections first and fail if connections exist |
| if not self.check_database_stopped(): |
| self.stop_database() |
| |
| #Restart the system |
| self.start_database() |
| self.stop_database() |
| |
| #StartGPDB in Master only mode |
| self.start_master_only() |
| |
| try: |
| |
| if not CheckSuperUser(dbconn.DbURL(username=self.user, password=self.pswd)).run(): |
| raise MoveFilespaceError('gpfilespace requires database superuser privileges.') |
| |
| gparray = GpArray.initFromCatalog(dbconn.DbURL(), utility=True) |
| |
| #CheckFilespace - Check if new filespace exists |
| if not CheckFilespace(self.new_filespace_name).run(): |
| raise MoveFilespaceError('Filespace %s does not exist' % self.new_filespace_name) |
| |
| #CheckFilespaceIsSame - Check if filespace is different from the old one |
| if CheckFilespaceIsSame(gparray, self.new_filespace_name, self.file_type).run(): |
| raise MoveFilespaceError('Filespace %s is same as the current filespace' % self.new_filespace_name) |
| except Exception, e: |
| #Bring GPDB Offline |
| self.stop_master_only() |
| #Restart the database |
| self.start_database() |
| raise |
| |
| #Bring GPDB Offline |
| if 'UTILITY' == get_local_db_mode(MASTER_DATA_DIR): |
| self.stop_master_only() |
| else: |
| raise MoveFilespaceError('Database state is invalid.') |
| |
| if not CheckConnectivity(gparray).run(): |
| raise MoveFilespaceError('Failed connectivity test') |
| |
| logger.info('Obtaining current filespace information') |
| #Find the filespace directory used for each segment |
| #query using pg_filespace, pg_filespace_entry, gp_segment_configuration |
| #If gp_transaction/temporary flat files exist and is not empty, then we know |
| #the filespace being used. Otherwise, we assume that it is the pg_system |
| #filespace by default |
| |
| cur_filespace_entries = GetFilespaceEntriesDict(GetCurrentFilespaceEntries(gparray, |
| self.file_type).run()).run() |
| new_filespace_entries = GetFilespaceEntriesDict(GetFilespaceEntries(gparray, |
| self.new_filespace_name).run()).run() |
| pg_system_filespace_entries = GetFilespaceEntriesDict(GetFilespaceEntries(gparray, |
| PG_SYSTEM_FILESPACE).run()).run() |
| |
| cur_filespace_name = gparray.getFileSpaceName(int(cur_filespace_entries[1][0])) |
| |
| logger.info('Obtaining segment information ...') |
| segments = gparray.getDbList() |
| |
| #MoveTemp/Trans files |
| |
| try: |
| operations = GetMoveOperationList(segments, |
| self.file_type, |
| self.new_filespace_name, |
| new_filespace_entries, |
| cur_filespace_entries, |
| pg_system_filespace_entries |
| ).run() |
| except Exception, e: |
| raise MoveFilespaceError('Failed to create operations list. %s' % str(e)) |
| |
| logger.info('Moving %s filespace from %s to %s ...' % (FileType.lookup[self.file_type], cur_filespace_name, self.new_filespace_name)) |
| |
| ParallelOperation(operations, NUM_WORKERS).run() |
| |
| try: |
| for operation in operations: |
| operation.get_ret() |
| except Exception, e: |
| logger.error('Failed to move files on remote host. %s' % str(e)) |
| try: |
| RollBackFilespaceChanges(segments, self.file_type, cur_filespace_name, cur_filespace_entries, new_filespace_entries, pg_system_filespace_entries).run() |
| except Exception, e: |
| raise MoveFilespaceError('Rollback Failed !') |
| |
| #Bring GPDB online in normal mode |
| self.start_database() |
| |
| finally: |
| #Enable Ctrl+C |
| signal.signal(signal.SIGINT,signal.default_int_handler) |
| |
| class CreateTempDirectories(Operation): |
| def __init__(self, data_dir): |
| self.directories_file = os.path.join(data_dir, 'gp_temporary_files_directories') |
| self.created_path_history = {} |
| |
| def execute(self): |
| # The temp directories file does not exist, nothing to do |
| if not os.path.exists(self.directories_file): |
| logger.info('temporary directory is set to default path') |
| return True |
| |
| logger.info('create temporary directories recorded in %s' % self.directories_file) |
| try: |
| with open(self.directories_file) as file: |
| for line in file: |
| path = line.strip() |
| # check the parent directories' permission |
| if os.path.exists(path): |
| # raise OSError('create temporary directory failed for %s(path already exists)' % path) |
| break |
| MakeDirectory.local('create temporary direcotry', path) |
| self.created_path_history[path] = True |
| |
| except Exception, e: |
| logger.error('create temporary directory failed. %s' % str(e)) |
| # cleanup the work, don't complain if this is an OSError. |
| for path in self.created_path_history: |
| try: |
| if self.created_path_history[path]: |
| RemoveFiles.local('remove temporary directory', path) |
| except Exception, e: |
| logger.error('temporary directory cannot be removed. %s' % str(e)) |
| return e |
| |
| return True |
| |
| class DeleteTempDirectories(Operation): |
| def __init__(self, data_dir): |
| self.directories_file = os.path.join(data_dir, 'gp_temporary_files_directories') |
| |
| def execute(self): |
| ret = True |
| # The temp directories file does not exist, nothing to do |
| if not os.path.exists(self.directories_file): |
| logger.info('temporary directory is set to default path') |
| return True |
| |
| logger.info('remove temporary directories recorded in %s' % self.directories_file) |
| with open(self.directories_file) as file: |
| for line in file: |
| path = line.strip() |
| try: |
| RemoveFiles.local('temporary directory cleanup', path) |
| except Exception, e: |
| logger.error('remove temporary directory failed. %s' % str(e)) |
| # return the exception |
| ret = e |
| |
| return ret |
| |
| def create_temporary_directories(host, dir): |
| logger.info('Try to create temporary directories') |
| op = RemoteOperation(CreateTempDirectories(dir), host) |
| op.run() |
| |
| try: |
| op.get_ret() |
| except Exception, e: |
| logger.error('create temporary directory failed. %s' % str(e)) |
| raise e |
| |
| def remove_temporary_directories(host, dir): |
| logger.info('Try to remove temporary directories') |
| op = RemoteOperation(DeleteTempDirectories(dir), host) |
| op.run() |
| |
| try: |
| op.get_ret() |
| except Exception, e: |
| logger.error('remove temporary directory failed. %s' % str(e)) |
| raise e |
| |
| def template_temporary_directories(data_dir, contentid): |
| directories_file = os.path.join(data_dir, 'gp_temporary_files_directories') |
| if not os.path.exists(directories_file): |
| logger.info('default temporary direcotry was used') |
| return |
| |
| directories = [] |
| with open(directories_file) as file: |
| for line in file: |
| path = line.strip() |
| # remove the contentid |
| contentid_pos = path.rfind(str(contentid)) |
| if contentid_pos == -1: |
| raise Exception('contentid is wrong in %s' % directories_file) |
| path = path[0:contentid_pos] |
| directories.append(path) |
| |
| os.remove(directories_file) |
| with open(directories_file, 'w' ) as file: |
| for d in directories: |
| file.write(d + '\n') |
| |
| |
| def update_temporary_directories(data_dir, contentid): |
| directories_file = os.path.join(data_dir, 'gp_temporary_files_directories') |
| if not os.path.exists(directories_file): |
| logger.info('default temporary direcotry was used') |
| return |
| |
| logger.info('updating temporary directory file') |
| directories = [] |
| with open(directories_file) as file: |
| for line in file: |
| path = line.strip() |
| # append the contentid |
| path = path + str(contentid) |
| directories.append(path) |
| |
| os.remove(directories_file) |
| with open(directories_file, 'w' ) as file: |
| for d in directories: |
| file.write(d + '\n') |
| |
| # create the temporary directories |
| for d in directories: |
| if os.path.exists(d): |
| logger.warn('temporary directory %s is already exists' % d) |
| |
| MakeDirectory.local('create temporary direcotry', d) |
| |
