Qa 7710

README.md

@@ -112,8 +112,9 @@ Configurations are stored in json files. Example:
 | node_list       | -               | Omit this property.                                  |
 
 ##### `kubernetes` - on-demand nodes in kubernetes cluster.
-| Strategy option           | Possible values | Description                 |
-|-------------------------- | --------------- | --------------------------- |
-| params                    | -               | Omit this property.         |
-| node_list.[].params.image | string          | Docker image with selenium. |
-| node_list.[].params.port  | string          | Port of selenium.           |
+| Strategy option             | Possible values | Description                                  |
+|--------------------------   | ---------------------- | ------------------------------------- |
+| params.namespace            | string                 | Namespace in k8s for on-demand nodes. |
+| params.pod_creation_timeout | string as `12m`, `60s` | Max waiting time for creating a pod.  |
+| node_list.[].params.image   | string                 | Docker image with selenium.           |
+| node_list.[].params.port    | string                 | Port of selenium.                     |

handlers/registerNode.go

@@ -50,9 +50,11 @@ func (h *RegisterNode) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
 	for i, value := range capabilitiesList {
 		poolCapabilitiesList[i] = capabilities.Capabilities(value)
 	}
+	hostPort := register.Configuration.Host + ":" + strconv.Itoa(register.Configuration.Port)
 	err = h.Pool.Add(
+		hostPort,
 		pool.NodeTypePersistent,
-		register.Configuration.Host+":"+strconv.Itoa(register.Configuration.Port),
+		hostPort,
 		poolCapabilitiesList,
 	)
 	if err != nil {

pool/mocks.go

@@ -72,3 +72,8 @@ func (s *StrategyListMock) FixNodeStatus(node Node) error {
 	args := s.Called(node)
 	return args.Error(0)
 }
+
+func (s *StorageMock) UpdateAddress(node Node, newAddress string) error {
+	args := s.Called(node, newAddress)
+	return args.Error(0)
+}

pool/node.go

@@ -18,6 +18,11 @@ const (
 )
 
 type Node struct {
+	// A unique key, by which we understand how to find this object in the outer world + for not adding the second time the same thing.
+	// The value may depend on the strategy:
+	// - for constant nodes ip: port
+	// - for temporary pod.name
+	Key             string
 	Type             NodeType
 	Address          string
 	Status           NodeStatus
@@ -28,10 +33,11 @@ type Node struct {
 }
 
 func (n *Node) String() string {
-	return "Node [" + n.Address + "]"
+	return "Node [" + n.Key + "]"
 }
 
 func NewNode(
+	key string,
 	t NodeType,
 	address string,
 	status NodeStatus,
@@ -41,6 +47,7 @@ func NewNode(
 	capabilitiesList []capabilities.Capabilities,
 ) *Node {
 	return &Node{
+		key,
 		t,
 		address,
 		status,

pool/pool.go

@@ -25,6 +25,7 @@ type StorageInterface interface {
 	GetByAddress(string) (Node, error)
 	GetAll() ([]Node, error)
 	Remove(Node) error
+	UpdateAddress(node Node, newAddress string) error
 }
 
 type StrategyInterface interface {
@@ -68,12 +69,12 @@ func (p *Pool) ReserveAvailableNode(caps capabilities.Capabilities) (*Node, erro
 	return &node, err
 }
 
-func (p *Pool) Add(t NodeType, address string, capabilitiesList []capabilities.Capabilities) error {
+func (p *Pool) Add(key string, t NodeType, address string, capabilitiesList []capabilities.Capabilities) error {
 	if len(capabilitiesList) == 0 {
 		return errors.New("[Pool/Add] Capabilities must contains more one element")
 	}
 	ts := time.Now().Unix()
-	return p.storage.Add(*NewNode(t, address, NodeStatusAvailable, "", ts, ts, capabilitiesList), 0)
+	return p.storage.Add(*NewNode(key, t, address, NodeStatusAvailable, "", ts, ts, capabilitiesList), 0)
 }
 
 func (p *Pool) RegisterSession(node *Node, sessionID string) error {
@@ -148,7 +149,7 @@ func (p *Pool) FixNodeStatuses() {
 			continue
 		}
 		if isFixed {
-			log.Infof("Node [%s] status fixed", node.Address)
+			log.Infof("Node [%s] status fixed", node.Key)
 		}
 	}
 }
@@ -169,7 +170,7 @@ func (p *Pool) fixNodeStatus(node *Node) (bool, error) {
 	}
 	err := p.strategyList.FixNodeStatus(*node)
 	if err != nil {
-		return false, fmt.Errorf("Can't fix node [%s] status, %s", node.Address, err.Error())
+		return false, fmt.Errorf("Can't fix node [%s] status, %s", node.Key, err.Error())
 	}
 	return true, nil
 }

pool/pool_test.go

@@ -47,7 +47,7 @@ func TestPool_Add_Positive(t *testing.T) {
 	p := NewPool(s, new(StrategyListMock))
 	eAddress := "127.0.0.1"
 	eNodeType := NodeTypePersistent
-	err := p.Add(eNodeType, eAddress, []capabilities.Capabilities{{"browserName": "ololo"}})
+	err := p.Add("1234", eNodeType, eAddress, []capabilities.Capabilities{{"browserName": "ololo"}})
 	a.Nil(err)
 }
 
@@ -59,7 +59,7 @@ func TestPool_Add_Negative(t *testing.T) {
 	p := NewPool(s, new(StrategyListMock))
 	eAddress := "127.0.0.1"
 	eNodeType := NodeTypePersistent
-	err := p.Add(eNodeType, eAddress, []capabilities.Capabilities{})
+	err := p.Add("1234", eNodeType, eAddress, []capabilities.Capabilities{})
 	a.Error(err)
 }
 
@@ -225,7 +225,7 @@ func TestPool_fixNodeStatus_Positive_BusyExpired(t *testing.T) {
 	slm := new(StrategyListMock)
 	slm.On("FixNodeStatus", mock.AnythingOfType("pool.Node")).Return(nil)
 	p := NewPool(new(StorageMock), slm)
-	node := NewNode(NodeTypePersistent, "", NodeStatusBusy, "", 0, 0, []capabilities.Capabilities{})
+	node := NewNode("123", NodeTypePersistent, "", NodeStatusBusy, "", 0, 0, []capabilities.Capabilities{})
 	isFixed, err := p.fixNodeStatus(node)
 	a.True(isFixed)
 	a.Nil(err)
@@ -236,7 +236,7 @@ func TestPool_fixNodeStatus_Positive_ReservedExpired(t *testing.T) {
 	slm := new(StrategyListMock)
 	slm.On("FixNodeStatus", mock.AnythingOfType("pool.Node")).Return(nil)
 	p := NewPool(new(StorageMock), slm)
-	node := NewNode(NodeTypePersistent, "", NodeStatusReserved, "", 0, 0, []capabilities.Capabilities{})
+	node := NewNode("123", NodeTypePersistent, "", NodeStatusReserved, "", 0, 0, []capabilities.Capabilities{})
 	isFixed, err := p.fixNodeStatus(node)
 	a.True(isFixed)
 	a.Nil(err)
@@ -247,7 +247,7 @@ func TestPool_fixNodeStatus_Positive_BusyNotNotExpired(t *testing.T) {
 	slm := new(StrategyListMock)
 	slm.On("FixNodeStatus", mock.AnythingOfType("pool.Node")).Return(nil)
 	p := NewPool(new(StorageMock), slm)
-	node := NewNode(NodeTypePersistent, "", NodeStatusBusy, "", time.Now().Unix(), 0, []capabilities.Capabilities{})
+	node := NewNode("123", NodeTypePersistent, "", NodeStatusBusy, "", time.Now().Unix(), 0, []capabilities.Capabilities{})
 	isFixed, err := p.fixNodeStatus(node)
 	a.False(isFixed)
 	a.Nil(err)
@@ -258,7 +258,7 @@ func TestPool_fixNodeStatus_Positive_ReservedNotNotExpired(t *testing.T) {
 	slm := new(StrategyListMock)
 	slm.On("FixNodeStatus", mock.AnythingOfType("pool.Node")).Return(nil)
 	p := NewPool(new(StorageMock), slm)
-	node := NewNode(NodeTypePersistent, "", NodeStatusReserved, "", time.Now().Unix(), 0, []capabilities.Capabilities{})
+	node := NewNode("123", NodeTypePersistent, "", NodeStatusReserved, "", time.Now().Unix(), 0, []capabilities.Capabilities{})
 	isFixed, err := p.fixNodeStatus(node)
 	a.False(isFixed)
 	a.Nil(err)
@@ -269,7 +269,7 @@ func TestPool_fixNodeStatus_Positive_AvailableExpired(t *testing.T) {
 	slm := new(StrategyListMock)
 	slm.On("FixNodeStatus", mock.AnythingOfType("pool.Node")).Return(nil)
 	p := NewPool(new(StorageMock), slm)
-	node := NewNode(NodeTypePersistent, "", NodeStatusAvailable, "", 0, 0, []capabilities.Capabilities{})
+	node := NewNode("123", NodeTypePersistent, "", NodeStatusAvailable, "", 0, 0, []capabilities.Capabilities{})
 	isFixed, err := p.fixNodeStatus(node)
 	a.False(isFixed)
 	a.Nil(err)
@@ -281,7 +281,7 @@ func TestPool_fixNodeStatus_NegativeBusy(t *testing.T) {
 	slm := new(StrategyListMock)
 	slm.On("FixNodeStatus", mock.AnythingOfType("pool.Node")).Return(eError)
 	p := NewPool(new(StorageMock), slm)
-	node := NewNode(NodeTypePersistent, "", NodeStatusBusy, "", 0, 0, []capabilities.Capabilities{})
+	node := NewNode("123", NodeTypePersistent, "", NodeStatusBusy, "", 0, 0, []capabilities.Capabilities{})
 	isFixed, err := p.fixNodeStatus(node)
 	a.False(isFixed)
 	a.Error(err)
@@ -293,7 +293,7 @@ func TestPool_fixNodeStatus_NegativeReserved(t *testing.T) {
 	slm := new(StrategyListMock)
 	slm.On("FixNodeStatus", mock.AnythingOfType("pool.Node")).Return(eError)
 	p := NewPool(new(StorageMock), slm)
-	node := NewNode(NodeTypePersistent, "", NodeStatusReserved, "", 0, 0, []capabilities.Capabilities{})
+	node := NewNode("123", NodeTypePersistent, "", NodeStatusReserved, "", 0, 0, []capabilities.Capabilities{})
 	isFixed, err := p.fixNodeStatus(node)
 	a.False(isFixed)
 	a.Error(err)

pool/strategy/kubernetes/config.go

@@ -4,9 +4,33 @@ import (
 	"encoding/json"
 	"errors"
 	"github.com/qa-dev/jsonwire-grid/config"
+	"time"
+	"fmt"
 )
 
 type strategyParams struct {
+	Namespace string
+	PodCreationTimeout time.Duration
+}
+
+func (sp *strategyParams) UnmarshalJSON(b []byte) error {
+	tempStruct :=  struct{
+		Namespace string `json:"namespace"`
+		PodCreationTimeout string `json:"pod_creation_timeout"`
+	} {
+		"default",
+		"1m",
+	}
+	if err := json.Unmarshal(b, &tempStruct); err != nil {
+		return err
+	}
+	podCreationTimeout, err := time.ParseDuration(tempStruct.PodCreationTimeout)
+	if err != nil {
+		return fmt.Errorf("invalid value strategy.pod_creation_timeout in config, given: %v", tempStruct.PodCreationTimeout)
+	}
+	sp.Namespace = tempStruct.Namespace
+	sp.PodCreationTimeout = podCreationTimeout
+	return nil
 }
 
 type strategyConfig struct {

pool/strategy/kubernetes/factory.go

@@ -8,6 +8,7 @@ import (
 	"github.com/qa-dev/jsonwire-grid/pool/capabilities"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
+	log "github.com/Sirupsen/logrus"
 )
 
 type StrategyFactory struct {
@@ -31,6 +32,8 @@ func (f *StrategyFactory) Create(
 		}
 	}
 
+	log.Debugf("strategy kubernetes config, %+v", strategyConfig)
+
 	//todo: выпилить этот говноклиент, когда будет работать нормальный
 	kubConfig, err := rest.InClusterConfig()
 	if err != nil {
@@ -42,9 +45,10 @@ func (f *StrategyFactory) Create(
 		return nil, errors.New("create k8s clientset, " + err.Error())
 	}
 
-	provider := &kubernetesProvider{
+	provider := &kubDnsProvider{
 		clientset:     clientset,
-		namespace:     "default", //todo: брать из конфига !!!
+		namespace:     strategyConfig.Params.Namespace,
+		podCreationTimeout: strategyConfig.Params.PodCreationTimeout,
 		clientFactory: clientFactory,
 	}
 

pool/strategy/kubernetes/provider.go

@@ -9,20 +9,24 @@ import (
 	"net"
 	"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
 }
 
-type kubernetesProvider struct {
+type kubDnsProvider struct {
 	clientset     *kubernetes.Clientset
 	namespace     string
+	podCreationTimeout     time.Duration
 	clientFactory jsonwire.ClientFactoryInterface
 }
 
-func (p *kubernetesProvider) 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}
@@ -32,62 +36,70 @@ func (p *kubernetesProvider) 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())
+	stop := time.After(p.podCreationTimeout)
+	log.Debugf("start waiting pod ip")
+	var createdPodIP string
+LoopWaitIP:
+	for {
+		select {
+		case <-stop:
+			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))
-	stop := time.After(40 * time.Second)
-	log.Debugln("start waiting")
-Loop:
+	nodeAddress = net.JoinHostPort(createdPodIP, nodeParams.Port)
+	client := p.clientFactory.Create(nodeAddress)
+	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
 }
 
-func (p *kubernetesProvider) Destroy(podName string) error {
+//Destroy - destroy all pod data (idempotent operation)
+func (p *kubDnsProvider) Destroy(podName string) error {
 	err := p.clientset.CoreV1Client.Pods(p.namespace).Delete(podName, &apiV1.DeleteOptions{})
-	if err != nil {
-		err = errors.New("send command pod/delete to k8s, " + err.Error())
-		return err
-	}
-	err = p.clientset.CoreV1Client.Services(p.namespace).Delete(podName, &apiV1.DeleteOptions{})
-	if err != nil {
-		err = errors.New("send command service/delete to k8s, " + err.Error())
-		return err
+	if err != nil && !strings.Contains(err.Error(), "not found") {
+		return errors.New("send command pod/delete to k8s, " + err.Error())
 	}
 	return nil
 }