QA-7710: change service address to pod ip in kubernetes strategy

Author: Mikhail Podtserkovskiy

pool/strategy/kubernetes/provider.go

@@ -10,10 +10,11 @@ import (
 	"strconv"
 	"time"
 	"strings"
+	"fmt"
 )
 
 type kubernetesProviderInterface interface {
-	Create(podName string, nodeParams nodeParams) error
+	Create(podName string, nodeParams nodeParams) (nodeAddress string, err error)
 	// idempotent operation
 	Destroy(podName string) error
 }
@@ -24,7 +25,7 @@ type kubDnsProvider struct {
 	clientFactory jsonwire.ClientFactoryInterface
 }
 
-func (p *kubDnsProvider) Create(podName string, nodeParams nodeParams) error {
+func (p *kubDnsProvider) Create(podName string, nodeParams nodeParams) (nodeAddress string, err error) {
 	pod := &apiV1.Pod{}
 	pod.ObjectMeta.Name = podName
 	pod.ObjectMeta.Labels = map[string]string{"name": podName}
@@ -34,50 +35,64 @@ func (p *kubDnsProvider) Create(podName string, nodeParams nodeParams) error {
 	container.Image = nodeParams.Image
 	port, err := strconv.Atoi(nodeParams.Port)
 	if err != nil {
-		return errors.New("convert to int nodeParams.Port, " + err.Error())
+		return "", errors.New("convert to int nodeParams.Port, " + err.Error())
 	}
 	container.Ports = []apiV1.ContainerPort{{ContainerPort: int32(port)}}
 	pod.Spec.Containers = append(pod.Spec.Containers, container)
 	_, err = p.clientset.CoreV1Client.Pods(p.namespace).Create(pod)
 	if err != nil {
-		return errors.New("send command pod/create to k8s, " + err.Error())
+		return "", errors.New("send command pod/create to k8s, " + err.Error())
 	}
 
-	service := &apiV1.Service{}
-	service.ObjectMeta.Name = podName
-	service.Spec.ClusterIP = "None"
-	service.Spec.Ports = []apiV1.ServicePort{{Port: int32(port)}}
-	service.Spec.Selector = map[string]string{"name": podName}
-	_, err = p.clientset.CoreV1Client.Services(p.namespace).Create(service)
-	if err != nil {
-		return errors.New("send command service/create to k8s, " + err.Error())
+	stopWaitIP := time.After(40 * time.Second)
+	log.Debugf("start waiting pod ip")
+	var createdPodIP string
+LoopWaitIP:
+	for {
+		select {
+		case <-stopWaitIP:
+			return "", fmt.Errorf("wait podIP stopped by timeout, %v", podName)
+		default:
+			time.Sleep(time.Second)
+			createdPod, err := p.clientset.CoreV1Client.Pods(p.namespace).Get(podName)
+			if err != nil {
+				log.Debugf("fail get created pod, %v, %v",podName, err)
+				continue
+			}
+			if createdPod.Status.PodIP == "" {
+				log.Debugf("empty pod ip, %v", podName)
+				continue
+			}
+			createdPodIP = createdPod.Status.PodIP
+			break LoopWaitIP
+		}
 	}
 
 	// todo: пока так ожидаем поднятие ноды, так как не понятно что конкретно означают статусы возвращаемые через апи
-	client := p.clientFactory.Create(net.JoinHostPort(podName, nodeParams.Port))
+	nodeAddress = net.JoinHostPort(createdPodIP, nodeParams.Port)
+	client := p.clientFactory.Create(nodeAddress)
 	stop := time.After(40 * time.Second)
-	log.Debugln("start waiting")
-Loop:
+	log.Debugln("start waiting selenium")
+LoopWaitSelenium:
 	for {
 		select {
 		case <-stop:
-			return errors.New("wait stopped by timeout")
+			return "", fmt.Errorf("wait selenium stopped by timeout, %v", podName)
 		default:
 			time.Sleep(time.Second)
-			log.Debugln("start request")
 			message, err := client.Health()
 			if err != nil {
 				log.Debugf("fail request, %v", err)
 				continue
 			}
 			log.Debugf("done request, status: %v", message.Status)
 			if message.Status == 0 {
-				break Loop
+				break LoopWaitSelenium
 			}
 		}
 	}
 
-	return nil
+	return nodeAddress, nil
 }
 
 //Destroy - destroy all pod data (idempotent operation)
@@ -90,12 +105,5 @@ func (p *kubDnsProvider) Destroy(podName string) error {
 		err = errors.New("send command pod/delete to k8s, " + err.Error())
 		return err
 	}
-	err = p.clientset.CoreV1Client.Services(p.namespace).Delete(podName, &apiV1.DeleteOptions{})
-	switch {
-	case err != nil && strings.Contains(err.Error(), "not found"):
-		// service already deleted
-	case err != nil:
-		return errors.New("send command service/delete to k8s, " + err.Error())
-	}
 	return nil
 }

pool/strategy/kubernetes/strategy.go

@@ -6,7 +6,6 @@ import (
 	"github.com/qa-dev/jsonwire-grid/pool/capabilities"
 	"github.com/qa-dev/jsonwire-grid/pool/strategy"
 	"github.com/satori/go.uuid"
-	"net"
 	"time"
 	"fmt"
 )
@@ -25,20 +24,25 @@ func (s *Strategy) Reserve(desiredCaps capabilities.Capabilities) (pool.Node, er
 	}
 	podName := "wd-node-" + uuid.NewV4().String()
 	ts := time.Now().Unix()
-	address := net.JoinHostPort(podName, nodeConfig.Params.Port)
-	node := pool.NewNode(podName, pool.NodeTypeKubernetes, address, pool.NodeStatusReserved, "", ts, ts, []capabilities.Capabilities{})
+	node := pool.NewNode(podName, pool.NodeTypeKubernetes, "temp-value-replace-me", pool.NodeStatusReserved, "", ts, ts, []capabilities.Capabilities{})
 	err := s.storage.Add(*node, s.config.Limit)
 	if err != nil {
 		return pool.Node{}, errors.New("add node to storage, " + err.Error())
 	}
-	err = s.provider.Create(podName, nodeConfig.Params)
+	nodeAddress, err := s.provider.Create(podName, nodeConfig.Params)
 	if err != nil {
 		go func(podName string) {
 			time.Sleep(time.Minute * 2)
 			_ = s.provider.Destroy(podName) // на случай если что то криво создалось
 		}(podName)
 		return pool.Node{}, errors.New("create node by provider, " + err.Error())
 	}
+
+	err = s.storage.UpdateAddress(*node, nodeAddress)
+	if err != nil {
+		return pool.Node{}, errors.New("update node address in storage, " + err.Error())
+	}
+	node.Address = nodeAddress
 	return *node, nil
 
 }

pool/strategy/kubernetes/strategy_test.go

@@ -14,9 +14,9 @@ type providerMock struct {
 	mock.Mock
 }
 
-func (p *providerMock) Create(podName string, nodeParams nodeParams) error {
+func (p *providerMock) Create(podName string, nodeParams nodeParams) (nodeAddress string, err error) {
 	args := p.Called(podName, nodeParams)
-	return args.Error(0)
+	return args.String(0), args.Error(1)
 }
 
 func (p *providerMock) Destroy(podName string) error {
@@ -32,14 +32,17 @@ func TestStrategy_Reserve_Positive(t *testing.T) {
 	}
 	sm := new(pool.StorageMock)
 	sm.On("Add", mock.AnythingOfType("pool.Node"), mock.AnythingOfType("int")).Return(nil)
+	sm.On("UpdateAddress", mock.AnythingOfType("pool.Node"), mock.AnythingOfType("string")).Return(nil)
 	cm := new(capabilities.ComparatorMock)
 	cm.On("Compare", mock.AnythingOfType("capabilities.Capabilities"), mock.AnythingOfType("capabilities.Capabilities")).Return(true)
 	pm := new(providerMock)
-	pm.On("Create", mock.AnythingOfType("string"), mock.AnythingOfType("nodeParams")).Return(nil)
+	expectedAddress := "addr"
+	pm.On("Create", mock.AnythingOfType("string"), mock.AnythingOfType("nodeParams")).Return(expectedAddress, nil)
 	str := Strategy{storage: sm, provider: pm, config: strategyConfig, capsComparator: cm}
 	node, err := str.Reserve(capabilities.Capabilities{})
 	assert.Nil(t, err)
 	assert.NotNil(t, node)
+	assert.Equal(t, expectedAddress, node.Address)
 }
 
 func TestStrategy_Reserve_Negative_NotMatchCapabilities(t *testing.T) {
@@ -61,7 +64,27 @@ func TestStrategy_Reserve_Negative_ReserveAvailable(t *testing.T) {
 	cm.On("Compare", mock.AnythingOfType("capabilities.Capabilities"), mock.AnythingOfType("capabilities.Capabilities")).Return(true)
 	pm := new(providerMock)
 	eError := errors.New("Error")
-	pm.On("Create", mock.AnythingOfType("string"), mock.AnythingOfType("nodeParams")).Return(eError)
+	pm.On("Create", mock.AnythingOfType("string"), mock.AnythingOfType("nodeParams")).Return("", eError)
+	pm.On("Destroy", mock.AnythingOfType("string")).Return(nil)
+	str := Strategy{storage: sm, provider: pm, config: strategyConfig, capsComparator: cm}
+	_, err := str.Reserve(capabilities.Capabilities{})
+	assert.NotNil(t, err)
+}
+
+func TestStrategy_Reserve_Negative_UpdateAddress(t *testing.T) {
+	nodeCfg := nodeConfig{}
+	nodeCfg.CapabilitiesList = []map[string]interface{}{{"cap1": "cal1"}}
+	strategyConfig := strategyConfig{
+		NodeList: []nodeConfig{nodeCfg},
+	}
+	sm := new(pool.StorageMock)
+	sm.On("Add", mock.AnythingOfType("pool.Node"), mock.AnythingOfType("int")).Return(nil)
+	sm.On("UpdateAddress", mock.AnythingOfType("pool.Node"), mock.AnythingOfType("string")).Return(errors.New("muhaha-error"))
+	cm := new(capabilities.ComparatorMock)
+	cm.On("Compare", mock.AnythingOfType("capabilities.Capabilities"), mock.AnythingOfType("capabilities.Capabilities")).Return(true)
+	pm := new(providerMock)
+	eError := errors.New("Error")
+	pm.On("Create", mock.AnythingOfType("string"), mock.AnythingOfType("nodeParams")).Return("", eError)
 	pm.On("Destroy", mock.AnythingOfType("string")).Return(nil)
 	str := Strategy{storage: sm, provider: pm, config: strategyConfig, capsComparator: cm}
 	_, err := str.Reserve(capabilities.Capabilities{})