github/k3k
1
0
Fork 0
mirror of https://github.com/rancher/k3k.git synced 2026-08-23 22:36:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

use statefulsets for servers (#67)

Browse Source 
* use statefulsets for servers

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* remove unused code

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* fix comments

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

---------

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>
This commit is contained in:
Hussein Galal
2023-11-28 23:51:50 +02:00
committed by GitHub
parent dc2f410c17
commit dd618e580a
4 changed files with 260 additions and 373 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/cmds/cluster/create.go
+5 -5
View File
@@ -12,7 +12,7 @@ import (
"github.com/rancher/k3k/cli/cmds"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
"github.com/rancher/k3k/pkg/controller/cluster"
"github.com/rancher/k3k/pkg/controller/cluster/server"
"github.com/rancher/k3k/pkg/controller/util"
"github.com/sirupsen/logrus"
"github.com/urfave/cli"
@@ -34,7 +34,7 @@ var (
Scheme = runtime.NewScheme()
backoff = wait.Backoff{
Steps: 5,
Duration: 10 * time.Second,
Duration: 20 * time.Second,
Factor: 2,
Jitter: 0.1,
}
@@ -93,7 +93,7 @@ var (
cli.StringFlag{
Name: "persistence-type",
Usage: "Persistence mode for the nodes (ephermal, static, dynamic)",
Value: cluster.EphermalNodesType,
Value: server.EphermalNodesType,
Destination: &persistenceType,
},
cli.StringFlag{
@@ -200,8 +200,8 @@ func createCluster(clx *cli.Context) error {
}
func validateCreateFlags(clx *cli.Context) error {
if persistenceType != cluster.EphermalNodesType &&
persistenceType != cluster.DynamicNodesType {
if persistenceType != server.EphermalNodesType &&
persistenceType != server.DynamicNodesType {
return errors.New("invalid persistence type")
}
if token == "" {
pkg/controller/cluster/cluster.go
+33 -115
View File
@@ -26,8 +26,6 @@ import (
const (
clusterController = "k3k-cluster-controller"
clusterFinalizerName = "cluster.k3k.io/finalizer"
EphermalNodesType = "ephermal"
DynamicNodesType = "dynamic"
maxConcurrentReconciles = 1
@@ -105,13 +103,12 @@ func (c *ClusterReconciler) Reconcile(ctx context.Context, req reconcile.Request
}
func (c *ClusterReconciler) createCluster(ctx context.Context, cluster *v1alpha1.Cluster) error {
server := server.New(cluster, c.Client)
agent := agent.New(cluster)
s := server.New(cluster, c.Client)
if cluster.Spec.Persistence == nil {
// default to ephermal nodes
cluster.Spec.Persistence = &v1alpha1.PersistenceConfig{
Type: EphermalNodesType,
Type: server.EphermalNodesType,
}
}
if err := c.Client.Update(ctx, cluster); err != nil {
@@ -133,7 +130,7 @@ func (c *ClusterReconciler) createCluster(ctx context.Context, cluster *v1alpha1
}
klog.Infof("creating cluster service")
serviceIP, err := c.createClusterService(ctx, cluster, server)
serviceIP, err := c.createClusterService(ctx, cluster, s)
if err != nil {
return util.LogAndReturnErr("failed to create cluster service", err)
}
@@ -143,23 +140,22 @@ func (c *ClusterReconciler) createCluster(ctx context.Context, cluster *v1alpha1
}
// creating statefulsets in case the user chose a persistence type other than ephermal
if cluster.Spec.Persistence.Type != EphermalNodesType {
if cluster.Spec.Persistence.StorageRequestSize == "" {
// default to 1G of request size
cluster.Spec.Persistence.StorageRequestSize = "1G"
}
if err := c.createStatefulSets(ctx, cluster, server, agent); err != nil {
return util.LogAndReturnErr("failed to create servers and agents statefulsets", err)
}
} else {
if err := c.createDeployments(ctx, cluster, server); err != nil {
return util.LogAndReturnErr("failed to create servers and agents deployment", err)
}
if cluster.Spec.Persistence.StorageRequestSize == "" {
// default to 1G of request size
cluster.Spec.Persistence.StorageRequestSize = "1G"
}
if err := c.server(ctx, cluster, s); err != nil {
return util.LogAndReturnErr("failed to create servers", err)
}
if err := c.agent(ctx, cluster); err != nil {
return util.LogAndReturnErr("failed to create agents", err)
}
if cluster.Spec.Expose != nil {
if cluster.Spec.Expose.Ingress != nil {
serverIngress, err := server.Ingress(ctx, c.Client)
serverIngress, err := s.Ingress(ctx, c.Client)
if err != nil {
return util.LogAndReturnErr("failed to create ingress object", err)
}
@@ -172,7 +168,7 @@ func (c *ClusterReconciler) createCluster(ctx context.Context, cluster *v1alpha1
}
}
kubeconfigSecret, err := server.GenerateNewKubeConfig(ctx, serviceIP)
kubeconfigSecret, err := s.GenerateNewKubeConfig(ctx, serviceIP)
if err != nil {
return util.LogAndReturnErr("failed to generate new kubeconfig", err)
}
@@ -277,40 +273,36 @@ func (c *ClusterReconciler) createClusterService(ctx context.Context, cluster *v
return service.Spec.ClusterIP, nil
}
func (c *ClusterReconciler) createDeployments(ctx context.Context, cluster *v1alpha1.Cluster, server *server.Server) error {
// create deployment for the init server
// the init deployment must have only 1 replica
initServerDeployment, err := server.Deploy(ctx, true)
func (c *ClusterReconciler) server(ctx context.Context, cluster *v1alpha1.Cluster, server *server.Server) error {
// create headless service for the statefulset
serverStatefulService := server.StatefulServerService(cluster)
if err := controllerutil.SetControllerReference(cluster, serverStatefulService, c.Scheme); err != nil {
return err
}
if err := c.Client.Create(ctx, serverStatefulService); err != nil {
if !apierrors.IsAlreadyExists(err) {
return err
}
}
ServerStatefulSet, err := server.StatefulServer(ctx, cluster)
if err != nil {
return err
}
if err := controllerutil.SetControllerReference(cluster, initServerDeployment, c.Scheme); err != nil {
if err := controllerutil.SetControllerReference(cluster, ServerStatefulSet, c.Scheme); err != nil {
return err
}
if err := c.Client.Create(ctx, initServerDeployment); err != nil {
if err := c.Client.Create(ctx, ServerStatefulSet); err != nil {
if !apierrors.IsAlreadyExists(err) {
return err
}
}
// create deployment for the rest of the servers
serversDeployment, err := server.Deploy(ctx, false)
if err != nil {
return err
}
if err := controllerutil.SetControllerReference(cluster, serversDeployment, c.Scheme); err != nil {
return err
}
if err := c.Client.Create(ctx, serversDeployment); err != nil {
if !apierrors.IsAlreadyExists(err) {
return err
}
}
return nil
}
func (c *ClusterReconciler) agent(ctx context.Context, cluster *v1alpha1.Cluster) error {
agent := agent.New(cluster)
agentsDeployment := agent.Deploy()
@@ -323,80 +315,6 @@ func (c *ClusterReconciler) createDeployments(ctx context.Context, cluster *v1al
return err
}
}
return nil
}
func (c *ClusterReconciler) createStatefulSets(ctx context.Context, cluster *v1alpha1.Cluster, server *server.Server, agent *agent.Agent) error {
// create headless service for the init statefulset
initServerStatefulService := server.StatefulServerService(cluster, true)
if err := controllerutil.SetControllerReference(cluster, initServerStatefulService, c.Scheme); err != nil {
return err
}
if err := c.Client.Create(ctx, initServerStatefulService); err != nil {
if !apierrors.IsAlreadyExists(err) {
return err
}
}
// create statefulsets for the init server
// the init statefulset must have only 1 replica
initServerStatefulSet := server.StatefulServer(ctx, cluster, true)
if err := controllerutil.SetControllerReference(cluster, initServerStatefulSet, c.Scheme); err != nil {
return err
}
if err := c.Client.Create(ctx, initServerStatefulSet); err != nil {
if !apierrors.IsAlreadyExists(err) {
return err
}
}
// create statefulset for the rest of the servers
// create headless service for the server statefulset
serverStatefulService := server.StatefulServerService(cluster, false)
if err := controllerutil.SetControllerReference(cluster, serverStatefulService, c.Scheme); err != nil {
return err
}
if err := c.Client.Create(ctx, serverStatefulService); err != nil {
if !apierrors.IsAlreadyExists(err) {
return err
}
}
serversStatefulSet := server.StatefulServer(ctx, cluster, false)
if err := controllerutil.SetControllerReference(cluster, serversStatefulSet, c.Scheme); err != nil {
return err
}
if err := c.Client.Create(ctx, serversStatefulSet); err != nil {
if !apierrors.IsAlreadyExists(err) {
return err
}
}
// create headless service for the agents statefulset
agentStatefulService := agent.StatefulAgentService(cluster)
if err := controllerutil.SetControllerReference(cluster, agentStatefulService, c.Scheme); err != nil {
return err
}
if err := c.Client.Create(ctx, agentStatefulService); err != nil {
if !apierrors.IsAlreadyExists(err) {
return err
}
}
agentsStatefulSet := agent.StatefulAgent(cluster)
if err := controllerutil.SetControllerReference(cluster, agentsStatefulSet, c.Scheme); err != nil {
return err
}
if err := c.Client.Create(ctx, agentsStatefulSet); err != nil {
if !apierrors.IsAlreadyExists(err) {
return err
}
}
return nil
}
pkg/controller/cluster/server/server.go
+221 -246
View File
@@ -2,7 +2,7 @@ package server
import (
"context"
"strconv"
"strings"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
"github.com/rancher/k3k/pkg/controller/util"
@@ -22,6 +22,9 @@ const (
initServerName = serverName + "init-server"
initContainerName = serverName + "server-check"
initContainerImage = "alpine/curl"
EphermalNodesType = "ephermal"
DynamicNodesType = "dynamic"
)
// Server
@@ -37,18 +40,220 @@ func New(cluster *v1alpha1.Cluster, client client.Client) *Server {
}
}
func (s *Server) Deploy(ctx context.Context, init bool) (*apps.Deployment, error) {
var replicas int32
image := util.K3SImage(s.cluster)
name := serverName + "server"
if init {
name = serverName + "init-server"
func (s *Server) podSpec(ctx context.Context, image, name string, persistent bool) v1.PodSpec {
podSpec := v1.PodSpec{
Volumes: []v1.Volume{
{
Name: "initconfig",
VolumeSource: v1.VolumeSource{
Secret: &v1.SecretVolumeSource{
SecretName: "k3k-init-server-config",
Items: []v1.KeyToPath{
{
Key: "config.yaml",
Path: "config.yaml",
},
},
},
},
},
{
Name: "config",
VolumeSource: v1.VolumeSource{
Secret: &v1.SecretVolumeSource{
SecretName: "k3k-server-config",
Items: []v1.KeyToPath{
{
Key: "config.yaml",
Path: "config.yaml",
},
},
},
},
},
{
Name: "run",
VolumeSource: v1.VolumeSource{
EmptyDir: &v1.EmptyDirVolumeSource{},
},
},
{
Name: "varrun",
VolumeSource: v1.VolumeSource{
EmptyDir: &v1.EmptyDirVolumeSource{},
},
},
{
Name: "varlibcni",
VolumeSource: v1.VolumeSource{
EmptyDir: &v1.EmptyDirVolumeSource{},
},
},
{
Name: "varlog",
VolumeSource: v1.VolumeSource{
EmptyDir: &v1.EmptyDirVolumeSource{},
},
},
},
Containers: []v1.Container{
{
Name: name,
Image: image,
Env: []v1.EnvVar{
{
Name: "POD_NAME",
ValueFrom: &v1.EnvVarSource{
FieldRef: &v1.ObjectFieldSelector{
FieldPath: "metadata.name",
},
},
},
},
SecurityContext: &v1.SecurityContext{
Privileged: pointer.Bool(true),
},
Command: []string{
"/bin/sh",
"-c",
`if [ ${POD_NAME: -1} == 0 ]; then
/bin/k3s server --config /opt/rancher/k3s/init/config.yaml ` + strings.Join(s.cluster.Spec.ServerArgs, " ") + `
else /bin/k3s server --config /opt/rancher/k3s/server/config.yaml ` + strings.Join(s.cluster.Spec.ServerArgs, " ") + `
fi
`,
},
VolumeMounts: []v1.VolumeMount{
{
Name: "config",
MountPath: "/opt/rancher/k3s/server",
ReadOnly: false,
},
{
Name: "initconfig",
MountPath: "/opt/rancher/k3s/init",
ReadOnly: false,
},
{
Name: "run",
MountPath: "/run",
ReadOnly: false,
},
{
Name: "varrun",
MountPath: "/var/run",
ReadOnly: false,
},
{
Name: "varlibcni",
MountPath: "/var/lib/cni",
ReadOnly: false,
},
{
Name: "varlibkubelet",
MountPath: "/var/lib/kubelet",
ReadOnly: false,
},
{
Name: "varlibrancherk3s",
MountPath: "/var/lib/rancher/k3s",
ReadOnly: false,
},
{
Name: "varlog",
MountPath: "/var/log",
ReadOnly: false,
},
},
},
},
}
replicas = *s.cluster.Spec.Servers - 1
if init {
replicas = 1
if !persistent {
podSpec.Volumes = append(podSpec.Volumes, v1.Volume{
Name: "varlibkubelet",
VolumeSource: v1.VolumeSource{
EmptyDir: &v1.EmptyDirVolumeSource{},
},
}, v1.Volume{
Name: "varlibrancherk3s",
VolumeSource: v1.VolumeSource{
EmptyDir: &v1.EmptyDirVolumeSource{},
},
},
)
}
// Adding readiness probes to deployment
podSpec.Containers[0].ReadinessProbe = &v1.Probe{
InitialDelaySeconds: 60,
FailureThreshold: 5,
TimeoutSeconds: 10,
ProbeHandler: v1.ProbeHandler{
TCPSocket: &v1.TCPSocketAction{
Port: intstr.FromInt(6443),
Host: "127.0.0.1",
},
},
}
return podSpec
}
func (s *Server) StatefulServer(ctx context.Context, cluster *v1alpha1.Cluster) (*apps.StatefulSet, error) {
var (
replicas int32
pvClaims []v1.PersistentVolumeClaim
persistent bool
)
image := util.K3SImage(cluster)
name := serverName + "server"
replicas = *cluster.Spec.Servers
if cluster.Spec.Persistence.Type != EphermalNodesType {
persistent = true
pvClaims = []v1.PersistentVolumeClaim{
{
TypeMeta: metav1.TypeMeta{
Kind: "PersistentVolumeClaim",
APIVersion: "v1",
},
ObjectMeta: metav1.ObjectMeta{
Name: "varlibrancherk3s",
Namespace: util.ClusterNamespace(cluster),
},
Spec: v1.PersistentVolumeClaimSpec{
AccessModes: []v1.PersistentVolumeAccessMode{v1.ReadWriteOnce},
StorageClassName: &cluster.Spec.Persistence.StorageClassName,
Resources: v1.ResourceRequirements{
Requests: v1.ResourceList{
"storage": resource.MustParse(cluster.Spec.Persistence.StorageRequestSize),
},
},
},
},
{
TypeMeta: metav1.TypeMeta{
Kind: "PersistentVolumeClaim",
APIVersion: "v1",
},
ObjectMeta: metav1.ObjectMeta{
Name: "varlibkubelet",
Namespace: util.ClusterNamespace(cluster),
},
Spec: v1.PersistentVolumeClaimSpec{
Resources: v1.ResourceRequirements{
Requests: v1.ResourceList{
"storage": resource.MustParse(cluster.Spec.Persistence.StorageRequestSize),
},
},
AccessModes: []v1.PersistentVolumeAccessMode{v1.ReadWriteOnce},
StorageClassName: &cluster.Spec.Persistence.StorageClassName,
},
},
}
}
var volumes []v1.Volume
@@ -109,202 +314,13 @@ func (s *Server) Deploy(ctx context.Context, init bool) (*apps.Deployment, error
volumeMounts = append(volumeMounts, volumeMount)
}
podSpec := s.podSpec(ctx, image, name, false, init)
podSpec := s.podSpec(ctx, image, name, persistent)
podSpec.Volumes = append(podSpec.Volumes, volumes...)
podSpec.Containers[0].VolumeMounts = append(podSpec.Containers[0].VolumeMounts, volumeMounts...)
return &apps.Deployment{
TypeMeta: metav1.TypeMeta{
Kind: "Deployment",
APIVersion: "apps/v1",
},
ObjectMeta: metav1.ObjectMeta{
Name: s.cluster.Name + "-" + name,
Namespace: util.ClusterNamespace(s.cluster),
},
Spec: apps.DeploymentSpec{
Replicas: &replicas,
Selector: &metav1.LabelSelector{
MatchLabels: map[string]string{
"cluster": s.cluster.Name,
"role": "server",
"init": strconv.FormatBool(init),
},
},
Template: v1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{
Labels: map[string]string{
"cluster": s.cluster.Name,
"role": "server",
"init": strconv.FormatBool(init),
},
},
Spec: podSpec,
},
},
}, nil
}
func (s *Server) podSpec(ctx context.Context, image, name string, statefulSet, init bool) v1.PodSpec {
args := append([]string{"server", "--config", "/opt/rancher/k3s/config.yaml"}, s.cluster.Spec.ServerArgs...)
podSpec := v1.PodSpec{
Volumes: []v1.Volume{
{
Name: "config",
VolumeSource: v1.VolumeSource{
Secret: &v1.SecretVolumeSource{
SecretName: name + "-config",
Items: []v1.KeyToPath{
{
Key: "config.yaml",
Path: "config.yaml",
},
},
},
},
},
{
Name: "run",
VolumeSource: v1.VolumeSource{
EmptyDir: &v1.EmptyDirVolumeSource{},
},
},
{
Name: "varrun",
VolumeSource: v1.VolumeSource{
EmptyDir: &v1.EmptyDirVolumeSource{},
},
},
{
Name: "varlibcni",
VolumeSource: v1.VolumeSource{
EmptyDir: &v1.EmptyDirVolumeSource{},
},
},
{
Name: "varlog",
VolumeSource: v1.VolumeSource{
EmptyDir: &v1.EmptyDirVolumeSource{},
},
},
},
Containers: []v1.Container{
{
Name: name,
Image: image,
SecurityContext: &v1.SecurityContext{
Privileged: pointer.Bool(true),
},
Command: []string{
"/bin/k3s",
},
Args: args,
VolumeMounts: []v1.VolumeMount{
{
Name: "config",
MountPath: "/opt/rancher/k3s/",
ReadOnly: false,
},
{
Name: "run",
MountPath: "/run",
ReadOnly: false,
},
{
Name: "varrun",
MountPath: "/var/run",
ReadOnly: false,
},
{
Name: "varlibcni",
MountPath: "/var/lib/cni",
ReadOnly: false,
},
{
Name: "varlibkubelet",
MountPath: "/var/lib/kubelet",
ReadOnly: false,
},
{
Name: "varlibrancherk3s",
MountPath: "/var/lib/rancher/k3s",
ReadOnly: false,
},
{
Name: "varlog",
MountPath: "/var/log",
ReadOnly: false,
},
},
},
},
}
if !statefulSet {
podSpec.Volumes = append(podSpec.Volumes, v1.Volume{
Name: "varlibkubelet",
VolumeSource: v1.VolumeSource{
EmptyDir: &v1.EmptyDirVolumeSource{},
},
}, v1.Volume{
Name: "varlibrancherk3s",
VolumeSource: v1.VolumeSource{
EmptyDir: &v1.EmptyDirVolumeSource{},
},
},
)
}
// Adding readiness probes to deployment
podSpec.Containers[0].ReadinessProbe = &v1.Probe{
InitialDelaySeconds: 60,
FailureThreshold: 5,
TimeoutSeconds: 10,
ProbeHandler: v1.ProbeHandler{
TCPSocket: &v1.TCPSocketAction{
Port: intstr.FromInt(6443),
Host: "127.0.0.1",
},
},
}
if !init {
podSpec.InitContainers = []v1.Container{
{
Name: initContainerName,
Image: initContainerImage,
Command: []string{
"sh",
"-c",
"until curl -qk https://k3k-server-service.$(cat /var/run/secrets/kubernetes.io/serviceaccount/namespace).svc.cluster.local:6443/v1-k3s/readyz; do echo waiting for init server to be up; sleep 2; done",
},
},
}
}
return podSpec
}
func (s *Server) StatefulServer(ctx context.Context, cluster *v1alpha1.Cluster, init bool) *apps.StatefulSet {
var replicas int32
image := util.K3SImage(cluster)
name := serverName
if init {
name = initServerName
}
replicas = *cluster.Spec.Servers - 1
if init {
replicas = 1
}
return &apps.StatefulSet{
TypeMeta: metav1.TypeMeta{
Kind: "Deployment",
Kind: "StatefulSet",
APIVersion: "apps/v1",
},
ObjectMeta: metav1.ObjectMeta{
@@ -318,59 +334,18 @@ func (s *Server) StatefulServer(ctx context.Context, cluster *v1alpha1.Cluster,
MatchLabels: map[string]string{
"cluster": cluster.Name,
"role": "server",
"init": strconv.FormatBool(init),
},
},
VolumeClaimTemplates: []v1.PersistentVolumeClaim{
{
TypeMeta: metav1.TypeMeta{
Kind: "PersistentVolumeClaim",
APIVersion: "v1",
},
ObjectMeta: metav1.ObjectMeta{
Name: "varlibrancherk3s",
Namespace: util.ClusterNamespace(cluster),
},
Spec: v1.PersistentVolumeClaimSpec{
AccessModes: []v1.PersistentVolumeAccessMode{v1.ReadWriteOnce},
StorageClassName: &cluster.Spec.Persistence.StorageClassName,
Resources: v1.ResourceRequirements{
Requests: v1.ResourceList{
"storage": resource.MustParse(cluster.Spec.Persistence.StorageRequestSize),
},
},
},
},
{
TypeMeta: metav1.TypeMeta{
Kind: "PersistentVolumeClaim",
APIVersion: "v1",
},
ObjectMeta: metav1.ObjectMeta{
Name: "varlibkubelet",
Namespace: util.ClusterNamespace(cluster),
},
Spec: v1.PersistentVolumeClaimSpec{
Resources: v1.ResourceRequirements{
Requests: v1.ResourceList{
"storage": resource.MustParse(cluster.Spec.Persistence.StorageRequestSize),
},
},
AccessModes: []v1.PersistentVolumeAccessMode{v1.ReadWriteOnce},
StorageClassName: &cluster.Spec.Persistence.StorageClassName,
},
},
},
VolumeClaimTemplates: pvClaims,
Template: v1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{
Labels: map[string]string{
"cluster": cluster.Name,
"role": "server",
"init": strconv.FormatBool(init),
},
},
Spec: s.podSpec(ctx, image, name, true, init),
Spec: podSpec,
},
},
}
}, nil
}
pkg/controller/cluster/server/service.go
+1 -7
View File
@@ -1,8 +1,6 @@
package server
import (
"strconv"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
"github.com/rancher/k3k/pkg/controller/util"
v1 "k8s.io/api/core/v1"
@@ -45,11 +43,8 @@ func (s *Server) Service(cluster *v1alpha1.Cluster) *v1.Service {
}
}
func (s *Server) StatefulServerService(cluster *v1alpha1.Cluster, init bool) *v1.Service {
func (s *Server) StatefulServerService(cluster *v1alpha1.Cluster) *v1.Service {
name := serverName
if init {
name = initServerName
}
return &v1.Service{
TypeMeta: metav1.TypeMeta{
Kind: "Service",
@@ -65,7 +60,6 @@ func (s *Server) StatefulServerService(cluster *v1alpha1.Cluster, init bool) *v1
Selector: map[string]string{
"cluster": cluster.Name,
"role": "server",
"init": strconv.FormatBool(init),
},
Ports: []v1.ServicePort{
{