4f1ad16c76
Support configuring kubelet credential provider config. Partially fixes: #7911 Signed-off-by: Noel Georgi <git@frezbo.dev>
538 lines
14 KiB
Go
538 lines
14 KiB
Go
// This Source Code Form is subject to the terms of the Mozilla Public
|
|
// License, v. 2.0. If a copy of the MPL was not distributed with this
|
|
// file, You can obtain one at http://mozilla.org/MPL/2.0/.
|
|
|
|
package k8s
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/x509"
|
|
"encoding/base64"
|
|
stdjson "encoding/json"
|
|
"encoding/pem"
|
|
"errors"
|
|
"fmt"
|
|
"net/url"
|
|
"os"
|
|
"path/filepath"
|
|
"text/template"
|
|
|
|
"github.com/cosi-project/runtime/pkg/controller"
|
|
"github.com/cosi-project/runtime/pkg/resource"
|
|
"github.com/cosi-project/runtime/pkg/safe"
|
|
"github.com/cosi-project/runtime/pkg/state"
|
|
"github.com/siderolabs/gen/optional"
|
|
"go.uber.org/zap"
|
|
"k8s.io/apimachinery/pkg/runtime"
|
|
"k8s.io/apimachinery/pkg/runtime/serializer/json"
|
|
"k8s.io/client-go/tools/clientcmd"
|
|
kubeletconfig "k8s.io/kubelet/config/v1beta1"
|
|
|
|
runtimetalos "github.com/siderolabs/talos/internal/app/machined/pkg/runtime"
|
|
"github.com/siderolabs/talos/internal/app/machined/pkg/system"
|
|
"github.com/siderolabs/talos/internal/app/machined/pkg/system/services"
|
|
"github.com/siderolabs/talos/pkg/machinery/constants"
|
|
"github.com/siderolabs/talos/pkg/machinery/resources/files"
|
|
"github.com/siderolabs/talos/pkg/machinery/resources/k8s"
|
|
runtimeres "github.com/siderolabs/talos/pkg/machinery/resources/runtime"
|
|
"github.com/siderolabs/talos/pkg/machinery/resources/secrets"
|
|
)
|
|
|
|
// ServiceManager is the interface to the v1alpha1 services subsystems.
|
|
type ServiceManager interface {
|
|
IsRunning(id string) (system.Service, bool, error)
|
|
Load(services ...system.Service) []string
|
|
Stop(ctx context.Context, serviceIDs ...string) (err error)
|
|
Start(serviceIDs ...string) error
|
|
}
|
|
|
|
// KubeletServiceController renders kubelet configuration files and controls kubelet service lifecycle.
|
|
type KubeletServiceController struct {
|
|
V1Alpha1Services ServiceManager
|
|
V1Alpha1Mode runtimetalos.Mode
|
|
}
|
|
|
|
// Name implements controller.Controller interface.
|
|
func (ctrl *KubeletServiceController) Name() string {
|
|
return "k8s.KubeletServiceController"
|
|
}
|
|
|
|
// Inputs implements controller.Controller interface.
|
|
func (ctrl *KubeletServiceController) Inputs() []controller.Input {
|
|
return nil
|
|
}
|
|
|
|
// Outputs implements controller.Controller interface.
|
|
func (ctrl *KubeletServiceController) Outputs() []controller.Output {
|
|
return nil
|
|
}
|
|
|
|
// Run implements controller.Controller interface.
|
|
//
|
|
//nolint:gocyclo,cyclop
|
|
func (ctrl *KubeletServiceController) Run(ctx context.Context, r controller.Runtime, logger *zap.Logger) error {
|
|
// initially, wait for the machine-id to be generated and /var to be mounted
|
|
if err := r.UpdateInputs([]controller.Input{
|
|
{
|
|
Namespace: files.NamespaceName,
|
|
Type: files.EtcFileStatusType,
|
|
ID: optional.Some("machine-id"),
|
|
Kind: controller.InputWeak,
|
|
},
|
|
{
|
|
Namespace: runtimeres.NamespaceName,
|
|
Type: runtimeres.MountStatusType,
|
|
ID: optional.Some(constants.EphemeralPartitionLabel),
|
|
Kind: controller.InputWeak,
|
|
},
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil
|
|
case <-r.EventCh():
|
|
}
|
|
|
|
_, err := r.Get(ctx, resource.NewMetadata(files.NamespaceName, files.EtcFileStatusType, "machine-id", resource.VersionUndefined))
|
|
if err != nil {
|
|
if state.IsNotFoundError(err) {
|
|
continue
|
|
}
|
|
|
|
return fmt.Errorf("error getting etc file status: %w", err)
|
|
}
|
|
|
|
_, err = r.Get(ctx, resource.NewMetadata(runtimeres.NamespaceName, runtimeres.MountStatusType, constants.EphemeralPartitionLabel, resource.VersionUndefined))
|
|
if err != nil {
|
|
if state.IsNotFoundError(err) {
|
|
// in container mode EPHEMERAL is always mounted
|
|
if ctrl.V1Alpha1Mode != runtimetalos.ModeContainer {
|
|
// wait for the EPHEMERAL to be mounted
|
|
continue
|
|
}
|
|
} else {
|
|
return fmt.Errorf("error getting ephemeral mount status: %w", err)
|
|
}
|
|
}
|
|
|
|
break
|
|
}
|
|
|
|
// normal reconcile loop
|
|
if err := r.UpdateInputs([]controller.Input{
|
|
{
|
|
Namespace: k8s.NamespaceName,
|
|
Type: k8s.KubeletSpecType,
|
|
ID: optional.Some(k8s.KubeletID),
|
|
Kind: controller.InputWeak,
|
|
},
|
|
{
|
|
Namespace: secrets.NamespaceName,
|
|
Type: secrets.KubeletType,
|
|
ID: optional.Some(secrets.KubeletID),
|
|
Kind: controller.InputWeak,
|
|
},
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
|
|
r.QueueReconcile()
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil
|
|
case <-r.EventCh():
|
|
}
|
|
|
|
cfg, err := safe.ReaderGetByID[*k8s.KubeletSpec](ctx, r, k8s.KubeletID)
|
|
if err != nil {
|
|
if state.IsNotFoundError(err) {
|
|
continue
|
|
}
|
|
|
|
return fmt.Errorf("error getting config: %w", err)
|
|
}
|
|
|
|
cfgSpec := cfg.TypedSpec()
|
|
|
|
secret, err := safe.ReaderGetByID[*secrets.Kubelet](ctx, r, secrets.KubeletID)
|
|
if err != nil {
|
|
if state.IsNotFoundError(err) {
|
|
continue
|
|
}
|
|
|
|
return fmt.Errorf("error getting secrets: %w", err)
|
|
}
|
|
|
|
secretSpec := secret.TypedSpec()
|
|
|
|
if err = ctrl.writePKI(secretSpec); err != nil {
|
|
return fmt.Errorf("error writing kubelet PKI: %w", err)
|
|
}
|
|
|
|
if err = ctrl.writeConfig(cfgSpec); err != nil {
|
|
return fmt.Errorf("error writing kubelet configuration: %w", err)
|
|
}
|
|
|
|
if err = ctrl.writeKubeletCredentialProviderConfig(cfgSpec); err != nil {
|
|
return fmt.Errorf("error writing kubelet credential provider configuration: %w", err)
|
|
}
|
|
|
|
_, running, err := ctrl.V1Alpha1Services.IsRunning("kubelet")
|
|
if err != nil {
|
|
ctrl.V1Alpha1Services.Load(&services.Kubelet{})
|
|
}
|
|
|
|
if running {
|
|
if err = ctrl.V1Alpha1Services.Stop(ctx, "kubelet"); err != nil {
|
|
return fmt.Errorf("error stopping kubelet service: %w", err)
|
|
}
|
|
}
|
|
|
|
// refresh certs only if we are managing the node name (not overridden by the user)
|
|
if cfgSpec.ExpectedNodename != "" {
|
|
err = ctrl.refreshKubeletCerts(cfgSpec.ExpectedNodename, logger)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
if err = ctrl.handlePolicyChange(cfgSpec, logger); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err = ctrl.refreshSelfServingCert(); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err = ctrl.updateKubeconfig(secretSpec.Endpoint, logger); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err = ctrl.V1Alpha1Services.Start("kubelet"); err != nil {
|
|
return fmt.Errorf("error starting kubelet service: %w", err)
|
|
}
|
|
|
|
r.ResetRestartBackoff()
|
|
}
|
|
}
|
|
|
|
// handlePolicyChange handles the cpuManagerPolicy change.
|
|
func (ctrl *KubeletServiceController) handlePolicyChange(cfgSpec *k8s.KubeletSpecSpec, logger *zap.Logger) error {
|
|
const managerFilename = "/var/lib/kubelet/cpu_manager_state"
|
|
|
|
oldPolicy, err := loadPolicyFromFile(managerFilename)
|
|
|
|
switch {
|
|
case errors.Is(err, os.ErrNotExist):
|
|
return nil // no cpu_manager_state file, nothing to do
|
|
case err != nil:
|
|
return fmt.Errorf("error loading cpu_manager_state file: %w", err)
|
|
}
|
|
|
|
policy, err := getFromMap[string](cfgSpec.Config, "cpuManagerPolicy")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
newPolicy := policy.ValueOrZero()
|
|
if equalPolicy(oldPolicy, newPolicy) {
|
|
return nil
|
|
}
|
|
|
|
logger.Info("cpuManagerPolicy changed", zap.String("old", oldPolicy), zap.String("new", newPolicy))
|
|
|
|
err = os.Remove(managerFilename)
|
|
if err != nil {
|
|
return fmt.Errorf("error removing cpu_manager_state file: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func loadPolicyFromFile(filename string) (string, error) {
|
|
raw, err := os.ReadFile(filename)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
|
|
cpuManagerState := struct {
|
|
Policy string `json:"policyName"`
|
|
}{}
|
|
|
|
if err = stdjson.Unmarshal(raw, &cpuManagerState); err != nil {
|
|
return "", err
|
|
}
|
|
|
|
return cpuManagerState.Policy, nil
|
|
}
|
|
|
|
func equalPolicy(current, newOne string) bool {
|
|
if current == "none" {
|
|
current = ""
|
|
}
|
|
|
|
if newOne == "none" {
|
|
newOne = ""
|
|
}
|
|
|
|
return current == newOne
|
|
}
|
|
|
|
func getFromMap[T any](m map[string]any, key string) (optional.Optional[T], error) {
|
|
var zero optional.Optional[T]
|
|
|
|
res, ok := m[key]
|
|
if !ok {
|
|
return zero, nil
|
|
}
|
|
|
|
if res, ok := res.(T); ok {
|
|
return optional.Some(res), nil
|
|
}
|
|
|
|
return zero, fmt.Errorf("unexpected type for key %q: found %T, expected %T", key, res, *new(T))
|
|
}
|
|
|
|
func (ctrl *KubeletServiceController) writePKI(secretSpec *secrets.KubeletSpec) error {
|
|
cfg := struct {
|
|
Server string
|
|
CACert string
|
|
BootstrapTokenID string
|
|
BootstrapTokenSecret string
|
|
}{
|
|
Server: secretSpec.Endpoint.String(),
|
|
CACert: base64.StdEncoding.EncodeToString(secretSpec.CA.Crt),
|
|
BootstrapTokenID: secretSpec.BootstrapTokenID,
|
|
BootstrapTokenSecret: secretSpec.BootstrapTokenSecret,
|
|
}
|
|
|
|
templ := template.Must(template.New("tmpl").Parse(string(kubeletKubeConfigTemplate)))
|
|
|
|
var buf bytes.Buffer
|
|
|
|
if err := templ.Execute(&buf, cfg); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := os.WriteFile(constants.KubeletBootstrapKubeconfig, buf.Bytes(), 0o600); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := os.MkdirAll(filepath.Dir(constants.KubernetesCACert), 0o700); err != nil {
|
|
return err
|
|
}
|
|
|
|
return os.WriteFile(constants.KubernetesCACert, secretSpec.CA.Crt, 0o400)
|
|
}
|
|
|
|
var kubeletKubeConfigTemplate = []byte(`apiVersion: v1
|
|
kind: Config
|
|
clusters:
|
|
- name: local
|
|
cluster:
|
|
server: {{ .Server }}
|
|
certificate-authority-data: {{ .CACert }}
|
|
users:
|
|
- name: kubelet
|
|
user:
|
|
token: {{ .BootstrapTokenID }}.{{ .BootstrapTokenSecret }}
|
|
contexts:
|
|
- context:
|
|
cluster: local
|
|
user: kubelet
|
|
`)
|
|
|
|
func (ctrl *KubeletServiceController) writeConfig(cfgSpec *k8s.KubeletSpecSpec) error {
|
|
var kubeletConfiguration kubeletconfig.KubeletConfiguration
|
|
|
|
if err := runtime.DefaultUnstructuredConverter.FromUnstructured(cfgSpec.Config, &kubeletConfiguration); err != nil {
|
|
return fmt.Errorf("error converting kubelet configuration from unstructured: %w", err)
|
|
}
|
|
|
|
serializer := json.NewSerializerWithOptions(
|
|
json.DefaultMetaFactory,
|
|
nil,
|
|
nil,
|
|
json.SerializerOptions{
|
|
Yaml: true,
|
|
},
|
|
)
|
|
|
|
var buf bytes.Buffer
|
|
|
|
if err := serializer.Encode(&kubeletConfiguration, &buf); err != nil {
|
|
return err
|
|
}
|
|
|
|
return os.WriteFile("/etc/kubernetes/kubelet.yaml", buf.Bytes(), 0o600)
|
|
}
|
|
|
|
func (ctrl *KubeletServiceController) writeKubeletCredentialProviderConfig(cfgSpec *k8s.KubeletSpecSpec) error {
|
|
if cfgSpec.CredentialProviderConfig == nil {
|
|
return os.RemoveAll(constants.KubeletCredentialProviderConfig)
|
|
}
|
|
|
|
var kubeletCredentialProviderConfig kubeletconfig.CredentialProviderConfig
|
|
|
|
if err := runtime.DefaultUnstructuredConverter.FromUnstructured(cfgSpec.CredentialProviderConfig, &kubeletCredentialProviderConfig); err != nil {
|
|
return fmt.Errorf("error converting kubelet credentialprovider configuration from unstructured: %w", err)
|
|
}
|
|
|
|
serializer := json.NewSerializerWithOptions(
|
|
json.DefaultMetaFactory,
|
|
nil,
|
|
nil,
|
|
json.SerializerOptions{
|
|
Yaml: true,
|
|
},
|
|
)
|
|
|
|
var buf bytes.Buffer
|
|
|
|
if err := serializer.Encode(&kubeletCredentialProviderConfig, &buf); err != nil {
|
|
return err
|
|
}
|
|
|
|
return os.WriteFile(constants.KubeletCredentialProviderConfig, buf.Bytes(), 0o600)
|
|
}
|
|
|
|
// updateKubeconfig updates the kubeconfig of kubelet with the given endpoint if it exists.
|
|
func (ctrl *KubeletServiceController) updateKubeconfig(newEndpoint *url.URL, logger *zap.Logger) error {
|
|
config, err := clientcmd.LoadFromFile(constants.KubeletKubeconfig)
|
|
if errors.Is(err, os.ErrNotExist) {
|
|
return nil
|
|
}
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
context := config.Contexts[config.CurrentContext]
|
|
if context == nil {
|
|
// this should never happen, but we can't fix kubeconfig if it is malformed
|
|
logger.Error("kubeconfig is missing current context", zap.String("context", config.CurrentContext))
|
|
|
|
return nil
|
|
}
|
|
|
|
cluster := config.Clusters[context.Cluster]
|
|
|
|
if cluster == nil {
|
|
// this should never happen, but we can't fix kubeconfig if it is malformed
|
|
logger.Error("kubeconfig is missing cluster", zap.String("context", config.CurrentContext), zap.String("cluster", context.Cluster))
|
|
|
|
return nil
|
|
}
|
|
|
|
if cluster.Server == newEndpoint.String() {
|
|
return nil
|
|
}
|
|
|
|
cluster.Server = newEndpoint.String()
|
|
|
|
return clientcmd.WriteToFile(*config, constants.KubeletKubeconfig)
|
|
}
|
|
|
|
// refreshKubeletCerts checks if the existing kubelet certificates match the node hostname.
|
|
// If they don't match, it clears the certificate directory and the removes kubelet's kubeconfig so that
|
|
// they can be regenerated next time kubelet is started.
|
|
func (ctrl *KubeletServiceController) refreshKubeletCerts(nodename string, logger *zap.Logger) error {
|
|
cert, err := ctrl.readKubeletClientCertificate()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if cert == nil {
|
|
return nil
|
|
}
|
|
|
|
expectedCommonName := fmt.Sprintf("system:node:%s", nodename)
|
|
|
|
valid := expectedCommonName == cert.Subject.CommonName
|
|
|
|
if valid {
|
|
// certificate looks good, no need to refresh
|
|
return nil
|
|
}
|
|
|
|
logger.Info("kubelet client certificate does not match expected nodename, removing",
|
|
zap.String("expected", expectedCommonName),
|
|
zap.String("actual", cert.Subject.CommonName),
|
|
)
|
|
|
|
// remove the pki directory
|
|
err = os.RemoveAll(constants.KubeletPKIDir)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// clear the kubelet kubeconfig
|
|
err = os.Remove(constants.KubeletKubeconfig)
|
|
if errors.Is(err, os.ErrNotExist) {
|
|
return nil
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
// refreshSelfServingCert removes the self-signed serving certificate (if exists) to force the kubelet to renew it.
|
|
func (ctrl *KubeletServiceController) refreshSelfServingCert() error {
|
|
for _, filename := range []string{
|
|
"kubelet.crt",
|
|
"kubelet.key",
|
|
} {
|
|
path := filepath.Join(constants.KubeletPKIDir, filename)
|
|
|
|
_, err := os.Stat(path)
|
|
if err == nil {
|
|
err = os.Remove(path)
|
|
if err != nil {
|
|
return fmt.Errorf("error removing self-signed certificate: %w", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (ctrl *KubeletServiceController) readKubeletClientCertificate() (*x509.Certificate, error) {
|
|
raw, err := os.ReadFile(filepath.Join(constants.KubeletPKIDir, "kubelet-client-current.pem"))
|
|
if errors.Is(err, os.ErrNotExist) {
|
|
return nil, nil
|
|
}
|
|
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
for {
|
|
block, rest := pem.Decode(raw)
|
|
if block == nil {
|
|
return nil, nil
|
|
}
|
|
|
|
raw = rest
|
|
|
|
if block.Type != "CERTIFICATE" {
|
|
continue
|
|
}
|
|
|
|
var cert *x509.Certificate
|
|
|
|
cert, err = x509.ParseCertificate(block.Bytes)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if !cert.IsCA {
|
|
return cert, nil
|
|
}
|
|
}
|
|
}
|