createTestBuckets creates a new set of integration test buckets and pushes them into the readier queue.
(ctx context.Context, numBuckets, bucketQuotaMB int, bucketInitFunc TBPBucketInitFunc, parallelBucketInit bool)
| 611 | |
| 612 | // createTestBuckets creates a new set of integration test buckets and pushes them into the readier queue. |
| 613 | func (tbp *TestBucketPool) createTestBuckets(ctx context.Context, numBuckets, bucketQuotaMB int, bucketInitFunc TBPBucketInitFunc, parallelBucketInit bool) { |
| 614 | |
| 615 | defer close(tbp.bucketCreationDoneChan) |
| 616 | start := time.Now() |
| 617 | wg := sync.WaitGroup{} |
| 618 | wg.Add(numBuckets) |
| 619 | |
| 620 | // Append a timestamp to all of the bucket names to ensure uniqueness across a single package. |
| 621 | // Not strictly required, but can help to prevent (index) resources from being incorrectly reused on the server side for recently deleted buckets. |
| 622 | bucketNameTimestamp := time.Now().UnixNano() |
| 623 | |
| 624 | // create required number of buckets (skipping any already existing ones) |
| 625 | for i := 0; i < numBuckets; i++ { |
| 626 | bucketName := fmt.Sprintf(tbpBucketNameFormat, tbpBucketNamePrefix, i, bucketNameTimestamp) |
| 627 | ctx := BucketNameCtx(ctx, bucketName) |
| 628 | |
| 629 | bucketInit := func() { |
| 630 | defer wg.Done() |
| 631 | ctx := BucketNameCtx(ctx, bucketName) |
| 632 | |
| 633 | tbp.Logf(ctx, "Creating new test bucket") |
| 634 | err := tbp.cluster.insertBucket(bucketName, bucketQuotaMB, tbp.xdcrConflictResolutionStrategy) |
| 635 | if ctx.Err() != nil { |
| 636 | return |
| 637 | } else if err != nil { |
| 638 | tbp.Fatalf(ctx, "Couldn't create test bucket: %v", err) |
| 639 | } |
| 640 | |
| 641 | bucket, err := tbp.cluster.openTestBucket(ctx, tbpBucketName(bucketName), waitForReadyBucketTimeout) |
| 642 | if err != nil { |
| 643 | tbp.Fatalf(ctx, "Timed out trying to open new bucket: %v", err) |
| 644 | } |
| 645 | |
| 646 | tbp.createCollections(ctx, bucket) |
| 647 | |
| 648 | tbp.emptyPreparedStatements(ctx, bucket) |
| 649 | if tbp.cluster.ee { |
| 650 | tbp.setXDCRBucketSetting(ctx, bucket) |
| 651 | } |
| 652 | |
| 653 | // All the buckets are created and opened, so now we can perform some synchronous setup (e.g. Creating GSI indexes) |
| 654 | |
| 655 | itemName := "bucket" |
| 656 | err, _ = RetryLoop(ctx, bucket.GetName()+"bucketInitRetry", func() (bool, error, interface{}) { |
| 657 | tbp.Logf(ctx, "Running %s through init function", itemName) |
| 658 | err := bucketInitFunc(ctx, bucket, tbp) |
| 659 | if err != nil { |
| 660 | if errors.Is(err, context.Canceled) { |
| 661 | return false, err, nil |
| 662 | } |
| 663 | tbp.Logf(ctx, "Couldn't init %s, got error: %v - Retrying", itemName, err) |
| 664 | return true, err, nil |
| 665 | } |
| 666 | return false, nil, nil |
| 667 | }, CreateSleeperFunc(5, 1000)) |
| 668 | if ctx.Err() != nil { |
| 669 | bucket.Close(ctx) |
| 670 | return |
no test coverage detected