k8s port forwarding

Author: Forest33

Diff for: README.md

@@ -17,7 +17,9 @@
 - `.proto` file discovery
 - Selection of multiple services and methods
 - Configuration of TLS, including disabling TLS (plain text)
-- Authentication: Basic, Bearer Token, JWT, GCE 
+- Authentication: Basic, Bearer Token, JWT, GCE
+- Kubernetes port forwarding
+- Authorization in Google Cloud services
 - Input generation for all scalar types
 - Input generation for nested and looped messages 
 - Input generation for enums, including nested
@@ -55,4 +57,11 @@ the `Applications` folder and run from `Applications`.
 
 ### Windows
 
-[Download](https://github.com/Forest33/warthog/releases) and run `Warthog*-windows-x86-64.exe`.
+[Download](https://github.com/Forest33/warthog/releases) and run `Warthog*-windows-x86-64.exe`.
+
+## Google Cloud services authorization
+- Enable Kubernetes Engine API and check quota for your project at [https://console.developers.google.com/apis/api/container](https://console.developers.google.com/apis/api/container)
+- Install gcloud CLI from [https://cloud.google.com/sdk/](https://cloud.google.com/sdk/) and run
+````
+  gcloud beta auth application-default login
+````

Diff for: adapter/database/settings.go

@@ -13,7 +13,7 @@ import (
 const (
 	settingsTable       = "settings"
 	settingsTableFields = `window_width, window_height, window_x, window_y, single_instance, connect_timeout,
-							request_timeout, non_blocking_connection, sort_methods_by_name, max_loop_depth`
+							request_timeout, k8s_request_timeout, non_blocking_connection, sort_methods_by_name, max_loop_depth`
 )
 
 // SettingsRepository object capable of interacting with SettingsRepository
@@ -38,6 +38,7 @@ type settingsDTO struct {
 	SingleInstance        bool `db:"single_instance"`
 	ConnectTimeout        int  `db:"connect_timeout"`
 	RequestTimeout        int  `db:"request_timeout"`
+	K8SRequestTimeout     int  `db:"k8s_request_timeout"`
 	NonBlockingConnection bool `db:"non_blocking_connection"`
 	SortMethodsByName     bool `db:"sort_methods_by_name"`
 	MaxLoopDepth          int  `db:"max_loop_depth"`
@@ -52,6 +53,7 @@ func (dto *settingsDTO) entity() *entity.Settings {
 		SingleInstance:        &dto.SingleInstance,
 		ConnectTimeout:        &dto.ConnectTimeout,
 		RequestTimeout:        &dto.RequestTimeout,
+		K8SRequestTimeout:     &dto.K8SRequestTimeout,
 		NonBlockingConnection: &dto.NonBlockingConnection,
 		SortMethodsByName:     &dto.SortMethodsByName,
 		MaxLoopDepth:          &dto.MaxLoopDepth,
@@ -74,7 +76,7 @@ func (repo *SettingsRepository) Get() (*entity.Settings, error) {
 func (repo *SettingsRepository) Update(in *entity.Settings) (*entity.Settings, error) {
 	dto := &settingsDTO{}
 	attrs := make([]string, 0, 10)
-	mapper := make(map[string]interface{}, 10)
+	mapper := make(map[string]interface{}, 11)
 
 	if in.WindowWidth > 0 {
 		attrs = append(attrs, "window_width = :window_width")
@@ -104,6 +106,10 @@ func (repo *SettingsRepository) Update(in *entity.Settings) (*entity.Settings, e
 		attrs = append(attrs, "request_timeout = :request_timeout")
 		mapper["request_timeout"] = in.RequestTimeout
 	}
+	if in.K8SRequestTimeout != nil {
+		attrs = append(attrs, "k8s_request_timeout = :k8s_request_timeout")
+		mapper["k8s_request_timeout"] = in.K8SRequestTimeout
+	}
 	if in.NonBlockingConnection != nil {
 		attrs = append(attrs, "non_blocking_connection = :non_blocking_connection")
 		mapper["non_blocking_connection"] = in.NonBlockingConnection

Diff for: adapter/grpc/client.go

@@ -34,7 +34,7 @@ type Client struct {
 	queryCtx         context.Context
 	queryCancel      context.CancelFunc
 	queryStartTime   time.Time
-	cancelQueryMux   sync.Mutex
+	connectionMux    sync.RWMutex
 	requestCh        chan *dynamic.Message
 	responseCh       chan *entity.QueryResponse
 	closeStreamCh    chan struct{}
@@ -90,7 +90,11 @@ func (c *Client) Connect(addr string, auth *entity.Auth, opts ...ClientOpt) erro
 		dialOptions = append(dialOptions, grpc.WithBlock())
 		if *c.cfg.ConnectTimeout > 0 {
 			ctx, cancel = context.WithTimeout(c.ctx, time.Second*time.Duration(*c.cfg.ConnectTimeout))
-			defer cancel()
+			defer func() {
+				c.connectionMux.Lock()
+				cancel()
+				c.connectionMux.Unlock()
+			}()
 		}
 	}
 
@@ -139,9 +143,19 @@ func (c *Client) loadTLSCredentials() (credentials.TransportCredentials, error)
 
 // Close closes connection to gRPC server
 func (c *Client) Close() {
+	c.connectionMux.Lock()
+	defer c.connectionMux.Unlock()
+
 	if c.conn != nil {
 		if err := c.conn.Close(); err != nil {
 			c.log.Error().Msgf("failed to close connection: %v", err)
 		}
 	}
 }
+
+func (c *Client) isConnected() bool {
+	c.connectionMux.RLock()
+	defer c.connectionMux.RUnlock()
+
+	return c.conn != nil
+}

Diff for: adapter/grpc/query.go

@@ -71,11 +71,15 @@ func (c *Client) createMessage(method *entity.Method, data map[string]interface{
 }
 
 // Query executes a gRPC request
-func (c *Client) Query(method *entity.Method, data map[string]interface{}, metadata []string) {
+func (c *Client) Query(method *entity.Method, data map[string]interface{}, metadata []string) error {
+	if !c.isConnected() {
+		return entity.ErrNotConnected
+	}
+
 	ms, err := c.createMessage(method, data, metadata)
 	if err != nil {
 		c.responseError(err, "")
-		return
+		return err
 	}
 
 	var isNew bool
@@ -92,12 +96,14 @@ func (c *Client) Query(method *entity.Method, data map[string]interface{}, metad
 	}
 
 	if err != nil {
-		return
+		return err
 	}
 
 	if isNew || method.Type == entity.MethodTypeClientStream || method.Type == entity.MethodTypeBidiStream {
 		c.request(ms)
 	}
+
+	return nil
 }
 
 // CancelQuery aborting a running gRPC request

Diff for: adapter/k8s/auth.go

@@ -0,0 +1,77 @@
+package k8s
+
+import (
+	"context"
+	"encoding/base64"
+	"fmt"
+
+	"golang.org/x/oauth2"
+	auth "golang.org/x/oauth2/google"
+	"google.golang.org/api/container/v1"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/client-go/tools/clientcmd/api"
+
+	"github.com/forest33/warthog/business/entity"
+)
+
+var gcsScopes = []string{"https://www.googleapis.com/auth/cloud-platform"}
+
+func (c *Client) gcsAuth(ctx context.Context, r *entity.GCSAuth) (*rest.Config, error) {
+	var token *oauth2.Token
+
+	credentials, err := auth.FindDefaultCredentials(ctx, gcsScopes...)
+	if err == nil {
+		token, err = credentials.TokenSource.Token()
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	containerService, _ := container.NewService(ctx)
+
+	name := fmt.Sprintf("projects/%s/locations/%s/clusters/%s", r.Project, r.Location, r.Cluster)
+	resp, err := containerService.Projects.Locations.Clusters.Get(name).Do()
+	if err != nil {
+		return nil, err
+	}
+
+	cert, err := base64.StdEncoding.DecodeString(resp.MasterAuth.ClusterCaCertificate)
+	if err != nil {
+		return nil, err
+	}
+
+	apiConfig := &api.Config{
+		APIVersion: "v1",
+		Kind:       "Config",
+		Clusters: map[string]*api.Cluster{
+			r.Cluster: {
+				CertificateAuthorityData: cert,
+				Server:                   fmt.Sprintf("https://%s", resp.Endpoint),
+			},
+		},
+		Contexts: map[string]*api.Context{
+			r.Cluster: {
+				Cluster:  r.Cluster,
+				AuthInfo: r.Cluster,
+			},
+		},
+		CurrentContext: r.Cluster,
+	}
+
+	restConfig, err := clientcmd.NewNonInteractiveClientConfig(
+		*apiConfig,
+		r.Cluster,
+		&clientcmd.ConfigOverrides{
+			CurrentContext: r.Cluster,
+		},
+		nil,
+	).ClientConfig()
+	if err != nil {
+		return nil, err
+	}
+
+	restConfig.BearerToken = token.AccessToken
+
+	return restConfig, nil
+}

Diff for: adapter/k8s/client.go

@@ -0,0 +1,172 @@
+package k8s
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net/http"
+	"net/url"
+	"os"
+	"path/filepath"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/client-go/tools/portforward"
+	"k8s.io/client-go/transport/spdy"
+
+	"github.com/forest33/warthog/business/entity"
+	"github.com/forest33/warthog/pkg/logger"
+)
+
+// Client object capable of interacting with Client
+type Client struct {
+	ctx context.Context
+	cfg *entity.Settings
+	log *logger.Zerolog
+}
+
+// New creates a new Client
+func New(ctx context.Context, log *logger.Zerolog) *Client {
+	return &Client{
+		ctx: ctx,
+		log: log,
+	}
+}
+
+// SetSettings sets application settings
+func (c *Client) SetSettings(cfg *entity.Settings) {
+	c.cfg = cfg
+}
+
+// PortForward port forward
+func (c *Client) PortForward(r *entity.K8SPortForward) (entity.PortForwardControl, error) {
+	config, client, err := c.createClient(r.ClientConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	var (
+		podName string
+	)
+
+	if r.PodName != "" {
+		podName = r.PodName
+	} else if r.PodNameSelector != "" {
+		if podName, err = c.findPod(client, r.Namespace, r.PodNameSelector); err != nil {
+			return nil, err
+		}
+	} else {
+		return nil, entity.ErrK8SPodNotFound
+	}
+
+	ctrl := &PortForwardControl{
+		stopCh: make(chan struct{}, 1),
+		out:    &bytes.Buffer{},
+		errOut: &bytes.Buffer{},
+	}
+
+	readyCh := make(chan struct{})
+
+	writeError := func(err error) {
+		if _, err := ctrl.errOut.Write([]byte(err.Error())); err != nil {
+			c.log.Error().Msgf("failed write to error stream: %v", err)
+		}
+	}
+
+	if r.ErrHandler != nil {
+		runtime.ErrorHandlers = []func(error){r.ErrHandler}
+	}
+
+	go func() {
+		path := fmt.Sprintf("/api/v1/namespaces/%s/pods/%s/portforward", r.Namespace, podName)
+
+		u, err := url.Parse(config.Host)
+		if err != nil {
+			writeError(err)
+			return
+		}
+
+		transport, upgrader, err := spdy.RoundTripperFor(config)
+		if err != nil {
+			writeError(err)
+			return
+		}
+
+		dialer := spdy.NewDialer(upgrader, &http.Client{Transport: transport}, http.MethodPost, &url.URL{Scheme: "https", Path: path, Host: u.Host})
+		fw, err := portforward.New(dialer, []string{fmt.Sprintf("%d:%d", r.LocalPort, r.PodPort)}, ctrl.stopCh, readyCh, ctrl.out, ctrl.errOut)
+		if err != nil {
+			writeError(err)
+			return
+		}
+
+		if err := fw.ForwardPorts(); err != nil {
+			writeError(err)
+		}
+	}()
+
+	<-readyCh
+
+	return ctrl, nil
+}
+
+func (c *Client) createClient(cfg *entity.K8SClientConfig) (*rest.Config, *kubernetes.Clientset, error) {
+	var (
+		restConfig *rest.Config
+		err        error
+	)
+
+	if cfg.GCSAuth != nil && cfg.GCSAuth.Enabled {
+		restConfig, err = c.gcsAuth(c.ctx, cfg.GCSAuth)
+		if err != nil {
+			return nil, nil, err
+		}
+	} else {
+		kubeConfig := cfg.KubeConfigFile
+		if kubeConfig == "" {
+			if home := homeDir(); home != "" {
+				kubeConfig = filepath.Join(home, ".kube", "config")
+			}
+		}
+
+		restConfig, err = clientcmd.BuildConfigFromFlags("", kubeConfig)
+		if err != nil {
+			return nil, nil, err
+		}
+
+		restConfig.BearerToken = cfg.BearerToken
+	}
+
+	client, err := kubernetes.NewForConfig(restConfig)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return restConfig, client, nil
+}
+
+func (c *Client) findPod(client *kubernetes.Clientset, namespace, selector string) (string, error) {
+	ctx, cancel := context.WithTimeout(c.ctx, time.Duration(*c.cfg.K8SRequestTimeout)*time.Second)
+	defer cancel()
+
+	pods, err := client.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{LabelSelector: selector})
+	if err != nil {
+		return "", err
+	}
+
+	if len(pods.Items) == 0 {
+		return "", entity.ErrK8SPodNotFound
+	}
+
+	return pods.Items[0].Name, nil
+}
+
+func homeDir() string {
+	if h := os.Getenv("HOME"); h != "" {
+		return h
+	}
+	return os.Getenv("USERPROFILE")
+}