Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commitc7a751d

Browse files
committed
[Issue#449] subdirectories for WAL archive
1 parent5b6ca62 commitc7a751d

File tree

7 files changed

+196
-29
lines changed

7 files changed

+196
-29
lines changed

‎src/archive.c

Lines changed: 126 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -15,14 +15,17 @@
1515

1616
staticintpush_file_internal_uncompressed(constchar*wal_file_name,constchar*pg_xlog_dir,
1717
constchar*archive_dir,booloverwrite,boolno_sync,
18-
uint32archive_timeout);
18+
uint32archive_timeout,xlogFileTypetype);
1919
#ifdefHAVE_LIBZ
2020
staticintpush_file_internal_gz(constchar*wal_file_name,constchar*pg_xlog_dir,
21-
constchar*archive_dir,booloverwrite,boolno_sync,
22-
intcompress_level,uint32archive_timeout);
21+
constchar*archive_dir,booloverwrite,boolno_sync,
22+
intcompress_level,uint32archive_timeout,xlogFileTypetype);
2323
#endif
2424
staticvoid*push_files(void*arg);
2525
staticvoid*get_files(void*arg);
26+
staticbool
27+
get_wal_file_wrapper(constchar*filename,constchar*archive_root_dir,
28+
constchar*to_fullpath,boolprefetch_mode);
2629
staticboolget_wal_file(constchar*filename,constchar*from_path,constchar*to_path,
2730
boolprefetch_mode);
2831
staticintget_wal_file_internal(constchar*from_path,constchar*to_path,FILE*out,
@@ -89,8 +92,9 @@ typedef struct
8992

9093
typedefstructWALSegno
9194
{
92-
charname[MAXFNAMELEN];
93-
volatilepg_atomic_flaglock;
95+
charname[MAXFNAMELEN];
96+
volatilepg_atomic_flaglock;
97+
xlogFileTypetype;
9498
}WALSegno;
9599

96100
staticintpush_file(WALSegno*xlogfile,constchar*archive_status_dir,
@@ -102,6 +106,28 @@ static int push_file(WALSegno *xlogfile, const char *archive_status_dir,
102106
staticparray*setup_push_filelist(constchar*archive_status_dir,
103107
constchar*first_file,intbatch_size);
104108

109+
staticxlogFileType
110+
get_xlogFileType(constchar*filename)
111+
{
112+
113+
ifIsXLogFileName(filename)
114+
returnSEGMENT;
115+
116+
elseifIsPartialXLogFileName(filename)
117+
returnPARTIAL_SEGMENT;
118+
119+
elseifIsBackupHistoryFileName(filename)
120+
returnBACKUP_HISTORY_FILE;
121+
122+
elseifIsTLHistoryFileName(filename)
123+
returnHISTORY_FILE;
124+
125+
elseifIsBackupHistoryFileName(filename)
126+
returnBACKUP_HISTORY_FILE;
127+
128+
returnUNKNOWN;
129+
}
130+
105131
/*
106132
* At this point, we already done one roundtrip to archive server
107133
* to get instance config.
@@ -185,6 +211,8 @@ do_archive_push(InstanceState *instanceState, InstanceConfig *instance, char *wa
185211
parray_num(batch_files),batch_size,
186212
is_compress ?"zlib" :"none");
187213

214+
/* TODO: create subdirectories here, not in internal functions */
215+
188216
num_threads=n_threads;
189217

190218
/* Single-thread push
@@ -366,12 +394,12 @@ push_file(WALSegno *xlogfile, const char *archive_status_dir,
366394
if (!is_compress)
367395
rc=push_file_internal_uncompressed(xlogfile->name,pg_xlog_dir,
368396
archive_dir,overwrite,no_sync,
369-
archive_timeout);
397+
archive_timeout,xlogfile->type);
370398
#ifdefHAVE_LIBZ
371399
else
372400
rc=push_file_internal_gz(xlogfile->name,pg_xlog_dir,archive_dir,
373401
overwrite,no_sync,compress_level,
374-
archive_timeout);
402+
archive_timeout,xlogfile->type);
375403
#endif
376404

377405
/* take '--no-ready-rename' flag into account */
@@ -408,13 +436,14 @@ push_file(WALSegno *xlogfile, const char *archive_status_dir,
408436
int
409437
push_file_internal_uncompressed(constchar*wal_file_name,constchar*pg_xlog_dir,
410438
constchar*archive_dir,booloverwrite,boolno_sync,
411-
uint32archive_timeout)
439+
uint32archive_timeout,xlogFileTypetype)
412440
{
413441
FILE*in=NULL;
414442
intout=-1;
415443
char*buf=pgut_malloc(OUT_BUF_SIZE);/* 1MB buffer */
416444
charfrom_fullpath[MAXPGPATH];
417445
charto_fullpath[MAXPGPATH];
446+
chararchive_subdir[MAXPGPATH];
418447
/* partial handling */
419448
structstatst;
420449
charto_fullpath_part[MAXPGPATH];
@@ -427,8 +456,16 @@ push_file_internal_uncompressed(const char *wal_file_name, const char *pg_xlog_d
427456
/* from path */
428457
join_path_components(from_fullpath,pg_xlog_dir,wal_file_name);
429458
canonicalize_path(from_fullpath);
459+
460+
/* calculate subdir in WAL archive */
461+
get_archive_subdir(archive_subdir,archive_dir,wal_file_name,type);
462+
463+
/* create subdirectory */
464+
if (fio_mkdir(archive_subdir,DIR_PERMISSION,FIO_BACKUP_HOST)!=0)
465+
elog(ERROR,"Cannot create subdirectory in WAL archive: '%s'",archive_subdir);
466+
430467
/* to path */
431-
join_path_components(to_fullpath,archive_dir,wal_file_name);
468+
join_path_components(to_fullpath,archive_subdir,wal_file_name);
432469
canonicalize_path(to_fullpath);
433470

434471
/* Open source file for read */
@@ -647,14 +684,15 @@ push_file_internal_uncompressed(const char *wal_file_name, const char *pg_xlog_d
647684
int
648685
push_file_internal_gz(constchar*wal_file_name,constchar*pg_xlog_dir,
649686
constchar*archive_dir,booloverwrite,boolno_sync,
650-
intcompress_level,uint32archive_timeout)
687+
intcompress_level,uint32archive_timeout,xlogFileTypetype)
651688
{
652689
FILE*in=NULL;
653690
gzFileout=NULL;
654691
char*buf=pgut_malloc(OUT_BUF_SIZE);
655692
charfrom_fullpath[MAXPGPATH];
656693
charto_fullpath[MAXPGPATH];
657694
charto_fullpath_gz[MAXPGPATH];
695+
chararchive_subdir[MAXPGPATH];
658696

659697
/* partial handling */
660698
structstatst;
@@ -669,8 +707,16 @@ push_file_internal_gz(const char *wal_file_name, const char *pg_xlog_dir,
669707
/* from path */
670708
join_path_components(from_fullpath,pg_xlog_dir,wal_file_name);
671709
canonicalize_path(from_fullpath);
710+
711+
/* calculate subdir in WAL archive */
712+
get_archive_subdir(archive_subdir,archive_dir,wal_file_name,type);
713+
714+
/* create subdirectory */
715+
if (fio_mkdir(archive_subdir,DIR_PERMISSION,FIO_BACKUP_HOST)!=0)
716+
elog(ERROR,"Cannot create subdirectory in WAL archive: '%s'",archive_subdir);
717+
672718
/* to path */
673-
join_path_components(to_fullpath,archive_dir,wal_file_name);
719+
join_path_components(to_fullpath,archive_subdir,wal_file_name);
674720
canonicalize_path(to_fullpath);
675721

676722
/* destination file with .gz suffix */
@@ -940,15 +986,17 @@ setup_push_filelist(const char *archive_status_dir, const char *first_file,
940986
{
941987
inti;
942988
WALSegno*xlogfile=NULL;
943-
parray*status_files=NULL;
944-
parray*batch_files=parray_new();
989+
parray*status_files=NULL;
990+
parray*batch_files=parray_new();
945991

946992
/* guarantee that first filename is in batch list */
947993
xlogfile=palloc(sizeof(WALSegno));
948994
pg_atomic_init_flag(&xlogfile->lock);
949995
snprintf(xlogfile->name,MAXFNAMELEN,"%s",first_file);
950996
parray_append(batch_files,xlogfile);
951997

998+
xlogfile->type=get_xlogFileType(xlogfile->name);
999+
9521000
if (batch_size<2)
9531001
returnbatch_files;
9541002

@@ -980,6 +1028,8 @@ setup_push_filelist(const char *archive_status_dir, const char *first_file,
9801028
pg_atomic_init_flag(&xlogfile->lock);
9811029

9821030
snprintf(xlogfile->name,MAXFNAMELEN,"%s",filename);
1031+
1032+
xlogfile->type=get_xlogFileType(xlogfile->name);
9831033
parray_append(batch_files,xlogfile);
9841034

9851035
if (parray_num(batch_files) >=batch_size)
@@ -1048,7 +1098,7 @@ do_archive_get(InstanceState *instanceState, InstanceConfig *instance, const cha
10481098

10491099
/* full filepath to WAL file in archive directory.
10501100
* $BACKUP_PATH/wal/instance_name/000000010000000000000001 */
1051-
join_path_components(backup_wal_file_path,instanceState->instance_wal_subdir_path,wal_file_name);
1101+
//join_path_components(backup_wal_file_path, instanceState->instance_wal_subdir_path, wal_file_name);
10521102

10531103
INSTR_TIME_SET_CURRENT(start_time);
10541104
if (num_threads>batch_size)
@@ -1177,7 +1227,7 @@ do_archive_get(InstanceState *instanceState, InstanceConfig *instance, const cha
11771227

11781228
while (fail_count<3)
11791229
{
1180-
if (get_wal_file(wal_file_name,backup_wal_file_path,absolute_wal_file_path, false))
1230+
if (get_wal_file_wrapper(wal_file_name,instanceState->instance_wal_subdir_path,absolute_wal_file_path, false))
11811231
{
11821232
fail_count=0;
11831233
elog(INFO,"pg_probackup archive-get copied WAL file %s",wal_file_name);
@@ -1260,7 +1310,7 @@ uint32 run_wal_prefetch(const char *prefetch_dir, const char *archive_dir,
12601310
/* It is ok, maybe requested batch is greater than the number of available
12611311
* files in the archive
12621312
*/
1263-
if (!get_wal_file(xlogfile->name,from_fullpath,to_fullpath, true))
1313+
if (!get_wal_file_wrapper(xlogfile->name,archive_dir,to_fullpath, true))
12641314
{
12651315
elog(LOG,"Thread [%d]: Failed to prefetch WAL segment %s",0,xlogfile->name);
12661316
break;
@@ -1334,7 +1384,7 @@ get_files(void *arg)
13341384
join_path_components(from_fullpath,args->archive_dir,xlogfile->name);
13351385
join_path_components(to_fullpath,args->prefetch_dir,xlogfile->name);
13361386

1337-
if (!get_wal_file(xlogfile->name,from_fullpath,to_fullpath, true))
1387+
if (!get_wal_file_wrapper(xlogfile->name,args->archive_dir,to_fullpath, true))
13381388
{
13391389
/* It is ok, maybe requested batch is greater than the number of available
13401390
* files in the archive
@@ -1353,6 +1403,38 @@ get_files(void *arg)
13531403
returnNULL;
13541404
}
13551405

1406+
/*
1407+
* First we try to copy from WAL archive subdirectory:
1408+
* Failing that, try WAL archive root directory
1409+
*/
1410+
bool
1411+
get_wal_file_wrapper(constchar*filename,constchar*archive_root_dir,
1412+
constchar*to_fullpath,boolprefetch_mode)
1413+
{
1414+
boolsuccess= false;
1415+
chararchive_subdir[MAXPGPATH];
1416+
charfrom_fullpath[MAXPGPATH];
1417+
xlogFileTypetype=get_xlogFileType(filename);
1418+
1419+
if (type==SEGMENT||type==PARTIAL_SEGMENT||type==BACKUP_HISTORY_FILE)
1420+
{
1421+
/* first try subdir ... */
1422+
get_archive_subdir(archive_subdir,archive_root_dir,filename,type);
1423+
join_path_components(from_fullpath,archive_subdir,filename);
1424+
1425+
success=get_wal_file(filename,from_fullpath,to_fullpath,prefetch_mode);
1426+
}
1427+
1428+
if (!success)
1429+
{
1430+
/* ... fallback to archive dir for backward compatibility purposes */
1431+
join_path_components(from_fullpath,archive_root_dir,filename);
1432+
success=get_wal_file(filename,from_fullpath,to_fullpath,prefetch_mode);
1433+
}
1434+
1435+
returnsuccess;
1436+
}
1437+
13561438
/*
13571439
* Copy WAL segment from archive catalog to pgdata with possible decompression.
13581440
* When running in prefetch mode, we should not error out.
@@ -1755,3 +1837,30 @@ uint32 maintain_prefetch(const char *prefetch_dir, XLogSegNo first_segno, uint32
17551837

17561838
returnn_files;
17571839
}
1840+
1841+
/* Calculate subdir path in WAL archive directory. Example:
1842+
* 000000010000000200000013 -> 00000002
1843+
*/
1844+
void
1845+
get_archive_subdir(char*archive_subdir,constchar*archive_dir,constchar*wal_file_name,xlogFileTypetype)
1846+
{
1847+
if (type==SEGMENT||type==PARTIAL_SEGMENT||type==BACKUP_HISTORY_FILE)
1848+
{
1849+
intrc=0;
1850+
chartli[MAXFNAMELEN];
1851+
charlog[MAXFNAMELEN];
1852+
charsuffix[MAXFNAMELEN];
1853+
1854+
rc=sscanf(wal_file_name,"%08s%08s%s",
1855+
(char*)&tli, (char*)&log, (char*)&suffix);
1856+
1857+
if (rc==3)
1858+
{
1859+
join_path_components(archive_subdir,archive_dir,log);
1860+
return;
1861+
}
1862+
}
1863+
1864+
/* for all other files just use root directory of WAL archive */
1865+
strcpy(archive_subdir,archive_dir);
1866+
}

‎src/backup.c

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1238,6 +1238,7 @@ wait_wal_lsn(const char *wal_segment_dir, XLogRecPtr target_lsn, bool is_start_l
12381238
{
12391239
XLogSegNotargetSegNo;
12401240
charwal_segment_path[MAXPGPATH],
1241+
wal_segment_subdir[MAXPGPATH],
12411242
wal_segment[MAXFNAMELEN];
12421243
boolfile_exists= false;
12431244
uint32try_count=0,
@@ -1255,7 +1256,13 @@ wait_wal_lsn(const char *wal_segment_dir, XLogRecPtr target_lsn, bool is_start_l
12551256
GetXLogFileName(wal_segment,tli,targetSegNo,
12561257
instance_config.xlog_seg_size);
12571258

1258-
join_path_components(wal_segment_path,wal_segment_dir,wal_segment);
1259+
// obtain WAL archive subdir for ARCHIVE backup
1260+
if (!in_stream_dir)
1261+
get_archive_subdir(wal_segment_subdir,wal_segment_dir,wal_segment,SEGMENT);
1262+
else
1263+
strcpy(wal_segment_subdir,wal_segment_dir);
1264+
1265+
join_path_components(wal_segment_path,wal_segment_subdir,wal_segment);
12591266
/*
12601267
* In pg_start_backup we wait for 'target_lsn' in 'pg_wal' directory if it is
12611268
* stream and non-page backup. Page backup needs archived WAL files, so we

‎src/catalog.c

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1628,6 +1628,7 @@ catalog_get_timelines(InstanceState *instanceState, InstanceConfig *instance)
16281628
elseif (strcmp(suffix,"gz")!=0)
16291629
{
16301630
elog(WARNING,"unexpected WAL file name \"%s\"",file->name);
1631+
// TODO: free file
16311632
continue;
16321633
}
16331634
}
@@ -1724,8 +1725,18 @@ catalog_get_timelines(InstanceState *instanceState, InstanceConfig *instance)
17241725
parray_walk(timelines,pfree);
17251726
parray_free(timelines);
17261727
}
1728+
/* add WAL archive subdirectories to filelist (used only in delete) */
1729+
elseif (S_ISDIR(file->mode)&&strspn(file->rel_path,"0123456789ABCDEF")==8)
1730+
{
1731+
if (instanceState->wal_archive_subdirs==NULL)
1732+
instanceState->wal_archive_subdirs=parray_new();
1733+
parray_append(instanceState->wal_archive_subdirs,file);
1734+
}
17271735
else
1736+
{
17281737
elog(WARNING,"unexpected WAL file name \"%s\"",file->name);
1738+
// TODO: free file
1739+
}
17291740
}
17301741

17311742
/* save information about backups belonging to each timeline */

‎src/delete.c

Lines changed: 23 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -932,14 +932,15 @@ delete_walfiles_in_tli(InstanceState *instanceState, XLogRecPtr keep_lsn, timeli
932932
if (interrupted)
933933
elog(ERROR,"interrupted during WAL archive purge");
934934

935-
/* Any segment equal or greater than EndSegNo must be kept
935+
/*
936+
* Any segment equal or greater than EndSegNo must be kept
936937
* unless it`s a 'purge all' scenario.
937938
*/
938939
if (purge_all||wal_file->segno<OldestToKeepSegNo)
939940
{
940941
charwal_fullpath[MAXPGPATH];
941942

942-
join_path_components(wal_fullpath,instanceState->instance_wal_subdir_path,wal_file->file.name);
943+
join_path_components(wal_fullpath,instanceState->instance_wal_subdir_path,wal_file->file.rel_path);
943944

944945
/* save segment from purging */
945946
if (instance_config.wal_depth >=0&&wal_file->keep)
@@ -970,6 +971,26 @@ delete_walfiles_in_tli(InstanceState *instanceState, XLogRecPtr keep_lsn, timeli
970971

971972
wal_deleted= true;
972973
}
974+
975+
//TODO: cleanup
976+
}
977+
978+
/* Remove empty subdirectories */
979+
if (!instanceState->wal_archive_subdirs)
980+
return;
981+
982+
for (i=0;i<parray_num(instanceState->wal_archive_subdirs);i++)
983+
{
984+
charfullpath[MAXPGPATH];
985+
pgFile*file= (pgFile*)parray_get(instanceState->wal_archive_subdirs,i);
986+
987+
join_path_components(fullpath,instanceState->instance_wal_subdir_path,file->name);
988+
989+
if (dir_is_empty(fullpath,FIO_LOCAL_HOST))
990+
pgFileDelete(file->mode,fullpath);
991+
992+
pgFileFree(file);
993+
// TODO: maintain instanceState->wal_archive_subdirs
973994
}
974995
}
975996

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp