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

Changing the cluster to be namespaced (#110)

Browse Source 
* Changing the cluster to be namespaced

Changes the cluster type to be namespaced (and changes the various
controllers to work with this new feature). Also adds crd generation and
docs to the core cluster type.

* CI fix
This commit is contained in:
Michael Bolot
2024-09-05 22:50:11 +03:00
committed by GitHub
parent bf72d39280
commit 9faab4f82d
20 changed files with 605 additions and 411 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Dockerfile.dapper
+6 -2
View File
@@ -6,16 +6,20 @@ ENV ARCH $DAPPER_HOST_ARCH
RUN apk -U add \bash git gcc musl-dev docker vim less file curl wget ca-certificates
RUN if [ "${ARCH}" == "amd64" ]; then \
curl -sL https://install.goreleaser.com/github.com/golangci/golangci-lint.sh | sh -s v1.15.0; \
curl -sL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s v1.59.0; \
fi
RUN curl -sL https://github.com/helm/chart-releaser/releases/download/v1.5.0/chart-releaser_1.5.0_linux_${ARCH}.tar.gz | tar -xz cr \
&& mv cr /bin/
# Tool for CRD generation.
ENV CONTROLLER_GEN_VERSION v0.14.0
RUN go install sigs.k8s.io/controller-tools/cmd/controller-gen@${CONTROLLER_GEN_VERSION}
ENV GO111MODULE on
ENV DAPPER_ENV REPO TAG DRONE_TAG CROSS GITHUB_TOKEN
ENV DAPPER_SOURCE /go/src/github.com/rancher/k3k/
ENV DAPPER_OUTPUT ./bin ./dist ./deploy
ENV DAPPER_OUTPUT ./bin ./dist ./deploy ./charts
ENV DAPPER_DOCKER_SOCKET true
ENV HOME ${DAPPER_SOURCE}
WORKDIR ${DAPPER_SOURCE}
charts/k3k/crds/cluster.yaml
-134
View File
@@ -1,134 +0,0 @@
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
name: clusters.k3k.io
spec:
group: k3k.io
versions:
- name: v1alpha1
served: true
storage: true
schema:
openAPIV3Schema:
type: object
properties:
spec:
type: object
properties:
name:
type: string
version:
type: string
servers:
type: integer
x-kubernetes-validations:
- message: cluster must have at least one server
rule: self >= 1
agents:
type: integer
x-kubernetes-validations:
- message: invalid value for agents
rule: self >= 0
token:
type: string
x-kubernetes-validations:
- message: token is immutable
rule: self == oldSelf
clusterCIDR:
type: string
x-kubernetes-validations:
- message: clusterCIDR is immutable
rule: self == oldSelf
serviceCIDR:
type: string
x-kubernetes-validations:
- message: serviceCIDR is immutable
rule: self == oldSelf
clusterDNS:
type: string
x-kubernetes-validations:
- message: clusterDNS is immutable
rule: self == oldSelf
serverArgs:
type: array
items:
type: string
agentArgs:
type: array
items:
type: string
tlsSANs:
type: array
items:
type: string
persistence:
type: object
properties:
type:
type: string
default: "ephermal"
storageClassName:
type: string
storageRequestSize:
type: string
addons:
type: array
items:
type: object
properties:
secretNamespace:
type: string
secretRef:
type: string
expose:
type: object
properties:
ingress:
type: object
properties:
enabled:
type: boolean
ingressClassName:
type: string
loadbalancer:
type: object
properties:
enabled:
type: boolean
nodePort:
type: object
properties:
enabled:
type: boolean
status:
type: object
properties:
overrideClusterCIDR:
type: boolean
clusterCIDR:
type: string
overrideServiceCIDR:
type: boolean
serviceCIDR:
type: string
clusterDNS:
type: string
tlsSANs:
type: array
items:
type: string
persistence:
type: object
properties:
type:
type: string
default: "ephermal"
storageClassName:
type: string
storageRequestSize:
type: string
scope: Cluster
names:
plural: clusters
singular: cluster
kind: Cluster
charts/k3k/crds/k3k.io_clusters.yaml
+206
View File
@@ -0,0 +1,206 @@
---
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
annotations:
controller-gen.kubebuilder.io/version: v0.14.0
name: clusters.k3k.io
spec:
group: k3k.io
names:
kind: Cluster
listKind: ClusterList
plural: clusters
singular: cluster
scope: Namespaced
versions:
- name: v1alpha1
schema:
openAPIV3Schema:
properties:
apiVersion:
description: |-
APIVersion defines the versioned schema of this representation of an object.
Servers should convert recognized schemas to the latest internal value, and
may reject unrecognized values.
More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources
type: string
kind:
description: |-
Kind is a string value representing the REST resource this object represents.
Servers may infer this from the endpoint the client submits requests to.
Cannot be updated.
In CamelCase.
More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds
type: string
metadata:
type: object
spec:
properties:
addons:
description: Addons is a list of secrets containing raw YAML which
will be deployed in the virtual K3k cluster on startup.
items:
properties:
secretNamespace:
type: string
secretRef:
type: string
type: object
type: array
agentArgs:
description: AgentArgs are the ordered key value pairs (e.x. "testArg",
"testValue") for the K3s pods running in agent mode.
items:
type: string
type: array
agents:
description: Agents is the number of K3s pods to run in agent (worker)
mode.
format: int32
type: integer
x-kubernetes-validations:
- message: invalid value for agents
rule: self >= 0
clusterCIDR:
description: ClusterCIDR is the CIDR range for the pods of the cluster.
Defaults to 10.42.0.0/16.
type: string
x-kubernetes-validations:
- message: clusterCIDR is immutable
rule: self == oldSelf
clusterDNS:
description: |-
ClusterDNS is the IP address for the coredns service. Needs to be in the range provided by ServiceCIDR or CoreDNS may not deploy.
Defaults to 10.43.0.10.
type: string
x-kubernetes-validations:
- message: clusterDNS is immutable
rule: self == oldSelf
expose:
description: |-
Expose contains options for exposing the apiserver inside/outside of the cluster. By default, this is only exposed as a
clusterIP which is relatively secure, but difficult to access outside of the cluster.
properties:
ingress:
properties:
enabled:
type: boolean
ingressClassName:
type: string
required:
- enabled
- ingressClassName
type: object
loadbalancer:
properties:
enabled:
type: boolean
required:
- enabled
type: object
nodePort:
properties:
enabled:
type: boolean
required:
- enabled
type: object
required:
- ingress
- loadbalancer
- nodePort
type: object
persistence:
description: |-
Persistence contains options controlling how the etcd data of the virtual cluster is persisted. By default, no data
persistence is guaranteed, so restart of a virtual cluster pod may result in data loss without this field.
properties:
storageClassName:
type: string
storageRequestSize:
type: string
type:
default: ephemeral
description: Type can be ephermal, static, dynamic
type: string
required:
- type
type: object
serverArgs:
description: ServerArgs are the ordered key value pairs (e.x. "testArg",
"testValue") for the K3s pods running in server mode.
items:
type: string
type: array
servers:
description: Servers is the number of K3s pods to run in server (controlplane)
mode.
format: int32
type: integer
x-kubernetes-validations:
- message: cluster must have at least one server
rule: self >= 1
serviceCIDR:
description: ServiceCIDR is the CIDR range for the services in the
cluster. Defaults to 10.43.0.0/16.
type: string
x-kubernetes-validations:
- message: serviceCIDR is immutable
rule: self == oldSelf
tlsSANs:
description: TLSSANs are the subjectAlternativeNames for the certificate
the K3s server will use.
items:
type: string
type: array
token:
description: Token is the token used to join the worker nodes to the
cluster.
type: string
x-kubernetes-validations:
- message: token is immutable
rule: self == oldSelf
version:
description: Version is a string representing the Kubernetes version
to be used by the virtual nodes.
type: string
required:
- agents
- servers
- token
- version
type: object
status:
properties:
clusterCIDR:
type: string
clusterDNS:
type: string
persistence:
properties:
storageClassName:
type: string
storageRequestSize:
type: string
type:
default: ephemeral
description: Type can be ephermal, static, dynamic
type: string
required:
- type
type: object
serviceCIDR:
type: string
tlsSANs:
items:
type: string
type: array
type: object
required:
- spec
type: object
served: true
storage: true
subresources:
status: {}
cli/cmds/cluster/create.go
+4 -2
View File
@@ -139,6 +139,7 @@ func create(clx *cli.Context) error {
logrus.Infof("Creating a new cluster [%s]", name)
cluster := newCluster(
name,
cmds.Namespace(),
token,
int32(servers),
int32(agents),
@@ -226,10 +227,11 @@ func validateCreateFlags(clx *cli.Context) error {
return nil
}
func newCluster(name, token string, servers, agents int32, clusterCIDR, serviceCIDR string, serverArgs, agentArgs []string) *v1alpha1.Cluster {
func newCluster(name, namespace, token string, servers, agents int32, clusterCIDR, serviceCIDR string, serverArgs, agentArgs []string) *v1alpha1.Cluster {
return &v1alpha1.Cluster{
ObjectMeta: metav1.ObjectMeta{
Name: name,
Name: name,
Namespace: namespace,
},
TypeMeta: metav1.TypeMeta{
Kind: "Cluster",
cli/cmds/cluster/delete.go
+2 -1
View File
@@ -40,7 +40,8 @@ func delete(clx *cli.Context) error {
logrus.Infof("deleting [%s] cluster", name)
cluster := v1alpha1.Cluster{
ObjectMeta: metav1.ObjectMeta{
Name: name,
Name: name,
Namespace: cmds.Namespace(),
},
}
return ctrlClient.Delete(ctx, &cluster)
cli/cmds/kubeconfig/kubeconfig.go
+2 -2
View File
@@ -114,9 +114,9 @@ func generate(clx *cli.Context) error {
if err != nil {
return err
}
clusterKey := types.NamespacedName{
Name: name,
Name: name,
Namespace: cmds.Namespace(),
}
if err := ctrlClient.Get(ctx, clusterKey, &cluster); err != nil {
cli/cmds/root.go
+17
View File
@@ -5,9 +5,14 @@ import (
"github.com/urfave/cli"
)
const (
defaultNamespace = "default"
)
var (
debug bool
Kubeconfig string
namespace string
CommonFlags = []cli.Flag{
cli.StringFlag{
Name: "kubeconfig",
@@ -15,6 +20,11 @@ var (
Usage: "Kubeconfig path",
Destination: &Kubeconfig,
},
cli.StringFlag{
Name: "namespace",
Usage: "Namespace to create the k3k cluster in",
Destination: &namespace,
},
}
)
@@ -40,3 +50,10 @@ func NewApp() *cli.App {
return app
}
func Namespace() string {
if namespace == "" {
return defaultNamespace
}
return namespace
}
main.go
+4
View File
@@ -46,6 +46,10 @@ func main() {
klog.Fatalf("Failed to add the new controller: %v", err)
}
if err := cluster.AddPodController(ctx, mgr); err != nil {
klog.Fatalf("Failed to add the new cluster controller: %v", err)
}
if err := mgr.Start(ctx); err != nil {
klog.Fatalf("Failed to start the manager: %v", err)
}
ops/build-crds
Executable
+8
View File
@@ -0,0 +1,8 @@
#! /bin/sh
cd $(dirname $0)/../
# This will return non-zero until all of our objects in ./pkg/apis can generate valid crds.
# allowDangerousTypes is needed for struct that use floats
controller-gen crd:generateEmbeddedObjectMeta=true,allowDangerousTypes=false paths=./pkg/apis/... output:crd:dir=./charts/k3k/crds
pkg/apis/k3k.io/v1alpha1/types.go
+46 -13
View File
@@ -1,35 +1,67 @@
package v1alpha1
import (
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
// +genclient
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
// +kubebuilder:storageversion
// +kubebuilder:subresource:status
type Cluster struct {
metav1.ObjectMeta `json:"metadata,omitempty"`
metav1.TypeMeta `json:",inline"`
Spec ClusterSpec `json:"spec"`
Status ClusterStatus `json:"status"`
Status ClusterStatus `json:"status,omitempty"`
}
type ClusterSpec struct {
Version string `json:"version"`
Servers *int32 `json:"servers"`
Agents *int32 `json:"agents"`
Token string `json:"token"`
ClusterCIDR string `json:"clusterCIDR,omitempty"`
ServiceCIDR string `json:"serviceCIDR,omitempty"`
ClusterDNS string `json:"clusterDNS,omitempty"`
ServerArgs []string `json:"serverArgs,omitempty"`
AgentArgs []string `json:"agentArgs,omitempty"`
TLSSANs []string `json:"tlsSANs,omitempty"`
Addons []Addon `json:"addons,omitempty"`
// Version is a string representing the Kubernetes version to be used by the virtual nodes.
Version string `json:"version"`
// +kubebuilder:validation:XValidation:message="cluster must have at least one server",rule="self >= 1"
// Servers is the number of K3s pods to run in server (controlplane) mode.
Servers *int32 `json:"servers"`
// +kubebuilder:validation:XValidation:message="invalid value for agents",rule="self >= 0"
// Agents is the number of K3s pods to run in agent (worker) mode.
Agents *int32 `json:"agents"`
// +kubebuilder:validation:XValidation:message="token is immutable",rule="self == oldSelf"
// Token is the token used to join the worker nodes to the cluster.
Token string `json:"token"`
// +kubebuilder:validation:XValidation:message="clusterCIDR is immutable",rule="self == oldSelf"
// ClusterCIDR is the CIDR range for the pods of the cluster. Defaults to 10.42.0.0/16.
ClusterCIDR string `json:"clusterCIDR,omitempty"`
// +kubebuilder:validation:XValidation:message="serviceCIDR is immutable",rule="self == oldSelf"
// ServiceCIDR is the CIDR range for the services in the cluster. Defaults to 10.43.0.0/16.
ServiceCIDR string `json:"serviceCIDR,omitempty"`
// +kubebuilder:validation:XValidation:message="clusterDNS is immutable",rule="self == oldSelf"
// ClusterDNS is the IP address for the coredns service. Needs to be in the range provided by ServiceCIDR or CoreDNS may not deploy.
// Defaults to 10.43.0.10.
ClusterDNS string `json:"clusterDNS,omitempty"`
// ServerArgs are the ordered key value pairs (e.x. "testArg", "testValue") for the K3s pods running in server mode.
ServerArgs []string `json:"serverArgs,omitempty"`
// AgentArgs are the ordered key value pairs (e.x. "testArg", "testValue") for the K3s pods running in agent mode.
AgentArgs []string `json:"agentArgs,omitempty"`
// TLSSANs are the subjectAlternativeNames for the certificate the K3s server will use.
TLSSANs []string `json:"tlsSANs,omitempty"`
// Addons is a list of secrets containing raw YAML which will be deployed in the virtual K3k cluster on startup.
Addons []Addon `json:"addons,omitempty"`
// Persistence contains options controlling how the etcd data of the virtual cluster is persisted. By default, no data
// persistence is guaranteed, so restart of a virtual cluster pod may result in data loss without this field.
Persistence *PersistenceConfig `json:"persistence,omitempty"`
Expose *ExposeConfig `json:"expose,omitempty"`
// Expose contains options for exposing the apiserver inside/outside of the cluster. By default, this is only exposed as a
// clusterIP which is relatively secure, but difficult to access outside of the cluster.
Expose *ExposeConfig `json:"expose,omitempty"`
}
type ClusterLimit struct {
// ServerLimit is the limits (cpu/mem) that apply to the server nodes
ServerLimit v1.ResourceList `json:"serverLimit,omitempty"`
// WorkerLimit is the limits (cpu/mem) that apply to the agent nodes
WorkerLimit v1.ResourceList `json:"workerLimit,omitempty"`
}
type Addon struct {
@@ -48,6 +80,7 @@ type ClusterList struct {
type PersistenceConfig struct {
// Type can be ephermal, static, dynamic
// +kubebuilder:default="ephemeral"
Type string `json:"type"`
StorageClassName string `json:"storageClassName,omitempty"`
StorageRequestSize string `json:"storageRequestSize,omitempty"`
pkg/controller/cluster/agent/agent.go
+35 -24
View File
@@ -26,7 +26,12 @@ func (a *Agent) Deploy() *apps.Deployment {
image := util.K3SImage(a.cluster)
const name = "k3k-agent"
selector := metav1.LabelSelector{
MatchLabels: map[string]string{
"cluster": a.cluster.Name,
"type": "agent",
},
}
return &apps.Deployment{
TypeMeta: metav1.TypeMeta{
Kind: "Deployment",
@@ -35,23 +40,16 @@ func (a *Agent) Deploy() *apps.Deployment {
ObjectMeta: metav1.ObjectMeta{
Name: a.cluster.Name + "-" + name,
Namespace: util.ClusterNamespace(a.cluster),
Labels: selector.MatchLabels,
},
Spec: apps.DeploymentSpec{
Replicas: a.cluster.Spec.Agents,
Selector: &metav1.LabelSelector{
MatchLabels: map[string]string{
"cluster": a.cluster.Name,
"type": "agent",
},
},
Selector: &selector,
Template: v1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{
Labels: map[string]string{
"cluster": a.cluster.Name,
"type": "agent",
},
Labels: selector.MatchLabels,
},
Spec: a.podSpec(image, name, a.cluster.Spec.AgentArgs, false),
Spec: a.podSpec(image, name, a.cluster.Spec.AgentArgs, false, &selector),
},
},
}
@@ -60,6 +58,12 @@ func (a *Agent) Deploy() *apps.Deployment {
func (a *Agent) StatefulAgent(cluster *v1alpha1.Cluster) *apps.StatefulSet {
image := util.K3SImage(cluster)
selector := metav1.LabelSelector{
MatchLabels: map[string]string{
"cluster": cluster.Name,
"type": "agent",
},
}
return &apps.StatefulSet{
TypeMeta: metav1.TypeMeta{
Kind: "Statefulset",
@@ -68,16 +72,12 @@ func (a *Agent) StatefulAgent(cluster *v1alpha1.Cluster) *apps.StatefulSet {
ObjectMeta: metav1.ObjectMeta{
Name: cluster.Name + "-" + agentName,
Namespace: util.ClusterNamespace(cluster),
Labels: selector.MatchLabels,
},
Spec: apps.StatefulSetSpec{
ServiceName: cluster.Name + "-" + agentName + "-headless",
Replicas: cluster.Spec.Agents,
Selector: &metav1.LabelSelector{
MatchLabels: map[string]string{
"cluster": cluster.Name,
"type": "agent",
},
},
Selector: &selector,
VolumeClaimTemplates: []v1.PersistentVolumeClaim{
{
TypeMeta: metav1.TypeMeta{
@@ -120,20 +120,28 @@ func (a *Agent) StatefulAgent(cluster *v1alpha1.Cluster) *apps.StatefulSet {
},
Template: v1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{
Labels: map[string]string{
"cluster": cluster.Name,
"type": "agent",
},
Labels: selector.MatchLabels,
},
Spec: a.podSpec(image, agentName, cluster.Spec.AgentArgs, true),
Spec: a.podSpec(image, agentName, cluster.Spec.AgentArgs, true, &selector),
},
},
}
}
func (a *Agent) podSpec(image, name string, args []string, statefulSet bool) v1.PodSpec {
func (a *Agent) podSpec(image, name string, args []string, statefulSet bool, affinitySelector *metav1.LabelSelector) v1.PodSpec {
var limit v1.ResourceList
args = append([]string{"agent", "--config", "/opt/rancher/k3s/config.yaml"}, args...)
podSpec := v1.PodSpec{
Affinity: &v1.Affinity{
PodAntiAffinity: &v1.PodAntiAffinity{
RequiredDuringSchedulingIgnoredDuringExecution: []v1.PodAffinityTerm{
{
LabelSelector: affinitySelector,
TopologyKey: "kubernetes.io/hostname",
},
},
},
},
Volumes: []v1.Volume{
{
Name: "config",
@@ -185,6 +193,9 @@ func (a *Agent) podSpec(image, name string, args []string, statefulSet bool) v1.
"/bin/k3s",
},
Args: args,
Resources: v1.ResourceRequirements{
Limits: limit,
},
VolumeMounts: []v1.VolumeMount{
{
Name: "config",
pkg/controller/cluster/cluster.go
+5 -208
View File
@@ -2,33 +2,22 @@ package cluster
import (
"context"
"crypto/tls"
"crypto/x509"
"errors"
"fmt"
"net/url"
"reflect"
"strings"
"time"
certutil "github.com/rancher/dynamiclistener/cert"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
"github.com/rancher/k3k/pkg/controller/cluster/agent"
"github.com/rancher/k3k/pkg/controller/cluster/config"
"github.com/rancher/k3k/pkg/controller/cluster/server"
"github.com/rancher/k3k/pkg/controller/cluster/server/bootstrap"
"github.com/rancher/k3k/pkg/controller/kubeconfig"
"github.com/rancher/k3k/pkg/controller/util"
"github.com/sirupsen/logrus"
"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
clientv3 "go.etcd.io/etcd/client/v3"
apps "k8s.io/api/apps/v1"
v1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/util/retry"
"k8s.io/klog"
ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/controller"
@@ -76,49 +65,15 @@ func Add(ctx context.Context, mgr manager.Manager) error {
return err
}
if err := controller.Watch(&source.Kind{Type: &v1alpha1.Cluster{}}, &handler.EnqueueRequestForObject{}); err != nil {
return err
}
return controller.Watch(&source.Kind{Type: &v1.Pod{}},
&handler.EnqueueRequestForOwner{IsController: true, OwnerType: &apps.StatefulSet{}})
return controller.Watch(&source.Kind{Type: &v1alpha1.Cluster{}}, &handler.EnqueueRequestForObject{})
}
func (c *ClusterReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
var (
cluster v1alpha1.Cluster
podList v1.PodList
clusterName string
cluster v1alpha1.Cluster
podList v1.PodList
)
if req.Namespace != "" {
s := strings.Split(req.Namespace, "-")
if len(s) <= 1 {
return reconcile.Result{}, util.LogAndReturnErr("failed to get cluster namespace", nil)
}
clusterName = s[1]
var cluster v1alpha1.Cluster
if err := c.Client.Get(ctx, types.NamespacedName{Name: clusterName}, &cluster); err != nil {
if !apierrors.IsNotFound(err) {
return reconcile.Result{}, err
}
}
matchingLabels := ctrlruntimeclient.MatchingLabels(map[string]string{"role": "server"})
listOpts := &ctrlruntimeclient.ListOptions{Namespace: req.Namespace}
matchingLabels.ApplyToList(listOpts)
if err := c.Client.List(ctx, &podList, listOpts); err != nil {
return reconcile.Result{}, ctrlruntimeclient.IgnoreNotFound(err)
}
for _, pod := range podList.Items {
klog.Infof("Handle etcd server pod [%s/%s]", pod.Namespace, pod.Name)
if err := c.handleServerPod(ctx, cluster, &pod); err != nil {
return reconcile.Result{}, util.LogAndReturnErr("failed to handle etcd pod", err)
}
}
return reconcile.Result{}, nil
}
if err := c.Client.Get(ctx, req.NamespacedName, &cluster); err != nil {
return reconcile.Result{}, ctrlruntimeclient.IgnoreNotFound(err)
@@ -132,17 +87,6 @@ func (c *ClusterReconciler) Reconcile(ctx context.Context, req reconcile.Request
}
}
// we create a namespace for each new cluster
var ns v1.Namespace
objKey := ctrlruntimeclient.ObjectKey{
Name: util.ClusterNamespace(&cluster),
}
if err := c.Client.Get(ctx, objKey, &ns); err != nil {
if !apierrors.IsNotFound(err) {
return reconcile.Result{}, util.LogAndReturnErr("failed to get cluster namespace "+util.ClusterNamespace(&cluster), err)
}
}
klog.Infof("enqueue cluster [%s]", cluster.Name)
if err := c.createCluster(ctx, &cluster); err != nil {
return reconcile.Result{}, util.LogAndReturnErr("failed to create cluster", err)
@@ -196,10 +140,6 @@ func (c *ClusterReconciler) createCluster(ctx context.Context, cluster *v1alpha1
if err := c.Client.Update(ctx, cluster); err != nil {
return util.LogAndReturnErr("failed to update cluster with persistence type", err)
}
// create a new namespace for the cluster
if err := c.createNamespace(ctx, cluster); err != nil {
return util.LogAndReturnErr("failed to create ns", err)
}
cluster.Status.ClusterCIDR = cluster.Spec.ClusterCIDR
if cluster.Status.ClusterCIDR == "" {
@@ -259,26 +199,6 @@ func (c *ClusterReconciler) createCluster(ctx context.Context, cluster *v1alpha1
return c.Client.Update(ctx, cluster)
}
func (c *ClusterReconciler) createNamespace(ctx context.Context, cluster *v1alpha1.Cluster) error {
// create a new namespace for the cluster
namespace := v1.Namespace{
ObjectMeta: metav1.ObjectMeta{
Name: util.ClusterNamespace(cluster),
},
}
if err := controllerutil.SetControllerReference(cluster, &namespace, c.Scheme); err != nil {
return err
}
if err := c.Client.Create(ctx, &namespace); err != nil {
if !apierrors.IsAlreadyExists(err) {
return util.LogAndReturnErr("failed to create ns", err)
}
}
return nil
}
func (c *ClusterReconciler) createClusterConfigs(ctx context.Context, cluster *v1alpha1.Cluster, serviceIP string) error {
// create init node config
initServerConfig, err := config.Server(cluster, true, serviceIP)
@@ -341,7 +261,7 @@ func (c *ClusterReconciler) createClusterService(ctx context.Context, cluster *v
objKey := ctrlruntimeclient.ObjectKey{
Namespace: util.ClusterNamespace(cluster),
Name: "k3k-server-service",
Name: util.ServerSvcName(cluster),
}
if err := c.Client.Get(ctx, objKey, &service); err != nil {
return "", err
@@ -400,7 +320,7 @@ func agentConfig(cluster *v1alpha1.Cluster, serviceIP string) v1.Secret {
APIVersion: "v1",
},
ObjectMeta: metav1.ObjectMeta{
Name: "k3k-agent-config",
Name: util.AgentConfigName(cluster),
Namespace: util.ClusterNamespace(cluster),
},
Data: map[string][]byte{
@@ -414,129 +334,6 @@ func agentData(serviceIP, token string) string {
token: %s`, serviceIP, token)
}
func (c *ClusterReconciler) handleServerPod(ctx context.Context, cluster v1alpha1.Cluster, pod *v1.Pod) error {
if _, ok := pod.Labels["role"]; ok {
if pod.Labels["role"] != "server" {
return nil
}
} else {
return errors.New("server pod has no role label")
}
// if etcd pod is marked for deletion then we need to remove it from the etcd member list before deletion
if !pod.DeletionTimestamp.IsZero() {
// check if cluster is deleted then remove the finalizer from the pod
if cluster.Name == "" {
if controllerutil.ContainsFinalizer(pod, etcdPodFinalizerName) {
controllerutil.RemoveFinalizer(pod, etcdPodFinalizerName)
if err := c.Client.Update(ctx, pod); err != nil {
return err
}
}
return nil
}
tlsConfig, err := c.getETCDTLS(&cluster)
if err != nil {
return err
}
// remove server from etcd
client, err := clientv3.New(clientv3.Config{
Endpoints: []string{
"https://k3k-server-service." + pod.Namespace + ":2379",
},
TLS: tlsConfig,
})
if err != nil {
return err
}
if err := removePeer(ctx, client, pod.Name, pod.Status.PodIP); err != nil {
return err
}
// remove our finalizer from the list and update it.
if controllerutil.ContainsFinalizer(pod, etcdPodFinalizerName) {
controllerutil.RemoveFinalizer(pod, etcdPodFinalizerName)
if err := c.Client.Update(ctx, pod); err != nil {
return err
}
}
}
if !controllerutil.ContainsFinalizer(pod, etcdPodFinalizerName) {
controllerutil.AddFinalizer(pod, etcdPodFinalizerName)
return c.Client.Update(ctx, pod)
}
return nil
}
// removePeer removes a peer from the cluster. The peer name and IP address must both match.
func removePeer(ctx context.Context, client *clientv3.Client, name, address string) error {
ctx, cancel := context.WithTimeout(ctx, memberRemovalTimeout)
defer cancel()
members, err := client.MemberList(ctx)
if err != nil {
return err
}
for _, member := range members.Members {
if !strings.Contains(member.Name, name) {
continue
}
for _, peerURL := range member.PeerURLs {
u, err := url.Parse(peerURL)
if err != nil {
return err
}
if u.Hostname() == address {
logrus.Infof("Removing name=%s id=%d address=%s from etcd", member.Name, member.ID, address)
_, err := client.MemberRemove(ctx, member.ID)
if errors.Is(err, rpctypes.ErrGRPCMemberNotFound) {
return nil
}
return err
}
}
}
return nil
}
func (c *ClusterReconciler) getETCDTLS(cluster *v1alpha1.Cluster) (*tls.Config, error) {
klog.Infof("generating etcd TLS client certificate for cluster [%s]", cluster.Name)
token := cluster.Spec.Token
endpoint := "k3k-server-service." + util.ClusterNamespace(cluster)
var b *bootstrap.ControlRuntimeBootstrap
if err := retry.OnError(retry.DefaultBackoff, func(err error) bool {
return true
}, func() error {
var err error
b, err = bootstrap.DecodedBootstrap(token, endpoint)
return err
}); err != nil {
return nil, err
}
etcdCert, etcdKey, err := kubeconfig.CreateClientCertKey("etcd-client", nil, nil, []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth}, 0, b.ETCDServerCA.Content, b.ETCDServerCAKey.Content)
if err != nil {
return nil, err
}
clientCert, err := tls.X509KeyPair(etcdCert, etcdKey)
if err != nil {
return nil, err
}
// create rootCA CertPool
cert, err := certutil.ParseCertsPEM([]byte(b.ETCDServerCA.Content))
if err != nil {
return nil, err
}
pool := x509.NewCertPool()
pool.AddCert(cert[0])
return &tls.Config{
RootCAs: pool,
Certificates: []tls.Certificate{clientCert},
}, nil
}
func (c *ClusterReconciler) validate(cluster *v1alpha1.Cluster) error {
if cluster.Name == ClusterInvalidName {
return errors.New("invalid cluster name " + cluster.Name + " no action will be taken")
pkg/controller/cluster/config/agent.go
+1 -1
View File
@@ -18,7 +18,7 @@ func Agent(cluster *v1alpha1.Cluster, serviceIP string) v1.Secret {
APIVersion: "v1",
},
ObjectMeta: metav1.ObjectMeta{
Name: "k3k-agent-config",
Name: util.AgentConfigName(cluster),
Namespace: util.ClusterNamespace(cluster),
},
Data: map[string][]byte{
pkg/controller/cluster/config/server.go
+8 -4
View File
@@ -1,22 +1,26 @@
package config
import (
"fmt"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
"github.com/rancher/k3k/pkg/controller/util"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
// Server returns the secret for the server's config. Note that this doesn't set the ownerRef on the secret
// to tie it back to the cluster.
func Server(cluster *v1alpha1.Cluster, init bool, serviceIP string) (*v1.Secret, error) {
name := "k3k-server-config"
name := util.ServerConfigName(cluster)
if init {
name = "k3k-init-server-config"
name = util.ServerInitConfigName(cluster)
}
cluster.Status.TLSSANs = append(cluster.Spec.TLSSANs,
serviceIP,
"k3k-server-service",
"k3k-server-service."+util.ClusterNamespace(cluster),
util.ServerSvcName(cluster),
fmt.Sprintf("%s.%s", util.ServerSvcName(cluster), util.ClusterNamespace(cluster)),
)
config := serverConfigData(serviceIP, cluster)
pkg/controller/cluster/pod.go
+214
View File
@@ -0,0 +1,214 @@
package cluster
import (
"context"
"crypto/tls"
"crypto/x509"
"errors"
"fmt"
"net/url"
"strings"
certutil "github.com/rancher/dynamiclistener/cert"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
"github.com/rancher/k3k/pkg/controller/cluster/server/bootstrap"
"github.com/rancher/k3k/pkg/controller/kubeconfig"
"github.com/rancher/k3k/pkg/controller/util"
"github.com/sirupsen/logrus"
"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
clientv3 "go.etcd.io/etcd/client/v3"
apps "k8s.io/api/apps/v1"
v1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/util/retry"
"k8s.io/klog"
ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/controller"
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
"sigs.k8s.io/controller-runtime/pkg/handler"
"sigs.k8s.io/controller-runtime/pkg/manager"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
"sigs.k8s.io/controller-runtime/pkg/source"
)
type PodReconciler struct {
Client ctrlruntimeclient.Client
Scheme *runtime.Scheme
}
// Add adds a new controller to the manager
func AddPodController(ctx context.Context, mgr manager.Manager) error {
// initialize a new Reconciler
reconciler := PodReconciler{
Client: mgr.GetClient(),
Scheme: mgr.GetScheme(),
}
// create a new controller and add it to the manager
//this can be replaced by the new builder functionality in controller-runtime
controller, err := controller.New(clusterController, mgr, controller.Options{
Reconciler: &reconciler,
MaxConcurrentReconciles: maxConcurrentReconciles,
})
if err != nil {
return err
}
return controller.Watch(&source.Kind{Type: &v1.Pod{}},
&handler.EnqueueRequestForOwner{IsController: true, OwnerType: &apps.StatefulSet{}})
}
func (p *PodReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
s := strings.Split(req.Namespace, "-")
if len(s) <= 1 {
return reconcile.Result{}, util.LogAndReturnErr("failed to get cluster namespace", nil)
}
clusterName := s[1]
var cluster v1alpha1.Cluster
if err := p.Client.Get(ctx, types.NamespacedName{Name: clusterName}, &cluster); err != nil {
if !apierrors.IsNotFound(err) {
return reconcile.Result{}, err
}
}
matchingLabels := ctrlruntimeclient.MatchingLabels(map[string]string{"role": "server"})
listOpts := &ctrlruntimeclient.ListOptions{Namespace: req.Namespace}
matchingLabels.ApplyToList(listOpts)
var podList v1.PodList
if err := p.Client.List(ctx, &podList, listOpts); err != nil {
return reconcile.Result{}, ctrlruntimeclient.IgnoreNotFound(err)
}
for _, pod := range podList.Items {
klog.Infof("Handle etcd server pod [%s/%s]", pod.Namespace, pod.Name)
if err := p.handleServerPod(ctx, cluster, &pod); err != nil {
return reconcile.Result{}, util.LogAndReturnErr("failed to handle etcd pod", err)
}
}
return reconcile.Result{}, nil
}
func (p *PodReconciler) handleServerPod(ctx context.Context, cluster v1alpha1.Cluster, pod *v1.Pod) error {
if _, ok := pod.Labels["role"]; ok {
if pod.Labels["role"] != "server" {
return nil
}
} else {
return fmt.Errorf("server pod has no role label")
}
// if etcd pod is marked for deletion then we need to remove it from the etcd member list before deletion
if !pod.DeletionTimestamp.IsZero() {
// check if cluster is deleted then remove the finalizer from the pod
if cluster.Name == "" {
if controllerutil.ContainsFinalizer(pod, etcdPodFinalizerName) {
controllerutil.RemoveFinalizer(pod, etcdPodFinalizerName)
if err := p.Client.Update(ctx, pod); err != nil {
return err
}
}
return nil
}
tlsConfig, err := p.getETCDTLS(&cluster)
if err != nil {
return err
}
// remove server from etcd
client, err := clientv3.New(clientv3.Config{
Endpoints: []string{
fmt.Sprintf("https://%s.%s:2379", util.ServerSvcName(&cluster), pod.Namespace),
},
TLS: tlsConfig,
})
if err != nil {
return err
}
if err := removePeer(ctx, client, pod.Name, pod.Status.PodIP); err != nil {
return err
}
// remove our finalizer from the list and update it.
if controllerutil.ContainsFinalizer(pod, etcdPodFinalizerName) {
controllerutil.RemoveFinalizer(pod, etcdPodFinalizerName)
if err := p.Client.Update(ctx, pod); err != nil {
return err
}
}
}
if !controllerutil.ContainsFinalizer(pod, etcdPodFinalizerName) {
controllerutil.AddFinalizer(pod, etcdPodFinalizerName)
return p.Client.Update(ctx, pod)
}
return nil
}
func (p *PodReconciler) getETCDTLS(cluster *v1alpha1.Cluster) (*tls.Config, error) {
klog.Infof("generating etcd TLS client certificate for cluster [%s]", cluster.Name)
token := cluster.Spec.Token
endpoint := fmt.Sprintf("%s.%s", util.ServerSvcName(cluster), util.ClusterNamespace(cluster))
var b *bootstrap.ControlRuntimeBootstrap
if err := retry.OnError(retry.DefaultBackoff, func(err error) bool {
return true
}, func() error {
var err error
b, err = bootstrap.DecodedBootstrap(token, endpoint)
return err
}); err != nil {
return nil, err
}
etcdCert, etcdKey, err := kubeconfig.CreateClientCertKey("etcd-client", nil, nil, []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth}, 0, b.ETCDServerCA.Content, b.ETCDServerCAKey.Content)
if err != nil {
return nil, err
}
clientCert, err := tls.X509KeyPair(etcdCert, etcdKey)
if err != nil {
return nil, err
}
// create rootCA CertPool
cert, err := certutil.ParseCertsPEM([]byte(b.ETCDServerCA.Content))
if err != nil {
return nil, err
}
pool := x509.NewCertPool()
pool.AddCert(cert[0])
return &tls.Config{
RootCAs: pool,
Certificates: []tls.Certificate{clientCert},
}, nil
}
// removePeer removes a peer from the cluster. The peer name and IP address must both match.
func removePeer(ctx context.Context, client *clientv3.Client, name, address string) error {
ctx, cancel := context.WithTimeout(ctx, memberRemovalTimeout)
defer cancel()
members, err := client.MemberList(ctx)
if err != nil {
return err
}
for _, member := range members.Members {
if !strings.Contains(member.Name, name) {
continue
}
for _, peerURL := range member.PeerURLs {
u, err := url.Parse(peerURL)
if err != nil {
return err
}
if u.Hostname() == address {
logrus.Infof("Removing name=%s id=%d address=%s from etcd", member.Name, member.ID, address)
_, err := client.MemberRemove(ctx, member.ID)
if errors.Is(err, rpctypes.ErrGRPCMemberNotFound) {
return nil
}
return err
}
}
}
return nil
}
pkg/controller/cluster/server/ingress.go
+1 -1
View File
@@ -59,7 +59,7 @@ func (s *Server) ingressRules(addresses []string) []networkingv1.IngressRule {
PathType: &pathTypePrefix,
Backend: networkingv1.IngressBackend{
Service: &networkingv1.IngressServiceBackend{
Name: "k3k-server-service",
Name: util.ServerSvcName(s.cluster),
Port: networkingv1.ServiceBackendPort{
Number: serverPort,
},
pkg/controller/cluster/server/server.go
+26 -16
View File
@@ -40,14 +40,24 @@ func New(cluster *v1alpha1.Cluster, client client.Client) *Server {
}
}
func (s *Server) podSpec(ctx context.Context, image, name string, persistent bool) v1.PodSpec {
func (s *Server) podSpec(ctx context.Context, image, name string, persistent bool, affinitySelector *metav1.LabelSelector) v1.PodSpec {
podSpec := v1.PodSpec{
Affinity: &v1.Affinity{
PodAntiAffinity: &v1.PodAntiAffinity{
RequiredDuringSchedulingIgnoredDuringExecution: []v1.PodAffinityTerm{
{
LabelSelector: affinitySelector,
TopologyKey: "kubernetes.io/hostname",
},
},
},
},
Volumes: []v1.Volume{
{
Name: "initconfig",
VolumeSource: v1.VolumeSource{
Secret: &v1.SecretVolumeSource{
SecretName: "k3k-init-server-config",
SecretName: util.ServerInitConfigName(s.cluster),
Items: []v1.KeyToPath{
{
Key: "config.yaml",
@@ -61,7 +71,7 @@ func (s *Server) podSpec(ctx context.Context, image, name string, persistent boo
Name: "config",
VolumeSource: v1.VolumeSource{
Secret: &v1.SecretVolumeSource{
SecretName: "k3k-server-config",
SecretName: util.ServerConfigName(s.cluster),
Items: []v1.KeyToPath{
{
Key: "config.yaml",
@@ -313,7 +323,14 @@ func (s *Server) StatefulServer(ctx context.Context, cluster *v1alpha1.Cluster)
volumeMounts = append(volumeMounts, volumeMount)
}
podSpec := s.podSpec(ctx, image, name, persistent)
selector := metav1.LabelSelector{
MatchLabels: map[string]string{
"cluster": cluster.Name,
"role": "server",
},
}
podSpec := s.podSpec(ctx, image, name, persistent, &selector)
podSpec.Volumes = append(podSpec.Volumes, volumes...)
podSpec.Containers[0].VolumeMounts = append(podSpec.Containers[0].VolumeMounts, volumeMounts...)
@@ -325,23 +342,16 @@ func (s *Server) StatefulServer(ctx context.Context, cluster *v1alpha1.Cluster)
ObjectMeta: metav1.ObjectMeta{
Name: cluster.Name + "-" + name,
Namespace: util.ClusterNamespace(cluster),
Labels: selector.MatchLabels,
},
Spec: apps.StatefulSetSpec{
Replicas: &replicas,
ServiceName: cluster.Name + "-" + name + "-headless",
Selector: &metav1.LabelSelector{
MatchLabels: map[string]string{
"cluster": cluster.Name,
"role": "server",
},
},
Replicas: &replicas,
ServiceName: cluster.Name + "-" + name + "-headless",
Selector: &selector,
VolumeClaimTemplates: pvClaims,
Template: v1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{
Labels: map[string]string{
"cluster": cluster.Name,
"role": "server",
},
Labels: selector.MatchLabels,
},
Spec: podSpec,
},
pkg/controller/cluster/server/service.go
+1 -1
View File
@@ -23,7 +23,7 @@ func (s *Server) Service(cluster *v1alpha1.Cluster) *v1.Service {
APIVersion: "v1",
},
ObjectMeta: metav1.ObjectMeta{
Name: "k3k-server-service",
Name: util.ServerSvcName(cluster),
Namespace: util.ClusterNamespace(cluster),
},
Spec: v1.ServiceSpec{
pkg/controller/kubeconfig/kubeconfig.go
+1 -1
View File
@@ -57,7 +57,7 @@ func (k *KubeConfig) Extract(ctx context.Context, client client.Client, cluster
}
// get the server service to extract the right IP
nn = types.NamespacedName{
Name: "k3k-server-service",
Name: util.ServerSvcName(cluster),
Namespace: util.ClusterNamespace(cluster),
}
pkg/controller/util/util.go
+18 -1
View File
@@ -2,6 +2,7 @@ package util
import (
"context"
"fmt"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
v1 "k8s.io/api/core/v1"
@@ -21,7 +22,23 @@ const (
)
func ClusterNamespace(cluster *v1alpha1.Cluster) string {
return namespacePrefix + cluster.Name
return cluster.Namespace
}
func ServerSvcName(cluster *v1alpha1.Cluster) string {
return fmt.Sprintf("k3k-%s-service", cluster.Name)
}
func ServerConfigName(cluster *v1alpha1.Cluster) string {
return fmt.Sprintf("k3k-%s-server-config", cluster.Name)
}
func ServerInitConfigName(cluster *v1alpha1.Cluster) string {
return fmt.Sprintf("k3k-init-%s-server-config", cluster.Name)
}
func AgentConfigName(cluster *v1alpha1.Cluster) string {
return fmt.Sprintf("k3k-%s-agent-config", cluster.Name)
}
func K3SImage(cluster *v1alpha1.Cluster) string {