Skip to content

Commit

Permalink
Merge pull request #109981 from marosset/automated-cherry-pick-of-#10…
Browse files Browse the repository at this point in the history 
…9124-upstream-release-1.23

Automated cherry pick of #109124: Winkernel proxier cache HNS data to improve syncProxyRules
  • Loading branch information
@k8s-ci-robot
k8s-ci-robot committed Jun 9, 2022
2 parents e160866 + 424dde3 commit 4094673
Show file tree
Hide file tree
Showing 5 changed files with 216 additions and 78 deletions.
102 changes: 76 additions & 26 deletions pkg/proxy/winkernel/hnsV1.go
View file
Expand Up @@ -31,11 +31,13 @@ import (

type HostNetworkService interface {
getNetworkByName(name string) (*hnsNetworkInfo, error)
getAllEndpointsByNetwork(networkName string) (map[string]*endpointsInfo, error)
getEndpointByID(id string) (*endpointsInfo, error)
getEndpointByIpAddress(ip string, networkName string) (*endpointsInfo, error)
createEndpoint(ep *endpointsInfo, networkName string) (*endpointsInfo, error)
deleteEndpoint(hnsID string) error
getLoadBalancer(endpoints []endpointsInfo, flags loadBalancerFlags, sourceVip string, vip string, protocol uint16, internalPort uint16, externalPort uint16) (*loadBalancerInfo, error)
getLoadBalancer(endpoints []endpointsInfo, flags loadBalancerFlags, sourceVip string, vip string, protocol uint16, internalPort uint16, externalPort uint16, previousLoadBalancers map[loadBalancerIdentifier]*loadBalancerInfo) (*loadBalancerInfo, error)
getAllLoadBalancers() (map[loadBalancerIdentifier]*loadBalancerInfo, error)
deleteLoadBalancer(hnsID string) error
}

Expand All @@ -55,6 +57,41 @@ func (hns hnsV1) getNetworkByName(name string) (*hnsNetworkInfo, error) {
networkType: hnsnetwork.Type,
}, nil
}

func (hns hnsV1) getAllEndpointsByNetwork(networkName string) (map[string]*(endpointsInfo), error) {
hnsnetwork, err := hcsshim.GetHNSNetworkByName(networkName)
if err != nil {
klog.ErrorS(err, "failed to get HNS network by name", "name", networkName)
return nil, err
}
endpoints, err := hcsshim.HNSListEndpointRequest()
if err != nil {
return nil, fmt.Errorf("failed to list endpoints: %w", err)
}
endpointInfos := make(map[string]*(endpointsInfo))
for _, endpoint := range endpoints {
if strings.EqualFold(endpoint.VirtualNetwork, hnsnetwork.Id) {
// 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 that can cause two endpoints to be created with the same IP address.
// TODO: Store by IP only and remove any lookups by endpoint ID.
endpointInfos[endpoint.Id] = &endpointsInfo{
ip: endpoint.IPAddress.String(),
isLocal: !endpoint.IsRemoteEndpoint,
macAddress: endpoint.MacAddress,
hnsID: endpoint.Id,
hns: hns,
// only ready and not terminating endpoints were added to HNS
ready: true,
serving: true,
terminating: false,
}
endpointInfos[endpoint.IPAddress.String()] = endpointInfos[endpoint.Id]
}
}
klog.V(3).InfoS("Queried endpoints from network", "network", networkName)
return endpointInfos, nil
}

