2021-02-12 15:35:57 +00:00
|
|
|
package etcd
|
|
|
|
|
|
|
|
import (
|
2021-03-08 22:10:00 +00:00
|
|
|
"context"
|
2024-03-19 22:01:36 +00:00
|
|
|
"crypto/tls"
|
|
|
|
"fmt"
|
|
|
|
"net"
|
|
|
|
"net/http"
|
|
|
|
"strconv"
|
|
|
|
"time"
|
2021-02-12 15:35:57 +00:00
|
|
|
|
2022-03-02 23:47:27 +00:00
|
|
|
"github.com/k3s-io/k3s/pkg/agent/loadbalancer"
|
2024-03-19 22:01:36 +00:00
|
|
|
"github.com/sirupsen/logrus"
|
|
|
|
"k8s.io/apimachinery/pkg/util/wait"
|
2021-02-12 15:35:57 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
type Proxy interface {
|
|
|
|
Update(addresses []string)
|
|
|
|
}
|
|
|
|
|
2024-03-19 22:01:36 +00:00
|
|
|
var httpClient = &http.Client{
|
|
|
|
Transport: &http.Transport{
|
|
|
|
TLSClientConfig: &tls.Config{
|
|
|
|
InsecureSkipVerify: true,
|
|
|
|
},
|
|
|
|
},
|
|
|
|
}
|
|
|
|
|
2021-02-12 15:35:57 +00:00
|
|
|
// NewETCDProxy initializes a new proxy structure that contain a load balancer
|
|
|
|
// which listens on port 2379 and proxy between etcd cluster members
|
2024-03-19 22:01:36 +00:00
|
|
|
func NewETCDProxy(ctx context.Context, supervisorPort int, dataDir, etcdURL string, isIPv6 bool) (Proxy, error) {
|
|
|
|
lb, err := loadbalancer.New(ctx, dataDir, loadbalancer.ETCDServerServiceName, etcdURL, 2379, isIPv6)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
2021-02-12 15:35:57 +00:00
|
|
|
}
|
|
|
|
|
2024-11-20 18:53:09 +00:00
|
|
|
return &etcdproxy{
|
|
|
|
supervisorPort: supervisorPort,
|
|
|
|
etcdLB: lb,
|
|
|
|
disconnect: map[string]context.CancelFunc{},
|
|
|
|
}, nil
|
2021-02-12 15:35:57 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
type etcdproxy struct {
|
2024-11-20 18:53:09 +00:00
|
|
|
supervisorPort int
|
|
|
|
etcdLB *loadbalancer.LoadBalancer
|
|
|
|
disconnect map[string]context.CancelFunc
|
2021-02-12 15:35:57 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (e *etcdproxy) Update(addresses []string) {
|
2024-03-19 22:01:36 +00:00
|
|
|
e.etcdLB.Update(addresses)
|
|
|
|
|
|
|
|
validEndpoint := map[string]bool{}
|
|
|
|
for _, address := range e.etcdLB.ServerAddresses {
|
|
|
|
validEndpoint[address] = true
|
|
|
|
if _, ok := e.disconnect[address]; !ok {
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
|
|
e.disconnect[address] = cancel
|
|
|
|
e.etcdLB.SetHealthCheck(address, e.createHealthCheck(ctx, address))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
for address, cancel := range e.disconnect {
|
|
|
|
if !validEndpoint[address] {
|
|
|
|
cancel()
|
|
|
|
delete(e.disconnect, address)
|
|
|
|
}
|
2021-02-12 15:35:57 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2024-03-19 22:01:36 +00:00
|
|
|
// start a polling routine that makes periodic requests to the etcd node's supervisor port.
|
|
|
|
// If the request fails, the node is marked unhealthy.
|
|
|
|
func (e etcdproxy) createHealthCheck(ctx context.Context, address string) func() bool {
|
|
|
|
// Assume that the connection to the server will succeed, to avoid failing health checks while attempting to connect.
|
|
|
|
// If we cannot connect, connected will be set to false when the initial connection attempt fails.
|
|
|
|
connected := true
|
|
|
|
|
|
|
|
host, _, _ := net.SplitHostPort(address)
|
|
|
|
url := fmt.Sprintf("https://%s/ping", net.JoinHostPort(host, strconv.Itoa(e.supervisorPort)))
|
|
|
|
|
|
|
|
go wait.JitterUntilWithContext(ctx, func(ctx context.Context) {
|
|
|
|
ctx, cancel := context.WithTimeout(ctx, time.Second)
|
|
|
|
defer cancel()
|
|
|
|
req, _ := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
|
|
|
|
resp, err := httpClient.Do(req)
|
|
|
|
var statusCode int
|
|
|
|
if resp != nil {
|
|
|
|
statusCode = resp.StatusCode
|
|
|
|
}
|
|
|
|
if err != nil || statusCode != http.StatusOK {
|
2024-05-29 18:17:29 +00:00
|
|
|
logrus.Debugf("Health check %s failed: %v (StatusCode: %d)", address, err, statusCode)
|
2024-03-19 22:01:36 +00:00
|
|
|
connected = false
|
|
|
|
} else {
|
|
|
|
connected = true
|
|
|
|
}
|
|
|
|
}, 5*time.Second, 1.0, true)
|
|
|
|
|
|
|
|
return func() bool {
|
|
|
|
return connected
|
|
|
|
}
|
|
|
|
}
|