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

feat: implement dokku enter for k8s scheduler

Browse Source
This commit is contained in:
Jose Diaz-Gonzalez
2024-01-17 20:56:30 -05:00
parent 6fb03eda4e
commit 737f54a5c7
7 changed files with 235 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
docs/processes/entering-containers.md
View File

@@ -40,3 +40,10 @@ dokku enter node-js-app web echo hi
# run a long-running command, as one might for a cron task
dokku enter node-js-app web python script/background-worker.py
```
If the custom command uses flags, it must be separated from the `dokku enter` command via `--` (double dashes):
```shell
# list files in the app
dokku enter node-js-app web -- ls -lah
```

2
plugins/scheduler-k3s/Makefile
View File

@@ -1,5 +1,5 @@
SUBCOMMANDS = subcommands/initialize subcommands/report subcommands/set
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 = 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
BUILD = commands subcommands triggers
PLUGIN_NAME = scheduler-k3s

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

@@ -9,8 +9,12 @@ import (
"github.com/dokku/dokku/plugins/config"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
"k8s.io/client-go/tools/clientcmd"
clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
"mvdan.cc/sh/v3/shell"
)
@@ -34,22 +38,56 @@ type StartCommandOutput struct {
Command []string
}
func NewClient() (*kubernetes.Clientset, error) {
config, err := clientcmd.BuildConfigFromFlags("", KubeConfigPath)
type KubernetesClient struct {
Client kubernetes.Clientset
RestClient rest.Interface
RestConfig rest.Config
}
func NewKubernetesClient() (KubernetesClient, error) {
clientConfig := KubernetesClientConfig()
restConf, err := clientConfig.ClientConfig()
if err != nil {
return &kubernetes.Clientset{}, err
return KubernetesClient{}, err
}
return kubernetes.NewForConfig(config)
restConf.GroupVersion = &schema.GroupVersion{
Group: "api",
Version: "v1",
}
client, err := kubernetes.NewForConfig(restConf)
if err != nil {
return KubernetesClient{}, err
}
restConf.NegotiatedSerializer = runtime.NewSimpleNegotiatedSerializer(runtime.SerializerInfo{})
restClient, err := rest.RESTClientFor(restConf)
if err != nil {
return KubernetesClient{}, err
}
return KubernetesClient{
Client: *client,
RestConfig: *restConf,
RestClient: restClient,
}, nil
}
func KubernetesClientConfig() clientcmd.ClientConfig {
return clientcmd.NewNonInteractiveDeferredLoadingClientConfig(
&clientcmd.ClientConfigLoadingRules{ExplicitPath: KubeConfigPath},
&clientcmd.ConfigOverrides{ClusterInfo: clientcmdapi.Cluster{Server: ""}})
}
func createKubernetesNamespace(ctx context.Context, namespaceName string) error {
client, err := NewClient()
clientset, err := NewKubernetesClient()
if err != nil {
return err
}
namespaces, err := client.CoreV1().Namespaces().List(ctx, metav1.ListOptions{})
namespaces, err := clientset.Client.CoreV1().Namespaces().List(ctx, metav1.ListOptions{})
if err != nil {
return err
}
@@ -65,7 +103,7 @@ func createKubernetesNamespace(ctx context.Context, namespaceName string) error
Name: namespaceName,
},
}
_, err = client.CoreV1().Namespaces().Create(ctx, namespace, metav1.CreateOptions{})
_, err = clientset.Client.CoreV1().Namespaces().Create(ctx, namespace, metav1.CreateOptions{})
if err != nil {
return err
}

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

@@ -16,6 +16,8 @@ require (
k8s.io/apimachinery v0.29.0
k8s.io/cli-runtime v0.29.0
k8s.io/client-go v0.29.0
k8s.io/klog/v2 v2.110.1
k8s.io/kubectl v0.28.4
k8s.io/utils v0.0.0-20240102154912-e7106e64919e
mvdan.cc/sh/v3 v3.7.0
)
@@ -154,9 +156,7 @@ require (
k8s.io/apiextensions-apiserver v0.28.4 // indirect
k8s.io/apiserver v0.28.4 // indirect
k8s.io/component-base v0.28.4 // indirect
k8s.io/klog/v2 v2.110.1 // indirect
k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00 // indirect
k8s.io/kubectl v0.28.4 // indirect
oras.land/oras-go v1.2.4 // indirect
sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
sigs.k8s.io/kustomize/api v0.13.5-0.20230601165947-6ce0bf390ce3 // indirect

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

@@ -6,6 +6,7 @@ import (
"strings"
flag "github.com/spf13/pflag"
"k8s.io/utils/ptr"
"github.com/dokku/dokku/plugins/common"
scheduler_k3s "github.com/dokku/dokku/plugins/scheduler-k3s"
@@ -15,6 +16,7 @@ import (
func main() {
parts := strings.Split(os.Args[0], "/")
trigger := parts[len(parts)-1]
podIdentifier := flag.String("container-id", "", "--container-id: A pod identifier")
flag.Parse()
var err error
@@ -45,6 +47,18 @@ func main() {
appName := flag.Arg(1)
imageTag := flag.Arg(2)
err = scheduler_k3s.TriggerSchedulerDeploy(scheduler, appName, imageTag)
case "scheduler-enter":
scheduler := flag.Arg(0)
appName := flag.Arg(1)
containerType := flag.Arg(2)
args := flag.Args()
if len(args) > 3 {
_, args = common.ShiftString(args)
_, args = common.ShiftString(args)
_, args = common.ShiftString(args)
}
err = scheduler_k3s.TriggerSchedulerEnter(scheduler, appName, containerType, ptr.Deref(podIdentifier, ""), args)
default:
err = fmt.Errorf("Invalid plugin trigger call: %s", trigger)
}

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

@@ -200,23 +200,24 @@ func printResource(input PrintInput) error {
func templateKubernetesDeployment(input Deployment) (appsv1.Deployment, error) {
labels := map[string]string{
"app": input.AppName,
"process-type": input.ProcessType,
"app-process-type": fmt.Sprintf("%s-%s", input.AppName, input.ProcessType),
"dokku.com/app-name": input.AppName,
"dokku.com/process-type": input.ProcessType,
"dokku.com/app-process-type": fmt.Sprintf("%s-%s", input.AppName, input.ProcessType),
}
annotations := map[string]string{
"dokku.com/app-name": input.AppName,
"dokku.com/deployment-id": "DEPLOYMENT_ID_QUOTED",
"dokku.com/process-type": input.ProcessType,
"dokku.com/builder-type": input.ImageSourceType,
}
secretName := fmt.Sprintf("env-%s.DEPLOYMENT_ID", input.AppName)
deployment := appsv1.Deployment{
ObjectMeta: metav1.ObjectMeta{
Name: fmt.Sprintf("%s-%s", input.AppName, input.ProcessType),
Namespace: input.Namespace,
Labels: labels,
Annotations: map[string]string{
"dokku.com/app-name": input.AppName,
"dokku.com/deployment-id": "DEPLOYMENT_ID_QUOTED",
"dokku.com/process-type": input.ProcessType,
"dokku.com/builder-type": input.ImageSourceType,
},
Name: fmt.Sprintf("%s-%s", input.AppName, input.ProcessType),
Namespace: input.Namespace,
Labels: labels,
Annotations: annotations,
},
Spec: appsv1.DeploymentSpec{
Replicas: ptr.To(input.Replicas),
@@ -226,7 +227,8 @@ func templateKubernetesDeployment(input Deployment) (appsv1.Deployment, error) {
},
Template: corev1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{
Labels: labels,
Labels: labels,
Annotations: annotations,
},
Spec: corev1.PodSpec{
Containers: []corev1.Container{
@@ -344,7 +346,7 @@ func templateKubernetesIngressRoute(input IngressRoute) traefikv1alpha1.IngressR
Name: fmt.Sprintf("%s-%s", input.ServiceName, port),
Namespace: input.Namespace,
Labels: map[string]string{
"dokku.com/app": input.AppName,
"dokku.com/app-name": input.AppName,
"dokku.com/process-type": input.ProcessType,
},
},
@@ -400,14 +402,14 @@ func templateKubernetesService(input Service) corev1.Service {
Name: fmt.Sprintf("%s-%s", input.AppName, "web"),
Namespace: input.Namespace,
Labels: map[string]string{
"dokku.com/app": input.AppName,
"dokku.com/app-name": input.AppName,
"dokku.com/process-type": "web",
"dokku.com/app-process-type": fmt.Sprintf("%s-%s", input.AppName, "web"),
},
},
Spec: corev1.ServiceSpec{
Selector: map[string]string{
"dokku.com/app": input.AppName,
"dokku.com/app-name": input.AppName,
"dokku.com/process-type": "web",
"dokku.com/app-process-type": fmt.Sprintf("%s-%s", input.AppName, "web"),
},

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

@@ -5,8 +5,12 @@ import (
"encoding/base64"
"fmt"
"os"
"os/signal"
"path/filepath"
"slices"
"strconv"
"strings"
"syscall"
"time"
"github.com/dokku/dokku/plugins/common"
@@ -15,6 +19,11 @@ import (
orderedmap "github.com/wk8/go-ordered-map/v2"
"gopkg.in/yaml.v3"
appsv1 "k8s.io/api/apps/v1"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
"k8s.io/client-go/tools/remotecommand"
"k8s.io/kubectl/pkg/util/term"
)
// TriggerInstall runs the install step for the scheduler-k3s plugin
@@ -354,3 +363,142 @@ data:
return nil
}
// TriggerSchedulerEnter enters a container for a given application
func TriggerSchedulerEnter(scheduler string, appName string, processType string, podName string, args []string) error {
if scheduler != "k3s" {
return nil
}
common.LogDebug(fmt.Sprintf("%s %s %s %s", scheduler, appName, processType, podName))
clientset, err := NewKubernetesClient()
if err != nil {
return fmt.Errorf("Error creating kubernetes client: %w", err)
}
ctx := context.Background()
ctx, cancel := context.WithCancel(ctx)
cSignal := make(chan os.Signal, 2)
signal.Notify(cSignal, os.Interrupt, syscall.SIGTERM)
go func() {
<-cSignal
common.LogWarn("Exiting pod")
cancel()
}()
namespace := common.PropertyGetDefault("scheduler-k3s", appName, "namespace", "default")
labelSelector := []string{fmt.Sprintf("dokku.com/app-name=%s", appName)}
processIndex := 1
if processType != "" {
parts := strings.SplitN(processType, ".", 2)
if len(parts) == 2 {
processType = parts[0]
processIndex, err = strconv.Atoi(parts[1])
if err != nil {
return fmt.Errorf("Error parsing process index: %w", err)
}
}
labelSelector = append(labelSelector, fmt.Sprintf("dokku.com/process-type=%s", processType))
}
podList, err := clientset.Client.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{
LabelSelector: strings.Join(labelSelector, ","),
})
if err != nil {
return fmt.Errorf("Error listing pods: %w", err)
}
if podList.Items == nil {
return fmt.Errorf("No pods found for app %s", appName)
}
if len(podList.Items) == 0 {
return fmt.Errorf("No pods found for app %s", appName)
}
pods := []string{}
for _, pod := range podList.Items {
pods = append(pods, pod.Name)
}
slices.Sort(pods)
processIndex--
if processIndex > len(podList.Items) {
return fmt.Errorf("Process index %d out of range for app %s", processIndex, appName)
}
var selectedPod corev1.Pod
if podName != "" {
for _, pod := range podList.Items {
if pod.Name == podName {
selectedPod = pod
break
}
}
} else {
selectedPod = podList.Items[processIndex]
}
processType, ok := selectedPod.Labels["dokku.com/process-type"]
if !ok {
return fmt.Errorf("Pod %s does not have a process type label", selectedPod.Name)
}
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}
}
}
coreclient, err := corev1client.NewForConfig(&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).
SubResource("exec")
req.Param("container", fmt.Sprintf("%s-%s", appName, 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")
}
for _, cmd := range command {
req.Param("command", cmd)
}
t := term.TTY{
In: os.Stdin,
Out: os.Stdout,
Raw: true,
}
size := t.GetSize()
sizeQueue := t.MonitorSize(size)
return t.Safe(func() error {
exec, err := remotecommand.NewSPDYExecutor(&clientset.RestConfig, "POST", req.URL())
if err != nil {
return fmt.Errorf("Error creating executor: %w", err)
}
return exec.StreamWithContext(ctx, remotecommand.StreamOptions{
Stdin: os.Stdin,
Stdout: os.Stdout,
Stderr: os.Stderr,
Tty: true,
TerminalSizeQueue: sizeQueue,
})
})
}