func (hns hnsV1) getEndpointByID(id string) (*endpointsInfo, error) {
hnsendpoint, err := hcsshim.GetHNSEndpointByID(id)
if err != nil {
Expand Down Expand Up @@ -167,39 +204,49 @@ func (hns hnsV1) deleteEndpoint(hnsID string) error {
return err
}

func (hns hnsV1) getLoadBalancer(endpoints []endpointsInfo, flags loadBalancerFlags, sourceVip string, vip string, protocol uint16, internalPort uint16, externalPort uint16) (*loadBalancerInfo, error) {
func (hns hnsV1) getAllLoadBalancers() (map[loadBalancerIdentifier]*loadBalancerInfo, error) {
plists, err := hcsshim.HNSListPolicyListRequest()
var id loadBalancerIdentifier
if err != nil {
return nil, err
}

if flags.isDSR {
klog.V(3).InfoS("DSR is not supported in V1. Using non DSR instead")
}

loadBalancers := make(map[loadBalancerIdentifier]*(loadBalancerInfo))
for _, plist := range plists {
if len(plist.EndpointReferences) != len(endpoints) {
continue
}
// Validate if input meets any of the policy lists
elbPolicy := hcsshim.ELBPolicy{}
if err = json.Unmarshal(plist.Policies[0], &elbPolicy); err != nil {
lb := hcsshim.ELBPolicy{}
if err = json.Unmarshal(plist.Policies[0], &lb); err != nil {
continue
}
if elbPolicy.Protocol == protocol && elbPolicy.InternalPort == internalPort && elbPolicy.ExternalPort == externalPort && elbPolicy.ILB == flags.isILB {
if len(vip) > 0 {
if len(elbPolicy.VIPs) == 0 || elbPolicy.VIPs[0] != vip {
continue
}
} else if len(elbPolicy.VIPs) != 0 {
continue
}
klog.V(1).InfoS("Found existing Hns loadbalancer policy resource", "policies", plist)
return &loadBalancerInfo{
hnsID: plist.ID,
}, nil
// Policy is ELB policy
portMap := lb.LBPolicy
if len(lb.VIPs) == 0 {
// Leave VIP uninitialized
id = loadBalancerIdentifier{protocol: uint16(portMap.Protocol), internalPort: portMap.InternalPort, externalPort: portMap.ExternalPort}
} else {
id = loadBalancerIdentifier{protocol: portMap.Protocol, internalPort: portMap.InternalPort, externalPort: portMap.ExternalPort, vip: lb.VIPs[0]}
}
loadBalancers[id] = &loadBalancerInfo{
hnsID: plist.ID,
}
}
return loadBalancers, nil
}

func (hns hnsV1) getLoadBalancer(endpoints []endpointsInfo, flags loadBalancerFlags, sourceVip string, vip string, protocol uint16, internalPort uint16, externalPort uint16, previousLoadBalancers map[loadBalancerIdentifier]*loadBalancerInfo) (*loadBalancerInfo, error) {
if flags.isDSR {
klog.V(3).InfoS("DSR is not supported in V1. Using non DSR instead")
}
var id loadBalancerIdentifier
if len(vip) > 0 {
id = loadBalancerIdentifier{protocol: protocol, internalPort: internalPort, externalPort: externalPort, vip: vip, endpointsCount: len(endpoints)}
} else {
id = loadBalancerIdentifier{protocol: protocol, internalPort: internalPort, externalPort: externalPort, endpointsCount: len(endpoints)}
}

if lb, found := previousLoadBalancers[id]; found {
klog.V(1).InfoS("Found existing Hns loadbalancer policy resource", "policies", lb)
return lb, nil
}

var hnsEndpoints []hcsshim.HNSEndpoint
for _, ep := range endpoints {
Expand All @@ -224,9 +271,12 @@ func (hns hnsV1) getLoadBalancer(endpoints []endpointsInfo, flags loadBalancerFl
} else {
return nil, err
}
return &loadBalancerInfo{
lbInfo := &loadBalancerInfo{
hnsID: lb.ID,
}, err
}
// Add to map of load balancers
previousLoadBalancers[id] = lbInfo
return lbInfo, err
}
func (hns hnsV1) deleteLoadBalancer(hnsID string) error {
if len(hnsID) == 0 {
Expand Down
104 changes: 69 additions & 35 deletions pkg/proxy/winkernel/hnsV2.go
View file
Expand Up @@ -68,6 +68,39 @@ func (hns hnsV2) getNetworkByName(name string) (*hnsNetworkInfo, error) {
remoteSubnets: remoteSubnets,
}, nil
}

func (hns hnsV2) getAllEndpointsByNetwork(networkName string) (map[string]*(endpointsInfo), error) {
hcnnetwork, err := hcn.GetNetworkByName(networkName)
if err != nil {
klog.ErrorS(err, "failed to get HNS network by name", "name", networkName)
return nil, err
}
endpoints, err := hcn.ListEndpointsOfNetwork(hcnnetwork.Id)
if err != nil {
return nil, fmt.Errorf("failed to list endpoints: %w", err)
}
endpointInfos := make(map[string]*(endpointsInfo))
for _, ep := range endpoints {
// 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 that can cause two endpoints to be created with the same IP address.
// TODO: Store by IP only and remove any lookups by endpoint ID.
endpointInfos[ep.Id] = &endpointsInfo{
ip: ep.IpConfigurations[0].IpAddress,
isLocal: uint32(ep.Flags&hcn.EndpointFlagsRemoteEndpoint) == 0,
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.IpConfigurations[0].IpAddress] = endpointInfos[ep.Id]
}
klog.V(3).InfoS("Queried endpoints from network", "network", networkName)
return endpointInfos, nil
}

func (hns hnsV2) getEndpointByID(id string) (*endpointsInfo, error) {
hnsendpoint, err := hcn.GetEndpointByID(id)
if err != nil {
Expand Down Expand Up @@ -111,7 +144,6 @@ func (hns hnsV2) getEndpointByIpAddress(ip string, networkName string) (*endpoin
}, nil
}
}

return nil, fmt.Errorf("Endpoint %v not found on network %s", ip, networkName)
}
func (hns hnsV2) createEndpoint(ep *endpointsInfo, networkName string) (*endpointsInfo, error) {
Expand Down Expand Up @@ -182,45 +214,43 @@ func (hns hnsV2) deleteEndpoint(hnsID string) error {
}
return err
}
func (hns hnsV2) getLoadBalancer(endpoints []endpointsInfo, flags loadBalancerFlags, sourceVip string, vip string, protocol uint16, internalPort uint16, externalPort uint16) (*loadBalancerInfo, error) {
plists, err := hcn.ListLoadBalancers()

func (hns hnsV2) getAllLoadBalancers() (map[loadBalancerIdentifier]*loadBalancerInfo, error) {
lbs, err := hcn.ListLoadBalancers()
var id loadBalancerIdentifier
if err != nil {
return nil, err
}

for _, plist := range plists {
if len(plist.HostComputeEndpoints) != len(endpoints) {
continue
loadBalancers := make(map[loadBalancerIdentifier]*(loadBalancerInfo))
for _, lb := range lbs {
portMap := lb.PortMappings[0]
if len(lb.FrontendVIPs) == 0 {
// Leave VIP uninitialized
id = loadBalancerIdentifier{protocol: uint16(portMap.Protocol), internalPort: portMap.InternalPort, externalPort: portMap.ExternalPort, endpointsCount: len(lb.HostComputeEndpoints)}
} else {
id = loadBalancerIdentifier{protocol: uint16(portMap.Protocol), internalPort: portMap.InternalPort, externalPort: portMap.ExternalPort, vip: lb.FrontendVIPs[0], endpointsCount: len(lb.HostComputeEndpoints)}
}
// Validate if input meets any of the policy lists
lbPortMapping := plist.PortMappings[0]
if lbPortMapping.Protocol == uint32(protocol) && lbPortMapping.InternalPort == internalPort && lbPortMapping.ExternalPort == externalPort && (lbPortMapping.Flags&1 != 0) == flags.isILB {
if len(vip) > 0 {
if len(plist.FrontendVIPs) == 0 || plist.FrontendVIPs[0] != vip {
continue
}
} else if len(plist.FrontendVIPs) != 0 {
continue
}
klog.V(1).InfoS("Found existing Hns loadbalancer policy resource", "policies", plist)
return &loadBalancerInfo{
hnsID: plist.Id,
}, nil
}
}

var hnsEndpoints []hcn.HostComputeEndpoint
for _, ep := range endpoints {
endpoint, err := hcn.GetEndpointByID(ep.hnsID)
if err != nil {
return nil, err
loadBalancers[id] = &loadBalancerInfo{
hnsID: lb.Id,
}
hnsEndpoints = append(hnsEndpoints, *endpoint)
}
klog.V(3).InfoS("Queried load balancers", "count", len(lbs))
return loadBalancers, nil
}

func (hns hnsV2) getLoadBalancer(endpoints []endpointsInfo, flags loadBalancerFlags, sourceVip string, vip string, protocol uint16, internalPort uint16, externalPort uint16, previousLoadBalancers map[loadBalancerIdentifier]*loadBalancerInfo) (*loadBalancerInfo, error) {
var id loadBalancerIdentifier
vips := []string{}
if len(vip) > 0 {
id = loadBalancerIdentifier{protocol: protocol, internalPort: internalPort, externalPort: externalPort, vip: vip, endpointsCount: len(endpoints)}
vips = append(vips, vip)
} else {
id = loadBalancerIdentifier{protocol: protocol, internalPort: internalPort, externalPort: externalPort, endpointsCount: len(endpoints)}
}

if lb, found := previousLoadBalancers[id]; found {
klog.V(1).InfoS("Found cached Hns loadbalancer policy resource", "policies", lb)
return lb, nil
}

lbPortMappingFlags := hcn.LoadBalancerPortMappingFlagsNone
Expand Down Expand Up @@ -271,8 +301,8 @@ func (hns hnsV2) getLoadBalancer(endpoints []endpointsInfo, flags loadBalancerFl
Flags: lbFlags,
}

for _, endpoint := range hnsEndpoints {
loadBalancer.HostComputeEndpoints = append(loadBalancer.HostComputeEndpoints, endpoint.Id)
for _, ep := range endpoints {
loadBalancer.HostComputeEndpoints = append(loadBalancer.HostComputeEndpoints, ep.hnsID)
}

lb, err := loadBalancer.Create()
Expand All @@ -281,11 +311,15 @@ func (hns hnsV2) getLoadBalancer(endpoints []endpointsInfo, flags loadBalancerFl
return nil, err
}

klog.V(1).InfoS("Hns loadbalancer policy resource", "loadBalancer", lb)
return &loadBalancerInfo{
klog.V(1).InfoS("Created Hns loadbalancer policy resource", "loadBalancer", lb)
lbInfo := &loadBalancerInfo{
hnsID: lb.Id,
}, err
}
// Add to map of load balancers
previousLoadBalancers[id] = lbInfo
return lbInfo, err
}

func (hns hnsV2) deleteLoadBalancer(hnsID string) error {
lb, err := hcn.GetLoadBalancerByID(hnsID)
if err != nil {
Expand Down
10 changes: 8 additions & 2 deletions pkg/proxy/winkernel/hns_test.go
View file
Expand Up @@ -316,6 +316,7 @@ func testDeleteEndpoint(t *testing.T, hns HostNetworkService) {

func testGetLoadBalancerExisting(t *testing.T, hns HostNetworkService) {
Network := mustTestNetwork(t)
lbs := make(map[loadBalancerIdentifier]*(loadBalancerInfo))

ipConfig := &hcn.IpConfig{
IpAddress: epIpAddress,
Expand Down Expand Up @@ -347,13 +348,16 @@ func testGetLoadBalancerExisting(t *testing.T, hns HostNetworkService) {
if err != nil {
t.Error(err)
}
// We populate this to ensure we test for getting existing load balancer
id := loadBalancerIdentifier{protocol: protocol, internalPort: internalPort, externalPort: externalPort, vip: serviceVip, endpointsCount: len(Endpoints)}
lbs[id] = &loadBalancerInfo{hnsID: LoadBalancer.Id}

endpoint := &endpointsInfo{
ip: Endpoint.IpConfigurations[0].IpAddress,
hnsID: Endpoint.Id,
}
endpoints := []endpointsInfo{*endpoint}
lb, err := hns.getLoadBalancer(endpoints, loadBalancerFlags{}, sourceVip, serviceVip, protocol, internalPort, externalPort)
lb, err := hns.getLoadBalancer(endpoints, loadBalancerFlags{}, sourceVip, serviceVip, protocol, internalPort, externalPort, lbs)
if err != nil {
t.Error(err)
}
Expand All @@ -377,6 +381,8 @@ func testGetLoadBalancerExisting(t *testing.T, hns HostNetworkService) {
}
func testGetLoadBalancerNew(t *testing.T, hns HostNetworkService) {
Network := mustTestNetwork(t)
// We keep this empty to ensure we test for new load balancer creation.
lbs := make(map[loadBalancerIdentifier]*(loadBalancerInfo))

ipConfig := &hcn.IpConfig{
IpAddress: epIpAddress,
Expand All @@ -398,7 +404,7 @@ func testGetLoadBalancerNew(t *testing.T, hns HostNetworkService) {
hnsID: Endpoint.Id,
}
endpoints := []endpointsInfo{*endpoint}
lb, err := hns.getLoadBalancer(endpoints, loadBalancerFlags{}, sourceVip, serviceVip, protocol, internalPort, externalPort)
lb, err := hns.getLoadBalancer(endpoints, loadBalancerFlags{}, sourceVip, serviceVip, protocol, internalPort, externalPort, lbs)
if err != nil {
t.Error(err)
}
Expand Down

0 comments on commit 4094673

Please sign in to comment.