Skip to content

K
k3s
Commit a8bc4124 authored by Brad Davidson's avatar Brad Davidson Committed by Brad Davidson
Browse files
Options

Move container runtime ready channel into executor

Move the container runtime ready channel into the executor interface, instead of passing it awkwardly between server and agent config structs Signed-off-by: 's avatarBrad Davidson <brad.davidson@rancher.com>
parent 529e748a
Hide whitespace changes
Inline Side-by-side
Showing with 123 additions and 173 deletions
pkg/agent/cri/cri.go
View file @ a8bc4124
......@@ -5,10 +5,37 @@ import (
"time"
"github.com/sirupsen/logrus"
"google.golang.org/grpc"
runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1"
k8sutil "k8s.io/cri-client/pkg/util"
)
const maxMsgSize = 1024 * 1024 * 16
// Connection connects to a CRI socket at the given path.
func Connection(ctx context.Context, address string) (*grpc.ClientConn, error) {
addr, dialer, err := k8sutil.GetAddressAndDialer(socketPrefix + address)
if err != nil {
return nil, err
}
conn, err := grpc.Dial(addr, grpc.WithInsecure(), grpc.WithTimeout(3*time.Second), grpc.WithContextDialer(dialer), grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMsgSize)))
if err != nil {
return nil, err
}
c := runtimeapi.NewRuntimeServiceClient(conn)
_, err = c.Version(ctx, &runtimeapi.VersionRequest{
Version: "0.1.0",
})
if err != nil {
conn.Close()
return nil, err
}
return conn, nil
}
// WaitForService blocks in a retry loop until the CRI service
// is functional at the provided socket address. It will return only on success,
// or when the context is cancelled.
......
pkg/agent/cri/cri_linux.go
View file @ a8bc4124
......@@ -3,37 +3,4 @@
package cri
import (
"context"
"time"
"google.golang.org/grpc"
runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1"
k8sutil "k8s.io/cri-client/pkg/util"
)
const socketPrefix = "unix://"
// Connection connects to a CRI socket at the given path.
func Connection(ctx context.Context, address string) (*grpc.ClientConn, error) {
addr, dialer, err := k8sutil.GetAddressAndDialer(socketPrefix + address)
if err != nil {
return nil, err
}
conn, err := grpc.Dial(addr, grpc.WithInsecure(), grpc.WithTimeout(3*time.Second), grpc.WithContextDialer(dialer), grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMsgSize)))
if err != nil {
return nil, err
}
c := runtimeapi.NewRuntimeServiceClient(conn)
_, err = c.Version(ctx, &runtimeapi.VersionRequest{
Version: "0.1.0",
})
if err != nil {
conn.Close()
return nil, err
}
return conn, nil
}
pkg/agent/cri/cri_windows.go
View file @ a8bc4124
......@@ -3,35 +3,4 @@
package cri
import (
"context"
"time"
"google.golang.org/grpc"
runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1"
"k8s.io/cri-client/pkg/util"
)
// Connection connects to a CRI socket at the given path.
func Connection(ctx context.Context, address string) (*grpc.ClientConn, error) {
addr, dialer, err := util.GetAddressAndDialer(address)
if err != nil {
return nil, err
}
conn, err := grpc.Dial(addr, grpc.WithInsecure(), grpc.WithTimeout(3*time.Second), grpc.WithContextDialer(dialer), grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMsgSize)))
if err != nil {
return nil, err
}
c := runtimeapi.NewRuntimeServiceClient(conn)
_, err = c.Version(ctx, &runtimeapi.VersionRequest{
Version: "0.1.0",
})
if err != nil {
conn.Close()
return nil, err
}
return conn, nil
}
const socketPrefix = "npipe://"
pkg/agent/run.go
View file @ a8bc4124
......@@ -158,13 +158,10 @@ func run(ctx context.Context, cfg cmds.Agent, proxy proxy.Proxy) error {
if err := executor.Containerd(ctx, nodeConfig); err != nil {
return err
}
}
// the container runtime is ready to host workloads when containerd is up and the airgap
// 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.
if cfg.ContainerRuntimeReady != nil {
close(cfg.ContainerRuntimeReady)
} else {
if err := executor.CRI(ctx, nodeConfig); err != nil {
return err
}
}
notifySocket := os.Getenv("NOTIFY_SOCKET")
......@@ -282,8 +279,9 @@ func RunStandalone(ctx context.Context, cfg cmds.Agent) error {
return err
}
if cfg.ContainerRuntimeReady != nil {
close(cfg.ContainerRuntimeReady)
// this is a no-op just to get the cri ready channel closed
if err := executor.CRI(ctx, nodeConfig); err != nil {
return err
}
if err := tunnelSetup(ctx, nodeConfig, cfg, proxy); err != nil {
......@@ -513,53 +511,34 @@ func updateAddressAnnotations(nodeConfig *daemonconfig.Node, nodeAnnotations map
return result, !equality.Semantic.DeepEqual(nodeAnnotations, result)
}
// 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
// setupTunnelAndRunAgent starts the agent tunnel, cert expiry monitoring, and
// runs the Agent (cri+kubelet). On etcd-only nodes, an extra goroutine is
// started to retrieve apiserver addresses from the datastore. On other node
// types, this is done later by the tunnel setup, which starts goroutines to
// watch apiserver endpoints.
func setupTunnelAndRunAgent(ctx context.Context, nodeConfig *daemonconfig.Node, cfg cmds.Agent, proxy proxy.Proxy) error {
var agentRan bool
// 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.
// only need to get apiserver addresses from the datastore on an etcd-only node that is not being reset
if !cfg.ClusterReset && cfg.ETCDAgent {
// 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.
if proxy.IsAPIServerLBEnabled() {
// On RKE2, the agent needs to be started early to run the etcd static pod.
if err := agent.Agent(ctx, nodeConfig, proxy); err != nil {
return err
}
agentRan = true
}
if err := waitForAPIServerAddresses(ctx, nodeConfig, cfg, proxy); err != nil {
return err
}
} else if cfg.ClusterReset && proxy.IsAPIServerLBEnabled() {
// 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 {
return err
}
agentRan = true
go waitForAPIServerAddresses(ctx, nodeConfig, cfg, proxy)
}
if err := tunnelSetup(ctx, nodeConfig, cfg, proxy); err != nil {
return err
}
if err := certMonitorSetup(ctx, nodeConfig, cfg); err != nil {
return err
}
if !agentRan {
return agent.Agent(ctx, nodeConfig, proxy)
}
return nil
return agent.Agent(ctx, nodeConfig, proxy)
}
func waitForAPIServerAddresses(ctx context.Context, nodeConfig *daemonconfig.Node, cfg cmds.Agent, proxy proxy.Proxy) error {
// waitForAPIServerAddresses syncs apiserver addresses from the datastore. This
// is also handled by the agent tunnel watch, but on etcd-only nodes we need to
// read apiserver addresses from APIAddressCh before the agent has a
// connection to the apiserver. This does not return until addresses or set,
// or the context is cancelled.
func waitForAPIServerAddresses(ctx context.Context, nodeConfig *daemonconfig.Node, cfg cmds.Agent, proxy proxy.Proxy) {
var localSupervisorDefault bool
if addresses := proxy.SupervisorAddresses(); len(addresses) > 0 {
host, _, _ := net.SplitHostPort(addresses[0])
......@@ -586,9 +565,9 @@ func waitForAPIServerAddresses(ctx context.Context, nodeConfig *daemonconfig.Nod
proxy.SetSupervisorDefault(addresses[0])
}
proxy.Update(addresses)
return nil
return
case <-ctx.Done():
return ctx.Err()
return
}
}
}
......
pkg/agent/tunnel/tunnel.go
View file @ a8bc4124
......@@ -8,7 +8,6 @@ import (
"net"
"os"
"strconv"
"sync"
"time"
"github.com/gorilla/websocket"
......@@ -49,6 +48,7 @@ var (
type agentTunnel struct {
client kubernetes.Interface
tlsConfig *tls.Config
cidrs cidranger.Ranger
ports map[string]bool
mode string
......@@ -69,6 +69,8 @@ func (p *podEntry) Network() net.IPNet {
return p.cidr
}
// Setup sets up the agent tunnel, which is reponsible for connecting websocket tunnels to
// control-plane nodes, syncing endpoints for the tunnel authorizer, and updating proxy endpoints.
func Setup(ctx context.Context, config *daemonconfig.Node, proxy proxy.Proxy) error {
client, err := util.GetClientSet(config.AgentConfig.KubeConfigK3sController)
if err != nil {
......@@ -87,6 +89,7 @@ func Setup(ctx context.Context, config *daemonconfig.Node, proxy proxy.Proxy) er
tunnel := &agentTunnel{
client: client,
tlsConfig: tlsConfig,
cidrs: cidranger.NewPCTrieRanger(),
ports: map[string]bool{},
mode: config.EgressSelectorMode,
......@@ -95,6 +98,14 @@ func Setup(ctx context.Context, config *daemonconfig.Node, proxy proxy.Proxy) er
startTime: time.Now().Truncate(time.Second),
}
go tunnel.startWatches(ctx, config, proxy)
return nil
}
// startWatches starts watching for changes to endpoints, both for the tunnel authorizer,
// and to sync supervisor addresses into the proxy. This will block until the context is cancelled.
func (a *agentTunnel) startWatches(ctx context.Context, config *daemonconfig.Node, proxy proxy.Proxy) {
rbacReady := make(chan struct{})
go func() {
<-executor.APIServerReadyChan()
......@@ -113,15 +124,15 @@ func Setup(ctx context.Context, config *daemonconfig.Node, proxy proxy.Proxy) er
// signifying that this is an agentless server that will not register a node.
if config.ContainerRuntimeEndpoint != "/dev/null" {
// Allow the kubelet port, as published via our node object.
go tunnel.setKubeletPort(ctx, rbacReady)
go a.setKubeletPort(ctx, rbacReady)
switch tunnel.mode {
switch a.mode {
case daemonconfig.EgressSelectorModeCluster:
// In Cluster mode, we allow the cluster CIDRs, and any connections to the node's IPs for pods using host network.
tunnel.clusterAuth(config)
a.clusterAuth(config)
case daemonconfig.EgressSelectorModePod:
// In Pod mode, we watch pods assigned to this node, and allow their addresses, as well as ports used by containers with host network.
go tunnel.watchPods(ctx, rbacReady, config)
go a.watchPods(ctx, rbacReady, config)
}
}
......@@ -150,7 +161,7 @@ func Setup(ctx context.Context, config *daemonconfig.Node, proxy proxy.Proxy) er
}
proxy.Update(addresses)
} else {
if endpoint, err := client.CoreV1().Endpoints(metav1.NamespaceDefault).Get(ctx, "kubernetes", metav1.GetOptions{}); err != nil {
if endpoint, err := a.client.CoreV1().Endpoints(metav1.NamespaceDefault).Get(ctx, "kubernetes", metav1.GetOptions{}); err != nil {
logrus.Errorf("Failed to get apiserver addresses from kubernetes endpoints: %v", err)
} else {
addresses := util.GetAddresses(endpoint)
......@@ -162,24 +173,7 @@ func Setup(ctx context.Context, config *daemonconfig.Node, proxy proxy.Proxy) er
}
}
wg := &sync.WaitGroup{}
go tunnel.watchEndpoints(ctx, rbacReady, wg, tlsConfig, config, proxy)
wait := make(chan int, 1)
go func() {
wg.Wait()
wait <- 0
}()
select {
case <-ctx.Done():
logrus.Error("Tunnel context canceled while waiting for connection")
return ctx.Err()
case <-wait:
}
return nil
a.watchEndpoints(ctx, rbacReady, config, proxy)
}
// setKubeletPort retrieves the configured kubelet port from our node object
......@@ -307,8 +301,8 @@ func (a *agentTunnel) watchPods(ctx context.Context, rbacReady <-chan struct{},
// WatchEndpoints attempts to create tunnels to all supervisor addresses. Once the
// apiserver is up, go into a watch loop, adding and removing tunnels as endpoints come
// and go from the cluster.
func (a *agentTunnel) watchEndpoints(ctx context.Context, rbacReady <-chan struct{}, wg *sync.WaitGroup, tlsConfig *tls.Config, node *daemonconfig.Node, proxy proxy.Proxy) {
syncProxyAddresses := a.getProxySyncer(ctx, wg, tlsConfig, proxy)
func (a *agentTunnel) watchEndpoints(ctx context.Context, rbacReady <-chan struct{}, node *daemonconfig.Node, proxy proxy.Proxy) {
syncProxyAddresses := a.getProxySyncer(ctx, proxy)
refreshFromSupervisor := getAPIServersRequester(node, proxy, syncProxyAddresses)
<-rbacReady
......@@ -395,17 +389,12 @@ type agentConnection struct {
// connect initiates a connection to the remotedialer server. Incoming dial requests from
// the server will be checked by the authorizer function prior to being fulfilled.
func (a *agentTunnel) connect(rootCtx context.Context, waitGroup *sync.WaitGroup, address string, tlsConfig *tls.Config) agentConnection {
func (a *agentTunnel) connect(rootCtx context.Context, address string) agentConnection {
var status loadbalancer.HealthCheckResult
wsURL := fmt.Sprintf("wss://%s/v1-"+version.Program+"/connect", address)
ws := &websocket.Dialer{
TLSClientConfig: tlsConfig,
}
once := sync.Once{}
if waitGroup != nil {
waitGroup.Add(1)
TLSClientConfig: a.tlsConfig,
}
ctx, cancel := context.WithCancel(rootCtx)
......@@ -416,9 +405,6 @@ func (a *agentTunnel) connect(rootCtx context.Context, waitGroup *sync.WaitGroup
onConnect := func(_ context.Context, _ *remotedialer.Session) error {
status = loadbalancer.HealthCheckResultOK
logrus.WithField("url", wsURL).Info("Remotedialer connected to proxy")
if waitGroup != nil {
once.Do(waitGroup.Done)
}
return nil
}
......@@ -435,9 +421,6 @@ func (a *agentTunnel) connect(rootCtx context.Context, waitGroup *sync.WaitGroup
}
// If the context has been cancelled, exit the goroutine instead of retrying
if ctx.Err() != nil {
if waitGroup != nil {
once.Do(waitGroup.Done)
}
return
}
}
......@@ -476,13 +459,13 @@ type proxySyncer func(addresses []string)
// getProxySyncer returns a function that can be called to update the list of supervisors.
// This function is responsible for connecting to or disconnecting websocket tunnels,
// as well as updating the proxy loadbalancer server list.
func (a *agentTunnel) getProxySyncer(ctx context.Context, wg *sync.WaitGroup, tlsConfig *tls.Config, proxy proxy.Proxy) proxySyncer {
func (a *agentTunnel) getProxySyncer(ctx context.Context, proxy proxy.Proxy) proxySyncer {
disconnect := map[string]context.CancelFunc{}
// Attempt to connect to inital list of addresses, storing their cancellation
// function for later when we need to disconnect.
for _, address := range proxy.SupervisorAddresses() {
if _, ok := disconnect[address]; !ok {
conn := a.connect(ctx, wg, address, tlsConfig)
conn := a.connect(ctx, address)
disconnect[address] = conn.cancel
proxy.SetHealthCheck(address, conn.healthCheck)
}
......@@ -538,7 +521,7 @@ func (a *agentTunnel) getProxySyncer(ctx context.Context, wg *sync.WaitGroup, tl
// add new servers
for address := range addedAddresses {
if _, ok := disconnect[address]; !ok {
conn := a.connect(ctx, nil, address, tlsConfig)
conn := a.connect(ctx, address)
logrus.Infof("Started tunnel to %s", address)
disconnect[address] = conn.cancel
proxy.SetHealthCheck(address, conn.healthCheck)
......
pkg/cli/cert/cert.go
View file @ a8bc4124
......@@ -46,7 +46,7 @@ func commandSetup(app *cli.Context, cfg *cmds.Server, sc *server.Config) (string
cfg.Token = string(bytes.TrimRight(tokenByte, "\n"))
}
sc.ControlConfig.Token = cfg.Token
sc.ControlConfig.Runtime = config.NewRuntime(nil)
sc.ControlConfig.Runtime = config.NewRuntime()
return dataDir, nil
}
......@@ -287,7 +287,7 @@ func rotateCA(app *cli.Context, cfg *cmds.Server, sync *cmds.CertRotateCA) error
// Set up dummy server config for reading new bootstrap data from disk.
tmpServer := &config.Control{
Runtime: config.NewRuntime(nil),
Runtime: config.NewRuntime(),
DataDir: sync.CACertPath,
}
deps.CreateRuntimeCertFiles(tmpServer)
......
pkg/cli/cmds/agent.go
View file @ a8bc4124
......@@ -56,7 +56,6 @@ type Agent struct {
Taints cli.StringSlice
ImageCredProvBinDir string
ImageCredProvConfig string
ContainerRuntimeReady chan<- struct{}
AgentShared
}
......
pkg/cli/server/server.go
View file @ a8bc4124
......@@ -113,11 +113,9 @@ func run(app *cli.Context, cfg *cmds.Server, leaderControllers server.CustomCont
}
}
containerRuntimeReady := make(chan struct{})
serverConfig := server.Config{}
serverConfig.DisableAgent = cfg.DisableAgent
serverConfig.ControlConfig.Runtime = config.NewRuntime(containerRuntimeReady)
serverConfig.ControlConfig.Runtime = config.NewRuntime()
serverConfig.ControlConfig.Token = cfg.Token
serverConfig.ControlConfig.AgentToken = cfg.AgentToken
serverConfig.ControlConfig.JoinURL = cfg.ServerURL
......@@ -525,7 +523,6 @@ func run(app *cli.Context, cfg *cmds.Server, leaderControllers server.CustomCont
}
agentConfig := cmds.AgentConfig
agentConfig.ContainerRuntimeReady = containerRuntimeReady
agentConfig.Debug = app.Bool("debug")
agentConfig.DataDir = filepath.Dir(serverConfig.ControlConfig.DataDir)
agentConfig.ServerURL = url
......
pkg/daemons/config/types.go
View file @ a8bc4124
......@@ -314,7 +314,6 @@ type ControlRuntimeBootstrap struct {
type ControlRuntime struct {
ControlRuntimeBootstrap
ContainerRuntimeReady <-chan struct{}
ETCDReady <-chan struct{}
StartupHooksWg *sync.WaitGroup
ClusterControllerStarts map[string]leader.Callback
......@@ -394,9 +393,8 @@ type CoreFactory interface {
Start(ctx context.Context, defaultThreadiness int) error
}
func NewRuntime(containerRuntimeReady <-chan struct{}) *ControlRuntime {
func NewRuntime() *ControlRuntime {
return &ControlRuntime{
ContainerRuntimeReady: containerRuntimeReady,
ClusterControllerStarts: map[string]leader.Callback{},
LeaderElectedClusterControllerStarts: map[string]leader.Callback{},
}
......
pkg/daemons/executor/embed.go
View file @ a8bc4124
......@@ -14,6 +14,7 @@ import (
"time"
"github.com/k3s-io/k3s/pkg/agent/containerd"
"github.com/k3s-io/k3s/pkg/agent/cri"
"github.com/k3s-io/k3s/pkg/agent/cridockerd"
"github.com/k3s-io/k3s/pkg/cli/cmds"
daemonconfig "github.com/k3s-io/k3s/pkg/daemons/config"
......@@ -44,6 +45,7 @@ func init() {
func (e *Embedded) Bootstrap(ctx context.Context, nodeConfig *daemonconfig.Node, cfg cmds.Agent) error {
e.apiServerReady = util.APIServerReadyChan(ctx, nodeConfig.AgentConfig.KubeConfigK3sController, util.DefaultAPIServerReadyTimeout)
e.criReady = make(chan struct{})
e.nodeConfig = nodeConfig
go func() {
......@@ -235,16 +237,34 @@ func (e *Embedded) CurrentETCDOptions() (InitialOptions, error) {
}
func (e *Embedded) Containerd(ctx context.Context, cfg *daemonconfig.Node) error {
defer close(e.criReady)
return containerd.Run(ctx, cfg)
}
func (e *Embedded) Docker(ctx context.Context, cfg *daemonconfig.Node) error {
defer close(e.criReady)
return cridockerd.Run(ctx, cfg)
}
func (e *Embedded) CRI(ctx context.Context, cfg *daemonconfig.Node) error {
defer close(e.criReady)
// agentless sets cri socket path to /dev/null to indicate no CRI is needed
if cfg.ContainerRuntimeEndpoint != "/dev/null" {
return cri.WaitForService(ctx, cfg.ContainerRuntimeEndpoint, "CRI")
}
return nil
}
func (e *Embedded) APIServerReadyChan() <-chan struct{} {
if e.apiServerReady == nil {
panic("executor not bootstrapped")
}
return e.apiServerReady
}
func (e *Embedded) CRIReadyChan() <-chan struct{} {
if e.criReady == nil {
panic("executor not bootstrapped")
}
return e.criReady
}
pkg/daemons/executor/etcd.go
View file @ a8bc4124
......@@ -17,6 +17,7 @@ import (
// of the embedded execututor is disabled by build flags
type Embedded struct {
apiServerReady <-chan struct{}
criReady chan struct{}
nodeConfig *daemonconfig.Node
}
......
pkg/daemons/executor/executor.go
View file @ a8bc4124
......@@ -33,7 +33,9 @@ type Executor interface {
CloudControllerManager(ctx context.Context, ccmRBACReady <-chan struct{}, args []string) error
Containerd(ctx context.Context, node *daemonconfig.Node) error
Docker(ctx context.Context, node *daemonconfig.Node) error
CRI(ctx context.Context, node *daemonconfig.Node) error
APIServerReadyChan() <-chan struct{}
CRIReadyChan() <-chan struct{}
}
type ETCDConfig struct {
......@@ -183,6 +185,14 @@ func Docker(ctx context.Context, config *daemonconfig.Node) error {
return executor.Docker(ctx, config)
}
func CRI(ctx context.Context, config *daemonconfig.Node) error {
return executor.CRI(ctx, config)
}
func APIServerReadyChan() <-chan struct{} {
return executor.APIServerReadyChan()
}
func CRIReadyChan() <-chan struct{} {
return executor.CRIReadyChan()
}
pkg/etcd/etcd.go
View file @ a8bc4124
......@@ -343,7 +343,7 @@ func (e *ETCD) IsInitialized() (bool, error) {
func (e *ETCD) Reset(ctx context.Context, rebootstrap func() error) error {
// Wait for etcd to come up as a new single-node cluster, then exit
go func() {
<-e.config.Runtime.ContainerRuntimeReady
<-executor.CRIReadyChan()
t := time.NewTicker(5 * time.Second)
defer t.Stop()
for range t.C {
......@@ -497,7 +497,7 @@ func (e *ETCD) Start(ctx context.Context, clientAccessInfo *clientaccess.Info) e
select {
case <-time.After(30 * time.Second):
logrus.Infof("Waiting for container runtime to become ready before joining etcd cluster")
case <-e.config.Runtime.ContainerRuntimeReady:
case <-executor.CRIReadyChan():
if err := wait.PollUntilContextCancel(ctx, time.Second, true, func(ctx context.Context) (bool, error) {
if err := e.join(ctx, clientAccessInfo); err != nil {
// Retry the join if waiting for another member to be promoted, or waiting for peers to connect after promotion
......@@ -726,7 +726,6 @@ func (e *ETCD) handler(next http.Handler) http.Handler {
}
// infoHandler returns etcd cluster information. This is used by new members when joining the cluster.
// If we can't retrieve an actual MemberList from etcd, we return a canned response with only the local node listed.
func (e *ETCD) infoHandler() http.Handler {
return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
if req.Method != http.MethodGet {
......@@ -734,6 +733,11 @@ func (e *ETCD) infoHandler() http.Handler {
return
}
if e.client == nil {
util.SendError(errors.New("failed to get etcd MemberList: etcd not started"), rw, req, http.StatusInternalServerError)
return
}
ctx, cancel := context.WithTimeout(req.Context(), 2*time.Second)
defer cancel()
......@@ -1157,7 +1161,7 @@ func (e *ETCD) RemovePeer(ctx context.Context, name, address string, allowSelfRe
// being promoted to full voting member. The checks only run on the cluster member that is
// the etcd leader.
func (e *ETCD) manageLearners(ctx context.Context) {
<-e.config.Runtime.ContainerRuntimeReady
<-executor.CRIReadyChan()
t := time.NewTicker(manageTickerTime)
defer t.Stop()
......
pkg/etcd/etcd_linux_test.go
View file @ a8bc4124
......@@ -61,7 +61,7 @@ func generateTestConfig() *config.Control {
}
return &config.Control{
ServerNodeName: hostname,
Runtime: config.NewRuntime(containerRuntimeReady),
Runtime: config.NewRuntime(),
HTTPSPort: 6443,
SupervisorPort: 6443,
AdvertisePort: 6443,
......
pkg/server/handlers/cert.go
View file @ a8bc4124
......@@ -56,7 +56,7 @@ func caCertReplace(control *config.Control, buf io.ReadCloser, force bool) error
}
defer os.RemoveAll(tmpdir)
runtime := config.NewRuntime(nil)
runtime := config.NewRuntime()
runtime.EtcdConfig = control.Runtime.EtcdConfig
runtime.ServerToken = control.Runtime.ServerToken
......
tests/unit.go
View file @ a8bc4124
......@@ -44,11 +44,7 @@ func CleanupDataDir(cnf *config.Control) {
// config.ControlRuntime with all the appropriate certificate keys.
func GenerateRuntime(cnf *config.Control) error {
// reuse ready channel from existing runtime if set
var readyCh <-chan struct{}
if cnf.Runtime != nil {
readyCh = cnf.Runtime.ContainerRuntimeReady
}
cnf.Runtime = config.NewRuntime(readyCh)
cnf.Runtime = config.NewRuntime()
if err := GenerateDataDir(cnf); err != nil {
return err
}
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment