Merge pull request #552 from digitalocean/asb/issues/551

vpcs: Protect against race conditions in IP range assignment.

digitalocean/provider.go

@@ -2,8 +2,13 @@ package digitalocean
 
 import (
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+
+	"github.com/digitalocean/terraform-provider-digitalocean/internal/mutexkv"
 )
 
+// Global MutexKV
+var mutexKV = mutexkv.NewMutexKV()
+
 // Provider returns a schema.Provider for DigitalOcean.
 func Provider() *schema.Provider {
 	p := &schema.Provider{

digitalocean/resource_digitalocean_vpc.go

@@ -79,9 +79,10 @@ func resourceDigitalOceanVPC() *schema.Resource {
 func resourceDigitalOceanVPCCreate(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
 	client := meta.(*CombinedConfig).godoClient()
 
+	region := d.Get("region").(string)
 	vpcRequest := &godo.VPCCreateRequest{
 		Name:       d.Get("name").(string),
-		RegionSlug: d.Get("region").(string),
+		RegionSlug: region,
 	}
 
 	if v, ok := d.GetOk("description"); ok {
@@ -92,6 +93,12 @@ func resourceDigitalOceanVPCCreate(ctx context.Context, d *schema.ResourceData,
 		vpcRequest.IPRange = v.(string)
 	}
 
+	// Prevent parallel creation of VPCs in the same region to protect
+	// against race conditions in IP range assignment.
+	key := fmt.Sprintf("resource_digitalocean_vpc/%s", region)
+	mutexKV.Lock(key)
+	defer mutexKV.Unlock(key)
+
 	log.Printf("[DEBUG] VPC create request: %#v", vpcRequest)
 	vpc, _, err := client.VPCs.Create(context.Background(), vpcRequest)
 	if err != nil {

digitalocean/resource_digitalocean_vpc_test.go

@@ -78,6 +78,36 @@ func TestAccDigitalOceanVPC_IPRange(t *testing.T) {
 	})
 }
 
+// https://github.com/digitalocean/terraform-provider-digitalocean/issues/551
+func TestAccDigitalOceanVPC_IPRangeRace(t *testing.T) {
+	vpcNameOne := randomTestName()
+	vpcNameTwo := randomTestName()
+	vpcCreateConfig := fmt.Sprintf(testAccCheckDigitalOceanVPCConfig_IPRangeRace, vpcNameOne, vpcNameTwo)
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:          func() { testAccPreCheck(t) },
+		ProviderFactories: testAccProviderFactories,
+		CheckDestroy:      testAccCheckDigitalOceanVPCDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: vpcCreateConfig,
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckDigitalOceanVPCExists("digitalocean_vpc.foo"),
+					testAccCheckDigitalOceanVPCExists("digitalocean_vpc.bar"),
+					resource.TestCheckResourceAttr(
+						"digitalocean_vpc.foo", "name", vpcNameOne),
+					resource.TestCheckResourceAttrSet(
+						"digitalocean_vpc.foo", "ip_range"),
+					resource.TestCheckResourceAttr(
+						"digitalocean_vpc.bar", "name", vpcNameTwo),
+					resource.TestCheckResourceAttrSet(
+						"digitalocean_vpc.bar", "ip_range"),
+				),
+			},
+		},
+	})
+}
+
 func testAccCheckDigitalOceanVPCDestroy(s *terraform.State) error {
 	client := testAccProvider.Meta().(*CombinedConfig).godoClient()
 
@@ -138,3 +168,15 @@ resource "digitalocean_vpc" "foobar" {
 	ip_range = "10.10.10.0/24"
 }
 `
+
+const testAccCheckDigitalOceanVPCConfig_IPRangeRace = `
+resource "digitalocean_vpc" "foo" {
+	name        = "%s"
+	region      = "nyc3"
+}
+
+resource "digitalocean_vpc" "bar" {
+	name        = "%s"
+	region      = "nyc3"
+}
+`

internal/mutexkv/mutexkv.go

@@ -0,0 +1,53 @@
+package mutexkv
+
+import (
+	"log"
+	"sync"
+)
+
+// MutexKV is a simple key/value store for arbitrary mutexes. It can be used to
+// serialize changes across arbitrary collaborators that share knowledge of the
+// keys they must serialize on.
+//
+// The initial use case is to let aws_security_group_rule resources serialize
+// their access to individual security groups based on SG ID.
+//
+// Originally from: https://github.com/hashicorp/terraform-plugin-sdk/blob/v1.12.0/helper/mutexkv/mutexkv.go
+type MutexKV struct {
+	lock  sync.Mutex
+	store map[string]*sync.Mutex
+}
+
+// Locks the mutex for the given key. Caller is responsible for calling Unlock
+// for the same key
+func (m *MutexKV) Lock(key string) {
+	log.Printf("[DEBUG] Locking %q", key)
+	m.get(key).Lock()
+	log.Printf("[DEBUG] Locked %q", key)
+}
+
+// Unlock the mutex for the given key. Caller must have called Lock for the same key first
+func (m *MutexKV) Unlock(key string) {
+	log.Printf("[DEBUG] Unlocking %q", key)
+	m.get(key).Unlock()
+	log.Printf("[DEBUG] Unlocked %q", key)
+}
+
+// Returns a mutex for the given key, no guarantee of its lock status
+func (m *MutexKV) get(key string) *sync.Mutex {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	mutex, ok := m.store[key]
+	if !ok {
+		mutex = &sync.Mutex{}
+		m.store[key] = mutex
+	}
+	return mutex
+}
+
+// Returns a properly initalized MutexKV
+func NewMutexKV() *MutexKV {
+	return &MutexKV{
+		store: make(map[string]*sync.Mutex),
+	}
+}

internal/mutexkv/mutexkv_test.go

@@ -0,0 +1,67 @@
+package mutexkv
+
+import (
+	"testing"
+	"time"
+)
+
+func TestMutexKVLock(t *testing.T) {
+	mkv := NewMutexKV()
+
+	mkv.Lock("foo")
+
+	doneCh := make(chan struct{})
+
+	go func() {
+		mkv.Lock("foo")
+		close(doneCh)
+	}()
+
+	select {
+	case <-doneCh:
+		t.Fatal("Second lock was able to be taken. This shouldn't happen.")
+	case <-time.After(50 * time.Millisecond):
+		// pass
+	}
+}
+
+func TestMutexKVUnlock(t *testing.T) {
+	mkv := NewMutexKV()
+
+	mkv.Lock("foo")
+	mkv.Unlock("foo")
+
+	doneCh := make(chan struct{})
+
+	go func() {
+		mkv.Lock("foo")
+		close(doneCh)
+	}()
+
+	select {
+	case <-doneCh:
+		// pass
+	case <-time.After(50 * time.Millisecond):
+		t.Fatal("Second lock blocked after unlock. This shouldn't happen.")
+	}
+}
+
+func TestMutexKVDifferentKeys(t *testing.T) {
+	mkv := NewMutexKV()
+
+	mkv.Lock("foo")
+
+	doneCh := make(chan struct{})
+
+	go func() {
+		mkv.Lock("bar")
+		close(doneCh)
+	}()
+
+	select {
+	case <-doneCh:
+		// pass
+	case <-time.After(50 * time.Millisecond):
+		t.Fatal("Second lock on a different key blocked. This shouldn't happen.")
+	}
+}