Skip to content

Commit cf6bce1

Browse files
ushitora-anqouushitora-anqou
authored
Merge pull request #273 from cybozu-go/add-max-import-jobs
controller: add --max-import-jobs to throttle concurrent import jobs
2 parents 249bcf7 + f98392d commit cf6bce1

5 files changed

Lines changed: 164 additions & 15 deletions

File tree

‎cmd/controller/main.go‎

Lines changed: 14 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,7 @@ var (
5555
role string
5656
mantleServiceEndpoint string
5757
maxExportJobs int
58+
maxImportJobs int
5859
maxUploadJobs int
5960
exportDataStorageClass string
6061
envSecret string
@@ -99,6 +100,8 @@ func init() {
99100
"this option is required and is interpreted as the address that the secondary mantle should listen to.")
100101
flags.IntVar(&maxExportJobs, "max-export-jobs", 8,
101102
"The maximum number of export jobs that can run simultaneously. If you set this to 0, there is no limit.")
103+
flags.IntVar(&maxImportJobs, "max-import-jobs", 8,
104+
"The maximum number of import jobs that can run simultaneously. If you set this to 0, there is no limit.")
102105
flags.IntVar(&maxUploadJobs, "max-upload-jobs", 8,
103106
"The maximum number of export jobs that can run simultaneously. If you set this to 0, there is no limit.")
104107
flags.StringVar(&exportDataStorageClass, "export-data-storage-class", "",
@@ -233,7 +236,11 @@ func checkCommandlineArgs() error {
233236
return nil
234237
}
235238

236-
func setupReconcilers(mgr manager.Manager, primarySettings *controller.PrimarySettings) error {
239+
func setupReconcilers(
240+
mgr manager.Manager,
241+
primarySettings *controller.PrimarySettings,
242+
secondarySettings *controller.SecondarySettings,
243+
) error {
237244
managedCephClusterID := os.Getenv("POD_NAMESPACE")
238245
if managedCephClusterID == "" {
239246
setupLog.Error(errors.New("POD_NAMESPACE is empty"), "POD_NAMESPACE is empty")
@@ -279,6 +286,7 @@ func setupReconcilers(mgr manager.Manager, primarySettings *controller.PrimarySe
279286
managedCephClusterID,
280287
role,
281288
primarySettings,
289+
secondarySettings,
282290
podImage,
283291
envSecret,
284292
&controller.ObjectStorageSettings{
@@ -348,7 +356,7 @@ func setupReconcilers(mgr manager.Manager, primarySettings *controller.PrimarySe
348356
}
349357

350358
func setupStandalone(mgr manager.Manager) error {
351-
return setupReconcilers(mgr, nil)
359+
return setupReconcilers(mgr, nil, nil)
352360
}
353361

354362
func setupPrimary(ctx context.Context, mgr manager.Manager, wg *sync.WaitGroup) error {
@@ -418,7 +426,7 @@ func setupPrimary(ctx context.Context, mgr manager.Manager, wg *sync.WaitGroup)
418426
ExportDataStorageClass: exportDataStorageClass,
419427
}
420428

421-
return setupReconcilers(mgr, primarySettings)
429+
return setupReconcilers(mgr, primarySettings, nil)
422430
}
423431

424432
func setupSecondary(ctx context.Context, mgr manager.Manager, wg *sync.WaitGroup, cancel context.CancelFunc) error {
@@ -476,7 +484,9 @@ func setupSecondary(ctx context.Context, mgr manager.Manager, wg *sync.WaitGroup
476484
serv.GracefulStop()
477485
}()
478486

479-
return setupReconcilers(mgr, nil)
487+
return setupReconcilers(mgr, nil, &controller.SecondarySettings{
488+
MaxImportJobs: maxImportJobs,
489+
})
480490
}
481491

482492
func subMain() error {

‎internal/controller/mantlebackup_controller.go‎

Lines changed: 28 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -117,7 +117,8 @@ type MantleBackupReconciler struct {
117117
ceph ceph.CephCmd
118118
managedCephClusterID string
119119
role string
120-
primarySettings *PrimarySettings // This should be non-nil if and only if role equals 'primary'.
120+
primarySettings *PrimarySettings // This should be non-nil if and only if role equals 'primary'.
121+
secondarySettings *SecondarySettings // This should be non-nil if and only if role equals 'secondary'.
121122
expireQueueCh chan event.GenericEvent
122123
podImage string
123124
envSecret string
@@ -134,6 +135,7 @@ func NewMantleBackupReconciler(
134135
managedCephClusterID,
135136
role string,
136137
primarySettings *PrimarySettings,
138+
secondarySettings *SecondarySettings,
137139
podImage string,
138140
envSecret string,
139141
objectStorageSettings *ObjectStorageSettings,
@@ -147,6 +149,7 @@ func NewMantleBackupReconciler(
147149
managedCephClusterID: managedCephClusterID,
148150
role: role,
149151
primarySettings: primarySettings,
152+
secondarySettings: secondarySettings,
150153
expireQueueCh: make(chan event.GenericEvent),
151154
podImage: podImage,
152155
envSecret: envSecret,
@@ -1345,8 +1348,8 @@ func (r *MantleBackupReconciler) handleCompletedExportJobs(ctx context.Context,
13451348
return r.handleCompletedJobsOfComponent(ctx, backup, labelComponentExportJob, MantleExportJobPrefix, nil)
13461349
}
13471350

1348-
func (r *MantleBackupReconciler) canNewExportJobBeCreated(ctx context.Context) (bool, error) {
1349-
if r.primarySettings.MaxExportJobs == 0 {
1351+
func (r *MantleBackupReconciler) canNewJobBeCreated(ctx context.Context, maxJobs int, component string) (bool, error) {
1352+
if maxJobs == 0 {
13501353
return true, nil
13511354
}
13521355

@@ -1355,10 +1358,10 @@ func (r *MantleBackupReconciler) canNewExportJobBeCreated(ctx context.Context) (
13551358
Namespace: r.managedCephClusterID,
13561359
LabelSelector: labels.SelectorFromSet(map[string]string{
13571360
"app.kubernetes.io/name": labelAppNameValue,
1358-
"app.kubernetes.io/component": labelComponentExportJob,
1361+
"app.kubernetes.io/component": component,
13591362
}),
13601363
}); err != nil {
1361-
return false, fmt.Errorf("failed to list export Jobs: %w", err)
1364+
return false, fmt.Errorf("failed to list %s Jobs: %w", component, err)
13621365
}
13631366

13641367
// exclude completed jobs
@@ -1369,11 +1372,19 @@ func (r *MantleBackupReconciler) canNewExportJobBeCreated(ctx context.Context) (
13691372
}
13701373
}
13711374

1372-
if count >= r.primarySettings.MaxExportJobs {
1373-
return false, nil
1375+
return count < maxJobs, nil
1376+
}
1377+
1378+
func (r *MantleBackupReconciler) canNewExportJobBeCreated(ctx context.Context) (bool, error) {
1379+
return r.canNewJobBeCreated(ctx, r.primarySettings.MaxExportJobs, labelComponentExportJob)
1380+
}
1381+
1382+
func (r *MantleBackupReconciler) canNewImportJobBeCreated(ctx context.Context) (bool, error) {
1383+
if r.secondarySettings == nil {
1384+
return true, nil
13741385
}
13751386

1376-
return true, nil
1387+
return r.canNewJobBeCreated(ctx, r.secondarySettings.MaxImportJobs, labelComponentImportJob)
13771388
}
13781389

13791390
func (r *MantleBackupReconciler) getPartNumRangeOfExpectedRunningUploadJobs(
@@ -2654,6 +2665,15 @@ func (r *MantleBackupReconciler) reconcileImportJob(
26542665
return requeueReconciliation(), nil
26552666
}
26562667

2668+
// check if a new import Job can be created
2669+
ok, err := r.canNewImportJobBeCreated(ctx)
2670+
if err != nil {
2671+
return ctrl.Result{}, fmt.Errorf("failed to check if a new import Job can be created: %w", err)
2672+
}
2673+
if !ok {
2674+
return requeueReconciliation(), nil
2675+
}
2676+
26572677
// create or update an import Job
26582678
if err := r.createOrUpdateImportJob(ctx, backup, snapshotTarget, partNum); err != nil {
26592679
return ctrl.Result{}, err

‎internal/controller/mantlebackup_controller_test.go‎

Lines changed: 117 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -200,6 +200,7 @@ var _ = Describe("MantleBackup controller", func() {
200200
resMgr.ClusterID,
201201
RoleStandalone,
202202
nil,
203+
nil,
203204
"dummy image",
204205
"",
205206
nil,
@@ -455,6 +456,7 @@ var _ = Describe("MantleBackup controller", func() {
455456
Client: grpcClient,
456457
ExportDataStorageClass: resMgr.StorageClassName,
457458
},
459+
nil,
458460
"dummy image",
459461
"dummy-secret-env",
460462
&ObjectStorageSettings{
@@ -979,7 +981,7 @@ var _ = Describe("prepareForDataSynchronization", func() {
979981
}
980982

981983
mbr := NewMantleBackupReconciler(ctrlClient,
982-
ctrlClient.Scheme(), "test", RolePrimary, nil, "dummy image", "", nil, nil, resource.MustParse("1Gi"))
984+
ctrlClient.Scheme(), "test", RolePrimary, nil, nil, "dummy image", "", nil, nil, resource.MustParse("1Gi"))
983985

984986
ret, err := mbr.prepareForDataSynchronization(context.Background(),
985987
backup, grpcClient)
@@ -1516,6 +1518,7 @@ var _ = Describe("export and upload", func() {
15161518
ExportDataStorageClass: resMgr.StorageClassName,
15171519
MaxExportJobs: 1,
15181520
},
1521+
nil,
15191522
"dummy image",
15201523
"dummy-secret",
15211524
&ObjectStorageSettings{},
@@ -1606,6 +1609,7 @@ var _ = Describe("export and upload", func() {
16061609
ExportDataStorageClass: resMgr.StorageClassName,
16071610
MaxExportJobs: 1,
16081611
},
1612+
nil,
16091613
"dummy image",
16101614
"",
16111615
nil,
@@ -1759,6 +1763,7 @@ var _ = Describe("import", func() {
17591763
nsController,
17601764
RoleSecondary,
17611765
nil,
1766+
nil,
17621767
"dummy-image",
17631768
"dummy-env-secret",
17641769
&ObjectStorageSettings{},
@@ -1777,6 +1782,32 @@ var _ = Describe("import", func() {
17771782
}
17781783
})
17791784

1785+
createAndReconcileImportJob := func(
1786+
ctx SpecContext,
1787+
mbr *MantleBackupReconciler,
1788+
name, ns string,
1789+
) {
1790+
GinkgoHelper()
1791+
1792+
target, err := createMantleBackupUsingDummyPVC(ctx, name, ns)
1793+
Expect(err).NotTo(HaveOccurred())
1794+
1795+
err = updateStatus(ctx, k8sClient, target, func() error {
1796+
target.Status.SnapSize = ptr.To(int64(testutil.FakeRBDSnapshotSize))
1797+
transferPartSize := resource.MustParse("1Gi")
1798+
target.Status.TransferPartSize = &transferPartSize
1799+
1800+
return nil
1801+
})
1802+
Expect(err).NotTo(HaveOccurred())
1803+
1804+
snapshotTarget, err := getSnapshotTargetByDummyMB(target)
1805+
Expect(err).NotTo(HaveOccurred())
1806+
1807+
_, err = mbr.reconcileImportJob(ctx, target, snapshotTarget, -1)
1808+
Expect(err).NotTo(HaveOccurred())
1809+
}
1810+
17801811
DescribeTable("isExportDataAlreadyUploaded: inputs and outputs",
17811812
func(ctx SpecContext, gotExist, gotError, expectUploaded, expectError bool) {
17821813
mockObjectStorage.EXPECT().Exists(gomock.Any(), gomock.Eq("name-uid-0.bin")).DoAndReturn(
@@ -1858,6 +1889,88 @@ var _ = Describe("import", func() {
18581889
})
18591890
})
18601891

1892+
Context("import job throttling", func() {
1893+
It("should throttle import jobs correctly", func(ctx SpecContext) {
1894+
nsController = resMgr.CreateNamespace()
1895+
mbr = NewMantleBackupReconciler(
1896+
k8sClient,
1897+
scheme.Scheme,
1898+
nsController,
1899+
RoleSecondary,
1900+
nil,
1901+
&SecondarySettings{
1902+
MaxImportJobs: 1,
1903+
},
1904+
"dummy-image",
1905+
"dummy-env-secret",
1906+
&ObjectStorageSettings{},
1907+
nil,
1908+
resource.MustParse("1Gi"),
1909+
)
1910+
mbr.objectStorageClient = mockObjectStorage
1911+
mbr.ceph = testutil.NewFakeRBD()
1912+
1913+
getNumOfImportJobs := func(ns string) (int, error) {
1914+
var jobs batchv1.JobList
1915+
err := k8sClient.List(ctx, &jobs, &client.ListOptions{
1916+
Namespace: ns,
1917+
LabelSelector: labels.SelectorFromSet(map[string]string{
1918+
"app.kubernetes.io/name": labelAppNameValue,
1919+
"app.kubernetes.io/component": labelComponentImportJob,
1920+
}),
1921+
})
1922+
1923+
return len(jobs.Items), err
1924+
}
1925+
1926+
mockObjectStorage.EXPECT().Exists(gomock.Any(), gomock.Any()).
1927+
Return(true, nil).AnyTimes()
1928+
1929+
// create 5 different MantleBackup resources and call reconcileImportJob for each of them
1930+
for i := range 5 {
1931+
createAndReconcileImportJob(ctx, mbr, fmt.Sprintf("target1-%d", i), ns)
1932+
}
1933+
1934+
// make sure that only 1 Job is created
1935+
Consistently(ctx, func(g Gomega) error {
1936+
numJobs, err := getNumOfImportJobs(nsController)
1937+
g.Expect(err).NotTo(HaveOccurred())
1938+
g.Expect(numJobs).To(Equal(1))
1939+
1940+
return nil
1941+
}, "1s").Should(Succeed())
1942+
1943+
// make sure that another mantle-controller existing in a different namespace can create an import Job
1944+
nsController2 := resMgr.CreateNamespace()
1945+
mbr2 := NewMantleBackupReconciler(
1946+
k8sClient,
1947+
scheme.Scheme,
1948+
nsController2,
1949+
RoleSecondary,
1950+
nil,
1951+
&SecondarySettings{
1952+
MaxImportJobs: 1,
1953+
},
1954+
"dummy-image",
1955+
"dummy-env-secret",
1956+
&ObjectStorageSettings{},
1957+
nil,
1958+
resource.MustParse("1Gi"),
1959+
)
1960+
mbr2.objectStorageClient = mockObjectStorage
1961+
mbr2.ceph = testutil.NewFakeRBD()
1962+
ns2 := resMgr.CreateNamespace()
1963+
createAndReconcileImportJob(ctx, mbr2, "target2", ns2)
1964+
Eventually(ctx, func(g Gomega) error {
1965+
numJobs, err := getNumOfImportJobs(nsController2)
1966+
g.Expect(err).NotTo(HaveOccurred())
1967+
g.Expect(numJobs).To(Equal(1))
1968+
1969+
return nil
1970+
}).Should(Succeed())
1971+
})
1972+
})
1973+
18611974
Context("markSecondarySnapshotCaptured", func() {
18621975
It("should work correctly", func(ctx SpecContext) {
18631976
backup, err := createMantleBackupUsingDummyPVC(ctx, "target", ns)
@@ -2580,7 +2693,7 @@ var _ = Describe("MantleBackupReconciler", func() {
25802693
}
25812694

25822695
It("setup", func(ctx SpecContext) {
2583-
reconciler = NewMantleBackupReconciler(nil, nil, "", "", nil, "", "", nil, nil, resource.Quantity{})
2696+
reconciler = NewMantleBackupReconciler(nil, nil, "", "", nil, nil, "", "", nil, nil, resource.Quantity{})
25842697
cephCmd = testutil.NewFakeRBD()
25852698
reconciler.ceph = cephCmd
25862699
})
@@ -2617,7 +2730,7 @@ var _ = Describe("MantleBackupReconciler", func() {
26172730
var cephCmd *testutil.FakeRBD
26182731

26192732
It("setup", func(ctx SpecContext) {
2620-
reconciler = NewMantleBackupReconciler(nil, nil, "", "", nil, "", "", nil, nil, resource.Quantity{})
2733+
reconciler = NewMantleBackupReconciler(nil, nil, "", "", nil, nil, "", "", nil, nil, resource.Quantity{})
26212734
cephCmd = testutil.NewFakeRBD()
26222735
reconciler.ceph = cephCmd
26232736
})
@@ -2834,6 +2947,7 @@ set -eux -o pipefail
28342947
podNamespace,
28352948
"", // unused
28362949
nil, // unused
2950+
nil, // unused
28372951
podImage,
28382952
"", // unused
28392953
nil, // unused

‎internal/controller/mantlerestore_controller_test.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@ func (test *mantleRestoreControllerUnitTest) setupEnv() {
5858
resMgr.ClusterID,
5959
RoleStandalone,
6060
nil,
61+
nil,
6162
"dummy image",
6263
"",
6364
nil,

‎internal/controller/replication.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,10 @@ type PrimarySettings struct {
3232
ExportDataStorageClass string
3333
}
3434

35+
type SecondarySettings struct {
36+
MaxImportJobs int
37+
}
38+
3539
type SecondaryServer struct {
3640
client client.Client
3741
proto.UnimplementedMantleServiceServer

0 commit comments

Comments
 (0)