Kube-503 - in-cluster deployment

cmd/proxy/main.go

@@ -1,16 +1,17 @@
 package main
 
 import (
-	"cloud-proxy/internal/cloud/gcp"
-	"cloud-proxy/internal/cloud/gcp/gcpauth"
-	proto "cloud-proxy/proto/v1alpha"
 	"context"
 	"fmt"
 	"net/http"
 	"path"
 	"runtime"
 	"time"
 
+	"cloud-proxy/internal/cloud/gcp"
+	"cloud-proxy/internal/cloud/gcp/gcpauth"
+	proto "cloud-proxy/proto/v1alpha"
+
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/backoff"
 	"google.golang.org/grpc/credentials"
@@ -29,6 +30,7 @@ var (
 )
 
 func main() {
+	logrus.Info("Starting proxy")
 	cfg := config.Get()
 
 	logger := logrus.New()
@@ -45,33 +47,44 @@ func main() {
 		"GitCommit": GitCommit,
 		"GitRef":    GitRef,
 		"Version":   Version,
-	}).Println("Starting cloud-proxy")
+	}).Info("Starting cloud-proxy")
 
 	dialOpts := make([]grpc.DialOption, 0)
-	dialOpts = append(dialOpts, grpc.WithConnectParams(grpc.ConnectParams{
+	if cfg.CastAI.DisableGRPCTLS {
+		// ONLY For testing purposes
+		dialOpts = append(dialOpts, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	} else {
+		dialOpts = append(dialOpts, grpc.WithTransportCredentials(credentials.NewTLS(nil)))
+	}
+
+	connectParams := grpc.ConnectParams{
 		Backoff: backoff.Config{
 			BaseDelay:  2 * time.Second,
 			Jitter:     0.1,
 			MaxDelay:   5 * time.Second,
 			Multiplier: 1.2,
 		},
-	}))
-	if cfg.CastAI.DisableGRPCTLS {
-		// ONLY For testing purposes
-		dialOpts = append(dialOpts, grpc.WithTransportCredentials(insecure.NewCredentials()))
-	} else {
-		dialOpts = append(dialOpts, grpc.WithTransportCredentials(credentials.NewTLS(nil)))
 	}
+	dialOpts = append(dialOpts, grpc.WithConnectParams(connectParams))
 
+	logger.Infof(
+		"Creating grpc channel against (%s) with connection config (%v) and TLS enabled=%v",
+		cfg.CastAI.GrpcURL,
+		connectParams,
+		!cfg.CastAI.DisableGRPCTLS,
+	)
 	conn, err := grpc.NewClient(cfg.CastAI.GrpcURL, dialOpts...)
 	if err != nil {
 		logger.Panicf("Failed to connect to server: %v", err)
+		panic(err)
 	}
 
 	defer func(conn *grpc.ClientConn) {
+		logger.Info("Closing grpc connection")
 		err := conn.Close()
 		if err != nil {
 			logger.Panicf("Failed to close gRPC connection: %v", err)
+			panic(err)
 		}
 	}(conn)
 
@@ -83,6 +96,7 @@ func main() {
 	err = client.Run(ctx, proto.NewCloudProxyAPIClient(conn))
 	if err != nil {
 		logger.Panicf("Failed to run client: %v", err)
+		panic(err)
 	}
 }
 

dummy_deploy.yaml

@@ -1,3 +1,4 @@
+# TODO: Remove this; just local testing for now. (or make it polished)
 apiVersion: apps/v1
 kind: Deployment
 metadata:
@@ -30,9 +31,36 @@ spec:
           args:
             - "-sanity-checks=false"
             - "-mockcast=true"
+          env:
+            - name: POD_NAME
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.name
+            - name: POD_NAMESPACE
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.namespace
+            - name: POD_IP
+              valueFrom:
+                fieldRef:
+                  fieldPath: status.podIP
+            - name: NODE_NAME
+              valueFrom:
+                fieldRef:
+                  fieldPath: spec.nodeName
+            - name: CAST_URL
+              value: https://api.dev-master.cast.ai
+            - name: CAST_GRPC_URL
+              value: api-grpc.dev.cast.ai
+            - name: CAST_API_KEY
+              value: ec82f1670aa140435ab483a3ff2e5f507d67f3a4e276197e6bb9339f1eda9350
+            - name: CLUSTER_ID
+              value: b8c18bba-ff52-442e-9f44-8dcc9becfc58
+            - name: LOG_LEVEL
+              value: "5"
 ---
 apiVersion: v1
 kind: ServiceAccount
 metadata:
   name: castai-cloud-proxy
-  namespace: default
+  namespace: default

internal/config/config.go

@@ -3,6 +3,7 @@ package config
 import (
 	"fmt"
 
+	"github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
 )
 
@@ -13,7 +14,7 @@ type Config struct {
 
 	//MetricsAddress  string               `mapstructure:"metricsaddress"`
 	//HealthAddress   string               `mapstructure:"healthaddress"`
-	Log Log
+	Log Log `mapstructure:"log"`
 }
 
 // PodMetadata stores metadata for the pod, mostly used for logging and debugging purposes.
@@ -45,12 +46,8 @@ type GCP struct {
 	UseMetadataServer bool
 }
 
-type TLSConfig struct {
-	Enabled bool
-}
-
 type Log struct {
-	Level int
+	Level int `mapstructure:"level"`
 }
 
 var cfg *Config = nil
@@ -74,13 +71,30 @@ func Get() Config {
 	v.MustBindEnv("podmetadata.nodename", "NODE_NAME")
 	v.MustBindEnv("podmetadata.podname", "POD_NAME")
 
-	// TODO: Logging
+	_ = v.BindEnv("log.level", "LOG_LEVEL")
 
 	cfg = &Config{}
 	if err := v.Unmarshal(cfg); err != nil {
 		panic(fmt.Errorf("while parsing config: %w", err))
 	}
 
+	if cfg.CastAI.ApiKey == "" {
+		required("CAST_API_KEY")
+	}
+	if cfg.CastAI.GrpcURL == "" {
+		required("CAST_GRPC_URL")
+	}
+	if cfg.CastAI.URL == "" {
+		required("CAST_URL")
+	}
+	if cfg.ClusterID == "" {
+		required("CLUSTER_ID")
+	}
+
+	if cfg.Log.Level == 0 {
+		cfg.Log.Level = int(logrus.InfoLevel)
+	}
+
 	return *cfg
 }
 

internal/config/config_test.go

@@ -26,6 +26,8 @@ func TestConfig(t *testing.T) {
 	t.Setenv("POD_NAME", "heavy-worker")
 	t.Setenv("NODE_NAME", "awesome-node")
 
+	t.Setenv("LOG_LEVEL", "3")
+
 	expected := Config{
 		CastAI: CastAPI{
 			ApiKey:         "API_KEY",
@@ -40,6 +42,9 @@ func TestConfig(t *testing.T) {
 			NodeName:     "awesome-node",
 			PodName:      "heavy-worker",
 		},
+		Log: Log{
+			Level: 3,
+		},
 	}
 
 	got := Get()

internal/proxy/client.go

@@ -4,16 +4,17 @@ package proxy
 
 import (
 	"bytes"
-	cloudproxyv1alpha "cloud-proxy/proto/v1alpha"
-	proto "cloud-proxy/proto/v1alpha"
 	"context"
 	"fmt"
-	"github.com/samber/lo"
-	"github.com/sirupsen/logrus"
 	"io"
 	"net/http"
 	"sync/atomic"
 	"time"
+
+	cloudproxyv1alpha "cloud-proxy/proto/v1alpha"
+	proto "cloud-proxy/proto/v1alpha"
+	"github.com/samber/lo"
+	"github.com/sirupsen/logrus"
 )
 
 const (
@@ -59,20 +60,22 @@ func (c *Client) Run(ctx context.Context, grpcClient cloudproxyv1alpha.CloudProx
 		if ctx.Err() != nil {
 			return nil
 		}
+		c.log.Info("Starting proxy client")
 		err := c.run(ctx, grpcClient)
 		if err != nil {
-			c.log.Printf("c.run: %v", err)
+			c.log.Printf("c.run exited: %v", err)
 		}
 	}
 }
 
 func (c *Client) initializeStream(ctx context.Context, proxyCastAIClient cloudproxyv1alpha.CloudProxyAPIClient) (StreamCloudProxyClient, error) {
-	c.log.Println("Connecting to castai")
+	c.log.Info("Connecting to castai")
 
 	stream, err := proxyCastAIClient.StreamCloudProxy(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("proxyCastAIClient.StreamCloudProxy: %w", err)
 	}
+	c.log.Info("Connected to castai, sending initial metadata")
 
 	err = stream.Send(&cloudproxyv1alpha.StreamCloudProxyRequest{
 		Request: &cloudproxyv1alpha.StreamCloudProxyRequest_InitialRequest{
@@ -89,6 +92,8 @@ func (c *Client) initializeStream(ctx context.Context, proxyCastAIClient cloudpr
 	}
 	c.lastSeen.Store(time.Now().UnixNano())
 
+	c.log.Info("Stream to castai started successfully")
+
 	return stream, nil
 }
 
@@ -101,7 +106,7 @@ func (c *Client) run(ctx context.Context, grpcClient cloudproxyv1alpha.CloudProx
 	defer func() {
 		err := stream.CloseSend()
 		if err != nil {
-			c.log.Println("error closing stream", err)
+			c.log.Errorf("error closing stream: %v", err)
 		}
 	}()
 
@@ -113,11 +118,13 @@ func (c *Client) run(ctx context.Context, grpcClient cloudproxyv1alpha.CloudProx
 		if !c.isAlive() {
 			return fmt.Errorf("last seen too old, closing stream")
 		}
+		c.log.Info("Polling stream for messages")
 		in, err := stream.Recv()
 		if err != nil {
 			return fmt.Errorf("stream.Recv: %w", err)
 		}
 
+		c.log.Info("Handling message from castai")
 		go c.handleMessage(in, stream)
 	}
 }
@@ -128,10 +135,17 @@ func (c *Client) handleMessage(in *cloudproxyv1alpha.StreamCloudProxyResponse, s
 	// skip processing http request if keep alive message
 	if in.GetMessageId() == KeepAliveMessageID {
 		c.lastSeen.Store(time.Now().UnixNano())
+		c.log.Debugf("Received keep-alive message from castai for %s", in.GetClientMetadata().GetClusterId())
 		return
 	}
 
+	c.log.Debugf("Received request for proxying msg_id=%v from castai", in.GetMessageId())
 	resp := c.processHttpRequest(in.GetHttpRequest())
+	if resp.GetError() != "" {
+		c.log.Errorf("Failed to proxy request msg_id=%v with %v", in.GetMessageId(), resp.GetError())
+	} else {
+		c.log.Debugf("Proxied request msg_id=%v, sending response to castai", in.GetMessageId())
+	}
 	err := stream.Send(&cloudproxyv1alpha.StreamCloudProxyRequest{
 		Request: &cloudproxyv1alpha.StreamCloudProxyRequest_Response{
 			Response: &cloudproxyv1alpha.ClusterResponse{
@@ -196,14 +210,18 @@ func (c *Client) sendKeepAlive(ctx context.Context, stream StreamCloudProxyClien
 	ticker := time.NewTimer(time.Duration(c.keepAlive.Load()))
 	defer ticker.Stop()
 
+	c.log.Info("Starting keep-alive loop")
 	for {
 		if !c.isAlive() {
+			c.log.Info("Stopping keep-alive loop: client connection is not alive")
 			return
 		}
 		select {
 		case <-ctx.Done():
+			c.log.Infof("Stopping keep-alive loop: context ended with %v", context.Cause(ctx))
 			return
 		case <-ticker.C:
+			c.log.Debug("Sending keep-alive to castai")
 			err := stream.Send(&cloudproxyv1alpha.StreamCloudProxyRequest{
 				Request: &cloudproxyv1alpha.StreamCloudProxyRequest_ClientStats{
 					ClientStats: &cloudproxyv1alpha.ClientStats{