From 8b8444fb1f1a2dcb805399471a6267c898858dfb Mon Sep 17 00:00:00 2001 From: Jose Diaz-Gonzalez Date: Thu, 18 Jan 2024 05:28:07 -0500 Subject: [PATCH] feat: implement run and run:detached This utilizes a Kubernetes Job to start a pod in order to take advantage of completed Job cleanup in Kubernetes. It also properly handles interactivity by ensuring the container has a TTY. Note that there is currently no provision for parsing docker arguments and translating them into their Kubernetes equivalents. --- docs/deployment/schedulers/k3s.md | 8 +- plugins/scheduler-k3s/Makefile | 2 +- plugins/scheduler-k3s/functions.go | 19 +- plugins/scheduler-k3s/go.mod | 13 +- plugins/scheduler-k3s/go.sum | 30 +- plugins/scheduler-k3s/helm.go | 93 +++- plugins/scheduler-k3s/scheduler_k3s.go | 2 + .../scheduler-k3s/src/triggers/triggers.go | 21 +- plugins/scheduler-k3s/template.go | 266 ++++++++-- plugins/scheduler-k3s/triggers.go | 494 +++++++++++++++++- 10 files changed, 833 insertions(+), 115 deletions(-) diff --git a/docs/deployment/schedulers/k3s.md b/docs/deployment/schedulers/k3s.md index 21ead7c48..ced87c825 100644 --- a/docs/deployment/schedulers/k3s.md +++ b/docs/deployment/schedulers/k3s.md @@ -162,6 +162,9 @@ The following sections describe implemented and unimplemented scheduler function This plugin implements various functionality through `plugn` triggers to integrate with Docker for running apps on a single server. The following functionality is supported by the `scheduler-docker-local` plugin. - `apps:destroy` +- `run` + - The `scheduler-post-run` trigger is not always triggered +- `run:detached` - `enter` - `deploy` - `logs` @@ -171,15 +174,14 @@ This plugin implements various functionality through `plugn` triggers to integra - `apps:clone` - `apps:rename` +- `run:list` +- `run:logs` - `ps:inspect` -- `run` -- `run:detached` The following Dokku functionality is not implemented at this time. - `cron` - `vector` log integration -- one-off tasks - persistent storage - healthchecks diff --git a/plugins/scheduler-k3s/Makefile b/plugins/scheduler-k3s/Makefile index fe33daf90..3ea295207 100644 --- a/plugins/scheduler-k3s/Makefile +++ b/plugins/scheduler-k3s/Makefile @@ -1,5 +1,5 @@ SUBCOMMANDS = subcommands/initialize subcommands/report subcommands/set subcommands/show-kubeconfig -TRIGGERS = triggers/install triggers/post-app-clone-setup triggers/post-app-rename-setup triggers/post-delete triggers/post-registry-login triggers/report triggers/scheduler-deploy triggers/scheduler-enter triggers/scheduler-logs triggers/scheduler-stop +TRIGGERS = triggers/install triggers/post-app-clone-setup triggers/post-app-rename-setup triggers/post-delete triggers/post-registry-login triggers/report triggers/scheduler-deploy triggers/scheduler-enter triggers/scheduler-logs triggers/scheduler-run triggers/scheduler-stop BUILD = commands subcommands triggers PLUGIN_NAME = scheduler-k3s diff --git a/plugins/scheduler-k3s/functions.go b/plugins/scheduler-k3s/functions.go index 314a6d6d2..d4fb376ac 100644 --- a/plugins/scheduler-k3s/functions.go +++ b/plugins/scheduler-k3s/functions.go @@ -7,6 +7,7 @@ import ( "github.com/dokku/dokku/plugins/common" "github.com/dokku/dokku/plugins/config" + batchv1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" @@ -81,6 +82,20 @@ func KubernetesClientConfig() clientcmd.ClientConfig { &clientcmd.ConfigOverrides{ClusterInfo: clientcmdapi.Cluster{Server: ""}}) } +func createKubernetesJob(ctx context.Context, job batchv1.Job) (batchv1.Job, error) { + clientset, err := NewKubernetesClient() + if err != nil { + return batchv1.Job{}, err + } + + createdJob, err := clientset.Client.BatchV1().Jobs(job.Namespace).Create(ctx, &job, metav1.CreateOptions{}) + if err != nil { + return batchv1.Job{}, err + } + + return *createdJob, nil +} + func createKubernetesNamespace(ctx context.Context, namespaceName string) error { clientset, err := NewKubernetesClient() if err != nil { @@ -98,7 +113,7 @@ func createKubernetesNamespace(ctx context.Context, namespaceName string) error } } - namespace := &corev1.Namespace{ + namespace := corev1.Namespace{ ObjectMeta: metav1.ObjectMeta{ Name: namespaceName, Annotations: map[string]string{ @@ -109,7 +124,7 @@ func createKubernetesNamespace(ctx context.Context, namespaceName string) error }, }, } - _, err = clientset.Client.CoreV1().Namespaces().Create(ctx, namespace, metav1.CreateOptions{}) + _, err = clientset.Client.CoreV1().Namespaces().Create(ctx, &namespace, metav1.CreateOptions{}) if err != nil { return err } diff --git a/plugins/scheduler-k3s/go.mod b/plugins/scheduler-k3s/go.mod index 1631b78cd..e5912efa4 100644 --- a/plugins/scheduler-k3s/go.mod +++ b/plugins/scheduler-k3s/go.mod @@ -7,6 +7,7 @@ require ( github.com/dokku/dokku/plugins/config v0.0.0-20240112043630-928c388f10fa github.com/fatih/color v1.16.0 github.com/go-resty/resty/v2 v2.11.0 + github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 github.com/rancher/wharfie v0.6.5 github.com/spf13/pflag v1.0.5 github.com/traefik/traefik/v2 v2.10.7 @@ -18,6 +19,7 @@ require ( k8s.io/cli-runtime v0.29.0 k8s.io/client-go v0.29.0 k8s.io/kubectl v0.28.4 + k8s.io/kubernetes v1.29.1 k8s.io/utils v0.0.0-20240102154912-e7106e64919e mvdan.cc/sh/v3 v3.7.0 ) @@ -133,12 +135,13 @@ require ( github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect github.com/xeipuuv/gojsonschema v1.2.0 // indirect github.com/xlab/treeprint v1.2.0 // indirect - go.opentelemetry.io/otel v1.14.0 // indirect - go.opentelemetry.io/otel/trace v1.14.0 // indirect + go.opentelemetry.io/otel v1.19.0 // indirect + go.opentelemetry.io/otel/metric v1.19.0 // indirect + go.opentelemetry.io/otel/trace v1.19.0 // indirect go.starlark.net v0.0.0-20230525235612-a134d8f9ddca // indirect go.uber.org/multierr v1.11.0 // indirect golang.org/x/crypto v0.16.0 // indirect - golang.org/x/mod v0.12.0 // indirect + golang.org/x/mod v0.14.0 // indirect golang.org/x/net v0.19.0 // indirect golang.org/x/oauth2 v0.10.0 // indirect golang.org/x/sync v0.6.0 // indirect @@ -146,9 +149,9 @@ require ( golang.org/x/term v0.15.0 // indirect golang.org/x/text v0.14.0 // indirect golang.org/x/time v0.3.0 // indirect - golang.org/x/tools v0.12.1-0.20230815132531-74c255bcf846 // indirect + golang.org/x/tools v0.16.1 // indirect google.golang.org/appengine v1.6.7 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20230711160842-782d3b101e98 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d // indirect google.golang.org/grpc v1.58.3 // indirect google.golang.org/protobuf v1.31.0 // indirect gopkg.in/inf.v0 v0.9.1 // indirect diff --git a/plugins/scheduler-k3s/go.sum b/plugins/scheduler-k3s/go.sum index fa4442fb9..79ff56cb4 100644 --- a/plugins/scheduler-k3s/go.sum +++ b/plugins/scheduler-k3s/go.sum @@ -241,8 +241,10 @@ github.com/json-iterator/go v1.1.7/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/u github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w= -github.com/karrick/godirwalk v1.16.1 h1:DynhcF+bztK8gooS0+NDJFrdNZjJ3gzVzC545UNA9iw= -github.com/karrick/godirwalk v1.16.1/go.mod h1:j4mkqPuvaLI8mp1DroR3P6ad7cyYd4c1qeJ3RV7ULlk= +github.com/karrick/godirwalk v1.17.0 h1:b4kY7nqDdioR/6qnbHQyDvmA17u5G1cZ6J+CZXwSWoI= +github.com/karrick/godirwalk v1.17.0/go.mod h1:j4mkqPuvaLI8mp1DroR3P6ad7cyYd4c1qeJ3RV7ULlk= +github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 h1:Z9n2FFNUXsshfwJMBgNA0RU6/i7WVaAegv3PtuIHPMs= +github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51/go.mod h1:CzGEWj7cYgsdH8dAjBGEr58BoE7ScuLd+fwFZ44+/x8= github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= github.com/klauspost/compress v1.17.1 h1:NE3C767s2ak2bweCZo3+rdP4U/HoyVXLv/X9f2gPS5g= @@ -438,10 +440,12 @@ github.com/yvasiyarov/newrelic_platform_go v0.0.0-20140908184405-b21fdbd4370f h1 github.com/yvasiyarov/newrelic_platform_go v0.0.0-20140908184405-b21fdbd4370f/go.mod h1:GlGEuHIJweS1mbCqG+7vt2nvWLzLLnRHbXz5JKd/Qbg= go.opencensus.io v0.24.0 h1:y73uSU6J157QMP2kn2r30vwW1A2W2WFwSCGnAVxeaD0= go.opencensus.io v0.24.0/go.mod h1:vNK8G9p7aAivkbmorf4v+7Hgx+Zs0yY+0fOtgBfjQKo= -go.opentelemetry.io/otel v1.14.0 h1:/79Huy8wbf5DnIPhemGB+zEPVwnN6fuQybr/SRXa6hM= -go.opentelemetry.io/otel v1.14.0/go.mod h1:o4buv+dJzx8rohcUeRmWUZhqupFvzWis188WlggnNeU= -go.opentelemetry.io/otel/trace v1.14.0 h1:wp2Mmvj41tDsyAJXiWDWpfNsOiIyd38fy85pyKcFq/M= -go.opentelemetry.io/otel/trace v1.14.0/go.mod h1:8avnQLK+CG77yNLUae4ea2JDQ6iT+gozhnZjy/rw9G8= +go.opentelemetry.io/otel v1.19.0 h1:MuS/TNf4/j4IXsZuJegVzI1cwut7Qc00344rgH7p8bs= +go.opentelemetry.io/otel v1.19.0/go.mod h1:i0QyjOq3UPoTzff0PJB2N66fb4S0+rSbSB15/oyH9fY= +go.opentelemetry.io/otel/metric v1.19.0 h1:aTzpGtV0ar9wlV4Sna9sdJyII5jTVJEvKETPiOKwvpE= +go.opentelemetry.io/otel/metric v1.19.0/go.mod h1:L5rUsV9kM1IxCj1MmSdS+JQAcVm319EUrDVLrt7jqt8= +go.opentelemetry.io/otel/trace v1.19.0 h1:DFVQmlVbfVeOuBRrwdtaehRrWiL1JoVs9CPIQ1Dzxpg= +go.opentelemetry.io/otel/trace v1.19.0/go.mod h1:mfaSyvGyEJEI0nyV2I4qhNQnbBOUUmYZpYojqMnX2vo= go.starlark.net v0.0.0-20230525235612-a134d8f9ddca h1:VdD38733bfYv5tUZwEIskMM93VanwNIi5bIKnDrJdEY= go.starlark.net v0.0.0-20230525235612-a134d8f9ddca/go.mod h1:jxU+3+j+71eXOW14274+SmmuW82qJzl6iZSeqEtTGds= go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= @@ -464,8 +468,8 @@ golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= -golang.org/x/mod v0.12.0 h1:rmsUpXtvNzj340zd98LZ4KntptpfRHwpFOHG188oHXc= -golang.org/x/mod v0.12.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= +golang.org/x/mod v0.14.0 h1:dGoOF9QVLYng8IHTm7BAyWqCqSheQ5pYWGhzW00YJr0= +golang.org/x/mod v0.14.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c= golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= @@ -550,8 +554,8 @@ golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roY golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU= -golang.org/x/tools v0.12.1-0.20230815132531-74c255bcf846 h1:Vve/L0v7CXXuxUmaMGIEK/dEeq7uiqb5qBgQrZzIE7E= -golang.org/x/tools v0.12.1-0.20230815132531-74c255bcf846/go.mod h1:Sc0INKfu04TlqNoRA1hgpFZbhYXHPr4V5DzpSBTPqQM= +golang.org/x/tools v0.16.1 h1:TLyB3WofjdOEepBHAU20JdNC1Zbg87elYofWYAY5oZA= +golang.org/x/tools v0.16.1/go.mod h1:kYVVN6I1mBNoB1OX+noeBjbRk4IUEPa7JJ+TJMEooJ0= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= @@ -563,8 +567,8 @@ google.golang.org/appengine v1.6.7/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCID google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc= google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98Agz4BDEuKkezgsaosCRResVns1a3J2ZsMNc= google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo= -google.golang.org/genproto/googleapis/rpc v0.0.0-20230711160842-782d3b101e98 h1:bVf09lpb+OJbByTj913DRJioFFAjf/ZGxEz7MajTp2U= -google.golang.org/genproto/googleapis/rpc v0.0.0-20230711160842-782d3b101e98/go.mod h1:TUfxEVdsvPg18p6AslUXFoLdpED4oBnGwyqk3dV1XzM= +google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d h1:uvYuEyMHKNt+lT4K3bN6fGswmK8qSvcreM3BwjDh+y4= +google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d/go.mod h1:+Bk1OCOj40wS2hwAMA+aCW9ypzm63QTBBHp6lQ3p+9M= google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c= google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg= google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk= @@ -624,6 +628,8 @@ k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00 h1:aVUu9fTY98ivBPKR9Y5w/A k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00/go.mod h1:AsvuZPBlUDVuCdzJ87iajxtXuR9oktsTctW/R9wwouA= k8s.io/kubectl v0.28.4 h1:gWpUXW/T7aFne+rchYeHkyB8eVDl5UZce8G4X//kjUQ= k8s.io/kubectl v0.28.4/go.mod h1:CKOccVx3l+3MmDbkXtIUtibq93nN2hkDR99XDCn7c/c= +k8s.io/kubernetes v1.29.1 h1:fxJFVb8uqbYZDYHpwIsAndBQs360cQGb0xa1gYFh3fo= +k8s.io/kubernetes v1.29.1/go.mod h1:xZPKU0yO0CBbLTnbd+XGyRmmtmaVuJykDb8gNCkeeUE= k8s.io/utils v0.0.0-20240102154912-e7106e64919e h1:eQ/4ljkx21sObifjzXwlPKpdGLrCfRziVtos3ofG/sQ= k8s.io/utils v0.0.0-20240102154912-e7106e64919e/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0= mvdan.cc/sh/v3 v3.7.0 h1:lSTjdP/1xsddtaKfGg7Myu7DnlHItd3/M2tomOcNNBg= diff --git a/plugins/scheduler-k3s/helm.go b/plugins/scheduler-k3s/helm.go index e5f462d6c..b846bf154 100644 --- a/plugins/scheduler-k3s/helm.go +++ b/plugins/scheduler-k3s/helm.go @@ -6,6 +6,7 @@ import ( "fmt" "os" "os/signal" + "sort" "strings" "syscall" "time" @@ -16,6 +17,7 @@ import ( "helm.sh/helm/v3/pkg/chart/loader" "helm.sh/helm/v3/pkg/kube" "helm.sh/helm/v3/pkg/storage/driver" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) var DevNullPrinter = func(format string, v ...interface{}) {} @@ -38,6 +40,21 @@ var DeployLogPrinter = func(format string, v ...interface{}) { } } +type ChartInput struct { + ChartPath string + Namespace string + ReleaseName string + RollbackOnFailure bool + Timeout time.Duration + Values map[string]interface{} +} + +type Release struct { + Name string + Namespace string + Version int +} + type HelmAgent struct { Configuration *action.Configuration Namespace string @@ -45,18 +62,6 @@ type HelmAgent struct { } func NewHelmAgent(namespace string, logger action.DebugLog) (*HelmAgent, error) { - actionConfig, err := getActionConfig(namespace, logger) - if err != nil { - return nil, err - } - - return &HelmAgent{ - Configuration: actionConfig, - Namespace: namespace, - }, nil -} - -func getActionConfig(namespace string, logger action.DebugLog) (*action.Configuration, error) { actionConfig := new(action.Configuration) helmDriver := os.Getenv("HELM_DRIVER") @@ -68,7 +73,12 @@ func getActionConfig(namespace string, logger action.DebugLog) (*action.Configur if err := actionConfig.Init(kubeConfig, namespace, helmDriver, logger); err != nil { return nil, err } - return actionConfig, nil + + return &HelmAgent{ + Configuration: actionConfig, + Namespace: namespace, + Logger: logger, + }, nil } func (h *HelmAgent) ChartExists(releaseName string) (bool, error) { @@ -89,19 +99,35 @@ func (h *HelmAgent) ChartExists(releaseName string) (bool, error) { return false, nil } -type ChartInput struct { - ChartPath string - Namespace string - ReleaseName string - RollbackOnFailure bool - Timeout time.Duration - Values map[string]interface{} +func (h *HelmAgent) DeleteRevision(releaseName string, revision int) error { + clientset, err := NewKubernetesClient() + if err != nil { + return fmt.Errorf("Error creating kubernetes client: %w", err) + } + + secretName := fmt.Sprintf("sh.helm.release.v1.%s.v%d", releaseName, revision) + err = clientset.Client.CoreV1().Secrets(h.Namespace).Delete(context.Background(), secretName, metav1.DeleteOptions{}) + if err != nil { + return fmt.Errorf("Error deleting secret: %w", err) + } + return nil +} + +func (h *HelmAgent) GetValues(releaseName string) (map[string]interface{}, error) { + client := action.NewGetValues(h.Configuration) + client.AllValues = true + values, err := client.Run(releaseName) + if err != nil { + return nil, fmt.Errorf("Error getting values: %w", err) + } + + return values, nil } func (h *HelmAgent) InstallOrUpgradeChart(input ChartInput) error { chartExists, err := h.ChartExists(input.ReleaseName) if err != nil { - return err + return fmt.Errorf("Error checking if chart exists: %w", err) } if chartExists { @@ -167,6 +193,29 @@ func (h *HelmAgent) InstallChart(input ChartInput) error { return nil } +func (h *HelmAgent) ListRevisions(releaseName string) ([]Release, error) { + client := action.NewHistory(h.Configuration) + response, err := client.Run(releaseName) + if err != nil { + return nil, fmt.Errorf("Error getting revisions: %w", err) + } + + releases := []Release{} + for _, release := range response { + releases = append(releases, Release{ + Name: release.Name, + Namespace: release.Namespace, + Version: release.Version, + }) + } + + sort.Slice(releases, func(i, j int) bool { + return releases[i].Version < releases[j].Version + }) + + return releases, nil +} + func (h *HelmAgent) UpgradeChart(input ChartInput) error { namespace := input.Namespace if namespace == "" { @@ -186,7 +235,9 @@ func (h *HelmAgent) UpgradeChart(input ChartInput) error { client := action.NewUpgrade(h.Configuration) client.Atomic = input.RollbackOnFailure client.ChartPathOptions = action.ChartPathOptions{} + client.CleanupOnFail = true client.DryRun = false + client.MaxHistory = 10 client.Namespace = namespace client.Timeout = input.Timeout client.Wait = false diff --git a/plugins/scheduler-k3s/scheduler_k3s.go b/plugins/scheduler-k3s/scheduler_k3s.go index 6837b91fe..02462ad14 100644 --- a/plugins/scheduler-k3s/scheduler_k3s.go +++ b/plugins/scheduler-k3s/scheduler_k3s.go @@ -5,6 +5,7 @@ import ( traefikv1alpha1 "github.com/traefik/traefik/v2/pkg/provider/kubernetes/crd/traefikio/v1alpha1" appsv1 "k8s.io/api/apps/v1" + batchv1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/serializer" @@ -43,6 +44,7 @@ var k8sNativeSchemeOnce sync.Once func init() { k8sNativeSchemeOnce.Do(func() { _ = appsv1.AddToScheme(runtimeScheme) + _ = batchv1.AddToScheme(runtimeScheme) _ = corev1.AddToScheme(runtimeScheme) _ = traefikv1alpha1.AddToScheme(runtimeScheme) }) diff --git a/plugins/scheduler-k3s/src/triggers/triggers.go b/plugins/scheduler-k3s/src/triggers/triggers.go index 7b8e74f0f..392dbb0e6 100644 --- a/plugins/scheduler-k3s/src/triggers/triggers.go +++ b/plugins/scheduler-k3s/src/triggers/triggers.go @@ -53,7 +53,10 @@ func main() { appName := flag.Arg(1) containerType := flag.Arg(2) args := flag.Args() - if len(args) > 3 { + if len(args) == 2 { + _, args = common.ShiftString(args) + _, args = common.ShiftString(args) + } else if len(args) >= 3 { _, args = common.ShiftString(args) _, args = common.ShiftString(args) _, args = common.ShiftString(args) @@ -86,6 +89,22 @@ func main() { scheduler := flag.Arg(0) appName := flag.Arg(1) err = scheduler_k3s.TriggerSchedulerStop(scheduler, appName) + case "scheduler-run": + var envCount int + scheduler := flag.Arg(0) + appName := flag.Arg(1) + envCount, err = strconv.Atoi(flag.Arg(2)) + if err != nil { + envCount = 0 + } + args := flag.Args() + if len(args) >= 3 { + _, args = common.ShiftString(args) + _, args = common.ShiftString(args) + _, args = common.ShiftString(args) + } + + err = scheduler_k3s.TriggerSchedulerRun(scheduler, appName, envCount, args) default: err = fmt.Errorf("Invalid plugin trigger call: %s", trigger) } diff --git a/plugins/scheduler-k3s/template.go b/plugins/scheduler-k3s/template.go index b1789a972..53c582ac4 100644 --- a/plugins/scheduler-k3s/template.go +++ b/plugins/scheduler-k3s/template.go @@ -11,6 +11,7 @@ import ( orderedmap "github.com/wk8/go-ordered-map/v2" "gopkg.in/yaml.v3" appsv1 "k8s.io/api/apps/v1" + batchv1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" v1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" @@ -48,6 +49,7 @@ type Deployment struct { PortMaps []PortMap ProcessType string Replicas int32 + WorkingDir string } type IngressRouteEntrypoint string @@ -67,6 +69,23 @@ type IngressRoute struct { ServiceName string } +type Job struct { + AppName string + Command []string + DeploymentID int64 + Entrypoint string + Env map[string]string + Image string + ImagePullSecrets string + ImageSourceType string + Interactive bool + Labels map[string]string + Namespace string + ProcessType string + RemoveContainer bool + WorkingDir string +} + type Secret struct { AppName string Env map[string]string @@ -79,11 +98,10 @@ type Service struct { PortMaps []PortMap } -type PrintInput struct { +type WriteResourceInput struct { AppendContents string Object runtime.Object Path string - Name string Replacements *orderedmap.OrderedMap[string, string] } @@ -133,10 +151,9 @@ func createIngressRoutesFiles(input CreateIngressRoutesInput) error { ServiceName: input.Service.Name, }) - err = printResource(PrintInput{ + err = writeResourceToFile(WriteResourceInput{ Object: &ingressRoute, Path: filepath.Join(input.ChartDir, fmt.Sprintf("templates/ingress-route-%s-%d-%d.yaml", portMap.Scheme, portMap.HostPort, portMap.ContainerPort)), - Name: ingressRoute.Name, }) if err != nil { return fmt.Errorf("Error printing ingress route: %w", err) @@ -146,56 +163,6 @@ func createIngressRoutesFiles(input CreateIngressRoutesInput) error { return nil } -func printResource(input PrintInput) error { - common.LogDebug(fmt.Sprintf("Printing resource: %s", input.Path)) - printr := printers.NewTypeSetter(runtimeScheme).ToPrinter(&printers.YAMLPrinter{}) - handle, err := os.Create(input.Path) - if err != nil { - return fmt.Errorf("Error creating template file: %w", err) - } - defer handle.Close() - - if err := printr.PrintObj(input.Object, handle); err != nil { - return fmt.Errorf("Error writing template file: %w", err) - } - - if input.Replacements != nil { - b, err := os.ReadFile(input.Path) - if err != nil { - return fmt.Errorf("Error reading template file: %w", err) - } - - contents := string(b) - for pair := input.Replacements.Oldest(); pair != nil; pair = pair.Next() { - contents = strings.ReplaceAll(string(contents), pair.Key, pair.Value) - } - - err = os.WriteFile(input.Path, []byte(contents), os.FileMode(0644)) - if err != nil { - return fmt.Errorf("Error updating template file with replacements: %w", err) - } - } - - if input.AppendContents != "" { - b, err := os.ReadFile(input.Path) - if err != nil { - return fmt.Errorf("Error reading template file: %w", err) - } - - contents := string(b) + "\n" + input.AppendContents - - err = os.WriteFile(input.Path, []byte(contents), os.FileMode(0644)) - if err != nil { - return fmt.Errorf("Error updating template file with replacements: %w", err) - } - } - - if os.Getenv("DOKKU_TRACE") == "1" { - common.CatFile(input.Path) - } - return nil -} - func templateKubernetesDeployment(input Deployment) (appsv1.Deployment, error) { labels := map[string]string{ "dokku.com/app-name": input.AppName, @@ -245,11 +212,12 @@ func templateKubernetesDeployment(input Deployment) (appsv1.Deployment, error) { }, }, Image: input.Image, - ImagePullPolicy: corev1.PullPolicy("Always"), + ImagePullPolicy: corev1.PullAlways, Resources: corev1.ResourceRequirements{ Limits: corev1.ResourceList{}, Requests: corev1.ResourceList{}, }, + WorkingDir: input.WorkingDir, }, }, }, @@ -382,6 +350,144 @@ func templateKubernetesIngressRoute(input IngressRoute) traefikv1alpha1.IngressR return ingressRoute } +func templateKubernetesJob(input Job) (batchv1.Job, error) { + labels := map[string]string{ + "dokku.com/app-name": input.AppName, + "dokku.com/app-process-type": fmt.Sprintf("%s-%s", input.AppName, input.ProcessType), + "dokku.com/process-type": input.ProcessType, + } + annotations := map[string]string{ + "dokku.com/app-name": input.AppName, + "dokku.com/builder-type": input.ImageSourceType, + "dokku.com/deployment-id": fmt.Sprint(input.DeploymentID), + "dokku.com/managed": "true", + "dokku.com/process-type": input.ProcessType, + } + + // todo: implement a controller to delete pods on completion + + for key, value := range input.Labels { + labels[key] = value + } + secretName := fmt.Sprintf("env-%s.%d", input.AppName, input.DeploymentID) + + env := []corev1.EnvVar{} + for key, value := range input.Env { + env = append(env, corev1.EnvVar{ + Name: key, + Value: value, + }) + } + + job := batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{ + GenerateName: fmt.Sprintf("%s-%s-", input.AppName, input.ProcessType), + Namespace: input.Namespace, + Labels: labels, + Annotations: annotations, + }, + Spec: batchv1.JobSpec{ + BackoffLimit: ptr.To(int32(0)), + Template: corev1.PodTemplateSpec{ + Spec: corev1.PodSpec{ + Containers: []corev1.Container{ + { + Args: input.Command, + Name: fmt.Sprintf("%s-%s", input.AppName, input.ProcessType), + Env: env, + // todo: pull secret ref here + EnvFrom: []corev1.EnvFromSource{ + { + SecretRef: &corev1.SecretEnvSource{ + LocalObjectReference: corev1.LocalObjectReference{ + Name: secretName, + }, + Optional: ptr.To(true), + }, + }, + }, + Image: input.Image, + ImagePullPolicy: corev1.PullAlways, + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{}, + Requests: corev1.ResourceList{}, + }, + WorkingDir: input.WorkingDir, + }, + }, + RestartPolicy: corev1.RestartPolicyNever, + }, + }, + }, + } + + if input.Entrypoint != "" { + job.Spec.Template.Spec.Containers[0].Command = []string{input.Entrypoint} + } + + if input.Interactive { + job.Spec.Template.Spec.Containers[0].Stdin = true + job.Spec.Template.Spec.Containers[0].StdinOnce = true + job.Spec.Template.Spec.Containers[0].TTY = true + } + + if input.RemoveContainer { + job.Spec.TTLSecondsAfterFinished = ptr.To(int32(60)) + } + + if input.ImagePullSecrets != "" { + job.Spec.Template.Spec.ImagePullSecrets = []corev1.LocalObjectReference{ + { + Name: input.ImagePullSecrets, + }, + } + } + + cpuLimit, err := common.PlugnTriggerOutputAsString("resource-get-property", []string{input.AppName, input.ProcessType, "limit", "cpu"}...) + if err != nil && cpuLimit != "" && cpuLimit != "0" { + cpuQuantity, err := resource.ParseQuantity(cpuLimit) + if err != nil { + return job, fmt.Errorf("Error parsing cpu limit: %w", err) + } + job.Spec.Template.Spec.Containers[0].Resources.Limits["cpu"] = cpuQuantity + } + nvidiaGpuLimit, err := common.PlugnTriggerOutputAsString("resource-get-property", []string{input.AppName, input.ProcessType, "limit", "nvidia-gpu"}...) + if err != nil && nvidiaGpuLimit != "" && nvidiaGpuLimit != "0" { + nvidiaGpuQuantity, err := resource.ParseQuantity(nvidiaGpuLimit) + if err != nil { + return job, fmt.Errorf("Error parsing nvidia-gpu limit: %w", err) + } + job.Spec.Template.Spec.Containers[0].Resources.Limits["nvidia.com/gpu"] = nvidiaGpuQuantity + } + memoryLimit, err := common.PlugnTriggerOutputAsString("resource-get-property", []string{input.AppName, input.ProcessType, "limit", "memory"}...) + if err != nil && memoryLimit != "" && memoryLimit != "0" { + memoryQuantity, err := resource.ParseQuantity(memoryLimit) + if err != nil { + return job, fmt.Errorf("Error parsing memory limit: %w", err) + } + job.Spec.Template.Spec.Containers[0].Resources.Limits["memory"] = memoryQuantity + } + + cpuRequest, err := common.PlugnTriggerOutputAsString("resource-get-property", []string{input.AppName, input.ProcessType, "reserve", "cpu"}...) + if err != nil && cpuRequest != "" && cpuRequest != "0" { + cpuQuantity, err := resource.ParseQuantity(cpuRequest) + if err != nil { + return job, fmt.Errorf("Error parsing cpu request: %w", err) + } + job.Spec.Template.Spec.Containers[0].Resources.Requests["cpu"] = cpuQuantity + } + memoryRequest, err := common.PlugnTriggerOutputAsString("resource-get-property", []string{input.AppName, input.ProcessType, "reserve", "memory"}...) + if err != nil && memoryRequest != "" && memoryRequest != "0" { + memoryQuantity, err := resource.ParseQuantity(memoryRequest) + if err != nil { + return job, fmt.Errorf("Error parsing memory request: %w", err) + } + job.Spec.Template.Spec.Containers[0].Resources.Requests["memory"] = memoryQuantity + } + + return job, nil +} + func templateKubernetesSecret(input Secret) corev1.Secret { secretName := fmt.Sprintf("env-%s.DEPLOYMENT_ID", input.AppName) secret := corev1.Secret{ @@ -441,6 +547,56 @@ func templateKubernetesService(input Service) corev1.Service { return service } +func writeResourceToFile(input WriteResourceInput) error { + common.LogDebug(fmt.Sprintf("Printing resource: %s", input.Path)) + printr := printers.NewTypeSetter(runtimeScheme).ToPrinter(&printers.YAMLPrinter{}) + handle, err := os.Create(input.Path) + if err != nil { + return fmt.Errorf("Error creating template file: %w", err) + } + defer handle.Close() + + if err := printr.PrintObj(input.Object, handle); err != nil { + return fmt.Errorf("Error writing template file: %w", err) + } + + if input.Replacements != nil { + b, err := os.ReadFile(input.Path) + if err != nil { + return fmt.Errorf("Error reading template file: %w", err) + } + + contents := string(b) + for pair := input.Replacements.Oldest(); pair != nil; pair = pair.Next() { + contents = strings.ReplaceAll(string(contents), pair.Key, pair.Value) + } + + err = os.WriteFile(input.Path, []byte(contents), os.FileMode(0644)) + if err != nil { + return fmt.Errorf("Error updating template file with replacements: %w", err) + } + } + + if input.AppendContents != "" { + b, err := os.ReadFile(input.Path) + if err != nil { + return fmt.Errorf("Error reading template file: %w", err) + } + + contents := string(b) + "\n" + input.AppendContents + + err = os.WriteFile(input.Path, []byte(contents), os.FileMode(0644)) + if err != nil { + return fmt.Errorf("Error updating template file with replacements: %w", err) + } + } + + if os.Getenv("DOKKU_TRACE") == "1" { + common.CatFile(input.Path) + } + return nil +} + func writeYaml(input WriteYamlInput) error { common.LogDebug(fmt.Sprintf("Printing resource: %s", input.Path)) data, err := yaml.Marshal(input.Object) diff --git a/plugins/scheduler-k3s/triggers.go b/plugins/scheduler-k3s/triggers.go index cc26c832e..cfff5cb92 100644 --- a/plugins/scheduler-k3s/triggers.go +++ b/plugins/scheduler-k3s/triggers.go @@ -4,6 +4,7 @@ import ( "bufio" "context" "encoding/base64" + "errors" "fmt" "io" "os" @@ -18,6 +19,7 @@ import ( "github.com/dokku/dokku/plugins/common" "github.com/dokku/dokku/plugins/config" "github.com/fatih/color" + "github.com/kballard/go-shellquote" "github.com/rancher/wharfie/pkg/registries" orderedmap "github.com/wk8/go-ordered-map/v2" "gopkg.in/yaml.v3" @@ -26,9 +28,11 @@ import ( corev1 "k8s.io/api/core/v1" v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/wait" corev1client "k8s.io/client-go/kubernetes/typed/core/v1" "k8s.io/client-go/tools/remotecommand" "k8s.io/kubectl/pkg/util/term" + "k8s.io/kubernetes/pkg/client/conditions" "k8s.io/utils/ptr" ) @@ -205,10 +209,9 @@ func TriggerSchedulerDeploy(scheduler string, appName string, imageTag string) e Env: env.Map(), Namespace: namespace, }) - err = printResource(PrintInput{ + err = writeResourceToFile(WriteResourceInput{ Object: &secret, Path: filepath.Join(chartDir, "templates/secret.yaml"), - Name: secret.Name, Replacements: replacements, AppendContents: `{{- with .Values.secrets }} data: @@ -265,6 +268,7 @@ data: PortMaps: portMaps, ProcessType: processType, Replicas: replicaCountPlaceholder, + WorkingDir: common.GetWorkingDir(appName, image), }) if err != nil { return fmt.Errorf("Error templating deployment: %w", err) @@ -272,10 +276,9 @@ data: replacements.Set(fmt.Sprintf("replicas: %d", replicaCountPlaceholder), fmt.Sprintf("replicas: {{.Values.processes.%s.replicas}}", processType)) deployments[processType] = deployment - err = printResource(PrintInput{ + err = writeResourceToFile(WriteResourceInput{ Object: &deployment, Path: filepath.Join(chartDir, fmt.Sprintf("templates/deployment-%s.yaml", deployment.Name)), - Name: deployment.Name, Replacements: replacements, }) replacements.Delete(fmt.Sprintf("replicas: %d", replicaCountPlaceholder)) @@ -292,10 +295,9 @@ data: PortMaps: portMaps, }) - err := printResource(PrintInput{ + err := writeResourceToFile(WriteResourceInput{ Object: &service, Path: filepath.Join(chartDir, "templates/service-web.yaml"), - Name: service.Name, Replacements: replacements, }) if err != nil { @@ -473,27 +475,52 @@ func TriggerSchedulerEnter(scheduler string, appName string, processType string, } } - coreclient, err := corev1client.NewForConfig(&clientset.RestConfig) + entrypoint := "" + if selectedPod.Annotations["dokku.com/builder-type"] == "herokuish" { + entrypoint = "/exec" + } + + return enterPod(ctx, EnterPodInput{ + AppName: appName, + Clientset: clientset, + Command: command, + Entrypoint: entrypoint, + ProcessType: processType, + SelectedPod: selectedPod, + }) +} + +type EnterPodInput struct { + AppName string + Clientset KubernetesClient + Command []string + Entrypoint string + ProcessType string + SelectedPod v1.Pod +} + +func enterPod(ctx context.Context, input EnterPodInput) error { + coreclient, err := corev1client.NewForConfig(&input.Clientset.RestConfig) if err != nil { return fmt.Errorf("Error creating corev1 client: %w", err) } req := coreclient.RESTClient().Post(). Resource("pods"). - Namespace(selectedPod.Namespace). - Name(selectedPod.Name). + Namespace(input.SelectedPod.Namespace). + Name(input.SelectedPod.Name). SubResource("exec") - req.Param("container", fmt.Sprintf("%s-%s", appName, processType)) + req.Param("container", fmt.Sprintf("%s-%s", input.AppName, input.ProcessType)) req.Param("stdin", "true") req.Param("stdout", "true") req.Param("stderr", "true") req.Param("tty", "true") - if selectedPod.Annotations["dokku.com/builder-type"] == "herokuish" { - req.Param("command", "/exec") + if input.Entrypoint != "" { + req.Param("command", input.Entrypoint) } - for _, cmd := range command { + for _, cmd := range input.Command { req.Param("command", cmd) } @@ -506,7 +533,7 @@ func TriggerSchedulerEnter(scheduler string, appName string, processType string, sizeQueue := t.MonitorSize(size) return t.Safe(func() error { - exec, err := remotecommand.NewSPDYExecutor(&clientset.RestConfig, "POST", req.URL()) + exec, err := remotecommand.NewSPDYExecutor(&input.Clientset.RestConfig, "POST", req.URL()) if err != nil { return fmt.Errorf("Error creating executor: %w", err) } @@ -571,7 +598,7 @@ func TriggerSchedulerLogs(scheduler string, appName string, processType string, ch := make(chan bool) podItems := podList.Items - if os.Getenv("FORCE_COLOR") == "1" { + if os.Getenv("FORCE_TTY") == "1" { color.NoColor = false } @@ -692,3 +719,440 @@ func TriggerSchedulerStop(scheduler string, appName string) error { return nil } + +// TriggerSchedulerRun runs a command in an ephemeral container +func TriggerSchedulerRun(scheduler string, appName string, envCount int, args []string) error { + if scheduler != "k3s" { + return nil + } + + extraEnv := map[string]string{} + if envCount > 0 { + var envPairs []string + envPairs, args = args[0:envCount], args[envCount:] + for _, envPair := range envPairs { + parts := strings.SplitN(envPair, "=", 2) + if len(parts) != 2 { + return fmt.Errorf("Invalid environment variable pair: %s", envPair) + } + extraEnv[parts[0]] = parts[1] + } + } + + imageTag, err := common.GetRunningImageTag(appName, "") + if err != nil { + return fmt.Errorf("Error getting running image tag: %w", err) + } + imageRepo := common.GetAppImageRepo(appName) + image, err := common.GetDeployingAppImageName(appName, imageTag, imageRepo) + if err != nil { + return fmt.Errorf("Error getting deploying app image name: %w", err) + } + + imageStage, err := common.DockerInspect(image, "{{ index .Config.Labels \"com.dokku.image-stage\" }}") + if err != nil { + return fmt.Errorf("Error getting image stage: %w", err) + } + if imageStage != "release" { + common.LogWarn("Invalid image stage detected: expected 'release', got '$IMAGE_STAGE'") + return fmt.Errorf("Successfully deploy your app to fix dokku run calls") + } + + dokkuRmContainer := os.Getenv("DOKKU_RM_CONTAINER") + if dokkuRmContainer == "" { + appRmContainer, err := common.CallPlugnTrigger(common.PlugnTriggerInput{ + Trigger: "config-get", + Args: []string{appName, "DOKKU_RM_CONTAINER"}, + CaptureOutput: true, + StreamStdio: false, + }) + if err != nil { + globalRmContainer, err := common.CallPlugnTrigger(common.PlugnTriggerInput{ + Trigger: "config-get-global", + Args: []string{"DOKKU_RM_CONTAINER"}, + CaptureOutput: true, + StreamStdio: false, + }) + if err == nil { + dokkuRmContainer = globalRmContainer.Stdout + } + } else { + dokkuRmContainer = appRmContainer.Stdout + } + } + if dokkuRmContainer == "" { + dokkuRmContainer = "true" + } + + rmContainer, err := strconv.ParseBool(dokkuRmContainer) + if err != nil { + return fmt.Errorf("Error parsing DOKKU_RM_CONTAINER value as boolean: %w", err) + } + + labels := map[string]string{ + "dokku.com/app-name": appName, + } + + if os.Getenv("DOKKU_TRACE") == "1" { + extraEnv["TRACE"] = "true" + } + + processType := "run" + if os.Getenv("DOKKU_CRON_ID") != "" { + processType = "cron" + labels["dokku.com/cron-id"] = os.Getenv("DOKKU_CRON_ID") + } + + dockerArgs := []string{} + if b, err := common.PlugnTriggerSetup("docker-args-run", []string{appName, imageTag}...).SetInput("").Output(); err == nil { + words, err := shellquote.Split(strings.TrimSpace(string(b[:]))) + if err != nil { + return err + } + + dockerArgs = append(dockerArgs, words...) + } + + imageSourceType, err := common.DockerInspect(image, "{{ index .Config.Labels \"com.dokku.builder-type\" }}") + if err != nil { + return fmt.Errorf("Error getting image builder type: %w", err) + } + + if b, err := common.PlugnTriggerSetup("docker-args-process-run", []string{appName, imageSourceType, imageTag}...).SetInput("").Output(); err == nil { + words, err := shellquote.Split(strings.TrimSpace(string(b[:]))) + if err != nil { + return err + } + + dockerArgs = append(dockerArgs, words...) + } + + // todo: do something with docker args + command := args + if len(args) == 0 { + command = []string{"/bin/bash"} + if globalShell, err := common.PlugnTriggerOutputAsString("config-get-global", []string{"DOKKU_APP_SHELL"}...); err == nil && globalShell != "" { + command = []string{globalShell} + } + if appShell, err := common.PlugnTriggerOutputAsString("config-get", []string{appName, "DOKKU_APP_SHELL"}...); err == nil && appShell != "" { + command = []string{appShell} + } + } else if len(args) == 1 { + resp, err := common.CallPlugnTrigger(common.PlugnTriggerInput{ + Trigger: "procfile-get-command", + Args: []string{appName, args[0], "5000"}, + CaptureOutput: true, + StreamStdio: false, + }) + if err == nil && resp.Stdout != "" { + common.LogInfo1Quiet(fmt.Sprintf("Found '%s' in Procfile, running that command", args[0])) + return err + } + // todo: run command in procfile + } + + entrypoint := "" + if imageSourceType == "herokuish" { + entrypoint = "/exec" + } + + namespace := common.PropertyGetDefault("scheduler-k3s", appName, "namespace", "default") + helmAgent, err := NewHelmAgent(namespace, DevNullPrinter) + if err != nil { + return fmt.Errorf("Error creating helm agent: %w", err) + } + + values, err := helmAgent.GetValues(fmt.Sprintf("dokku-%s", appName)) + if err != nil { + return fmt.Errorf("Error getting helm values: %w", err) + } + + deploymentIDValue, ok := values["deploment_id"].(string) + if !ok { + return fmt.Errorf("Deployment ID is not a string") + } + if len(deploymentIDValue) == 0 { + return fmt.Errorf("Deployment ID is empty") + } + deploymentID, err := strconv.ParseInt(deploymentIDValue, 10, 64) + if err != nil { + return fmt.Errorf("Error parsing deployment ID: %w", err) + } + + attachToPod := os.Getenv("DOKKU_DETACH_CONTAINER") != "1" + imagePullSecrets := common.PropertyGetDefault("scheduler-k3s", appName, "image-pull-secrets", "") + workingDir := common.GetWorkingDir(appName, image) + job, err := templateKubernetesJob(Job{ + AppName: appName, + Command: command, + DeploymentID: deploymentID, + Entrypoint: entrypoint, + Env: extraEnv, + Image: image, + ImagePullSecrets: imagePullSecrets, + ImageSourceType: imageSourceType, + Interactive: attachToPod, + Labels: labels, + Namespace: namespace, + ProcessType: processType, + RemoveContainer: rmContainer, + WorkingDir: workingDir, + }) + if err != nil { + return fmt.Errorf("Error templating job: %w", err) + } + + if os.Getenv("FORCE_TTY") == "1" { + color.NoColor = false + } + + err = writeResourceToFile(WriteResourceInput{ + Object: &job, + Path: "/tmp/job.yaml", + }) + if err != nil { + return fmt.Errorf("Error writing job: %w", err) + } + common.CatFile("/tmp/job.yaml") + + ctx, cancel := context.WithCancel(context.Background()) + signals := make(chan os.Signal, 1) + signal.Notify(signals, os.Interrupt, syscall.SIGHUP, + syscall.SIGINT, + syscall.SIGQUIT, + syscall.SIGTERM) + go func() { + <-signals + cancel() + }() + + createdJob, err := createKubernetesJob(ctx, job) + if err != nil { + return fmt.Errorf("Error creating job: %w", err) + } + + clientset, err := NewKubernetesClient() + if err != nil { + return fmt.Errorf("Error creating kubernetes client: %w", err) + } + + batchJobSelector := fmt.Sprintf("batch.kubernetes.io/job-name=%s", createdJob.Name) + podList, err := WaitForPodToExist(ctx, WaitForPodToExistInput{ + Clientset: clientset, + Namespace: namespace, + RetryCount: 3, + Selector: batchJobSelector, + }) + if err != nil { + return fmt.Errorf("Error waiting for pod to exist: %w", err) + } + + for _, pod := range podList.Items { + common.LogQuiet(pod.Name) + } + + if !attachToPod { + return nil + } + + err = WaitForPodBySelectorRunning(ctx, WaitForPodBySelectorRunningInput{ + Clientset: clientset, + Namespace: namespace, + Selector: batchJobSelector, + Timeout: 300, + Waiter: isPodReady, + }) + if err != nil { + if errors.Is(err, conditions.ErrPodCompleted) { + podList, podErr := GetPod(ctx, GetPodInput{ + Clientset: clientset, + Namespace: namespace, + Selector: batchJobSelector, + }) + if podErr != nil { + return fmt.Errorf("Error completed pod: %w", err) + } + selectedPod := podList.Items[0] + if selectedPod.Status.Phase == v1.PodFailed { + for _, status := range selectedPod.Status.ContainerStatuses { + if status.Name != fmt.Sprintf("%s-%s", appName, processType) { + continue + } + if status.State.Terminated == nil { + continue + } + return fmt.Errorf("Unable to attach as the pod has already exited with a failed exit code: %s", status.State.Terminated.Message) + } + + return fmt.Errorf("Unable to attach as the pod has already exited with a failed exit code") + } else if selectedPod.Status.Phase == v1.PodSucceeded { + return errors.New("Unable to attach as the pod has already exited with a successful exit code") + } + } + return fmt.Errorf("Error waiting for pod to be running: %w", err) + } + + podList, err = GetPod(ctx, GetPodInput{ + Clientset: clientset, + Namespace: namespace, + Selector: batchJobSelector, + }) + if err != nil { + return fmt.Errorf("Error getting pod: %w", err) + } + selectedPod := podList.Items[0] + + switch selectedPod.Status.Phase { + case v1.PodFailed, v1.PodSucceeded: + if selectedPod.Status.Phase == v1.PodFailed { + for _, status := range selectedPod.Status.ContainerStatuses { + if status.Name != fmt.Sprintf("%s-%s", appName, processType) { + continue + } + if status.State.Terminated == nil { + continue + } + return fmt.Errorf("Unable to attach as the pod has already exited with a failed exit code: %s", status.State.Terminated.Message) + } + + return fmt.Errorf("Unable to attach as the pod has already exited with a failed exit code") + } else if selectedPod.Status.Phase == v1.PodSucceeded { + return errors.New("Unable to attach as the pod has already exited with a successful exit code") + } + case v1.PodRunning: + err := enterPod(ctx, EnterPodInput{ + AppName: appName, + Clientset: clientset, + Command: command, + Entrypoint: entrypoint, + ProcessType: processType, + SelectedPod: selectedPod, + }) + if err != nil { + return err + } + + err = clientset.Client.BatchV1().Jobs(namespace).Delete(ctx, createdJob.Name, metav1.DeleteOptions{ + PropagationPolicy: ptr.To(metav1.DeletePropagationForeground), + }) + if err != nil { + return fmt.Errorf("Error deleting pod: %w", err) + } + + return nil + default: + return fmt.Errorf("Unable to attach as the pod is in an unknown state: %s", selectedPod.Status.Phase) + } + // todo: support scheduler-post-run + + return nil +} + +func isPodReady(ctx context.Context, clientset KubernetesClient, podName, namespace string) wait.ConditionWithContextFunc { + return func(ctx context.Context) (bool, error) { + fmt.Printf(".") // progress bar! + + pod, err := clientset.Client.CoreV1().Pods(namespace).Get(ctx, podName, metav1.GetOptions{}) + if err != nil { + return false, err + } + + switch pod.Status.Phase { + case v1.PodRunning: + return true, nil + case v1.PodFailed, v1.PodSucceeded: + return false, conditions.ErrPodCompleted + } + return false, nil + } +} + +func isPodComplete(ctx context.Context, clientset KubernetesClient, podName, namespace string) wait.ConditionWithContextFunc { + return func(ctx context.Context) (bool, error) { + fmt.Printf(".") // progress bar! + + pod, err := clientset.Client.CoreV1().Pods(namespace).Get(ctx, podName, metav1.GetOptions{}) + if err != nil { + return false, err + } + + switch pod.Status.Phase { + case v1.PodFailed, v1.PodSucceeded: + return true, nil + } + return false, nil + } +} + +type WaitForPodBySelectorRunningInput struct { + Clientset KubernetesClient + Namespace string + Selector string + Timeout int + Waiter func(ctx context.Context, clientset KubernetesClient, podName, namespace string) wait.ConditionWithContextFunc +} + +type WaitForPodToExistInput struct { + Clientset KubernetesClient + Namespace string + RetryCount int + Selector string +} + +func WaitForPodToExist(ctx context.Context, input WaitForPodToExistInput) (v1.PodList, error) { + var podList v1.PodList + var err error + for i := 0; i < input.RetryCount; i++ { + podList, err = GetPod(ctx, GetPodInput{ + Clientset: input.Clientset, + Namespace: input.Namespace, + Selector: input.Selector, + }) + if err == nil { + break + } + time.Sleep(1 * time.Second) + } + if err != nil { + return podList, fmt.Errorf("Error listing pods: %w", err) + } + return podList, nil +} + +type GetPodInput struct { + Clientset KubernetesClient + Namespace string + Selector string +} + +func GetPod(ctx context.Context, input GetPodInput) (v1.PodList, error) { + listOptions := metav1.ListOptions{LabelSelector: input.Selector} + podList, err := input.Clientset.Client.CoreV1().Pods(input.Namespace).List(ctx, listOptions) + return *podList, err +} + +func WaitForPodBySelectorRunning(ctx context.Context, input WaitForPodBySelectorRunningInput) error { + podList, err := WaitForPodToExist(ctx, WaitForPodToExistInput{ + Clientset: input.Clientset, + Namespace: input.Namespace, + RetryCount: 3, + Selector: input.Selector, + }) + if err != nil { + return fmt.Errorf("Error waiting for pod to exist: %w", err) + } + + if len(podList.Items) == 0 { + return fmt.Errorf("no pods in %s with selector %s", input.Namespace, input.Selector) + } + + timeout := time.Duration(input.Timeout) * time.Second + for _, pod := range podList.Items { + if err := wait.PollUntilContextTimeout(ctx, time.Second, timeout, false, input.Waiter(ctx, input.Clientset, pod.Name, pod.Namespace)); err != nil { + print("\n") + return err + } + } + print("\n") + return nil +}