1 Star 0 Fork 0

zhuchance/kubernetes

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
克隆/下载
aws_loadbalancer.go 17.75 KB
一键复制 编辑 原始数据 按行查看 历史
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544
/*
Copyright 2014 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 aws
import (
"fmt"
"reflect"
"strconv"
"strings"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/aws/aws-sdk-go/service/elb"
"github.com/golang/glog"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/kubernetes/pkg/api/v1"
)
const ProxyProtocolPolicyName = "k8s-proxyprotocol-enabled"
// getLoadBalancerAdditionalTags converts the comma separated list of key-value
// pairs in the ServiceAnnotationLoadBalancerAdditionalTags annotation and returns
// it as a map.
func getLoadBalancerAdditionalTags(annotations map[string]string) map[string]string {
additionalTags := make(map[string]string)
if additionalTagsList, ok := annotations[ServiceAnnotationLoadBalancerAdditionalTags]; ok {
additionalTagsList = strings.TrimSpace(additionalTagsList)
// Break up list of "Key1=Val,Key2=Val2"
tagList := strings.Split(additionalTagsList, ",")
// Break up "Key=Val"
for _, tagSet := range tagList {
tag := strings.Split(strings.TrimSpace(tagSet), "=")
// Accept "Key=val" or "Key=" or just "Key"
if len(tag) >= 2 && len(tag[0]) != 0 {
// There is a key and a value, so save it
additionalTags[tag[0]] = tag[1]
} else if len(tag) == 1 && len(tag[0]) != 0 {
// Just "Key"
additionalTags[tag[0]] = ""
}
}
}
return additionalTags
}
func (c *Cloud) ensureLoadBalancer(namespacedName types.NamespacedName, loadBalancerName string, listeners []*elb.Listener, subnetIDs []string, securityGroupIDs []string, internalELB, proxyProtocol bool, loadBalancerAttributes *elb.LoadBalancerAttributes, annotations map[string]string) (*elb.LoadBalancerDescription, error) {
loadBalancer, err := c.describeLoadBalancer(loadBalancerName)
if err != nil {
return nil, err
}
dirty := false
if loadBalancer == nil {
createRequest := &elb.CreateLoadBalancerInput{}
createRequest.LoadBalancerName = aws.String(loadBalancerName)
createRequest.Listeners = listeners
if internalELB {
createRequest.Scheme = aws.String("internal")
}
// We are supposed to specify one subnet per AZ.
// TODO: What happens if we have more than one subnet per AZ?
createRequest.Subnets = stringPointerArray(subnetIDs)
createRequest.SecurityGroups = stringPointerArray(securityGroupIDs)
// Get additional tags set by the user
tags := getLoadBalancerAdditionalTags(annotations)
// Add default tags
tags[TagNameKubernetesService] = namespacedName.String()
tags = c.tagging.buildTags(ResourceLifecycleOwned, tags)
for k, v := range tags {
createRequest.Tags = append(createRequest.Tags, &elb.Tag{
Key: aws.String(k), Value: aws.String(v),
})
}
glog.Infof("Creating load balancer for %v with name: %s", namespacedName, loadBalancerName)
_, err := c.elb.CreateLoadBalancer(createRequest)
if err != nil {
return nil, err
}
if proxyProtocol {
err = c.createProxyProtocolPolicy(loadBalancerName)
if err != nil {
return nil, err
}
for _, listener := range listeners {
glog.V(2).Infof("Adjusting AWS loadbalancer proxy protocol on node port %d. Setting to true", *listener.InstancePort)
err := c.setBackendPolicies(loadBalancerName, *listener.InstancePort, []*string{aws.String(ProxyProtocolPolicyName)})
if err != nil {
return nil, err
}
}
}
dirty = true
} else {
// TODO: Sync internal vs non-internal
{
// Sync subnets
expected := sets.NewString(subnetIDs...)
actual := stringSetFromPointers(loadBalancer.Subnets)
additions := expected.Difference(actual)
removals := actual.Difference(expected)
if removals.Len() != 0 {
request := &elb.DetachLoadBalancerFromSubnetsInput{}
request.LoadBalancerName = aws.String(loadBalancerName)
request.Subnets = stringSetToPointers(removals)
glog.V(2).Info("Detaching load balancer from removed subnets")
_, err := c.elb.DetachLoadBalancerFromSubnets(request)
if err != nil {
return nil, fmt.Errorf("error detaching AWS loadbalancer from subnets: %v", err)
}
dirty = true
}
if additions.Len() != 0 {
request := &elb.AttachLoadBalancerToSubnetsInput{}
request.LoadBalancerName = aws.String(loadBalancerName)
request.Subnets = stringSetToPointers(additions)
glog.V(2).Info("Attaching load balancer to added subnets")
_, err := c.elb.AttachLoadBalancerToSubnets(request)
if err != nil {
return nil, fmt.Errorf("error attaching AWS loadbalancer to subnets: %v", err)
}
dirty = true
}
}
{
// Sync security groups
expected := sets.NewString(securityGroupIDs...)
actual := stringSetFromPointers(loadBalancer.SecurityGroups)
if !expected.Equal(actual) {
// This call just replaces the security groups, unlike e.g. subnets (!)
request := &elb.ApplySecurityGroupsToLoadBalancerInput{}
request.LoadBalancerName = aws.String(loadBalancerName)
request.SecurityGroups = stringPointerArray(securityGroupIDs)
glog.V(2).Info("Applying updated security groups to load balancer")
_, err := c.elb.ApplySecurityGroupsToLoadBalancer(request)
if err != nil {
return nil, fmt.Errorf("error applying AWS loadbalancer security groups: %v", err)
}
dirty = true
}
}
{
// Sync listeners
listenerDescriptions := loadBalancer.ListenerDescriptions
foundSet := make(map[int]bool)
removals := []*int64{}
for _, listenerDescription := range listenerDescriptions {
actual := listenerDescription.Listener
if actual == nil {
glog.Warning("Ignoring empty listener in AWS loadbalancer: ", loadBalancerName)
continue
}
found := -1
for i, expected := range listeners {
if elbProtocolsAreEqual(actual.Protocol, expected.Protocol) {
continue
}
if elbProtocolsAreEqual(actual.InstanceProtocol, expected.InstanceProtocol) {
continue
}
if orZero(actual.InstancePort) != orZero(expected.InstancePort) {
continue
}
if orZero(actual.LoadBalancerPort) != orZero(expected.LoadBalancerPort) {
continue
}
if awsArnEquals(actual.SSLCertificateId, expected.SSLCertificateId) {
continue
}
found = i
}
if found != -1 {
foundSet[found] = true
} else {
removals = append(removals, actual.LoadBalancerPort)
}
}
additions := []*elb.Listener{}
for i := range listeners {
if foundSet[i] {
continue
}
additions = append(additions, listeners[i])
}
if len(removals) != 0 {
request := &elb.DeleteLoadBalancerListenersInput{}
request.LoadBalancerName = aws.String(loadBalancerName)
request.LoadBalancerPorts = removals
glog.V(2).Info("Deleting removed load balancer listeners")
_, err := c.elb.DeleteLoadBalancerListeners(request)
if err != nil {
return nil, fmt.Errorf("error deleting AWS loadbalancer listeners: %v", err)
}
dirty = true
}
if len(additions) != 0 {
request := &elb.CreateLoadBalancerListenersInput{}
request.LoadBalancerName = aws.String(loadBalancerName)
request.Listeners = additions
glog.V(2).Info("Creating added load balancer listeners")
_, err := c.elb.CreateLoadBalancerListeners(request)
if err != nil {
return nil, fmt.Errorf("error creating AWS loadbalancer listeners: %v", err)
}
dirty = true
}
}
{
// Sync proxy protocol state for new and existing listeners
proxyPolicies := make([]*string, 0)
if proxyProtocol {
// Ensure the backend policy exists
// NOTE The documentation for the AWS API indicates we could get an HTTP 400
// back if a policy of the same name already exists. However, the aws-sdk does not
// seem to return an error to us in these cases. Therefore, this will issue an API
// request every time.
err := c.createProxyProtocolPolicy(loadBalancerName)
if err != nil {
return nil, err
}
proxyPolicies = append(proxyPolicies, aws.String(ProxyProtocolPolicyName))
}
foundBackends := make(map[int64]bool)
proxyProtocolBackends := make(map[int64]bool)
for _, backendListener := range loadBalancer.BackendServerDescriptions {
foundBackends[*backendListener.InstancePort] = false
proxyProtocolBackends[*backendListener.InstancePort] = proxyProtocolEnabled(backendListener)
}
for _, listener := range listeners {
setPolicy := false
instancePort := *listener.InstancePort
if currentState, ok := proxyProtocolBackends[instancePort]; !ok {
// This is a new ELB backend so we only need to worry about
// potentially adding a policy and not removing an
// existing one
setPolicy = proxyProtocol
} else {
foundBackends[instancePort] = true
// This is an existing ELB backend so we need to determine
// if the state changed
setPolicy = (currentState != proxyProtocol)
}
if setPolicy {
glog.V(2).Infof("Adjusting AWS loadbalancer proxy protocol on node port %d. Setting to %t", instancePort, proxyProtocol)
err := c.setBackendPolicies(loadBalancerName, instancePort, proxyPolicies)
if err != nil {
return nil, err
}
dirty = true
}
}
// We now need to figure out if any backend policies need removed
// because these old policies will stick around even if there is no
// corresponding listener anymore
for instancePort, found := range foundBackends {
if !found {
glog.V(2).Infof("Adjusting AWS loadbalancer proxy protocol on node port %d. Setting to false", instancePort)
err := c.setBackendPolicies(loadBalancerName, instancePort, []*string{})
if err != nil {
return nil, err
}
dirty = true
}
}
}
}
// Whether the ELB was new or existing, sync attributes regardless. This accounts for things
// that cannot be specified at the time of creation and can only be modified after the fact,
// e.g. idle connection timeout.
{
describeAttributesRequest := &elb.DescribeLoadBalancerAttributesInput{}
describeAttributesRequest.LoadBalancerName = aws.String(loadBalancerName)
describeAttributesOutput, err := c.elb.DescribeLoadBalancerAttributes(describeAttributesRequest)
if err != nil {
glog.Warning("Unable to retrieve load balancer attributes during attribute sync")
return nil, err
}
foundAttributes := &describeAttributesOutput.LoadBalancerAttributes
// Update attributes if they're dirty
if !reflect.DeepEqual(loadBalancerAttributes, foundAttributes) {
glog.V(2).Infof("Updating load-balancer attributes for %q", loadBalancerName)
modifyAttributesRequest := &elb.ModifyLoadBalancerAttributesInput{}
modifyAttributesRequest.LoadBalancerName = aws.String(loadBalancerName)
modifyAttributesRequest.LoadBalancerAttributes = loadBalancerAttributes
_, err = c.elb.ModifyLoadBalancerAttributes(modifyAttributesRequest)
if err != nil {
return nil, fmt.Errorf("Unable to update load balancer attributes during attribute sync: %v", err)
}
dirty = true
}
}
if dirty {
loadBalancer, err = c.describeLoadBalancer(loadBalancerName)
if err != nil {
glog.Warning("Unable to retrieve load balancer after creation/update")
return nil, err
}
}
return loadBalancer, nil
}
// elbProtocolsAreEqual checks if two ELB protocol strings are considered the same
// Comparison is case insensitive
func elbProtocolsAreEqual(l, r *string) bool {
if l == nil || r == nil {
return l == r
}
return strings.EqualFold(aws.StringValue(l), aws.StringValue(r))
}
// awsArnEquals checks if two ARN strings are considered the same
// Comparison is case insensitive
func awsArnEquals(l, r *string) bool {
if l == nil || r == nil {
return l == r
}
return strings.EqualFold(aws.StringValue(l), aws.StringValue(r))
}
// Makes sure that the health check for an ELB matches the configured health check node port
func (c *Cloud) ensureLoadBalancerHealthCheck(loadBalancer *elb.LoadBalancerDescription, protocol string, port int32, path string) error {
name := aws.StringValue(loadBalancer.LoadBalancerName)
actual := loadBalancer.HealthCheck
// Default AWS settings
expectedHealthyThreshold := int64(2)
expectedUnhealthyThreshold := int64(6)
expectedTimeout := int64(5)
expectedInterval := int64(10)
expectedTarget := protocol + ":" + strconv.FormatInt(int64(port), 10) + path
if expectedTarget == orEmpty(actual.Target) &&
expectedHealthyThreshold == orZero(actual.HealthyThreshold) &&
expectedUnhealthyThreshold == orZero(actual.UnhealthyThreshold) &&
expectedTimeout == orZero(actual.Timeout) &&
expectedInterval == orZero(actual.Interval) {
return nil
}
glog.V(2).Infof("Updating load-balancer health-check for %q", name)
healthCheck := &elb.HealthCheck{}
healthCheck.HealthyThreshold = &expectedHealthyThreshold
healthCheck.UnhealthyThreshold = &expectedUnhealthyThreshold
healthCheck.Timeout = &expectedTimeout
healthCheck.Interval = &expectedInterval
healthCheck.Target = &expectedTarget
request := &elb.ConfigureHealthCheckInput{}
request.HealthCheck = healthCheck
request.LoadBalancerName = loadBalancer.LoadBalancerName
_, err := c.elb.ConfigureHealthCheck(request)
if err != nil {
return fmt.Errorf("error configuring load-balancer health-check for %q: %v", name, err)
}
return nil
}
// Makes sure that exactly the specified hosts are registered as instances with the load balancer
func (c *Cloud) ensureLoadBalancerInstances(loadBalancerName string, lbInstances []*elb.Instance, instanceIDs map[awsInstanceID]*ec2.Instance) error {
expected := sets.NewString()
for id := range instanceIDs {
expected.Insert(string(id))
}
actual := sets.NewString()
for _, lbInstance := range lbInstances {
actual.Insert(orEmpty(lbInstance.InstanceId))
}
additions := expected.Difference(actual)
removals := actual.Difference(expected)
addInstances := []*elb.Instance{}
for _, instanceId := range additions.List() {
addInstance := &elb.Instance{}
addInstance.InstanceId = aws.String(instanceId)
addInstances = append(addInstances, addInstance)
}
removeInstances := []*elb.Instance{}
for _, instanceId := range removals.List() {
removeInstance := &elb.Instance{}
removeInstance.InstanceId = aws.String(instanceId)
removeInstances = append(removeInstances, removeInstance)
}
if len(addInstances) > 0 {
registerRequest := &elb.RegisterInstancesWithLoadBalancerInput{}
registerRequest.Instances = addInstances
registerRequest.LoadBalancerName = aws.String(loadBalancerName)
_, err := c.elb.RegisterInstancesWithLoadBalancer(registerRequest)
if err != nil {
return err
}
glog.V(1).Infof("Instances added to load-balancer %s", loadBalancerName)
}
if len(removeInstances) > 0 {
deregisterRequest := &elb.DeregisterInstancesFromLoadBalancerInput{}
deregisterRequest.Instances = removeInstances
deregisterRequest.LoadBalancerName = aws.String(loadBalancerName)
_, err := c.elb.DeregisterInstancesFromLoadBalancer(deregisterRequest)
if err != nil {
return err
}
glog.V(1).Infof("Instances removed from load-balancer %s", loadBalancerName)
}
return nil
}
func (c *Cloud) createProxyProtocolPolicy(loadBalancerName string) error {
request := &elb.CreateLoadBalancerPolicyInput{
LoadBalancerName: aws.String(loadBalancerName),
PolicyName: aws.String(ProxyProtocolPolicyName),
PolicyTypeName: aws.String("ProxyProtocolPolicyType"),
PolicyAttributes: []*elb.PolicyAttribute{
{
AttributeName: aws.String("ProxyProtocol"),
AttributeValue: aws.String("true"),
},
},
}
glog.V(2).Info("Creating proxy protocol policy on load balancer")
_, err := c.elb.CreateLoadBalancerPolicy(request)
if err != nil {
return fmt.Errorf("error creating proxy protocol policy on load balancer: %v", err)
}
return nil
}
func (c *Cloud) setBackendPolicies(loadBalancerName string, instancePort int64, policies []*string) error {
request := &elb.SetLoadBalancerPoliciesForBackendServerInput{
InstancePort: aws.Int64(instancePort),
LoadBalancerName: aws.String(loadBalancerName),
PolicyNames: policies,
}
if len(policies) > 0 {
glog.V(2).Infof("Adding AWS loadbalancer backend policies on node port %d", instancePort)
} else {
glog.V(2).Infof("Removing AWS loadbalancer backend policies on node port %d", instancePort)
}
_, err := c.elb.SetLoadBalancerPoliciesForBackendServer(request)
if err != nil {
return fmt.Errorf("error adjusting AWS loadbalancer backend policies: %v", err)
}
return nil
}
func proxyProtocolEnabled(backend *elb.BackendServerDescription) bool {
for _, policy := range backend.PolicyNames {
if aws.StringValue(policy) == ProxyProtocolPolicyName {
return true
}
}
return false
}
// findInstancesForELB gets the EC2 instances corresponding to the Nodes, for setting up an ELB
// We ignore Nodes (with a log message) where the instanceid cannot be determined from the provider,
// and we ignore instances which are not found
func (c *Cloud) findInstancesForELB(nodes []*v1.Node) (map[awsInstanceID]*ec2.Instance, error) {
// Map to instance ids ignoring Nodes where we cannot find the id (but logging)
instanceIDs := mapToAWSInstanceIDsTolerant(nodes)
cacheCriteria := cacheCriteria{
// MaxAge not required, because we only care about security groups, which should not change
HasInstances: instanceIDs, // Refresh if any of the instance ids are missing
}
snapshot, err := c.instanceCache.describeAllInstancesCached(cacheCriteria)
if err != nil {
return nil, err
}
instances := snapshot.FindInstances(instanceIDs)
// We ignore instances that cannot be found
return instances, nil
}
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
Go
1
https://gitee.com/meoom/kubernetes.git
git@gitee.com:meoom/kubernetes.git
meoom
kubernetes
kubernetes
v1.7.8

搜索帮助

Cb406eda 1850385 E526c682 1850385