Skip to content

K
k3s
Commit c355a2ac authored by m1093782566's avatar m1093782566
Browse files
Options

Paramaterize stickyMaxAgeMinutes for service in API

parent 41aeccd0
Hide whitespace changes
Inline Side-by-side
Showing with 349 additions and 69 deletions
examples/meteor/meteor-service.json
View file @ c355a2ac
...@@ -15,6 +15,11 @@ ...@@ -15,6 +15,11 @@
"name": "meteor" "name": "meteor"
}, },
"sessionAffinity": "ClientIP", "sessionAffinity": "ClientIP",
"sessionAffinityConfig": {
"clientIP": {
"timeoutSeconds": 90
}
},
"type": "LoadBalancer" "type": "LoadBalancer"
} }
} }
pkg/api/types.go
View file @ c355a2ac
...@@ -2660,6 +2660,31 @@ const ( ...@@ -2660,6 +2660,31 @@ const (
ServiceAffinityNone ServiceAffinity = "None" ServiceAffinityNone ServiceAffinity = "None"
) )
const (
// DefaultClientIPServiceAffinitySeconds is the default timeout seconds
// of Client IP based session affinity - 3 hours.
DefaultClientIPServiceAffinitySeconds int32 = 10800
// MaxClientIPServiceAffinitySeconds is the max timeout seconds
// of Client IP based session affinity - 1 day.
MaxClientIPServiceAffinitySeconds int32 = 86400
)
// SessionAffinityConfig represents the configurations of session affinity.
type SessionAffinityConfig struct {
// clientIP contains the configurations of Client IP based session affinity.
// +optional
ClientIP *ClientIPConfig
}
// ClientIPConfig represents the configurations of Client IP based session affinity.
type ClientIPConfig struct {
// timeoutSeconds specifies the seconds of ClientIP type session sticky time.
// The value must be >0 && <=86400(for 1 day) if ServiceAffinity == "ClientIP".
// Default value is 10800(for 3 hours).
// +optional
TimeoutSeconds *int32
}
// Service Type string describes ingress methods for a service // Service Type string describes ingress methods for a service
type ServiceType string type ServiceType string
...@@ -2787,6 +2812,10 @@ type ServiceSpec struct { ...@@ -2787,6 +2812,10 @@ type ServiceSpec struct {
// +optional // +optional
SessionAffinity ServiceAffinity SessionAffinity ServiceAffinity
// sessionAffinityConfig contains the configurations of session affinity.
// +optional
SessionAffinityConfig *SessionAffinityConfig
// Optional: If specified and supported by the platform, this will restrict traffic through the cloud-provider // Optional: If specified and supported by the platform, this will restrict traffic through the cloud-provider
// load-balancer will be restricted to the specified client IPs. This field will be ignored if the // load-balancer will be restricted to the specified client IPs. This field will be ignored if the
// cloud-provider does not support the feature." // cloud-provider does not support the feature."
......
pkg/api/validation/validation.go
View file @ c355a2ac
...@@ -19,6 +19,7 @@ package validation ...@@ -19,6 +19,7 @@ package validation
import ( import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"math"
"net" "net"
"path" "path"
"path/filepath" "path/filepath"
...@@ -28,8 +29,6 @@ import ( ...@@ -28,8 +29,6 @@ import (
"github.com/golang/glog" "github.com/golang/glog"
"math"
"k8s.io/api/core/v1" "k8s.io/api/core/v1"
apiequality "k8s.io/apimachinery/pkg/api/equality" apiequality "k8s.io/apimachinery/pkg/api/equality"
"k8s.io/apimachinery/pkg/api/resource" "k8s.io/apimachinery/pkg/api/resource"
...@@ -1873,6 +1872,33 @@ func validateProbe(probe *api.Probe, fldPath *field.Path) field.ErrorList { ...@@ -1873,6 +1872,33 @@ func validateProbe(probe *api.Probe, fldPath *field.Path) field.ErrorList {
return allErrs return allErrs
} }
func validateClientIPAffinityConfig(config *api.SessionAffinityConfig, fldPath *field.Path) field.ErrorList {
allErrs := field.ErrorList{}
if config == nil {
allErrs = append(allErrs, field.Required(fldPath, fmt.Sprintf("when session affinity type is %s", api.ServiceAffinityClientIP)))
return allErrs
}
if config.ClientIP == nil {
allErrs = append(allErrs, field.Required(fldPath.Child("clientIP"), fmt.Sprintf("when session affinity type is %s", api.ServiceAffinityClientIP)))
return allErrs
}
if config.ClientIP.TimeoutSeconds == nil {
allErrs = append(allErrs, field.Required(fldPath.Child("clientIP").Child("timeoutSeconds"), fmt.Sprintf("when session affinity type is %s", api.ServiceAffinityClientIP)))
return allErrs
}
allErrs = append(allErrs, validateAffinityTimeout(config.ClientIP.TimeoutSeconds, fldPath.Child("clientIP").Child("timeoutSeconds"))...)
return allErrs
}
func validateAffinityTimeout(timeout *int32, fldPath *field.Path) field.ErrorList {
allErrs := field.ErrorList{}
if *timeout <= 0 || *timeout > api.MaxClientIPServiceAffinitySeconds {
allErrs = append(allErrs, field.Invalid(fldPath, timeout, fmt.Sprintf("must be greater than 0 and less than %d", api.MaxClientIPServiceAffinitySeconds)))
}
return allErrs
}
// AccumulateUniqueHostPorts extracts each HostPort of each Container, // AccumulateUniqueHostPorts extracts each HostPort of each Container,
// accumulating the results and returning an error if any ports conflict. // accumulating the results and returning an error if any ports conflict.
func AccumulateUniqueHostPorts(containers []api.Container, accumulator *sets.String, fldPath *field.Path) field.ErrorList { func AccumulateUniqueHostPorts(containers []api.Container, accumulator *sets.String, fldPath *field.Path) field.ErrorList {
...@@ -2914,6 +2940,14 @@ func ValidateService(service *api.Service) field.ErrorList { ...@@ -2914,6 +2940,14 @@ func ValidateService(service *api.Service) field.ErrorList {
allErrs = append(allErrs, field.NotSupported(specPath.Child("sessionAffinity"), service.Spec.SessionAffinity, supportedSessionAffinityType.List())) allErrs = append(allErrs, field.NotSupported(specPath.Child("sessionAffinity"), service.Spec.SessionAffinity, supportedSessionAffinityType.List()))
} }
if service.Spec.SessionAffinity == api.ServiceAffinityClientIP {
allErrs = append(allErrs, validateClientIPAffinityConfig(service.Spec.SessionAffinityConfig, specPath.Child("sessionAffinityConfig"))...)
} else if service.Spec.SessionAffinity == api.ServiceAffinityNone {
if service.Spec.SessionAffinityConfig != nil {
allErrs = append(allErrs, field.Forbidden(specPath.Child("sessionAffinityConfig"), fmt.Sprintf("must not be set when session affinity is %s", string(api.ServiceAffinityNone))))
}
}
if helper.IsServiceIPSet(service) { if helper.IsServiceIPSet(service) {
if ip := net.ParseIP(service.Spec.ClusterIP); ip == nil { if ip := net.ParseIP(service.Spec.ClusterIP); ip == nil {
allErrs = append(allErrs, field.Invalid(specPath.Child("clusterIP"), service.Spec.ClusterIP, "must be empty, 'None', or a valid IP address")) allErrs = append(allErrs, field.Invalid(specPath.Child("clusterIP"), service.Spec.ClusterIP, "must be empty, 'None', or a valid IP address"))
......
pkg/api/validation/validation_test.go
View file @ c355a2ac
...@@ -6792,6 +6792,32 @@ func TestValidateService(t *testing.T) { ...@@ -6792,6 +6792,32 @@ func TestValidateService(t *testing.T) {
numErrs: 0, numErrs: 0,
}, },
// ESIPP section ends. // ESIPP section ends.
{
name: "invalid timeoutSeconds field",
tweakSvc: func(s *api.Service) {
s.Spec.Type = api.ServiceTypeClusterIP
s.Spec.SessionAffinity = api.ServiceAffinityClientIP
s.Spec.SessionAffinityConfig = &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: newInt32(-1),
},
}
},
numErrs: 1,
},
{
name: "sessionAffinityConfig can't be set when session affinity is None",
tweakSvc: func(s *api.Service) {
s.Spec.Type = api.ServiceTypeLoadBalancer
s.Spec.SessionAffinity = api.ServiceAffinityNone
s.Spec.SessionAffinityConfig = &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: newInt32(90),
},
}
},
numErrs: 1,
},
} }
for _, tc := range testCases { for _, tc := range testCases {
...@@ -8193,6 +8219,11 @@ func TestValidateServiceUpdate(t *testing.T) { ...@@ -8193,6 +8219,11 @@ func TestValidateServiceUpdate(t *testing.T) {
name: "change affinity", name: "change affinity",
tweakSvc: func(oldSvc, newSvc *api.Service) { tweakSvc: func(oldSvc, newSvc *api.Service) {
newSvc.Spec.SessionAffinity = "ClientIP" newSvc.Spec.SessionAffinity = "ClientIP"
newSvc.Spec.SessionAffinityConfig = &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: newInt32(90),
},
}
}, },
numErrs: 0, numErrs: 0,
}, },
...@@ -10314,3 +10345,62 @@ func TestValidateFlexVolumeSource(t *testing.T) { ...@@ -10314,3 +10345,62 @@ func TestValidateFlexVolumeSource(t *testing.T) {
} }
} }
} }
func TestValidateOrSetClientIPAffinityConfig(t *testing.T) {
successCases := map[string]*api.SessionAffinityConfig{
"non-empty config, valid timeout: 1": {
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: newInt32(1),
},
},
"non-empty config, valid timeout: api.MaxClientIPServiceAffinitySeconds-1": {
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: newInt32(int(api.MaxClientIPServiceAffinitySeconds - 1)),
},
},
"non-empty config, valid timeout: api.MaxClientIPServiceAffinitySeconds": {
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: newInt32(int(api.MaxClientIPServiceAffinitySeconds)),
},
},
}
for name, test := range successCases {
if errs := validateClientIPAffinityConfig(test, field.NewPath("field")); len(errs) != 0 {
t.Errorf("case: %s, expected success: %v", name, errs)
}
}
errorCases := map[string]*api.SessionAffinityConfig{
"empty session affinity config": nil,
"empty client IP config": {
ClientIP: nil,
},
"empty timeoutSeconds": {
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: nil,
},
},
"non-empty config, invalid timeout: api.MaxClientIPServiceAffinitySeconds+1": {
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: newInt32(int(api.MaxClientIPServiceAffinitySeconds + 1)),
},
},
"non-empty config, invalid timeout: -1": {
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: newInt32(-1),
},
},
"non-empty config, invalid timeout: 0": {
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: newInt32(0),
},
},
}
for name, test := range errorCases {
if errs := validateClientIPAffinityConfig(test, field.NewPath("field")); len(errs) == 0 {
t.Errorf("case: %v, expected failures: %v", name, errs)
}
}
}
pkg/master/controller.go
View file @ c355a2ac
...@@ -250,6 +250,7 @@ func (c *Controller) CreateOrUpdateMasterServiceIfNeeded(serviceName string, ser ...@@ -250,6 +250,7 @@ func (c *Controller) CreateOrUpdateMasterServiceIfNeeded(serviceName string, ser
} }
return nil return nil
} }
timeoutSeconds := api.DefaultClientIPServiceAffinitySeconds
svc := &api.Service{ svc := &api.Service{
ObjectMeta: metav1.ObjectMeta{ ObjectMeta: metav1.ObjectMeta{
Name: serviceName, Name: serviceName,
...@@ -263,6 +264,11 @@ func (c *Controller) CreateOrUpdateMasterServiceIfNeeded(serviceName string, ser ...@@ -263,6 +264,11 @@ func (c *Controller) CreateOrUpdateMasterServiceIfNeeded(serviceName string, ser
ClusterIP: serviceIP.String(), ClusterIP: serviceIP.String(),
SessionAffinity: api.ServiceAffinityClientIP, SessionAffinity: api.ServiceAffinityClientIP,
Type: serviceType, Type: serviceType,
SessionAffinityConfig: &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: &timeoutSeconds,
},
},
}, },
} }
......
pkg/master/controller_test.go
View file @ c355a2ac
...@@ -546,6 +546,7 @@ func TestCreateOrUpdateMasterService(t *testing.T) { ...@@ -546,6 +546,7 @@ func TestCreateOrUpdateMasterService(t *testing.T) {
om := func(name string) metav1.ObjectMeta { om := func(name string) metav1.ObjectMeta {
return metav1.ObjectMeta{Namespace: ns, Name: name} return metav1.ObjectMeta{Namespace: ns, Name: name}
} }
timeoutSeconds := api.DefaultClientIPServiceAffinitySeconds
create_tests := []struct { create_tests := []struct {
testName string testName string
...@@ -570,7 +571,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) { ...@@ -570,7 +571,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) {
Selector: nil, Selector: nil,
ClusterIP: "1.2.3.4", ClusterIP: "1.2.3.4",
SessionAffinity: api.ServiceAffinityClientIP, SessionAffinity: api.ServiceAffinityClientIP,
Type: api.ServiceTypeClusterIP, SessionAffinityConfig: &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: &timeoutSeconds,
},
},
Type: api.ServiceTypeClusterIP,
}, },
}, },
}, },
...@@ -625,7 +631,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) { ...@@ -625,7 +631,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) {
Selector: nil, Selector: nil,
ClusterIP: "1.2.3.4", ClusterIP: "1.2.3.4",
SessionAffinity: api.ServiceAffinityClientIP, SessionAffinity: api.ServiceAffinityClientIP,
Type: api.ServiceTypeClusterIP, SessionAffinityConfig: &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: &timeoutSeconds,
},
},
Type: api.ServiceTypeClusterIP,
}, },
}, },
expectUpdate: &api.Service{ expectUpdate: &api.Service{
...@@ -637,7 +648,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) { ...@@ -637,7 +648,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) {
Selector: nil, Selector: nil,
ClusterIP: "1.2.3.4", ClusterIP: "1.2.3.4",
SessionAffinity: api.ServiceAffinityClientIP, SessionAffinity: api.ServiceAffinityClientIP,
Type: api.ServiceTypeClusterIP, SessionAffinityConfig: &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: &timeoutSeconds,
},
},
Type: api.ServiceTypeClusterIP,
}, },
}, },
}, },
...@@ -658,7 +674,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) { ...@@ -658,7 +674,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) {
Selector: nil, Selector: nil,
ClusterIP: "1.2.3.4", ClusterIP: "1.2.3.4",
SessionAffinity: api.ServiceAffinityClientIP, SessionAffinity: api.ServiceAffinityClientIP,
Type: api.ServiceTypeClusterIP, SessionAffinityConfig: &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: &timeoutSeconds,
},
},
Type: api.ServiceTypeClusterIP,
}, },
}, },
expectUpdate: &api.Service{ expectUpdate: &api.Service{
...@@ -671,7 +692,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) { ...@@ -671,7 +692,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) {
Selector: nil, Selector: nil,
ClusterIP: "1.2.3.4", ClusterIP: "1.2.3.4",
SessionAffinity: api.ServiceAffinityClientIP, SessionAffinity: api.ServiceAffinityClientIP,
Type: api.ServiceTypeClusterIP, SessionAffinityConfig: &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: &timeoutSeconds,
},
},
Type: api.ServiceTypeClusterIP,
}, },
}, },
}, },
...@@ -691,7 +717,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) { ...@@ -691,7 +717,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) {
Selector: nil, Selector: nil,
ClusterIP: "1.2.3.4", ClusterIP: "1.2.3.4",
SessionAffinity: api.ServiceAffinityClientIP, SessionAffinity: api.ServiceAffinityClientIP,
Type: api.ServiceTypeClusterIP, SessionAffinityConfig: &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: &timeoutSeconds,
},
},
Type: api.ServiceTypeClusterIP,
}, },
}, },
expectUpdate: &api.Service{ expectUpdate: &api.Service{
...@@ -703,7 +734,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) { ...@@ -703,7 +734,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) {
Selector: nil, Selector: nil,
ClusterIP: "1.2.3.4", ClusterIP: "1.2.3.4",
SessionAffinity: api.ServiceAffinityClientIP, SessionAffinity: api.ServiceAffinityClientIP,
Type: api.ServiceTypeClusterIP, SessionAffinityConfig: &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: &timeoutSeconds,
},
},
Type: api.ServiceTypeClusterIP,
}, },
}, },
}, },
...@@ -723,7 +759,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) { ...@@ -723,7 +759,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) {
Selector: nil, Selector: nil,
ClusterIP: "1.2.3.4", ClusterIP: "1.2.3.4",
SessionAffinity: api.ServiceAffinityClientIP, SessionAffinity: api.ServiceAffinityClientIP,
Type: api.ServiceTypeClusterIP, SessionAffinityConfig: &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: &timeoutSeconds,
},
},
Type: api.ServiceTypeClusterIP,
}, },
}, },
expectUpdate: &api.Service{ expectUpdate: &api.Service{
...@@ -735,7 +776,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) { ...@@ -735,7 +776,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) {
Selector: nil, Selector: nil,
ClusterIP: "1.2.3.4", ClusterIP: "1.2.3.4",
SessionAffinity: api.ServiceAffinityClientIP, SessionAffinity: api.ServiceAffinityClientIP,
Type: api.ServiceTypeClusterIP, SessionAffinityConfig: &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: &timeoutSeconds,
},
},
Type: api.ServiceTypeClusterIP,
}, },
}, },
}, },
...@@ -755,7 +801,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) { ...@@ -755,7 +801,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) {
Selector: nil, Selector: nil,
ClusterIP: "1.2.3.4", ClusterIP: "1.2.3.4",
SessionAffinity: api.ServiceAffinityClientIP, SessionAffinity: api.ServiceAffinityClientIP,
Type: api.ServiceTypeClusterIP, SessionAffinityConfig: &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: &timeoutSeconds,
},
},
Type: api.ServiceTypeClusterIP,
}, },
}, },
expectUpdate: &api.Service{ expectUpdate: &api.Service{
...@@ -767,7 +818,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) { ...@@ -767,7 +818,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) {
Selector: nil, Selector: nil,
ClusterIP: "1.2.3.4", ClusterIP: "1.2.3.4",
SessionAffinity: api.ServiceAffinityClientIP, SessionAffinity: api.ServiceAffinityClientIP,
Type: api.ServiceTypeClusterIP, SessionAffinityConfig: &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: &timeoutSeconds,
},
},
Type: api.ServiceTypeClusterIP,
}, },
}, },
}, },
...@@ -787,7 +843,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) { ...@@ -787,7 +843,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) {
Selector: nil, Selector: nil,
ClusterIP: "1.2.3.4", ClusterIP: "1.2.3.4",
SessionAffinity: api.ServiceAffinityClientIP, SessionAffinity: api.ServiceAffinityClientIP,
Type: api.ServiceTypeClusterIP, SessionAffinityConfig: &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: &timeoutSeconds,
},
},
Type: api.ServiceTypeClusterIP,
}, },
}, },
expectUpdate: &api.Service{ expectUpdate: &api.Service{
...@@ -799,7 +860,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) { ...@@ -799,7 +860,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) {
Selector: nil, Selector: nil,
ClusterIP: "1.2.3.4", ClusterIP: "1.2.3.4",
SessionAffinity: api.ServiceAffinityClientIP, SessionAffinity: api.ServiceAffinityClientIP,
Type: api.ServiceTypeClusterIP, SessionAffinityConfig: &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: &timeoutSeconds,
},
},
Type: api.ServiceTypeClusterIP,
}, },
}, },
}, },
...@@ -819,7 +885,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) { ...@@ -819,7 +885,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) {
Selector: nil, Selector: nil,
ClusterIP: "1.2.3.4", ClusterIP: "1.2.3.4",
SessionAffinity: api.ServiceAffinityClientIP, SessionAffinity: api.ServiceAffinityClientIP,
Type: api.ServiceTypeNodePort, SessionAffinityConfig: &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: &timeoutSeconds,
},
},
Type: api.ServiceTypeNodePort,
}, },
}, },
expectUpdate: &api.Service{ expectUpdate: &api.Service{
...@@ -831,7 +902,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) { ...@@ -831,7 +902,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) {
Selector: nil, Selector: nil,
ClusterIP: "1.2.3.4", ClusterIP: "1.2.3.4",
SessionAffinity: api.ServiceAffinityClientIP, SessionAffinity: api.ServiceAffinityClientIP,
Type: api.ServiceTypeClusterIP, SessionAffinityConfig: &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: &timeoutSeconds,
},
},
Type: api.ServiceTypeClusterIP,
}, },
}, },
}, },
...@@ -851,7 +927,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) { ...@@ -851,7 +927,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) {
Selector: nil, Selector: nil,
ClusterIP: "1.2.3.4", ClusterIP: "1.2.3.4",
SessionAffinity: api.ServiceAffinityClientIP, SessionAffinity: api.ServiceAffinityClientIP,
Type: api.ServiceTypeClusterIP, SessionAffinityConfig: &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: &timeoutSeconds,
},
},
Type: api.ServiceTypeClusterIP,
}, },
}, },
expectUpdate: nil, expectUpdate: nil,
...@@ -910,7 +991,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) { ...@@ -910,7 +991,12 @@ func TestCreateOrUpdateMasterService(t *testing.T) {
Selector: nil, Selector: nil,
ClusterIP: "1.2.3.4", ClusterIP: "1.2.3.4",
SessionAffinity: api.ServiceAffinityClientIP, SessionAffinity: api.ServiceAffinityClientIP,
Type: api.ServiceTypeClusterIP, SessionAffinityConfig: &api.SessionAffinityConfig{
ClientIP: &api.ClientIPConfig{
TimeoutSeconds: &timeoutSeconds,
},
},
Type: api.ServiceTypeClusterIP,
}, },
}, },
expectUpdate: nil, expectUpdate: nil,
......
pkg/proxy/iptables/proxier.go
View file @ c355a2ac
...@@ -143,7 +143,7 @@ type serviceInfo struct { ...@@ -143,7 +143,7 @@ type serviceInfo struct {
nodePort int nodePort int
loadBalancerStatus api.LoadBalancerStatus loadBalancerStatus api.LoadBalancerStatus
sessionAffinityType api.ServiceAffinity sessionAffinityType api.ServiceAffinity
stickyMaxAgeMinutes int stickyMaxAgeSeconds int
externalIPs []string externalIPs []string
loadBalancerSourceRanges []string loadBalancerSourceRanges []string
onlyNodeLocalEndpoints bool onlyNodeLocalEndpoints bool
...@@ -194,6 +194,10 @@ func newServiceInfo(svcPortName proxy.ServicePortName, port *api.ServicePort, se ...@@ -194,6 +194,10 @@ func newServiceInfo(svcPortName proxy.ServicePortName, port *api.ServicePort, se
apiservice.RequestsOnlyLocalTraffic(service) { apiservice.RequestsOnlyLocalTraffic(service) {
onlyNodeLocalEndpoints = true onlyNodeLocalEndpoints = true
} }
var stickyMaxAgeSeconds int
if service.Spec.SessionAffinity == api.ServiceAffinityClientIP {
stickyMaxAgeSeconds = int(*service.Spec.SessionAffinityConfig.ClientIP.TimeoutSeconds)
}
info := &serviceInfo{ info := &serviceInfo{
clusterIP: net.ParseIP(service.Spec.ClusterIP), clusterIP: net.ParseIP(service.Spec.ClusterIP),
port: int(port.Port), port: int(port.Port),
...@@ -202,11 +206,12 @@ func newServiceInfo(svcPortName proxy.ServicePortName, port *api.ServicePort, se ...@@ -202,11 +206,12 @@ func newServiceInfo(svcPortName proxy.ServicePortName, port *api.ServicePort, se
// Deep-copy in case the service instance changes // Deep-copy in case the service instance changes
loadBalancerStatus: *helper.LoadBalancerStatusDeepCopy(&service.Status.LoadBalancer), loadBalancerStatus: *helper.LoadBalancerStatusDeepCopy(&service.Status.LoadBalancer),
sessionAffinityType: service.Spec.SessionAffinity, sessionAffinityType: service.Spec.SessionAffinity,
stickyMaxAgeMinutes: 180, // TODO: paramaterize this in the API. stickyMaxAgeSeconds: stickyMaxAgeSeconds,
externalIPs: make([]string, len(service.Spec.ExternalIPs)), externalIPs: make([]string, len(service.Spec.ExternalIPs)),
loadBalancerSourceRanges: make([]string, len(service.Spec.LoadBalancerSourceRanges)), loadBalancerSourceRanges: make([]string, len(service.Spec.LoadBalancerSourceRanges)),
onlyNodeLocalEndpoints: onlyNodeLocalEndpoints, onlyNodeLocalEndpoints: onlyNodeLocalEndpoints,
} }
copy(info.loadBalancerSourceRanges, service.Spec.LoadBalancerSourceRanges) copy(info.loadBalancerSourceRanges, service.Spec.LoadBalancerSourceRanges)
copy(info.externalIPs, service.Spec.ExternalIPs) copy(info.externalIPs, service.Spec.ExternalIPs)
...@@ -1440,7 +1445,7 @@ func (proxier *Proxier) syncProxyRules() { ...@@ -1440,7 +1445,7 @@ func (proxier *Proxier) syncProxyRules() {
"-A", string(svcChain), "-A", string(svcChain),
"-m", "comment", "--comment", svcNameString, "-m", "comment", "--comment", svcNameString,
"-m", "recent", "--name", string(endpointChain), "-m", "recent", "--name", string(endpointChain),
"--rcheck", "--seconds", strconv.Itoa(svcInfo.stickyMaxAgeMinutes*60), "--reap", "--rcheck", "--seconds", strconv.Itoa(svcInfo.stickyMaxAgeSeconds), "--reap",
"-j", string(endpointChain)) "-j", string(endpointChain))
} }
} }
......
pkg/proxy/iptables/proxier_test.go
View file @ c355a2ac
...@@ -178,8 +178,8 @@ func TestGetChainLinesMultipleTables(t *testing.T) { ...@@ -178,8 +178,8 @@ func TestGetChainLinesMultipleTables(t *testing.T) {
func newFakeServiceInfo(service proxy.ServicePortName, ip net.IP, port int, protocol api.Protocol, onlyNodeLocalEndpoints bool) *serviceInfo { func newFakeServiceInfo(service proxy.ServicePortName, ip net.IP, port int, protocol api.Protocol, onlyNodeLocalEndpoints bool) *serviceInfo {
return &serviceInfo{ return &serviceInfo{
sessionAffinityType: api.ServiceAffinityNone, // default sessionAffinityType: api.ServiceAffinityNone, // default
stickyMaxAgeMinutes: 180, // TODO: paramaterize this in the API. stickyMaxAgeSeconds: int(api.DefaultClientIPServiceAffinitySeconds), // default
clusterIP: ip, clusterIP: ip,
port: port, port: port,
protocol: protocol, protocol: protocol,
......
pkg/proxy/userspace/loadbalancer.go
View file @ c355a2ac
...@@ -28,7 +28,7 @@ type LoadBalancer interface { ...@@ -28,7 +28,7 @@ type LoadBalancer interface {
// NextEndpoint returns the endpoint to handle a request for the given // NextEndpoint returns the endpoint to handle a request for the given
// service-port and source address. // service-port and source address.
NextEndpoint(service proxy.ServicePortName, srcAddr net.Addr, sessionAffinityReset bool) (string, error) NextEndpoint(service proxy.ServicePortName, srcAddr net.Addr, sessionAffinityReset bool) (string, error)
NewService(service proxy.ServicePortName, sessionAffinityType api.ServiceAffinity, stickyMaxAgeMinutes int) error NewService(service proxy.ServicePortName, sessionAffinityType api.ServiceAffinity, stickyMaxAgeSeconds int) error
DeleteService(service proxy.ServicePortName) DeleteService(service proxy.ServicePortName)
CleanupStaleStickySessions(service proxy.ServicePortName) CleanupStaleStickySessions(service proxy.ServicePortName)
ServiceHasEndpoints(service proxy.ServicePortName) bool ServiceHasEndpoints(service proxy.ServicePortName) bool
......
pkg/proxy/userspace/proxier.go
View file @ c355a2ac
...@@ -61,7 +61,7 @@ type ServiceInfo struct { ...@@ -61,7 +61,7 @@ type ServiceInfo struct {
nodePort int nodePort int
loadBalancerStatus api.LoadBalancerStatus loadBalancerStatus api.LoadBalancerStatus
sessionAffinityType api.ServiceAffinity sessionAffinityType api.ServiceAffinity
stickyMaxAgeMinutes int stickyMaxAgeSeconds int
// Deprecated, but required for back-compat (including e2e) // Deprecated, but required for back-compat (including e2e)
externalIPs []string externalIPs []string
} }
...@@ -378,15 +378,13 @@ func (proxier *Proxier) addServiceOnPort(service proxy.ServicePortName, protocol ...@@ -378,15 +378,13 @@ func (proxier *Proxier) addServiceOnPort(service proxy.ServicePortName, protocol
return nil, err return nil, err
} }
si := &ServiceInfo{ si := &ServiceInfo{
Timeout: timeout, Timeout: timeout,
ActiveClients: newClientCache(), ActiveClients: newClientCache(),
isAliveAtomic: 1, isAliveAtomic: 1,
proxyPort: portNum, proxyPort: portNum,
protocol: protocol, protocol: protocol,
socket: sock, socket: sock,
sessionAffinityType: api.ServiceAffinityNone, // default sessionAffinityType: api.ServiceAffinityNone, // default
stickyMaxAgeMinutes: 180, // TODO: parameterize this in the API.
} }
proxier.setServiceInfo(service, si) proxier.setServiceInfo(service, si)
...@@ -450,12 +448,17 @@ func (proxier *Proxier) mergeService(service *api.Service) sets.String { ...@@ -450,12 +448,17 @@ func (proxier *Proxier) mergeService(service *api.Service) sets.String {
info.loadBalancerStatus = *helper.LoadBalancerStatusDeepCopy(&service.Status.LoadBalancer) info.loadBalancerStatus = *helper.LoadBalancerStatusDeepCopy(&service.Status.LoadBalancer)
info.nodePort = int(servicePort.NodePort) info.nodePort = int(servicePort.NodePort)
info.sessionAffinityType = service.Spec.SessionAffinity info.sessionAffinityType = service.Spec.SessionAffinity
// Set session affinity timeout value when sessionAffinity==ClientIP
if service.Spec.SessionAffinity == api.ServiceAffinityClientIP {
info.stickyMaxAgeSeconds = int(*service.Spec.SessionAffinityConfig.ClientIP.TimeoutSeconds)
}
glog.V(4).Infof("info: %#v", info) glog.V(4).Infof("info: %#v", info)
if err := proxier.openPortal(serviceName, info); err != nil { if err := proxier.openPortal(serviceName, info); err != nil {
glog.Errorf("Failed to open portal for %q: %v", serviceName, err) glog.Errorf("Failed to open portal for %q: %v", serviceName, err)
} }
proxier.loadBalancer.NewService(serviceName, info.sessionAffinityType, info.stickyMaxAgeMinutes) proxier.loadBalancer.NewService(serviceName, info.sessionAffinityType, info.stickyMaxAgeSeconds)
} }
return existingPorts return existingPorts
......
pkg/proxy/userspace/roundrobin.go
View file @ c355a2ac
...@@ -48,7 +48,7 @@ type affinityState struct { ...@@ -48,7 +48,7 @@ type affinityState struct {
type affinityPolicy struct { type affinityPolicy struct {
affinityType api.ServiceAffinity affinityType api.ServiceAffinity
affinityMap map[string]*affinityState // map client IP -> affinity info affinityMap map[string]*affinityState // map client IP -> affinity info
ttlMinutes int ttlSeconds int
} }
// LoadBalancerRR is a round-robin load balancer. // LoadBalancerRR is a round-robin load balancer.
...@@ -66,11 +66,11 @@ type balancerState struct { ...@@ -66,11 +66,11 @@ type balancerState struct {
affinity affinityPolicy affinity affinityPolicy
} }
func newAffinityPolicy(affinityType api.ServiceAffinity, ttlMinutes int) *affinityPolicy { func newAffinityPolicy(affinityType api.ServiceAffinity, ttlSeconds int) *affinityPolicy {
return &affinityPolicy{ return &affinityPolicy{
affinityType: affinityType, affinityType: affinityType,
affinityMap: make(map[string]*affinityState), affinityMap: make(map[string]*affinityState),
ttlMinutes: ttlMinutes, ttlSeconds: ttlSeconds,
} }
} }
...@@ -81,22 +81,22 @@ func NewLoadBalancerRR() *LoadBalancerRR { ...@@ -81,22 +81,22 @@ func NewLoadBalancerRR() *LoadBalancerRR {
} }
} }
func (lb *LoadBalancerRR) NewService(svcPort proxy.ServicePortName, affinityType api.ServiceAffinity, ttlMinutes int) error { func (lb *LoadBalancerRR) NewService(svcPort proxy.ServicePortName, affinityType api.ServiceAffinity, ttlSeconds int) error {
glog.V(4).Infof("LoadBalancerRR NewService %q", svcPort) glog.V(4).Infof("LoadBalancerRR NewService %q", svcPort)
lb.lock.Lock() lb.lock.Lock()
defer lb.lock.Unlock() defer lb.lock.Unlock()
lb.newServiceInternal(svcPort, affinityType, ttlMinutes) lb.newServiceInternal(svcPort, affinityType, ttlSeconds)
return nil return nil
} }
// This assumes that lb.lock is already held. // This assumes that lb.lock is already held.
func (lb *LoadBalancerRR) newServiceInternal(svcPort proxy.ServicePortName, affinityType api.ServiceAffinity, ttlMinutes int) *balancerState { func (lb *LoadBalancerRR) newServiceInternal(svcPort proxy.ServicePortName, affinityType api.ServiceAffinity, ttlSeconds int) *balancerState {
if ttlMinutes == 0 { if ttlSeconds == 0 {
ttlMinutes = 180 //default to 3 hours if not specified. Should 0 be unlimited instead???? ttlSeconds = int(api.DefaultClientIPServiceAffinitySeconds) //default to 3 hours if not specified. Should 0 be unlimited instead????
} }
if _, exists := lb.services[svcPort]; !exists { if _, exists := lb.services[svcPort]; !exists {
lb.services[svcPort] = &balancerState{affinity: *newAffinityPolicy(affinityType, ttlMinutes)} lb.services[svcPort] = &balancerState{affinity: *newAffinityPolicy(affinityType, ttlSeconds)}
glog.V(4).Infof("LoadBalancerRR service %q did not exist, created", svcPort) glog.V(4).Infof("LoadBalancerRR service %q did not exist, created", svcPort)
} else if affinityType != "" { } else if affinityType != "" {
lb.services[svcPort].affinity.affinityType = affinityType lb.services[svcPort].affinity.affinityType = affinityType
...@@ -159,7 +159,7 @@ func (lb *LoadBalancerRR) NextEndpoint(svcPort proxy.ServicePortName, srcAddr ne ...@@ -159,7 +159,7 @@ func (lb *LoadBalancerRR) NextEndpoint(svcPort proxy.ServicePortName, srcAddr ne
} }
if !sessionAffinityReset { if !sessionAffinityReset {
sessionAffinity, exists := state.affinity.affinityMap[ipaddr] sessionAffinity, exists := state.affinity.affinityMap[ipaddr]
if exists && int(time.Now().Sub(sessionAffinity.lastUsed).Minutes()) < state.affinity.ttlMinutes { if exists && int(time.Now().Sub(sessionAffinity.lastUsed).Seconds()) < state.affinity.ttlSeconds {
// Affinity wins. // Affinity wins.
endpoint := sessionAffinity.endpoint endpoint := sessionAffinity.endpoint
sessionAffinity.lastUsed = time.Now() sessionAffinity.lastUsed = time.Now()
...@@ -378,7 +378,7 @@ func (lb *LoadBalancerRR) CleanupStaleStickySessions(svcPort proxy.ServicePortNa ...@@ -378,7 +378,7 @@ func (lb *LoadBalancerRR) CleanupStaleStickySessions(svcPort proxy.ServicePortNa
return return
} }
for ip, affinity := range state.affinity.affinityMap { for ip, affinity := range state.affinity.affinityMap {
if int(time.Now().Sub(affinity.lastUsed).Minutes()) >= state.affinity.ttlMinutes { if int(time.Now().Sub(affinity.lastUsed).Seconds()) >= state.affinity.ttlSeconds {
glog.V(4).Infof("Removing client %s from affinityMap for service %q", affinity.clientIP, svcPort) glog.V(4).Infof("Removing client %s from affinityMap for service %q", affinity.clientIP, svcPort)
delete(state.affinity.affinityMap, ip) delete(state.affinity.affinityMap, ip)
} }
......
pkg/proxy/userspace/roundrobin_test.go
View file @ c355a2ac
...@@ -357,7 +357,7 @@ func TestStickyLoadBalanceWorksWithNewServiceCalledFirst(t *testing.T) { ...@@ -357,7 +357,7 @@ func TestStickyLoadBalanceWorksWithNewServiceCalledFirst(t *testing.T) {
} }
// Call NewService() before OnEndpointsUpdate() // Call NewService() before OnEndpointsUpdate()
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, int(api.DefaultClientIPServiceAffinitySeconds))
endpoints := &api.Endpoints{ endpoints := &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
...@@ -421,7 +421,7 @@ func TestStickyLoadBalanceWorksWithNewServiceCalledSecond(t *testing.T) { ...@@ -421,7 +421,7 @@ func TestStickyLoadBalanceWorksWithNewServiceCalledSecond(t *testing.T) {
}, },
} }
loadBalancer.OnEndpointsAdd(endpoints) loadBalancer.OnEndpointsAdd(endpoints)
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, int(api.DefaultClientIPServiceAffinitySeconds))
client1 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0} client1 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0}
client2 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 2), Port: 0} client2 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 2), Port: 0}
...@@ -473,7 +473,7 @@ func TestStickyLoadBalanaceWorksWithMultipleEndpointsRemoveOne(t *testing.T) { ...@@ -473,7 +473,7 @@ func TestStickyLoadBalanaceWorksWithMultipleEndpointsRemoveOne(t *testing.T) {
t.Errorf("Didn't fail with non-existent service") t.Errorf("Didn't fail with non-existent service")
} }
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, int(api.DefaultClientIPServiceAffinitySeconds))
endpointsv1 := &api.Endpoints{ endpointsv1 := &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
...@@ -546,7 +546,7 @@ func TestStickyLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) { ...@@ -546,7 +546,7 @@ func TestStickyLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
t.Errorf("Didn't fail with non-existent service") t.Errorf("Didn't fail with non-existent service")
} }
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, int(api.DefaultClientIPServiceAffinitySeconds))
endpointsv1 := &api.Endpoints{ endpointsv1 := &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
...@@ -605,7 +605,7 @@ func TestStickyLoadBalanceWorksWithServiceRemoval(t *testing.T) { ...@@ -605,7 +605,7 @@ func TestStickyLoadBalanceWorksWithServiceRemoval(t *testing.T) {
if err == nil || len(endpoint) != 0 { if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service") t.Errorf("Didn't fail with non-existent service")
} }
loadBalancer.NewService(fooService, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(fooService, api.ServiceAffinityClientIP, int(api.DefaultClientIPServiceAffinitySeconds))
endpoints1 := &api.Endpoints{ endpoints1 := &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: fooService.Name, Namespace: fooService.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: fooService.Name, Namespace: fooService.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
...@@ -616,7 +616,7 @@ func TestStickyLoadBalanceWorksWithServiceRemoval(t *testing.T) { ...@@ -616,7 +616,7 @@ func TestStickyLoadBalanceWorksWithServiceRemoval(t *testing.T) {
}, },
} }
barService := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "bar"}, Port: ""} barService := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "bar"}, Port: ""}
loadBalancer.NewService(barService, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(barService, api.ServiceAffinityClientIP, int(api.DefaultClientIPServiceAffinitySeconds))
endpoints2 := &api.Endpoints{ endpoints2 := &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: barService.Name, Namespace: barService.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: barService.Name, Namespace: barService.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
...@@ -674,7 +674,7 @@ func TestStickyLoadBalanceWorksWithEndpointFails(t *testing.T) { ...@@ -674,7 +674,7 @@ func TestStickyLoadBalanceWorksWithEndpointFails(t *testing.T) {
} }
// Call NewService() before OnEndpointsUpdate() // Call NewService() before OnEndpointsUpdate()
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, int(api.DefaultClientIPServiceAffinitySeconds))
endpoints := &api.Endpoints{ endpoints := &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
......
pkg/proxy/winuserspace/proxier.go
View file @ c355a2ac
...@@ -36,7 +36,6 @@ import ( ...@@ -36,7 +36,6 @@ import (
) )
const allAvailableInterfaces string = "" const allAvailableInterfaces string = ""
const stickyMaxAgeMinutes int = 180 // TODO: parameterize this in the API.
type portal struct { type portal struct {
ip string ip string
...@@ -360,7 +359,11 @@ func (proxier *Proxier) mergeService(service *api.Service) map[ServicePortPortal ...@@ -360,7 +359,11 @@ func (proxier *Proxier) mergeService(service *api.Service) map[ServicePortPortal
}, },
Port: servicePort.Name, Port: servicePort.Name,
} }
proxier.loadBalancer.NewService(servicePortName, service.Spec.SessionAffinity, stickyMaxAgeMinutes) timeoutSeconds := 0
if service.Spec.SessionAffinity == api.ServiceAffinityClientIP {
timeoutSeconds = int(*service.Spec.SessionAffinityConfig.ClientIP.TimeoutSeconds)
}
proxier.loadBalancer.NewService(servicePortName, service.Spec.SessionAffinity, timeoutSeconds)
} }
} }
......
pkg/proxy/winuserspace/roundrobin.go
View file @ c355a2ac
...@@ -48,7 +48,7 @@ type affinityState struct { ...@@ -48,7 +48,7 @@ type affinityState struct {
type affinityPolicy struct { type affinityPolicy struct {
affinityType api.ServiceAffinity affinityType api.ServiceAffinity
affinityMap map[string]*affinityState // map client IP -> affinity info affinityMap map[string]*affinityState // map client IP -> affinity info
ttlMinutes int ttlSeconds int
} }
// LoadBalancerRR is a round-robin load balancer. // LoadBalancerRR is a round-robin load balancer.
...@@ -66,11 +66,11 @@ type balancerState struct { ...@@ -66,11 +66,11 @@ type balancerState struct {
affinity affinityPolicy affinity affinityPolicy
} }
func newAffinityPolicy(affinityType api.ServiceAffinity, ttlMinutes int) *affinityPolicy { func newAffinityPolicy(affinityType api.ServiceAffinity, ttlSeconds int) *affinityPolicy {
return &affinityPolicy{ return &affinityPolicy{
affinityType: affinityType, affinityType: affinityType,
affinityMap: make(map[string]*affinityState), affinityMap: make(map[string]*affinityState),
ttlMinutes: ttlMinutes, ttlSeconds: ttlSeconds,
} }
} }
...@@ -81,22 +81,22 @@ func NewLoadBalancerRR() *LoadBalancerRR { ...@@ -81,22 +81,22 @@ func NewLoadBalancerRR() *LoadBalancerRR {
} }
} }
func (lb *LoadBalancerRR) NewService(svcPort proxy.ServicePortName, affinityType api.ServiceAffinity, ttlMinutes int) error { func (lb *LoadBalancerRR) NewService(svcPort proxy.ServicePortName, affinityType api.ServiceAffinity, ttlSeconds int) error {
glog.V(4).Infof("LoadBalancerRR NewService %q", svcPort) glog.V(4).Infof("LoadBalancerRR NewService %q", svcPort)
lb.lock.Lock() lb.lock.Lock()
defer lb.lock.Unlock() defer lb.lock.Unlock()
lb.newServiceInternal(svcPort, affinityType, ttlMinutes) lb.newServiceInternal(svcPort, affinityType, ttlSeconds)
return nil return nil
} }
// This assumes that lb.lock is already held. // This assumes that lb.lock is already held.
func (lb *LoadBalancerRR) newServiceInternal(svcPort proxy.ServicePortName, affinityType api.ServiceAffinity, ttlMinutes int) *balancerState { func (lb *LoadBalancerRR) newServiceInternal(svcPort proxy.ServicePortName, affinityType api.ServiceAffinity, ttlSeconds int) *balancerState {
if ttlMinutes == 0 { if ttlSeconds == 0 {
ttlMinutes = 180 //default to 3 hours if not specified. Should 0 be unlimited instead???? ttlSeconds = int(api.DefaultClientIPServiceAffinitySeconds) //default to 3 hours if not specified. Should 0 be unlimited instead????
} }
if _, exists := lb.services[svcPort]; !exists { if _, exists := lb.services[svcPort]; !exists {
lb.services[svcPort] = &balancerState{affinity: *newAffinityPolicy(affinityType, ttlMinutes)} lb.services[svcPort] = &balancerState{affinity: *newAffinityPolicy(affinityType, ttlSeconds)}
glog.V(4).Infof("LoadBalancerRR service %q did not exist, created", svcPort) glog.V(4).Infof("LoadBalancerRR service %q did not exist, created", svcPort)
} else if affinityType != "" { } else if affinityType != "" {
lb.services[svcPort].affinity.affinityType = affinityType lb.services[svcPort].affinity.affinityType = affinityType
...@@ -149,7 +149,7 @@ func (lb *LoadBalancerRR) NextEndpoint(svcPort proxy.ServicePortName, srcAddr ne ...@@ -149,7 +149,7 @@ func (lb *LoadBalancerRR) NextEndpoint(svcPort proxy.ServicePortName, srcAddr ne
} }
if !sessionAffinityReset { if !sessionAffinityReset {
sessionAffinity, exists := state.affinity.affinityMap[ipaddr] sessionAffinity, exists := state.affinity.affinityMap[ipaddr]
if exists && int(time.Now().Sub(sessionAffinity.lastUsed).Minutes()) < state.affinity.ttlMinutes { if exists && int(time.Now().Sub(sessionAffinity.lastUsed).Seconds()) < state.affinity.ttlSeconds {
// Affinity wins. // Affinity wins.
endpoint := sessionAffinity.endpoint endpoint := sessionAffinity.endpoint
sessionAffinity.lastUsed = time.Now() sessionAffinity.lastUsed = time.Now()
...@@ -366,7 +366,7 @@ func (lb *LoadBalancerRR) CleanupStaleStickySessions(svcPort proxy.ServicePortNa ...@@ -366,7 +366,7 @@ func (lb *LoadBalancerRR) CleanupStaleStickySessions(svcPort proxy.ServicePortNa
return return
} }
for ip, affinity := range state.affinity.affinityMap { for ip, affinity := range state.affinity.affinityMap {
if int(time.Now().Sub(affinity.lastUsed).Minutes()) >= state.affinity.ttlMinutes { if int(time.Now().Sub(affinity.lastUsed).Seconds()) >= state.affinity.ttlSeconds {
glog.V(4).Infof("Removing client %s from affinityMap for service %q", affinity.clientIP, svcPort) glog.V(4).Infof("Removing client %s from affinityMap for service %q", affinity.clientIP, svcPort)
delete(state.affinity.affinityMap, ip) delete(state.affinity.affinityMap, ip)
} }
......
pkg/proxy/winuserspace/roundrobin_test.go
View file @ c355a2ac
...@@ -357,7 +357,7 @@ func TestStickyLoadBalanceWorksWithNewServiceCalledFirst(t *testing.T) { ...@@ -357,7 +357,7 @@ func TestStickyLoadBalanceWorksWithNewServiceCalledFirst(t *testing.T) {
} }
// Call NewService() before OnEndpointsUpdate() // Call NewService() before OnEndpointsUpdate()
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, int(api.DefaultClientIPServiceAffinitySeconds))
endpoints := &api.Endpoints{ endpoints := &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
...@@ -421,7 +421,7 @@ func TestStickyLoadBalanceWorksWithNewServiceCalledSecond(t *testing.T) { ...@@ -421,7 +421,7 @@ func TestStickyLoadBalanceWorksWithNewServiceCalledSecond(t *testing.T) {
}, },
} }
loadBalancer.OnEndpointsAdd(endpoints) loadBalancer.OnEndpointsAdd(endpoints)
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, int(api.DefaultClientIPServiceAffinitySeconds))
client1 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0} client1 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0}
client2 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 2), Port: 0} client2 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 2), Port: 0}
...@@ -473,7 +473,7 @@ func TestStickyLoadBalanaceWorksWithMultipleEndpointsRemoveOne(t *testing.T) { ...@@ -473,7 +473,7 @@ func TestStickyLoadBalanaceWorksWithMultipleEndpointsRemoveOne(t *testing.T) {
t.Errorf("Didn't fail with non-existent service") t.Errorf("Didn't fail with non-existent service")
} }
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, int(api.DefaultClientIPServiceAffinitySeconds))
endpointsv1 := &api.Endpoints{ endpointsv1 := &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
...@@ -546,7 +546,7 @@ func TestStickyLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) { ...@@ -546,7 +546,7 @@ func TestStickyLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
t.Errorf("Didn't fail with non-existent service") t.Errorf("Didn't fail with non-existent service")
} }
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, int(api.DefaultClientIPServiceAffinitySeconds))
endpointsv1 := &api.Endpoints{ endpointsv1 := &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
...@@ -605,7 +605,7 @@ func TestStickyLoadBalanceWorksWithServiceRemoval(t *testing.T) { ...@@ -605,7 +605,7 @@ func TestStickyLoadBalanceWorksWithServiceRemoval(t *testing.T) {
if err == nil || len(endpoint) != 0 { if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service") t.Errorf("Didn't fail with non-existent service")
} }
loadBalancer.NewService(fooService, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(fooService, api.ServiceAffinityClientIP, int(api.DefaultClientIPServiceAffinitySeconds))
endpoints1 := &api.Endpoints{ endpoints1 := &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: fooService.Name, Namespace: fooService.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: fooService.Name, Namespace: fooService.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
...@@ -616,7 +616,7 @@ func TestStickyLoadBalanceWorksWithServiceRemoval(t *testing.T) { ...@@ -616,7 +616,7 @@ func TestStickyLoadBalanceWorksWithServiceRemoval(t *testing.T) {
}, },
} }
barService := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "bar"}, Port: ""} barService := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "bar"}, Port: ""}
loadBalancer.NewService(barService, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(barService, api.ServiceAffinityClientIP, int(api.DefaultClientIPServiceAffinitySeconds))
endpoints2 := &api.Endpoints{ endpoints2 := &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: barService.Name, Namespace: barService.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: barService.Name, Namespace: barService.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
...@@ -674,7 +674,7 @@ func TestStickyLoadBalanceWorksWithEndpointFails(t *testing.T) { ...@@ -674,7 +674,7 @@ func TestStickyLoadBalanceWorksWithEndpointFails(t *testing.T) {
} }
// Call NewService() before OnEndpointsUpdate() // Call NewService() before OnEndpointsUpdate()
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, int(api.DefaultClientIPServiceAffinitySeconds))
endpoints := &api.Endpoints{ endpoints := &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
......
staging/src/k8s.io/api/core/v1/types.go
View file @ c355a2ac
...@@ -2998,6 +2998,22 @@ const ( ...@@ -2998,6 +2998,22 @@ const (
ServiceAffinityNone ServiceAffinity = "None" ServiceAffinityNone ServiceAffinity = "None"
) )
// SessionAffinityConfig represents the configurations of session affinity.
type SessionAffinityConfig struct {
// clientIP contains the configurations of Client IP based session affinity.
// +optional
ClientIP *ClientIPConfig `json:"clientIP,omitempty" protobuf:"bytes,1,opt,name=clientIP"`
}
// ClientIPConfig represents the configurations of Client IP based session affinity.
type ClientIPConfig struct {
// timeoutSeconds specifies the seconds of ClientIP type session sticky time.
// The value must be >0 && <=86400(for 1 day) if ServiceAffinity == "ClientIP".
// Default value is 10800(for 3 hours).
// +optional
TimeoutSeconds *int32 `json:"timeoutSeconds,omitempty" protobuf:"varint,1,opt,name=timeoutSeconds"`
}
// Service Type string describes ingress methods for a service // Service Type string describes ingress methods for a service
type ServiceType string type ServiceType string
...@@ -3172,6 +3188,9 @@ type ServiceSpec struct { ...@@ -3172,6 +3188,9 @@ type ServiceSpec struct {
// field. // field.
// +optional // +optional
PublishNotReadyAddresses bool `json:"publishNotReadyAddresses,omitempty" protobuf:"varint,13,opt,name=publishNotReadyAddresses"` PublishNotReadyAddresses bool `json:"publishNotReadyAddresses,omitempty" protobuf:"varint,13,opt,name=publishNotReadyAddresses"`
// sessionAffinityConfig contains the configurations of session affinity.
// +optional
SessionAffinityConfig *SessionAffinityConfig `json:"sessionAffinityConfig,omitempty" protobuf:"bytes,14,opt,name=sessionAffinityConfig"`
} }
// ServicePort contains information on service's port. // ServicePort contains information on service's port.
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment