func()

in pkg/providers/apisix/apisix_cluster_config.go [86:213]

• 111 lines of code
• 17 McCabe index (conditional complexity)

func (c *apisixClusterConfigController) sync(ctx context.Context, ev *types.Event) error {
	event := ev.Object.(kube.ApisixClusterConfigEvent)
	key := event.Key
	_, name, err := cache.SplitMetaNamespaceKey(key)
	if err != nil {
		log.Errorf("found ApisixClusterConfig resource with invalid meta key %s: %s", key, err)
		return err
	}

	var multiVersioned kube.ApisixClusterConfig
	switch event.GroupVersion {
	case config.ApisixV2:
		multiVersioned, err = c.ApisixClusterConfigLister.V2(name)
	default:
		return fmt.Errorf("unsupported ApisixClusterConfig group version %s", event.GroupVersion)
	}

	if err != nil {
		if !k8serrors.IsNotFound(err) {
			log.Errorw("failed to get ApisixClusterConfig",
				zap.Error(err),
				zap.String("key", key),
				zap.String("version", event.GroupVersion),
			)
			return err
		}
		if ev.Type == types.EventSync {
			// ignore not found error in delay sync
			return nil
		}
		if ev.Type != types.EventDelete {
			log.Warnw("ApisixClusterConfig was deleted before it can be delivered",
				zap.String("key", key),
				zap.String("version", event.GroupVersion),
			)
			return nil
		}
	}
	if ev.Type == types.EventDelete {
		if multiVersioned != nil {
			// We still find the resource while we are processing the DELETE event,
			// that means object with same namespace and name was created, discarding
			// this stale DELETE event.
			log.Warnf("discard the stale ApisixClusterConfig delete event since the %s exists", key)
			return nil
		}
		multiVersioned = ev.Tombstone.(kube.ApisixClusterConfig)
	}

	switch event.GroupVersion {
	case config.ApisixV2:
		acc := multiVersioned.V2()
		// Currently we don't handle multiple cluster, so only process
		// the default apisix cluster.
		if acc.Name != c.Config.APISIX.DefaultClusterName {
			log.Infow("ignore non-default apisix cluster config",
				zap.String("default_cluster_name", c.Config.APISIX.DefaultClusterName),
				zap.Any("ApisixClusterConfig", acc),
			)
			return nil
		}
		// Cluster delete is dangerous.
		// TODO handle delete?
		if ev.Type == types.EventDelete {
			log.Error("ApisixClusterConfig delete event for default apisix cluster will be ignored")
			return nil
		}

		if acc.Spec.Admin != nil {
			clusterOpts := &apisix.ClusterOptions{
				Name:     acc.Name,
				BaseURL:  acc.Spec.Admin.BaseURL,
				AdminKey: acc.Spec.Admin.AdminKey,
			}
			log.Infow("updating cluster",
				zap.Any("opts", clusterOpts),
			)
			// TODO we may first call AddCluster.
			// Since now we already have the default cluster, we just call UpdateCluster.
			if err := c.APISIX.UpdateCluster(ctx, clusterOpts); err != nil {
				log.Errorw("failed to update cluster",
					zap.String("cluster_name", acc.Name),
					zap.Error(err),
					zap.Any("opts", clusterOpts),
				)
				c.RecordEvent(acc, corev1.EventTypeWarning, utils.ResourceSyncAborted, err)
				c.recordStatus(acc, utils.ResourceSyncAborted, err, metav1.ConditionFalse, acc.GetGeneration())
				return err
			}
		}

		globalRule, err := c.translator.TranslateClusterConfigV2(acc)
		if err != nil {
			log.Errorw("failed to translate ApisixClusterConfig",
				zap.Error(err),
				zap.String("key", key),
				zap.Any("object", acc),
			)
			c.RecordEvent(acc, corev1.EventTypeWarning, utils.ResourceSyncAborted, err)
			c.recordStatus(acc, utils.ResourceSyncAborted, err, metav1.ConditionFalse, acc.GetGeneration())
			return err
		}
		log.Debugw("translated global_rule",
			zap.Any("object", globalRule),
		)

		// TODO multiple cluster support
		if ev.Type.IsAddEvent() {
			_, err = c.APISIX.Cluster(acc.Name).GlobalRule().Create(ctx, globalRule, ev.Type.IsSyncEvent())
		} else {
			_, err = c.APISIX.Cluster(acc.Name).GlobalRule().Update(ctx, globalRule, false)
		}
		if err != nil {
			log.Errorw("failed to reflect global_rule changes to apisix cluster",
				zap.Any("global_rule", globalRule),
				zap.Any("cluster", acc.Name),
			)
			c.RecordEvent(acc, corev1.EventTypeWarning, utils.ResourceSyncAborted, err)
			c.recordStatus(acc, utils.ResourceSyncAborted, err, metav1.ConditionFalse, acc.GetGeneration())
			return err
		}
		c.RecordEvent(acc, corev1.EventTypeNormal, utils.ResourceSynced, nil)
		c.recordStatus(acc, utils.ResourceSynced, nil, metav1.ConditionTrue, acc.GetGeneration())
		return nil
	default:
		return fmt.Errorf("unsupported ApisixClusterConfig group version %s", event.GroupVersion)
	}
}