OSPP feat(divide): support label-based canary routing and partition metrics

shenyu-admin/src/main/java/org/apache/shenyu/admin/service/impl/DiscoveryUpstreamServiceImpl.java

@@ -42,6 +42,7 @@
 import org.apache.shenyu.admin.service.DiscoveryUpstreamService;
 import org.apache.shenyu.admin.service.configs.ConfigsImportContext;
 import org.apache.shenyu.admin.transfer.DiscoveryTransfer;
+import org.apache.shenyu.admin.utils.DiscoveryUpstreamPropsValidator;
 import org.apache.shenyu.admin.utils.Assert;
 import org.apache.shenyu.admin.utils.ShenyuResultMessage;
 import org.apache.shenyu.common.dto.DiscoverySyncData;
@@ -138,6 +139,8 @@ public void createOrUpdateBatch(final List<DiscoveryUpstreamDTO> upstreams) {
     @Override
     @Transactional
     public int updateBatch(final String discoveryHandlerId, final List<DiscoveryUpstreamDTO> discoveryUpstreamDTOList) {
+        // Validate the complete replacement before deleting anything or emitting a sync event.
+        discoveryUpstreamDTOList.forEach(upstream -> DiscoveryUpstreamPropsValidator.validate(upstream.getProps()));
         discoveryUpstreamMapper.deleteByDiscoveryHandlerId(discoveryHandlerId);
         for (DiscoveryUpstreamDTO discoveryUpstreamDTO : discoveryUpstreamDTOList) {
             discoveryUpstreamDTO.setId(null);
@@ -160,6 +163,7 @@ public void afterCommit() {
 
     @Override
     public void nativeCreateOrUpdate(final DiscoveryUpstreamDTO discoveryUpstreamDTO) {
+        DiscoveryUpstreamPropsValidator.validate(discoveryUpstreamDTO.getProps());
         DiscoveryUpstreamDO discoveryUpstreamDO = DiscoveryUpstreamDO.buildDiscoveryUpstreamDO(discoveryUpstreamDTO);
         if (StringUtils.hasLength(discoveryUpstreamDTO.getId())) {
             discoveryUpstreamMapper.updateSelective(discoveryUpstreamDO);
@@ -264,6 +268,7 @@ public void refreshBySelectorId(final String selectorId) {
      * @return the string
      */
     private String create(final DiscoveryUpstreamDTO discoveryUpstreamDTO) {
+        DiscoveryUpstreamPropsValidator.validate(discoveryUpstreamDTO.getProps());
         DiscoveryUpstreamDO discoveryUpstreamDO = DiscoveryUpstreamDO.buildDiscoveryUpstreamDO(discoveryUpstreamDTO);
         discoveryUpstreamMapper.insert(discoveryUpstreamDO);
         fetchAll(discoveryUpstreamDTO.getDiscoveryHandlerId());
@@ -277,6 +282,7 @@ private String create(final DiscoveryUpstreamDTO discoveryUpstreamDTO) {
      * @return the string
      */
     private String update(final DiscoveryUpstreamDTO discoveryUpstreamDTO) {
+        DiscoveryUpstreamPropsValidator.validate(discoveryUpstreamDTO.getProps());
         DiscoveryUpstreamDO discoveryUpstreamDO = DiscoveryUpstreamDO.buildDiscoveryUpstreamDO(discoveryUpstreamDTO);
         discoveryUpstreamMapper.update(discoveryUpstreamDO);
         fetchAll(discoveryUpstreamDTO.getDiscoveryHandlerId());
@@ -317,6 +323,7 @@ public ConfigImportResult importData(final List<DiscoveryUpstreamDTO> discoveryU
         if (CollectionUtils.isEmpty(discoveryUpstreamList)) {
             return ConfigImportResult.success();
         }
+        discoveryUpstreamList.forEach(upstream -> DiscoveryUpstreamPropsValidator.validate(upstream.getProps()));
         int successCount = 0;
         StringBuilder errorMsgBuilder = new StringBuilder();
         Map<String, List<DiscoveryUpstreamDO>> discoveryHandlerUpstreamMap = discoveryUpstreamMapper
@@ -363,6 +370,7 @@ public ConfigImportResult importData(final String namespace, final List<Discover
             return ConfigImportResult.success();
         }
         Map<String, String> discoveryHandlerIdMapping = context.getDiscoveryHandlerIdMapping();
+        discoveryUpstreamList.forEach(upstream -> DiscoveryUpstreamPropsValidator.validate(upstream.getProps()));
         int successCount = 0;
         StringBuilder errorMsgBuilder = new StringBuilder();
         Map<String, Set<String>> discoveryHandlerUpstreamMap = discoveryUpstreamMapper

shenyu-admin/src/main/java/org/apache/shenyu/admin/service/impl/ProxySelectorServiceImpl.java

@@ -49,6 +49,7 @@
 import org.apache.shenyu.admin.service.ProxySelectorService;
 import org.apache.shenyu.admin.service.configs.ConfigsImportContext;
 import org.apache.shenyu.admin.transfer.DiscoveryTransfer;
+import org.apache.shenyu.admin.utils.DiscoveryUpstreamPropsValidator;
 import org.apache.shenyu.admin.utils.ShenyuResultMessage;
 import org.apache.shenyu.admin.utils.Assert;
 import org.apache.shenyu.common.dto.ProxySelectorData;
@@ -239,6 +240,7 @@ public String delete(final List<String> ids) {
     @Override
     @Transactional(rollbackFor = Exception.class)
     public String create(final ProxySelectorAddDTO proxySelectorAddDTO) {
+        validateUpstreamProps(proxySelectorAddDTO);
         Timestamp currentTime = new Timestamp(System.currentTimeMillis());
         ProxySelectorDO proxySelectorDO = ProxySelectorDO.buildProxySelectorDO(proxySelectorAddDTO);
         String proxySelectorId = proxySelectorDO.getId();
@@ -341,6 +343,7 @@ private static DiscoveryDO buildDiscovery(final ProxySelectorAddDTO proxySelecto
     @Override
     @Transactional(rollbackFor = Exception.class)
     public String bindingDiscoveryHandler(final ProxySelectorAddDTO proxySelectorAddDTO) {
+        validateUpstreamProps(proxySelectorAddDTO);
         Timestamp currentTime = new Timestamp(System.currentTimeMillis());
         String selectorId = proxySelectorAddDTO.getSelectorId();
         final ProxySelectorAddDTO.Discovery discovery = proxySelectorAddDTO.getDiscovery();
@@ -392,6 +395,7 @@ public String bindingDiscoveryHandler(final ProxySelectorAddDTO proxySelectorAdd
      */
     @Transactional(rollbackFor = Exception.class)
     public String update(final ProxySelectorAddDTO proxySelectorAddDTO) {
+        validateUpstreamProps(proxySelectorAddDTO);
         ProxySelectorAddDTO.Discovery discovery = proxySelectorAddDTO.getDiscovery();
         Assert.notNull(discovery, "Discovery configuration is required");
         ProxySelectorDO proxySelectorDO = ProxySelectorDO.buildProxySelectorDO(proxySelectorAddDTO);
@@ -461,6 +465,12 @@ public void afterCompletion(final int status) {
         return ShenyuResultMessage.UPDATE_SUCCESS;
     }
 
+    private void validateUpstreamProps(final ProxySelectorAddDTO dto) {
+        if (Objects.nonNull(dto.getDiscoveryUpstreams())) {
+            dto.getDiscoveryUpstreams().forEach(upstream -> DiscoveryUpstreamPropsValidator.validate(upstream.getProps()));
+        }
+    }
+
     @Override
     public void fetchData(final String discoveryHandlerId) {
         DiscoveryHandlerDO discoveryHandlerDO = discoveryHandlerMapper.selectById(discoveryHandlerId);

shenyu-admin/src/main/java/org/apache/shenyu/admin/service/impl/RuleServiceImpl.java

@@ -30,12 +30,14 @@
 import org.apache.shenyu.admin.mapper.RuleConditionMapper;
 import org.apache.shenyu.admin.mapper.RuleMapper;
 import org.apache.shenyu.admin.mapper.SelectorMapper;
+import org.apache.shenyu.admin.mapper.ShenyuDictMapper;
 import org.apache.shenyu.admin.model.dto.RuleConditionDTO;
 import org.apache.shenyu.admin.model.dto.RuleDTO;
 import org.apache.shenyu.admin.model.entity.PluginDO;
 import org.apache.shenyu.admin.model.entity.RuleConditionDO;
 import org.apache.shenyu.admin.model.entity.RuleDO;
 import org.apache.shenyu.admin.model.entity.SelectorDO;
+import org.apache.shenyu.admin.model.entity.ShenyuDictDO;
 import org.apache.shenyu.admin.model.event.rule.RuleCreatedEvent;
 import org.apache.shenyu.admin.model.event.selector.BatchSelectorDeletedEvent;
 import org.apache.shenyu.admin.model.page.CommonPager;
@@ -57,6 +59,8 @@
 import org.apache.shenyu.common.constant.AdminConstants;
 import org.apache.shenyu.common.dto.ConditionData;
 import org.apache.shenyu.common.dto.RuleData;
+import org.apache.shenyu.common.dto.convert.rule.canary.CanaryConfigValidator;
+import org.apache.shenyu.common.dto.convert.rule.impl.DivideRuleHandle;
 import org.apache.shenyu.common.enums.MatchModeEnum;
 import org.apache.shenyu.common.enums.PluginEnum;
 import org.apache.shenyu.common.utils.JsonUtils;
@@ -96,16 +100,20 @@ public class RuleServiceImpl implements RuleService {
 
     private final RuleEventPublisher ruleEventPublisher;
 
+    private final ShenyuDictMapper shenyuDictMapper;
+
     public RuleServiceImpl(final RuleMapper ruleMapper,
                            final RuleConditionMapper ruleConditionMapper,
                            final SelectorMapper selectorMapper,
                            final PluginMapper pluginMapper,
-                           final RuleEventPublisher ruleEventPublisher) {
+                           final RuleEventPublisher ruleEventPublisher,
+                           final ShenyuDictMapper shenyuDictMapper) {
         this.ruleMapper = ruleMapper;
         this.ruleConditionMapper = ruleConditionMapper;
         this.selectorMapper = selectorMapper;
         this.pluginMapper = pluginMapper;
         this.ruleEventPublisher = ruleEventPublisher;
+        this.shenyuDictMapper = shenyuDictMapper;
     }
 
     @Override
@@ -156,6 +164,7 @@ public String registerDefault(final RuleDTO ruleDTO) {
             return "";
         }
         validateRuleHandle(ruleDTO);
+        validateCanaryConfig(ruleDTO.getSelectorId(), ruleDTO.getHandle());
         RuleDO ruleDO = RuleDO.buildRuleDO(ruleDTO);
         if (StringUtils.isEmpty(ruleDTO.getId())) {
             ruleMapper.insertSelective(ruleDO);
@@ -181,6 +190,7 @@ public int createOrUpdate(final RuleDTO ruleDTO) {
     @Override
     public int create(final RuleDTO ruleDTO) {
         validateRuleHandle(ruleDTO);
+        validateCanaryConfig(ruleDTO.getSelectorId(), ruleDTO.getHandle());
         RuleDO ruleDO = RuleDO.buildRuleDO(ruleDTO);
         final int ruleCount = ruleMapper.insertSelective(ruleDO);
         addCondition(ruleDO, ruleDTO.getRuleConditions());
@@ -195,6 +205,7 @@ public int update(final RuleDTO ruleDTO) {
         validateRuleHandle(ruleDTO);
         final RuleDO before = ruleMapper.selectById(ruleDTO.getId());
         Assert.notNull(before, "the updated rule is not found");
+        validateCanaryConfig(StringUtils.defaultIfBlank(ruleDTO.getSelectorId(), before.getSelectorId()), ruleDTO.getHandle());
         RuleDO ruleDO = RuleDO.buildRuleDO(ruleDTO);
         final int ruleCount = ruleMapper.updateSelective(ruleDO);
 
@@ -230,6 +241,38 @@ public int update(final RuleDTO ruleDTO) {
         return ruleCount;
     }
 
+    private void validateCanaryConfig(final String selectorId, final String handle) {
+        if (StringUtils.isBlank(handle)) {
+            return;
+        }
+        SelectorDO selector = selectorMapper.selectById(selectorId);
+        if (Objects.isNull(selector)) {
+            return;
+        }
+        PluginDO plugin = pluginMapper.selectById(selector.getPluginId());
+        if (Objects.isNull(plugin) || !PluginEnum.DIVIDE.getName().equals(plugin.getName())) {
+            return;
+        }
+        try {
+            DivideRuleHandle divide = CanaryConfigValidator.parseHandle(handle);
+            if (Objects.nonNull(divide) && Objects.nonNull(divide.getCanary())) {
+                CanaryConfigValidator.validateExtensions(divide.getCanary(),
+                        name -> validateCanaryDictionary("paramType", name), name -> validateCanaryDictionary("operator", name));
+            }
+        } catch (IllegalArgumentException ex) {
+            throw new ShenyuAdminException(ex.getMessage(), ex);
+        }
+    }
+
+    private void validateCanaryDictionary(final String type, final String name) {
+        boolean supported = shenyuDictMapper.findByType(type).stream()
+                .filter(dict -> Boolean.TRUE.equals(dict.getEnabled()))
+                .map(ShenyuDictDO::getDictValue).anyMatch(name::equals);
+        if (!supported) {
+            throw new IllegalArgumentException("Unsupported Canary " + type + ": " + name);
+        }
+    }
+
     /**
      * find rule by id and namespaceId.
      *
@@ -302,6 +345,7 @@ public ConfigImportResult importData(final List<RuleDTO> ruleList) {
             return ConfigImportResult.success();
         }
 
+        ruleList.forEach(rule -> validateCanaryConfig(rule.getSelectorId(), rule.getHandle()));
         Map<String, List<RuleDO>> selectorRuleMap = ruleMapper
                 .selectAll()
                 .stream()
@@ -347,6 +391,8 @@ public ConfigImportResult importData(final String namespace, final List<RuleDTO>
             return ConfigImportResult.success();
         }
         Map<String, String> selectorIdMapping = context.getSelectorIdMapping();
+        ruleList.stream().filter(rule -> selectorIdMapping.containsKey(rule.getSelectorId()))
+                .forEach(rule -> validateCanaryConfig(selectorIdMapping.get(rule.getSelectorId()), rule.getHandle()));
         Map<String, Set<String>> selectorRuleMap = ruleMapper
                 .selectAllByNamespaceId(namespace)
                 .stream()

shenyu-admin/src/main/java/org/apache/shenyu/admin/transfer/DiscoveryTransfer.java

@@ -17,6 +17,7 @@
 
 package org.apache.shenyu.admin.transfer;
 
+import com.google.gson.JsonElement;
 import org.apache.shenyu.admin.model.dto.DiscoveryDTO;
 import org.apache.shenyu.admin.model.dto.DiscoveryHandlerDTO;
 import org.apache.shenyu.admin.model.dto.DiscoveryRelDTO;
@@ -80,11 +81,16 @@ public CommonUpstream mapToCommonUpstream(DiscoveryUpstreamData discoveryUpstrea
             String url = data.getUrl();
             CommonUpstream commonUpstream = new CommonUpstream(data.getProtocol(), url.split(":")[0], url, false,
                     data.getDateCreated().getTime());
-            Properties properties = Optional.ofNullable(data.getProps())
-                    .map(props -> GsonUtils.getInstance().fromJson(props, Properties.class))
-                    .orElse(new Properties());
-            commonUpstream
-                    .setHealthCheckEnabled(Boolean.parseBoolean(properties.getProperty("healthCheckEnabled", "true")));
+            // Other properties, such as labels, may contain nested JSON objects.
+            boolean healthCheckEnabled = Optional.ofNullable(data.getProps())
+                    .map(props -> GsonUtils.getInstance().fromJson(props, JsonElement.class))
+                    .filter(props -> !props.isJsonNull())
+                    .map(JsonElement::getAsJsonObject)
+                    .map(props -> props.get("healthCheckEnabled"))
+                    .filter(value -> !value.isJsonNull())
+                    .map(value -> Boolean.parseBoolean(value.getAsString()))
+                    .orElse(true);
+            commonUpstream.setHealthCheckEnabled(healthCheckEnabled);
             return commonUpstream;
         }).orElse(null);
     }

shenyu-admin/src/main/java/org/apache/shenyu/admin/utils/DiscoveryUpstreamPropsValidator.java

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.admin.utils;
+
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParseException;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.shenyu.admin.exception.ShenyuAdminException;
+import org.apache.shenyu.common.dto.convert.rule.canary.CanaryConfigValidator;
+import org.apache.shenyu.common.utils.GsonUtils;
+
+import java.util.Objects;
+
+/**
+ * Validation of submitted discovery upstream properties without changing replacement semantics.
+ */
+public final class DiscoveryUpstreamPropsValidator {
+
+    private DiscoveryUpstreamPropsValidator() {
+    }
+
+    /**
+     * Validate the submitted properties without merging or rewriting them.
+     * Omitted labels and an empty labels object are both valid; null labels are invalid.
+     * Empty properties retain the existing API behavior for optional properties.
+     *
+     * @param props the complete submitted properties
+     */
+    public static void validate(final String props) {
+        if (StringUtils.isBlank(props)) {
+            return;
+        }
+        try {
+            JsonElement parsed = GsonUtils.getInstance().fromJson(props, JsonElement.class);
+            if (Objects.isNull(parsed) || !parsed.isJsonObject()) {
+                throw new IllegalArgumentException("props must be a JSON object");
+            }
+            JsonObject object = parsed.getAsJsonObject();
+            if (object.has("labels")) {
+                CanaryConfigValidator.validateLabels(object.get("labels"), "props.labels", true);
+            }
+        } catch (IllegalArgumentException | JsonParseException ex) {
+            throw new ShenyuAdminException("Invalid upstream props: " + ex.getMessage(), ex);
+        }
+    }
+}

shenyu-admin/src/test/java/org/apache/shenyu/admin/service/ConfigsImportDuplicateTest.java

@@ -160,7 +160,8 @@ public void testSelectorsDeduplicateAfterPluginRemapping(final boolean existing)
     @ParameterizedTest
     @ValueSource(booleans = {false, true})
     public void testRulesDeduplicateAfterSelectorRemapping(final boolean existing) {
-        final RuleServiceImpl service = new RuleServiceImpl(ruleMapper, mock(RuleConditionMapper.class), selectorMapper, pluginMapper, mock(RuleEventPublisher.class));
+        final RuleServiceImpl service = new RuleServiceImpl(ruleMapper, mock(RuleConditionMapper.class), selectorMapper, pluginMapper,
+                mock(RuleEventPublisher.class), dictMapper);
         final ConfigsImportContext context = new ConfigsImportContext();
         context.getSelectorIdMapping().put("source", "target");
         context.getSelectorIdMapping().put("alias", "target");

shenyu-admin/src/test/java/org/apache/shenyu/admin/service/ConfigsServiceTest.java

@@ -22,6 +22,7 @@
 import org.apache.shenyu.admin.mapper.PluginMapper;
 import org.apache.shenyu.admin.mapper.RuleConditionMapper;
 import org.apache.shenyu.admin.mapper.RuleMapper;
+import org.apache.shenyu.admin.mapper.ShenyuDictMapper;
 import org.apache.shenyu.admin.mapper.SelectorMapper;
 import org.apache.shenyu.admin.model.dto.RuleDTO;
 import org.apache.shenyu.admin.model.entity.PluginDO;
@@ -170,7 +171,7 @@ public void testRuleBeforeSelectorInZipCannotBypassHandleValidation() {
         SelectorMapper selectorMapper = mock(SelectorMapper.class);
         PluginMapper pluginMapper = mock(PluginMapper.class);
         RuleService ruleServiceImpl = new RuleServiceImpl(ruleMapper, mock(RuleConditionMapper.class),
-                selectorMapper, pluginMapper, mock(RuleEventPublisher.class));
+                selectorMapper, pluginMapper, mock(RuleEventPublisher.class), mock(ShenyuDictMapper.class));
         ConfigsServiceImpl importService = new ConfigsServiceImpl(appAuthService, pluginService, namespacePluginService,
                 pluginHandleService, selectorService, ruleServiceImpl, metaDataService, shenyuDictService,
                 proxySelectorService, discoveryService, discoveryUpstreamService, Collections.emptyList());