fanhuadesenlinnn/kube-vip
1
0
Fork 0
mirror of https://hubproxy.babadafafafafa.cn/https://github.com/kube-vip/kube-vip.git synced 2026-09-20 08:03:47 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fixed services error handling (#1478)

Browse Source 
* Fixed services error handling

Signed-off-by: Patryk Strusiewicz-Surmacki <patryk.pawel.strusiewicz-surmacki@external.telekom.de>

* Removed redundant code

Signed-off-by: Patryk Strusiewicz-Surmacki <patryk.pawel.strusiewicz-surmacki@external.telekom.de>

---------

Signed-off-by: Patryk Strusiewicz-Surmacki <patryk.pawel.strusiewicz-surmacki@external.telekom.de>
Co-authored-by: Patryk Strusiewicz-Surmacki <patryk.pawel.strusiewicz-surmacki@external.telekom.de>
This commit is contained in:
Patryk Strusiewicz-Surmacki
2026-03-20 10:44:32 +01:00
committed by GitHub
parent dcd8fe0392
commit ca47abfc3a
4 changed files with 11 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
pkg/endpoints/endpoints.go
View File

@@ -75,14 +75,7 @@ func (p *Processor) AddOrModify(svcCtx *servicecontext.Context, event watch.Even
// start leader election if it's enabled and not already started
if !svcCtx.IsActive && p.config.EnableServicesElection {
wg.Go(func() {
for {
select {
case <-svcCtx.Ctx.Done():
return
default:
startLeaderElection(svcCtx, service, serviceFunc, wg)
}
}
startLeaderElection(svcCtx, service, serviceFunc, wg)
})
}
@@ -177,10 +170,6 @@ func startLeaderElection(svcCtx *servicecontext.Context, service *v1.Service, se
if err != nil {
log.Error(err.Error())
}
if !svcCtx.HasEndpoints.Load() {
log.Debug("there are no available endpoints for this service, exiting watch", "service", service.Name, "uid", service.UID)
return
}
}
}
}

3
pkg/manager/manager.go
View File

@@ -389,7 +389,8 @@ func (sm *Manager) startMode(ctx context.Context) error {
w.ConfigureServices()
if err = w.StartServices(modeCtx); err != nil {
return fmt.Errorf("failed to start services: %w", err)
sm.Kill()
return fmt.Errorf("failed to reconcile services: %w", err)
}
}

4
pkg/services/processor.go
View File

@@ -131,8 +131,8 @@ func (p *Processor) AddOrModify(ctx context.Context, event watch.Event, serviceF
_, usesCommonLease := svc.Annotations[kubevip.ServiceLease]
if usesCommonLease && svc.Spec.ExternalTrafficPolicy != v1.ServiceExternalTrafficPolicyTypeCluster {
return false, fmt.Errorf("annotation %q cannot be used with service traffic policy other than %q",
kubevip.ServiceLease, v1.ServiceExternalTrafficPolicyTypeCluster)
return false, fmt.Errorf("annotation %q cannot be used with service traffic policy other than %q, service %s/%s",
kubevip.ServiceLease, v1.ServiceExternalTrafficPolicyTypeCluster, svc.Namespace, svc.Name)
}
svcCtx, err := p.getServiceContext(svc.UID)

9
pkg/services/watch_services.go
View File

@@ -55,9 +55,12 @@ func (p *Processor) ServicesWatcher(ctx context.Context, serviceFunc func(*servi
wg := sync.WaitGroup{}
defer wg.Wait()
watcherCtx, watcherCancel := context.WithCancel(ctx)
defer watcherCancel()
wg.Go(func() {
<-ctx.Done()
log.Debug("(svcs) context cancelled")
<-watcherCtx.Done()
log.Debug("(svcs) watcher context cancelled")
rw.Stop()
p.Stop()
})
@@ -70,7 +73,7 @@ func (p *Processor) ServicesWatcher(ctx context.Context, serviceFunc func(*servi
// We need to inspect the event and get ResourceVersion out of it
switch event.Type {
case watch.Added, watch.Modified:
restart, err := p.AddOrModify(ctx, event, serviceFunc, &wg)
restart, err := p.AddOrModify(watcherCtx, event, serviceFunc, &wg)
if restart {
break
}