2019-01-09 16:54:15 +00:00
package agent
import (
"context"
2022-04-22 14:04:22 +00:00
"fmt"
2022-02-16 22:19:58 +00:00
"net"
2019-01-09 16:54:15 +00:00
"os"
"path/filepath"
2023-10-16 18:53:09 +00:00
goruntime "runtime"
2022-02-16 22:19:58 +00:00
"strconv"
2019-02-08 04:13:43 +00:00
"strings"
2019-01-09 16:54:15 +00:00
"time"
2024-03-28 17:09:48 +00:00
systemd "github.com/coreos/go-systemd/v22/daemon"
2022-03-02 23:47:27 +00:00
"github.com/k3s-io/k3s/pkg/agent/config"
2024-02-21 18:22:03 +00:00
"github.com/k3s-io/k3s/pkg/agent/containerd"
2022-03-02 23:47:27 +00:00
"github.com/k3s-io/k3s/pkg/agent/flannel"
"github.com/k3s-io/k3s/pkg/agent/netpol"
"github.com/k3s-io/k3s/pkg/agent/proxy"
"github.com/k3s-io/k3s/pkg/agent/syssetup"
"github.com/k3s-io/k3s/pkg/agent/tunnel"
2024-03-23 00:13:08 +00:00
"github.com/k3s-io/k3s/pkg/certmonitor"
2022-03-02 23:47:27 +00:00
"github.com/k3s-io/k3s/pkg/cgroups"
"github.com/k3s-io/k3s/pkg/cli/cmds"
"github.com/k3s-io/k3s/pkg/clientaccess"
cp "github.com/k3s-io/k3s/pkg/cloudprovider"
"github.com/k3s-io/k3s/pkg/daemons/agent"
daemonconfig "github.com/k3s-io/k3s/pkg/daemons/config"
"github.com/k3s-io/k3s/pkg/daemons/executor"
2024-04-25 01:02:05 +00:00
"github.com/k3s-io/k3s/pkg/metrics"
2022-03-02 23:47:27 +00:00
"github.com/k3s-io/k3s/pkg/nodeconfig"
2024-04-25 01:02:05 +00:00
"github.com/k3s-io/k3s/pkg/profile"
2022-03-02 23:47:27 +00:00
"github.com/k3s-io/k3s/pkg/rootless"
2023-12-06 00:25:49 +00:00
"github.com/k3s-io/k3s/pkg/spegel"
2022-03-02 23:47:27 +00:00
"github.com/k3s-io/k3s/pkg/util"
2022-06-15 16:00:52 +00:00
"github.com/k3s-io/k3s/pkg/version"
2021-04-21 22:56:20 +00:00
"github.com/pkg/errors"
2019-01-22 21:14:58 +00:00
"github.com/sirupsen/logrus"
2022-04-21 20:56:39 +00:00
v1 "k8s.io/api/core/v1"
2019-10-27 05:53:25 +00:00
"k8s.io/apimachinery/pkg/api/equality"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
2021-10-29 08:59:03 +00:00
"k8s.io/apimachinery/pkg/fields"
2020-04-28 22:00:30 +00:00
"k8s.io/apimachinery/pkg/labels"
2022-04-21 20:56:39 +00:00
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/watch"
2021-10-29 08:59:03 +00:00
typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
2022-04-21 20:56:39 +00:00
"k8s.io/client-go/tools/cache"
toolswatch "k8s.io/client-go/tools/watch"
2023-11-21 00:07:20 +00:00
"k8s.io/component-base/cli/globalflag"
"k8s.io/component-base/logs"
2021-05-17 20:30:55 +00:00
app2 "k8s.io/kubernetes/cmd/kube-proxy/app"
kubeproxyconfig "k8s.io/kubernetes/pkg/proxy/apis/config"
2021-04-21 22:56:20 +00:00
utilsnet "k8s.io/utils/net"
2024-03-28 17:07:02 +00:00
utilsptr "k8s.io/utils/ptr"
2019-01-09 16:54:15 +00:00
)
2020-06-24 21:16:44 +00:00
func run ( ctx context . Context , cfg cmds . Agent , proxy proxy . Proxy ) error {
2023-12-08 00:13:53 +00:00
nodeConfig , err := config . Get ( ctx , cfg , proxy )
if err != nil {
return errors . Wrap ( err , "failed to retrieve agent configuration" )
}
2020-06-24 21:16:44 +00:00
2021-04-21 22:56:20 +00:00
dualCluster , err := utilsnet . IsDualStackCIDRs ( nodeConfig . AgentConfig . ClusterCIDRs )
if err != nil {
return errors . Wrap ( err , "failed to validate cluster-cidr" )
}
dualService , err := utilsnet . IsDualStackCIDRs ( nodeConfig . AgentConfig . ServiceCIDRs )
if err != nil {
return errors . Wrap ( err , "failed to validate service-cidr" )
}
dualNode , err := utilsnet . IsDualStackIPs ( nodeConfig . AgentConfig . NodeIPs )
if err != nil {
return errors . Wrap ( err , "failed to validate node-ip" )
}
2022-04-20 14:01:49 +00:00
serviceIPv4 := utilsnet . IsIPv4CIDR ( nodeConfig . AgentConfig . ServiceCIDR )
clusterIPv4 := utilsnet . IsIPv4CIDR ( nodeConfig . AgentConfig . ClusterCIDR )
2022-04-22 14:04:22 +00:00
nodeIPv4 := utilsnet . IsIPv4String ( nodeConfig . AgentConfig . NodeIP )
2021-11-10 19:23:05 +00:00
serviceIPv6 := utilsnet . IsIPv6CIDR ( nodeConfig . AgentConfig . ServiceCIDR )
clusterIPv6 := utilsnet . IsIPv6CIDR ( nodeConfig . AgentConfig . ClusterCIDR )
2022-04-22 14:04:22 +00:00
nodeIPv6 := utilsnet . IsIPv6String ( nodeConfig . AgentConfig . NodeIP )
2023-09-21 13:39:05 +00:00
// check that cluster-cidr and service-cidr have the same IP versions
2022-04-22 14:04:22 +00:00
if ( serviceIPv6 != clusterIPv6 ) || ( dualCluster != dualService ) || ( serviceIPv4 != clusterIPv4 ) {
return fmt . Errorf ( "cluster-cidr: %v and service-cidr: %v, must share the same IP version (IPv4, IPv6 or dual-stack)" , nodeConfig . AgentConfig . ClusterCIDRs , nodeConfig . AgentConfig . ServiceCIDRs )
}
2023-09-21 13:39:05 +00:00
// check that node-ip has the IP versions set in cluster-cidr
if ( clusterIPv6 && ! ( nodeIPv6 || dualNode ) ) || ( dualCluster && ! dualNode ) || ( clusterIPv4 && ! ( nodeIPv4 || dualNode ) ) {
2022-04-22 14:04:22 +00:00
return fmt . Errorf ( "cluster-cidr: %v and node-ip: %v, must share the same IP version (IPv4, IPv6 or dual-stack)" , nodeConfig . AgentConfig . ClusterCIDRs , nodeConfig . AgentConfig . NodeIPs )
}
2023-09-21 13:39:05 +00:00
2022-04-22 14:04:22 +00:00
enableIPv6 := dualCluster || clusterIPv6
enableIPv4 := dualCluster || clusterIPv4
2021-04-21 22:56:20 +00:00
2023-10-16 18:53:09 +00:00
// dualStack or IPv6 are not supported on Windows node
if ( goruntime . GOOS == "windows" ) && enableIPv6 {
return fmt . Errorf ( "dual-stack or IPv6 are not supported on Windows node" )
}
2021-05-17 20:30:55 +00:00
conntrackConfig , err := getConntrackConfig ( nodeConfig )
if err != nil {
return errors . Wrap ( err , "failed to validate kube-proxy conntrack configuration" )
}
syssetup . Configure ( enableIPv6 , conntrackConfig )
2022-04-20 14:01:49 +00:00
nodeConfig . AgentConfig . EnableIPv4 = enableIPv4
2021-11-09 15:44:34 +00:00
nodeConfig . AgentConfig . EnableIPv6 = enableIPv6
2021-04-21 22:56:20 +00:00
2023-12-06 00:25:49 +00:00
if nodeConfig . EmbeddedRegistry {
if nodeConfig . Docker || nodeConfig . ContainerRuntimeEndpoint != "" {
return errors . New ( "embedded registry mirror requires embedded containerd" )
}
if err := spegel . DefaultRegistry . Start ( ctx , nodeConfig ) ; err != nil {
return errors . Wrap ( err , "failed to start embedded registry" )
}
}
2024-04-25 01:02:05 +00:00
if nodeConfig . SupervisorMetrics {
if err := metrics . DefaultMetrics . Start ( ctx , nodeConfig ) ; err != nil {
return errors . Wrap ( err , "failed to serve metrics" )
}
}
if nodeConfig . EnablePProf {
if err := profile . DefaultProfiler . Start ( ctx , nodeConfig ) ; err != nil {
return errors . Wrap ( err , "failed to serve pprof" )
}
}
2020-06-24 21:16:44 +00:00
if err := setupCriCtlConfig ( cfg , nodeConfig ) ; err != nil {
return err
}
2021-05-11 19:50:08 +00:00
if err := executor . Bootstrap ( ctx , nodeConfig , cfg ) ; err != nil {
return err
}
2019-01-09 16:54:15 +00:00
if ! nodeConfig . NoFlannel {
2022-10-31 17:07:43 +00:00
if ( nodeConfig . FlannelExternalIP ) && ( len ( nodeConfig . AgentConfig . NodeExternalIPs ) == 0 ) {
logrus . Warnf ( "Server has flannel-external-ip flag set but this node does not set node-external-ip. Flannel will use internal address when connecting to this node." )
2023-08-25 05:09:13 +00:00
} else if ( nodeConfig . FlannelExternalIP ) && ( nodeConfig . FlannelBackend != daemonconfig . FlannelBackendWireguardNative ) {
2022-10-31 17:07:43 +00:00
logrus . Warnf ( "Flannel is using external addresses with an insecure backend: %v. Please consider using an encrypting flannel backend." , nodeConfig . FlannelBackend )
}
2019-01-09 16:54:15 +00:00
if err := flannel . Prepare ( ctx , nodeConfig ) ; err != nil {
return err
}
}
2024-02-21 18:22:03 +00:00
if nodeConfig . Docker {
if err := executor . Docker ( ctx , nodeConfig ) ; err != nil {
return err
}
} else if nodeConfig . ContainerRuntimeEndpoint == "" {
if err := containerd . SetupContainerdConfig ( nodeConfig ) ; err != nil {
return err
}
if err := executor . Containerd ( ctx , nodeConfig ) ; err != nil {
return err
}
2019-01-09 16:54:15 +00:00
}
2024-02-21 18:26:13 +00:00
// the container runtime is ready to host workloads when containerd is up and the airgap
2021-10-12 06:13:10 +00:00
// images have finished loading, as that portion of startup may block for an arbitrary
// amount of time depending on how long it takes to import whatever the user has placed
// in the images directory.
2024-02-21 18:26:13 +00:00
if cfg . ContainerRuntimeReady != nil {
close ( cfg . ContainerRuntimeReady )
2021-10-12 06:13:10 +00:00
}
2024-02-21 18:22:03 +00:00
notifySocket := os . Getenv ( "NOTIFY_SOCKET" )
os . Unsetenv ( "NOTIFY_SOCKET" )
if err := setupTunnelAndRunAgent ( ctx , nodeConfig , cfg , proxy ) ; err != nil {
return err
}
2022-04-29 19:57:38 +00:00
if err := util . WaitForAPIServerReady ( ctx , nodeConfig . AgentConfig . KubeConfigKubelet , util . DefaultAPIServerReadyTimeout ) ; err != nil {
return errors . Wrap ( err , "failed to wait for apiserver ready" )
}
2022-12-07 18:08:53 +00:00
coreClient , err := util . GetClientSet ( nodeConfig . AgentConfig . KubeConfigKubelet )
2019-10-27 05:53:25 +00:00
if err != nil {
return err
}
2021-02-26 18:37:27 +00:00
2022-10-22 00:22:01 +00:00
if err := configureNode ( ctx , nodeConfig , coreClient . CoreV1 ( ) . Nodes ( ) ) ; err != nil {
2021-10-29 08:59:03 +00:00
return err
}
2019-01-09 16:54:15 +00:00
if ! nodeConfig . NoFlannel {
2019-10-27 05:53:25 +00:00
if err := flannel . Run ( ctx , nodeConfig , coreClient . CoreV1 ( ) . Nodes ( ) ) ; err != nil {
2019-01-09 16:54:15 +00:00
return err
}
}
2019-10-17 21:46:15 +00:00
if ! nodeConfig . AgentConfig . DisableNPC {
if err := netpol . Run ( ctx , nodeConfig ) ; err != nil {
return err
}
}
2022-06-15 16:00:52 +00:00
// By default, the server is responsible for notifying systemd
// On agent-only nodes, the agent will notify systemd
if notifySocket != "" {
logrus . Info ( version . Program + " agent is up and running" )
os . Setenv ( "NOTIFY_SOCKET" , notifySocket )
systemd . SdNotify ( true , "READY=1\n" )
}
2021-06-15 11:20:26 +00:00
2019-01-09 16:54:15 +00:00
<- ctx . Done ( )
return ctx . Err ( )
}
2021-05-17 20:30:55 +00:00
// getConntrackConfig uses the kube-proxy code to parse the user-provided kube-proxy-arg values, and
// extract the conntrack settings so that K3s can set them itself. This allows us to soft-fail when
// running K3s in Docker, where kube-proxy is no longer allowed to set conntrack sysctls on newer kernels.
// When running rootless, we do not attempt to set conntrack sysctls - this behavior is copied from kubeadm.
func getConntrackConfig ( nodeConfig * daemonconfig . Node ) ( * kubeproxyconfig . KubeProxyConntrackConfiguration , error ) {
ctConfig := & kubeproxyconfig . KubeProxyConntrackConfiguration {
2024-03-28 17:07:02 +00:00
MaxPerCore : utilsptr . To ( int32 ( 0 ) ) ,
Min : utilsptr . To ( int32 ( 0 ) ) ,
2021-05-17 20:30:55 +00:00
TCPEstablishedTimeout : & metav1 . Duration { } ,
TCPCloseWaitTimeout : & metav1 . Duration { } ,
}
if nodeConfig . AgentConfig . Rootless {
return ctConfig , nil
}
cmd := app2 . NewProxyCommand ( )
2023-11-21 00:07:20 +00:00
globalflag . AddGlobalFlags ( cmd . Flags ( ) , cmd . Name ( ) , logs . SkipLoggingConfigurationFlags ( ) )
2021-07-28 20:04:19 +00:00
if err := cmd . ParseFlags ( daemonconfig . GetArgs ( map [ string ] string { } , nodeConfig . AgentConfig . ExtraKubeProxyArgs ) ) ; err != nil {
2021-05-17 20:30:55 +00:00
return nil , err
}
maxPerCore , err := cmd . Flags ( ) . GetInt32 ( "conntrack-max-per-core" )
if err != nil {
return nil , err
}
ctConfig . MaxPerCore = & maxPerCore
min , err := cmd . Flags ( ) . GetInt32 ( "conntrack-min" )
if err != nil {
return nil , err
}
ctConfig . Min = & min
establishedTimeout , err := cmd . Flags ( ) . GetDuration ( "conntrack-tcp-timeout-established" )
if err != nil {
return nil , err
}
ctConfig . TCPEstablishedTimeout . Duration = establishedTimeout
closeWaitTimeout , err := cmd . Flags ( ) . GetDuration ( "conntrack-tcp-timeout-close-wait" )
if err != nil {
return nil , err
}
ctConfig . TCPCloseWaitTimeout . Duration = closeWaitTimeout
return ctConfig , nil
}
2022-02-15 00:34:49 +00:00
// RunStandalone bootstraps the executor, but does not run the kubelet or containerd.
// This allows other bits of code that expect the executor to be set up properly to function
// even when the agent is disabled. It will only return in case of error or context
// cancellation.
func RunStandalone ( ctx context . Context , cfg cmds . Agent ) error {
proxy , err := createProxyAndValidateToken ( ctx , & cfg )
if err != nil {
return err
}
2023-12-08 00:13:53 +00:00
nodeConfig , err := config . Get ( ctx , cfg , proxy )
if err != nil {
return errors . Wrap ( err , "failed to retrieve agent configuration" )
}
2022-02-15 00:34:49 +00:00
if err := executor . Bootstrap ( ctx , nodeConfig , cfg ) ; err != nil {
return err
}
2024-02-21 18:26:13 +00:00
if cfg . ContainerRuntimeReady != nil {
close ( cfg . ContainerRuntimeReady )
2022-02-15 00:34:49 +00:00
}
2022-04-27 20:44:15 +00:00
if err := tunnelSetup ( ctx , nodeConfig , cfg , proxy ) ; err != nil {
2022-03-29 18:36:48 +00:00
return err
}
2024-03-23 00:13:08 +00:00
if err := certMonitorSetup ( ctx , nodeConfig , cfg ) ; err != nil {
return err
}
2022-03-29 18:36:48 +00:00
2024-07-26 18:23:57 +00:00
if nodeConfig . SupervisorMetrics {
if err := metrics . DefaultMetrics . Start ( ctx , nodeConfig ) ; err != nil {
return errors . Wrap ( err , "failed to serve metrics" )
}
}
if nodeConfig . EnablePProf {
if err := profile . DefaultProfiler . Start ( ctx , nodeConfig ) ; err != nil {
return errors . Wrap ( err , "failed to serve pprof" )
}
}
2022-02-15 00:34:49 +00:00
<- ctx . Done ( )
return ctx . Err ( )
}
// Run sets up cgroups, configures the LB proxy, and triggers startup
// of containerd and kubelet. It will only return in case of error or context
// cancellation.
2019-01-09 16:54:15 +00:00
func Run ( ctx context . Context , cfg cmds . Agent ) error {
2021-06-01 19:29:46 +00:00
if err := cgroups . Validate ( ) ; err != nil {
2019-02-08 04:13:43 +00:00
return err
}
2019-10-19 10:18:51 +00:00
if cfg . Rootless && ! cfg . RootlessAlreadyUnshared {
2022-11-15 08:10:12 +00:00
dualNode , err := utilsnet . IsDualStackIPStrings ( cfg . NodeIP )
if err != nil {
return err
}
if err := rootless . Rootless ( cfg . DataDir , dualNode ) ; err != nil {
2019-03-08 22:47:44 +00:00
return err
}
}
2022-02-15 00:34:49 +00:00
proxy , err := createProxyAndValidateToken ( ctx , & cfg )
if err != nil {
return err
}
return run ( ctx , cfg , proxy )
}
func createProxyAndValidateToken ( ctx context . Context , cfg * cmds . Agent ) ( proxy . Proxy , error ) {
2020-11-03 19:19:26 +00:00
agentDir := filepath . Join ( cfg . DataDir , "agent" )
2022-12-08 23:59:21 +00:00
clientKubeletCert := filepath . Join ( agentDir , "client-kubelet.crt" )
clientKubeletKey := filepath . Join ( agentDir , "client-kubelet.key" )
2020-11-03 19:19:26 +00:00
if err := os . MkdirAll ( agentDir , 0700 ) ; err != nil {
2022-02-15 00:34:49 +00:00
return nil , err
2020-04-27 16:42:15 +00:00
}
2024-07-13 22:57:53 +00:00
isIPv6 := utilsnet . IsIPv6 ( net . ParseIP ( util . GetFirstValidIPString ( cfg . NodeIP ) ) )
2019-01-09 16:54:15 +00:00
2022-03-31 09:49:30 +00:00
proxy , err := proxy . NewSupervisorProxy ( ctx , ! cfg . DisableLoadBalancer , agentDir , cfg . ServerURL , cfg . LBServerPort , isIPv6 )
2019-07-24 07:22:31 +00:00
if err != nil {
2022-02-15 00:34:49 +00:00
return nil , err
2019-07-24 07:22:31 +00:00
}
2022-12-08 23:59:21 +00:00
options := [ ] clientaccess . ValidationOption {
clientaccess . WithUser ( "node" ) ,
clientaccess . WithClientCertificate ( clientKubeletCert , clientKubeletKey ) ,
}
2019-01-09 16:54:15 +00:00
for {
2022-12-08 23:59:21 +00:00
newToken , err := clientaccess . ParseAndValidateToken ( proxy . SupervisorURL ( ) , cfg . Token , options ... )
2019-01-09 16:54:15 +00:00
if err != nil {
logrus . Error ( err )
select {
case <- ctx . Done ( ) :
2022-02-15 00:34:49 +00:00
return nil , ctx . Err ( )
2019-01-09 16:54:15 +00:00
case <- time . After ( 2 * time . Second ) :
}
continue
}
2020-09-24 06:47:17 +00:00
cfg . Token = newToken . String ( )
2019-01-09 16:54:15 +00:00
break
}
2022-02-15 00:34:49 +00:00
return proxy , nil
2019-01-09 16:54:15 +00:00
}
2019-02-08 04:13:43 +00:00
2022-04-21 20:56:39 +00:00
// configureNode waits for the node object to be created, and if/when it does,
// ensures that the labels and annotations are up to date.
2022-10-22 00:22:01 +00:00
func configureNode ( ctx context . Context , nodeConfig * daemonconfig . Node , nodes typedcorev1 . NodeInterface ) error {
agentConfig := & nodeConfig . AgentConfig
2021-10-29 08:59:03 +00:00
fieldSelector := fields . Set { metav1 . ObjectNameField : agentConfig . NodeName } . String ( )
2022-04-21 20:56:39 +00:00
lw := & cache . ListWatch {
ListFunc : func ( options metav1 . ListOptions ) ( object runtime . Object , e error ) {
options . FieldSelector = fieldSelector
return nodes . List ( ctx , options )
} ,
WatchFunc : func ( options metav1 . ListOptions ) ( i watch . Interface , e error ) {
options . FieldSelector = fieldSelector
return nodes . Watch ( ctx , options )
} ,
}
condition := func ( ev watch . Event ) ( bool , error ) {
node , ok := ev . Object . ( * v1 . Node )
2021-10-29 08:59:03 +00:00
if ! ok {
2022-04-21 20:56:39 +00:00
return false , errors . New ( "event object not of type v1.Node" )
2019-10-15 21:17:26 +00:00
}
2019-10-27 05:53:25 +00:00
2021-04-21 22:56:20 +00:00
updateNode := false
if labels , changed := updateMutableLabels ( agentConfig , node . Labels ) ; changed {
node . Labels = labels
updateNode = true
}
2019-12-09 22:54:56 +00:00
2021-04-21 22:56:20 +00:00
if ! agentConfig . DisableCCM {
2022-10-22 00:22:01 +00:00
if annotations , changed := updateAddressAnnotations ( nodeConfig , node . Annotations ) ; changed {
2021-04-21 22:56:20 +00:00
node . Annotations = annotations
updateNode = true
}
if labels , changed := updateLegacyAddressLabels ( agentConfig , node . Labels ) ; changed {
node . Labels = labels
updateNode = true
}
2019-12-09 22:54:56 +00:00
}
2020-02-11 23:27:43 +00:00
// inject node config
2023-01-26 00:43:57 +00:00
if changed , err := nodeconfig . SetNodeConfigAnnotations ( nodeConfig , node ) ; err != nil {
2022-04-21 20:56:39 +00:00
return false , err
2021-04-21 22:56:20 +00:00
} else if changed {
2020-02-11 23:27:43 +00:00
updateNode = true
}
2021-04-21 22:56:20 +00:00
2023-01-26 00:43:57 +00:00
if changed , err := nodeconfig . SetNodeConfigLabels ( nodeConfig , node ) ; err != nil {
2022-04-04 21:54:50 +00:00
return false , err
} else if changed {
updateNode = true
}
2020-02-11 23:27:43 +00:00
if updateNode {
2020-03-26 21:08:47 +00:00
if _ , err := nodes . Update ( ctx , node , metav1 . UpdateOptions { } ) ; err != nil {
2022-04-04 21:54:50 +00:00
logrus . Infof ( "Failed to set annotations and labels on node %s: %v" , agentConfig . NodeName , err )
2022-04-21 20:56:39 +00:00
return false , nil
2019-10-15 21:17:26 +00:00
}
2022-04-04 21:54:50 +00:00
logrus . Infof ( "Annotations and labels have been set successfully on node: %s" , agentConfig . NodeName )
2022-04-21 20:56:39 +00:00
return true , nil
2019-10-15 21:17:26 +00:00
}
2022-04-04 21:54:50 +00:00
logrus . Infof ( "Annotations and labels have already set on node: %s" , agentConfig . NodeName )
2022-04-21 20:56:39 +00:00
return true , nil
2019-10-15 21:17:26 +00:00
}
2022-04-21 20:56:39 +00:00
if _ , err := toolswatch . UntilWithSync ( ctx , lw , & v1 . Node { } , nil , condition ) ; err != nil {
return errors . Wrap ( err , "failed to configure node" )
}
2019-10-27 05:53:25 +00:00
return nil
2019-10-15 21:17:26 +00:00
}
2019-12-09 22:54:56 +00:00
func updateMutableLabels ( agentConfig * daemonconfig . Agent , nodeLabels map [ string ] string ) ( map [ string ] string , bool ) {
2019-10-27 05:53:25 +00:00
result := map [ string ] string { }
2019-12-09 22:54:56 +00:00
for _ , m := range agentConfig . NodeLabels {
var (
v string
p = strings . SplitN ( m , ` = ` , 2 )
k = p [ 0 ]
)
if len ( p ) > 1 {
v = p [ 1 ]
}
2019-10-27 05:53:25 +00:00
result [ k ] = v
2019-10-15 21:17:26 +00:00
}
2019-12-09 22:54:56 +00:00
result = labels . Merge ( nodeLabels , result )
return result , ! equality . Semantic . DeepEqual ( nodeLabels , result )
}
2019-10-27 05:53:25 +00:00
2021-04-21 22:56:20 +00:00
func updateLegacyAddressLabels ( agentConfig * daemonconfig . Agent , nodeLabels map [ string ] string ) ( map [ string ] string , bool ) {
ls := labels . Set ( nodeLabels )
if ls . Has ( cp . InternalIPKey ) || ls . Has ( cp . HostnameKey ) {
result := map [ string ] string {
cp . InternalIPKey : agentConfig . NodeIP ,
2024-02-27 22:43:39 +00:00
cp . HostnameKey : getHostname ( agentConfig ) ,
2021-04-21 22:56:20 +00:00
}
if agentConfig . NodeExternalIP != "" {
result [ cp . ExternalIPKey ] = agentConfig . NodeExternalIP
}
result = labels . Merge ( nodeLabels , result )
return result , ! equality . Semantic . DeepEqual ( nodeLabels , result )
}
return nil , false
}
2022-09-26 12:56:55 +00:00
// updateAddressAnnotations updates the node annotations with important information about IP addresses of the node
2022-10-22 00:22:01 +00:00
func updateAddressAnnotations ( nodeConfig * daemonconfig . Node , nodeAnnotations map [ string ] string ) ( map [ string ] string , bool ) {
agentConfig := & nodeConfig . AgentConfig
2019-12-09 22:54:56 +00:00
result := map [ string ] string {
2021-04-21 22:56:20 +00:00
cp . InternalIPKey : util . JoinIPs ( agentConfig . NodeIPs ) ,
2024-02-27 22:43:39 +00:00
cp . HostnameKey : getHostname ( agentConfig ) ,
2019-12-09 22:54:56 +00:00
}
if agentConfig . NodeExternalIP != "" {
2021-04-21 22:56:20 +00:00
result [ cp . ExternalIPKey ] = util . JoinIPs ( agentConfig . NodeExternalIPs )
2022-10-22 00:22:01 +00:00
if nodeConfig . FlannelExternalIP {
for _ , ipAddress := range agentConfig . NodeExternalIPs {
if utilsnet . IsIPv4 ( ipAddress ) {
result [ flannel . FlannelExternalIPv4Annotation ] = ipAddress . String ( )
}
if utilsnet . IsIPv6 ( ipAddress ) {
result [ flannel . FlannelExternalIPv6Annotation ] = ipAddress . String ( )
}
2022-09-26 12:56:55 +00:00
}
}
2019-10-15 21:17:26 +00:00
}
2019-10-27 05:53:25 +00:00
2024-09-06 21:15:19 +00:00
if len ( agentConfig . NodeInternalDNSs ) > 0 {
result [ cp . InternalDNSKey ] = strings . Join ( agentConfig . NodeInternalDNSs , "," )
} else {
delete ( result , cp . InternalDNSKey )
}
if len ( agentConfig . NodeExternalDNSs ) > 0 {
result [ cp . ExternalDNSKey ] = strings . Join ( agentConfig . NodeExternalDNSs , "," )
} else {
delete ( result , cp . ExternalDNSKey )
}
2021-04-21 22:56:20 +00:00
result = labels . Merge ( nodeAnnotations , result )
return result , ! equality . Semantic . DeepEqual ( nodeAnnotations , result )
2019-10-15 21:17:26 +00:00
}
2021-02-12 15:35:57 +00:00
// setupTunnelAndRunAgent should start the setup tunnel before starting kubelet and kubeproxy
// there are special case for etcd agents, it will wait until it can find the apiaddress from
// the address channel and update the proxy with the servers addresses, if in rke2 we need to
// start the agent before the tunnel is setup to allow kubelet to start first and start the pods
func setupTunnelAndRunAgent ( ctx context . Context , nodeConfig * daemonconfig . Node , cfg cmds . Agent , proxy proxy . Proxy ) error {
var agentRan bool
2021-07-27 21:56:05 +00:00
// IsAPIServerLBEnabled is used as a shortcut for detecting RKE2, where the kubelet needs to
// be run earlier in order to manage static pods. This should probably instead query a
// flag on the executor or something.
2022-04-27 20:44:15 +00:00
if ! cfg . ClusterReset && cfg . ETCDAgent {
2021-07-27 21:56:05 +00:00
// ETCDAgent is only set to true on servers that are started with --disable-apiserver.
// In this case, we may be running without an apiserver available in the cluster, and need
// to wait for one to register and post it's address into APIAddressCh so that we can update
// the LB proxy with its address.
2021-02-12 15:35:57 +00:00
if proxy . IsAPIServerLBEnabled ( ) {
2021-07-27 21:56:05 +00:00
// On RKE2, the agent needs to be started early to run the etcd static pod.
if err := agent . Agent ( ctx , nodeConfig , proxy ) ; err != nil {
2021-02-12 15:35:57 +00:00
return err
}
agentRan = true
}
2022-02-16 22:19:58 +00:00
if err := waitForAPIServerAddresses ( ctx , nodeConfig , cfg , proxy ) ; err != nil {
return err
2021-02-12 15:35:57 +00:00
}
2021-05-05 15:40:04 +00:00
} else if cfg . ClusterReset && proxy . IsAPIServerLBEnabled ( ) {
2021-07-27 21:56:05 +00:00
// If we're doing a cluster-reset on RKE2, the kubelet needs to be started early to clean
// up static pods.
if err := agent . Agent ( ctx , nodeConfig , proxy ) ; err != nil {
2021-05-05 15:40:04 +00:00
return err
}
agentRan = true
2021-02-12 15:35:57 +00:00
}
2022-04-27 20:44:15 +00:00
if err := tunnelSetup ( ctx , nodeConfig , cfg , proxy ) ; err != nil {
2021-02-12 15:35:57 +00:00
return err
}
2024-03-23 00:13:08 +00:00
if err := certMonitorSetup ( ctx , nodeConfig , cfg ) ; err != nil {
return err
}
2021-02-12 15:35:57 +00:00
if ! agentRan {
2021-07-27 21:56:05 +00:00
return agent . Agent ( ctx , nodeConfig , proxy )
2021-02-12 15:35:57 +00:00
}
return nil
}
2022-02-16 22:19:58 +00:00
func waitForAPIServerAddresses ( ctx context . Context , nodeConfig * daemonconfig . Node , cfg cmds . Agent , proxy proxy . Proxy ) error {
2024-07-12 17:24:52 +00:00
var localSupervisorDefault bool
if addresses := proxy . SupervisorAddresses ( ) ; len ( addresses ) > 0 {
host , _ , _ := net . SplitHostPort ( addresses [ 0 ] )
if host == "127.0.0.1" || host == "::1" {
localSupervisorDefault = true
}
}
2022-02-16 22:19:58 +00:00
for {
select {
case <- time . After ( 5 * time . Second ) :
2024-07-12 17:24:52 +00:00
logrus . Info ( "Waiting for control-plane node to register apiserver addresses in etcd" )
2022-02-16 22:19:58 +00:00
case addresses := <- cfg . APIAddressCh :
for i , a := range addresses {
host , _ , err := net . SplitHostPort ( a )
if err == nil {
addresses [ i ] = net . JoinHostPort ( host , strconv . Itoa ( nodeConfig . ServerHTTPSPort ) )
}
}
2024-07-12 17:24:52 +00:00
// If this is an etcd-only node that started up using its local supervisor,
// switch to using a control-plane node as the supervisor. Otherwise, leave the
// configured server address as the default.
if localSupervisorDefault && len ( addresses ) > 0 {
proxy . SetSupervisorDefault ( addresses [ 0 ] )
}
2022-02-16 22:19:58 +00:00
proxy . Update ( addresses )
return nil
case <- ctx . Done ( ) :
return ctx . Err ( )
}
}
}
2022-04-27 20:44:15 +00:00
// tunnelSetup calls tunnel setup, unless the embedded etc cluster is being reset/restored, in which case
// this is unnecessary as the kubelet is only needed to manage static pods and does not need to establish
// tunneled connections to other cluster members.
func tunnelSetup ( ctx context . Context , nodeConfig * daemonconfig . Node , cfg cmds . Agent , proxy proxy . Proxy ) error {
if cfg . ClusterReset {
return nil
}
return tunnel . Setup ( ctx , nodeConfig , proxy )
}
2024-02-27 22:43:39 +00:00
2024-03-23 00:13:08 +00:00
func certMonitorSetup ( ctx context . Context , nodeConfig * daemonconfig . Node , cfg cmds . Agent ) error {
if cfg . ClusterReset {
return nil
}
return certmonitor . Setup ( ctx , nodeConfig , cfg . DataDir )
}
2024-02-27 22:43:39 +00:00
// getHostname returns the actual system hostname.
// If the hostname cannot be determined, or is invalid, the node name is used.
func getHostname ( agentConfig * daemonconfig . Agent ) string {
hostname , err := os . Hostname ( )
if err != nil || hostname == "" || strings . Contains ( hostname , "localhost" ) {
return agentConfig . NodeName
}
return hostname
}