github/dokku
0
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat: implement run and run:detached

Browse Source 
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.
This commit is contained in:
Jose Diaz-Gonzalez
2024-01-18 05:28:07 -05:00
parent 6b85374857
commit 8b8444fb1f
10 changed files with 833 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
docs/deployment/schedulers/k3s.md
View File

@@ -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

2
plugins/scheduler-k3s/Makefile
View File

@@ -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

19
plugins/scheduler-k3s/functions.go
View File

@@ -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
}

13
plugins/scheduler-k3s/go.mod
View File

@@ -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

30
plugins/scheduler-k3s/go.sum
View File

@@ -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=

93
plugins/scheduler-k3s/helm.go
View File

@@ -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

2
plugins/scheduler-k3s/scheduler_k3s.go
View File

@@ -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)
})

21
plugins/scheduler-k3s/src/triggers/triggers.go
View File

@@ -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)
}

266
plugins/scheduler-k3s/template.go
View File

@@ -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)

494
plugins/scheduler-k3s/triggers.go
View File

@@ -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
}