Initial k8s support in libplayground (#216)

* Initial k8s support in libplayground

* Make tasks never give up + custom event for k8s cluster status

k8s/factory.go

@@ -0,0 +1,140 @@
+package k8s
+
+import (
+	"crypto/tls"
+	"encoding/json"
+	"fmt"
+	"net"
+	"net/http"
+	"net/url"
+	"time"
+
+	"github.com/docker/go-connections/tlsconfig"
+	"github.com/play-with-docker/play-with-docker/pwd/types"
+	"github.com/play-with-docker/play-with-docker/router"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
+)
+
+type FactoryApi interface {
+	GetForInstance(instance *types.Instance) (*kubernetes.Clientset, error)
+	GetKubeletForInstance(instance *types.Instance) (*kubeletClient, error)
+}
+
+func NewClient(instance *types.Instance, proxyHost string) (*kubernetes.Clientset, error) {
+	var durl string
+
+	host := router.EncodeHost(instance.SessionId, instance.RoutableIP, router.HostOpts{EncodedPort: 6443})
+
+	var tlsConfig *tls.Config
+	tlsConfig = tlsconfig.ClientDefault()
+	tlsConfig.InsecureSkipVerify = true
+	tlsConfig.ServerName = host
+
+	var transport http.RoundTripper
+	transport = &http.Transport{
+		DialContext: (&net.Dialer{
+			Timeout:   1 * time.Second,
+			KeepAlive: 30 * time.Second,
+		}).DialContext,
+		TLSClientConfig:     tlsConfig,
+		MaxIdleConnsPerHost: 5,
+	}
+
+	durl = fmt.Sprintf("https://%s", proxyHost)
+
+	cc := rest.ContentConfig{
+		ContentType:          "application/json",
+		GroupVersion:         &schema.GroupVersion{Version: "v1"},
+		NegotiatedSerializer: serializer.DirectCodecFactory{CodecFactory: scheme.Codecs},
+	}
+	restConfig := &rest.Config{
+		Host:          durl,
+		APIPath:       "/api/",
+		BearerToken:   "31ada4fd-adec-460c-809a-9e56ceb75269",
+		ContentConfig: cc,
+	}
+
+	transport, err := rest.HTTPWrappersForConfig(restConfig, transport)
+	if err != nil {
+		return nil, fmt.Errorf("Error wrapping transport %v", err)
+	}
+	cli := &http.Client{
+		Transport: transport,
+	}
+
+	rc, err := rest.RESTClientFor(restConfig)
+	rc.Client = cli
+	if err != nil {
+		return nil, fmt.Errorf("Error creating K8s client %v", err)
+	}
+
+	return kubernetes.New(rc), nil
+}
+
+func NewKubeletClient(instance *types.Instance, proxyHost string) (*kubeletClient, error) {
+	var durl string
+
+	host := router.EncodeHost(instance.SessionId, instance.RoutableIP, router.HostOpts{EncodedPort: 10255})
+
+	transport := &http.Transport{
+		DialContext: (&net.Dialer{
+			Timeout:   1 * time.Second,
+			KeepAlive: 30 * time.Second,
+		}).DialContext,
+		MaxIdleConnsPerHost: 5,
+	}
+
+	durl = fmt.Sprintf("http://%s", host)
+	transport.Proxy = http.ProxyURL(&url.URL{Host: proxyHost})
+
+	cli := &http.Client{
+		Transport: transport,
+	}
+	kc := &kubeletClient{client: cli, baseURL: durl}
+	return kc, nil
+}
+
+type kubeletClient struct {
+	client  *http.Client
+	baseURL string
+}
+
+func (c *kubeletClient) Get(path string) (*http.Response, error) {
+	return c.client.Get(c.baseURL + path)
+}
+
+type metadata struct {
+	Labels map[string]string
+}
+
+type item struct {
+	Metadata metadata
+}
+
+type kubeletPodsResponse struct {
+	Items []item
+}
+
+func (c *kubeletClient) IsManager() (bool, error) {
+	res, err := c.client.Get(c.baseURL + "/pods")
+	if err != nil {
+		return false, err
+	}
+	podsData := &kubeletPodsResponse{}
+
+	json.NewDecoder(res.Body).Decode(podsData)
+
+	for _, i := range podsData.Items {
+		for _, v := range i.Metadata.Labels {
+			if v == "kube-apiserver" {
+				return true, nil
+			}
+		}
+	}
+
+	return false, nil
+}

k8s/factory_mock.go

@@ -0,0 +1,21 @@
+package k8s
+
+import (
+	"github.com/play-with-docker/play-with-docker/pwd/types"
+	"github.com/stretchr/testify/mock"
+	"k8s.io/client-go/kubernetes"
+)
+
+type FactoryMock struct {
+	mock.Mock
+}
+
+func (m *FactoryMock) GetKubeletForInstance(i *types.Instance) (*kubeletClient, error) {
+	args := m.Called(i)
+	return args.Get(0).(*kubeletClient), args.Error(1)
+}
+
+func (m *FactoryMock) GetForInstance(instance *types.Instance) (*kubernetes.Clientset, error) {
+	args := m.Called(instance)
+	return args.Get(0).(*kubernetes.Clientset), args.Error(1)
+}

k8s/local_cached_factory.go

@@ -0,0 +1,124 @@
+package k8s
+
+import (
+	"fmt"
+	"log"
+	"sync"
+	"time"
+
+	"github.com/play-with-docker/play-with-docker/pwd/types"
+	"github.com/play-with-docker/play-with-docker/storage"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+)
+
+type localCachedFactory struct {
+	rw              sync.Mutex
+	irw             sync.Mutex
+	sessionClient   *kubernetes.Clientset
+	instanceClients map[string]*instanceEntry
+	storage         storage.StorageApi
+}
+
+type instanceEntry struct {
+	rw            sync.Mutex
+	client        *kubernetes.Clientset
+	kubeletClient *kubeletClient
+}
+
+func (f *localCachedFactory) GetForInstance(instance *types.Instance) (*kubernetes.Clientset, error) {
+	key := instance.Name
+
+	f.irw.Lock()
+	c, found := f.instanceClients[key]
+	if !found {
+		c := &instanceEntry{}
+		f.instanceClients[key] = c
+	}
+	c = f.instanceClients[key]
+	f.irw.Unlock()
+
+	c.rw.Lock()
+	defer c.rw.Unlock()
+
+	if c.client == nil {
+		kc, err := NewClient(instance, "l2:443")
+		if err != nil {
+			return nil, err
+		}
+		c.client = kc
+	}
+
+	err := f.check(func() error {
+		_, err := c.client.CoreV1().Pods("").List(metav1.ListOptions{})
+		return err
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return c.client, nil
+}
+
+func (f *localCachedFactory) GetKubeletForInstance(instance *types.Instance) (*kubeletClient, error) {
+	key := instance.Name
+
+	f.irw.Lock()
+	c, found := f.instanceClients[key]
+	if !found {
+		c := &instanceEntry{}
+		f.instanceClients[key] = c
+	}
+	c = f.instanceClients[key]
+	f.irw.Unlock()
+
+	c.rw.Lock()
+	defer c.rw.Unlock()
+
+	if c.kubeletClient == nil {
+		kc, err := NewKubeletClient(instance, "l2:443")
+		if err != nil {
+			return nil, err
+		}
+		c.kubeletClient = kc
+	}
+
+	err := f.check(func() error {
+		r, err := c.kubeletClient.Get("/pods")
+		if err != nil {
+			return err
+		}
+		defer r.Body.Close()
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return c.kubeletClient, nil
+}
+
+func (f *localCachedFactory) check(fn func() error) error {
+	ok := false
+	for i := 0; i < 5; i++ {
+		err := fn()
+		if err != nil {
+			log.Printf("Connection to k8s api has failed, maybe instance is not ready yet, sleeping and retrying in 1 second. Try #%d. Got: %v\n", i+1, err)
+			time.Sleep(time.Second)
+			continue
+		}
+		ok = true
+		break
+	}
+	if !ok {
+		return fmt.Errorf("Connection to k8s api was not established.")
+	}
+	return nil
+}
+
+func NewLocalCachedFactory(s storage.StorageApi) *localCachedFactory {
+	return &localCachedFactory{
+		instanceClients: make(map[string]*instanceEntry),
+		storage:         s,
+	}
+}