mirror of https://github.com/k3d-io/k3d
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
253 lines
9.5 KiB
253 lines
9.5 KiB
/*
|
|
Copyright © 2020-2023 The k3d Author(s)
|
|
|
|
Permission is hereby granted, free of charge, to any person obtaining a copy
|
|
of this software and associated documentation files (the "Software"), to deal
|
|
in the Software without restriction, including without limitation the rights
|
|
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
|
copies of the Software, and to permit persons to whom the Software is
|
|
furnished to do so, subject to the following conditions:
|
|
|
|
The above copyright notice and this permission notice shall be included in
|
|
all copies or substantial portions of the Software.
|
|
|
|
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
|
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
|
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
|
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
|
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
|
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
|
THE SOFTWARE.
|
|
*/
|
|
package client
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/docker/go-connections/nat"
|
|
"github.com/go-test/deep"
|
|
"github.com/imdario/mergo"
|
|
"github.com/spf13/viper"
|
|
"sigs.k8s.io/yaml"
|
|
|
|
l "github.com/k3d-io/k3d/v5/pkg/logger"
|
|
"github.com/k3d-io/k3d/v5/pkg/runtimes"
|
|
k3d "github.com/k3d-io/k3d/v5/pkg/types"
|
|
)
|
|
|
|
var (
|
|
ErrLBConfigHostNotFound error = errors.New("lbconfig: host not found")
|
|
ErrLBConfigFailedTest error = errors.New("lbconfig: failed to test")
|
|
ErrLBConfigEntryExists error = errors.New("lbconfig: entry exists in config")
|
|
)
|
|
|
|
// UpdateLoadbalancerConfig updates the loadbalancer config with an updated list of servers belonging to that cluster
|
|
func UpdateLoadbalancerConfig(ctx context.Context, runtime runtimes.Runtime, cluster *k3d.Cluster) error {
|
|
var err error
|
|
// update cluster details to ensure that we have the latest node list
|
|
cluster, err = ClusterGet(ctx, runtime, cluster)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to update details for cluster '%s': %w", cluster.Name, err)
|
|
}
|
|
|
|
currentConfig, err := GetLoadbalancerConfig(ctx, runtime, cluster)
|
|
if err != nil {
|
|
return fmt.Errorf("error getting current config from loadbalancer: %w", err)
|
|
}
|
|
|
|
l.Log().Tracef("Current loadbalancer config:\n%+v", currentConfig)
|
|
|
|
newLBConfig, err := LoadbalancerGenerateConfig(cluster)
|
|
if err != nil {
|
|
return fmt.Errorf("error generating new loadbalancer config: %w", err)
|
|
}
|
|
l.Log().Tracef("New loadbalancer config:\n%+v", currentConfig)
|
|
|
|
if diff := deep.Equal(currentConfig, newLBConfig); diff != nil {
|
|
l.Log().Debugf("Updating the loadbalancer with this diff: %+v", diff)
|
|
}
|
|
|
|
newLbConfigYaml, err := yaml.Marshal(&newLBConfig)
|
|
if err != nil {
|
|
return fmt.Errorf("error marshalling the new loadbalancer config: %w", err)
|
|
}
|
|
l.Log().Debugf("Writing lb config:\n%s", string(newLbConfigYaml))
|
|
startTime := time.Now().Truncate(time.Second).UTC()
|
|
if err := runtime.WriteToNode(ctx, newLbConfigYaml, k3d.DefaultLoadbalancerConfigPath, 0744, cluster.ServerLoadBalancer.Node); err != nil {
|
|
return fmt.Errorf("error writing new loadbalancer config to container: %w", err)
|
|
}
|
|
|
|
successCtx, successCtxCancel := context.WithDeadline(ctx, time.Now().Add(5*time.Second))
|
|
defer successCtxCancel()
|
|
err = NodeWaitForLogMessage(successCtx, runtime, cluster.ServerLoadBalancer.Node, k3d.GetReadyLogMessage(cluster.ServerLoadBalancer.Node, k3d.IntentAny), startTime)
|
|
if err != nil {
|
|
if errors.Is(err, context.DeadlineExceeded) {
|
|
failureCtx, failureCtxCancel := context.WithDeadline(ctx, time.Now().Add(5*time.Second))
|
|
defer failureCtxCancel()
|
|
err = NodeWaitForLogMessage(failureCtx, runtime, cluster.ServerLoadBalancer.Node, "host not found in upstream", startTime)
|
|
if err != nil {
|
|
l.Log().Warnf("Failed to check if the loadbalancer was configured correctly or if it broke. Please check it manually or try again: %v", err)
|
|
return ErrLBConfigFailedTest
|
|
} else {
|
|
l.Log().Warnln("Failed to configure loadbalancer because one of the nodes seems to be down! Run `k3d node list` to see which one it could be.")
|
|
return ErrLBConfigHostNotFound
|
|
}
|
|
} else {
|
|
l.Log().Warnf("Failed to ensure that loadbalancer was configured correctly. Please check it manually or try again: %v", err)
|
|
return ErrLBConfigFailedTest
|
|
}
|
|
}
|
|
l.Log().Infof("Successfully configured loadbalancer %s!", cluster.ServerLoadBalancer.Node.Name)
|
|
|
|
time.Sleep(1 * time.Second) // waiting for a second, to avoid issues with too fast lb updates which would screw up the log waits
|
|
|
|
return nil
|
|
}
|
|
|
|
func GetLoadbalancerConfig(ctx context.Context, runtime runtimes.Runtime, cluster *k3d.Cluster) (k3d.LoadbalancerConfig, error) {
|
|
var cfg k3d.LoadbalancerConfig
|
|
|
|
if cluster.ServerLoadBalancer == nil || cluster.ServerLoadBalancer.Node == nil {
|
|
cluster.ServerLoadBalancer = &k3d.Loadbalancer{}
|
|
for _, node := range cluster.Nodes {
|
|
if node.Role == k3d.LoadBalancerRole {
|
|
var err error
|
|
cluster.ServerLoadBalancer.Node, err = NodeGet(ctx, runtime, node)
|
|
if err != nil {
|
|
return cfg, fmt.Errorf("failed to get loadbalancer node '%s': %w", node.Name, err)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
reader, err := runtime.ReadFromNode(ctx, k3d.DefaultLoadbalancerConfigPath, cluster.ServerLoadBalancer.Node)
|
|
if err != nil {
|
|
return cfg, fmt.Errorf("runtime failed to read loadbalancer config '%s' from node '%s': %w", k3d.DefaultLoadbalancerConfigPath, cluster.ServerLoadBalancer.Node.Name, err)
|
|
}
|
|
defer reader.Close()
|
|
|
|
file, err := io.ReadAll(reader)
|
|
if err != nil {
|
|
return cfg, fmt.Errorf("failed to read loadbalancer config file: %w", err)
|
|
}
|
|
|
|
file = bytes.Trim(file[512:], "\x00") // trim control characters, etc.
|
|
|
|
if err := yaml.Unmarshal(file, &cfg); err != nil {
|
|
return cfg, fmt.Errorf("error unmarshalling loadbalancer config: %w", err)
|
|
}
|
|
|
|
return cfg, nil
|
|
}
|
|
|
|
func LoadbalancerGenerateConfig(cluster *k3d.Cluster) (k3d.LoadbalancerConfig, error) {
|
|
lbConfig := k3d.LoadbalancerConfig{
|
|
Ports: map[string][]string{},
|
|
Settings: k3d.LoadBalancerSettings{},
|
|
}
|
|
|
|
// get list of server nodes
|
|
servers := []string{}
|
|
for _, node := range cluster.Nodes {
|
|
if node.Role == k3d.ServerRole {
|
|
servers = append(servers, node.Name)
|
|
}
|
|
}
|
|
|
|
// Default API Port proxied to the server nodes
|
|
lbConfig.Ports[fmt.Sprintf("%s.tcp", k3d.DefaultAPIPort)] = servers
|
|
|
|
// generate comma-separated list of extra ports to forward // TODO: no default targets?
|
|
for exposedPort := range cluster.ServerLoadBalancer.Node.Ports {
|
|
// TODO: catch duplicates here?
|
|
lbConfig.Ports[fmt.Sprintf("%s.%s", exposedPort.Port(), exposedPort.Proto())] = servers
|
|
}
|
|
|
|
// some additional nginx settings
|
|
lbConfig.Settings.WorkerConnections = k3d.DefaultLoadbalancerWorkerConnections + len(cluster.ServerLoadBalancer.Node.Ports)*len(servers)
|
|
|
|
return lbConfig, nil
|
|
}
|
|
|
|
func LoadbalancerPrepare(ctx context.Context, runtime runtimes.Runtime, cluster *k3d.Cluster, opts *k3d.LoadbalancerCreateOpts) (*k3d.Node, error) {
|
|
labels := map[string]string{}
|
|
|
|
if opts != nil && opts.Labels == nil && len(opts.Labels) == 0 {
|
|
labels = opts.Labels
|
|
}
|
|
|
|
if cluster.ServerLoadBalancer.Node.Ports == nil {
|
|
cluster.ServerLoadBalancer.Node.Ports = nat.PortMap{}
|
|
}
|
|
cluster.ServerLoadBalancer.Node.Ports[k3d.DefaultAPIPort] = []nat.PortBinding{cluster.KubeAPI.Binding}
|
|
|
|
if cluster.ServerLoadBalancer.Config == nil {
|
|
cluster.ServerLoadBalancer.Config = &k3d.LoadbalancerConfig{
|
|
Ports: map[string][]string{},
|
|
}
|
|
}
|
|
|
|
if opts != nil && opts.ConfigOverrides != nil && len(opts.ConfigOverrides) > 0 {
|
|
tmpViper := viper.New()
|
|
for _, override := range opts.ConfigOverrides {
|
|
kv := strings.SplitN(override, "=", 2)
|
|
l.Log().Tracef("Overriding LB config with %s...", kv)
|
|
tmpViper.Set(kv[0], kv[1])
|
|
}
|
|
lbConfigOverride := &k3d.LoadbalancerConfig{}
|
|
if err := tmpViper.Unmarshal(lbConfigOverride); err != nil {
|
|
return nil, fmt.Errorf("failed to unmarshal loadbalancer config override into loadbalancer config: %w", err)
|
|
}
|
|
if err := mergo.MergeWithOverwrite(cluster.ServerLoadBalancer.Config, lbConfigOverride); err != nil {
|
|
return nil, fmt.Errorf("failed to override loadbalancer config: %w", err)
|
|
}
|
|
}
|
|
|
|
// Create LB as a modified node with loadbalancerRole
|
|
lbNode := &k3d.Node{
|
|
Name: fmt.Sprintf("%s-%s-serverlb", k3d.DefaultObjectNamePrefix, cluster.Name),
|
|
Image: k3d.GetLoadbalancerImage(),
|
|
Ports: cluster.ServerLoadBalancer.Node.Ports,
|
|
Role: k3d.LoadBalancerRole,
|
|
RuntimeLabels: labels, // TODO: createLoadBalancer: add more expressive labels
|
|
Networks: []string{cluster.Network.Name},
|
|
Restart: true,
|
|
}
|
|
|
|
return lbNode, nil
|
|
}
|
|
|
|
func loadbalancerAddPortConfigs(loadbalancer *k3d.Loadbalancer, portmapping nat.PortMapping, targetNodes []*k3d.Node) error {
|
|
portconfig := fmt.Sprintf("%s.%s", portmapping.Port.Port(), portmapping.Port.Proto())
|
|
nodenames := []string{}
|
|
for _, node := range targetNodes {
|
|
if node.Role == k3d.LoadBalancerRole {
|
|
return fmt.Errorf("cannot add port config referencing the loadbalancer itself (loop)")
|
|
}
|
|
nodenames = append(nodenames, node.Name)
|
|
}
|
|
|
|
// entry for that port doesn't exist yet, so we simply create it with the list of node names
|
|
if _, ok := loadbalancer.Config.Ports[portconfig]; !ok {
|
|
loadbalancer.Config.Ports[portconfig] = nodenames
|
|
return nil
|
|
}
|
|
|
|
nodenameLoop:
|
|
for _, nodename := range nodenames {
|
|
for _, existingNames := range loadbalancer.Config.Ports[portconfig] {
|
|
if nodename == existingNames {
|
|
continue nodenameLoop
|
|
}
|
|
loadbalancer.Config.Ports[portconfig] = append(loadbalancer.Config.Ports[portconfig], nodename)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|