When we're backing up a PG cluster that is not running, we can't query it for information like the current restartpoint's WAL index, the current PG version, etc. Fortunately, we can use pg_controldata, which provides this information and doesn't require a running PG pro
| 7 | |
| 8 | |
| 9 | class PgControlDataParser(object): |
| 10 | """ |
| 11 | When we're backing up a PG cluster that is not |
| 12 | running, we can't query it for information like |
| 13 | the current restartpoint's WAL index, |
| 14 | the current PG version, etc. |
| 15 | |
| 16 | Fortunately, we can use pg_controldata, which |
| 17 | provides this information and doesn't require |
| 18 | a running PG process |
| 19 | """ |
| 20 | |
| 21 | def __init__(self, data_directory): |
| 22 | self.data_directory = data_directory |
| 23 | pg_config_proc = popen_sp([CONFIG_BIN], |
| 24 | stdout=PIPE) |
| 25 | output = pg_config_proc.communicate()[0].decode('utf-8') |
| 26 | for line in output.split('\n'): |
| 27 | parts = line.split('=') |
| 28 | if len(parts) != 2: |
| 29 | continue |
| 30 | key, val = [x.strip() for x in parts] |
| 31 | if key == 'BINDIR': |
| 32 | self._controldata_bin = os.path.join(val, CONTROLDATA_BIN) |
| 33 | elif key == 'VERSION': |
| 34 | self._pg_version = val |
| 35 | |
| 36 | def _read_controldata(self): |
| 37 | controldata_proc = popen_sp( |
| 38 | [self._controldata_bin, self.data_directory], stdout=PIPE) |
| 39 | stdout = controldata_proc.communicate()[0].decode('utf-8') |
| 40 | controldata = {} |
| 41 | for line in stdout.split('\n'): |
| 42 | split_values = line.split(':') |
| 43 | if len(split_values) == 2: |
| 44 | key, val = split_values |
| 45 | controldata[key.strip()] = val.strip() |
| 46 | return controldata |
| 47 | |
| 48 | def controldata_bin(self): |
| 49 | return self._controldata_bin |
| 50 | |
| 51 | def pg_version(self): |
| 52 | return self._pg_version |
| 53 | |
| 54 | def last_xlog_file_name_and_offset(self): |
| 55 | controldata = self._read_controldata() |
| 56 | last_checkpoint_offset = \ |
| 57 | controldata["Latest checkpoint's REDO location"] |
| 58 | current_timeline = controldata["Latest checkpoint's TimeLineID"] |
| 59 | x, offset = last_checkpoint_offset.split('/') |
| 60 | timeline = current_timeline.zfill(8) |
| 61 | wal = x.zfill(8) |
| 62 | offset = offset[0:2].zfill(8) |
| 63 | return { |
| 64 | 'file_name': ''.join([timeline, wal, offset]), |
| 65 | 'file_offset': offset.zfill(8)} |
no outgoing calls
no test coverage detected