Make FakeAPIClient threadsafe (#4790)

Author: briandealwis

pkg/skaffold/build/buildpacks/fetcher_test.go

@@ -54,7 +54,7 @@ func TestFetcher(t *testing.T) {
 			f := newFetcher(&out, docker)
 			f.Fetch(context.Background(), "image", true, test.pull)
 
-			t.CheckDeepEqual(test.expectedPulled, api.Pulled)
+			t.CheckDeepEqual(test.expectedPulled, api.Pulled())
 		})
 	}
 }

pkg/skaffold/build/local/local_test.go

@@ -260,7 +260,7 @@ func TestLocalRun(t *testing.T) {
 
 			t.CheckErrorAndDeepEqual(test.shouldErr, err, test.expected, res)
 			t.CheckDeepEqual(test.expectedWarnings, fakeWarner.Warnings)
-			t.CheckDeepEqual(test.expectedPushed, test.api.Pushed)
+			t.CheckDeepEqual(test.expectedPushed, test.api.Pushed())
 		})
 	}
 }

testutil/fake_image_api.go

@@ -25,6 +25,8 @@ import (
 	"io/ioutil"
 	"os"
 	"strings"
+	"sync"
+	"sync/atomic"
 
 	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/registry"
@@ -44,18 +46,20 @@ const (
 type FakeAPIClient struct {
 	client.CommonAPIClient
 
-	tagToImageID    map[string]string
 	ErrImageBuild   bool
 	ErrImageInspect bool
 	ErrImagePush    bool
 	ErrImagePull    bool
 	ErrStream       bool
 	ErrVersion      bool
 
-	nextImageID int
-	Pushed      map[string]string
-	Pulled      []string
-	Built       []types.ImageBuildOptions
+	nextImageID  int32
+	tagToImageID sync.Map // map[string]string
+	pushed       sync.Map // map[string]string
+	pulled       sync.Map // map[string]string
+
+	mux   sync.Mutex
+	Built []types.ImageBuildOptions
 }
 
 func (f *FakeAPIClient) ServerVersion(ctx context.Context) (types.Version, error) {
@@ -66,18 +70,35 @@ func (f *FakeAPIClient) ServerVersion(ctx context.Context) (types.Version, error
 }
 
 func (f *FakeAPIClient) Add(tag, imageID string) *FakeAPIClient {
-	if f.tagToImageID == nil {
-		f.tagToImageID = make(map[string]string)
-	}
-
-	f.tagToImageID[imageID] = imageID
-	f.tagToImageID[tag] = imageID
+	f.tagToImageID.Store(imageID, imageID)
+	f.tagToImageID.Store(tag, imageID)
 	if !strings.Contains(tag, ":") {
-		f.tagToImageID[tag+":latest"] = imageID
+		f.tagToImageID.Store(tag+":latest", imageID)
 	}
 	return f
 }
 
+func (f *FakeAPIClient) Pulled() []string {
+	var p []string
+	f.pulled.Range(func(ref, _ interface{}) bool {
+		p = append(p, ref.(string))
+		return true
+	})
+	return p
+}
+
+func (f *FakeAPIClient) Pushed() map[string]string {
+	p := make(map[string]string)
+	f.pushed.Range(func(ref, id interface{}) bool {
+		p[ref.(string)] = id.(string)
+		return true
+	})
+	if len(p) == 0 {
+		return nil
+	}
+	return p
+}
+
 type notFoundError struct {
 	error
 }
@@ -103,49 +124,69 @@ func (f *FakeAPIClient) ImageBuild(_ context.Context, _ io.Reader, options types
 		return types.ImageBuildResponse{}, fmt.Errorf("")
 	}
 
-	f.nextImageID++
-	imageID := fmt.Sprintf("sha256:%d", f.nextImageID)
+	next := atomic.AddInt32(&f.nextImageID, 1)
+	imageID := fmt.Sprintf("sha256:%d", next)
 
 	for _, tag := range options.Tags {
 		f.Add(tag, imageID)
 	}
 
+	f.mux.Lock()
 	f.Built = append(f.Built, options)
+	f.mux.Unlock()
 
 	return types.ImageBuildResponse{
 		Body: f.body(imageID),
 	}, nil
 }
 
-func (f *FakeAPIClient) ImageInspectWithRaw(_ context.Context, ref string) (types.ImageInspect, []byte, error) {
+func (f *FakeAPIClient) ImageInspectWithRaw(_ context.Context, refOrID string) (types.ImageInspect, []byte, error) {
 	if f.ErrImageInspect {
 		return types.ImageInspect{}, nil, fmt.Errorf("")
 	}
 
-	for tag, imageID := range f.tagToImageID {
-		if tag == ref || imageID == ref {
-			rawConfig := []byte(fmt.Sprintf(`{"Config":{"Image":"%s"}}`, imageID))
+	ref, imageID, err := f.findImageID(refOrID)
+	if err != nil {
+		return types.ImageInspect{}, nil, err
+	}
 
-			var repoDigests []string
-			if digest, found := f.Pushed[ref]; found {
-				repoDigests = append(repoDigests, ref+"@"+digest)
-			}
+	rawConfig := []byte(fmt.Sprintf(`{"Config":{"Image":"%s"}}`, imageID))
 
-			return types.ImageInspect{
-				ID:          imageID,
-				RepoDigests: repoDigests,
-			}, rawConfig, nil
-		}
+	var repoDigests []string
+	if digest, found := f.pushed.Load(ref); found {
+		repoDigests = append(repoDigests, ref+"@"+digest.(string))
 	}
 
-	return types.ImageInspect{}, nil, &notFoundError{}
+	return types.ImageInspect{
+		ID:          imageID,
+		RepoDigests: repoDigests,
+	}, rawConfig, nil
+}
+
+func (f *FakeAPIClient) findImageID(refOrID string) (string, string, error) {
+	if id, found := f.tagToImageID.Load(refOrID); found {
+		return refOrID, id.(string), nil
+	}
+	var ref, id string
+	f.tagToImageID.Range(func(r, i interface{}) bool {
+		if r == refOrID || i == refOrID {
+			ref = r.(string)
+			id = i.(string)
+			return false
+		}
+		return true
+	})
+	if ref == "" {
+		return "", "", &notFoundError{}
+	}
+	return ref, id, nil
 }
 
 func (f *FakeAPIClient) DistributionInspect(ctx context.Context, ref, encodedRegistryAuth string) (registry.DistributionInspect, error) {
-	if sha, found := f.Pushed[ref]; found {
+	if sha, found := f.pushed.Load(ref); found {
 		return registry.DistributionInspect{
 			Descriptor: v1.Descriptor{
-				Digest: digest.Digest(sha),
+				Digest: digest.Digest(sha.(string)),
 			},
 		}, nil
 	}
@@ -154,12 +195,12 @@ func (f *FakeAPIClient) DistributionInspect(ctx context.Context, ref, encodedReg
 }
 
 func (f *FakeAPIClient) ImageTag(_ context.Context, image, ref string) error {
-	imageID, ok := f.tagToImageID[image]
+	imageID, ok := f.tagToImageID.Load(image)
 	if !ok {
 		return fmt.Errorf("image %s not found", image)
 	}
 
-	f.Add(ref, imageID)
+	f.Add(ref, imageID.(string))
 	return nil
 }
 
@@ -168,22 +209,24 @@ func (f *FakeAPIClient) ImagePush(_ context.Context, ref string, _ types.ImagePu
 		return nil, fmt.Errorf("")
 	}
 
+	// use the digest if previously pushed
+	imageID, found := f.tagToImageID.Load(ref)
+	if !found {
+		imageID = ""
+	}
 	sha256Digester := sha256.New()
-	if _, err := sha256Digester.Write([]byte(f.tagToImageID[ref])); err != nil {
+	if _, err := sha256Digester.Write([]byte(imageID.(string))); err != nil {
 		return nil, err
 	}
 
 	digest := "sha256:" + fmt.Sprintf("%x", sha256Digester.Sum(nil))[0:64]
-	if f.Pushed == nil {
-		f.Pushed = make(map[string]string)
-	}
-	f.Pushed[ref] = digest
 
+	f.pushed.Store(ref, digest)
 	return f.body(digest), nil
 }
 
 func (f *FakeAPIClient) ImagePull(_ context.Context, ref string, _ types.ImagePullOptions) (io.ReadCloser, error) {
-	f.Pulled = append(f.Pulled, ref)
+	f.pulled.Store(ref, ref)
 	if f.ErrImagePull {
 		return nil, fmt.Errorf("")
 	}
@@ -203,8 +246,8 @@ func (f *FakeAPIClient) ImageLoad(ctx context.Context, input io.Reader, quiet bo
 		return types.ImageLoadResponse{}, fmt.Errorf("reading tar")
 	}
 
-	f.nextImageID++
-	imageID := fmt.Sprintf("sha256:%d", f.nextImageID)
+	next := atomic.AddInt32(&f.nextImageID, 1)
+	imageID := fmt.Sprintf("sha256:%d", next)
 	f.Add(ref, imageID)
 
 	return types.ImageLoadResponse{