Merge branch 'bug/timeout' into 'master'

pull image for as long as you like

See merge request !140

Author: CMGS

cluster/calcium/create_container.go

@@ -7,7 +7,6 @@ import (
 	"strconv"
 	"strings"
 	"sync"
-	"time"
 
 	log "github.com/Sirupsen/logrus"
 	enginetypes "github.com/docker/docker/api/types"
@@ -104,7 +103,7 @@ func (c *calcium) doCreateContainerWithMemoryPrior(nodeInfo types.NodeInfo, spec
 		return ms
 	}
 
-	if err := pullImage(node, opts.Image, c.config.Timeout.CreateContainer); err != nil {
+	if err := pullImage(node, opts.Image); err != nil {
 		for i := 0; i < nodeInfo.Deploy; i++ {
 			if err != nil {
 				log.Errorf("Error during pullImage %s for %s: %v", opts.Image, nodeInfo.Name, err)
@@ -279,7 +278,7 @@ func (c *calcium) doCreateContainerWithCPUPrior(nodeName string, cpuMap []types.
 		return ms
 	}
 
-	if err := pullImage(node, opts.Image, c.config.Timeout.CreateContainer); err != nil {
+	if err := pullImage(node, opts.Image); err != nil {
 		log.Errorf("Error during pullImage: %v", err)
 		for i := 0; i < len(ms); i++ {
 			ms[i].Error = err.Error()
@@ -566,13 +565,12 @@ func (c *calcium) makeContainerOptions(index int, quota types.CPUMap, specs type
 
 // Pull an image
 // Blocks until it finishes.
-func pullImage(node *types.Node, image string, timeout time.Duration) error {
+func pullImage(node *types.Node, image string) error {
 	log.Debugf("Pulling image %s", image)
 	if image == "" {
 		return fmt.Errorf("Goddamn empty image, WTF?")
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
-	defer cancel()
+	ctx := context.Background()
 	outStream, err := node.Engine.ImagePull(ctx, image, enginetypes.ImagePullOptions{})
 	if err != nil {
 		log.Errorf("Error during pulling image %s: %v", image, err)

cluster/calcium/create_container_test.go

@@ -3,7 +3,6 @@ package calcium
 import (
 	"fmt"
 	"testing"
-	"time"
 
 	"github.com/docker/docker/client"
 	"github.com/stretchr/testify/assert"
@@ -19,7 +18,7 @@ func TestPullImage(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if err := pullImage(nodes[0], image, 5*time.Second); err != nil {
+	if err := pullImage(nodes[0], image); err != nil {
 		t.Fatal(err)
 	}
 }
@@ -90,42 +89,3 @@ func TestCreateContainerWithCPUPrior(t *testing.T) {
 		fmt.Printf("Get Container ID: %s\n", msg.ContainerID)
 	}
 }
-
-func TestCreateContainerMemTimeoutError(t *testing.T) {
-	initMockConfig()
-	mockTimeoutError = true
-
-	// update node
-	mockStore.On("UpdateNode", mock.MatchedBy(func(input *types.Node) bool {
-		return true
-	})).Return(nil)
-
-	// Create Container with memory prior
-	testlogF("Create containers with memory prior")
-	createCh, err := mockc.createContainerWithMemoryPrior(specs, opts)
-	assert.NoError(t, err)
-	for msg := range createCh {
-		assert.False(t, msg.Success)
-		assert.Contains(t, msg.Error, "context deadline exceeded")
-	}
-
-}
-
-func TestCreateContainerCPUTimeoutError(t *testing.T) {
-	initMockConfig()
-	mockTimeoutError = true
-
-	// update node
-	mockStore.On("UpdateNode", mock.MatchedBy(func(input *types.Node) bool {
-		return true
-	})).Return(nil)
-
-	// Create Container with cpu prior
-	testlogF("Create containers with cpu prior")
-	createCh, err := mockc.createContainerWithCPUPrior(specs, opts)
-	assert.NoError(t, err)
-	for msg := range createCh {
-		assert.False(t, msg.Success)
-		assert.Contains(t, msg.Error, "context deadline exceeded")
-	}
-}

cluster/calcium/image.go

@@ -36,7 +36,7 @@ func (c *calcium) cacheImage(podname, image string) error {
 		wg.Add(1)
 		go func(node *types.Node) {
 			defer wg.Done()
-			pullImage(node, image, c.config.Timeout.CreateContainer)
+			pullImage(node, image)
 		}(node)
 	}
 

cluster/calcium/mock_test.go

@@ -2,23 +2,19 @@ package calcium
 
 import (
 	"bytes"
-	"context"
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
-	"io"
 	"io/ioutil"
 	"net/http"
 	"runtime"
 	"strings"
-	"time"
 
 	"github.com/stretchr/testify/mock"
 
 	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/client"
-	"github.com/docker/docker/pkg/ioutils"
 	"gitlab.ricebook.net/platform/core/store/mock"
 	coretypes "gitlab.ricebook.net/platform/core/types"
 	"gitlab.ricebook.net/platform/core/utils"
@@ -110,35 +106,6 @@ func mockContainerID() string {
 	return utils.RandomString(64)
 }
 
-func mockReadCloser(ctx context.Context) io.ReadCloser {
-	pr, pw := io.Pipe()
-	w := ioutils.NewWriteFlusher(pw)
-	msgChan := make(chan string)
-
-	go func() {
-		for {
-			msgChan <- "pulling image...\n"
-			time.Sleep(1000 * time.Millisecond)
-		}
-	}()
-	go func() {
-		for {
-			select {
-			case <-ctx.Done():
-				testlogF("Error!! Context canceld!!!")
-				w.Close()
-				pw.Close()
-				pr.Close()
-				return
-			case msg := <-msgChan:
-				w.Write([]byte(msg))
-			}
-		}
-	}()
-
-	return pr
-}
-
 func mockDockerDoer(r *http.Request) (*http.Response, error) {
 	var b []byte
 	prefix := fmt.Sprintf("/%s", APIVersion)
@@ -176,13 +143,6 @@ func mockDockerDoer(r *http.Request) (*http.Response, error) {
 		tag := query.Get("tag")
 		testlogF("mock docker create image: %s:%s", fromImage, tag)
 		b = []byte("body")
-		if mockTimeoutError {
-			ctx := r.Context()
-			return &http.Response{
-				StatusCode: http.StatusOK,
-				Body:       mockReadCloser(ctx),
-			}, nil
-		}
 	case "/images/json": // docker images
 		testlogF("mock docker list images")
 		b, _ = json.Marshal([]types.ImageSummary{