-
Notifications
You must be signed in to change notification settings - Fork 331
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: Jakub Dyszkiewicz <jakub.dyszkiewicz@gmail.com>
- Loading branch information
1 parent
b94f248
commit 3220769
Showing
11 changed files
with
462 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,146 @@ | ||
| package vip | ||
|
|
||
| import ( | ||
| "context" | ||
| "net" | ||
| "time" | ||
|
|
||
| "github.com/Nordix/simple-ipam/pkg/ipam" | ||
| "github.com/go-logr/logr" | ||
| "github.com/pkg/errors" | ||
| "github.com/prometheus/client_golang/prometheus" | ||
|
|
||
| meshservice_api "github.com/kumahq/kuma/pkg/core/resources/apis/meshservice/api/v1alpha1" | ||
| "github.com/kumahq/kuma/pkg/core/resources/manager" | ||
| "github.com/kumahq/kuma/pkg/core/resources/store" | ||
| "github.com/kumahq/kuma/pkg/core/runtime/component" | ||
| "github.com/kumahq/kuma/pkg/core/user" | ||
| core_metrics "github.com/kumahq/kuma/pkg/metrics" | ||
| ) | ||
|
|
||
| // Allocator manages IPs for MeshServices. | ||
| // Each time allocator starts it initiates the IPAM based on existing MeshServices | ||
| // We don't free addresses explicitly, but we always allocate next free IP to avoid a problem when we | ||
| // 1) Remove Service A with IP X | ||
| // 2) Add new Service B that gets IP X | ||
| // 3) Clients that were sending the traffic to A now sends the traffic to B for brief amount of time | ||
| // IPAM is kept in memory to avoid state management, so technically this problem can still happen when leader changes | ||
| // However, leader should not change before TTL of a DNS that serves this VIP. | ||
| // | ||
| // It's technically possible to allocate all addresses by creating and removing services in the loop. | ||
| // However, CIDR has range of 16M addresses, after that the component will just restart. | ||
| type Allocator struct { | ||
| logger logr.Logger | ||
| cidr string | ||
| interval time.Duration | ||
| metric prometheus.Summary | ||
| resManager manager.ResourceManager | ||
| } | ||
|
|
||
| var _ component.Component = &Allocator{} | ||
|
|
||
| func NewAllocator( | ||
| logger logr.Logger, | ||
| metrics core_metrics.Metrics, | ||
| resManager manager.ResourceManager, | ||
| cidr string, | ||
| interval time.Duration, | ||
| ) (*Allocator, error) { | ||
| metric := prometheus.NewSummary(prometheus.SummaryOpts{ | ||
| Name: "component_ms_vip_allocator", | ||
| Help: "Summary of Inter CP Heartbeat component interval", | ||
| Objectives: core_metrics.DefaultObjectives, | ||
| }) | ||
| if err := metrics.Register(metric); err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| return &Allocator{ | ||
| logger: logger, | ||
| resManager: resManager, | ||
| cidr: cidr, | ||
| interval: interval, | ||
| metric: metric, | ||
| }, nil | ||
| } | ||
|
|
||
| func (a *Allocator) Start(stop <-chan struct{}) error { | ||
| a.logger.Info("starting") | ||
| ticker := time.NewTicker(a.interval) | ||
| ctx := user.Ctx(context.Background(), user.ControlPlane) | ||
|
|
||
| kumaIpam, err := a.initIpam(ctx) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| for { | ||
| select { | ||
| case <-ticker.C: | ||
| start := time.Now() | ||
| if err := a.allocateVips(ctx, kumaIpam); err != nil { | ||
| return err | ||
| } | ||
| a.metric.Observe(float64(time.Since(start).Milliseconds())) | ||
| case <-stop: | ||
| a.logger.Info("stopping") | ||
| return nil | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (a *Allocator) NeedLeaderElection() bool { | ||
| return true | ||
| } | ||
|
|
||
| func (a *Allocator) initIpam(ctx context.Context) (*ipam.IPAM, error) { | ||
| newIPAM, err := ipam.New(a.cidr) | ||
| if err != nil { | ||
| return nil, errors.Wrapf(err, "could not allocate IPAM of CIDR %s", a.cidr) | ||
| } | ||
|
|
||
| services := &meshservice_api.MeshServiceResourceList{} | ||
| if err := a.resManager.List(ctx, services); err != nil { | ||
| return nil, errors.Wrap(err, "could not list mesh services for initialization of ipam") | ||
| } | ||
| for _, service := range services.Items { | ||
| for _, vip := range service.Status.VIPs { | ||
| _ = newIPAM.Reserve(net.ParseIP(vip.IP)) // ignore error for outside of range | ||
| } | ||
| } | ||
|
|
||
| return newIPAM, nil | ||
| } | ||
|
|
||
| func (a *Allocator) allocateVips(ctx context.Context, kumaIpam *ipam.IPAM) error { | ||
| services := &meshservice_api.MeshServiceResourceList{} | ||
| if err := a.resManager.List(ctx, services); err != nil { | ||
| return errors.Wrap(err, "could not list mesh services for ip allocation") | ||
| } | ||
|
|
||
| for _, svc := range services.Items { | ||
| if len(svc.Status.VIPs) == 0 { | ||
| log := a.logger.WithValues("service", svc.Meta.GetName(), "mesh", svc.Meta.GetMesh()) | ||
| ip, err := kumaIpam.Allocate() | ||
| if err != nil { | ||
| return errors.Wrapf(err, "could not allocate the address for svc %s", svc.Meta.GetName()) | ||
| } | ||
| log.Info("allocating IP for a service", "ip", ip.String()) | ||
| svc.Status.VIPs = []meshservice_api.VIP{ | ||
| { | ||
| IP: ip.String(), | ||
| }, | ||
| } | ||
|
|
||
| if err := a.resManager.Update(ctx, svc, store.UpdateWithLabels(svc.GetMeta().GetLabels())); err != nil { | ||
| msg := "could not update service with allocated Kuma VIP. Will try to update in the next allocation window" | ||
| if errors.Is(err, &store.ResourceConflictError{}) { | ||
| log.Info(msg, "cause", "conflict", "interval", a.interval) | ||
| } else { | ||
| log.Error(err, msg, "interval", a.interval) | ||
| } | ||
| } | ||
| } | ||
| } | ||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,92 @@ | ||
| package vip | ||
|
|
||
| import ( | ||
| "context" | ||
| "time" | ||
|
|
||
| "github.com/go-logr/logr" | ||
| . "github.com/onsi/ginkgo/v2" | ||
| . "github.com/onsi/gomega" | ||
|
|
||
| meshservice_api "github.com/kumahq/kuma/pkg/core/resources/apis/meshservice/api/v1alpha1" | ||
| "github.com/kumahq/kuma/pkg/core/resources/manager" | ||
| "github.com/kumahq/kuma/pkg/core/resources/model" | ||
| "github.com/kumahq/kuma/pkg/core/resources/store" | ||
| core_metrics "github.com/kumahq/kuma/pkg/metrics" | ||
| "github.com/kumahq/kuma/pkg/plugins/resources/memory" | ||
| test_metrics "github.com/kumahq/kuma/pkg/test/metrics" | ||
| "github.com/kumahq/kuma/pkg/test/resources/samples" | ||
| ) | ||
|
|
||
| var _ = Describe("VIP Allocator", func() { | ||
| var stopCh chan struct{} | ||
| var resManager manager.ResourceManager | ||
| var metrics core_metrics.Metrics | ||
|
|
||
| BeforeEach(func() { | ||
| m, err := core_metrics.NewMetrics("") | ||
| Expect(err).ToNot(HaveOccurred()) | ||
| metrics = m | ||
| resManager = manager.NewResourceManager(memory.NewStore()) | ||
| allocator, err := NewAllocator(logr.Discard(), m, resManager, "241.0.0.0/8", 50*time.Millisecond) | ||
| Expect(err).ToNot(HaveOccurred()) | ||
| stopCh = make(chan struct{}) | ||
| go func() { | ||
| defer GinkgoRecover() | ||
| Expect(allocator.Start(stopCh)).To(Succeed()) | ||
| }() | ||
|
|
||
| Expect(samples.MeshDefaultBuilder().Create(resManager)).To(Succeed()) | ||
| }) | ||
|
|
||
| AfterEach(func() { | ||
| close(stopCh) | ||
| }) | ||
|
|
||
| vipOfMeshService := func(name string) string { | ||
| ms := meshservice_api.NewMeshServiceResource() | ||
| err := resManager.Get(context.Background(), ms, store.GetByKey(name, model.DefaultMesh)) | ||
| Expect(err).ToNot(HaveOccurred()) | ||
| if len(ms.Status.VIPs) == 0 { | ||
| return "" | ||
| } | ||
| return ms.Status.VIPs[0].IP | ||
| } | ||
|
|
||
| It("should allocate vip for service without vip", func() { | ||
| // when | ||
| err := samples.MeshServiceBackendBuilder().WithoutVIP().Create(resManager) | ||
| Expect(err).ToNot(HaveOccurred()) | ||
|
|
||
| // then | ||
| Eventually(func(g Gomega) { | ||
| g.Expect(vipOfMeshService("backend")).Should(Equal("241.0.0.0")) | ||
| }, "10s", "100ms").Should(Succeed()) | ||
| }) | ||
|
|
||
| It("should not reuse IPs", func() { | ||
| // given | ||
| err := samples.MeshServiceBackendBuilder().WithoutVIP().Create(resManager) | ||
| Expect(err).ToNot(HaveOccurred()) | ||
| Eventually(func(g Gomega) { | ||
| g.Expect(vipOfMeshService("backend")).Should(Equal("241.0.0.0")) | ||
| }, "10s", "100ms").Should(Succeed()) | ||
|
|
||
| // when resource is reapplied | ||
| err = resManager.Delete(context.Background(), meshservice_api.NewMeshServiceResource(), store.DeleteByKey("backend", model.DefaultMesh)) | ||
| Expect(err).ToNot(HaveOccurred()) | ||
| err = samples.MeshServiceBackendBuilder().WithoutVIP().Create(resManager) | ||
| Expect(err).ToNot(HaveOccurred()) | ||
|
|
||
| // then | ||
| Eventually(func(g Gomega) { | ||
| g.Expect(vipOfMeshService("backend")).Should(Equal("241.0.0.1")) | ||
| }, "10s", "100ms").Should(Succeed()) | ||
| }) | ||
|
|
||
| It("should emit metric", func() { | ||
| Eventually(func(g Gomega) { | ||
| g.Expect(test_metrics.FindMetric(metrics, "component_ms_vip_allocator")).ToNot(BeNil()) | ||
| }, "10s", "100ms").Should(Succeed()) | ||
| }) | ||
| }) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| package vip_test | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| "github.com/kumahq/kuma/pkg/test" | ||
| ) | ||
|
|
||
| func TestVIP(t *testing.T) { | ||
| test.RunSpecs(t, "MeshService VIP Suite") | ||
| } |
Oops, something went wrong.