New Demo image (#370)

# TL;DR
This will let flytectl demo use the new bundled sandbox image instead.  See the issue for additional information.

`flytectl demo start`
* Brings up the new container
* kubeconfig will now be published to `~/.flyte/state/kubeconfig` (but the context will still be copied to the user's main kubeconfig with "flyte-sandbox" as the context name).

`flytectl demo reload`
Kills the Flyte pod, allowing the new one to come up.

Signed-off-by: Yee Hing Tong <[email protected]>

cmd/config/subcommand/sandbox/sandbox_config.go

@@ -4,7 +4,7 @@ import "github.com/flyteorg/flytectl/pkg/docker"
 
 // Config holds configuration flags for sandbox command.
 type Config struct {
-	Source string `json:"source" pflag:",Path of your source code"`
+	DeprecatedSource string `json:"source" pflag:",deprecated, path of your source code, please build images with local daemon"`
 
 	// Flytectl sandbox only supports Flyte version available in Github release https://github.com/flyteorg/flyte/tags.
 	// Flytectl sandbox will only work for v0.10.0+.

cmd/demo/demo.go

@@ -6,6 +6,11 @@ import (
 	"github.com/spf13/cobra"
 )
 
+const (
+	flyteNs     = "flyte"
+	K8sEndpoint = "https://127.0.0.1:6443"
+)
+
 // Long descriptions are whitespace sensitive when generating docs using sphinx.
 const (
 	demoShort = `Helps with demo interactions like start, teardown, status, and exec.`
@@ -47,6 +52,9 @@ func CreateDemoCommand() *cobra.Command {
 		"start": {CmdFunc: startDemoCluster, Aliases: []string{}, ProjectDomainNotRequired: true,
 			Short: startShort,
 			Long:  startLong, PFlagProvider: sandboxCmdConfig.DefaultConfig, DisableFlyteClient: true},
+		"reload": {CmdFunc: reloadDemoCluster, Aliases: []string{}, ProjectDomainNotRequired: true,
+			Short: reloadShort,
+			Long:  reloadLong, PFlagProvider: sandboxCmdConfig.DefaultConfig, DisableFlyteClient: true},
 		"teardown": {CmdFunc: teardownDemoCluster, Aliases: []string{}, ProjectDomainNotRequired: true,
 			Short: teardownShort,
 			Long:  teardownLong, DisableFlyteClient: true},

cmd/demo/demo_test.go

@@ -13,7 +13,8 @@ func TestCreateDemoCommand(t *testing.T) {
 	assert.Equal(t, demoCommand.Use, "demo")
 	assert.Equal(t, demoCommand.Short, "Helps with demo interactions like start, teardown, status, and exec.")
 	fmt.Println(demoCommand.Commands())
-	assert.Equal(t, len(demoCommand.Commands()), 4)
+
+	assert.Equal(t, len(demoCommand.Commands()), 5)
 	cmdNouns := demoCommand.Commands()
 	// Sort by Use value.
 	sort.Slice(cmdNouns, func(i, j int) bool {
@@ -24,16 +25,19 @@ func TestCreateDemoCommand(t *testing.T) {
 	assert.Equal(t, cmdNouns[0].Short, execShort)
 	assert.Equal(t, cmdNouns[0].Long, execLong)
 
-	assert.Equal(t, cmdNouns[1].Use, "start")
-	assert.Equal(t, cmdNouns[1].Short, startShort)
-	assert.Equal(t, cmdNouns[1].Long, startLong)
+	assert.Equal(t, cmdNouns[1].Use, "reload")
+	assert.Equal(t, cmdNouns[1].Short, reloadShort)
+	assert.Equal(t, cmdNouns[1].Long, reloadLong)
 
-	assert.Equal(t, cmdNouns[2].Use, "status")
-	assert.Equal(t, cmdNouns[2].Short, statusShort)
-	assert.Equal(t, cmdNouns[2].Long, statusLong)
+	assert.Equal(t, cmdNouns[2].Use, "start")
+	assert.Equal(t, cmdNouns[2].Short, startShort)
+	assert.Equal(t, cmdNouns[2].Long, startLong)
 
-	assert.Equal(t, cmdNouns[3].Use, "teardown")
-	assert.Equal(t, cmdNouns[3].Short, teardownShort)
-	assert.Equal(t, cmdNouns[3].Long, teardownLong)
+	assert.Equal(t, cmdNouns[3].Use, "status")
+	assert.Equal(t, cmdNouns[3].Short, statusShort)
+	assert.Equal(t, cmdNouns[3].Long, statusLong)
 
+	assert.Equal(t, cmdNouns[4].Use, "teardown")
+	assert.Equal(t, cmdNouns[4].Short, teardownShort)
+	assert.Equal(t, cmdNouns[4].Long, teardownLong)
 }

cmd/demo/reload.go

@@ -0,0 +1,58 @@
+package demo
+
+import (
+	"context"
+	"fmt"
+
+	cmdCore "github.com/flyteorg/flytectl/cmd/core"
+	"github.com/flyteorg/flytectl/pkg/docker"
+	"github.com/flyteorg/flytectl/pkg/k8s"
+	"github.com/flyteorg/flytestdlib/logger"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+const (
+	labelSelector = "app=flyte"
+)
+const (
+	reloadShort = "Power cycle the Flyte executable pod, effectively picking up an updated config."
+	reloadLong  = `
+If you've changed the ~/.flyte/state/flyte.yaml file, run this command to restart the Flyte binary pod, effectively
+picking up the new settings:
+
+Usage
+::
+
+ flytectl demo reload
+
+`
+)
+
+// reloadDemoCluster will kill the flyte binary pod so the new one can pick up a new config file
+func reloadDemoCluster(ctx context.Context, args []string, cmdCtx cmdCore.CommandContext) error {
+	k8sClient, err := k8s.GetK8sClient(docker.Kubeconfig, K8sEndpoint)
+	if err != nil {
+		fmt.Println("Could not get K8s client")
+		return err
+	}
+	pi := k8sClient.CoreV1().Pods(flyteNs)
+	podList, err := pi.List(ctx, v1.ListOptions{LabelSelector: labelSelector})
+	if err != nil {
+		fmt.Println("could not list pods")
+		return err
+	}
+	if len(podList.Items) != 1 {
+		return fmt.Errorf("should only have one pod running, %d found, %v", len(podList.Items), podList.Items)
+	}
+	logger.Debugf(ctx, "Found %d pods\n", len(podList.Items))
+	var grace = int64(0)
+	err = pi.Delete(ctx, podList.Items[0].Name, v1.DeleteOptions{
+		GracePeriodSeconds: &grace,
+	})
+	if err != nil {
+		fmt.Printf("Could not delete Flyte pod, old configuration may still be in effect. Err: %s\n", err)
+		return err
+	}
+
+	return nil
+}

cmd/demo/reload_test.go

@@ -0,0 +1,51 @@
+package demo
+
+import (
+	"context"
+	"testing"
+
+	cmdCore "github.com/flyteorg/flytectl/cmd/core"
+	"github.com/flyteorg/flytectl/pkg/k8s"
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	testclient "k8s.io/client-go/kubernetes/fake"
+)
+
+var fakePod = corev1.Pod{
+	Status: corev1.PodStatus{
+		Phase:      corev1.PodRunning,
+		Conditions: []corev1.PodCondition{},
+	},
+	ObjectMeta: metav1.ObjectMeta{
+		Name:   "dummyflytepod",
+		Labels: map[string]string{"app": "flyte"},
+	},
+}
+
+func TestDemoReload(t *testing.T) {
+	ctx := context.Background()
+	commandCtx := cmdCore.CommandContext{}
+
+	t.Run("No errors", func(t *testing.T) {
+		client := testclient.NewSimpleClientset()
+		_, err := client.CoreV1().Pods("flyte").Create(ctx, &fakePod, v1.CreateOptions{})
+		assert.NoError(t, err)
+		k8s.Client = client
+		err = reloadDemoCluster(ctx, []string{}, commandCtx)
+		assert.NoError(t, err)
+	})
+
+	t.Run("Multiple pods will error", func(t *testing.T) {
+		client := testclient.NewSimpleClientset()
+		_, err := client.CoreV1().Pods("flyte").Create(ctx, &fakePod, v1.CreateOptions{})
+		assert.NoError(t, err)
+		fakePod.SetName("othername")
+		_, err = client.CoreV1().Pods("flyte").Create(ctx, &fakePod, v1.CreateOptions{})
+		assert.NoError(t, err)
+		k8s.Client = client
+		err = reloadDemoCluster(ctx, []string{}, commandCtx)
+		assert.Errorf(t, err, "should only have one pod")
+	})
+}

cmd/demo/start.go

@@ -3,6 +3,8 @@ package demo
 import (
 	"context"
 
+	"github.com/flyteorg/flytectl/pkg/docker"
+
 	"github.com/flyteorg/flytectl/pkg/sandbox"
 
 	sandboxCmdConfig "github.com/flyteorg/flytectl/cmd/config/subcommand/sandbox"
@@ -75,18 +77,21 @@ eg : for passing multiple environment variables
  flytectl demo start --env USER=foo --env PASSWORD=bar
 
 
-For just printing the docker commands for bringingup the demo container
+For just printing the docker commands for bringing up the demo container
 ::
 
  flytectl demo start --dryRun
 
-
 Usage
 `
 )
 
 func startDemoCluster(ctx context.Context, args []string, cmdCtx cmdCore.CommandContext) error {
-	sandboxDefaultConfig := sandboxCmdConfig.DefaultConfig
-	return sandbox.StartDemoCluster(ctx, args, sandboxDefaultConfig)
+	cfg := sandboxCmdConfig.DefaultConfig
+	err := cfg.ImagePullPolicy.Set(docker.ImagePullPolicyIfNotPresent.String())
+	if err != nil {
+		return err
+	}
 
+	return sandbox.StartDemoCluster(ctx, args, cfg)
 }

cmd/register/files.go

@@ -151,7 +151,7 @@ func Register(ctx context.Context, args []string, cfg *config.Config, cmdCtx cmd
 			return fmt.Errorf("failed to upload source code from [%v]. Error: %w", sourceCodePath, err)
 		}
 
-		logger.Infof(ctx, "Source code successfully uploaded to [%v]", uploadLocation)
+		logger.Infof(ctx, "DeprecatedSource code successfully uploaded to [%v]", uploadLocation)
 	}
 
 	var registerResults []Result

go.mod

@@ -3,7 +3,6 @@ module github.com/flyteorg/flytectl
 go 1.18
 
 require (
-	github.com/avast/retry-go v3.0.0+incompatible
 	github.com/awalterschulze/gographviz v2.0.3+incompatible
 	github.com/disiqueira/gotree v1.0.0
 	github.com/docker/docker v20.10.7+incompatible
@@ -44,6 +43,7 @@ require (
 )
 
 require (
+	github.com/avast/retry-go v3.0.0+incompatible
 	github.com/flyteorg/flytepropeller v1.1.1
 	golang.org/x/text v0.3.7
 )

pkg/docker/docker.go

@@ -25,6 +25,8 @@ type Docker interface {
 	ContainerExecAttach(ctx context.Context, execID string, config types.ExecStartCheck) (types.HijackedResponse, error)
 	ContainerExecInspect(ctx context.Context, execID string) (types.ContainerExecInspect, error)
 	ImageList(ctx context.Context, listOption types.ImageListOptions) ([]types.ImageSummary, error)
+	ContainerStatPath(ctx context.Context, containerID, path string) (types.ContainerPathStat, error)
+	CopyFromContainer(ctx context.Context, containerID, srcPath string) (io.ReadCloser, types.ContainerPathStat, error)
 }
 
 type FlyteDocker struct {

pkg/docker/docker_util.go

@@ -24,11 +24,14 @@ import (
 )
 
 var (
-	Kubeconfig              = f.FilePathJoin(f.UserHomeDir(), ".flyte", "k3s", "k3s.yaml")
+	FlyteStateDir           = f.FilePathJoin(f.UserHomeDir(), ".flyte", "state")
+	Kubeconfig              = f.FilePathJoin(FlyteStateDir, "kubeconfig")
+	SandboxKubeconfig       = f.FilePathJoin(f.UserHomeDir(), ".flyte", "k3s", "k3s.yaml")
 	SuccessMessage          = "Deploying Flyte..."
 	FlyteSandboxClusterName = "flyte-sandbox"
 	Environment             = []string{"SANDBOX=1", "KUBERNETES_API_PORT=30086", "FLYTE_HOST=localhost:30081", "FLYTE_AWS_ENDPOINT=http://localhost:30084"}
 	Source                  = "/root"
+	StateDirMountDest       = "/srv/flyte"
 	K3sDir                  = "/etc/rancher/"
 	Client                  Docker
 	Volumes                 = []mount.Mount{
@@ -126,20 +129,18 @@ func GetSandboxPorts() (map[nat.Port]struct{}, map[nat.Port][]nat.PortBinding, e
 // GetDemoPorts will return demo ports
 func GetDemoPorts() (map[nat.Port]struct{}, map[nat.Port][]nat.PortBinding, error) {
 	return nat.ParsePortSpecs([]string{
-		"0.0.0.0:30080:30080", // Flyteconsole Port
-		"0.0.0.0:30081:30081", // Flyteadmin Port
-		"0.0.0.0:30082:30082", // K8s Dashboard Port
-		"0.0.0.0:30084:30084", // Minio API Port
-		"0.0.0.0:30086:30086", // K8s cluster
-		"0.0.0.0:30088:30088", // Minio Console Port
-		"0.0.0.0:30089:30089", // Postgres Port
-		"0.0.0.0:30090:30090", // Webhook service
+		"0.0.0.0:6443:6443",   // K3s API Port
+		"0.0.0.0:30080:30080", // HTTP Port
+		"0.0.0.0:30000:30000", // Registry Port
+		"0.0.0.0:30001:30001", // Postgres Port
+		"0.0.0.0:30002:30002", // Minio API Port (use HTTP port for minio console)
 	})
 }
 
 // PullDockerImage will Pull docker image
 func PullDockerImage(ctx context.Context, cli Docker, image string, pullPolicy ImagePullPolicy,
 	imagePullOptions ImagePullOptions, dryRun bool) error {
+
 	if dryRun {
 		PrintPullImage(image, imagePullOptions)
 		return nil
@@ -222,6 +223,7 @@ func PrintCreateContainer(volumes []mount.Mount, portBindings map[nat.Port][]nat
 // StartContainer will create and start docker container
 func StartContainer(ctx context.Context, cli Docker, volumes []mount.Mount, exposedPorts map[nat.Port]struct{},
 	portBindings map[nat.Port][]nat.PortBinding, name, image string, additionalEnvVars []string, dryRun bool) (string, error) {
+
 	// Append the additional env variables to the default list of env
 	Environment = append(Environment, additionalEnvVars...)
 	if dryRun {
@@ -252,6 +254,45 @@ func StartContainer(ctx context.Context, cli Docker, volumes []mount.Mount, expo
 	return resp.ID, nil
 }
 
+// CopyContainerFile try to create the container, see if the source file is there, copy it to the destination
+func CopyContainerFile(ctx context.Context, cli Docker, source, destination, name, image string) error {
+	resp, err := cli.ContainerCreate(ctx, &container.Config{Image: image}, &container.HostConfig{}, nil, nil, name)
+	if err != nil {
+		return err
+	}
+	var removeErr error
+	defer func() {
+		removeErr = cli.ContainerRemove(context.Background(), resp.ID, types.ContainerRemoveOptions{
+			Force: true,
+		})
+	}()
+	_, err = cli.ContainerStatPath(ctx, resp.ID, source)
+	if err != nil {
+		return err
+	}
+	reader, _, err := cli.CopyFromContainer(ctx, resp.ID, source)
+	if err != nil {
+		return err
+	}
+	tarFile := destination + ".tar"
+	outFile, err := os.Create(tarFile)
+	if err != nil {
+		return err
+	}
+	defer outFile.Close()
+	defer reader.Close()
+	_, err = io.Copy(outFile, reader)
+	if err != nil {
+		return err
+	}
+	r, _ := os.Open(tarFile)
+	err = f.ExtractTar(r, destination)
+	if err != nil {
+		return err
+	}
+	return removeErr
+}
+
 // ReadLogs will return io scanner for reading the logs of a container
 func ReadLogs(ctx context.Context, cli Docker, id string) (*bufio.Scanner, error) {
 	reader, err := cli.ContainerLogs(ctx, id, types.ContainerLogsOptions{