294
// 5 - Update, add version info and make sure it's persisted and can be retrieved.
295
// 6 - Delete, no data persisted
297
>
clusterNameToPersist := "testing"
298
>
historyShardsToPersist := int32(43)
299
>
clusterIdToPersist := "12345"
300
>
clusterAddress := "cluster-address"
301
>
clusterHttpAddress := "cluster-http-address"
302
>
failoverVersionIncrement := int64(10)
303
>
initialFailoverVersion := int64(1)
304
>
305
>
// Case 1 - Get, mo data persisted
306
>
// Fetch the persisted values, there should be nothing on start.
307
>
// This doesn't error on no row found, but returns an empty record.
308
>
getResp, err := s.ClusterMetadataManager.GetClusterMetadata(s.ctx, &p.GetClusterMetadataRequest{ClusterName: clusterNameToPersist})
309
>
310
>
// Validate they match our initializations
311
>
s.NotNil(err)
312
>
s.IsType(&serviceerror.NotFound{}, err)
313
>
s.Nil(getResp)
314
>
315
>
// Case 2 - Init, no data persisted yet
316
>
// First commit, this should be persisted
317
>
initialResp, err := s.ClusterMetadataManager.SaveClusterMetadata(
318
>
s.ctx,
319
>
&p.SaveClusterMetadataRequest{
320
>
ClusterMetadata: &persistencespb.ClusterMetadata{
321
>
ClusterName: clusterNameToPersist,
322
>
HistoryShardCount: historyShardsToPersist,
323
>
ClusterId: clusterIdToPersist,
324
>
ClusterAddress: clusterAddress,
325
>
HttpAddress: clusterHttpAddress,
326
>
FailoverVersionIncrement: failoverVersionIncrement,
327
>
InitialFailoverVersion: initialFailoverVersion,
328
>
IsGlobalNamespaceEnabled: true,
329
>
IsConnectionEnabled: true,
330
>
}})
331
>
332
>
s.Nil(err)
333
>
s.True(initialResp) // request should be applied as this is first initialize
334
>
335
>
// Case 3 - Get, data persisted
336
>
// Fetch the persisted values
337
>
getResp, err = s.ClusterMetadataManager.GetClusterMetadata(s.ctx, &p.GetClusterMetadataRequest{ClusterName: clusterNameToPersist})
338
>
339
>
// Validate they match our initializations
340
>
s.Nil(err)
341
>
s.True(getResp != nil)
342
>
s.Equal(clusterNameToPersist, getResp.ClusterName)
343
>
s.Equal(historyShardsToPersist, getResp.HistoryShardCount)
344
>
s.Equal(clusterIdToPersist, getResp.ClusterId)
345
>
s.Equal(clusterAddress, getResp.ClusterAddress)
346
>
s.Equal(clusterHttpAddress, getResp.HttpAddress)
347
>
s.Equal(failoverVersionIncrement, getResp.FailoverVersionIncrement)
348
>
s.Equal(initialFailoverVersion, getResp.InitialFailoverVersion)
349
>
s.True(getResp.IsGlobalNamespaceEnabled)
350
>
s.True(getResp.IsConnectionEnabled)
351
>
352
>
// Case 4 - Init, data persisted
353
>
// Attempt to overwrite with new values
354
>
secondResp, err := s.ClusterMetadataManager.SaveClusterMetadata(s.ctx, &p.SaveClusterMetadataRequest{
355
>
ClusterMetadata: &persistencespb.ClusterMetadata{
356
>
ClusterName: clusterNameToPersist,
357
>
HistoryShardCount: int32(77),
358
>
}})
359
>
360
>
s.Nil(err)
361
>
s.False(secondResp) // Should not have applied, and should match values from first request
362
>
363
>
// Refetch persisted
364
>
getResp, err = s.ClusterMetadataManager.GetClusterMetadata(s.ctx, &p.GetClusterMetadataRequest{ClusterName: clusterNameToPersist})
365
>
366
>
// Validate they match our initial values
367
>
s.Nil(err)
368
>
s.NotNil(getResp)
369
>
s.Equal(clusterNameToPersist, getResp.ClusterName)
370
>
s.Equal(historyShardsToPersist, getResp.HistoryShardCount)
371
>
s.Equal(clusterIdToPersist, getResp.ClusterId)
372
>
s.Equal(clusterAddress, getResp.ClusterAddress)
373
>
s.Equal(clusterHttpAddress, getResp.HttpAddress)
374
>
s.Equal(failoverVersionIncrement, getResp.FailoverVersionIncrement)
375
>
s.Equal(initialFailoverVersion, getResp.InitialFailoverVersion)
376
>
s.True(getResp.IsGlobalNamespaceEnabled)
377
>
s.True(getResp.IsConnectionEnabled)
378
>
379
>
// Case 5 - Update version info
380
>
getResp.VersionInfo = &versionpb.VersionInfo{
381
>
Current: &versionpb.ReleaseInfo{
382
>
Version: "1.0",
383
>
},
384
>
}
385
>
thirdResp, err := s.ClusterMetadataManager.SaveClusterMetadata(s.ctx, &p.SaveClusterMetadataRequest{
386
>
ClusterMetadata: getResp.ClusterMetadata,
387
>
Version: getResp.Version,
388
>
})
389
>
s.Nil(err)
390
>
s.True(thirdResp)
391
>
getResp, err = s.ClusterMetadataManager.GetClusterMetadata(s.ctx, &p.GetClusterMetadataRequest{ClusterName: clusterNameToPersist})
392
>
s.Nil(err)
393
>
s.NotNil(getResp)
394
>
s.Equal("1.0", getResp.VersionInfo.Current.Version)
395
>
396
>
// Case 6 - Delete Cluster Metadata
397
>
err = s.ClusterMetadataManager.DeleteClusterMetadata(s.ctx, &p.DeleteClusterMetadataRequest{ClusterName: clusterNameToPersist})
398
>
s.Nil(err)
399
>
getResp, err = s.ClusterMetadataManager.GetClusterMetadata(s.ctx, &p.GetClusterMetadataRequest{ClusterName: clusterNameToPersist})
400
>
401
>
// Validate they match our initializations
402
>
s.NotNil(err)
403
>
s.IsType(&serviceerror.NotFound{}, err)
404
>
s.Nil(getResp)
405
>
406
>
// Case 7 - Update current cluster metadata
407
>
clusterNameToPersist = "active"
408
>
initialResp, err = s.ClusterMetadataManager.SaveClusterMetadata(
409
>
s.ctx,
410
>
&p.SaveClusterMetadataRequest{
411
>
ClusterMetadata: &persistencespb.ClusterMetadata{
412
>
ClusterName: clusterNameToPersist,
413
>
HistoryShardCount: historyShardsToPersist,
414
>
ClusterId: clusterIdToPersist,
415
>
ClusterAddress: clusterAddress,
416
>
HttpAddress: clusterHttpAddress,
417
>
FailoverVersionIncrement: failoverVersionIncrement,
418
>
InitialFailoverVersion: initialFailoverVersion,
419
>
IsGlobalNamespaceEnabled: true,
420
>
IsConnectionEnabled: true,
421
>
}})
422
>
s.Nil(err)
423
>
s.True(initialResp)
424
>
425
>
// Case 8 - Get, data persisted
426
>
// Fetch the persisted values
427
>
getResp, err = s.ClusterMetadataManager.GetClusterMetadata(s.ctx, &p.GetClusterMetadataRequest{ClusterName: clusterNameToPersist})
428
>
429
>
// Validate they match our initializations
430
>
s.Nil(err)
431
>
s.True(getResp != nil)
432
>
s.Equal(clusterNameToPersist, getResp.ClusterName)
433
>
s.Equal(historyShardsToPersist, getResp.HistoryShardCount)
434
>
s.Equal(clusterIdToPersist, getResp.ClusterId)
435
>
s.Equal(clusterAddress, getResp.ClusterAddress)
436
>
s.Equal(clusterHttpAddress, getResp.HttpAddress)
437
>
s.Equal(failoverVersionIncrement, getResp.FailoverVersionIncrement)
438
>
s.Equal(initialFailoverVersion, getResp.InitialFailoverVersion)
439
>
s.True(getResp.IsGlobalNamespaceEnabled)
440
>
s.True(getResp.IsConnectionEnabled)
441
>
442
>
// Case 9 - Update current cluster metadata
443
>
getResp.VersionInfo = &versionpb.VersionInfo{
444
>
Current: &versionpb.ReleaseInfo{
445
>
Version: "2.0",
446
>
},
447
>
}
448
>
applied, err := s.ClusterMetadataManager.SaveClusterMetadata(s.ctx, &p.SaveClusterMetadataRequest{
449
>
ClusterMetadata: getResp.ClusterMetadata,
450
>
Version: getResp.Version,
451
>
})
452
>
s.True(applied)
453
>
s.NoError(err)
454
>
455
>
// Case 10 - Get, data persisted
456
>
// Fetch the persisted values
457
>
getResp, err = s.ClusterMetadataManager.GetClusterMetadata(s.ctx, &p.GetClusterMetadataRequest{ClusterName: clusterNameToPersist})
458
>
s.NoError(err)
459
>
s.Equal("2.0", getResp.VersionInfo.Current.Version)
460
>
461
>
// Case 11 - List
462
>
_, err = s.ClusterMetadataManager.SaveClusterMetadata(
463
>
s.ctx,
464
>
&p.SaveClusterMetadataRequest{
465
>
ClusterMetadata: &persistencespb.ClusterMetadata{
466
>
ClusterName: clusterNameToPersist + "2",
467
>
HistoryShardCount: historyShardsToPersist,
468
>
ClusterId: clusterIdToPersist,
469
>
ClusterAddress: clusterAddress,
470
>
HttpAddress: clusterHttpAddress,
471
>
FailoverVersionIncrement: failoverVersionIncrement,
472
>
InitialFailoverVersion: initialFailoverVersion,
473
>
IsGlobalNamespaceEnabled: true,
474
>
IsConnectionEnabled: true,
475
>
}})
476
>
s.NoError(err)
477
>
478
>
resp, err := s.ClusterMetadataManager.ListClusterMetadata(s.ctx, &p.ListClusterMetadataRequest{PageSize: 1})
479
>
s.NoError(err)
480
>
s.Equal(1, len(resp.ClusterMetadata))
481
>
resp, err = s.ClusterMetadataManager.ListClusterMetadata(s.ctx, &p.ListClusterMetadataRequest{PageSize: 1, NextPageToken: resp.NextPageToken})
482
>
s.NoError(err)
483
>
s.Equal(1, len(resp.ClusterMetadata))
484
>
}