mirror of
https://github.com/kubernetes/kubernetes.git
synced 2026-08-08 10:40:38 +00:00
688 lines
25 KiB
Go
688 lines
25 KiB
Go
//go:build windows
|
|
|
|
/*
|
|
Copyright 2018 The Kubernetes Authors.
|
|
|
|
Licensed under the Apache License, Version 2.0 (the "License");
|
|
you may not use this file except in compliance with the License.
|
|
You may obtain a copy of the License at
|
|
|
|
http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
Unless required by applicable law or agreed to in writing, software
|
|
distributed under the License is distributed on an "AS IS" BASIS,
|
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
See the License for the specific language governing permissions and
|
|
limitations under the License.
|
|
*/
|
|
|
|
package winkernel
|
|
|
|
import (
|
|
"crypto/sha1"
|
|
"encoding/json"
|
|
"fmt"
|
|
"strings"
|
|
|
|
"github.com/Microsoft/hnslib/hcn"
|
|
|
|
"k8s.io/klog/v2"
|
|
)
|
|
|
|
type HostNetworkService interface {
|
|
getNetworkByName(name string) (*hnsNetworkInfo, error)
|
|
// Returns a map of endpoints keyed by both endpoint ID and IP address for all endpoints on the specified network, and a map of remote endpoints with duplicate IPs to be deleted.
|
|
getAllEndpointsByNetwork(networkName string) (map[string]*endpointInfo, map[string]bool, error)
|
|
// deleteAllRemoteEndpointsWithDupIP deletes all remote endpoints with duplicate IPs that were found in getAllEndpointsByNetwork. This is needed to clean up stale remote endpoints that can be left behind due to a Windows bug.
|
|
deleteAllRemoteEndpointsWithDupIP(remoteEPsWithDupIP map[string]bool)
|
|
getEndpointByID(id string) (*endpointInfo, error)
|
|
getEndpointByIpAddress(ip string, networkName string) (*endpointInfo, error)
|
|
getEndpointByName(id string) (*endpointInfo, error)
|
|
createEndpoint(ep *endpointInfo, networkName string) (*endpointInfo, error)
|
|
deleteEndpoint(hnsID string) error
|
|
getLoadBalancer(endpoints []endpointInfo, flags loadBalancerFlags, sourceVip string, vip string, protocol uint16, internalPort uint16, externalPort uint16, previousLoadBalancers map[loadBalancerIdentifier]*loadBalancerInfo) (*loadBalancerInfo, error)
|
|
getAllLoadBalancers() (map[loadBalancerIdentifier]*loadBalancerInfo, error)
|
|
createOrReplaceLoadbalancer(proposedLB *hcn.HostComputeLoadBalancer, existingLBs map[loadBalancerIdentifier]*loadBalancerInfo) (*hcn.HostComputeLoadBalancer, error)
|
|
updateLoadBalancer(hnsID string, sourceVip, vip string, endpoints []endpointInfo, flags loadBalancerFlags, protocol, internalPort, externalPort uint16, previousLoadBalancers map[loadBalancerIdentifier]*loadBalancerInfo) (*loadBalancerInfo, error)
|
|
deleteLoadBalancer(hnsID string) error
|
|
}
|
|
|
|
type hns struct {
|
|
hcn HcnService
|
|
}
|
|
|
|
var (
|
|
// LoadBalancerFlagsIPv6 enables IPV6.
|
|
LoadBalancerFlagsIPv6 hcn.LoadBalancerFlags = 2
|
|
// LoadBalancerPortMappingFlagsVipExternalIP enables VipExternalIP.
|
|
LoadBalancerPortMappingFlagsVipExternalIP hcn.LoadBalancerPortMappingFlags = 16
|
|
)
|
|
|
|
const (
|
|
|
|
// 0x6b5 is a standard Win32 RPC error (RPC_S_UNKNOWN_IF).
|
|
// It is returned by the RPC runtime when the target service's
|
|
// RPC interface is not registered. In this context, it typically
|
|
// indicates that the HNS service is not running or not yet initialized.
|
|
errorCodeHnsNotRunning = "0x6b5"
|
|
|
|
// 0xb7 (ERROR_ALREADY_EXISTS): Object already exists.
|
|
// Returned when attempting to create a resource that already exists.
|
|
errorCodeFileAlreadyExists = "0xb7"
|
|
)
|
|
|
|
// IsHnsNotRunningError checks if the error is due to HNS service not running by looking for the specific error code in the error message.
|
|
func IsHnsNotRunningError(err error) bool {
|
|
return err != nil && strings.Contains(err.Error(), errorCodeHnsNotRunning)
|
|
}
|
|
|
|
// IsPolicyAlreadyExists checks if the error is due to a load balancer policy already existing with
|
|
// the same frontend configuration by checking for specific error conditions, including the presence
|
|
// of a specific error code or message indicating a resource already exists.
|
|
func IsPolicyAlreadyExists(err error) bool {
|
|
if err == nil {
|
|
return false
|
|
}
|
|
return hcn.IsPortAlreadyExistsError(err) || strings.Contains(err.Error(), errorCodeFileAlreadyExists)
|
|
}
|
|
|
|
// findExistingLBIdByFrontend finds the ID of an existing load balancer that matches the frontend configuration of the proposed load balancer.
|
|
func findExistingLBIdByFrontend(proposedLB *hcn.HostComputeLoadBalancer, existingLBs map[loadBalancerIdentifier]*loadBalancerInfo) string {
|
|
if len(proposedLB.PortMappings) == 0 {
|
|
return ""
|
|
}
|
|
lbID := ""
|
|
frontEndVIP, isProposedLbIPv6 := "", (proposedLB.Flags&LoadBalancerFlagsIPv6) == LoadBalancerFlagsIPv6
|
|
if len(proposedLB.FrontendVIPs) != 0 {
|
|
frontEndVIP = proposedLB.FrontendVIPs[0]
|
|
}
|
|
for id, lbInfo := range existingLBs {
|
|
if id.vip == frontEndVIP && id.protocol == uint16(proposedLB.PortMappings[0].Protocol) && id.internalPort == proposedLB.PortMappings[0].InternalPort && id.externalPort == proposedLB.PortMappings[0].ExternalPort && id.isIPv6 == isProposedLbIPv6 {
|
|
if lbID != "" {
|
|
// More than 1 existing LB has the same frontend configuration, return no match to avoid deleting any of them
|
|
klog.Warning("More than 1 existing lb has matching frontend, will not delete any existing lbs", "existingLBID1", lbID, "existingLBID2", lbInfo.hnsID, "frontendVIP", frontEndVIP, "protocol", proposedLB.PortMappings[0].Protocol, "internalPort", proposedLB.PortMappings[0].InternalPort, "externalPort", proposedLB.PortMappings[0].ExternalPort, "isIPv6", isProposedLbIPv6)
|
|
return ""
|
|
}
|
|
lbID = lbInfo.hnsID
|
|
}
|
|
}
|
|
klog.V(4).InfoS("Search for existing lb with matching frontend completed", "lbID", lbID, "frontendVIP", frontEndVIP, "protocol", proposedLB.PortMappings[0].Protocol, "internalPort", proposedLB.PortMappings[0].InternalPort, "externalPort", proposedLB.PortMappings[0].ExternalPort, "isIPv6", isProposedLbIPv6)
|
|
return lbID
|
|
}
|
|
|
|
func getLoadBalancerPolicyFlags(flags loadBalancerFlags) (lbPortMappingFlags hcn.LoadBalancerPortMappingFlags, lbFlags hcn.LoadBalancerFlags) {
|
|
lbPortMappingFlags = hcn.LoadBalancerPortMappingFlagsNone
|
|
if flags.isILB {
|
|
lbPortMappingFlags |= hcn.LoadBalancerPortMappingFlagsILB
|
|
}
|
|
if flags.useMUX {
|
|
lbPortMappingFlags |= hcn.LoadBalancerPortMappingFlagsUseMux
|
|
}
|
|
if flags.preserveDIP {
|
|
lbPortMappingFlags |= hcn.LoadBalancerPortMappingFlagsPreserveDIP
|
|
}
|
|
if flags.localRoutedVIP {
|
|
lbPortMappingFlags |= hcn.LoadBalancerPortMappingFlagsLocalRoutedVIP
|
|
}
|
|
if flags.isVipExternalIP {
|
|
lbPortMappingFlags |= LoadBalancerPortMappingFlagsVipExternalIP
|
|
}
|
|
lbFlags = hcn.LoadBalancerFlagsNone
|
|
if flags.isDSR {
|
|
lbFlags |= hcn.LoadBalancerFlagsDSR
|
|
}
|
|
if flags.isIPv6 {
|
|
lbFlags |= LoadBalancerFlagsIPv6
|
|
}
|
|
return
|
|
}
|
|
|
|
func (hns hns) getNetworkByName(name string) (*hnsNetworkInfo, error) {
|
|
hnsnetwork, err := hns.hcn.GetNetworkByName(name)
|
|
if err != nil {
|
|
klog.ErrorS(err, "Error getting network by name")
|
|
return nil, err
|
|
}
|
|
|
|
var remoteSubnets []*remoteSubnetInfo
|
|
for _, policy := range hnsnetwork.Policies {
|
|
if policy.Type == hcn.RemoteSubnetRoute {
|
|
policySettings := hcn.RemoteSubnetRoutePolicySetting{}
|
|
err = json.Unmarshal(policy.Settings, &policySettings)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to unmarshal Remote Subnet policy settings")
|
|
}
|
|
rs := &remoteSubnetInfo{
|
|
destinationPrefix: policySettings.DestinationPrefix,
|
|
isolationID: policySettings.IsolationId,
|
|
providerAddress: policySettings.ProviderAddress,
|
|
drMacAddress: policySettings.DistributedRouterMacAddress,
|
|
}
|
|
remoteSubnets = append(remoteSubnets, rs)
|
|
}
|
|
}
|
|
|
|
return &hnsNetworkInfo{
|
|
id: hnsnetwork.Id,
|
|
name: hnsnetwork.Name,
|
|
networkType: string(hnsnetwork.Type),
|
|
remoteSubnets: remoteSubnets,
|
|
}, nil
|
|
}
|
|
|
|
func (hns hns) getAllEndpointsByNetwork(networkName string) (map[string]*(endpointInfo), map[string]bool, error) {
|
|
hcnnetwork, err := hns.hcn.GetNetworkByName(networkName)
|
|
if err != nil {
|
|
klog.ErrorS(err, "failed to get HNS network by name", "name", networkName)
|
|
return nil, nil, err
|
|
}
|
|
endpoints, err := hns.hcn.ListEndpointsOfNetwork(hcnnetwork.Id)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to list endpoints: %w", err)
|
|
}
|
|
|
|
endpointInfos := make(map[string]*(endpointInfo))
|
|
remoteEPsWithDupIP := make(map[string]bool)
|
|
|
|
for _, ep := range endpoints {
|
|
|
|
if len(ep.IpConfigurations) == 0 {
|
|
klog.V(3).InfoS("No IpConfigurations found in endpoint info of queried endpoints", "endpoint", ep)
|
|
continue
|
|
}
|
|
|
|
for index, ipConfig := range ep.IpConfigurations {
|
|
|
|
if index > 1 {
|
|
// Expecting only ipv4 and ipv6 ipaddresses
|
|
// This is highly unlikely to happen, but if it does, we should log a warning
|
|
// and break out of the loop
|
|
klog.Warning("Endpoint ipconfiguration holds more than 2 IP addresses.", "hnsID", ep.Id, "IP", ipConfig.IpAddress, "ipConfigCount", len(ep.IpConfigurations))
|
|
break
|
|
}
|
|
|
|
curEpIsLocal := uint32(ep.Flags&hcn.EndpointFlagsRemoteEndpoint) == 0
|
|
|
|
if existingEp, ok := endpointInfos[ipConfig.IpAddress]; ok {
|
|
if curEpIsLocal && !existingEp.isLocal {
|
|
// Local found, stale remote in map → delete remote from HNS, overwrite
|
|
remoteEPsWithDupIP[existingEp.hnsID] = true
|
|
delete(endpointInfos, existingEp.hnsID)
|
|
delete(endpointInfos, existingEp.ip)
|
|
// fall through to add local
|
|
} else if !curEpIsLocal && existingEp.isLocal {
|
|
// Local already in map, remote arriving → delete remote from HNS, skip
|
|
remoteEPsWithDupIP[ep.Id] = true
|
|
continue
|
|
} else {
|
|
continue // same type, keep existing
|
|
}
|
|
}
|
|
|
|
// Add to map with key endpoint ID or IP address
|
|
// Storing this is expensive in terms of memory, however there is a bug in Windows Server 2019 and 2022 that can cause two endpoints (local and remote) to be created with the same IP address.
|
|
// TODO: Store by IP only and remove any lookups by endpoint ID.
|
|
epInfo := &endpointInfo{
|
|
ip: ipConfig.IpAddress,
|
|
isLocal: curEpIsLocal,
|
|
macAddress: ep.MacAddress,
|
|
hnsID: ep.Id,
|
|
hns: hns,
|
|
// only ready and not terminating endpoints were added to HNS
|
|
ready: true,
|
|
serving: true,
|
|
terminating: false,
|
|
}
|
|
endpointInfos[ep.Id] = epInfo
|
|
endpointInfos[ipConfig.IpAddress] = epInfo
|
|
}
|
|
}
|
|
klog.V(3).InfoS("Queried endpoints from network", "network", networkName, "count", len(endpointInfos))
|
|
klog.V(5).InfoS("Queried endpoints details", "network", networkName, "endpointInfos", endpointInfos)
|
|
return endpointInfos, remoteEPsWithDupIP, nil
|
|
}
|
|
|
|
func (hns hns) deleteAllRemoteEndpointsWithDupIP(remoteEPsWithDupIP map[string]bool) {
|
|
for hnsID := range remoteEPsWithDupIP {
|
|
klog.V(3).InfoS("Deleting stale remote endpoint with duplicate IP", "hnsID", hnsID)
|
|
err := hns.deleteEndpoint(hnsID)
|
|
if err != nil {
|
|
klog.ErrorS(err, "Failed to delete stale remote endpoint with duplicate IP", "hnsID", hnsID)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (hns hns) getEndpointByID(id string) (*endpointInfo, error) {
|
|
hnsendpoint, err := hns.hcn.GetEndpointByID(id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &endpointInfo{ //TODO: fill out PA
|
|
ip: hnsendpoint.IpConfigurations[0].IpAddress,
|
|
isLocal: uint32(hnsendpoint.Flags&hcn.EndpointFlagsRemoteEndpoint) == 0, //TODO: Change isLocal to isRemote
|
|
macAddress: hnsendpoint.MacAddress,
|
|
hnsID: hnsendpoint.Id,
|
|
hns: hns,
|
|
}, nil
|
|
}
|
|
func (hns hns) getEndpointByIpAddress(ip string, networkName string) (*endpointInfo, error) {
|
|
hnsnetwork, err := hns.hcn.GetNetworkByName(networkName)
|
|
if err != nil {
|
|
klog.ErrorS(err, "Error getting network by name")
|
|
return nil, err
|
|
}
|
|
|
|
endpoints, err := hns.hcn.ListEndpoints()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to list endpoints: %w", err)
|
|
}
|
|
for _, endpoint := range endpoints {
|
|
equal := false
|
|
if len(endpoint.IpConfigurations) > 0 {
|
|
equal = endpoint.IpConfigurations[0].IpAddress == ip
|
|
|
|
if !equal && len(endpoint.IpConfigurations) > 1 {
|
|
equal = endpoint.IpConfigurations[1].IpAddress == ip
|
|
}
|
|
}
|
|
if equal && strings.EqualFold(endpoint.HostComputeNetwork, hnsnetwork.Id) {
|
|
return &endpointInfo{
|
|
ip: ip,
|
|
isLocal: uint32(endpoint.Flags&hcn.EndpointFlagsRemoteEndpoint) == 0, //TODO: Change isLocal to isRemote
|
|
macAddress: endpoint.MacAddress,
|
|
hnsID: endpoint.Id,
|
|
hns: hns,
|
|
}, nil
|
|
}
|
|
}
|
|
return nil, fmt.Errorf("Endpoint %v not found on network %s", ip, networkName)
|
|
}
|
|
func (hns hns) getEndpointByName(name string) (*endpointInfo, error) {
|
|
hnsendpoint, err := hns.hcn.GetEndpointByName(name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &endpointInfo{ //TODO: fill out PA
|
|
ip: hnsendpoint.IpConfigurations[0].IpAddress,
|
|
isLocal: uint32(hnsendpoint.Flags&hcn.EndpointFlagsRemoteEndpoint) == 0, //TODO: Change isLocal to isRemote
|
|
macAddress: hnsendpoint.MacAddress,
|
|
hnsID: hnsendpoint.Id,
|
|
hns: hns,
|
|
}, nil
|
|
}
|
|
func (hns hns) createEndpoint(ep *endpointInfo, networkName string) (*endpointInfo, error) {
|
|
hnsNetwork, err := hns.hcn.GetNetworkByName(networkName)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var flags hcn.EndpointFlags
|
|
if !ep.isLocal {
|
|
flags |= hcn.EndpointFlagsRemoteEndpoint
|
|
}
|
|
ipConfig := &hcn.IpConfig{
|
|
IpAddress: ep.ip,
|
|
}
|
|
hnsEndpoint := &hcn.HostComputeEndpoint{
|
|
IpConfigurations: []hcn.IpConfig{*ipConfig},
|
|
MacAddress: ep.macAddress,
|
|
Flags: flags,
|
|
SchemaVersion: hcn.SchemaVersion{
|
|
Major: 2,
|
|
Minor: 0,
|
|
},
|
|
}
|
|
|
|
var createdEndpoint *hcn.HostComputeEndpoint
|
|
if !ep.isLocal {
|
|
if len(ep.providerAddress) != 0 {
|
|
policySettings := hcn.ProviderAddressEndpointPolicySetting{
|
|
ProviderAddress: ep.providerAddress,
|
|
}
|
|
policySettingsJson, err := json.Marshal(policySettings)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("PA Policy creation failed: %v", err)
|
|
}
|
|
paPolicy := hcn.EndpointPolicy{
|
|
Type: hcn.NetworkProviderAddress,
|
|
Settings: policySettingsJson,
|
|
}
|
|
hnsEndpoint.Policies = append(hnsEndpoint.Policies, paPolicy)
|
|
}
|
|
createdEndpoint, err = hns.hcn.CreateRemoteEndpoint(hnsNetwork, hnsEndpoint)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
klog.V(3).InfoS("Created remote endpoint resource", "hnsID", createdEndpoint.Id)
|
|
} else {
|
|
createdEndpoint, err = hns.hcn.CreateEndpoint(hnsNetwork, hnsEndpoint)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
klog.V(3).InfoS("Created local endpoint resource", "hnsID", createdEndpoint.Id)
|
|
}
|
|
return &endpointInfo{
|
|
ip: createdEndpoint.IpConfigurations[0].IpAddress,
|
|
isLocal: uint32(createdEndpoint.Flags&hcn.EndpointFlagsRemoteEndpoint) == 0,
|
|
macAddress: createdEndpoint.MacAddress,
|
|
hnsID: createdEndpoint.Id,
|
|
providerAddress: ep.providerAddress, //TODO get from createdEndpoint
|
|
hns: hns,
|
|
}, nil
|
|
}
|
|
func (hns hns) deleteEndpoint(hnsID string) error {
|
|
hnsendpoint, err := hns.hcn.GetEndpointByID(hnsID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
err = hns.hcn.DeleteEndpoint(hnsendpoint)
|
|
if err == nil {
|
|
klog.V(3).InfoS("Remote endpoint resource deleted", "hnsID", hnsID)
|
|
}
|
|
return err
|
|
}
|
|
|
|
// findLoadBalancerID will construct a id from the provided loadbalancer fields
|
|
func findLoadBalancerID(endpoints []endpointInfo, vip string, protocol, internalPort, externalPort uint16, isIpv6 bool) (loadBalancerIdentifier, error) {
|
|
// Compute hash from backends (endpoint IDs)
|
|
hash, err := hashEndpoints(endpoints)
|
|
if err != nil {
|
|
klog.V(2).ErrorS(err, "Error hashing endpoints", "endpoints", endpoints)
|
|
return loadBalancerIdentifier{}, err
|
|
}
|
|
return loadBalancerIdentifier{protocol: protocol, internalPort: internalPort, externalPort: externalPort, vip: vip, endpointsHash: hash, isIPv6: isIpv6}, nil
|
|
}
|
|
|
|
func (hns hns) getAllLoadBalancers() (map[loadBalancerIdentifier]*loadBalancerInfo, error) {
|
|
lbs, err := hns.hcn.ListLoadBalancers()
|
|
var id loadBalancerIdentifier
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
loadBalancers := make(map[loadBalancerIdentifier]*(loadBalancerInfo))
|
|
for _, lb := range lbs {
|
|
isIPv6 := (lb.Flags & LoadBalancerFlagsIPv6) == LoadBalancerFlagsIPv6
|
|
portMap := lb.PortMappings[0]
|
|
// Compute hash from backends (endpoint IDs)
|
|
hash, err := hashEndpoints(lb.HostComputeEndpoints)
|
|
if err != nil {
|
|
klog.V(2).ErrorS(err, "Error hashing endpoints", "policy", lb)
|
|
return nil, err
|
|
}
|
|
if len(lb.FrontendVIPs) == 0 {
|
|
// Leave VIP uninitialized
|
|
id = loadBalancerIdentifier{protocol: uint16(portMap.Protocol), internalPort: portMap.InternalPort, externalPort: portMap.ExternalPort, endpointsHash: hash, isIPv6: isIPv6}
|
|
} else {
|
|
id = loadBalancerIdentifier{protocol: uint16(portMap.Protocol), internalPort: portMap.InternalPort, externalPort: portMap.ExternalPort, vip: lb.FrontendVIPs[0], endpointsHash: hash, isIPv6: isIPv6}
|
|
}
|
|
loadBalancers[id] = &loadBalancerInfo{
|
|
hnsID: lb.Id,
|
|
}
|
|
}
|
|
klog.V(3).InfoS("Queried load balancers", "count", len(lbs))
|
|
return loadBalancers, nil
|
|
}
|
|
|
|
func (hns hns) createOrReplaceLoadbalancer(proposedLB *hcn.HostComputeLoadBalancer, existingLBs map[loadBalancerIdentifier]*loadBalancerInfo) (*hcn.HostComputeLoadBalancer, error) {
|
|
lb, err := hns.hcn.CreateLoadBalancer(proposedLB)
|
|
if IsPolicyAlreadyExists(err) && len(proposedLB.PortMappings) > 0 {
|
|
frontEndVIP := ""
|
|
if len(proposedLB.FrontendVIPs) != 0 {
|
|
frontEndVIP = proposedLB.FrontendVIPs[0]
|
|
}
|
|
existingLbID := findExistingLBIdByFrontend(proposedLB, existingLBs)
|
|
if existingLbID != "" {
|
|
klog.V(4).InfoS("Deleting matching load balancer and retrying create load balancer again", "existingLBID", existingLbID, "frontendVIP", frontEndVIP, "protocol", proposedLB.PortMappings[0].Protocol, "internalPort", proposedLB.PortMappings[0].InternalPort, "externalPort", proposedLB.PortMappings[0].ExternalPort)
|
|
err = hns.deleteLoadBalancer(existingLbID)
|
|
if err != nil {
|
|
klog.V(1).InfoS("Failed to delete existing load balancer", "lbID", existingLbID, "error", err)
|
|
return nil, err
|
|
}
|
|
lb, err = hns.hcn.CreateLoadBalancer(proposedLB)
|
|
} else {
|
|
klog.V(4).InfoS("Did not find a unique existing load balancer with matching frontend configuration", "frontendVIP", frontEndVIP, "protocol", proposedLB.PortMappings[0].Protocol, "internalPort", proposedLB.PortMappings[0].InternalPort, "externalPort", proposedLB.PortMappings[0].ExternalPort)
|
|
}
|
|
}
|
|
return lb, err
|
|
}
|
|
|
|
func (hns hns) getLoadBalancer(endpoints []endpointInfo, flags loadBalancerFlags, sourceVip string, vip string, protocol uint16, internalPort uint16, externalPort uint16, previousLoadBalancers map[loadBalancerIdentifier]*loadBalancerInfo) (*loadBalancerInfo, error) {
|
|
var id loadBalancerIdentifier
|
|
vips := []string{}
|
|
id, lbIdErr := findLoadBalancerID(
|
|
endpoints,
|
|
vip,
|
|
protocol,
|
|
internalPort,
|
|
externalPort,
|
|
flags.isIPv6,
|
|
)
|
|
|
|
if lbIdErr != nil {
|
|
klog.V(2).ErrorS(lbIdErr, "Error hashing endpoints", "endpoints", endpoints)
|
|
return nil, lbIdErr
|
|
}
|
|
|
|
if len(vip) > 0 {
|
|
vips = append(vips, vip)
|
|
}
|
|
|
|
if lb, found := previousLoadBalancers[id]; found {
|
|
klog.V(1).InfoS("Found cached Hns loadbalancer policy resource", "policies", lb)
|
|
return lb, nil
|
|
}
|
|
|
|
lbPortMappingFlags := hcn.LoadBalancerPortMappingFlagsNone
|
|
if flags.isILB {
|
|
lbPortMappingFlags |= hcn.LoadBalancerPortMappingFlagsILB
|
|
}
|
|
if flags.useMUX {
|
|
lbPortMappingFlags |= hcn.LoadBalancerPortMappingFlagsUseMux
|
|
}
|
|
if flags.preserveDIP {
|
|
lbPortMappingFlags |= hcn.LoadBalancerPortMappingFlagsPreserveDIP
|
|
}
|
|
if flags.localRoutedVIP {
|
|
lbPortMappingFlags |= hcn.LoadBalancerPortMappingFlagsLocalRoutedVIP
|
|
}
|
|
if flags.isVipExternalIP {
|
|
lbPortMappingFlags |= LoadBalancerPortMappingFlagsVipExternalIP
|
|
}
|
|
|
|
lbFlags := hcn.LoadBalancerFlagsNone
|
|
if flags.isDSR {
|
|
lbFlags |= hcn.LoadBalancerFlagsDSR
|
|
}
|
|
|
|
if flags.isIPv6 {
|
|
lbFlags |= LoadBalancerFlagsIPv6
|
|
}
|
|
|
|
lbDistributionType := hcn.LoadBalancerDistributionNone
|
|
|
|
if flags.sessionAffinity {
|
|
lbDistributionType = hcn.LoadBalancerDistributionSourceIP
|
|
}
|
|
|
|
loadBalancer := &hcn.HostComputeLoadBalancer{
|
|
SourceVIP: sourceVip,
|
|
PortMappings: []hcn.LoadBalancerPortMapping{
|
|
{
|
|
Protocol: uint32(protocol),
|
|
InternalPort: internalPort,
|
|
ExternalPort: externalPort,
|
|
DistributionType: lbDistributionType,
|
|
Flags: lbPortMappingFlags,
|
|
},
|
|
},
|
|
FrontendVIPs: vips,
|
|
SchemaVersion: hcn.SchemaVersion{
|
|
Major: 2,
|
|
Minor: 0,
|
|
},
|
|
Flags: lbFlags,
|
|
}
|
|
|
|
for _, ep := range endpoints {
|
|
loadBalancer.HostComputeEndpoints = append(loadBalancer.HostComputeEndpoints, ep.hnsID)
|
|
}
|
|
|
|
lb, err := hns.createOrReplaceLoadbalancer(loadBalancer, previousLoadBalancers)
|
|
|
|
if err != nil {
|
|
klog.V(2).ErrorS(err, "Error creating Hns loadbalancer policy resource", "error", err, "endpoints", endpoints)
|
|
return nil, err
|
|
}
|
|
|
|
klog.V(1).InfoS("Created Hns loadbalancer policy resource", "loadBalancer", lb)
|
|
lbInfo := &loadBalancerInfo{
|
|
hnsID: lb.Id,
|
|
}
|
|
// Add to map of load balancers
|
|
previousLoadBalancers[id] = lbInfo
|
|
return lbInfo, err
|
|
}
|
|
|
|
func (hns hns) updateLoadBalancer(hnsID string,
|
|
sourceVip,
|
|
vip string,
|
|
endpoints []endpointInfo,
|
|
flags loadBalancerFlags,
|
|
protocol,
|
|
internalPort,
|
|
externalPort uint16,
|
|
previousLoadBalancers map[loadBalancerIdentifier]*loadBalancerInfo) (*loadBalancerInfo, error) {
|
|
klog.V(3).InfoS("Updating existing loadbalancer called", "hnsLbID", hnsID, "endpointCount", len(endpoints), "vip", vip, "sourceVip", sourceVip, "internalPort", internalPort, "externalPort", externalPort)
|
|
|
|
var id loadBalancerIdentifier
|
|
vips := []string{}
|
|
// Compute hash from backends (endpoint IDs)
|
|
hash, err := hashEndpoints(endpoints)
|
|
if err != nil {
|
|
klog.V(2).ErrorS(err, "Error hashing endpoints", "endpoints", endpoints)
|
|
return nil, err
|
|
}
|
|
if len(vip) > 0 {
|
|
id = loadBalancerIdentifier{protocol: protocol, internalPort: internalPort, externalPort: externalPort, vip: vip, endpointsHash: hash, isIPv6: flags.isIPv6}
|
|
vips = append(vips, vip)
|
|
} else {
|
|
id = loadBalancerIdentifier{protocol: protocol, internalPort: internalPort, externalPort: externalPort, endpointsHash: hash, isIPv6: flags.isIPv6}
|
|
}
|
|
|
|
if lb, found := previousLoadBalancers[id]; found {
|
|
klog.V(1).InfoS("Found cached Hns loadbalancer policy resource", "policies", lb)
|
|
return lb, nil
|
|
}
|
|
|
|
lbPortMappingFlags, lbFlags := getLoadBalancerPolicyFlags(flags)
|
|
|
|
lbDistributionType := hcn.LoadBalancerDistributionNone
|
|
|
|
if flags.sessionAffinity {
|
|
lbDistributionType = hcn.LoadBalancerDistributionSourceIP
|
|
}
|
|
|
|
loadBalancer := &hcn.HostComputeLoadBalancer{
|
|
SourceVIP: sourceVip,
|
|
PortMappings: []hcn.LoadBalancerPortMapping{
|
|
{
|
|
Protocol: uint32(protocol),
|
|
InternalPort: internalPort,
|
|
ExternalPort: externalPort,
|
|
DistributionType: lbDistributionType,
|
|
Flags: lbPortMappingFlags,
|
|
},
|
|
},
|
|
FrontendVIPs: vips,
|
|
SchemaVersion: hcn.SchemaVersion{
|
|
Major: 2,
|
|
Minor: 0,
|
|
},
|
|
Flags: lbFlags,
|
|
}
|
|
|
|
for _, ep := range endpoints {
|
|
loadBalancer.HostComputeEndpoints = append(loadBalancer.HostComputeEndpoints, ep.hnsID)
|
|
}
|
|
|
|
lb, err := hns.hcn.UpdateLoadBalancer(loadBalancer, hnsID)
|
|
|
|
if err != nil {
|
|
klog.V(2).ErrorS(err, "Error updating existing loadbalancer", "hnsLbID", hnsID, "error", err, "endpoints", endpoints)
|
|
return nil, err
|
|
}
|
|
|
|
klog.V(1).InfoS("Update loadbalancer is successful", "loadBalancer", lb)
|
|
lbInfo := &loadBalancerInfo{
|
|
hnsID: lb.Id,
|
|
}
|
|
// Add to map of load balancers
|
|
previousLoadBalancers[id] = lbInfo
|
|
return lbInfo, err
|
|
}
|
|
|
|
func (hns hns) deleteLoadBalancer(hnsID string) error {
|
|
lb, err := hns.hcn.GetLoadBalancerByID(hnsID)
|
|
if err != nil {
|
|
if hcn.IsNotFoundError(err) {
|
|
klog.V(1).InfoS("LoadBalancer policy resource not found, may have already been deleted", "lbID", hnsID)
|
|
// Return silently
|
|
return nil
|
|
}
|
|
if IsHnsNotRunningError(err) {
|
|
klog.V(1).ErrorS(err, "HNS is not running, skipping delete loadbalancer", "lbID", hnsID)
|
|
return err
|
|
}
|
|
klog.V(2).ErrorS(err, "Error getting Hns loadbalancer policy resource by ID", "lbID", hnsID)
|
|
return err
|
|
}
|
|
|
|
err = hns.hcn.DeleteLoadBalancer(lb)
|
|
if err != nil {
|
|
// There is a bug in Windows Server 2019, that can cause the delete call to fail sometimes. We retry one more time.
|
|
// TODO: The logic in syncProxyRules should be rewritten in the future to better stage and handle a call like this failing using the policyApplied fields.
|
|
klog.V(1).ErrorS(err, "Error deleting Hns loadbalancer policy resource. Attempting one more time...", "loadBalancer", lb)
|
|
err = hns.hcn.DeleteLoadBalancer(lb)
|
|
}
|
|
if err != nil {
|
|
klog.V(2).ErrorS(err, "Error deleting Hns loadbalancer policy resource again.", "hnsID", hnsID)
|
|
return err
|
|
}
|
|
klog.V(3).InfoS("Deleted Hns loadbalancer policy resource", "hnsID", hnsID)
|
|
return err
|
|
}
|
|
|
|
// Calculates a hash from the given endpoint IDs.
|
|
func hashEndpoints[T string | endpointInfo](endpoints []T) (hash [20]byte, err error) {
|
|
var id string
|
|
// Recover in case something goes wrong. Return error and null byte array.
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
err = r.(error)
|
|
hash = [20]byte{}
|
|
}
|
|
}()
|
|
|
|
// Iterate over endpoints, compute hash
|
|
for _, ep := range endpoints {
|
|
switch x := any(ep).(type) {
|
|
case endpointInfo:
|
|
id = strings.ToUpper(x.hnsID)
|
|
case string:
|
|
id = strings.ToUpper(x)
|
|
}
|
|
if len(id) > 0 {
|
|
// We XOR the hashes of endpoints, since they are an unordered set.
|
|
// This can cause collisions, but is sufficient since we are using other keys to identify the load balancer.
|
|
hash = xor(hash, sha1.Sum(([]byte(id))))
|
|
}
|
|
}
|
|
return
|
|
}
|
|
|
|
func xor(b1 [20]byte, b2 [20]byte) (xorbytes [20]byte) {
|
|
for i := 0; i < 20; i++ {
|
|
xorbytes[i] = b1[i] ^ b2[i]
|
|
}
|
|
return xorbytes
|
|
}
|