func()

in pkg/providers/apisix/apisix_route.go [627:716]
86 lines of code
16 McCabe index (conditional complexity)

func (c *apisixRouteController) ResourceSync(interval time.Duration, namespace string) {
	objs := c.ApisixRouteInformer.GetIndexer().List()
	delay := GetSyncDelay(interval, len(objs))

	c.svcLock.Lock()
	c.apisixUpstreamLock.Lock()
	defer c.svcLock.Unlock()
	defer c.apisixUpstreamLock.Unlock()

	c.svcMap = make(map[string]map[string]struct{})
	c.apisixUpstreamMap = make(map[string]map[string]struct{})

	for i, obj := range objs {
		key, err := cache.MetaNamespaceKeyFunc(obj)
		if err != nil {
			log.Errorw("ApisixRoute sync failed, found ApisixRoute resource with bad meta namespace key",
				zap.Error(err),
			)
			continue
		}
		if !c.namespaceProvider.IsWatchingNamespace(key) {
			continue
		}
		ns, _, err := cache.SplitMetaNamespaceKey(key)
		if err != nil {
			log.Errorw("split ApisixRoute meta key failed",
				zap.Error(err),
				zap.String("key", key),
			)
			continue
		}
		if namespace != "" && ns != namespace {
			continue
		}
		ar := kube.MustNewApisixRoute(obj)
		if !c.isEffective(ar) {
			log.Debugw("ignore noneffective ApisixRoute sync event arrived",
				zap.Any("final state", ar),
			)
			continue
		}
		log.Debugw("ResourceSync",
			zap.String("resource", "ApisixRoute"),
			zap.String("key", key),
			zap.Duration("calc_delay", delay),
			zap.Int("i", i),
			zap.Duration("delay", delay*time.Duration(i)),
		)
		c.workqueue.AddAfter(&types.Event{
			Type: types.EventSync,
			Object: kube.ApisixRouteEvent{
				Key:          key,
				GroupVersion: ar.GroupVersion(),
			},
		}, delay*time.Duration(i))

		var (
			backends  []string
			upstreams []string
		)
		switch ar.GroupVersion() {
		case config.ApisixV2:
			for _, rule := range ar.V2().Spec.HTTP {
				for _, backend := range rule.Backends {
					backends = append(backends, ns+"/"+backend.ServiceName)
				}
				for _, upstream := range rule.Upstreams {
					upstreams = append(upstreams, ns+"/"+upstream.Name)
				}
			}
		default:
			log.Errorw("unknown ApisixRoute version",
				zap.String("version", ar.GroupVersion()),
				zap.String("key", key),
			)
		}
		for _, svcKey := range backends {
			if _, ok := c.svcMap[svcKey]; !ok {
				c.svcMap[svcKey] = make(map[string]struct{})
			}
			c.svcMap[svcKey][key] = struct{}{}
		}
		for _, upstreamKey := range upstreams {
			if _, ok := c.apisixUpstreamMap[upstreamKey]; !ok {
				c.apisixUpstreamMap[upstreamKey] = make(map[string]struct{})
			}
			c.apisixUpstreamMap[upstreamKey][key] = struct{}{}
		}
	}
}