proxier_test.go 114 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490249124922493249424952496249724982499250025012502250325042505250625072508250925102511251225132514251525162517251825192520252125222523252425252526252725282529253025312532253325342535253625372538253925402541254225432544254525462547254825492550255125522553255425552556255725582559256025612562256325642565256625672568256925702571257225732574257525762577257825792580258125822583258425852586258725882589259025912592259325942595259625972598259926002601260226032604260526062607260826092610261126122613261426152616261726182619262026212622262326242625262626272628262926302631263226332634263526362637263826392640264126422643264426452646264726482649265026512652265326542655265626572658265926602661266226632664266526662667266826692670267126722673267426752676267726782679268026812682268326842685268626872688268926902691269226932694269526962697269826992700270127022703270427052706270727082709271027112712271327142715271627172718271927202721272227232724272527262727272827292730273127322733273427352736273727382739274027412742274327442745274627472748274927502751275227532754275527562757275827592760276127622763276427652766276727682769277027712772277327742775277627772778277927802781278227832784278527862787278827892790279127922793279427952796279727982799280028012802280328042805280628072808280928102811281228132814281528162817281828192820282128222823282428252826282728282829283028312832283328342835283628372838283928402841284228432844284528462847284828492850285128522853285428552856285728582859286028612862286328642865286628672868286928702871287228732874287528762877287828792880288128822883288428852886288728882889289028912892289328942895289628972898289929002901290229032904290529062907290829092910291129122913291429152916291729182919292029212922292329242925292629272928292929302931293229332934293529362937293829392940294129422943294429452946294729482949295029512952295329542955295629572958295929602961296229632964296529662967296829692970297129722973297429752976297729782979298029812982298329842985298629872988298929902991299229932994299529962997299829993000300130023003300430053006300730083009301030113012301330143015301630173018301930203021302230233024302530263027302830293030303130323033303430353036303730383039304030413042304330443045304630473048304930503051305230533054305530563057305830593060306130623063306430653066306730683069307030713072307330743075307630773078307930803081308230833084308530863087308830893090309130923093309430953096309730983099310031013102310331043105310631073108310931103111311231133114311531163117311831193120312131223123312431253126312731283129313031313132313331343135313631373138313931403141314231433144314531463147314831493150315131523153315431553156315731583159316031613162316331643165316631673168316931703171317231733174317531763177317831793180318131823183318431853186318731883189319031913192319331943195319631973198319932003201320232033204320532063207320832093210321132123213321432153216321732183219322032213222322332243225322632273228322932303231323232333234323532363237323832393240324132423243324432453246324732483249325032513252325332543255325632573258325932603261326232633264326532663267326832693270327132723273327432753276327732783279328032813282328332843285328632873288328932903291329232933294329532963297329832993300330133023303330433053306330733083309331033113312331333143315331633173318331933203321332233233324332533263327332833293330333133323333333433353336333733383339334033413342334333443345334633473348334933503351335233533354335533563357335833593360336133623363336433653366336733683369337033713372337333743375337633773378337933803381338233833384338533863387338833893390339133923393339433953396339733983399340034013402340334043405340634073408340934103411341234133414341534163417341834193420342134223423342434253426342734283429343034313432343334343435343634373438343934403441344234433444344534463447344834493450345134523453345434553456345734583459346034613462346334643465346634673468346934703471347234733474347534763477347834793480348134823483348434853486348734883489349034913492349334943495349634973498349935003501350235033504350535063507350835093510351135123513351435153516351735183519352035213522352335243525352635273528352935303531353235333534353535363537353835393540354135423543354435453546354735483549355035513552355335543555355635573558355935603561356235633564356535663567356835693570357135723573357435753576357735783579358035813582358335843585358635873588358935903591359235933594359535963597359835993600360136023603360436053606360736083609361036113612361336143615361636173618361936203621362236233624362536263627362836293630363136323633363436353636363736383639364036413642364336443645364636473648364936503651365236533654365536563657365836593660366136623663366436653666366736683669367036713672367336743675367636773678367936803681368236833684368536863687368836893690369136923693369436953696369736983699370037013702370337043705370637073708370937103711371237133714371537163717371837193720372137223723372437253726372737283729373037313732373337343735373637373738373937403741374237433744374537463747374837493750375137523753375437553756375737583759376037613762376337643765376637673768376937703771377237733774377537763777377837793780378137823783378437853786378737883789379037913792379337943795379637973798379938003801380238033804
  1. /*
  2. Copyright 2017 The Kubernetes Authors.
  3. Licensed under the Apache License, Version 2.0 (the "License");
  4. you may not use this file except in compliance with the License.
  5. You may obtain a copy of the License at
  6. http://www.apache.org/licenses/LICENSE-2.0
  7. Unless required by applicable law or agreed to in writing, software
  8. distributed under the License is distributed on an "AS IS" BASIS,
  9. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  10. See the License for the specific language governing permissions and
  11. limitations under the License.
  12. */
  13. package ipvs
  14. import (
  15. "bytes"
  16. "fmt"
  17. "net"
  18. "reflect"
  19. "sort"
  20. "strings"
  21. "testing"
  22. "time"
  23. "github.com/stretchr/testify/assert"
  24. v1 "k8s.io/api/core/v1"
  25. discovery "k8s.io/api/discovery/v1beta1"
  26. metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  27. "k8s.io/apimachinery/pkg/types"
  28. "k8s.io/apimachinery/pkg/util/intstr"
  29. "k8s.io/apimachinery/pkg/util/sets"
  30. "k8s.io/kubernetes/pkg/proxy"
  31. "k8s.io/kubernetes/pkg/proxy/healthcheck"
  32. netlinktest "k8s.io/kubernetes/pkg/proxy/ipvs/testing"
  33. utilproxy "k8s.io/kubernetes/pkg/proxy/util"
  34. proxyutiltest "k8s.io/kubernetes/pkg/proxy/util/testing"
  35. "k8s.io/kubernetes/pkg/util/async"
  36. utilipset "k8s.io/kubernetes/pkg/util/ipset"
  37. ipsettest "k8s.io/kubernetes/pkg/util/ipset/testing"
  38. utiliptables "k8s.io/kubernetes/pkg/util/iptables"
  39. iptablestest "k8s.io/kubernetes/pkg/util/iptables/testing"
  40. utilipvs "k8s.io/kubernetes/pkg/util/ipvs"
  41. ipvstest "k8s.io/kubernetes/pkg/util/ipvs/testing"
  42. "k8s.io/utils/exec"
  43. fakeexec "k8s.io/utils/exec/testing"
  44. utilpointer "k8s.io/utils/pointer"
  45. )
  46. const testHostname = "test-hostname"
  47. type fakeIPGetter struct {
  48. nodeIPs []net.IP
  49. }
  50. func (f *fakeIPGetter) NodeIPs() ([]net.IP, error) {
  51. return f.nodeIPs, nil
  52. }
  53. // fakePortOpener implements portOpener.
  54. type fakePortOpener struct {
  55. openPorts []*utilproxy.LocalPort
  56. }
  57. // OpenLocalPort fakes out the listen() and bind() used by syncProxyRules
  58. // to lock a local port.
  59. func (f *fakePortOpener) OpenLocalPort(lp *utilproxy.LocalPort) (utilproxy.Closeable, error) {
  60. f.openPorts = append(f.openPorts, lp)
  61. return nil, nil
  62. }
  63. // fakeKernelHandler implements KernelHandler.
  64. type fakeKernelHandler struct {
  65. modules []string
  66. kernelVersion string
  67. }
  68. func (fake *fakeKernelHandler) GetModules() ([]string, error) {
  69. return fake.modules, nil
  70. }
  71. func (fake *fakeKernelHandler) GetKernelVersion() (string, error) {
  72. return fake.kernelVersion, nil
  73. }
  74. // fakeKernelHandler implements KernelHandler.
  75. type fakeIPSetVersioner struct {
  76. version string
  77. err error
  78. }
  79. func (fake *fakeIPSetVersioner) GetVersion() (string, error) {
  80. return fake.version, fake.err
  81. }
  82. func NewFakeProxier(ipt utiliptables.Interface, ipvs utilipvs.Interface, ipset utilipset.Interface, nodeIPs []net.IP, excludeCIDRs []*net.IPNet, endpointSlicesEnabled bool) *Proxier {
  83. fcmd := fakeexec.FakeCmd{
  84. CombinedOutputScript: []fakeexec.FakeAction{
  85. func() ([]byte, []byte, error) { return []byte("dummy device have been created"), nil, nil },
  86. func() ([]byte, []byte, error) { return []byte(""), nil, nil },
  87. },
  88. }
  89. fexec := &fakeexec.FakeExec{
  90. CommandScript: []fakeexec.FakeCommandAction{
  91. func(cmd string, args ...string) exec.Cmd { return fakeexec.InitFakeCmd(&fcmd, cmd, args...) },
  92. func(cmd string, args ...string) exec.Cmd { return fakeexec.InitFakeCmd(&fcmd, cmd, args...) },
  93. },
  94. LookPathFunc: func(cmd string) (string, error) { return cmd, nil },
  95. }
  96. // initialize ipsetList with all sets we needed
  97. ipsetList := make(map[string]*IPSet)
  98. for _, is := range ipsetInfo {
  99. ipsetList[is.name] = NewIPSet(ipset, is.name, is.setType, false, is.comment)
  100. }
  101. p := &Proxier{
  102. exec: fexec,
  103. serviceMap: make(proxy.ServiceMap),
  104. serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, nil, nil),
  105. endpointsMap: make(proxy.EndpointsMap),
  106. endpointsChanges: proxy.NewEndpointChangeTracker(testHostname, nil, nil, nil, endpointSlicesEnabled),
  107. excludeCIDRs: excludeCIDRs,
  108. iptables: ipt,
  109. ipvs: ipvs,
  110. ipset: ipset,
  111. clusterCIDR: "10.0.0.0/24",
  112. strictARP: false,
  113. hostname: testHostname,
  114. portsMap: make(map[utilproxy.LocalPort]utilproxy.Closeable),
  115. portMapper: &fakePortOpener{[]*utilproxy.LocalPort{}},
  116. serviceHealthServer: healthcheck.NewFakeServiceHealthServer(),
  117. ipvsScheduler: DefaultScheduler,
  118. ipGetter: &fakeIPGetter{nodeIPs: nodeIPs},
  119. iptablesData: bytes.NewBuffer(nil),
  120. filterChainsData: bytes.NewBuffer(nil),
  121. natChains: bytes.NewBuffer(nil),
  122. natRules: bytes.NewBuffer(nil),
  123. filterChains: bytes.NewBuffer(nil),
  124. filterRules: bytes.NewBuffer(nil),
  125. netlinkHandle: netlinktest.NewFakeNetlinkHandle(),
  126. ipsetList: ipsetList,
  127. nodePortAddresses: make([]string, 0),
  128. networkInterfacer: proxyutiltest.NewFakeNetwork(),
  129. gracefuldeleteManager: NewGracefulTerminationManager(ipvs),
  130. }
  131. p.setInitialized(true)
  132. p.syncRunner = async.NewBoundedFrequencyRunner("test-sync-runner", p.syncProxyRules, 0, time.Minute, 1)
  133. return p
  134. }
  135. func makeNSN(namespace, name string) types.NamespacedName {
  136. return types.NamespacedName{Namespace: namespace, Name: name}
  137. }
  138. func makeServiceMap(proxier *Proxier, allServices ...*v1.Service) {
  139. for i := range allServices {
  140. proxier.OnServiceAdd(allServices[i])
  141. }
  142. proxier.mu.Lock()
  143. defer proxier.mu.Unlock()
  144. proxier.servicesSynced = true
  145. }
  146. func makeTestService(namespace, name string, svcFunc func(*v1.Service)) *v1.Service {
  147. svc := &v1.Service{
  148. ObjectMeta: metav1.ObjectMeta{
  149. Name: name,
  150. Namespace: namespace,
  151. Annotations: map[string]string{},
  152. },
  153. Spec: v1.ServiceSpec{},
  154. Status: v1.ServiceStatus{},
  155. }
  156. svcFunc(svc)
  157. return svc
  158. }
  159. func makeEndpointsMap(proxier *Proxier, allEndpoints ...*v1.Endpoints) {
  160. for i := range allEndpoints {
  161. proxier.OnEndpointsAdd(allEndpoints[i])
  162. }
  163. proxier.mu.Lock()
  164. defer proxier.mu.Unlock()
  165. proxier.endpointsSynced = true
  166. }
  167. func makeTestEndpoints(namespace, name string, eptFunc func(*v1.Endpoints)) *v1.Endpoints {
  168. ept := &v1.Endpoints{
  169. ObjectMeta: metav1.ObjectMeta{
  170. Name: name,
  171. Namespace: namespace,
  172. },
  173. }
  174. eptFunc(ept)
  175. return ept
  176. }
  177. func TestCleanupLeftovers(t *testing.T) {
  178. ipt := iptablestest.NewFake()
  179. ipvs := ipvstest.NewFake()
  180. ipset := ipsettest.NewFake(testIPSetVersion)
  181. fp := NewFakeProxier(ipt, ipvs, ipset, nil, nil, false)
  182. svcIP := "10.20.30.41"
  183. svcPort := 80
  184. svcNodePort := 3001
  185. svcPortName := proxy.ServicePortName{
  186. NamespacedName: makeNSN("ns1", "svc1"),
  187. Port: "p80",
  188. }
  189. makeServiceMap(fp,
  190. makeTestService(svcPortName.Namespace, svcPortName.Name, func(svc *v1.Service) {
  191. svc.Spec.Type = "NodePort"
  192. svc.Spec.ClusterIP = svcIP
  193. svc.Spec.Ports = []v1.ServicePort{{
  194. Name: svcPortName.Port,
  195. Port: int32(svcPort),
  196. Protocol: v1.ProtocolTCP,
  197. NodePort: int32(svcNodePort),
  198. }}
  199. }),
  200. )
  201. epIP := "10.180.0.1"
  202. makeEndpointsMap(fp,
  203. makeTestEndpoints(svcPortName.Namespace, svcPortName.Name, func(ept *v1.Endpoints) {
  204. ept.Subsets = []v1.EndpointSubset{{
  205. Addresses: []v1.EndpointAddress{{
  206. IP: epIP,
  207. }},
  208. Ports: []v1.EndpointPort{{
  209. Name: svcPortName.Port,
  210. Port: int32(svcPort),
  211. }},
  212. }}
  213. }),
  214. )
  215. fp.syncProxyRules()
  216. // test cleanup left over
  217. if CleanupLeftovers(ipvs, ipt, ipset, true) {
  218. t.Errorf("Cleanup leftovers failed")
  219. }
  220. }
  221. func TestCanUseIPVSProxier(t *testing.T) {
  222. testCases := []struct {
  223. mods []string
  224. kernelVersion string
  225. kernelErr error
  226. ipsetVersion string
  227. ipsetErr error
  228. ok bool
  229. }{
  230. // case 0, kernel error
  231. {
  232. mods: []string{"foo", "bar", "baz"},
  233. kernelVersion: "4.19",
  234. kernelErr: fmt.Errorf("oops"),
  235. ipsetVersion: "0.0",
  236. ok: false,
  237. },
  238. // case 1, ipset error
  239. {
  240. mods: []string{"foo", "bar", "baz"},
  241. kernelVersion: "4.19",
  242. ipsetVersion: MinIPSetCheckVersion,
  243. ipsetErr: fmt.Errorf("oops"),
  244. ok: false,
  245. },
  246. // case 2, missing required kernel modules and ipset version too low
  247. {
  248. mods: []string{"foo", "bar", "baz"},
  249. kernelVersion: "4.19",
  250. ipsetVersion: "1.1",
  251. ok: false,
  252. },
  253. // case 3, missing required ip_vs_* kernel modules
  254. {
  255. mods: []string{"ip_vs", "a", "bc", "def"},
  256. kernelVersion: "4.19",
  257. ipsetVersion: MinIPSetCheckVersion,
  258. ok: false,
  259. },
  260. // case 4, ipset version too low
  261. {
  262. mods: []string{"ip_vs", "ip_vs_rr", "ip_vs_wrr", "ip_vs_sh", "nf_conntrack"},
  263. kernelVersion: "4.19",
  264. ipsetVersion: "4.3.0",
  265. ok: false,
  266. },
  267. // case 5, ok for linux kernel 4.19
  268. {
  269. mods: []string{"ip_vs", "ip_vs_rr", "ip_vs_wrr", "ip_vs_sh", "nf_conntrack"},
  270. kernelVersion: "4.19",
  271. ipsetVersion: MinIPSetCheckVersion,
  272. ok: true,
  273. },
  274. // case 6, ok for linux kernel 4.18
  275. {
  276. mods: []string{"ip_vs", "ip_vs_rr", "ip_vs_wrr", "ip_vs_sh", "nf_conntrack_ipv4"},
  277. kernelVersion: "4.18",
  278. ipsetVersion: MinIPSetCheckVersion,
  279. ok: true,
  280. },
  281. // case 7. ok when module list has extra modules
  282. {
  283. mods: []string{"foo", "ip_vs", "ip_vs_rr", "ip_vs_wrr", "ip_vs_sh", "nf_conntrack", "bar"},
  284. kernelVersion: "4.19",
  285. ipsetVersion: "6.19",
  286. ok: true,
  287. },
  288. }
  289. for i := range testCases {
  290. handle := &fakeKernelHandler{modules: testCases[i].mods, kernelVersion: testCases[i].kernelVersion}
  291. versioner := &fakeIPSetVersioner{version: testCases[i].ipsetVersion, err: testCases[i].ipsetErr}
  292. ok, err := CanUseIPVSProxier(handle, versioner)
  293. if ok != testCases[i].ok {
  294. t.Errorf("Case [%d], expect %v, got %v: err: %v", i, testCases[i].ok, ok, err)
  295. }
  296. }
  297. }
  298. func TestGetNodeIPs(t *testing.T) {
  299. testCases := []struct {
  300. devAddresses map[string][]string
  301. expectIPs []string
  302. }{
  303. // case 0
  304. {
  305. devAddresses: map[string][]string{"eth0": {"1.2.3.4"}, "lo": {"127.0.0.1"}},
  306. expectIPs: []string{"1.2.3.4", "127.0.0.1"},
  307. },
  308. // case 1
  309. {
  310. devAddresses: map[string][]string{"lo": {"127.0.0.1"}},
  311. expectIPs: []string{"127.0.0.1"},
  312. },
  313. // case 2
  314. {
  315. devAddresses: map[string][]string{},
  316. expectIPs: []string{},
  317. },
  318. // case 3
  319. {
  320. devAddresses: map[string][]string{"encap0": {"10.20.30.40"}, "lo": {"127.0.0.1"}, "docker0": {"172.17.0.1"}},
  321. expectIPs: []string{"10.20.30.40", "127.0.0.1", "172.17.0.1"},
  322. },
  323. // case 4
  324. {
  325. devAddresses: map[string][]string{"encaps9": {"10.20.30.40"}, "lo": {"127.0.0.1"}, "encap7": {"10.20.30.31"}},
  326. expectIPs: []string{"10.20.30.40", "127.0.0.1", "10.20.30.31"},
  327. },
  328. // case 5
  329. {
  330. devAddresses: map[string][]string{"kube-ipvs0": {"1.2.3.4"}, "lo": {"127.0.0.1"}, "encap7": {"10.20.30.31"}},
  331. expectIPs: []string{"127.0.0.1", "10.20.30.31"},
  332. },
  333. // case 6
  334. {
  335. devAddresses: map[string][]string{"kube-ipvs0": {"1.2.3.4", "2.3.4.5"}, "lo": {"127.0.0.1"}},
  336. expectIPs: []string{"127.0.0.1"},
  337. },
  338. // case 7
  339. {
  340. devAddresses: map[string][]string{"kube-ipvs0": {"1.2.3.4", "2.3.4.5"}},
  341. expectIPs: []string{},
  342. },
  343. // case 8
  344. {
  345. devAddresses: map[string][]string{"kube-ipvs0": {"1.2.3.4", "2.3.4.5"}, "eth5": {"3.4.5.6"}, "lo": {"127.0.0.1"}},
  346. expectIPs: []string{"127.0.0.1", "3.4.5.6"},
  347. },
  348. // case 9
  349. {
  350. devAddresses: map[string][]string{"ipvs0": {"1.2.3.4"}, "lo": {"127.0.0.1"}, "encap7": {"10.20.30.31"}},
  351. expectIPs: []string{"127.0.0.1", "10.20.30.31", "1.2.3.4"},
  352. },
  353. }
  354. for i := range testCases {
  355. fake := netlinktest.NewFakeNetlinkHandle()
  356. for dev, addresses := range testCases[i].devAddresses {
  357. fake.SetLocalAddresses(dev, addresses...)
  358. }
  359. r := realIPGetter{nl: fake}
  360. ips, err := r.NodeIPs()
  361. if err != nil {
  362. t.Errorf("Unexpected error: %v", err)
  363. }
  364. ipStrs := sets.NewString()
  365. for _, ip := range ips {
  366. ipStrs.Insert(ip.String())
  367. }
  368. if !ipStrs.Equal(sets.NewString(testCases[i].expectIPs...)) {
  369. t.Errorf("case[%d], unexpected mismatch, expected: %v, got: %v", i, testCases[i].expectIPs, ips)
  370. }
  371. }
  372. }
  373. func TestNodePort(t *testing.T) {
  374. tests := []struct {
  375. name string
  376. services []*v1.Service
  377. endpoints []*v1.Endpoints
  378. nodeIPs []net.IP
  379. nodePortAddresses []string
  380. expectedIPVS *ipvstest.FakeIPVS
  381. expectedIPSets netlinktest.ExpectedIPSet
  382. expectedIptablesChains netlinktest.ExpectedIptablesChain
  383. }{
  384. {
  385. name: "1 service with node port, has 2 endpoints",
  386. services: []*v1.Service{
  387. makeTestService("ns1", "svc1", func(svc *v1.Service) {
  388. svc.Spec.Type = "NodePort"
  389. svc.Spec.ClusterIP = "10.20.30.41"
  390. svc.Spec.Ports = []v1.ServicePort{{
  391. Name: "p80",
  392. Port: int32(80),
  393. Protocol: v1.ProtocolTCP,
  394. NodePort: int32(3001),
  395. }}
  396. }),
  397. },
  398. endpoints: []*v1.Endpoints{
  399. makeTestEndpoints("ns1", "svc1", func(ept *v1.Endpoints) {
  400. ept.Subsets = []v1.EndpointSubset{{
  401. Addresses: []v1.EndpointAddress{{
  402. IP: "10.180.0.1",
  403. }, {
  404. IP: "1002:ab8::2:10",
  405. }},
  406. Ports: []v1.EndpointPort{{
  407. Name: "p80",
  408. Port: int32(80),
  409. Protocol: v1.ProtocolTCP,
  410. }},
  411. }}
  412. }),
  413. },
  414. nodeIPs: []net.IP{
  415. net.ParseIP("100.101.102.103"),
  416. net.ParseIP("2001:db8::1:1"),
  417. },
  418. nodePortAddresses: []string{},
  419. expectedIPVS: &ipvstest.FakeIPVS{
  420. Services: map[ipvstest.ServiceKey]*utilipvs.VirtualServer{
  421. {
  422. IP: "10.20.30.41",
  423. Port: 80,
  424. Protocol: "TCP",
  425. }: {
  426. Address: net.ParseIP("10.20.30.41"),
  427. Protocol: "TCP",
  428. Port: uint16(80),
  429. Scheduler: "rr",
  430. },
  431. {
  432. IP: "100.101.102.103",
  433. Port: 3001,
  434. Protocol: "TCP",
  435. }: {
  436. Address: net.ParseIP("100.101.102.103"),
  437. Protocol: "TCP",
  438. Port: uint16(3001),
  439. Scheduler: "rr",
  440. },
  441. {
  442. IP: "2001:db8::1:1",
  443. Port: 3001,
  444. Protocol: "TCP",
  445. }: {
  446. Address: net.ParseIP("2001:db8::1:1"),
  447. Protocol: "TCP",
  448. Port: uint16(3001),
  449. Scheduler: "rr",
  450. },
  451. },
  452. Destinations: map[ipvstest.ServiceKey][]*utilipvs.RealServer{
  453. {
  454. IP: "10.20.30.41",
  455. Port: 80,
  456. Protocol: "TCP",
  457. }: {
  458. {
  459. Address: net.ParseIP("10.180.0.1"),
  460. Port: uint16(80),
  461. Weight: 1,
  462. },
  463. {
  464. Address: net.ParseIP("1002:ab8::2:10"),
  465. Port: uint16(80),
  466. Weight: 1,
  467. },
  468. },
  469. {
  470. IP: "100.101.102.103",
  471. Port: 3001,
  472. Protocol: "TCP",
  473. }: {
  474. {
  475. Address: net.ParseIP("10.180.0.1"),
  476. Port: uint16(80),
  477. Weight: 1,
  478. },
  479. {
  480. Address: net.ParseIP("1002:ab8::2:10"),
  481. Port: uint16(80),
  482. Weight: 1,
  483. },
  484. },
  485. {
  486. IP: "2001:db8::1:1",
  487. Port: 3001,
  488. Protocol: "TCP",
  489. }: {
  490. {
  491. Address: net.ParseIP("10.180.0.1"),
  492. Port: uint16(80),
  493. Weight: 1,
  494. },
  495. {
  496. Address: net.ParseIP("1002:ab8::2:10"),
  497. Port: uint16(80),
  498. Weight: 1,
  499. },
  500. },
  501. },
  502. },
  503. },
  504. {
  505. name: "1 UDP service with node port, has endpoints",
  506. services: []*v1.Service{
  507. makeTestService("ns1", "svc1", func(svc *v1.Service) {
  508. svc.Spec.Type = "NodePort"
  509. svc.Spec.ClusterIP = "10.20.30.41"
  510. svc.Spec.Ports = []v1.ServicePort{{
  511. Name: "p80",
  512. Port: int32(80),
  513. Protocol: v1.ProtocolUDP,
  514. NodePort: int32(3001),
  515. }}
  516. }),
  517. },
  518. endpoints: []*v1.Endpoints{
  519. makeTestEndpoints("ns1", "svc1", func(ept *v1.Endpoints) {
  520. ept.Subsets = []v1.EndpointSubset{{
  521. Addresses: []v1.EndpointAddress{{
  522. IP: "10.180.0.1",
  523. }},
  524. Ports: []v1.EndpointPort{{
  525. Name: "p80",
  526. Port: int32(80),
  527. Protocol: v1.ProtocolUDP,
  528. }},
  529. }}
  530. }),
  531. },
  532. nodeIPs: []net.IP{
  533. net.ParseIP("100.101.102.103"),
  534. },
  535. nodePortAddresses: []string{"0.0.0.0/0"},
  536. expectedIPVS: &ipvstest.FakeIPVS{
  537. Services: map[ipvstest.ServiceKey]*utilipvs.VirtualServer{
  538. {
  539. IP: "10.20.30.41",
  540. Port: 80,
  541. Protocol: "UDP",
  542. }: {
  543. Address: net.ParseIP("10.20.30.41"),
  544. Protocol: "UDP",
  545. Port: uint16(80),
  546. Scheduler: "rr",
  547. },
  548. {
  549. IP: "100.101.102.103",
  550. Port: 3001,
  551. Protocol: "UDP",
  552. }: {
  553. Address: net.ParseIP("100.101.102.103"),
  554. Protocol: "UDP",
  555. Port: uint16(3001),
  556. Scheduler: "rr",
  557. },
  558. },
  559. Destinations: map[ipvstest.ServiceKey][]*utilipvs.RealServer{
  560. {
  561. IP: "10.20.30.41",
  562. Port: 80,
  563. Protocol: "UDP",
  564. }: {
  565. {
  566. Address: net.ParseIP("10.180.0.1"),
  567. Port: uint16(80),
  568. Weight: 1,
  569. },
  570. },
  571. {
  572. IP: "100.101.102.103",
  573. Port: 3001,
  574. Protocol: "UDP",
  575. }: {
  576. {
  577. Address: net.ParseIP("10.180.0.1"),
  578. Port: uint16(80),
  579. Weight: 1,
  580. },
  581. },
  582. },
  583. },
  584. expectedIPSets: netlinktest.ExpectedIPSet{
  585. kubeNodePortSetUDP: {{
  586. Port: 3001,
  587. Protocol: strings.ToLower(string(v1.ProtocolUDP)),
  588. SetType: utilipset.BitmapPort,
  589. }},
  590. },
  591. expectedIptablesChains: netlinktest.ExpectedIptablesChain{
  592. string(KubeNodePortChain): {{
  593. JumpChain: string(KubeMarkMasqChain), MatchSet: kubeNodePortSetUDP,
  594. }},
  595. string(kubeServicesChain): {{
  596. JumpChain: string(KubeNodePortChain), MatchSet: "",
  597. }},
  598. },
  599. },
  600. {
  601. name: "service has node port but no endpoints",
  602. services: []*v1.Service{
  603. makeTestService("ns1", "svc1", func(svc *v1.Service) {
  604. svc.Spec.Type = "NodePort"
  605. svc.Spec.ClusterIP = "10.20.30.41"
  606. svc.Spec.Ports = []v1.ServicePort{{
  607. Name: "p80",
  608. Port: int32(80),
  609. Protocol: v1.ProtocolTCP,
  610. NodePort: int32(3001),
  611. }}
  612. }),
  613. },
  614. endpoints: []*v1.Endpoints{},
  615. nodeIPs: []net.IP{
  616. net.ParseIP("100.101.102.103"),
  617. },
  618. nodePortAddresses: []string{},
  619. expectedIPVS: &ipvstest.FakeIPVS{
  620. Services: map[ipvstest.ServiceKey]*utilipvs.VirtualServer{
  621. {
  622. IP: "10.20.30.41",
  623. Port: 80,
  624. Protocol: "TCP",
  625. }: {
  626. Address: net.ParseIP("10.20.30.41"),
  627. Protocol: "TCP",
  628. Port: uint16(80),
  629. Scheduler: "rr",
  630. },
  631. {
  632. IP: "100.101.102.103",
  633. Port: 3001,
  634. Protocol: "TCP",
  635. }: {
  636. Address: net.ParseIP("100.101.102.103"),
  637. Protocol: "TCP",
  638. Port: uint16(3001),
  639. Scheduler: "rr",
  640. },
  641. },
  642. Destinations: map[ipvstest.ServiceKey][]*utilipvs.RealServer{
  643. {
  644. IP: "10.20.30.41",
  645. Port: 80,
  646. Protocol: "TCP",
  647. }: {}, // no real servers corresponding to no endpoints
  648. {
  649. IP: "100.101.102.103",
  650. Port: 3001,
  651. Protocol: "TCP",
  652. }: {}, // no real servers corresponding to no endpoints
  653. },
  654. },
  655. },
  656. {
  657. name: "node port service with protocol sctp on a node with multiple nodeIPs",
  658. services: []*v1.Service{
  659. makeTestService("ns1", "svc1", func(svc *v1.Service) {
  660. svc.Spec.Type = "NodePort"
  661. svc.Spec.ClusterIP = "10.20.30.41"
  662. svc.Spec.Ports = []v1.ServicePort{{
  663. Name: "p80",
  664. Port: int32(80),
  665. Protocol: v1.ProtocolSCTP,
  666. NodePort: int32(3001),
  667. }}
  668. }),
  669. },
  670. endpoints: []*v1.Endpoints{
  671. makeTestEndpoints("ns1", "svc1", func(ept *v1.Endpoints) {
  672. ept.Subsets = []v1.EndpointSubset{{
  673. Addresses: []v1.EndpointAddress{{
  674. IP: "10.180.0.1",
  675. }},
  676. Ports: []v1.EndpointPort{{
  677. Name: "p80",
  678. Port: int32(80),
  679. Protocol: v1.ProtocolSCTP,
  680. }},
  681. }}
  682. }),
  683. },
  684. nodeIPs: []net.IP{
  685. net.ParseIP("100.101.102.103"),
  686. net.ParseIP("100.101.102.104"),
  687. net.ParseIP("100.101.102.105"),
  688. net.ParseIP("2001:db8::1:1"),
  689. net.ParseIP("2001:db8::1:2"),
  690. net.ParseIP("2001:db8::1:3"),
  691. },
  692. nodePortAddresses: []string{},
  693. expectedIPVS: &ipvstest.FakeIPVS{
  694. Services: map[ipvstest.ServiceKey]*utilipvs.VirtualServer{
  695. {
  696. IP: "10.20.30.41",
  697. Port: 80,
  698. Protocol: "SCTP",
  699. }: {
  700. Address: net.ParseIP("10.20.30.41"),
  701. Protocol: "SCTP",
  702. Port: uint16(80),
  703. Scheduler: "rr",
  704. },
  705. {
  706. IP: "100.101.102.103",
  707. Port: 3001,
  708. Protocol: "SCTP",
  709. }: {
  710. Address: net.ParseIP("100.101.102.103"),
  711. Protocol: "SCTP",
  712. Port: uint16(3001),
  713. Scheduler: "rr",
  714. },
  715. {
  716. IP: "100.101.102.104",
  717. Port: 3001,
  718. Protocol: "SCTP",
  719. }: {
  720. Address: net.ParseIP("100.101.102.104"),
  721. Protocol: "SCTP",
  722. Port: uint16(3001),
  723. Scheduler: "rr",
  724. },
  725. {
  726. IP: "100.101.102.105",
  727. Port: 3001,
  728. Protocol: "SCTP",
  729. }: {
  730. Address: net.ParseIP("100.101.102.105"),
  731. Protocol: "SCTP",
  732. Port: uint16(3001),
  733. Scheduler: "rr",
  734. },
  735. {
  736. IP: "2001:db8::1:1",
  737. Port: 3001,
  738. Protocol: "SCTP",
  739. }: {
  740. Address: net.ParseIP("2001:db8::1:1"),
  741. Protocol: "SCTP",
  742. Port: uint16(3001),
  743. Scheduler: "rr",
  744. },
  745. {
  746. IP: "2001:db8::1:2",
  747. Port: 3001,
  748. Protocol: "SCTP",
  749. }: {
  750. Address: net.ParseIP("2001:db8::1:2"),
  751. Protocol: "SCTP",
  752. Port: uint16(3001),
  753. Scheduler: "rr",
  754. },
  755. {
  756. IP: "2001:db8::1:3",
  757. Port: 3001,
  758. Protocol: "SCTP",
  759. }: {
  760. Address: net.ParseIP("2001:db8::1:3"),
  761. Protocol: "SCTP",
  762. Port: uint16(3001),
  763. Scheduler: "rr",
  764. },
  765. },
  766. Destinations: map[ipvstest.ServiceKey][]*utilipvs.RealServer{
  767. {
  768. IP: "10.20.30.41",
  769. Port: 80,
  770. Protocol: "SCTP",
  771. }: {
  772. {
  773. Address: net.ParseIP("10.180.0.1"),
  774. Port: uint16(80),
  775. Weight: 1,
  776. },
  777. },
  778. {
  779. IP: "100.101.102.103",
  780. Port: 3001,
  781. Protocol: "SCTP",
  782. }: {
  783. {
  784. Address: net.ParseIP("10.180.0.1"),
  785. Port: uint16(80),
  786. Weight: 1,
  787. },
  788. },
  789. {
  790. IP: "100.101.102.104",
  791. Port: 3001,
  792. Protocol: "SCTP",
  793. }: {
  794. {
  795. Address: net.ParseIP("10.180.0.1"),
  796. Port: uint16(80),
  797. Weight: 1,
  798. },
  799. },
  800. {
  801. IP: "100.101.102.105",
  802. Port: 3001,
  803. Protocol: "SCTP",
  804. }: {
  805. {
  806. Address: net.ParseIP("10.180.0.1"),
  807. Port: uint16(80),
  808. Weight: 1,
  809. },
  810. },
  811. {
  812. IP: "2001:db8::1:1",
  813. Port: 3001,
  814. Protocol: "SCTP",
  815. }: {
  816. {
  817. Address: net.ParseIP("10.180.0.1"),
  818. Port: uint16(80),
  819. Weight: 1,
  820. },
  821. },
  822. {
  823. IP: "2001:db8::1:2",
  824. Port: 3001,
  825. Protocol: "SCTP",
  826. }: {
  827. {
  828. Address: net.ParseIP("10.180.0.1"),
  829. Port: uint16(80),
  830. Weight: 1,
  831. },
  832. },
  833. {
  834. IP: "2001:db8::1:3",
  835. Port: 3001,
  836. Protocol: "SCTP",
  837. }: {
  838. {
  839. Address: net.ParseIP("10.180.0.1"),
  840. Port: uint16(80),
  841. Weight: 1,
  842. },
  843. },
  844. },
  845. },
  846. expectedIPSets: netlinktest.ExpectedIPSet{
  847. kubeNodePortSetSCTP: {
  848. {
  849. IP: "100.101.102.103",
  850. Port: 3001,
  851. Protocol: strings.ToLower(string(v1.ProtocolSCTP)),
  852. SetType: utilipset.HashIPPort,
  853. },
  854. {
  855. IP: "100.101.102.104",
  856. Port: 3001,
  857. Protocol: strings.ToLower(string(v1.ProtocolSCTP)),
  858. SetType: utilipset.HashIPPort,
  859. },
  860. {
  861. IP: "100.101.102.105",
  862. Port: 3001,
  863. Protocol: strings.ToLower(string(v1.ProtocolSCTP)),
  864. SetType: utilipset.HashIPPort,
  865. },
  866. {
  867. IP: "2001:db8::1:1",
  868. Port: 3001,
  869. Protocol: strings.ToLower(string(v1.ProtocolSCTP)),
  870. SetType: utilipset.HashIPPort,
  871. },
  872. {
  873. IP: "2001:db8::1:2",
  874. Port: 3001,
  875. Protocol: strings.ToLower(string(v1.ProtocolSCTP)),
  876. SetType: utilipset.HashIPPort,
  877. },
  878. {
  879. IP: "2001:db8::1:3",
  880. Port: 3001,
  881. Protocol: strings.ToLower(string(v1.ProtocolSCTP)),
  882. SetType: utilipset.HashIPPort,
  883. },
  884. },
  885. },
  886. },
  887. }
  888. for _, test := range tests {
  889. t.Run(test.name, func(t *testing.T) {
  890. ipt := iptablestest.NewFake()
  891. ipvs := ipvstest.NewFake()
  892. ipset := ipsettest.NewFake(testIPSetVersion)
  893. fp := NewFakeProxier(ipt, ipvs, ipset, test.nodeIPs, nil, false)
  894. fp.nodePortAddresses = test.nodePortAddresses
  895. makeServiceMap(fp, test.services...)
  896. makeEndpointsMap(fp, test.endpoints...)
  897. fp.syncProxyRules()
  898. if !reflect.DeepEqual(ipvs, test.expectedIPVS) {
  899. t.Logf("actual ipvs state: %v", ipvs)
  900. t.Logf("expected ipvs state: %v", test.expectedIPVS)
  901. t.Errorf("unexpected IPVS state")
  902. }
  903. if test.expectedIPSets != nil {
  904. checkIPSet(t, fp, test.expectedIPSets)
  905. }
  906. if test.expectedIptablesChains != nil {
  907. checkIptables(t, ipt, test.expectedIptablesChains)
  908. }
  909. })
  910. }
  911. }
  912. func TestClusterIP(t *testing.T) {
  913. tests := []struct {
  914. name string
  915. services []*v1.Service
  916. endpoints []*v1.Endpoints
  917. expectedIPVS *ipvstest.FakeIPVS
  918. }{
  919. {
  920. name: "2 services with Cluster IP, each with endpoints",
  921. services: []*v1.Service{
  922. makeTestService("ns1", "svc1", func(svc *v1.Service) {
  923. svc.Spec.ClusterIP = "10.20.30.41"
  924. svc.Spec.Ports = []v1.ServicePort{{
  925. Name: "p80",
  926. Port: int32(80),
  927. Protocol: v1.ProtocolTCP,
  928. }}
  929. }),
  930. makeTestService("ns2", "svc2", func(svc *v1.Service) {
  931. svc.Spec.ClusterIP = "1002:ab8::2:1"
  932. svc.Spec.Ports = []v1.ServicePort{{
  933. Name: "p8080",
  934. Port: int32(8080),
  935. Protocol: v1.ProtocolTCP,
  936. }}
  937. }),
  938. },
  939. endpoints: []*v1.Endpoints{
  940. makeTestEndpoints("ns1", "svc1", func(ept *v1.Endpoints) {
  941. ept.Subsets = []v1.EndpointSubset{{
  942. Addresses: []v1.EndpointAddress{{
  943. IP: "10.180.0.1",
  944. }},
  945. Ports: []v1.EndpointPort{{
  946. Name: "p80",
  947. Port: int32(80),
  948. Protocol: v1.ProtocolTCP,
  949. }},
  950. }}
  951. }),
  952. makeTestEndpoints("ns2", "svc2", func(ept *v1.Endpoints) {
  953. ept.Subsets = []v1.EndpointSubset{{
  954. Addresses: []v1.EndpointAddress{{
  955. IP: "1009:ab8::5:6",
  956. }},
  957. Ports: []v1.EndpointPort{{
  958. Name: "p8080",
  959. Port: int32(8080),
  960. Protocol: v1.ProtocolTCP,
  961. }},
  962. }}
  963. }),
  964. },
  965. expectedIPVS: &ipvstest.FakeIPVS{
  966. Services: map[ipvstest.ServiceKey]*utilipvs.VirtualServer{
  967. {
  968. IP: "10.20.30.41",
  969. Port: 80,
  970. Protocol: "TCP",
  971. }: {
  972. Address: net.ParseIP("10.20.30.41"),
  973. Protocol: "TCP",
  974. Port: uint16(80),
  975. Scheduler: "rr",
  976. },
  977. {
  978. IP: "1002:ab8::2:1",
  979. Port: 8080,
  980. Protocol: "TCP",
  981. }: {
  982. Address: net.ParseIP("1002:ab8::2:1"),
  983. Protocol: "TCP",
  984. Port: uint16(8080),
  985. Scheduler: "rr",
  986. },
  987. },
  988. Destinations: map[ipvstest.ServiceKey][]*utilipvs.RealServer{
  989. {
  990. IP: "10.20.30.41",
  991. Port: 80,
  992. Protocol: "TCP",
  993. }: {
  994. {
  995. Address: net.ParseIP("10.180.0.1"),
  996. Port: uint16(80),
  997. Weight: 1,
  998. },
  999. },
  1000. {
  1001. IP: "1002:ab8::2:1",
  1002. Port: 8080,
  1003. Protocol: "TCP",
  1004. }: {
  1005. {
  1006. Address: net.ParseIP("1009:ab8::5:6"),
  1007. Port: uint16(8080),
  1008. Weight: 1,
  1009. },
  1010. },
  1011. },
  1012. },
  1013. },
  1014. {
  1015. name: "cluster IP service with no endpoints",
  1016. services: []*v1.Service{
  1017. makeTestService("ns1", "svc1", func(svc *v1.Service) {
  1018. svc.Spec.ClusterIP = "10.20.30.41"
  1019. svc.Spec.Ports = []v1.ServicePort{{
  1020. Name: "p80",
  1021. Port: int32(80),
  1022. Protocol: v1.ProtocolTCP,
  1023. }}
  1024. }),
  1025. },
  1026. endpoints: []*v1.Endpoints{},
  1027. expectedIPVS: &ipvstest.FakeIPVS{
  1028. Services: map[ipvstest.ServiceKey]*utilipvs.VirtualServer{
  1029. {
  1030. IP: "10.20.30.41",
  1031. Port: 80,
  1032. Protocol: "TCP",
  1033. }: {
  1034. Address: net.ParseIP("10.20.30.41"),
  1035. Protocol: "TCP",
  1036. Port: uint16(80),
  1037. Scheduler: "rr",
  1038. },
  1039. },
  1040. Destinations: map[ipvstest.ServiceKey][]*utilipvs.RealServer{
  1041. {
  1042. IP: "10.20.30.41",
  1043. Port: 80,
  1044. Protocol: "TCP",
  1045. }: {},
  1046. },
  1047. },
  1048. },
  1049. }
  1050. for _, test := range tests {
  1051. t.Run(test.name, func(t *testing.T) {
  1052. ipt := iptablestest.NewFake()
  1053. ipvs := ipvstest.NewFake()
  1054. ipset := ipsettest.NewFake(testIPSetVersion)
  1055. fp := NewFakeProxier(ipt, ipvs, ipset, nil, nil, false)
  1056. makeServiceMap(fp, test.services...)
  1057. makeEndpointsMap(fp, test.endpoints...)
  1058. fp.syncProxyRules()
  1059. if !reflect.DeepEqual(ipvs, test.expectedIPVS) {
  1060. t.Logf("actual ipvs state: %v", ipvs)
  1061. t.Logf("expected ipvs state: %v", test.expectedIPVS)
  1062. t.Errorf("unexpected IPVS state")
  1063. }
  1064. })
  1065. }
  1066. }
  1067. func TestMasqueradeRule(t *testing.T) {
  1068. for _, testcase := range []bool{false, true} {
  1069. ipt := iptablestest.NewFake().SetHasRandomFully(testcase)
  1070. ipvs := ipvstest.NewFake()
  1071. ipset := ipsettest.NewFake(testIPSetVersion)
  1072. fp := NewFakeProxier(ipt, ipvs, ipset, nil, nil, false)
  1073. makeServiceMap(fp)
  1074. makeEndpointsMap(fp)
  1075. fp.syncProxyRules()
  1076. postRoutingRules := ipt.GetRules(string(kubePostroutingChain))
  1077. if !hasJump(postRoutingRules, "MASQUERADE", "") {
  1078. t.Errorf("Failed to find -j MASQUERADE in %s chain", kubePostroutingChain)
  1079. }
  1080. if hasMasqRandomFully(postRoutingRules) != testcase {
  1081. probs := map[bool]string{false: "found", true: "did not find"}
  1082. t.Errorf("%s --random-fully in -j MASQUERADE rule in %s chain for HasRandomFully()=%v", probs[testcase], kubePostroutingChain, testcase)
  1083. }
  1084. }
  1085. }
  1086. func TestExternalIPsNoEndpoint(t *testing.T) {
  1087. ipt := iptablestest.NewFake()
  1088. ipvs := ipvstest.NewFake()
  1089. ipset := ipsettest.NewFake(testIPSetVersion)
  1090. fp := NewFakeProxier(ipt, ipvs, ipset, nil, nil, false)
  1091. svcIP := "10.20.30.41"
  1092. svcPort := 80
  1093. svcExternalIPs := "50.60.70.81"
  1094. svcPortName := proxy.ServicePortName{
  1095. NamespacedName: makeNSN("ns1", "svc1"),
  1096. Port: "p80",
  1097. }
  1098. makeServiceMap(fp,
  1099. makeTestService(svcPortName.Namespace, svcPortName.Name, func(svc *v1.Service) {
  1100. svc.Spec.Type = "ClusterIP"
  1101. svc.Spec.ClusterIP = svcIP
  1102. svc.Spec.ExternalIPs = []string{svcExternalIPs}
  1103. svc.Spec.Ports = []v1.ServicePort{{
  1104. Name: svcPortName.Port,
  1105. Port: int32(svcPort),
  1106. Protocol: v1.ProtocolTCP,
  1107. TargetPort: intstr.FromInt(svcPort),
  1108. }}
  1109. }),
  1110. )
  1111. makeEndpointsMap(fp)
  1112. fp.syncProxyRules()
  1113. // check ipvs service and destinations
  1114. services, err := ipvs.GetVirtualServers()
  1115. if err != nil {
  1116. t.Errorf("Failed to get ipvs services, err: %v", err)
  1117. }
  1118. if len(services) != 2 {
  1119. t.Errorf("Expect 2 ipvs services, got %d", len(services))
  1120. }
  1121. found := false
  1122. for _, svc := range services {
  1123. if svc.Address.String() == svcExternalIPs && svc.Port == uint16(svcPort) && svc.Protocol == string(v1.ProtocolTCP) {
  1124. found = true
  1125. destinations, _ := ipvs.GetRealServers(svc)
  1126. if len(destinations) != 0 {
  1127. t.Errorf("Unexpected %d destinations, expect 0 destinations", len(destinations))
  1128. }
  1129. break
  1130. }
  1131. }
  1132. if !found {
  1133. t.Errorf("Expect external ip type service, got none")
  1134. }
  1135. }
  1136. func TestExternalIPs(t *testing.T) {
  1137. ipt := iptablestest.NewFake()
  1138. ipvs := ipvstest.NewFake()
  1139. ipset := ipsettest.NewFake(testIPSetVersion)
  1140. fp := NewFakeProxier(ipt, ipvs, ipset, nil, nil, false)
  1141. svcIP := "10.20.30.41"
  1142. svcPort := 80
  1143. svcExternalIPs := sets.NewString("50.60.70.81", "2012::51", "127.0.0.1")
  1144. svcPortName := proxy.ServicePortName{
  1145. NamespacedName: makeNSN("ns1", "svc1"),
  1146. Port: "p80",
  1147. }
  1148. makeServiceMap(fp,
  1149. makeTestService(svcPortName.Namespace, svcPortName.Name, func(svc *v1.Service) {
  1150. svc.Spec.Type = "ClusterIP"
  1151. svc.Spec.ClusterIP = svcIP
  1152. svc.Spec.ExternalIPs = svcExternalIPs.UnsortedList()
  1153. svc.Spec.Ports = []v1.ServicePort{{
  1154. Name: svcPortName.Port,
  1155. Port: int32(svcPort),
  1156. Protocol: v1.ProtocolTCP,
  1157. TargetPort: intstr.FromInt(svcPort),
  1158. }}
  1159. }),
  1160. )
  1161. epIP := "10.180.0.1"
  1162. makeEndpointsMap(fp,
  1163. makeTestEndpoints(svcPortName.Namespace, svcPortName.Name, func(ept *v1.Endpoints) {
  1164. ept.Subsets = []v1.EndpointSubset{{
  1165. Addresses: []v1.EndpointAddress{{
  1166. IP: epIP,
  1167. }},
  1168. Ports: []v1.EndpointPort{{
  1169. Name: svcPortName.Port,
  1170. Port: int32(svcPort),
  1171. }},
  1172. }}
  1173. }),
  1174. )
  1175. fp.syncProxyRules()
  1176. // check ipvs service and destinations
  1177. services, err := ipvs.GetVirtualServers()
  1178. if err != nil {
  1179. t.Errorf("Failed to get ipvs services, err: %v", err)
  1180. }
  1181. if len(services) != 4 {
  1182. t.Errorf("Expect 4 ipvs services, got %d", len(services))
  1183. }
  1184. found := false
  1185. for _, svc := range services {
  1186. if svcExternalIPs.Has(svc.Address.String()) && svc.Port == uint16(svcPort) && svc.Protocol == string(v1.ProtocolTCP) {
  1187. found = true
  1188. destinations, _ := ipvs.GetRealServers(svc)
  1189. for _, dest := range destinations {
  1190. if dest.Address.String() != epIP || dest.Port != uint16(svcPort) {
  1191. t.Errorf("service Endpoint mismatch ipvs service destination")
  1192. }
  1193. }
  1194. break
  1195. }
  1196. }
  1197. if !found {
  1198. t.Errorf("Expect external ip type service, got none")
  1199. }
  1200. }
  1201. func TestLoadBalancer(t *testing.T) {
  1202. ipt, fp := buildFakeProxier()
  1203. svcIP := "10.20.30.41"
  1204. svcPort := 80
  1205. svcNodePort := 3001
  1206. svcLBIP := "1.2.3.4"
  1207. svcPortName := proxy.ServicePortName{
  1208. NamespacedName: makeNSN("ns1", "svc1"),
  1209. Port: "p80",
  1210. }
  1211. makeServiceMap(fp,
  1212. makeTestService(svcPortName.Namespace, svcPortName.Name, func(svc *v1.Service) {
  1213. svc.Spec.Type = "LoadBalancer"
  1214. svc.Spec.ClusterIP = svcIP
  1215. svc.Spec.Ports = []v1.ServicePort{{
  1216. Name: svcPortName.Port,
  1217. Port: int32(svcPort),
  1218. Protocol: v1.ProtocolTCP,
  1219. NodePort: int32(svcNodePort),
  1220. }}
  1221. svc.Status.LoadBalancer.Ingress = []v1.LoadBalancerIngress{{
  1222. IP: svcLBIP,
  1223. }}
  1224. }),
  1225. )
  1226. epIP := "10.180.0.1"
  1227. makeEndpointsMap(fp,
  1228. makeTestEndpoints(svcPortName.Namespace, svcPortName.Name, func(ept *v1.Endpoints) {
  1229. ept.Subsets = []v1.EndpointSubset{{
  1230. Addresses: []v1.EndpointAddress{{
  1231. IP: epIP,
  1232. }},
  1233. Ports: []v1.EndpointPort{{
  1234. Name: svcPortName.Port,
  1235. Port: int32(svcPort),
  1236. }},
  1237. }}
  1238. }),
  1239. )
  1240. fp.syncProxyRules()
  1241. // Expect 2 services and 1 destination
  1242. epVS := &netlinktest.ExpectedVirtualServer{
  1243. VSNum: 2, IP: svcLBIP, Port: uint16(svcNodePort), Protocol: string(v1.ProtocolTCP),
  1244. RS: []netlinktest.ExpectedRealServer{{
  1245. IP: epIP, Port: uint16(svcPort),
  1246. }}}
  1247. checkIPVS(t, fp, epVS)
  1248. // check ipSet rules
  1249. epIPSet := netlinktest.ExpectedIPSet{
  1250. kubeLoadBalancerSet: {{
  1251. IP: svcLBIP,
  1252. Port: svcPort,
  1253. Protocol: strings.ToLower(string(v1.ProtocolTCP)),
  1254. SetType: utilipset.HashIPPort,
  1255. }},
  1256. }
  1257. checkIPSet(t, fp, epIPSet)
  1258. // Check iptables chain and rules
  1259. epIpt := netlinktest.ExpectedIptablesChain{
  1260. string(kubeServicesChain): {{
  1261. JumpChain: string(KubeLoadBalancerChain), MatchSet: kubeLoadBalancerSet,
  1262. }},
  1263. string(kubeLoadBalancerSet): {{
  1264. JumpChain: string(KubeMarkMasqChain), MatchSet: "",
  1265. }},
  1266. }
  1267. checkIptables(t, ipt, epIpt)
  1268. }
  1269. func TestOnlyLocalNodePorts(t *testing.T) {
  1270. nodeIP := net.ParseIP("100.101.102.103")
  1271. ipt, fp := buildFakeProxier()
  1272. svcIP := "10.20.30.41"
  1273. svcPort := 80
  1274. svcNodePort := 3001
  1275. svcPortName := proxy.ServicePortName{
  1276. NamespacedName: makeNSN("ns1", "svc1"),
  1277. Port: "p80",
  1278. }
  1279. makeServiceMap(fp,
  1280. makeTestService(svcPortName.Namespace, svcPortName.Name, func(svc *v1.Service) {
  1281. svc.Spec.Type = "NodePort"
  1282. svc.Spec.ClusterIP = svcIP
  1283. svc.Spec.Ports = []v1.ServicePort{{
  1284. Name: svcPortName.Port,
  1285. Port: int32(svcPort),
  1286. Protocol: v1.ProtocolTCP,
  1287. NodePort: int32(svcNodePort),
  1288. }}
  1289. svc.Spec.ExternalTrafficPolicy = v1.ServiceExternalTrafficPolicyTypeLocal
  1290. }),
  1291. )
  1292. epIP := "10.180.0.1"
  1293. epIP1 := "10.180.1.1"
  1294. thisHostname := testHostname
  1295. otherHostname := "other-hostname"
  1296. makeEndpointsMap(fp,
  1297. makeTestEndpoints(svcPortName.Namespace, svcPortName.Name, func(ept *v1.Endpoints) {
  1298. ept.Subsets = []v1.EndpointSubset{
  1299. { // **local** endpoint address, should be added as RS
  1300. Addresses: []v1.EndpointAddress{{
  1301. IP: epIP,
  1302. NodeName: &thisHostname,
  1303. }},
  1304. Ports: []v1.EndpointPort{{
  1305. Name: svcPortName.Port,
  1306. Port: int32(svcPort),
  1307. Protocol: v1.ProtocolTCP,
  1308. }}},
  1309. { // **remote** endpoint address, should not be added as RS
  1310. Addresses: []v1.EndpointAddress{{
  1311. IP: epIP1,
  1312. NodeName: &otherHostname,
  1313. }},
  1314. Ports: []v1.EndpointPort{{
  1315. Name: svcPortName.Port,
  1316. Port: int32(svcPort),
  1317. Protocol: v1.ProtocolTCP,
  1318. }},
  1319. }}
  1320. }),
  1321. )
  1322. itf := net.Interface{Index: 0, MTU: 0, Name: "eth0", HardwareAddr: nil, Flags: 0}
  1323. addrs := []net.Addr{proxyutiltest.AddrStruct{Val: "100.101.102.103/24"}}
  1324. itf1 := net.Interface{Index: 1, MTU: 0, Name: "eth1", HardwareAddr: nil, Flags: 0}
  1325. addrs1 := []net.Addr{proxyutiltest.AddrStruct{Val: "2001:db8::0/64"}}
  1326. fp.networkInterfacer.(*proxyutiltest.FakeNetwork).AddInterfaceAddr(&itf, addrs)
  1327. fp.networkInterfacer.(*proxyutiltest.FakeNetwork).AddInterfaceAddr(&itf1, addrs1)
  1328. fp.nodePortAddresses = []string{"100.101.102.0/24", "2001:db8::0/64"}
  1329. fp.syncProxyRules()
  1330. // Expect 3 services and 1 destination
  1331. epVS := &netlinktest.ExpectedVirtualServer{
  1332. VSNum: 3, IP: nodeIP.String(), Port: uint16(svcNodePort), Protocol: string(v1.ProtocolTCP),
  1333. RS: []netlinktest.ExpectedRealServer{{
  1334. IP: epIP, Port: uint16(svcPort),
  1335. }}}
  1336. checkIPVS(t, fp, epVS)
  1337. // check ipSet rules
  1338. epEntry := &utilipset.Entry{
  1339. Port: svcNodePort,
  1340. Protocol: strings.ToLower(string(v1.ProtocolTCP)),
  1341. SetType: utilipset.BitmapPort,
  1342. }
  1343. epIPSet := netlinktest.ExpectedIPSet{
  1344. kubeNodePortSetTCP: {epEntry},
  1345. kubeNodePortLocalSetTCP: {epEntry},
  1346. }
  1347. checkIPSet(t, fp, epIPSet)
  1348. // Check iptables chain and rules
  1349. epIpt := netlinktest.ExpectedIptablesChain{
  1350. string(kubeServicesChain): {{
  1351. JumpChain: string(KubeNodePortChain), MatchSet: "",
  1352. }},
  1353. string(KubeNodePortChain): {{
  1354. JumpChain: "RETURN", MatchSet: kubeNodePortLocalSetTCP,
  1355. }, {
  1356. JumpChain: string(KubeMarkMasqChain), MatchSet: kubeNodePortSetTCP,
  1357. }},
  1358. }
  1359. checkIptables(t, ipt, epIpt)
  1360. }
  1361. func TestLoadBalanceSourceRanges(t *testing.T) {
  1362. ipt, fp := buildFakeProxier()
  1363. svcIP := "10.20.30.41"
  1364. svcPort := 80
  1365. svcLBIP := "1.2.3.4"
  1366. svcLBSource := "10.0.0.0/8"
  1367. svcPortName := proxy.ServicePortName{
  1368. NamespacedName: makeNSN("ns1", "svc1"),
  1369. Port: "p80",
  1370. }
  1371. epIP := "10.180.0.1"
  1372. makeServiceMap(fp,
  1373. makeTestService(svcPortName.Namespace, svcPortName.Name, func(svc *v1.Service) {
  1374. svc.Spec.Type = "LoadBalancer"
  1375. svc.Spec.ClusterIP = svcIP
  1376. svc.Spec.Ports = []v1.ServicePort{{
  1377. Name: svcPortName.Port,
  1378. Port: int32(svcPort),
  1379. Protocol: v1.ProtocolTCP,
  1380. }}
  1381. svc.Status.LoadBalancer.Ingress = []v1.LoadBalancerIngress{{
  1382. IP: svcLBIP,
  1383. }}
  1384. svc.Spec.LoadBalancerSourceRanges = []string{
  1385. svcLBSource,
  1386. }
  1387. }),
  1388. )
  1389. makeEndpointsMap(fp,
  1390. makeTestEndpoints(svcPortName.Namespace, svcPortName.Name, func(ept *v1.Endpoints) {
  1391. ept.Subsets = []v1.EndpointSubset{{
  1392. Addresses: []v1.EndpointAddress{{
  1393. IP: epIP,
  1394. NodeName: nil,
  1395. }},
  1396. Ports: []v1.EndpointPort{{
  1397. Name: svcPortName.Port,
  1398. Port: int32(svcPort),
  1399. Protocol: v1.ProtocolTCP,
  1400. }},
  1401. }}
  1402. }),
  1403. )
  1404. fp.syncProxyRules()
  1405. // Check ipvs service and destinations
  1406. epVS := &netlinktest.ExpectedVirtualServer{
  1407. VSNum: 2, IP: svcLBIP, Port: uint16(svcPort), Protocol: string(v1.ProtocolTCP),
  1408. RS: []netlinktest.ExpectedRealServer{{
  1409. IP: epIP, Port: uint16(svcPort),
  1410. }}}
  1411. checkIPVS(t, fp, epVS)
  1412. // Check ipset entry
  1413. epIPSet := netlinktest.ExpectedIPSet{
  1414. kubeLoadBalancerSet: {{
  1415. IP: svcLBIP,
  1416. Port: svcPort,
  1417. Protocol: strings.ToLower(string(v1.ProtocolTCP)),
  1418. SetType: utilipset.HashIPPort,
  1419. }},
  1420. kubeLoadbalancerFWSet: {{
  1421. IP: svcLBIP,
  1422. Port: svcPort,
  1423. Protocol: strings.ToLower(string(v1.ProtocolTCP)),
  1424. SetType: utilipset.HashIPPort,
  1425. }},
  1426. kubeLoadBalancerSourceCIDRSet: {{
  1427. IP: svcLBIP,
  1428. Port: svcPort,
  1429. Protocol: strings.ToLower(string(v1.ProtocolTCP)),
  1430. Net: svcLBSource,
  1431. SetType: utilipset.HashIPPortNet,
  1432. }},
  1433. }
  1434. checkIPSet(t, fp, epIPSet)
  1435. // Check iptables chain and rules
  1436. epIpt := netlinktest.ExpectedIptablesChain{
  1437. string(kubeServicesChain): {{
  1438. JumpChain: string(KubeLoadBalancerChain), MatchSet: kubeLoadBalancerSet,
  1439. }},
  1440. string(KubeLoadBalancerChain): {{
  1441. JumpChain: string(KubeFireWallChain), MatchSet: kubeLoadbalancerFWSet,
  1442. }, {
  1443. JumpChain: string(KubeMarkMasqChain), MatchSet: "",
  1444. }},
  1445. string(KubeFireWallChain): {{
  1446. JumpChain: "RETURN", MatchSet: kubeLoadBalancerSourceCIDRSet,
  1447. }, {
  1448. JumpChain: string(KubeMarkDropChain), MatchSet: "",
  1449. }},
  1450. }
  1451. checkIptables(t, ipt, epIpt)
  1452. }
  1453. func TestAcceptIPVSTraffic(t *testing.T) {
  1454. ipt, fp := buildFakeProxier()
  1455. ingressIP := "1.2.3.4"
  1456. externalIP := []string{"5.6.7.8"}
  1457. svcInfos := []struct {
  1458. svcType v1.ServiceType
  1459. svcIP string
  1460. svcName string
  1461. epIP string
  1462. }{
  1463. {v1.ServiceTypeClusterIP, "10.20.30.40", "svc1", "10.180.0.1"},
  1464. {v1.ServiceTypeLoadBalancer, "10.20.30.41", "svc2", "10.180.0.2"},
  1465. {v1.ServiceTypeNodePort, "10.20.30.42", "svc3", "10.180.0.3"},
  1466. }
  1467. for _, svcInfo := range svcInfos {
  1468. makeServiceMap(fp,
  1469. makeTestService("ns1", svcInfo.svcName, func(svc *v1.Service) {
  1470. svc.Spec.Type = svcInfo.svcType
  1471. svc.Spec.ClusterIP = svcInfo.svcIP
  1472. svc.Spec.Ports = []v1.ServicePort{{
  1473. Name: "p80",
  1474. Port: 80,
  1475. Protocol: v1.ProtocolTCP,
  1476. NodePort: 80,
  1477. }}
  1478. if svcInfo.svcType == v1.ServiceTypeLoadBalancer {
  1479. svc.Status.LoadBalancer.Ingress = []v1.LoadBalancerIngress{{
  1480. IP: ingressIP,
  1481. }}
  1482. }
  1483. if svcInfo.svcType == v1.ServiceTypeClusterIP {
  1484. svc.Spec.ExternalIPs = externalIP
  1485. }
  1486. }),
  1487. )
  1488. makeEndpointsMap(fp,
  1489. makeTestEndpoints("ns1", "p80", func(ept *v1.Endpoints) {
  1490. ept.Subsets = []v1.EndpointSubset{{
  1491. Addresses: []v1.EndpointAddress{{
  1492. IP: svcInfo.epIP,
  1493. }},
  1494. Ports: []v1.EndpointPort{{
  1495. Name: "p80",
  1496. Port: 80,
  1497. }},
  1498. }}
  1499. }),
  1500. )
  1501. }
  1502. fp.syncProxyRules()
  1503. // Check iptables chain and rules
  1504. epIpt := netlinktest.ExpectedIptablesChain{
  1505. string(kubeServicesChain): {
  1506. {JumpChain: "ACCEPT", MatchSet: kubeClusterIPSet},
  1507. {JumpChain: "ACCEPT", MatchSet: kubeLoadBalancerSet},
  1508. {JumpChain: "ACCEPT", MatchSet: kubeExternalIPSet},
  1509. },
  1510. }
  1511. checkIptables(t, ipt, epIpt)
  1512. }
  1513. func TestOnlyLocalLoadBalancing(t *testing.T) {
  1514. ipt, fp := buildFakeProxier()
  1515. svcIP := "10.20.30.41"
  1516. svcPort := 80
  1517. svcNodePort := 3001
  1518. svcLBIP := "1.2.3.4"
  1519. svcPortName := proxy.ServicePortName{
  1520. NamespacedName: makeNSN("ns1", "svc1"),
  1521. Port: "p80",
  1522. }
  1523. makeServiceMap(fp,
  1524. makeTestService(svcPortName.Namespace, svcPortName.Name, func(svc *v1.Service) {
  1525. svc.Spec.Type = "LoadBalancer"
  1526. svc.Spec.ClusterIP = svcIP
  1527. svc.Spec.Ports = []v1.ServicePort{{
  1528. Name: svcPortName.Port,
  1529. Port: int32(svcPort),
  1530. Protocol: v1.ProtocolTCP,
  1531. NodePort: int32(svcNodePort),
  1532. }}
  1533. svc.Status.LoadBalancer.Ingress = []v1.LoadBalancerIngress{{
  1534. IP: svcLBIP,
  1535. }}
  1536. svc.Spec.ExternalTrafficPolicy = v1.ServiceExternalTrafficPolicyTypeLocal
  1537. }),
  1538. )
  1539. epIP := "10.180.0.1"
  1540. epIP1 := "10.180.1.1"
  1541. thisHostname := testHostname
  1542. otherHostname := "other-hostname"
  1543. makeEndpointsMap(fp,
  1544. makeTestEndpoints(svcPortName.Namespace, svcPortName.Name, func(ept *v1.Endpoints) {
  1545. ept.Subsets = []v1.EndpointSubset{
  1546. { // **local** endpoint address, should be added as RS
  1547. Addresses: []v1.EndpointAddress{{
  1548. IP: epIP,
  1549. NodeName: &thisHostname,
  1550. }},
  1551. Ports: []v1.EndpointPort{{
  1552. Name: svcPortName.Port,
  1553. Port: int32(svcPort),
  1554. Protocol: v1.ProtocolTCP,
  1555. }}},
  1556. { // **remote** endpoint address, should not be added as RS
  1557. Addresses: []v1.EndpointAddress{{
  1558. IP: epIP1,
  1559. NodeName: &otherHostname,
  1560. }},
  1561. Ports: []v1.EndpointPort{{
  1562. Name: svcPortName.Port,
  1563. Port: int32(svcPort),
  1564. Protocol: v1.ProtocolTCP,
  1565. }},
  1566. }}
  1567. }),
  1568. )
  1569. fp.syncProxyRules()
  1570. // Expect 2 services and 1 destination
  1571. epVS := &netlinktest.ExpectedVirtualServer{
  1572. VSNum: 2, IP: svcLBIP, Port: uint16(svcPort), Protocol: string(v1.ProtocolTCP),
  1573. RS: []netlinktest.ExpectedRealServer{{
  1574. IP: epIP, Port: uint16(svcPort),
  1575. }}}
  1576. checkIPVS(t, fp, epVS)
  1577. // check ipSet rules
  1578. epIPSet := netlinktest.ExpectedIPSet{
  1579. kubeLoadBalancerSet: {{
  1580. IP: svcLBIP,
  1581. Port: svcPort,
  1582. Protocol: strings.ToLower(string(v1.ProtocolTCP)),
  1583. SetType: utilipset.HashIPPort,
  1584. }},
  1585. kubeLoadBalancerLocalSet: {{
  1586. IP: svcLBIP,
  1587. Port: svcPort,
  1588. Protocol: strings.ToLower(string(v1.ProtocolTCP)),
  1589. SetType: utilipset.HashIPPort,
  1590. }},
  1591. }
  1592. checkIPSet(t, fp, epIPSet)
  1593. // Check iptables chain and rules
  1594. epIpt := netlinktest.ExpectedIptablesChain{
  1595. string(kubeServicesChain): {{
  1596. JumpChain: string(KubeLoadBalancerChain), MatchSet: kubeLoadBalancerSet,
  1597. }},
  1598. string(KubeLoadBalancerChain): {{
  1599. JumpChain: "RETURN", MatchSet: kubeLoadBalancerLocalSet,
  1600. }, {
  1601. JumpChain: string(KubeMarkMasqChain), MatchSet: "",
  1602. }},
  1603. }
  1604. checkIptables(t, ipt, epIpt)
  1605. }
  1606. func addTestPort(array []v1.ServicePort, name string, protocol v1.Protocol, port, nodeport int32, targetPort int) []v1.ServicePort {
  1607. svcPort := v1.ServicePort{
  1608. Name: name,
  1609. Protocol: protocol,
  1610. Port: port,
  1611. NodePort: nodeport,
  1612. TargetPort: intstr.FromInt(targetPort),
  1613. }
  1614. return append(array, svcPort)
  1615. }
  1616. func TestBuildServiceMapAddRemove(t *testing.T) {
  1617. ipt := iptablestest.NewFake()
  1618. ipvs := ipvstest.NewFake()
  1619. ipset := ipsettest.NewFake(testIPSetVersion)
  1620. fp := NewFakeProxier(ipt, ipvs, ipset, nil, nil, false)
  1621. services := []*v1.Service{
  1622. makeTestService("somewhere-else", "cluster-ip", func(svc *v1.Service) {
  1623. svc.Spec.Type = v1.ServiceTypeClusterIP
  1624. svc.Spec.ClusterIP = "172.16.55.4"
  1625. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "something", "UDP", 1234, 4321, 0)
  1626. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "somethingelse", "UDP", 1235, 5321, 0)
  1627. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "somesctp", "SCTP", 1236, 6321, 0)
  1628. }),
  1629. makeTestService("somewhere-else", "node-port", func(svc *v1.Service) {
  1630. svc.Spec.Type = v1.ServiceTypeNodePort
  1631. svc.Spec.ClusterIP = "172.16.55.10"
  1632. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "blahblah", "UDP", 345, 678, 0)
  1633. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "moreblahblah", "TCP", 344, 677, 0)
  1634. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "sctpblah", "SCTP", 343, 676, 0)
  1635. }),
  1636. makeTestService("somewhere", "load-balancer", func(svc *v1.Service) {
  1637. svc.Spec.Type = v1.ServiceTypeLoadBalancer
  1638. svc.Spec.ClusterIP = "172.16.55.11"
  1639. svc.Spec.LoadBalancerIP = "5.6.7.8"
  1640. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "foobar", "UDP", 8675, 30061, 7000)
  1641. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "baz", "UDP", 8676, 30062, 7001)
  1642. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "sctpfoo", "SCTP", 8677, 30063, 7002)
  1643. svc.Status.LoadBalancer = v1.LoadBalancerStatus{
  1644. Ingress: []v1.LoadBalancerIngress{
  1645. {IP: "10.1.2.4"},
  1646. },
  1647. }
  1648. }),
  1649. makeTestService("somewhere", "only-local-load-balancer", func(svc *v1.Service) {
  1650. svc.Spec.Type = v1.ServiceTypeLoadBalancer
  1651. svc.Spec.ClusterIP = "172.16.55.12"
  1652. svc.Spec.LoadBalancerIP = "5.6.7.8"
  1653. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "foobar2", "UDP", 8677, 30063, 7002)
  1654. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "baz", "UDP", 8678, 30064, 7003)
  1655. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "sctpbaz", "SCTP", 8679, 30065, 7004)
  1656. svc.Status.LoadBalancer = v1.LoadBalancerStatus{
  1657. Ingress: []v1.LoadBalancerIngress{
  1658. {IP: "10.1.2.3"},
  1659. },
  1660. }
  1661. svc.Spec.ExternalTrafficPolicy = v1.ServiceExternalTrafficPolicyTypeLocal
  1662. svc.Spec.HealthCheckNodePort = 345
  1663. }),
  1664. }
  1665. for i := range services {
  1666. fp.OnServiceAdd(services[i])
  1667. }
  1668. result := proxy.UpdateServiceMap(fp.serviceMap, fp.serviceChanges)
  1669. if len(fp.serviceMap) != 12 {
  1670. t.Errorf("expected service map length 12, got %v", fp.serviceMap)
  1671. }
  1672. // The only-local-loadbalancer ones get added
  1673. if len(result.HCServiceNodePorts) != 1 {
  1674. t.Errorf("expected 1 healthcheck port, got %v", result.HCServiceNodePorts)
  1675. } else {
  1676. nsn := makeNSN("somewhere", "only-local-load-balancer")
  1677. if port, found := result.HCServiceNodePorts[nsn]; !found || port != 345 {
  1678. t.Errorf("expected healthcheck port [%q]=345: got %v", nsn, result.HCServiceNodePorts)
  1679. }
  1680. }
  1681. if len(result.UDPStaleClusterIP) != 0 {
  1682. // Services only added, so nothing stale yet
  1683. t.Errorf("expected stale UDP services length 0, got %d", len(result.UDPStaleClusterIP))
  1684. }
  1685. // Remove some stuff
  1686. // oneService is a modification of services[0] with removed first port.
  1687. oneService := makeTestService("somewhere-else", "cluster-ip", func(svc *v1.Service) {
  1688. svc.Spec.Type = v1.ServiceTypeClusterIP
  1689. svc.Spec.ClusterIP = "172.16.55.4"
  1690. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "somethingelse", "UDP", 1235, 5321, 0)
  1691. })
  1692. fp.OnServiceUpdate(services[0], oneService)
  1693. fp.OnServiceDelete(services[1])
  1694. fp.OnServiceDelete(services[2])
  1695. fp.OnServiceDelete(services[3])
  1696. result = proxy.UpdateServiceMap(fp.serviceMap, fp.serviceChanges)
  1697. if len(fp.serviceMap) != 1 {
  1698. t.Errorf("expected service map length 1, got %v", fp.serviceMap)
  1699. }
  1700. if len(result.HCServiceNodePorts) != 0 {
  1701. t.Errorf("expected 0 healthcheck ports, got %v", result.HCServiceNodePorts)
  1702. }
  1703. // All services but one were deleted. While you'd expect only the ClusterIPs
  1704. // from the three deleted services here, we still have the ClusterIP for
  1705. // the not-deleted service, because one of it's ServicePorts was deleted.
  1706. expectedStaleUDPServices := []string{"172.16.55.10", "172.16.55.4", "172.16.55.11", "172.16.55.12"}
  1707. if len(result.UDPStaleClusterIP) != len(expectedStaleUDPServices) {
  1708. t.Errorf("expected stale UDP services length %d, got %v", len(expectedStaleUDPServices), result.UDPStaleClusterIP.List())
  1709. }
  1710. for _, ip := range expectedStaleUDPServices {
  1711. if !result.UDPStaleClusterIP.Has(ip) {
  1712. t.Errorf("expected stale UDP service service %s", ip)
  1713. }
  1714. }
  1715. }
  1716. func TestBuildServiceMapServiceHeadless(t *testing.T) {
  1717. ipt := iptablestest.NewFake()
  1718. ipvs := ipvstest.NewFake()
  1719. ipset := ipsettest.NewFake(testIPSetVersion)
  1720. fp := NewFakeProxier(ipt, ipvs, ipset, nil, nil, false)
  1721. makeServiceMap(fp,
  1722. makeTestService("somewhere-else", "headless", func(svc *v1.Service) {
  1723. svc.Spec.Type = v1.ServiceTypeClusterIP
  1724. svc.Spec.ClusterIP = v1.ClusterIPNone
  1725. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "rpc", "UDP", 1234, 0, 0)
  1726. }),
  1727. makeTestService("somewhere-else", "headless-without-port", func(svc *v1.Service) {
  1728. svc.Spec.Type = v1.ServiceTypeClusterIP
  1729. svc.Spec.ClusterIP = v1.ClusterIPNone
  1730. }),
  1731. makeTestService("somewhere-else", "headless-sctp", func(svc *v1.Service) {
  1732. svc.Spec.Type = v1.ServiceTypeClusterIP
  1733. svc.Spec.ClusterIP = v1.ClusterIPNone
  1734. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "sip", "SCTP", 1235, 0, 0)
  1735. }),
  1736. )
  1737. // Headless service should be ignored
  1738. result := proxy.UpdateServiceMap(fp.serviceMap, fp.serviceChanges)
  1739. if len(fp.serviceMap) != 0 {
  1740. t.Errorf("expected service map length 0, got %d", len(fp.serviceMap))
  1741. }
  1742. // No proxied services, so no healthchecks
  1743. if len(result.HCServiceNodePorts) != 0 {
  1744. t.Errorf("expected healthcheck ports length 0, got %d", len(result.HCServiceNodePorts))
  1745. }
  1746. if len(result.UDPStaleClusterIP) != 0 {
  1747. t.Errorf("expected stale UDP services length 0, got %d", len(result.UDPStaleClusterIP))
  1748. }
  1749. }
  1750. func TestBuildServiceMapServiceTypeExternalName(t *testing.T) {
  1751. ipt := iptablestest.NewFake()
  1752. ipvs := ipvstest.NewFake()
  1753. ipset := ipsettest.NewFake(testIPSetVersion)
  1754. fp := NewFakeProxier(ipt, ipvs, ipset, nil, nil, false)
  1755. makeServiceMap(fp,
  1756. makeTestService("somewhere-else", "external-name", func(svc *v1.Service) {
  1757. svc.Spec.Type = v1.ServiceTypeExternalName
  1758. svc.Spec.ClusterIP = "172.16.55.4" // Should be ignored
  1759. svc.Spec.ExternalName = "foo2.bar.com"
  1760. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "blah", "UDP", 1235, 5321, 0)
  1761. }),
  1762. )
  1763. result := proxy.UpdateServiceMap(fp.serviceMap, fp.serviceChanges)
  1764. if len(fp.serviceMap) != 0 {
  1765. t.Errorf("expected service map length 0, got %v", fp.serviceMap)
  1766. }
  1767. // No proxied services, so no healthchecks
  1768. if len(result.HCServiceNodePorts) != 0 {
  1769. t.Errorf("expected healthcheck ports length 0, got %v", result.HCServiceNodePorts)
  1770. }
  1771. if len(result.UDPStaleClusterIP) != 0 {
  1772. t.Errorf("expected stale UDP services length 0, got %v", result.UDPStaleClusterIP)
  1773. }
  1774. }
  1775. func TestBuildServiceMapServiceUpdate(t *testing.T) {
  1776. ipt := iptablestest.NewFake()
  1777. ipvs := ipvstest.NewFake()
  1778. ipset := ipsettest.NewFake(testIPSetVersion)
  1779. fp := NewFakeProxier(ipt, ipvs, ipset, nil, nil, false)
  1780. servicev1 := makeTestService("somewhere", "some-service", func(svc *v1.Service) {
  1781. svc.Spec.Type = v1.ServiceTypeClusterIP
  1782. svc.Spec.ClusterIP = "172.16.55.4"
  1783. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "something", "UDP", 1234, 4321, 0)
  1784. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "somethingelse", "TCP", 1235, 5321, 0)
  1785. })
  1786. servicev2 := makeTestService("somewhere", "some-service", func(svc *v1.Service) {
  1787. svc.Spec.Type = v1.ServiceTypeLoadBalancer
  1788. svc.Spec.ClusterIP = "172.16.55.4"
  1789. svc.Spec.LoadBalancerIP = "5.6.7.8"
  1790. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "something", "UDP", 1234, 4321, 7002)
  1791. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "somethingelse", "TCP", 1235, 5321, 7003)
  1792. svc.Status.LoadBalancer = v1.LoadBalancerStatus{
  1793. Ingress: []v1.LoadBalancerIngress{
  1794. {IP: "10.1.2.3"},
  1795. },
  1796. }
  1797. svc.Spec.ExternalTrafficPolicy = v1.ServiceExternalTrafficPolicyTypeLocal
  1798. svc.Spec.HealthCheckNodePort = 345
  1799. })
  1800. fp.OnServiceAdd(servicev1)
  1801. result := proxy.UpdateServiceMap(fp.serviceMap, fp.serviceChanges)
  1802. if len(fp.serviceMap) != 2 {
  1803. t.Errorf("expected service map length 2, got %v", fp.serviceMap)
  1804. }
  1805. if len(result.HCServiceNodePorts) != 0 {
  1806. t.Errorf("expected healthcheck ports length 0, got %v", result.HCServiceNodePorts)
  1807. }
  1808. if len(result.UDPStaleClusterIP) != 0 {
  1809. // Services only added, so nothing stale yet
  1810. t.Errorf("expected stale UDP services length 0, got %d", len(result.UDPStaleClusterIP))
  1811. }
  1812. // Change service to load-balancer
  1813. fp.OnServiceUpdate(servicev1, servicev2)
  1814. result = proxy.UpdateServiceMap(fp.serviceMap, fp.serviceChanges)
  1815. if len(fp.serviceMap) != 2 {
  1816. t.Errorf("expected service map length 2, got %v", fp.serviceMap)
  1817. }
  1818. if len(result.HCServiceNodePorts) != 1 {
  1819. t.Errorf("expected healthcheck ports length 1, got %v", result.HCServiceNodePorts)
  1820. }
  1821. if len(result.UDPStaleClusterIP) != 0 {
  1822. t.Errorf("expected stale UDP services length 0, got %v", result.UDPStaleClusterIP.List())
  1823. }
  1824. // No change; make sure the service map stays the same and there are
  1825. // no health-check changes
  1826. fp.OnServiceUpdate(servicev2, servicev2)
  1827. result = proxy.UpdateServiceMap(fp.serviceMap, fp.serviceChanges)
  1828. if len(fp.serviceMap) != 2 {
  1829. t.Errorf("expected service map length 2, got %v", fp.serviceMap)
  1830. }
  1831. if len(result.HCServiceNodePorts) != 1 {
  1832. t.Errorf("expected healthcheck ports length 1, got %v", result.HCServiceNodePorts)
  1833. }
  1834. if len(result.UDPStaleClusterIP) != 0 {
  1835. t.Errorf("expected stale UDP services length 0, got %v", result.UDPStaleClusterIP.List())
  1836. }
  1837. // And back to ClusterIP
  1838. fp.OnServiceUpdate(servicev2, servicev1)
  1839. result = proxy.UpdateServiceMap(fp.serviceMap, fp.serviceChanges)
  1840. if len(fp.serviceMap) != 2 {
  1841. t.Errorf("expected service map length 2, got %v", fp.serviceMap)
  1842. }
  1843. if len(result.HCServiceNodePorts) != 0 {
  1844. t.Errorf("expected healthcheck ports length 0, got %v", result.HCServiceNodePorts)
  1845. }
  1846. if len(result.UDPStaleClusterIP) != 0 {
  1847. // Services only added, so nothing stale yet
  1848. t.Errorf("expected stale UDP services length 0, got %d", len(result.UDPStaleClusterIP))
  1849. }
  1850. }
  1851. func TestSessionAffinity(t *testing.T) {
  1852. ipt := iptablestest.NewFake()
  1853. ipvs := ipvstest.NewFake()
  1854. ipset := ipsettest.NewFake(testIPSetVersion)
  1855. nodeIP := net.ParseIP("100.101.102.103")
  1856. fp := NewFakeProxier(ipt, ipvs, ipset, []net.IP{nodeIP}, nil, false)
  1857. svcIP := "10.20.30.41"
  1858. svcPort := 80
  1859. svcNodePort := 3001
  1860. svcExternalIPs := "50.60.70.81"
  1861. svcPortName := proxy.ServicePortName{
  1862. NamespacedName: makeNSN("ns1", "svc1"),
  1863. Port: "p80",
  1864. }
  1865. timeoutSeconds := v1.DefaultClientIPServiceAffinitySeconds
  1866. makeServiceMap(fp,
  1867. makeTestService(svcPortName.Namespace, svcPortName.Name, func(svc *v1.Service) {
  1868. svc.Spec.Type = "NodePort"
  1869. svc.Spec.ClusterIP = svcIP
  1870. svc.Spec.ExternalIPs = []string{svcExternalIPs}
  1871. svc.Spec.SessionAffinity = v1.ServiceAffinityClientIP
  1872. svc.Spec.SessionAffinityConfig = &v1.SessionAffinityConfig{
  1873. ClientIP: &v1.ClientIPConfig{
  1874. TimeoutSeconds: &timeoutSeconds,
  1875. },
  1876. }
  1877. svc.Spec.Ports = []v1.ServicePort{{
  1878. Name: svcPortName.Port,
  1879. Port: int32(svcPort),
  1880. Protocol: v1.ProtocolTCP,
  1881. NodePort: int32(svcNodePort),
  1882. }}
  1883. }),
  1884. )
  1885. makeEndpointsMap(fp)
  1886. fp.syncProxyRules()
  1887. // check ipvs service and destinations
  1888. services, err := ipvs.GetVirtualServers()
  1889. if err != nil {
  1890. t.Errorf("Failed to get ipvs services, err: %v", err)
  1891. }
  1892. for _, svc := range services {
  1893. if svc.Timeout != uint32(v1.DefaultClientIPServiceAffinitySeconds) {
  1894. t.Errorf("Unexpected mismatch ipvs service session affinity timeout: %d, expected: %d", svc.Timeout, v1.DefaultClientIPServiceAffinitySeconds)
  1895. }
  1896. }
  1897. }
  1898. func makeServicePortName(ns, name, port string, protocol v1.Protocol) proxy.ServicePortName {
  1899. return proxy.ServicePortName{
  1900. NamespacedName: makeNSN(ns, name),
  1901. Port: port,
  1902. Protocol: protocol,
  1903. }
  1904. }
  1905. func Test_updateEndpointsMap(t *testing.T) {
  1906. var nodeName = testHostname
  1907. emptyEndpoint := func(ept *v1.Endpoints) {
  1908. ept.Subsets = []v1.EndpointSubset{}
  1909. }
  1910. unnamedPort := func(ept *v1.Endpoints) {
  1911. ept.Subsets = []v1.EndpointSubset{{
  1912. Addresses: []v1.EndpointAddress{{
  1913. IP: "1.1.1.1",
  1914. }},
  1915. Ports: []v1.EndpointPort{{
  1916. Port: 11,
  1917. Protocol: v1.ProtocolUDP,
  1918. }},
  1919. }}
  1920. }
  1921. unnamedPortLocal := func(ept *v1.Endpoints) {
  1922. ept.Subsets = []v1.EndpointSubset{{
  1923. Addresses: []v1.EndpointAddress{{
  1924. IP: "1.1.1.1",
  1925. NodeName: &nodeName,
  1926. }},
  1927. Ports: []v1.EndpointPort{{
  1928. Port: 11,
  1929. Protocol: v1.ProtocolUDP,
  1930. }},
  1931. }}
  1932. }
  1933. namedPortLocal := func(ept *v1.Endpoints) {
  1934. ept.Subsets = []v1.EndpointSubset{{
  1935. Addresses: []v1.EndpointAddress{{
  1936. IP: "1.1.1.1",
  1937. NodeName: &nodeName,
  1938. }},
  1939. Ports: []v1.EndpointPort{{
  1940. Name: "p11",
  1941. Port: 11,
  1942. Protocol: v1.ProtocolUDP,
  1943. }},
  1944. }}
  1945. }
  1946. namedPort := func(ept *v1.Endpoints) {
  1947. ept.Subsets = []v1.EndpointSubset{{
  1948. Addresses: []v1.EndpointAddress{{
  1949. IP: "1.1.1.1",
  1950. }},
  1951. Ports: []v1.EndpointPort{{
  1952. Name: "p11",
  1953. Port: 11,
  1954. Protocol: v1.ProtocolUDP,
  1955. }},
  1956. }}
  1957. }
  1958. namedPortRenamed := func(ept *v1.Endpoints) {
  1959. ept.Subsets = []v1.EndpointSubset{{
  1960. Addresses: []v1.EndpointAddress{{
  1961. IP: "1.1.1.1",
  1962. }},
  1963. Ports: []v1.EndpointPort{{
  1964. Name: "p11-2",
  1965. Port: 11,
  1966. Protocol: v1.ProtocolUDP,
  1967. }},
  1968. }}
  1969. }
  1970. namedPortRenumbered := func(ept *v1.Endpoints) {
  1971. ept.Subsets = []v1.EndpointSubset{{
  1972. Addresses: []v1.EndpointAddress{{
  1973. IP: "1.1.1.1",
  1974. }},
  1975. Ports: []v1.EndpointPort{{
  1976. Name: "p11",
  1977. Port: 22,
  1978. Protocol: v1.ProtocolUDP,
  1979. }},
  1980. }}
  1981. }
  1982. namedPortsLocalNoLocal := func(ept *v1.Endpoints) {
  1983. ept.Subsets = []v1.EndpointSubset{{
  1984. Addresses: []v1.EndpointAddress{{
  1985. IP: "1.1.1.1",
  1986. }, {
  1987. IP: "1.1.1.2",
  1988. NodeName: &nodeName,
  1989. }},
  1990. Ports: []v1.EndpointPort{{
  1991. Name: "p11",
  1992. Port: 11,
  1993. Protocol: v1.ProtocolUDP,
  1994. }, {
  1995. Name: "p12",
  1996. Port: 12,
  1997. Protocol: v1.ProtocolUDP,
  1998. }},
  1999. }}
  2000. }
  2001. multipleSubsets := func(ept *v1.Endpoints) {
  2002. ept.Subsets = []v1.EndpointSubset{{
  2003. Addresses: []v1.EndpointAddress{{
  2004. IP: "1.1.1.1",
  2005. }},
  2006. Ports: []v1.EndpointPort{{
  2007. Name: "p11",
  2008. Port: 11,
  2009. Protocol: v1.ProtocolUDP,
  2010. }},
  2011. }, {
  2012. Addresses: []v1.EndpointAddress{{
  2013. IP: "1.1.1.2",
  2014. }},
  2015. Ports: []v1.EndpointPort{{
  2016. Name: "p12",
  2017. Port: 12,
  2018. Protocol: v1.ProtocolUDP,
  2019. }},
  2020. }}
  2021. }
  2022. multipleSubsetsWithLocal := func(ept *v1.Endpoints) {
  2023. ept.Subsets = []v1.EndpointSubset{{
  2024. Addresses: []v1.EndpointAddress{{
  2025. IP: "1.1.1.1",
  2026. }},
  2027. Ports: []v1.EndpointPort{{
  2028. Name: "p11",
  2029. Port: 11,
  2030. Protocol: v1.ProtocolUDP,
  2031. }},
  2032. }, {
  2033. Addresses: []v1.EndpointAddress{{
  2034. IP: "1.1.1.2",
  2035. NodeName: &nodeName,
  2036. }},
  2037. Ports: []v1.EndpointPort{{
  2038. Name: "p12",
  2039. Port: 12,
  2040. Protocol: v1.ProtocolUDP,
  2041. }},
  2042. }}
  2043. }
  2044. multipleSubsetsMultiplePortsLocal := func(ept *v1.Endpoints) {
  2045. ept.Subsets = []v1.EndpointSubset{{
  2046. Addresses: []v1.EndpointAddress{{
  2047. IP: "1.1.1.1",
  2048. NodeName: &nodeName,
  2049. }},
  2050. Ports: []v1.EndpointPort{{
  2051. Name: "p11",
  2052. Port: 11,
  2053. Protocol: v1.ProtocolUDP,
  2054. }, {
  2055. Name: "p12",
  2056. Port: 12,
  2057. Protocol: v1.ProtocolUDP,
  2058. }},
  2059. }, {
  2060. Addresses: []v1.EndpointAddress{{
  2061. IP: "1.1.1.3",
  2062. }},
  2063. Ports: []v1.EndpointPort{{
  2064. Name: "p13",
  2065. Port: 13,
  2066. Protocol: v1.ProtocolUDP,
  2067. }},
  2068. }}
  2069. }
  2070. multipleSubsetsIPsPorts1 := func(ept *v1.Endpoints) {
  2071. ept.Subsets = []v1.EndpointSubset{{
  2072. Addresses: []v1.EndpointAddress{{
  2073. IP: "1.1.1.1",
  2074. }, {
  2075. IP: "1.1.1.2",
  2076. NodeName: &nodeName,
  2077. }},
  2078. Ports: []v1.EndpointPort{{
  2079. Name: "p11",
  2080. Port: 11,
  2081. Protocol: v1.ProtocolUDP,
  2082. }, {
  2083. Name: "p12",
  2084. Port: 12,
  2085. Protocol: v1.ProtocolUDP,
  2086. }},
  2087. }, {
  2088. Addresses: []v1.EndpointAddress{{
  2089. IP: "1.1.1.3",
  2090. }, {
  2091. IP: "1.1.1.4",
  2092. NodeName: &nodeName,
  2093. }},
  2094. Ports: []v1.EndpointPort{{
  2095. Name: "p13",
  2096. Port: 13,
  2097. Protocol: v1.ProtocolUDP,
  2098. }, {
  2099. Name: "p14",
  2100. Port: 14,
  2101. Protocol: v1.ProtocolUDP,
  2102. }},
  2103. }}
  2104. }
  2105. multipleSubsetsIPsPorts2 := func(ept *v1.Endpoints) {
  2106. ept.Subsets = []v1.EndpointSubset{{
  2107. Addresses: []v1.EndpointAddress{{
  2108. IP: "2.2.2.1",
  2109. }, {
  2110. IP: "2.2.2.2",
  2111. NodeName: &nodeName,
  2112. }},
  2113. Ports: []v1.EndpointPort{{
  2114. Name: "p21",
  2115. Port: 21,
  2116. Protocol: v1.ProtocolUDP,
  2117. }, {
  2118. Name: "p22",
  2119. Port: 22,
  2120. Protocol: v1.ProtocolUDP,
  2121. }},
  2122. }}
  2123. }
  2124. complexBefore1 := func(ept *v1.Endpoints) {
  2125. ept.Subsets = []v1.EndpointSubset{{
  2126. Addresses: []v1.EndpointAddress{{
  2127. IP: "1.1.1.1",
  2128. }},
  2129. Ports: []v1.EndpointPort{{
  2130. Name: "p11",
  2131. Port: 11,
  2132. Protocol: v1.ProtocolUDP,
  2133. }},
  2134. }}
  2135. }
  2136. complexBefore2 := func(ept *v1.Endpoints) {
  2137. ept.Subsets = []v1.EndpointSubset{{
  2138. Addresses: []v1.EndpointAddress{{
  2139. IP: "2.2.2.2",
  2140. NodeName: &nodeName,
  2141. }, {
  2142. IP: "2.2.2.22",
  2143. NodeName: &nodeName,
  2144. }},
  2145. Ports: []v1.EndpointPort{{
  2146. Name: "p22",
  2147. Port: 22,
  2148. Protocol: v1.ProtocolUDP,
  2149. }},
  2150. }, {
  2151. Addresses: []v1.EndpointAddress{{
  2152. IP: "2.2.2.3",
  2153. NodeName: &nodeName,
  2154. }},
  2155. Ports: []v1.EndpointPort{{
  2156. Name: "p23",
  2157. Port: 23,
  2158. Protocol: v1.ProtocolUDP,
  2159. }},
  2160. }}
  2161. }
  2162. complexBefore4 := func(ept *v1.Endpoints) {
  2163. ept.Subsets = []v1.EndpointSubset{{
  2164. Addresses: []v1.EndpointAddress{{
  2165. IP: "4.4.4.4",
  2166. NodeName: &nodeName,
  2167. }, {
  2168. IP: "4.4.4.5",
  2169. NodeName: &nodeName,
  2170. }},
  2171. Ports: []v1.EndpointPort{{
  2172. Name: "p44",
  2173. Port: 44,
  2174. Protocol: v1.ProtocolUDP,
  2175. }},
  2176. }, {
  2177. Addresses: []v1.EndpointAddress{{
  2178. IP: "4.4.4.6",
  2179. NodeName: &nodeName,
  2180. }},
  2181. Ports: []v1.EndpointPort{{
  2182. Name: "p45",
  2183. Port: 45,
  2184. Protocol: v1.ProtocolUDP,
  2185. }},
  2186. }}
  2187. }
  2188. complexAfter1 := func(ept *v1.Endpoints) {
  2189. ept.Subsets = []v1.EndpointSubset{{
  2190. Addresses: []v1.EndpointAddress{{
  2191. IP: "1.1.1.1",
  2192. }, {
  2193. IP: "1.1.1.11",
  2194. }},
  2195. Ports: []v1.EndpointPort{{
  2196. Name: "p11",
  2197. Port: 11,
  2198. Protocol: v1.ProtocolUDP,
  2199. }},
  2200. }, {
  2201. Addresses: []v1.EndpointAddress{{
  2202. IP: "1.1.1.2",
  2203. }},
  2204. Ports: []v1.EndpointPort{{
  2205. Name: "p12",
  2206. Port: 12,
  2207. Protocol: v1.ProtocolUDP,
  2208. }, {
  2209. Name: "p122",
  2210. Port: 122,
  2211. Protocol: v1.ProtocolUDP,
  2212. }},
  2213. }}
  2214. }
  2215. complexAfter3 := func(ept *v1.Endpoints) {
  2216. ept.Subsets = []v1.EndpointSubset{{
  2217. Addresses: []v1.EndpointAddress{{
  2218. IP: "3.3.3.3",
  2219. }},
  2220. Ports: []v1.EndpointPort{{
  2221. Name: "p33",
  2222. Port: 33,
  2223. Protocol: v1.ProtocolUDP,
  2224. }},
  2225. }}
  2226. }
  2227. complexAfter4 := func(ept *v1.Endpoints) {
  2228. ept.Subsets = []v1.EndpointSubset{{
  2229. Addresses: []v1.EndpointAddress{{
  2230. IP: "4.4.4.4",
  2231. NodeName: &nodeName,
  2232. }},
  2233. Ports: []v1.EndpointPort{{
  2234. Name: "p44",
  2235. Port: 44,
  2236. Protocol: v1.ProtocolUDP,
  2237. }},
  2238. }}
  2239. }
  2240. testCases := []struct {
  2241. // previousEndpoints and currentEndpoints are used to call appropriate
  2242. // handlers OnEndpoints* (based on whether corresponding values are nil
  2243. // or non-nil) and must be of equal length.
  2244. previousEndpoints []*v1.Endpoints
  2245. currentEndpoints []*v1.Endpoints
  2246. oldEndpoints map[proxy.ServicePortName][]*proxy.BaseEndpointInfo
  2247. expectedResult map[proxy.ServicePortName][]*proxy.BaseEndpointInfo
  2248. expectedStaleEndpoints []proxy.ServiceEndpoint
  2249. expectedStaleServiceNames map[proxy.ServicePortName]bool
  2250. expectedHealthchecks map[types.NamespacedName]int
  2251. }{{
  2252. // Case[0]: nothing
  2253. oldEndpoints: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{},
  2254. expectedResult: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{},
  2255. expectedStaleEndpoints: []proxy.ServiceEndpoint{},
  2256. expectedStaleServiceNames: map[proxy.ServicePortName]bool{},
  2257. expectedHealthchecks: map[types.NamespacedName]int{},
  2258. }, {
  2259. // Case[1]: no change, unnamed port
  2260. previousEndpoints: []*v1.Endpoints{
  2261. makeTestEndpoints("ns1", "ep1", unnamedPort),
  2262. },
  2263. currentEndpoints: []*v1.Endpoints{
  2264. makeTestEndpoints("ns1", "ep1", unnamedPort),
  2265. },
  2266. oldEndpoints: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2267. makeServicePortName("ns1", "ep1", "", v1.ProtocolUDP): {
  2268. {Endpoint: "1.1.1.1:11", IsLocal: false},
  2269. },
  2270. },
  2271. expectedResult: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2272. makeServicePortName("ns1", "ep1", "", v1.ProtocolUDP): {
  2273. {Endpoint: "1.1.1.1:11", IsLocal: false},
  2274. },
  2275. },
  2276. expectedStaleEndpoints: []proxy.ServiceEndpoint{},
  2277. expectedStaleServiceNames: map[proxy.ServicePortName]bool{},
  2278. expectedHealthchecks: map[types.NamespacedName]int{},
  2279. }, {
  2280. // Case[2]: no change, named port, local
  2281. previousEndpoints: []*v1.Endpoints{
  2282. makeTestEndpoints("ns1", "ep1", namedPortLocal),
  2283. },
  2284. currentEndpoints: []*v1.Endpoints{
  2285. makeTestEndpoints("ns1", "ep1", namedPortLocal),
  2286. },
  2287. oldEndpoints: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2288. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2289. {Endpoint: "1.1.1.1:11", IsLocal: true},
  2290. },
  2291. },
  2292. expectedResult: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2293. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2294. {Endpoint: "1.1.1.1:11", IsLocal: true},
  2295. },
  2296. },
  2297. expectedStaleEndpoints: []proxy.ServiceEndpoint{},
  2298. expectedStaleServiceNames: map[proxy.ServicePortName]bool{},
  2299. expectedHealthchecks: map[types.NamespacedName]int{
  2300. makeNSN("ns1", "ep1"): 1,
  2301. },
  2302. }, {
  2303. // Case[3]: no change, multiple subsets
  2304. previousEndpoints: []*v1.Endpoints{
  2305. makeTestEndpoints("ns1", "ep1", multipleSubsets),
  2306. },
  2307. currentEndpoints: []*v1.Endpoints{
  2308. makeTestEndpoints("ns1", "ep1", multipleSubsets),
  2309. },
  2310. oldEndpoints: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2311. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2312. {Endpoint: "1.1.1.1:11", IsLocal: false},
  2313. },
  2314. makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): {
  2315. {Endpoint: "1.1.1.2:12", IsLocal: false},
  2316. },
  2317. },
  2318. expectedResult: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2319. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2320. {Endpoint: "1.1.1.1:11", IsLocal: false},
  2321. },
  2322. makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): {
  2323. {Endpoint: "1.1.1.2:12", IsLocal: false},
  2324. },
  2325. },
  2326. expectedStaleEndpoints: []proxy.ServiceEndpoint{},
  2327. expectedStaleServiceNames: map[proxy.ServicePortName]bool{},
  2328. expectedHealthchecks: map[types.NamespacedName]int{},
  2329. }, {
  2330. // Case[4]: no change, multiple subsets, multiple ports, local
  2331. previousEndpoints: []*v1.Endpoints{
  2332. makeTestEndpoints("ns1", "ep1", multipleSubsetsMultiplePortsLocal),
  2333. },
  2334. currentEndpoints: []*v1.Endpoints{
  2335. makeTestEndpoints("ns1", "ep1", multipleSubsetsMultiplePortsLocal),
  2336. },
  2337. oldEndpoints: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2338. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2339. {Endpoint: "1.1.1.1:11", IsLocal: true},
  2340. },
  2341. makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): {
  2342. {Endpoint: "1.1.1.1:12", IsLocal: true},
  2343. },
  2344. makeServicePortName("ns1", "ep1", "p13", v1.ProtocolUDP): {
  2345. {Endpoint: "1.1.1.3:13", IsLocal: false},
  2346. },
  2347. },
  2348. expectedResult: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2349. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2350. {Endpoint: "1.1.1.1:11", IsLocal: true},
  2351. },
  2352. makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): {
  2353. {Endpoint: "1.1.1.1:12", IsLocal: true},
  2354. },
  2355. makeServicePortName("ns1", "ep1", "p13", v1.ProtocolUDP): {
  2356. {Endpoint: "1.1.1.3:13", IsLocal: false},
  2357. },
  2358. },
  2359. expectedStaleEndpoints: []proxy.ServiceEndpoint{},
  2360. expectedStaleServiceNames: map[proxy.ServicePortName]bool{},
  2361. expectedHealthchecks: map[types.NamespacedName]int{
  2362. makeNSN("ns1", "ep1"): 1,
  2363. },
  2364. }, {
  2365. // Case[5]: no change, multiple endpoints, subsets, IPs, and ports
  2366. previousEndpoints: []*v1.Endpoints{
  2367. makeTestEndpoints("ns1", "ep1", multipleSubsetsIPsPorts1),
  2368. makeTestEndpoints("ns2", "ep2", multipleSubsetsIPsPorts2),
  2369. },
  2370. currentEndpoints: []*v1.Endpoints{
  2371. makeTestEndpoints("ns1", "ep1", multipleSubsetsIPsPorts1),
  2372. makeTestEndpoints("ns2", "ep2", multipleSubsetsIPsPorts2),
  2373. },
  2374. oldEndpoints: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2375. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2376. {Endpoint: "1.1.1.1:11", IsLocal: false},
  2377. {Endpoint: "1.1.1.2:11", IsLocal: true},
  2378. },
  2379. makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): {
  2380. {Endpoint: "1.1.1.1:12", IsLocal: false},
  2381. {Endpoint: "1.1.1.2:12", IsLocal: true},
  2382. },
  2383. makeServicePortName("ns1", "ep1", "p13", v1.ProtocolUDP): {
  2384. {Endpoint: "1.1.1.3:13", IsLocal: false},
  2385. {Endpoint: "1.1.1.4:13", IsLocal: true},
  2386. },
  2387. makeServicePortName("ns1", "ep1", "p14", v1.ProtocolUDP): {
  2388. {Endpoint: "1.1.1.3:14", IsLocal: false},
  2389. {Endpoint: "1.1.1.4:14", IsLocal: true},
  2390. },
  2391. makeServicePortName("ns2", "ep2", "p21", v1.ProtocolUDP): {
  2392. {Endpoint: "2.2.2.1:21", IsLocal: false},
  2393. {Endpoint: "2.2.2.2:21", IsLocal: true},
  2394. },
  2395. makeServicePortName("ns2", "ep2", "p22", v1.ProtocolUDP): {
  2396. {Endpoint: "2.2.2.1:22", IsLocal: false},
  2397. {Endpoint: "2.2.2.2:22", IsLocal: true},
  2398. },
  2399. },
  2400. expectedResult: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2401. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2402. {Endpoint: "1.1.1.1:11", IsLocal: false},
  2403. {Endpoint: "1.1.1.2:11", IsLocal: true},
  2404. },
  2405. makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): {
  2406. {Endpoint: "1.1.1.1:12", IsLocal: false},
  2407. {Endpoint: "1.1.1.2:12", IsLocal: true},
  2408. },
  2409. makeServicePortName("ns1", "ep1", "p13", v1.ProtocolUDP): {
  2410. {Endpoint: "1.1.1.3:13", IsLocal: false},
  2411. {Endpoint: "1.1.1.4:13", IsLocal: true},
  2412. },
  2413. makeServicePortName("ns1", "ep1", "p14", v1.ProtocolUDP): {
  2414. {Endpoint: "1.1.1.3:14", IsLocal: false},
  2415. {Endpoint: "1.1.1.4:14", IsLocal: true},
  2416. },
  2417. makeServicePortName("ns2", "ep2", "p21", v1.ProtocolUDP): {
  2418. {Endpoint: "2.2.2.1:21", IsLocal: false},
  2419. {Endpoint: "2.2.2.2:21", IsLocal: true},
  2420. },
  2421. makeServicePortName("ns2", "ep2", "p22", v1.ProtocolUDP): {
  2422. {Endpoint: "2.2.2.1:22", IsLocal: false},
  2423. {Endpoint: "2.2.2.2:22", IsLocal: true},
  2424. },
  2425. },
  2426. expectedStaleEndpoints: []proxy.ServiceEndpoint{},
  2427. expectedStaleServiceNames: map[proxy.ServicePortName]bool{},
  2428. expectedHealthchecks: map[types.NamespacedName]int{
  2429. makeNSN("ns1", "ep1"): 2,
  2430. makeNSN("ns2", "ep2"): 1,
  2431. },
  2432. }, {
  2433. // Case[6]: add an Endpoints
  2434. previousEndpoints: []*v1.Endpoints{
  2435. nil,
  2436. },
  2437. currentEndpoints: []*v1.Endpoints{
  2438. makeTestEndpoints("ns1", "ep1", unnamedPortLocal),
  2439. },
  2440. oldEndpoints: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{},
  2441. expectedResult: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2442. makeServicePortName("ns1", "ep1", "", v1.ProtocolUDP): {
  2443. {Endpoint: "1.1.1.1:11", IsLocal: true},
  2444. },
  2445. },
  2446. expectedStaleEndpoints: []proxy.ServiceEndpoint{},
  2447. expectedStaleServiceNames: map[proxy.ServicePortName]bool{
  2448. makeServicePortName("ns1", "ep1", "", v1.ProtocolUDP): true,
  2449. },
  2450. expectedHealthchecks: map[types.NamespacedName]int{
  2451. makeNSN("ns1", "ep1"): 1,
  2452. },
  2453. }, {
  2454. // Case[7]: remove an Endpoints
  2455. previousEndpoints: []*v1.Endpoints{
  2456. makeTestEndpoints("ns1", "ep1", unnamedPortLocal),
  2457. },
  2458. currentEndpoints: []*v1.Endpoints{
  2459. nil,
  2460. },
  2461. oldEndpoints: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2462. makeServicePortName("ns1", "ep1", "", v1.ProtocolUDP): {
  2463. {Endpoint: "1.1.1.1:11", IsLocal: true},
  2464. },
  2465. },
  2466. expectedResult: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{},
  2467. expectedStaleEndpoints: []proxy.ServiceEndpoint{{
  2468. Endpoint: "1.1.1.1:11",
  2469. ServicePortName: makeServicePortName("ns1", "ep1", "", v1.ProtocolUDP),
  2470. }},
  2471. expectedStaleServiceNames: map[proxy.ServicePortName]bool{},
  2472. expectedHealthchecks: map[types.NamespacedName]int{},
  2473. }, {
  2474. // Case[8]: add an IP and port
  2475. previousEndpoints: []*v1.Endpoints{
  2476. makeTestEndpoints("ns1", "ep1", namedPort),
  2477. },
  2478. currentEndpoints: []*v1.Endpoints{
  2479. makeTestEndpoints("ns1", "ep1", namedPortsLocalNoLocal),
  2480. },
  2481. oldEndpoints: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2482. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2483. {Endpoint: "1.1.1.1:11", IsLocal: false},
  2484. },
  2485. },
  2486. expectedResult: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2487. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2488. {Endpoint: "1.1.1.1:11", IsLocal: false},
  2489. {Endpoint: "1.1.1.2:11", IsLocal: true},
  2490. },
  2491. makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): {
  2492. {Endpoint: "1.1.1.1:12", IsLocal: false},
  2493. {Endpoint: "1.1.1.2:12", IsLocal: true},
  2494. },
  2495. },
  2496. expectedStaleEndpoints: []proxy.ServiceEndpoint{},
  2497. expectedStaleServiceNames: map[proxy.ServicePortName]bool{
  2498. makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): true,
  2499. },
  2500. expectedHealthchecks: map[types.NamespacedName]int{
  2501. makeNSN("ns1", "ep1"): 1,
  2502. },
  2503. }, {
  2504. // Case[9]: remove an IP and port
  2505. previousEndpoints: []*v1.Endpoints{
  2506. makeTestEndpoints("ns1", "ep1", namedPortsLocalNoLocal),
  2507. },
  2508. currentEndpoints: []*v1.Endpoints{
  2509. makeTestEndpoints("ns1", "ep1", namedPort),
  2510. },
  2511. oldEndpoints: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2512. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2513. {Endpoint: "1.1.1.1:11", IsLocal: false},
  2514. {Endpoint: "1.1.1.2:11", IsLocal: true},
  2515. },
  2516. makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): {
  2517. {Endpoint: "1.1.1.1:12", IsLocal: false},
  2518. {Endpoint: "1.1.1.2:12", IsLocal: true},
  2519. },
  2520. },
  2521. expectedResult: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2522. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2523. {Endpoint: "1.1.1.1:11", IsLocal: false},
  2524. },
  2525. },
  2526. expectedStaleEndpoints: []proxy.ServiceEndpoint{{
  2527. Endpoint: "1.1.1.2:11",
  2528. ServicePortName: makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP),
  2529. }, {
  2530. Endpoint: "1.1.1.1:12",
  2531. ServicePortName: makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP),
  2532. }, {
  2533. Endpoint: "1.1.1.2:12",
  2534. ServicePortName: makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP),
  2535. }},
  2536. expectedStaleServiceNames: map[proxy.ServicePortName]bool{},
  2537. expectedHealthchecks: map[types.NamespacedName]int{},
  2538. }, {
  2539. // Case[10]: add a subset
  2540. previousEndpoints: []*v1.Endpoints{
  2541. makeTestEndpoints("ns1", "ep1", namedPort),
  2542. },
  2543. currentEndpoints: []*v1.Endpoints{
  2544. makeTestEndpoints("ns1", "ep1", multipleSubsetsWithLocal),
  2545. },
  2546. oldEndpoints: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2547. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2548. {Endpoint: "1.1.1.1:11", IsLocal: false},
  2549. },
  2550. },
  2551. expectedResult: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2552. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2553. {Endpoint: "1.1.1.1:11", IsLocal: false},
  2554. },
  2555. makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): {
  2556. {Endpoint: "1.1.1.2:12", IsLocal: true},
  2557. },
  2558. },
  2559. expectedStaleEndpoints: []proxy.ServiceEndpoint{},
  2560. expectedStaleServiceNames: map[proxy.ServicePortName]bool{
  2561. makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): true,
  2562. },
  2563. expectedHealthchecks: map[types.NamespacedName]int{
  2564. makeNSN("ns1", "ep1"): 1,
  2565. },
  2566. }, {
  2567. // Case[11]: remove a subset
  2568. previousEndpoints: []*v1.Endpoints{
  2569. makeTestEndpoints("ns1", "ep1", multipleSubsets),
  2570. },
  2571. currentEndpoints: []*v1.Endpoints{
  2572. makeTestEndpoints("ns1", "ep1", namedPort),
  2573. },
  2574. oldEndpoints: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2575. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2576. {Endpoint: "1.1.1.1:11", IsLocal: false},
  2577. },
  2578. makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): {
  2579. {Endpoint: "1.1.1.2:12", IsLocal: false},
  2580. },
  2581. },
  2582. expectedResult: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2583. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2584. {Endpoint: "1.1.1.1:11", IsLocal: false},
  2585. },
  2586. },
  2587. expectedStaleEndpoints: []proxy.ServiceEndpoint{{
  2588. Endpoint: "1.1.1.2:12",
  2589. ServicePortName: makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP),
  2590. }},
  2591. expectedStaleServiceNames: map[proxy.ServicePortName]bool{},
  2592. expectedHealthchecks: map[types.NamespacedName]int{},
  2593. }, {
  2594. // Case[12]: rename a port
  2595. previousEndpoints: []*v1.Endpoints{
  2596. makeTestEndpoints("ns1", "ep1", namedPort),
  2597. },
  2598. currentEndpoints: []*v1.Endpoints{
  2599. makeTestEndpoints("ns1", "ep1", namedPortRenamed),
  2600. },
  2601. oldEndpoints: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2602. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2603. {Endpoint: "1.1.1.1:11", IsLocal: false},
  2604. },
  2605. },
  2606. expectedResult: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2607. makeServicePortName("ns1", "ep1", "p11-2", v1.ProtocolUDP): {
  2608. {Endpoint: "1.1.1.1:11", IsLocal: false},
  2609. },
  2610. },
  2611. expectedStaleEndpoints: []proxy.ServiceEndpoint{{
  2612. Endpoint: "1.1.1.1:11",
  2613. ServicePortName: makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP),
  2614. }},
  2615. expectedStaleServiceNames: map[proxy.ServicePortName]bool{
  2616. makeServicePortName("ns1", "ep1", "p11-2", v1.ProtocolUDP): true,
  2617. },
  2618. expectedHealthchecks: map[types.NamespacedName]int{},
  2619. }, {
  2620. // Case[13]: renumber a port
  2621. previousEndpoints: []*v1.Endpoints{
  2622. makeTestEndpoints("ns1", "ep1", namedPort),
  2623. },
  2624. currentEndpoints: []*v1.Endpoints{
  2625. makeTestEndpoints("ns1", "ep1", namedPortRenumbered),
  2626. },
  2627. oldEndpoints: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2628. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2629. {Endpoint: "1.1.1.1:11", IsLocal: false},
  2630. },
  2631. },
  2632. expectedResult: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2633. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2634. {Endpoint: "1.1.1.1:22", IsLocal: false},
  2635. },
  2636. },
  2637. expectedStaleEndpoints: []proxy.ServiceEndpoint{{
  2638. Endpoint: "1.1.1.1:11",
  2639. ServicePortName: makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP),
  2640. }},
  2641. expectedStaleServiceNames: map[proxy.ServicePortName]bool{},
  2642. expectedHealthchecks: map[types.NamespacedName]int{},
  2643. }, {
  2644. // Case[14]: complex add and remove
  2645. previousEndpoints: []*v1.Endpoints{
  2646. makeTestEndpoints("ns1", "ep1", complexBefore1),
  2647. makeTestEndpoints("ns2", "ep2", complexBefore2),
  2648. nil,
  2649. makeTestEndpoints("ns4", "ep4", complexBefore4),
  2650. },
  2651. currentEndpoints: []*v1.Endpoints{
  2652. makeTestEndpoints("ns1", "ep1", complexAfter1),
  2653. nil,
  2654. makeTestEndpoints("ns3", "ep3", complexAfter3),
  2655. makeTestEndpoints("ns4", "ep4", complexAfter4),
  2656. },
  2657. oldEndpoints: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2658. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2659. {Endpoint: "1.1.1.1:11", IsLocal: false},
  2660. },
  2661. makeServicePortName("ns2", "ep2", "p22", v1.ProtocolUDP): {
  2662. {Endpoint: "2.2.2.2:22", IsLocal: true},
  2663. {Endpoint: "2.2.2.22:22", IsLocal: true},
  2664. },
  2665. makeServicePortName("ns2", "ep2", "p23", v1.ProtocolUDP): {
  2666. {Endpoint: "2.2.2.3:23", IsLocal: true},
  2667. },
  2668. makeServicePortName("ns4", "ep4", "p44", v1.ProtocolUDP): {
  2669. {Endpoint: "4.4.4.4:44", IsLocal: true},
  2670. {Endpoint: "4.4.4.5:44", IsLocal: true},
  2671. },
  2672. makeServicePortName("ns4", "ep4", "p45", v1.ProtocolUDP): {
  2673. {Endpoint: "4.4.4.6:45", IsLocal: true},
  2674. },
  2675. },
  2676. expectedResult: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2677. makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
  2678. {Endpoint: "1.1.1.1:11", IsLocal: false},
  2679. {Endpoint: "1.1.1.11:11", IsLocal: false},
  2680. },
  2681. makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): {
  2682. {Endpoint: "1.1.1.2:12", IsLocal: false},
  2683. },
  2684. makeServicePortName("ns1", "ep1", "p122", v1.ProtocolUDP): {
  2685. {Endpoint: "1.1.1.2:122", IsLocal: false},
  2686. },
  2687. makeServicePortName("ns3", "ep3", "p33", v1.ProtocolUDP): {
  2688. {Endpoint: "3.3.3.3:33", IsLocal: false},
  2689. },
  2690. makeServicePortName("ns4", "ep4", "p44", v1.ProtocolUDP): {
  2691. {Endpoint: "4.4.4.4:44", IsLocal: true},
  2692. },
  2693. },
  2694. expectedStaleEndpoints: []proxy.ServiceEndpoint{{
  2695. Endpoint: "2.2.2.2:22",
  2696. ServicePortName: makeServicePortName("ns2", "ep2", "p22", v1.ProtocolUDP),
  2697. }, {
  2698. Endpoint: "2.2.2.22:22",
  2699. ServicePortName: makeServicePortName("ns2", "ep2", "p22", v1.ProtocolUDP),
  2700. }, {
  2701. Endpoint: "2.2.2.3:23",
  2702. ServicePortName: makeServicePortName("ns2", "ep2", "p23", v1.ProtocolUDP),
  2703. }, {
  2704. Endpoint: "4.4.4.5:44",
  2705. ServicePortName: makeServicePortName("ns4", "ep4", "p44", v1.ProtocolUDP),
  2706. }, {
  2707. Endpoint: "4.4.4.6:45",
  2708. ServicePortName: makeServicePortName("ns4", "ep4", "p45", v1.ProtocolUDP),
  2709. }},
  2710. expectedStaleServiceNames: map[proxy.ServicePortName]bool{
  2711. makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): true,
  2712. makeServicePortName("ns1", "ep1", "p122", v1.ProtocolUDP): true,
  2713. makeServicePortName("ns3", "ep3", "p33", v1.ProtocolUDP): true,
  2714. },
  2715. expectedHealthchecks: map[types.NamespacedName]int{
  2716. makeNSN("ns4", "ep4"): 1,
  2717. },
  2718. }, {
  2719. // Case[15]: change from 0 endpoint address to 1 unnamed port
  2720. previousEndpoints: []*v1.Endpoints{
  2721. makeTestEndpoints("ns1", "ep1", emptyEndpoint),
  2722. },
  2723. currentEndpoints: []*v1.Endpoints{
  2724. makeTestEndpoints("ns1", "ep1", unnamedPort),
  2725. },
  2726. oldEndpoints: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{},
  2727. expectedResult: map[proxy.ServicePortName][]*proxy.BaseEndpointInfo{
  2728. makeServicePortName("ns1", "ep1", "", v1.ProtocolUDP): {
  2729. {Endpoint: "1.1.1.1:11", IsLocal: false},
  2730. },
  2731. },
  2732. expectedStaleEndpoints: []proxy.ServiceEndpoint{},
  2733. expectedStaleServiceNames: map[proxy.ServicePortName]bool{
  2734. makeServicePortName("ns1", "ep1", "", v1.ProtocolUDP): true,
  2735. },
  2736. expectedHealthchecks: map[types.NamespacedName]int{},
  2737. },
  2738. }
  2739. for tci, tc := range testCases {
  2740. ipt := iptablestest.NewFake()
  2741. ipvs := ipvstest.NewFake()
  2742. ipset := ipsettest.NewFake(testIPSetVersion)
  2743. fp := NewFakeProxier(ipt, ipvs, ipset, nil, nil, false)
  2744. fp.hostname = nodeName
  2745. // First check that after adding all previous versions of endpoints,
  2746. // the fp.oldEndpoints is as we expect.
  2747. for i := range tc.previousEndpoints {
  2748. if tc.previousEndpoints[i] != nil {
  2749. fp.OnEndpointsAdd(tc.previousEndpoints[i])
  2750. }
  2751. }
  2752. fp.endpointsMap.Update(fp.endpointsChanges)
  2753. compareEndpointsMaps(t, tci, fp.endpointsMap, tc.oldEndpoints)
  2754. // Now let's call appropriate handlers to get to state we want to be.
  2755. if len(tc.previousEndpoints) != len(tc.currentEndpoints) {
  2756. t.Fatalf("[%d] different lengths of previous and current endpoints", tci)
  2757. continue
  2758. }
  2759. for i := range tc.previousEndpoints {
  2760. prev, curr := tc.previousEndpoints[i], tc.currentEndpoints[i]
  2761. switch {
  2762. case prev == nil:
  2763. fp.OnEndpointsAdd(curr)
  2764. case curr == nil:
  2765. fp.OnEndpointsDelete(prev)
  2766. default:
  2767. fp.OnEndpointsUpdate(prev, curr)
  2768. }
  2769. }
  2770. result := fp.endpointsMap.Update(fp.endpointsChanges)
  2771. newMap := fp.endpointsMap
  2772. compareEndpointsMaps(t, tci, newMap, tc.expectedResult)
  2773. if len(result.StaleEndpoints) != len(tc.expectedStaleEndpoints) {
  2774. t.Errorf("[%d] expected %d staleEndpoints, got %d: %v", tci, len(tc.expectedStaleEndpoints), len(result.StaleEndpoints), result.StaleEndpoints)
  2775. }
  2776. for _, x := range tc.expectedStaleEndpoints {
  2777. found := false
  2778. for _, stale := range result.StaleEndpoints {
  2779. if stale == x {
  2780. found = true
  2781. break
  2782. }
  2783. }
  2784. if !found {
  2785. t.Errorf("[%d] expected staleEndpoints[%v], but didn't find it: %v", tci, x, result.StaleEndpoints)
  2786. }
  2787. }
  2788. if len(result.StaleServiceNames) != len(tc.expectedStaleServiceNames) {
  2789. t.Errorf("[%d] expected %d staleServiceNames, got %d: %v", tci, len(tc.expectedStaleServiceNames), len(result.StaleServiceNames), result.StaleServiceNames)
  2790. }
  2791. for svcName := range tc.expectedStaleServiceNames {
  2792. found := false
  2793. for _, stale := range result.StaleServiceNames {
  2794. if stale == svcName {
  2795. found = true
  2796. break
  2797. }
  2798. }
  2799. if !found {
  2800. t.Errorf("[%d] expected staleServiceNames[%v], but didn't find it: %v", tci, svcName, result.StaleServiceNames)
  2801. }
  2802. }
  2803. if !reflect.DeepEqual(result.HCEndpointsLocalIPSize, tc.expectedHealthchecks) {
  2804. t.Errorf("[%d] expected healthchecks %v, got %v", tci, tc.expectedHealthchecks, result.HCEndpointsLocalIPSize)
  2805. }
  2806. }
  2807. }
  2808. func compareEndpointsMaps(t *testing.T, tci int, newMap proxy.EndpointsMap, expected map[proxy.ServicePortName][]*proxy.BaseEndpointInfo) {
  2809. if len(newMap) != len(expected) {
  2810. t.Errorf("[%d] expected %d results, got %d: %v", tci, len(expected), len(newMap), newMap)
  2811. }
  2812. for x := range expected {
  2813. if len(newMap[x]) != len(expected[x]) {
  2814. t.Errorf("[%d] expected %d endpoints for %v, got %d", tci, len(expected[x]), x, len(newMap[x]))
  2815. } else {
  2816. for i := range expected[x] {
  2817. newEp, ok := newMap[x][i].(*proxy.BaseEndpointInfo)
  2818. if !ok {
  2819. t.Errorf("Failed to cast proxy.BaseEndpointInfo")
  2820. continue
  2821. }
  2822. if !reflect.DeepEqual(*newEp, *(expected[x][i])) {
  2823. t.Errorf("[%d] expected new[%v][%d] to be %v, got %v", tci, x, i, expected[x][i], newEp)
  2824. }
  2825. }
  2826. }
  2827. }
  2828. }
  2829. func Test_syncService(t *testing.T) {
  2830. testCases := []struct {
  2831. oldVirtualServer *utilipvs.VirtualServer
  2832. svcName string
  2833. newVirtualServer *utilipvs.VirtualServer
  2834. bindAddr bool
  2835. }{
  2836. {
  2837. // case 0, old virtual server is same as new virtual server
  2838. oldVirtualServer: &utilipvs.VirtualServer{
  2839. Address: net.ParseIP("1.2.3.4"),
  2840. Protocol: string(v1.ProtocolTCP),
  2841. Port: 80,
  2842. Scheduler: "rr",
  2843. Flags: utilipvs.FlagHashed,
  2844. },
  2845. svcName: "foo",
  2846. newVirtualServer: &utilipvs.VirtualServer{
  2847. Address: net.ParseIP("1.2.3.4"),
  2848. Protocol: string(v1.ProtocolTCP),
  2849. Port: 80,
  2850. Scheduler: "rr",
  2851. Flags: utilipvs.FlagHashed,
  2852. },
  2853. bindAddr: false,
  2854. },
  2855. {
  2856. // case 1, old virtual server is different from new virtual server
  2857. oldVirtualServer: &utilipvs.VirtualServer{
  2858. Address: net.ParseIP("1.2.3.4"),
  2859. Protocol: string(v1.ProtocolTCP),
  2860. Port: 8080,
  2861. Scheduler: "rr",
  2862. Flags: utilipvs.FlagHashed,
  2863. },
  2864. svcName: "bar",
  2865. newVirtualServer: &utilipvs.VirtualServer{
  2866. Address: net.ParseIP("1.2.3.4"),
  2867. Protocol: string(v1.ProtocolTCP),
  2868. Port: 8080,
  2869. Scheduler: "rr",
  2870. Flags: utilipvs.FlagPersistent,
  2871. },
  2872. bindAddr: false,
  2873. },
  2874. {
  2875. // case 2, old virtual server is different from new virtual server
  2876. oldVirtualServer: &utilipvs.VirtualServer{
  2877. Address: net.ParseIP("1.2.3.4"),
  2878. Protocol: string(v1.ProtocolTCP),
  2879. Port: 8080,
  2880. Scheduler: "rr",
  2881. Flags: utilipvs.FlagHashed,
  2882. },
  2883. svcName: "bar",
  2884. newVirtualServer: &utilipvs.VirtualServer{
  2885. Address: net.ParseIP("1.2.3.4"),
  2886. Protocol: string(v1.ProtocolTCP),
  2887. Port: 8080,
  2888. Scheduler: "wlc",
  2889. Flags: utilipvs.FlagHashed,
  2890. },
  2891. bindAddr: false,
  2892. },
  2893. {
  2894. // case 3, old virtual server is nil, and create new virtual server
  2895. oldVirtualServer: nil,
  2896. svcName: "baz",
  2897. newVirtualServer: &utilipvs.VirtualServer{
  2898. Address: net.ParseIP("1.2.3.4"),
  2899. Protocol: string(v1.ProtocolUDP),
  2900. Port: 53,
  2901. Scheduler: "rr",
  2902. Flags: utilipvs.FlagHashed,
  2903. },
  2904. bindAddr: true,
  2905. },
  2906. {
  2907. // case 4, SCTP, old virtual server is same as new virtual server
  2908. oldVirtualServer: &utilipvs.VirtualServer{
  2909. Address: net.ParseIP("1.2.3.4"),
  2910. Protocol: string(v1.ProtocolSCTP),
  2911. Port: 80,
  2912. Scheduler: "rr",
  2913. Flags: utilipvs.FlagHashed,
  2914. },
  2915. svcName: "foo",
  2916. newVirtualServer: &utilipvs.VirtualServer{
  2917. Address: net.ParseIP("1.2.3.4"),
  2918. Protocol: string(v1.ProtocolSCTP),
  2919. Port: 80,
  2920. Scheduler: "rr",
  2921. Flags: utilipvs.FlagHashed,
  2922. },
  2923. bindAddr: false,
  2924. },
  2925. {
  2926. // case 5, old virtual server is different from new virtual server
  2927. oldVirtualServer: &utilipvs.VirtualServer{
  2928. Address: net.ParseIP("1.2.3.4"),
  2929. Protocol: string(v1.ProtocolSCTP),
  2930. Port: 8080,
  2931. Scheduler: "rr",
  2932. Flags: utilipvs.FlagHashed,
  2933. },
  2934. svcName: "bar",
  2935. newVirtualServer: &utilipvs.VirtualServer{
  2936. Address: net.ParseIP("1.2.3.4"),
  2937. Protocol: string(v1.ProtocolSCTP),
  2938. Port: 8080,
  2939. Scheduler: "rr",
  2940. Flags: utilipvs.FlagPersistent,
  2941. },
  2942. bindAddr: false,
  2943. },
  2944. {
  2945. // case 6, old virtual server is different from new virtual server
  2946. oldVirtualServer: &utilipvs.VirtualServer{
  2947. Address: net.ParseIP("1.2.3.4"),
  2948. Protocol: string(v1.ProtocolSCTP),
  2949. Port: 8080,
  2950. Scheduler: "rr",
  2951. Flags: utilipvs.FlagHashed,
  2952. },
  2953. svcName: "bar",
  2954. newVirtualServer: &utilipvs.VirtualServer{
  2955. Address: net.ParseIP("1.2.3.4"),
  2956. Protocol: string(v1.ProtocolSCTP),
  2957. Port: 8080,
  2958. Scheduler: "wlc",
  2959. Flags: utilipvs.FlagHashed,
  2960. },
  2961. bindAddr: false,
  2962. },
  2963. {
  2964. // case 7, old virtual server is nil, and create new virtual server
  2965. oldVirtualServer: nil,
  2966. svcName: "baz",
  2967. newVirtualServer: &utilipvs.VirtualServer{
  2968. Address: net.ParseIP("1.2.3.4"),
  2969. Protocol: string(v1.ProtocolSCTP),
  2970. Port: 53,
  2971. Scheduler: "rr",
  2972. Flags: utilipvs.FlagHashed,
  2973. },
  2974. bindAddr: true,
  2975. },
  2976. }
  2977. for i := range testCases {
  2978. ipt := iptablestest.NewFake()
  2979. ipvs := ipvstest.NewFake()
  2980. ipset := ipsettest.NewFake(testIPSetVersion)
  2981. proxier := NewFakeProxier(ipt, ipvs, ipset, nil, nil, false)
  2982. proxier.netlinkHandle.EnsureDummyDevice(DefaultDummyDevice)
  2983. if testCases[i].oldVirtualServer != nil {
  2984. if err := proxier.ipvs.AddVirtualServer(testCases[i].oldVirtualServer); err != nil {
  2985. t.Errorf("Case [%d], unexpected add IPVS virtual server error: %v", i, err)
  2986. }
  2987. }
  2988. if err := proxier.syncService(testCases[i].svcName, testCases[i].newVirtualServer, testCases[i].bindAddr); err != nil {
  2989. t.Errorf("Case [%d], unexpected sync IPVS virtual server error: %v", i, err)
  2990. }
  2991. // check
  2992. list, err := proxier.ipvs.GetVirtualServers()
  2993. if err != nil {
  2994. t.Errorf("Case [%d], unexpected list IPVS virtual server error: %v", i, err)
  2995. }
  2996. if len(list) != 1 {
  2997. t.Errorf("Case [%d], expect %d virtual servers, got %d", i, 1, len(list))
  2998. continue
  2999. }
  3000. if !list[0].Equal(testCases[i].newVirtualServer) {
  3001. t.Errorf("Case [%d], unexpected mismatch, expect: %#v, got: %#v", i, testCases[i].newVirtualServer, list[0])
  3002. }
  3003. }
  3004. }
  3005. func buildFakeProxier() (*iptablestest.FakeIPTables, *Proxier) {
  3006. ipt := iptablestest.NewFake()
  3007. ipvs := ipvstest.NewFake()
  3008. ipset := ipsettest.NewFake(testIPSetVersion)
  3009. return ipt, NewFakeProxier(ipt, ipvs, ipset, nil, nil, false)
  3010. }
  3011. func hasJump(rules []iptablestest.Rule, destChain, ipSet string) bool {
  3012. for _, r := range rules {
  3013. if r[iptablestest.Jump] == destChain {
  3014. if ipSet == "" {
  3015. return true
  3016. }
  3017. if strings.Contains(r[iptablestest.MatchSet], ipSet) {
  3018. return true
  3019. }
  3020. }
  3021. }
  3022. return false
  3023. }
  3024. func hasMasqRandomFully(rules []iptablestest.Rule) bool {
  3025. for _, r := range rules {
  3026. if r[iptablestest.Masquerade] == "--random-fully" {
  3027. return true
  3028. }
  3029. }
  3030. return false
  3031. }
  3032. // checkIptabless to check expected iptables chain and rules
  3033. func checkIptables(t *testing.T, ipt *iptablestest.FakeIPTables, epIpt netlinktest.ExpectedIptablesChain) {
  3034. for epChain, epRules := range epIpt {
  3035. rules := ipt.GetRules(epChain)
  3036. for _, epRule := range epRules {
  3037. if !hasJump(rules, epRule.JumpChain, epRule.MatchSet) {
  3038. t.Errorf("Didn't find jump from chain %v match set %v to %v", epChain, epRule.MatchSet, epRule.JumpChain)
  3039. }
  3040. }
  3041. }
  3042. }
  3043. // checkIPSet to check expected ipset and entries
  3044. func checkIPSet(t *testing.T, fp *Proxier, ipSet netlinktest.ExpectedIPSet) {
  3045. for set, entries := range ipSet {
  3046. ents, err := fp.ipset.ListEntries(set)
  3047. if err != nil || len(ents) != len(entries) {
  3048. t.Errorf("Check ipset entries failed for ipset: %q, expect %d, got %d", set, len(entries), len(ents))
  3049. continue
  3050. }
  3051. expectedEntries := []string{}
  3052. for _, entry := range entries {
  3053. expectedEntries = append(expectedEntries, entry.String())
  3054. }
  3055. sort.Strings(ents)
  3056. sort.Strings(expectedEntries)
  3057. if !reflect.DeepEqual(ents, expectedEntries) {
  3058. t.Errorf("Check ipset entries failed for ipset: %q", set)
  3059. }
  3060. }
  3061. }
  3062. // checkIPVS to check expected ipvs service and destination
  3063. func checkIPVS(t *testing.T, fp *Proxier, vs *netlinktest.ExpectedVirtualServer) {
  3064. services, err := fp.ipvs.GetVirtualServers()
  3065. if err != nil {
  3066. t.Errorf("Failed to get ipvs services, err: %v", err)
  3067. }
  3068. if len(services) != vs.VSNum {
  3069. t.Errorf("Expect %d ipvs services, got %d", vs.VSNum, len(services))
  3070. }
  3071. for _, svc := range services {
  3072. if svc.Address.String() == vs.IP && svc.Port == vs.Port && svc.Protocol == vs.Protocol {
  3073. destinations, _ := fp.ipvs.GetRealServers(svc)
  3074. if len(destinations) != len(vs.RS) {
  3075. t.Errorf("Expected %d destinations, got %d destinations", len(vs.RS), len(destinations))
  3076. }
  3077. if len(vs.RS) == 1 {
  3078. if destinations[0].Address.String() != vs.RS[0].IP || destinations[0].Port != vs.RS[0].Port {
  3079. t.Errorf("Unexpected mismatch destinations")
  3080. }
  3081. }
  3082. }
  3083. }
  3084. }
  3085. func TestCleanLegacyService(t *testing.T) {
  3086. ipt := iptablestest.NewFake()
  3087. ipvs := ipvstest.NewFake()
  3088. ipset := ipsettest.NewFake(testIPSetVersion)
  3089. fp := NewFakeProxier(ipt, ipvs, ipset, nil, parseExcludedCIDRs([]string{"3.3.3.0/24", "4.4.4.0/24"}), false)
  3090. // All ipvs services that were processed in the latest sync loop.
  3091. activeServices := map[string]bool{"ipvs0": true, "ipvs1": true}
  3092. // All ipvs services in the system.
  3093. currentServices := map[string]*utilipvs.VirtualServer{
  3094. // Created by kube-proxy.
  3095. "ipvs0": {
  3096. Address: net.ParseIP("1.1.1.1"),
  3097. Protocol: string(v1.ProtocolUDP),
  3098. Port: 53,
  3099. Scheduler: "rr",
  3100. Flags: utilipvs.FlagHashed,
  3101. },
  3102. // Created by kube-proxy.
  3103. "ipvs1": {
  3104. Address: net.ParseIP("2.2.2.2"),
  3105. Protocol: string(v1.ProtocolUDP),
  3106. Port: 54,
  3107. Scheduler: "rr",
  3108. Flags: utilipvs.FlagHashed,
  3109. },
  3110. // Created by an external party.
  3111. "ipvs2": {
  3112. Address: net.ParseIP("3.3.3.3"),
  3113. Protocol: string(v1.ProtocolUDP),
  3114. Port: 55,
  3115. Scheduler: "rr",
  3116. Flags: utilipvs.FlagHashed,
  3117. },
  3118. // Created by an external party.
  3119. "ipvs3": {
  3120. Address: net.ParseIP("4.4.4.4"),
  3121. Protocol: string(v1.ProtocolUDP),
  3122. Port: 56,
  3123. Scheduler: "rr",
  3124. Flags: utilipvs.FlagHashed,
  3125. },
  3126. // Created by an external party.
  3127. "ipvs4": {
  3128. Address: net.ParseIP("5.5.5.5"),
  3129. Protocol: string(v1.ProtocolUDP),
  3130. Port: 57,
  3131. Scheduler: "rr",
  3132. Flags: utilipvs.FlagHashed,
  3133. },
  3134. // Created by kube-proxy, but now stale.
  3135. "ipvs5": {
  3136. Address: net.ParseIP("6.6.6.6"),
  3137. Protocol: string(v1.ProtocolUDP),
  3138. Port: 58,
  3139. Scheduler: "rr",
  3140. Flags: utilipvs.FlagHashed,
  3141. },
  3142. }
  3143. for v := range currentServices {
  3144. fp.ipvs.AddVirtualServer(currentServices[v])
  3145. }
  3146. fp.netlinkHandle.EnsureDummyDevice(DefaultDummyDevice)
  3147. activeBindAddrs := map[string]bool{"1.1.1.1": true, "2.2.2.2": true, "3.3.3.3": true, "4.4.4.4": true}
  3148. // This is ipv4-only so ipv6 addresses should be ignored
  3149. currentBindAddrs := []string{"1.1.1.1", "2.2.2.2", "3.3.3.3", "4.4.4.4", "5.5.5.5", "6.6.6.6", "fd80::1:2:3", "fd80::1:2:4"}
  3150. for i := range currentBindAddrs {
  3151. fp.netlinkHandle.EnsureAddressBind(currentBindAddrs[i], DefaultDummyDevice)
  3152. }
  3153. fp.cleanLegacyService(activeServices, currentServices, map[string]bool{"5.5.5.5": true, "6.6.6.6": true})
  3154. // ipvs4 and ipvs5 should have been cleaned.
  3155. remainingVirtualServers, _ := fp.ipvs.GetVirtualServers()
  3156. if len(remainingVirtualServers) != 4 {
  3157. t.Errorf("Expected number of remaining IPVS services after cleanup to be %v. Got %v", 4, len(remainingVirtualServers))
  3158. }
  3159. for _, vs := range remainingVirtualServers {
  3160. // Checking that ipvs4 and ipvs5 were removed.
  3161. if vs.Port == 57 {
  3162. t.Errorf("Expected ipvs4 to be removed after cleanup. It still remains")
  3163. }
  3164. if vs.Port == 58 {
  3165. t.Errorf("Expected ipvs5 to be removed after cleanup. It still remains")
  3166. }
  3167. }
  3168. // Addresses 5.5.5.5 and 6.6.6.6 should not be bound any more, but the ipv6 addresses should remain
  3169. remainingAddrs, _ := fp.netlinkHandle.ListBindAddress(DefaultDummyDevice)
  3170. if len(remainingAddrs) != 6 {
  3171. t.Errorf("Expected number of remaining bound addrs after cleanup to be %v. Got %v", 6, len(remainingAddrs))
  3172. }
  3173. // check that address "1.1.1.1", "2.2.2.2", "3.3.3.3", "4.4.4.4" are bound, ignore ipv6 addresses
  3174. remainingAddrsMap := make(map[string]bool)
  3175. for _, a := range remainingAddrs {
  3176. if net.ParseIP(a).To4() == nil {
  3177. continue
  3178. }
  3179. remainingAddrsMap[a] = true
  3180. }
  3181. if !reflect.DeepEqual(activeBindAddrs, remainingAddrsMap) {
  3182. t.Errorf("Expected remainingAddrsMap %v, got %v", activeBindAddrs, remainingAddrsMap)
  3183. }
  3184. }
  3185. func TestCleanLegacyServiceWithRealServers(t *testing.T) {
  3186. ipt := iptablestest.NewFake()
  3187. ipvs := ipvstest.NewFake()
  3188. ipset := ipsettest.NewFake(testIPSetVersion)
  3189. fp := NewFakeProxier(ipt, ipvs, ipset, nil, nil, false)
  3190. // all deleted expect ipvs2
  3191. activeServices := map[string]bool{"ipvs2": true}
  3192. // All ipvs services in the system.
  3193. currentServices := map[string]*utilipvs.VirtualServer{
  3194. "ipvs0": { // deleted with real servers
  3195. Address: net.ParseIP("1.1.1.1"),
  3196. Protocol: string(v1.ProtocolUDP),
  3197. Port: 53,
  3198. Scheduler: "rr",
  3199. Flags: utilipvs.FlagHashed,
  3200. },
  3201. "ipvs1": { // deleted no real server
  3202. Address: net.ParseIP("2.2.2.2"),
  3203. Protocol: string(v1.ProtocolUDP),
  3204. Port: 54,
  3205. Scheduler: "rr",
  3206. Flags: utilipvs.FlagHashed,
  3207. },
  3208. "ipvs2": { // not deleted
  3209. Address: net.ParseIP("3.3.3.3"),
  3210. Protocol: string(v1.ProtocolUDP),
  3211. Port: 54,
  3212. Scheduler: "rr",
  3213. Flags: utilipvs.FlagHashed,
  3214. },
  3215. }
  3216. // "ipvs0" has a real server, but it should still be deleted since the Service is deleted
  3217. realServers := map[*utilipvs.VirtualServer]*utilipvs.RealServer{
  3218. {
  3219. Address: net.ParseIP("1.1.1.1"),
  3220. Protocol: string(v1.ProtocolUDP),
  3221. Port: 53,
  3222. Scheduler: "rr",
  3223. Flags: utilipvs.FlagHashed,
  3224. }: {
  3225. Address: net.ParseIP("10.180.0.1"),
  3226. Port: uint16(53),
  3227. Weight: 1,
  3228. },
  3229. }
  3230. for v := range currentServices {
  3231. fp.ipvs.AddVirtualServer(currentServices[v])
  3232. }
  3233. for v, r := range realServers {
  3234. fp.ipvs.AddRealServer(v, r)
  3235. }
  3236. fp.netlinkHandle.EnsureDummyDevice(DefaultDummyDevice)
  3237. activeBindAddrs := map[string]bool{"3.3.3.3": true}
  3238. currentBindAddrs := []string{"1.1.1.1", "2.2.2.2", "3.3.3.3"}
  3239. for i := range currentBindAddrs {
  3240. fp.netlinkHandle.EnsureAddressBind(currentBindAddrs[i], DefaultDummyDevice)
  3241. }
  3242. fp.cleanLegacyService(activeServices, currentServices, map[string]bool{"1.1.1.1": true, "2.2.2.2": true})
  3243. remainingVirtualServers, _ := fp.ipvs.GetVirtualServers()
  3244. if len(remainingVirtualServers) != 1 {
  3245. t.Errorf("Expected number of remaining IPVS services after cleanup to be %v. Got %v", 1, len(remainingVirtualServers))
  3246. }
  3247. if remainingVirtualServers[0] != currentServices["ipvs2"] {
  3248. t.Logf("actual virtual server: %v", remainingVirtualServers[0])
  3249. t.Logf("expected virtual server: %v", currentServices["ipvs0"])
  3250. t.Errorf("unexpected IPVS service")
  3251. }
  3252. remainingAddrs, _ := fp.netlinkHandle.ListBindAddress(DefaultDummyDevice)
  3253. if len(remainingAddrs) != 1 {
  3254. t.Errorf("Expected number of remaining bound addrs after cleanup to be %v. Got %v", 1, len(remainingAddrs))
  3255. }
  3256. // check that address is "3.3.3.3"
  3257. remainingAddrsMap := make(map[string]bool)
  3258. for _, a := range remainingAddrs {
  3259. if net.ParseIP(a).To4() == nil {
  3260. continue
  3261. }
  3262. remainingAddrsMap[a] = true
  3263. }
  3264. if !reflect.DeepEqual(activeBindAddrs, remainingAddrsMap) {
  3265. t.Errorf("Expected remainingAddrsMap %v, got %v", activeBindAddrs, remainingAddrsMap)
  3266. }
  3267. }
  3268. func TestCleanLegacyRealServersExcludeCIDRs(t *testing.T) {
  3269. ipt := iptablestest.NewFake()
  3270. ipvs := ipvstest.NewFake()
  3271. ipset := ipsettest.NewFake(testIPSetVersion)
  3272. gtm := NewGracefulTerminationManager(ipvs)
  3273. fp := NewFakeProxier(ipt, ipvs, ipset, nil, parseExcludedCIDRs([]string{"4.4.4.4/32"}), false)
  3274. fp.gracefuldeleteManager = gtm
  3275. vs := &utilipvs.VirtualServer{
  3276. Address: net.ParseIP("4.4.4.4"),
  3277. Protocol: string(v1.ProtocolUDP),
  3278. Port: 56,
  3279. Scheduler: "rr",
  3280. Flags: utilipvs.FlagHashed,
  3281. }
  3282. fp.ipvs.AddVirtualServer(vs)
  3283. rss := []*utilipvs.RealServer{
  3284. {
  3285. Address: net.ParseIP("10.10.10.10"),
  3286. Port: 56,
  3287. ActiveConn: 0,
  3288. InactiveConn: 0,
  3289. },
  3290. {
  3291. Address: net.ParseIP("11.11.11.11"),
  3292. Port: 56,
  3293. ActiveConn: 0,
  3294. InactiveConn: 0,
  3295. },
  3296. }
  3297. for _, rs := range rss {
  3298. fp.ipvs.AddRealServer(vs, rs)
  3299. }
  3300. fp.netlinkHandle.EnsureDummyDevice(DefaultDummyDevice)
  3301. fp.netlinkHandle.EnsureAddressBind("4.4.4.4", DefaultDummyDevice)
  3302. fp.cleanLegacyService(
  3303. map[string]bool{},
  3304. map[string]*utilipvs.VirtualServer{"ipvs0": vs},
  3305. map[string]bool{"4.4.4.4": true},
  3306. )
  3307. fp.gracefuldeleteManager.tryDeleteRs()
  3308. remainingRealServers, _ := fp.ipvs.GetRealServers(vs)
  3309. if len(remainingRealServers) != 2 {
  3310. t.Errorf("Expected number of remaining IPVS real servers after cleanup should be %v. Got %v", 2, len(remainingRealServers))
  3311. }
  3312. }
  3313. func TestCleanLegacyService6(t *testing.T) {
  3314. ipt := iptablestest.NewFake()
  3315. ipvs := ipvstest.NewFake()
  3316. ipset := ipsettest.NewFake(testIPSetVersion)
  3317. fp := NewFakeProxier(ipt, ipvs, ipset, nil, parseExcludedCIDRs([]string{"3000::/64", "4000::/64"}), false)
  3318. fp.nodeIP = net.ParseIP("::1")
  3319. // All ipvs services that were processed in the latest sync loop.
  3320. activeServices := map[string]bool{"ipvs0": true, "ipvs1": true}
  3321. // All ipvs services in the system.
  3322. currentServices := map[string]*utilipvs.VirtualServer{
  3323. // Created by kube-proxy.
  3324. "ipvs0": {
  3325. Address: net.ParseIP("1000::1"),
  3326. Protocol: string(v1.ProtocolUDP),
  3327. Port: 53,
  3328. Scheduler: "rr",
  3329. Flags: utilipvs.FlagHashed,
  3330. },
  3331. // Created by kube-proxy.
  3332. "ipvs1": {
  3333. Address: net.ParseIP("1000::2"),
  3334. Protocol: string(v1.ProtocolUDP),
  3335. Port: 54,
  3336. Scheduler: "rr",
  3337. Flags: utilipvs.FlagHashed,
  3338. },
  3339. // Created by an external party.
  3340. "ipvs2": {
  3341. Address: net.ParseIP("3000::1"),
  3342. Protocol: string(v1.ProtocolUDP),
  3343. Port: 55,
  3344. Scheduler: "rr",
  3345. Flags: utilipvs.FlagHashed,
  3346. },
  3347. // Created by an external party.
  3348. "ipvs3": {
  3349. Address: net.ParseIP("4000::1"),
  3350. Protocol: string(v1.ProtocolUDP),
  3351. Port: 56,
  3352. Scheduler: "rr",
  3353. Flags: utilipvs.FlagHashed,
  3354. },
  3355. // Created by an external party.
  3356. "ipvs4": {
  3357. Address: net.ParseIP("5000::1"),
  3358. Protocol: string(v1.ProtocolUDP),
  3359. Port: 57,
  3360. Scheduler: "rr",
  3361. Flags: utilipvs.FlagHashed,
  3362. },
  3363. // Created by kube-proxy, but now stale.
  3364. "ipvs5": {
  3365. Address: net.ParseIP("1000::6"),
  3366. Protocol: string(v1.ProtocolUDP),
  3367. Port: 58,
  3368. Scheduler: "rr",
  3369. Flags: utilipvs.FlagHashed,
  3370. },
  3371. }
  3372. for v := range currentServices {
  3373. fp.ipvs.AddVirtualServer(currentServices[v])
  3374. }
  3375. fp.netlinkHandle.EnsureDummyDevice(DefaultDummyDevice)
  3376. activeBindAddrs := map[string]bool{"1000::1": true, "1000::2": true, "3000::1": true, "4000::1": true}
  3377. // This is ipv6-only so ipv4 addresses should be ignored
  3378. currentBindAddrs := []string{"1000::1", "1000::2", "3000::1", "4000::1", "5000::1", "1000::6", "1.1.1.1", "2.2.2.2"}
  3379. for i := range currentBindAddrs {
  3380. fp.netlinkHandle.EnsureAddressBind(currentBindAddrs[i], DefaultDummyDevice)
  3381. }
  3382. fp.cleanLegacyService(activeServices, currentServices, map[string]bool{"5000::1": true, "1000::6": true})
  3383. // ipvs4 and ipvs5 should have been cleaned.
  3384. remainingVirtualServers, _ := fp.ipvs.GetVirtualServers()
  3385. if len(remainingVirtualServers) != 4 {
  3386. t.Errorf("Expected number of remaining IPVS services after cleanup to be %v. Got %v", 4, len(remainingVirtualServers))
  3387. }
  3388. for _, vs := range remainingVirtualServers {
  3389. // Checking that ipvs4 and ipvs5 were removed.
  3390. if vs.Port == 57 {
  3391. t.Errorf("Expected ipvs4 to be removed after cleanup. It still remains")
  3392. }
  3393. if vs.Port == 58 {
  3394. t.Errorf("Expected ipvs5 to be removed after cleanup. It still remains")
  3395. }
  3396. }
  3397. // Addresses 5000::1 and 1000::6 should not be bound any more, but the ipv4 addresses should remain
  3398. remainingAddrs, _ := fp.netlinkHandle.ListBindAddress(DefaultDummyDevice)
  3399. if len(remainingAddrs) != 6 {
  3400. t.Errorf("Expected number of remaining bound addrs after cleanup to be %v. Got %v", 6, len(remainingAddrs))
  3401. }
  3402. // check that address "1000::1", "1000::2", "3000::1", "4000::1" are still bound, ignore ipv4 addresses
  3403. remainingAddrsMap := make(map[string]bool)
  3404. for _, a := range remainingAddrs {
  3405. if net.ParseIP(a).To4() != nil {
  3406. continue
  3407. }
  3408. remainingAddrsMap[a] = true
  3409. }
  3410. if !reflect.DeepEqual(activeBindAddrs, remainingAddrsMap) {
  3411. t.Errorf("Expected remainingAddrsMap %v, got %v", activeBindAddrs, remainingAddrsMap)
  3412. }
  3413. }
  3414. func TestMultiPortServiceBindAddr(t *testing.T) {
  3415. ipt := iptablestest.NewFake()
  3416. ipvs := ipvstest.NewFake()
  3417. ipset := ipsettest.NewFake(testIPSetVersion)
  3418. fp := NewFakeProxier(ipt, ipvs, ipset, nil, nil, false)
  3419. service1 := makeTestService("ns1", "svc1", func(svc *v1.Service) {
  3420. svc.Spec.Type = v1.ServiceTypeClusterIP
  3421. svc.Spec.ClusterIP = "172.16.55.4"
  3422. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "port1", "TCP", 1234, 0, 0)
  3423. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "port2", "TCP", 1235, 0, 0)
  3424. })
  3425. service2 := makeTestService("ns1", "svc1", func(svc *v1.Service) {
  3426. svc.Spec.Type = v1.ServiceTypeClusterIP
  3427. svc.Spec.ClusterIP = "172.16.55.4"
  3428. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "port1", "TCP", 1234, 0, 0)
  3429. })
  3430. service3 := makeTestService("ns1", "svc1", func(svc *v1.Service) {
  3431. svc.Spec.Type = v1.ServiceTypeClusterIP
  3432. svc.Spec.ClusterIP = "172.16.55.4"
  3433. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "port1", "TCP", 1234, 0, 0)
  3434. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "port2", "TCP", 1235, 0, 0)
  3435. svc.Spec.Ports = addTestPort(svc.Spec.Ports, "port3", "UDP", 1236, 0, 0)
  3436. })
  3437. fp.servicesSynced = true
  3438. fp.endpointsSynced = true
  3439. // first, add multi-port service1
  3440. fp.OnServiceAdd(service1)
  3441. fp.syncProxyRules()
  3442. remainingAddrs, _ := fp.netlinkHandle.ListBindAddress(DefaultDummyDevice)
  3443. // should only remain address "172.16.55.4"
  3444. if len(remainingAddrs) != 1 {
  3445. t.Errorf("Expected number of remaining bound addrs after cleanup to be %v. Got %v", 1, len(remainingAddrs))
  3446. }
  3447. if remainingAddrs[0] != "172.16.55.4" {
  3448. t.Errorf("Expected remaining address should be %s, got %s", "172.16.55.4", remainingAddrs[0])
  3449. }
  3450. // update multi-port service1 to single-port service2
  3451. fp.OnServiceUpdate(service1, service2)
  3452. fp.syncProxyRules()
  3453. remainingAddrs, _ = fp.netlinkHandle.ListBindAddress(DefaultDummyDevice)
  3454. // should still only remain address "172.16.55.4"
  3455. if len(remainingAddrs) != 1 {
  3456. t.Errorf("Expected number of remaining bound addrs after cleanup to be %v. Got %v", 1, len(remainingAddrs))
  3457. } else if remainingAddrs[0] != "172.16.55.4" {
  3458. t.Errorf("Expected remaining address should be %s, got %s", "172.16.55.4", remainingAddrs[0])
  3459. }
  3460. // update single-port service2 to multi-port service3
  3461. fp.OnServiceUpdate(service2, service3)
  3462. fp.syncProxyRules()
  3463. remainingAddrs, _ = fp.netlinkHandle.ListBindAddress(DefaultDummyDevice)
  3464. // should still only remain address "172.16.55.4"
  3465. if len(remainingAddrs) != 1 {
  3466. t.Errorf("Expected number of remaining bound addrs after cleanup to be %v. Got %v", 1, len(remainingAddrs))
  3467. } else if remainingAddrs[0] != "172.16.55.4" {
  3468. t.Errorf("Expected remaining address should be %s, got %s", "172.16.55.4", remainingAddrs[0])
  3469. }
  3470. // delete multi-port service3
  3471. fp.OnServiceDelete(service3)
  3472. fp.syncProxyRules()
  3473. remainingAddrs, _ = fp.netlinkHandle.ListBindAddress(DefaultDummyDevice)
  3474. // all addresses should be unbound
  3475. if len(remainingAddrs) != 0 {
  3476. t.Errorf("Expected number of remaining bound addrs after cleanup to be %v. Got %v", 0, len(remainingAddrs))
  3477. }
  3478. }
  3479. func Test_getFirstColumn(t *testing.T) {
  3480. testCases := []struct {
  3481. name string
  3482. fileContent string
  3483. want []string
  3484. wantErr bool
  3485. }{
  3486. {
  3487. name: "valid content",
  3488. fileContent: `libiscsi_tcp 28672 1 iscsi_tcp, Live 0xffffffffc07ae000
  3489. libiscsi 57344 3 ib_iser,iscsi_tcp,libiscsi_tcp, Live 0xffffffffc079a000
  3490. raid10 57344 0 - Live 0xffffffffc0597000`,
  3491. want: []string{"libiscsi_tcp", "libiscsi", "raid10"},
  3492. wantErr: false,
  3493. },
  3494. }
  3495. for _, test := range testCases {
  3496. t.Run(test.name, func(t *testing.T) {
  3497. got, err := getFirstColumn(strings.NewReader(test.fileContent))
  3498. if (err != nil) != test.wantErr {
  3499. t.Errorf("getFirstColumn() error = %v, wantErr %v", err, test.wantErr)
  3500. return
  3501. }
  3502. if !reflect.DeepEqual(got, test.want) {
  3503. t.Errorf("getFirstColumn() = %v, want %v", got, test.want)
  3504. }
  3505. })
  3506. }
  3507. }
  3508. // The majority of EndpointSlice specific tests are not ipvs specific and focus on
  3509. // the shared EndpointChangeTracker and EndpointSliceCache. This test ensures that the
  3510. // ipvs proxier supports translating EndpointSlices to ipvs output.
  3511. func TestEndpointSliceE2E(t *testing.T) {
  3512. ipt := iptablestest.NewFake()
  3513. ipvs := ipvstest.NewFake()
  3514. ipset := ipsettest.NewFake(testIPSetVersion)
  3515. fp := NewFakeProxier(ipt, ipvs, ipset, nil, nil, true)
  3516. fp.servicesSynced = true
  3517. fp.endpointsSynced = true
  3518. fp.endpointSlicesSynced = true
  3519. // Add initial service
  3520. serviceName := "svc1"
  3521. namespaceName := "ns1"
  3522. fp.OnServiceAdd(&v1.Service{
  3523. ObjectMeta: metav1.ObjectMeta{Name: serviceName, Namespace: namespaceName},
  3524. Spec: v1.ServiceSpec{
  3525. ClusterIP: "172.20.1.1",
  3526. Selector: map[string]string{"foo": "bar"},
  3527. Ports: []v1.ServicePort{{Name: "", TargetPort: intstr.FromInt(80), Protocol: v1.ProtocolTCP}},
  3528. },
  3529. })
  3530. // Add initial endpoint slice
  3531. tcpProtocol := v1.ProtocolTCP
  3532. endpointSlice := &discovery.EndpointSlice{
  3533. ObjectMeta: metav1.ObjectMeta{
  3534. Name: fmt.Sprintf("%s-1", serviceName),
  3535. Namespace: namespaceName,
  3536. Labels: map[string]string{discovery.LabelServiceName: serviceName},
  3537. },
  3538. Ports: []discovery.EndpointPort{{
  3539. Name: utilpointer.StringPtr(""),
  3540. Port: utilpointer.Int32Ptr(80),
  3541. Protocol: &tcpProtocol,
  3542. }},
  3543. AddressType: discovery.AddressTypeIPv4,
  3544. Endpoints: []discovery.Endpoint{{
  3545. Addresses: []string{"10.0.1.1"},
  3546. Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
  3547. Topology: map[string]string{"kubernetes.io/hostname": testHostname},
  3548. }, {
  3549. Addresses: []string{"10.0.1.2"},
  3550. Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
  3551. Topology: map[string]string{"kubernetes.io/hostname": "node2"},
  3552. }, {
  3553. Addresses: []string{"10.0.1.3"},
  3554. Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
  3555. Topology: map[string]string{"kubernetes.io/hostname": "node3"},
  3556. }},
  3557. }
  3558. fp.OnEndpointSliceAdd(endpointSlice)
  3559. fp.syncProxyRules()
  3560. // Ensure that Proxier updates ipvs appropriately after EndpointSlice update
  3561. assert.NotNil(t, fp.ipsetList["KUBE-LOOP-BACK"])
  3562. activeEntries1 := fp.ipsetList["KUBE-LOOP-BACK"].activeEntries
  3563. assert.Equal(t, 1, activeEntries1.Len(), "Expected 1 active entry in KUBE-LOOP-BACK")
  3564. assert.Equal(t, true, activeEntries1.Has("10.0.1.1,tcp:80,10.0.1.1"), "Expected activeEntries to reference first (local) pod")
  3565. virtualServers1, vsErr1 := ipvs.GetVirtualServers()
  3566. assert.Nil(t, vsErr1, "Expected no error getting virtual servers")
  3567. assert.Len(t, virtualServers1, 1, "Expected 1 virtual server")
  3568. realServers1, rsErr1 := ipvs.GetRealServers(virtualServers1[0])
  3569. assert.Nil(t, rsErr1, "Expected no error getting real servers")
  3570. assert.Len(t, realServers1, 3, "Expected 3 real servers")
  3571. assert.Equal(t, realServers1[0].String(), "10.0.1.1:80")
  3572. assert.Equal(t, realServers1[1].String(), "10.0.1.2:80")
  3573. assert.Equal(t, realServers1[2].String(), "10.0.1.3:80")
  3574. fp.OnEndpointSliceDelete(endpointSlice)
  3575. fp.syncProxyRules()
  3576. // Ensure that Proxier updates ipvs appropriately after EndpointSlice delete
  3577. assert.NotNil(t, fp.ipsetList["KUBE-LOOP-BACK"])
  3578. activeEntries2 := fp.ipsetList["KUBE-LOOP-BACK"].activeEntries
  3579. assert.Equal(t, 0, activeEntries2.Len(), "Expected 0 active entries in KUBE-LOOP-BACK")
  3580. virtualServers2, vsErr2 := ipvs.GetVirtualServers()
  3581. assert.Nil(t, vsErr2, "Expected no error getting virtual servers")
  3582. assert.Len(t, virtualServers2, 1, "Expected 1 virtual server")
  3583. realServers2, rsErr2 := ipvs.GetRealServers(virtualServers2[0])
  3584. assert.Nil(t, rsErr2, "Expected no error getting real servers")
  3585. assert.Len(t, realServers2, 0, "Expected 0 real servers")
  3586. }
  3587. func TestFilterCIDRs(t *testing.T) {
  3588. var cidrList []string
  3589. var cidrs []string
  3590. var expected []string
  3591. cidrs = filterCIDRs(true, []string{})
  3592. if len(cidrs) > 0 {
  3593. t.Errorf("An empty list produces a non-empty return %v", cidrs)
  3594. }
  3595. cidrList = []string{"1000::/64", "10.0.0.0/16", "11.0.0.0/16", "2000::/64"}
  3596. expected = []string{"1000::/64", "2000::/64"}
  3597. cidrs = filterCIDRs(true, cidrList)
  3598. if !reflect.DeepEqual(cidrs, expected) {
  3599. t.Errorf("cidrs %v is not expected %v", cidrs, expected)
  3600. }
  3601. expected = []string{"10.0.0.0/16", "11.0.0.0/16"}
  3602. cidrs = filterCIDRs(false, cidrList)
  3603. if !reflect.DeepEqual(cidrs, expected) {
  3604. t.Errorf("cidrs %v is not expected %v", cidrs, expected)
  3605. }
  3606. cidrList = []string{"1000::/64", "2000::/64"}
  3607. expected = []string{}
  3608. cidrs = filterCIDRs(false, cidrList)
  3609. if len(cidrs) > 0 {
  3610. t.Errorf("cidrs %v is not expected %v", cidrs, expected)
  3611. }
  3612. }