diff --git a/api/src/main/java/com/alibaba/nacos/api/common/Constants.java b/api/src/main/java/com/alibaba/nacos/api/common/Constants.java index 509972f07f0..b8c8e3eeaca 100644 --- a/api/src/main/java/com/alibaba/nacos/api/common/Constants.java +++ b/api/src/main/java/com/alibaba/nacos/api/common/Constants.java @@ -194,8 +194,12 @@ public class Constants { public static final int WRITE_REDIRECT_CODE = 307; + public static final String NAMESPACE_ID_SPLITER = ">>"; + public static final String SERVICE_INFO_SPLITER = "@@"; + public static final String MATCH_PATTERN_SPLITER = "##"; + public static final int SERVICE_INFO_SPLIT_COUNT = 2; public static final String NULL_STRING = "null"; @@ -212,7 +216,7 @@ public class Constants { public static final String ALL_PATTERN = "*"; - public static final String FUZZY_LISTEN_PATTERN_WILDCARD = "*"; + public static final String FUZZY_WATCH_PATTERN_WILDCARD = "*"; public static final String COLON = ":"; @@ -292,6 +296,26 @@ public static class Naming { public static final String CMDB_CONTEXT_TYPE = "CMDB"; } + /** + * The constants in fuzzy watch event type directory. + */ + public static class ServiceChangedType { + + public static final String ADD_SERVICE = "ADD_SERVICE"; + + public static final String DELETE_SERVICE = "DELETE_SERVICE"; + + public static final String INSTANCE_CHANGED = "INSTANCE_CHANGED"; + + public static final String HEART_BEAT = "HEART_BEAT"; + + public static final String SERVICE_SUBSCRIBED = "SERVICE_SUBSCRIBED"; + + public static final String WATCH_INITIAL_MATCH = "WATCH_INITIAL_MATCH"; + + public static final String FINISH_WATCH_INIT = "FINISH_WATCH_INIT"; + } + /** * The constants in remote directory. */ diff --git a/api/src/main/java/com/alibaba/nacos/api/naming/NamingService.java b/api/src/main/java/com/alibaba/nacos/api/naming/NamingService.java index f10c25d3029..6edc947f578 100644 --- a/api/src/main/java/com/alibaba/nacos/api/naming/NamingService.java +++ b/api/src/main/java/com/alibaba/nacos/api/naming/NamingService.java @@ -17,6 +17,7 @@ package com.alibaba.nacos.api.naming; import com.alibaba.nacos.api.exception.NacosException; +import com.alibaba.nacos.api.naming.listener.AbstractFuzzyWatchEventListener; import com.alibaba.nacos.api.naming.listener.EventListener; import com.alibaba.nacos.api.naming.pojo.Instance; import com.alibaba.nacos.api.naming.pojo.ListView; @@ -555,6 +556,50 @@ void subscribe(String serviceName, String groupName, NamingSelector selector, Ev void unsubscribe(String serviceName, String groupName, List clusters, EventListener listener) throws NacosException; + /** + * According to matching rules, watch services within a specific scope, and receive notifications when + * changes occur in the services within the scope. + * When given a fixed group name, watch changes in all services under this group. + * + * @param fixedGroupName fixed group name for fuzzy watch + * @param listener event listener + * @throws NacosException nacos exception + */ + void fuzzyWatch(String fixedGroupName, AbstractFuzzyWatchEventListener listener) throws NacosException; + + /** + * According to matching rules, watch services within a specific scope, and receive notifications when + * changes occur in the services within the scope. + * When provided with a fixed group name and pattern of service name, watch changes in services under + * this group that match the specified pattern. + * + * @param serviceNamePattern service name pattern for fuzzy watch + * @param fixedGroupName fixed group name for fuzzy watch + * @param listener event listener + * @throws NacosException nacos exception + */ + void fuzzyWatch(String serviceNamePattern, String fixedGroupName, + AbstractFuzzyWatchEventListener listener) throws NacosException; + + /** + * Cancel fuzzy watch, and remove event listener of a pattern. + * + * @param fixedGroupName fixed group name for fuzzy watch + * @param listener event listener + * @throws NacosException nacos exception + */ + void cancelFuzzyWatch(String fixedGroupName, AbstractFuzzyWatchEventListener listener) throws NacosException; + + /** + * Cancel fuzzy watch, and remove event listener of a pattern. + * + * @param serviceNamePattern service name pattern for fuzzy watch + * @param fixedGroupName fixed group name for fuzzy watch + * @param listener event listener + * @throws NacosException nacos exception + */ + void cancelFuzzyWatch(String serviceNamePattern, String fixedGroupName, AbstractFuzzyWatchEventListener listener) throws NacosException; + /** * Unsubscribe event listener of service. * diff --git a/api/src/main/java/com/alibaba/nacos/api/naming/listener/AbstractFuzzyWatchEventListener.java b/api/src/main/java/com/alibaba/nacos/api/naming/listener/AbstractFuzzyWatchEventListener.java new file mode 100644 index 00000000000..e4c23dbe43b --- /dev/null +++ b/api/src/main/java/com/alibaba/nacos/api/naming/listener/AbstractFuzzyWatchEventListener.java @@ -0,0 +1,41 @@ +/* + * Copyright 1999-2023 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.nacos.api.naming.listener; + +import java.util.concurrent.Executor; + +/** + * Abstract fuzzy watch event listener, to support handle event by user custom executor. + * + * @author tanyongquan + */ +public abstract class AbstractFuzzyWatchEventListener implements FuzzyWatchListener { + + String uuid; + + public Executor getExecutor() { + return null; + } + + public final void setUuid(String uuid) { + this.uuid = uuid; + } + + public final String getUuid() { + return uuid; + } +} diff --git a/api/src/main/java/com/alibaba/nacos/api/naming/listener/FuzzyWatchListener.java b/api/src/main/java/com/alibaba/nacos/api/naming/listener/FuzzyWatchListener.java new file mode 100644 index 00000000000..ffd2090a799 --- /dev/null +++ b/api/src/main/java/com/alibaba/nacos/api/naming/listener/FuzzyWatchListener.java @@ -0,0 +1,32 @@ +/* + * Copyright 1999-2023 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.nacos.api.naming.listener; + +/** + * Fuzzy Watch Listener. + * + * @author tanyongquan + */ +public interface FuzzyWatchListener { + + /** + * callback event. + * + * @param event event + */ + void onEvent(FuzzyWatchNotifyEvent event); +} diff --git a/api/src/main/java/com/alibaba/nacos/api/naming/listener/FuzzyWatchNotifyEvent.java b/api/src/main/java/com/alibaba/nacos/api/naming/listener/FuzzyWatchNotifyEvent.java new file mode 100644 index 00000000000..29f07309389 --- /dev/null +++ b/api/src/main/java/com/alibaba/nacos/api/naming/listener/FuzzyWatchNotifyEvent.java @@ -0,0 +1,55 @@ +/* + * Copyright 1999-2023 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.nacos.api.naming.listener; + +import com.alibaba.nacos.api.naming.pojo.Service; + +/** + * Fuzzy Watch Notify Event. + * + * @author tanyongquan + */ +public class FuzzyWatchNotifyEvent implements Event { + + private Service service; + + private String changeType; + + public FuzzyWatchNotifyEvent() { + } + + public FuzzyWatchNotifyEvent(Service service, String changeType) { + this.service = service; + this.changeType = changeType; + } + + public Service getService() { + return service; + } + + public void setService(Service service) { + this.service = service; + } + + public String getChangeType() { + return changeType; + } + + public void setChangeType(String changeType) { + this.changeType = changeType; + } +} diff --git a/api/src/main/java/com/alibaba/nacos/api/naming/pojo/Service.java b/api/src/main/java/com/alibaba/nacos/api/naming/pojo/Service.java index 08fa6d269ca..3c800dd0bda 100644 --- a/api/src/main/java/com/alibaba/nacos/api/naming/pojo/Service.java +++ b/api/src/main/java/com/alibaba/nacos/api/naming/pojo/Service.java @@ -16,9 +16,12 @@ package com.alibaba.nacos.api.naming.pojo; +import com.alibaba.nacos.api.naming.utils.NamingUtils; + import java.io.Serializable; import java.util.HashMap; import java.util.Map; +import java.util.Objects; /** * Service of Nacos. @@ -63,6 +66,11 @@ public Service(String name) { this.name = name; } + public Service(String name, String groupName) { + this.name = name; + this.groupName = groupName; + } + public String getName() { return name; } @@ -95,6 +103,10 @@ public void setGroupName(String groupName) { this.groupName = groupName; } + public String getGroupedServiceName() { + return NamingUtils.getGroupedName(name, groupName); + } + public Map getMetadata() { return metadata; } @@ -112,4 +124,21 @@ public String toString() { return "Service{" + "name='" + name + '\'' + ", protectThreshold=" + protectThreshold + ", appName='" + appName + '\'' + ", groupName='" + groupName + '\'' + ", metadata=" + metadata + '}'; } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + Service service = (Service) o; + return name.equals(service.name) && groupName.equals(service.groupName); + } + + @Override + public int hashCode() { + return Objects.hash(name, groupName); + } } diff --git a/api/src/main/java/com/alibaba/nacos/api/naming/remote/NamingRemoteConstants.java b/api/src/main/java/com/alibaba/nacos/api/naming/remote/NamingRemoteConstants.java index 4eb9cccc30c..36ee9f64034 100644 --- a/api/src/main/java/com/alibaba/nacos/api/naming/remote/NamingRemoteConstants.java +++ b/api/src/main/java/com/alibaba/nacos/api/naming/remote/NamingRemoteConstants.java @@ -30,6 +30,10 @@ public class NamingRemoteConstants { public static final String DE_REGISTER_INSTANCE = "deregisterInstance"; + public static final String FUZZY_WATCH_SERVICE = "fuzzyWatchService"; + + public static final String CANCEL_FUZZY_WATCH_SERVICE = "cancelFuzzyWatchService"; + public static final String QUERY_SERVICE = "queryService"; public static final String SUBSCRIBE_SERVICE = "subscribeService"; diff --git a/api/src/main/java/com/alibaba/nacos/api/naming/remote/request/AbstractFuzzyWatchNotifyRequest.java b/api/src/main/java/com/alibaba/nacos/api/naming/remote/request/AbstractFuzzyWatchNotifyRequest.java new file mode 100644 index 00000000000..b1678a8001d --- /dev/null +++ b/api/src/main/java/com/alibaba/nacos/api/naming/remote/request/AbstractFuzzyWatchNotifyRequest.java @@ -0,0 +1,61 @@ +/* + * Copyright 1999-2023 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.nacos.api.naming.remote.request; + +import com.alibaba.nacos.api.remote.request.ServerRequest; + +import static com.alibaba.nacos.api.common.Constants.Naming.NAMING_MODULE; + +/** + * Abstract fuzzy watch notify request, including basic fuzzy watch notify information. + * + * @author tanyongquan + */ +public abstract class AbstractFuzzyWatchNotifyRequest extends ServerRequest { + private String namespace; + + private String serviceChangedType; + + public AbstractFuzzyWatchNotifyRequest(){ + } + + public AbstractFuzzyWatchNotifyRequest(String namespace, String serviceChangedType) { + this.namespace = namespace; + this.serviceChangedType = serviceChangedType; + } + + public String getNamespace() { + return namespace; + } + + public void setNamespace(String namespace) { + this.namespace = namespace; + } + + public String getServiceChangedType() { + return serviceChangedType; + } + + public void setServiceChangedType(String serviceChangedType) { + this.serviceChangedType = serviceChangedType; + } + + @Override + public String getModule() { + return NAMING_MODULE; + } +} diff --git a/api/src/main/java/com/alibaba/nacos/api/naming/remote/request/FuzzyWatchNotifyChangeRequest.java b/api/src/main/java/com/alibaba/nacos/api/naming/remote/request/FuzzyWatchNotifyChangeRequest.java new file mode 100644 index 00000000000..bf31ecb58cd --- /dev/null +++ b/api/src/main/java/com/alibaba/nacos/api/naming/remote/request/FuzzyWatchNotifyChangeRequest.java @@ -0,0 +1,55 @@ +/* + * Copyright 1999-2023 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.nacos.api.naming.remote.request; + +/** + * Nacos fuzzy watch notify service change request, use it when one of the services changes. + * + * @author tanyongquan + */ +public class FuzzyWatchNotifyChangeRequest extends AbstractFuzzyWatchNotifyRequest { + + String serviceName; + + String groupName; + + public FuzzyWatchNotifyChangeRequest() { + } + + public FuzzyWatchNotifyChangeRequest(String namespace, String serviceName, + String groupName, String serviceChangedType) { + super(namespace, serviceChangedType); + this.serviceName = serviceName; + this.groupName = groupName; + } + + public String getServiceName() { + return serviceName; + } + + public void setServiceName(String serviceName) { + this.serviceName = serviceName; + } + + public String getGroupName() { + return groupName; + } + + public void setGroupName(String groupName) { + this.groupName = groupName; + } +} diff --git a/api/src/main/java/com/alibaba/nacos/api/naming/remote/request/FuzzyWatchNotifyInitRequest.java b/api/src/main/java/com/alibaba/nacos/api/naming/remote/request/FuzzyWatchNotifyInitRequest.java new file mode 100644 index 00000000000..0983df802d7 --- /dev/null +++ b/api/src/main/java/com/alibaba/nacos/api/naming/remote/request/FuzzyWatchNotifyInitRequest.java @@ -0,0 +1,67 @@ +/* + * Copyright 1999-2023 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.nacos.api.naming.remote.request; + +import com.alibaba.nacos.api.common.Constants; + +import java.util.Collection; +import java.util.HashSet; + +/** + * Nacos fuzzy watch initial notify request, use it when init a watch request, push service by batch. + * + * @author tanyongquan + */ +public class FuzzyWatchNotifyInitRequest extends AbstractFuzzyWatchNotifyRequest { + + private String pattern; + + private Collection servicesName; + + public FuzzyWatchNotifyInitRequest() { + } + + private FuzzyWatchNotifyInitRequest(String namespace, String pattern, String serviceChangedType, Collection servicesName) { + super(namespace, serviceChangedType); + this.servicesName = servicesName; + this.pattern = pattern; + } + + public static FuzzyWatchNotifyInitRequest buildInitRequest(String namespace, String pattern, Collection servicesName) { + return new FuzzyWatchNotifyInitRequest(namespace, pattern, Constants.ServiceChangedType.WATCH_INITIAL_MATCH, servicesName); + } + + public static FuzzyWatchNotifyInitRequest buildInitFinishRequest(String namespace, String pattern) { + return new FuzzyWatchNotifyInitRequest(namespace, pattern, Constants.ServiceChangedType.FINISH_WATCH_INIT, new HashSet<>(1)); + } + + public String getPattern() { + return pattern; + } + + public void setPattern(String pattern) { + this.pattern = pattern; + } + + public Collection getServicesName() { + return servicesName; + } + + public void setServicesName(Collection servicesName) { + this.servicesName = servicesName; + } +} diff --git a/api/src/main/java/com/alibaba/nacos/api/naming/remote/request/FuzzyWatchRequest.java b/api/src/main/java/com/alibaba/nacos/api/naming/remote/request/FuzzyWatchRequest.java new file mode 100644 index 00000000000..dd5015abd8b --- /dev/null +++ b/api/src/main/java/com/alibaba/nacos/api/naming/remote/request/FuzzyWatchRequest.java @@ -0,0 +1,43 @@ +/* + * Copyright 1999-2023 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.nacos.api.naming.remote.request; + +/** + * Nacos naming fuzzy watch service request. + * + * @author tanyongquan + */ +public class FuzzyWatchRequest extends AbstractNamingRequest { + + private String type; + + public FuzzyWatchRequest() { + } + + public FuzzyWatchRequest(String namespace, String serviceNamePattern, String groupNamePattern, String type) { + super(namespace, serviceNamePattern, groupNamePattern); + this.type = type; + } + + public void setType(String type) { + this.type = type; + } + + public String getType() { + return this.type; + } +} diff --git a/api/src/main/java/com/alibaba/nacos/api/naming/remote/response/FuzzyWatchResponse.java b/api/src/main/java/com/alibaba/nacos/api/naming/remote/response/FuzzyWatchResponse.java new file mode 100644 index 00000000000..6abbad6ba54 --- /dev/null +++ b/api/src/main/java/com/alibaba/nacos/api/naming/remote/response/FuzzyWatchResponse.java @@ -0,0 +1,61 @@ +/* + * Copyright 1999-2023 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.nacos.api.naming.remote.response; + +import com.alibaba.nacos.api.remote.response.Response; +import com.alibaba.nacos.api.remote.response.ResponseCode; + +/** + * Nacos naming fuzzy watch service response. + * + * @author tanyongquan + */ +public class FuzzyWatchResponse extends Response { + + private String type; + + public FuzzyWatchResponse(){ + } + + public FuzzyWatchResponse(String type) { + this.type = type; + } + + public static FuzzyWatchResponse buildSuccessResponse(String type) { + return new FuzzyWatchResponse(type); + } + + /** + * Build fail response. + * + * @param message error message + * @return fail response + */ + public static FuzzyWatchResponse buildFailResponse(String message) { + FuzzyWatchResponse result = new FuzzyWatchResponse(); + result.setErrorInfo(ResponseCode.FAIL.getCode(), message); + return result; + } + + public String getType() { + return type; + } + + public void setType(String type) { + this.type = type; + } +} diff --git a/api/src/main/java/com/alibaba/nacos/api/naming/remote/response/NotifyFuzzyWatcherResponse.java b/api/src/main/java/com/alibaba/nacos/api/naming/remote/response/NotifyFuzzyWatcherResponse.java new file mode 100644 index 00000000000..ac9d55298db --- /dev/null +++ b/api/src/main/java/com/alibaba/nacos/api/naming/remote/response/NotifyFuzzyWatcherResponse.java @@ -0,0 +1,28 @@ +/* + * Copyright 1999-2023 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.nacos.api.naming.remote.response; + +import com.alibaba.nacos.api.remote.response.Response; + +/** + * Response for notify fuzzy watcher. + * + * @author tanyongquan + */ +public class NotifyFuzzyWatcherResponse extends Response { + +} diff --git a/api/src/main/java/com/alibaba/nacos/api/naming/utils/NamingUtils.java b/api/src/main/java/com/alibaba/nacos/api/naming/utils/NamingUtils.java index 074c41749e2..69176918c7d 100644 --- a/api/src/main/java/com/alibaba/nacos/api/naming/utils/NamingUtils.java +++ b/api/src/main/java/com/alibaba/nacos/api/naming/utils/NamingUtils.java @@ -23,12 +23,14 @@ import com.alibaba.nacos.api.naming.pojo.Instance; import com.alibaba.nacos.api.utils.StringUtils; +import java.util.Collection; import java.util.HashSet; import java.util.List; import java.util.Set; import java.util.regex.Pattern; import static com.alibaba.nacos.api.common.Constants.CLUSTER_NAME_PATTERN_STRING; +import static com.alibaba.nacos.api.common.Constants.FUZZY_WATCH_PATTERN_WILDCARD; import static com.alibaba.nacos.api.common.Constants.NUMBER_PATTERN_STRING; /** @@ -189,6 +191,124 @@ public static void batchCheckInstanceIsLegal(List instances) throws Na } } + public static String getPatternWithNamespace(final String namespaceId, final String groupedPattern) { + if (StringUtils.isBlank(namespaceId)) { + throw new IllegalArgumentException("Param 'namespaceId' is illegal, namespaceId is blank"); + } + if (StringUtils.isBlank(groupedPattern)) { + throw new IllegalArgumentException("Param 'groupedPattern' is illegal, groupedPattern is blank"); + } + final String resultGroupedPattern = namespaceId + Constants.NAMESPACE_ID_SPLITER + groupedPattern; + return resultGroupedPattern.intern(); + } + + public static String getNamespaceFromPattern(String completedPattern) { + if (StringUtils.isBlank(completedPattern)) { + return StringUtils.EMPTY; + } + if (!completedPattern.contains(Constants.NAMESPACE_ID_SPLITER)) { + return Constants.DEFAULT_NAMESPACE_ID; + } + return completedPattern.split(Constants.NAMESPACE_ID_SPLITER)[0]; + } + + public static String getPatternRemovedNamespace(String completedPattern) { + if (StringUtils.isBlank(completedPattern)) { + return StringUtils.EMPTY; + } + if (!completedPattern.contains(Constants.NAMESPACE_ID_SPLITER)) { + return completedPattern; + } + return completedPattern.split(Constants.NAMESPACE_ID_SPLITER)[1]; + } + + /** + * Get the pattern watched under given namespace id. + * + * @param namespaceId name space id + * @param completedPatterns a set of all watched pattern(with namespace id) + * @return filtered pattern set + */ + public static Set filterPatternWithNamespace(String namespaceId, Set completedPatterns) { + Set patternsOfGivenNamespace = new HashSet<>(); + for (String each : completedPatterns) { + String nameSpaceOfPattern = getNamespaceFromPattern(each); + if (namespaceId.equals(nameSpaceOfPattern)) { + patternsOfGivenNamespace.add(getPatternRemovedNamespace(each)); + } + } + return patternsOfGivenNamespace; + } + + /** + * Given a service, and a list of watched patterns, return the patterns that the service can match. + * + * @param serviceName service Name + * @param groupName group Name + * @param watchPattern a list of completed watch patterns + * @return the patterns list that the service can match. + */ + public static Set getServiceMatchedPatterns(String serviceName, String groupName, Collection watchPattern) { + if (watchPattern == null || watchPattern.isEmpty()) { + return new HashSet<>(1); + } + Set matchedPatternList = new HashSet<>(); + for (String eachPattern : watchPattern) { + if (isMatchPattern(serviceName, groupName, getServiceName(eachPattern), getGroupName(eachPattern))) { + matchedPatternList.add(eachPattern); + } + } + return matchedPatternList; + } + + /** + * Given a list of service's name, and a pattern to watch, return the services that can match the pattern. + * + * @param servicesList a list of service's name + * @param serviceNamePattern service name Pattern + * @param groupNamePattern group name Pattern + * @return the patterns list that the service can match. + */ + public static Set getPatternMatchedServices(Collection servicesList, String serviceNamePattern, + String groupNamePattern) { + if (servicesList == null || servicesList.isEmpty()) { + return new HashSet<>(1); + } + Set matchList = new HashSet<>(); + for (String eachService : servicesList) { + if (isMatchPattern(getServiceName(eachService), getGroupName(eachService), serviceNamePattern, groupNamePattern)) { + matchList.add(eachService); + } + } + return matchList; + } + + /** + * Given a service name and a pattern to match, determine whether it can match. + * TODOļ¼šIf want to add a matching method, can implement in here. + * + * @param serviceName service name to judge + * @param groupName group name to judge + * @param serviceNamePattern service name Pattern + * @param groupNamePattern group name Pattern + * @return matching result + */ + public static boolean isMatchPattern(String serviceName, String groupName, String serviceNamePattern, String groupNamePattern) { + // Only support prefix match or all match service name right now + // Only support fixed group name right now + if (serviceNamePattern.equals(FUZZY_WATCH_PATTERN_WILDCARD)) { + return groupName.equals(groupNamePattern); + } else if (serviceNamePattern.endsWith(FUZZY_WATCH_PATTERN_WILDCARD)) { + String serviceMatchName = serviceNamePattern.substring(0, serviceNamePattern.length() - 1); + return prefixMatchWithFixedGroupName(serviceName, serviceMatchName, groupName, groupNamePattern); + } + return false; + } + + private static boolean prefixMatchWithFixedGroupName(String serviceName, String serviceNamePrefix, String groupName, String fixedGroupName) { + return groupName.equals(fixedGroupName) && serviceName.startsWith(serviceNamePrefix); + } + /** * Check string is a number or not. * diff --git a/api/src/main/resources/META-INF/services/com.alibaba.nacos.api.remote.Payload b/api/src/main/resources/META-INF/services/com.alibaba.nacos.api.remote.Payload index eba07c73d27..3b4927ec774 100644 --- a/api/src/main/resources/META-INF/services/com.alibaba.nacos.api.remote.Payload +++ b/api/src/main/resources/META-INF/services/com.alibaba.nacos.api.remote.Payload @@ -49,15 +49,20 @@ com.alibaba.nacos.api.naming.remote.request.BatchInstanceRequest com.alibaba.nacos.api.naming.remote.request.InstanceRequest com.alibaba.nacos.api.naming.remote.request.PersistentInstanceRequest com.alibaba.nacos.api.naming.remote.request.NotifySubscriberRequest +com.alibaba.nacos.api.naming.remote.request.FuzzyWatchNotifyChangeRequest +com.alibaba.nacos.api.naming.remote.request.FuzzyWatchNotifyInitRequest com.alibaba.nacos.api.naming.remote.request.ServiceListRequest com.alibaba.nacos.api.naming.remote.request.ServiceQueryRequest com.alibaba.nacos.api.naming.remote.request.SubscribeServiceRequest +com.alibaba.nacos.api.naming.remote.request.FuzzyWatchRequest com.alibaba.nacos.api.naming.remote.response.BatchInstanceResponse com.alibaba.nacos.api.naming.remote.response.InstanceResponse com.alibaba.nacos.api.naming.remote.response.NotifySubscriberResponse +com.alibaba.nacos.api.naming.remote.response.NotifyFuzzyWatcherResponse com.alibaba.nacos.api.naming.remote.response.QueryServiceResponse com.alibaba.nacos.api.naming.remote.response.ServiceListResponse com.alibaba.nacos.api.naming.remote.response.SubscribeServiceResponse +com.alibaba.nacos.api.naming.remote.response.FuzzyWatchResponse com.alibaba.nacos.api.config.remote.request.ConfigBatchFuzzyListenRequest com.alibaba.nacos.api.config.remote.response.ConfigBatchFuzzyListenResponse com.alibaba.nacos.api.config.remote.request.FuzzyListenNotifyChangeRequest diff --git a/client/src/main/java/com/alibaba/nacos/client/naming/NacosNamingService.java b/client/src/main/java/com/alibaba/nacos/client/naming/NacosNamingService.java index bf382f02e27..3125502b114 100644 --- a/client/src/main/java/com/alibaba/nacos/client/naming/NacosNamingService.java +++ b/client/src/main/java/com/alibaba/nacos/client/naming/NacosNamingService.java @@ -20,6 +20,7 @@ import com.alibaba.nacos.api.common.Constants; import com.alibaba.nacos.api.exception.NacosException; import com.alibaba.nacos.api.naming.NamingService; +import com.alibaba.nacos.api.naming.listener.AbstractFuzzyWatchEventListener; import com.alibaba.nacos.api.naming.listener.EventListener; import com.alibaba.nacos.api.naming.pojo.Instance; import com.alibaba.nacos.api.naming.pojo.ListView; @@ -31,9 +32,12 @@ import com.alibaba.nacos.api.selector.AbstractSelector; import com.alibaba.nacos.client.env.NacosClientProperties; import com.alibaba.nacos.client.naming.cache.ServiceInfoHolder; +import com.alibaba.nacos.client.naming.cache.FuzzyWatchServiceListHolder; import com.alibaba.nacos.client.naming.core.Balancer; import com.alibaba.nacos.client.naming.event.InstancesChangeEvent; import com.alibaba.nacos.client.naming.event.InstancesChangeNotifier; +import com.alibaba.nacos.client.naming.event.FuzzyWatchNotifyEvent; +import com.alibaba.nacos.client.naming.event.ServicesChangeNotifier; import com.alibaba.nacos.client.naming.event.InstancesDiff; import com.alibaba.nacos.client.naming.remote.NamingClientProxy; import com.alibaba.nacos.client.naming.remote.NamingClientProxyDelegate; @@ -56,6 +60,7 @@ import java.util.Properties; import java.util.UUID; +import static com.alibaba.nacos.api.common.Constants.FUZZY_WATCH_PATTERN_WILDCARD; import static com.alibaba.nacos.client.naming.selector.NamingSelectorFactory.getUniqueClusterString; import static com.alibaba.nacos.client.utils.LogUtils.NAMING_LOGGER; @@ -83,8 +88,12 @@ public class NacosNamingService implements NamingService { private ServiceInfoHolder serviceInfoHolder; + private FuzzyWatchServiceListHolder fuzzyWatchServiceListHolder; + private InstancesChangeNotifier changeNotifier; + private ServicesChangeNotifier servicesChangeNotifier; + private NamingClientProxy clientProxy; private String notifierEventScope; @@ -113,9 +122,14 @@ private void init(Properties properties) throws NacosException { this.changeNotifier = new InstancesChangeNotifier(this.notifierEventScope); NotifyCenter.registerToPublisher(InstancesChangeEvent.class, 16384); NotifyCenter.registerSubscriber(changeNotifier); + this.servicesChangeNotifier = new ServicesChangeNotifier(this.notifierEventScope); + NotifyCenter.registerToPublisher(FuzzyWatchNotifyEvent.class, 16384); + NotifyCenter.registerSubscriber(servicesChangeNotifier); + this.serviceInfoHolder = new ServiceInfoHolder(namespace, this.notifierEventScope, nacosClientProperties); - this.clientProxy = new NamingClientProxyDelegate(this.namespace, serviceInfoHolder, nacosClientProperties, - changeNotifier); + this.fuzzyWatchServiceListHolder = new FuzzyWatchServiceListHolder(this.notifierEventScope, nacosClientProperties); + this.clientProxy = new NamingClientProxyDelegate(this.namespace, serviceInfoHolder, fuzzyWatchServiceListHolder, + nacosClientProperties, changeNotifier); } @Deprecated @@ -525,6 +539,58 @@ private void doUnsubscribe(String serviceName, String groupName, NamingSelector } } + @Override + public void fuzzyWatch(String fixedGroupName, AbstractFuzzyWatchEventListener listener) throws NacosException { + doFuzzyWatch(FUZZY_WATCH_PATTERN_WILDCARD, fixedGroupName, listener); + } + + @Override + public void fuzzyWatch(String serviceNamePattern, String fixedGroupName, + AbstractFuzzyWatchEventListener listener) throws NacosException { + // only support prefix match right now + if (!serviceNamePattern.endsWith(FUZZY_WATCH_PATTERN_WILDCARD)) { + if (serviceNamePattern.startsWith(FUZZY_WATCH_PATTERN_WILDCARD)) { + throw new UnsupportedOperationException("Suffix matching for service names is not supported yet." + + " It will be supported in future updates if needed."); + } else { + throw new UnsupportedOperationException("Illegal service name pattern, please read the documentation and pass a valid pattern."); + } + } + doFuzzyWatch(serviceNamePattern, fixedGroupName, listener); + } + + private void doFuzzyWatch(String serviceNamePattern, String groupNamePattern, + AbstractFuzzyWatchEventListener listener) throws NacosException { + if (null == listener) { + return; + } + String uuid = UUID.randomUUID().toString(); + listener.setUuid(uuid); + servicesChangeNotifier.registerFuzzyWatchListener(serviceNamePattern, groupNamePattern, listener); + clientProxy.fuzzyWatch(serviceNamePattern, groupNamePattern, uuid); + } + + @Override + public void cancelFuzzyWatch(String fixedGroupName, AbstractFuzzyWatchEventListener listener) throws NacosException { + doCancelFuzzyWatch(FUZZY_WATCH_PATTERN_WILDCARD, fixedGroupName, listener); + } + + @Override + public void cancelFuzzyWatch(String serviceNamePattern, String fixedGroupName, AbstractFuzzyWatchEventListener listener) throws NacosException { + doCancelFuzzyWatch(serviceNamePattern, fixedGroupName, listener); + } + + private void doCancelFuzzyWatch(String serviceNamePattern, String groupNamePattern, + AbstractFuzzyWatchEventListener listener) throws NacosException { + if (null == listener) { + return; + } + servicesChangeNotifier.deregisterFuzzyWatchListener(serviceNamePattern, groupNamePattern, listener); + if (!servicesChangeNotifier.isWatched(serviceNamePattern, groupNamePattern)) { + clientProxy.cancelFuzzyWatch(serviceNamePattern, groupNamePattern); + } + } + @Override public ListView getServicesOfServer(int pageNo, int pageSize) throws NacosException { return getServicesOfServer(pageNo, pageSize, Constants.DEFAULT_GROUP); diff --git a/client/src/main/java/com/alibaba/nacos/client/naming/cache/FuzzyWatchServiceListHolder.java b/client/src/main/java/com/alibaba/nacos/client/naming/cache/FuzzyWatchServiceListHolder.java new file mode 100644 index 00000000000..dfd0e2bd240 --- /dev/null +++ b/client/src/main/java/com/alibaba/nacos/client/naming/cache/FuzzyWatchServiceListHolder.java @@ -0,0 +1,142 @@ +/* + * Copyright 1999-2023 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.nacos.client.naming.cache; + +import com.alibaba.nacos.api.common.Constants; +import com.alibaba.nacos.api.naming.pojo.Service; +import com.alibaba.nacos.api.naming.remote.request.AbstractFuzzyWatchNotifyRequest; +import com.alibaba.nacos.api.naming.remote.request.FuzzyWatchNotifyChangeRequest; +import com.alibaba.nacos.api.naming.remote.request.FuzzyWatchNotifyInitRequest; +import com.alibaba.nacos.api.naming.utils.NamingUtils; +import com.alibaba.nacos.client.env.NacosClientProperties; +import com.alibaba.nacos.client.naming.event.FuzzyWatchNotifyEvent; +import com.alibaba.nacos.client.naming.utils.CollectionUtils; +import com.alibaba.nacos.common.notify.NotifyCenter; +import com.alibaba.nacos.common.utils.ConcurrentHashSet; + +import java.util.Collection; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Naming client fuzzy watch service list holder. + * + * @author tanyongquan + */ +public class FuzzyWatchServiceListHolder { + + private String notifierEventScope; + + /** + * The contents of {@code patternMatchMap} are Map{pattern -> Set[matched services]}. + */ + private Map> patternMatchMap = new ConcurrentHashMap<>(); + + public FuzzyWatchServiceListHolder(String notifierEventScope, NacosClientProperties properties) { + this.notifierEventScope = notifierEventScope; + } + + /** + * Publish service change event notify from nacos server about fuzzy watch. + * + * @param request watch notify request from Nacos server + */ + public void processFuzzyWatchNotify(AbstractFuzzyWatchNotifyRequest request) { + if (request instanceof FuzzyWatchNotifyInitRequest) { + FuzzyWatchNotifyInitRequest watchNotifyInitRequest = (FuzzyWatchNotifyInitRequest) request; + Set matchedServiceSet = patternMatchMap.computeIfAbsent(watchNotifyInitRequest.getPattern(), + keyInner -> new ConcurrentHashSet<>()); + Collection servicesName = watchNotifyInitRequest.getServicesName(); + for (String groupedName : servicesName) { + Service service = new Service(NamingUtils.getServiceName(groupedName), NamingUtils.getGroupName(groupedName)); + // may have a 'change event' sent to client before 'init event' + if (matchedServiceSet.add(service)) { + NotifyCenter.publishEvent(FuzzyWatchNotifyEvent.buildNotifyPatternAllListenersEvent(notifierEventScope, + service, watchNotifyInitRequest.getPattern(), Constants.ServiceChangedType.ADD_SERVICE)); + } + } + } else if (request instanceof FuzzyWatchNotifyChangeRequest) { + FuzzyWatchNotifyChangeRequest notifyChangeRequest = (FuzzyWatchNotifyChangeRequest) request; + Collection matchedPattern = NamingUtils.getServiceMatchedPatterns(notifyChangeRequest.getServiceName(), + notifyChangeRequest.getGroupName(), patternMatchMap.keySet()); + Service service = new Service(notifyChangeRequest.getServiceName(), notifyChangeRequest.getGroupName()); + String serviceChangeType = request.getServiceChangedType(); + + switch (serviceChangeType) { + case Constants.ServiceChangedType.ADD_SERVICE: + case Constants.ServiceChangedType.INSTANCE_CHANGED: + for (String pattern : matchedPattern) { + Set matchedServiceSet = patternMatchMap.get(pattern); + if (matchedServiceSet != null && matchedServiceSet.add(service)) { + NotifyCenter.publishEvent( + FuzzyWatchNotifyEvent.buildNotifyPatternAllListenersEvent(notifierEventScope, + service, pattern, Constants.ServiceChangedType.ADD_SERVICE)); + } + } + break; + case Constants.ServiceChangedType.DELETE_SERVICE: + for (String pattern : matchedPattern) { + Set matchedServiceSet = patternMatchMap.get(pattern); + if (matchedServiceSet != null && matchedServiceSet.remove(service)) { + NotifyCenter.publishEvent( + FuzzyWatchNotifyEvent.buildNotifyPatternAllListenersEvent(notifierEventScope, + service, pattern, Constants.ServiceChangedType.DELETE_SERVICE)); + } + } + break; + default: + break; + } + } + } + + /** + * For a duplicate fuzzy watch of a certain pattern, initiate an initialization event to the corresponding Listener. + * + * @param serviceNamePattern service name pattern. + * @param groupNamePattern group name pattern. + * @param uuid The UUID that identifies the Listener. + */ + public void duplicateFuzzyWatchInit(String serviceNamePattern, String groupNamePattern, String uuid) { + String pattern = NamingUtils.getGroupedName(serviceNamePattern, groupNamePattern); + Collection cacheServices = patternMatchMap.get(pattern); + if (cacheServices == null) { + return; + } + for (Service service : cacheServices) { + NotifyCenter.publishEvent( + FuzzyWatchNotifyEvent.buildNotifyPatternSpecificListenerEvent(notifierEventScope, service, + pattern, uuid, Constants.ServiceChangedType.ADD_SERVICE)); + } + } + + public boolean containsPatternMatchCache(String serviceNamePattern, String groupNamePattern) { + String pattern = NamingUtils.getGroupedName(serviceNamePattern, groupNamePattern); + return CollectionUtils.isEmpty(patternMatchMap.get(pattern)); + } + + public void removePatternMatchCache(String serviceNamePattern, String groupNamePattern) { + String pattern = NamingUtils.getGroupedName(serviceNamePattern, groupNamePattern); + patternMatchMap.remove(pattern); + } + + public void addPatternMatchCache(String serviceNamePattern, String groupNamePattern) { + String pattern = NamingUtils.getGroupedName(serviceNamePattern, groupNamePattern); + patternMatchMap.computeIfAbsent(pattern, keyInner -> new ConcurrentHashSet<>()); + } +} diff --git a/client/src/main/java/com/alibaba/nacos/client/naming/event/FuzzyWatchNotifyEvent.java b/client/src/main/java/com/alibaba/nacos/client/naming/event/FuzzyWatchNotifyEvent.java new file mode 100644 index 00000000000..4f39dc30645 --- /dev/null +++ b/client/src/main/java/com/alibaba/nacos/client/naming/event/FuzzyWatchNotifyEvent.java @@ -0,0 +1,85 @@ +/* + * Copyright 1999-2023 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.nacos.client.naming.event; + +import com.alibaba.nacos.api.naming.pojo.Service; +import com.alibaba.nacos.common.notify.Event; + +/** + * Watch notify event, including service change/watch initial. + * + * @author tanyongquan + */ +public class FuzzyWatchNotifyEvent extends Event { + + private final String eventScope; + + private final Service changedService; + + private String pattern; + + private String uuid; + + private final String serviceChangedType; + + private FuzzyWatchNotifyEvent(String eventScope, Service changedService, String pattern, String uuid, String serviceChangedType) { + this(eventScope, changedService, pattern, serviceChangedType); + this.uuid = uuid; + } + + private FuzzyWatchNotifyEvent(String eventScope, Service changedService, String pattern, String serviceChangedType) { + this.eventScope = eventScope; + this.changedService = changedService; + this.serviceChangedType = serviceChangedType; + this.pattern = pattern; + } + + public static FuzzyWatchNotifyEvent buildNotifyPatternSpecificListenerEvent(String eventScope, Service changedService, + String pattern, String uuid, String serviceChangedType) { + return new FuzzyWatchNotifyEvent(eventScope, changedService, pattern, uuid, serviceChangedType); + } + + public static FuzzyWatchNotifyEvent buildNotifyPatternAllListenersEvent(String eventScope, Service changedService, + String pattern, String serviceChangedType) { + return new FuzzyWatchNotifyEvent(eventScope, changedService, pattern, serviceChangedType); + } + + public Service getChangedService() { + return changedService; + } + + public String getPattern() { + return pattern; + } + + public void setPattern(String pattern) { + this.pattern = pattern; + } + + public String getUuid() { + return uuid; + } + + public String getServiceChangedType() { + return serviceChangedType; + } + + @Override + public String scope() { + return this.eventScope; + } +} diff --git a/client/src/main/java/com/alibaba/nacos/client/naming/event/ServicesChangeNotifier.java b/client/src/main/java/com/alibaba/nacos/client/naming/event/ServicesChangeNotifier.java new file mode 100644 index 00000000000..26bc5741387 --- /dev/null +++ b/client/src/main/java/com/alibaba/nacos/client/naming/event/ServicesChangeNotifier.java @@ -0,0 +1,147 @@ +/* + * Copyright 1999-2023 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.nacos.client.naming.event; + +import com.alibaba.nacos.api.naming.listener.AbstractFuzzyWatchEventListener; +import com.alibaba.nacos.api.naming.utils.NamingUtils; +import com.alibaba.nacos.common.JustForTest; +import com.alibaba.nacos.common.notify.Event; +import com.alibaba.nacos.common.notify.listener.Subscriber; +import com.alibaba.nacos.common.utils.CollectionUtils; +import com.alibaba.nacos.common.utils.ConcurrentHashSet; +import com.alibaba.nacos.common.utils.StringUtils; + +import java.util.Collection; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; + +/** + * A watcher to notify service change event Listener callback. + * + * @author tanyongquan + */ +public class ServicesChangeNotifier extends Subscriber { + + private final String eventScope; + + @JustForTest + public ServicesChangeNotifier() { + this.eventScope = UUID.randomUUID().toString(); + } + + public ServicesChangeNotifier(String eventScope) { + this.eventScope = eventScope; + } + + /** + * The content of map is {pattern -> Set[Listener]}. + */ + private final Map> fuzzyWatchListenerMap = new ConcurrentHashMap<>(); + + /** register fuzzy watch listener. + * This listener responds to changes of the services (not the instance's change). + * + * @param serviceNamePattern service name pattern + * @param groupNamePattern group name pattern + * @param listener custom listener + */ + public void registerFuzzyWatchListener(String serviceNamePattern, String groupNamePattern, AbstractFuzzyWatchEventListener listener) { + String key = NamingUtils.getGroupedName(serviceNamePattern, groupNamePattern); + Set eventListeners = fuzzyWatchListenerMap.computeIfAbsent(key, keyInner -> new ConcurrentHashSet<>()); + eventListeners.add(listener); + } + + /** remove fuzzy watch listener. + * + * @param serviceNamePattern service name pattern + * @param groupNamePattern group name pattern + */ + public void deregisterFuzzyWatchListener(String serviceNamePattern, String groupNamePattern, AbstractFuzzyWatchEventListener listener) { + String key = NamingUtils.getGroupedName(serviceNamePattern, groupNamePattern); + ConcurrentHashSet eventListeners = fuzzyWatchListenerMap.get(key); + if (eventListeners == null) { + return; + } + eventListeners.remove(listener); + if (CollectionUtils.isEmpty(eventListeners)) { + fuzzyWatchListenerMap.remove(key); + } + } + + /** + * check pattern is watched. + * + * @param serviceNamePattern service name pattern + * @param groupNamePattern group name pattern + * @return is pattern watched + */ + public boolean isWatched(String serviceNamePattern, String groupNamePattern) { + String key = NamingUtils.getGroupedName(serviceNamePattern, groupNamePattern); + ConcurrentHashSet eventListeners = fuzzyWatchListenerMap.get(key); + return CollectionUtils.isNotEmpty(eventListeners); + } + + /** + * receive fuzzy watch notify (fuzzy watch init or service change) from nacos server, notify all listener watch this pattern. + * If the event contains a UUID, then the event is used to notify the specified Listener when there are + * multiple watches for a particular pattern. + * + * @param event watch notify event + */ + @Override + public void onEvent(FuzzyWatchNotifyEvent event) { + String uuid = event.getUuid(); + Collection listeners = fuzzyWatchListenerMap.get(event.getPattern()); + final com.alibaba.nacos.api.naming.listener.FuzzyWatchNotifyEvent fuzzyWatchNotifyEvent = transferToWatchNotifyEvent(event); + for (AbstractFuzzyWatchEventListener each : listeners) { + // notify all listener watch this pattern + if (StringUtils.isEmpty(uuid)) { + if (each.getExecutor() != null) { + each.getExecutor().execute(() -> each.onEvent(fuzzyWatchNotifyEvent)); + } else { + each.onEvent(fuzzyWatchNotifyEvent); + } + } else if (uuid.equals(each.getUuid())) { + // notify specific listener by uuid, use in duplicate watch a same pattern + if (each.getExecutor() != null) { + each.getExecutor().execute(() -> each.onEvent(fuzzyWatchNotifyEvent)); + } else { + each.onEvent(fuzzyWatchNotifyEvent); + } + return; + } + } + } + + private com.alibaba.nacos.api.naming.listener.FuzzyWatchNotifyEvent transferToWatchNotifyEvent( + FuzzyWatchNotifyEvent fuzzyWatchNotifyEvent) { + return new com.alibaba.nacos.api.naming.listener.FuzzyWatchNotifyEvent(fuzzyWatchNotifyEvent.getChangedService(), + fuzzyWatchNotifyEvent.getServiceChangedType()); + } + + @Override + public Class subscribeType() { + return FuzzyWatchNotifyEvent.class; + } + + @Override + public boolean scopeMatches(FuzzyWatchNotifyEvent event) { + return this.eventScope.equals(event.scope()); + } +} diff --git a/client/src/main/java/com/alibaba/nacos/client/naming/remote/NamingClientProxy.java b/client/src/main/java/com/alibaba/nacos/client/naming/remote/NamingClientProxy.java index 6fc40c35c91..924f45f937e 100644 --- a/client/src/main/java/com/alibaba/nacos/client/naming/remote/NamingClientProxy.java +++ b/client/src/main/java/com/alibaba/nacos/client/naming/remote/NamingClientProxy.java @@ -181,6 +181,33 @@ ListView getServiceList(int pageNo, int pageSize, String groupName, Abst */ boolean isSubscribed(String serviceName, String groupName, String clusters) throws NacosException; + /** + * Fuzzy watch services change by pattern. + * + * @param serviceNamePattern service name pattern + * @param groupNamePattern group name pattern + * @param uuid UUID used to identify the Listener. Used for local initialization when repeating fuzzy watch + * @throws NacosException nacos exception + */ + void fuzzyWatch(String serviceNamePattern, String groupNamePattern, String uuid) throws NacosException; + + /** Judge whether pattern has been watched. + * + * @param serviceNamePattern service name pattern + * @param groupNamePattern group name pattern + * @return {@code true} if subscribed, otherwise {@code false} + * @throws NacosException nacos exception + */ + boolean isFuzzyWatched(String serviceNamePattern, String groupNamePattern) throws NacosException; + + /** Cancel fuzzy watch pattern. + * + * @param serviceNamePattern service name pattern + * @param groupNamePattern group name pattern + * @throws NacosException nacos exception + */ + void cancelFuzzyWatch(String serviceNamePattern, String groupNamePattern) throws NacosException; + /** * Check Server healthy. * diff --git a/client/src/main/java/com/alibaba/nacos/client/naming/remote/NamingClientProxyDelegate.java b/client/src/main/java/com/alibaba/nacos/client/naming/remote/NamingClientProxyDelegate.java index efd7bdabf46..408babbca72 100644 --- a/client/src/main/java/com/alibaba/nacos/client/naming/remote/NamingClientProxyDelegate.java +++ b/client/src/main/java/com/alibaba/nacos/client/naming/remote/NamingClientProxyDelegate.java @@ -26,6 +26,8 @@ import com.alibaba.nacos.api.selector.AbstractSelector; import com.alibaba.nacos.client.env.NacosClientProperties; import com.alibaba.nacos.client.naming.cache.ServiceInfoHolder; +import com.alibaba.nacos.client.naming.cache.FuzzyWatchServiceListHolder; +import com.alibaba.nacos.client.naming.core.ServerListManager; import com.alibaba.nacos.client.naming.core.NamingServerListManager; import com.alibaba.nacos.client.naming.core.ServiceInfoUpdateService; import com.alibaba.nacos.client.naming.event.InstancesChangeNotifier; @@ -58,6 +60,8 @@ public class NamingClientProxyDelegate implements NamingClientProxy { private final ServiceInfoUpdateService serviceInfoUpdateService; private final ServiceInfoHolder serviceInfoHolder; + + private final FuzzyWatchServiceListHolder fuzzyWatchServiceListHolder; private final NamingHttpClientProxy httpClientProxy; @@ -67,19 +71,20 @@ public class NamingClientProxyDelegate implements NamingClientProxy { private ScheduledExecutorService executorService; - public NamingClientProxyDelegate(String namespace, ServiceInfoHolder serviceInfoHolder, + public NamingClientProxyDelegate(String namespace, ServiceInfoHolder serviceInfoHolder, FuzzyWatchServiceListHolder fuzzyWatchServiceListHolder, NacosClientProperties properties, InstancesChangeNotifier changeNotifier) throws NacosException { this.serviceInfoUpdateService = new ServiceInfoUpdateService(properties, serviceInfoHolder, this, changeNotifier); this.serverListManager = new NamingServerListManager(properties, namespace); this.serverListManager.start(); this.serviceInfoHolder = serviceInfoHolder; + this.fuzzyWatchServiceListHolder = fuzzyWatchServiceListHolder; this.securityProxy = new SecurityProxy(this.serverListManager, NamingHttpClientManager.getInstance().getNacosRestTemplate()); initSecurityProxy(properties); this.httpClientProxy = new NamingHttpClientProxy(namespace, securityProxy, serverListManager, properties); this.grpcClientProxy = new NamingGrpcClientProxy(namespace, securityProxy, serverListManager, properties, - serviceInfoHolder); + serviceInfoHolder, fuzzyWatchServiceListHolder); } private void initSecurityProxy(NacosClientProperties properties) { @@ -187,6 +192,31 @@ public boolean isSubscribed(String serviceName, String groupName, String cluster return grpcClientProxy.isSubscribed(serviceName, groupName, clusters); } + @Override + public void fuzzyWatch(String serviceNamePattern, String groupNamePattern, String uuid) throws NacosException { + NAMING_LOGGER.info("[FUZZY-WATCH] serviceNamePattern:{}, groupNamePattern:{}", serviceNamePattern, groupNamePattern); + if (!fuzzyWatchServiceListHolder.containsPatternMatchCache(serviceNamePattern, groupNamePattern) + || !isFuzzyWatched(serviceNamePattern, groupNamePattern)) { + fuzzyWatchServiceListHolder.addPatternMatchCache(serviceNamePattern, groupNamePattern); + grpcClientProxy.fuzzyWatch(serviceNamePattern, groupNamePattern, ""); + } else { + fuzzyWatchServiceListHolder.duplicateFuzzyWatchInit(serviceNamePattern, groupNamePattern, uuid); + } + } + + @Override + public boolean isFuzzyWatched(String serviceNamePattern, String groupNamePattern) { + return grpcClientProxy.isFuzzyWatched(serviceNamePattern, groupNamePattern); + } + + @Override + public void cancelFuzzyWatch(String serviceNamePattern, String groupNamePattern) throws NacosException { + NAMING_LOGGER + .debug("[CANCEL-FUZZY-WATCH] serviceNamePattern:{}, groupNamePattern:{} ", serviceNamePattern, groupNamePattern); + fuzzyWatchServiceListHolder.removePatternMatchCache(serviceNamePattern, groupNamePattern); + grpcClientProxy.cancelFuzzyWatch(serviceNamePattern, groupNamePattern); + } + @Override public boolean serverHealthy() { return grpcClientProxy.serverHealthy() || httpClientProxy.serverHealthy(); diff --git a/client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/NamingGrpcClientProxy.java b/client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/NamingGrpcClientProxy.java index 508b76f0ab3..3a301406898 100644 --- a/client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/NamingGrpcClientProxy.java +++ b/client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/NamingGrpcClientProxy.java @@ -33,10 +33,12 @@ import com.alibaba.nacos.api.naming.remote.request.ServiceListRequest; import com.alibaba.nacos.api.naming.remote.request.ServiceQueryRequest; import com.alibaba.nacos.api.naming.remote.request.SubscribeServiceRequest; +import com.alibaba.nacos.api.naming.remote.request.FuzzyWatchRequest; import com.alibaba.nacos.api.naming.remote.response.BatchInstanceResponse; import com.alibaba.nacos.api.naming.remote.response.QueryServiceResponse; import com.alibaba.nacos.api.naming.remote.response.ServiceListResponse; import com.alibaba.nacos.api.naming.remote.response.SubscribeServiceResponse; +import com.alibaba.nacos.api.naming.remote.response.FuzzyWatchResponse; import com.alibaba.nacos.api.naming.utils.NamingUtils; import com.alibaba.nacos.api.remote.RemoteConstants; import com.alibaba.nacos.api.remote.response.Response; @@ -46,6 +48,8 @@ import com.alibaba.nacos.client.env.NacosClientProperties; import com.alibaba.nacos.client.monitor.MetricsMonitor; import com.alibaba.nacos.client.naming.cache.ServiceInfoHolder; +import com.alibaba.nacos.client.naming.cache.FuzzyWatchServiceListHolder; +import com.alibaba.nacos.client.naming.event.ServerListChangedEvent; import com.alibaba.nacos.client.address.ServerListChangeEvent; import com.alibaba.nacos.client.naming.remote.AbstractNamingClientProxy; import com.alibaba.nacos.client.naming.remote.gprc.redo.NamingGrpcRedoService; @@ -94,7 +98,8 @@ public class NamingGrpcClientProxy extends AbstractNamingClientProxy { private final NamingGrpcRedoService redoService; public NamingGrpcClientProxy(String namespaceId, SecurityProxy securityProxy, ServerListFactory serverListFactory, - NacosClientProperties properties, ServiceInfoHolder serviceInfoHolder) throws NacosException { + NacosClientProperties properties, ServiceInfoHolder serviceInfoHolder, + FuzzyWatchServiceListHolder fuzzyWatchServiceListHolder) throws NacosException { super(securityProxy); this.namespaceId = namespaceId; this.uuid = UUID.randomUUID().toString(); @@ -107,13 +112,15 @@ public NamingGrpcClientProxy(String namespaceId, SecurityProxy securityProxy, Se RpcClientTlsConfigFactory.getInstance().createSdkConfig(properties.asProperties())); this.redoService = new NamingGrpcRedoService(this, properties); NAMING_LOGGER.info("Create naming rpc client for uuid->{}", uuid); - start(serverListFactory, serviceInfoHolder); + start(serverListFactory, serviceInfoHolder, fuzzyWatchServiceListHolder); } - private void start(ServerListFactory serverListFactory, ServiceInfoHolder serviceInfoHolder) throws NacosException { + private void start(ServerListFactory serverListFactory, ServiceInfoHolder serviceInfoHolder, + FuzzyWatchServiceListHolder fuzzyWatchServiceListHolder) throws NacosException { rpcClient.serverListFactory(serverListFactory); rpcClient.registerConnectionListener(redoService); - rpcClient.registerServerRequestHandler(new NamingPushRequestHandler(serviceInfoHolder)); + rpcClient.registerServerRequestHandler(new NamingPushRequestHandler(serviceInfoHolder, + fuzzyWatchServiceListHolder)); rpcClient.start(); NotifyCenter.registerSubscriber(this); } @@ -423,6 +430,58 @@ public void doUnsubscribe(String serviceName, String groupName, String clusters) redoService.removeSubscriberForRedo(serviceName, groupName, clusters); } + @Override + public void fuzzyWatch(String serviceNamePattern, String groupNamePattern, String watcherUuid) throws NacosException { + if (NAMING_LOGGER.isDebugEnabled()) { + NAMING_LOGGER.debug("[GRPC-FUZZY-WATCH] servicePattern:{}, groupPattern:{}", serviceNamePattern, groupNamePattern); + } + redoService.cacheFuzzyWatcherForRedo(serviceNamePattern, groupNamePattern); + doFuzzyWatch(serviceNamePattern, groupNamePattern); + } + + /** + * Execute fuzzy watch operation. + * + * @param serviceNamePattern service name pattern + * @param groupNamePattern group name pattern + * @throws NacosException nacos exception + */ + public void doFuzzyWatch(String serviceNamePattern, String groupNamePattern) throws NacosException { + FuzzyWatchRequest request = new FuzzyWatchRequest(namespaceId, serviceNamePattern, groupNamePattern, + NamingRemoteConstants.FUZZY_WATCH_SERVICE); + requestToServer(request, FuzzyWatchResponse.class); + redoService.fuzzyWatcherRegistered(serviceNamePattern, groupNamePattern); + } + + @Override + public boolean isFuzzyWatched(String serviceNamePattern, String groupNamePattern) { + return redoService.isFuzzyWatcherRegistered(serviceNamePattern, groupNamePattern); + } + + @Override + public void cancelFuzzyWatch(String serviceNamePattern, String groupNamePattern) throws NacosException { + if (NAMING_LOGGER.isDebugEnabled()) { + NAMING_LOGGER + .debug("[GRPC-CANCEL-FUZZY-WATCH] serviceNamePattern:{}, groupNamePattern:{}", serviceNamePattern, groupNamePattern); + } + redoService.fuzzyWatcherDeregister(serviceNamePattern, groupNamePattern); + doCancelFuzzyWatch(serviceNamePattern, groupNamePattern); + } + + /** + * Execute cancel fuzzy watch operation. + * + * @param serviceNamePattern service name pattern + * @param groupNamePattern group name pattern + * @throws NacosException nacos exception + */ + public void doCancelFuzzyWatch(String serviceNamePattern, String groupNamePattern) throws NacosException { + FuzzyWatchRequest request = new FuzzyWatchRequest(namespaceId, serviceNamePattern, groupNamePattern, + NamingRemoteConstants.CANCEL_FUZZY_WATCH_SERVICE); + requestToServer(request, FuzzyWatchResponse.class); + redoService.removeFuzzyWatcherForRedo(serviceNamePattern, groupNamePattern); + } + @Override public boolean serverHealthy() { return rpcClient.isRunning(); @@ -508,4 +567,8 @@ private void shutDownAndRemove(String uuid) { public boolean isEnable() { return rpcClient.isRunning(); } + + public String getNamespaceId() { + return namespaceId; + } } diff --git a/client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/NamingPushRequestHandler.java b/client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/NamingPushRequestHandler.java index 83c828c802d..4448158890b 100644 --- a/client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/NamingPushRequestHandler.java +++ b/client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/NamingPushRequestHandler.java @@ -16,11 +16,14 @@ package com.alibaba.nacos.client.naming.remote.gprc; +import com.alibaba.nacos.api.naming.remote.request.AbstractFuzzyWatchNotifyRequest; import com.alibaba.nacos.api.naming.remote.request.NotifySubscriberRequest; import com.alibaba.nacos.api.naming.remote.response.NotifySubscriberResponse; +import com.alibaba.nacos.api.naming.remote.response.NotifyFuzzyWatcherResponse; import com.alibaba.nacos.api.remote.request.Request; import com.alibaba.nacos.api.remote.response.Response; import com.alibaba.nacos.client.naming.cache.ServiceInfoHolder; +import com.alibaba.nacos.client.naming.cache.FuzzyWatchServiceListHolder; import com.alibaba.nacos.common.remote.client.Connection; import com.alibaba.nacos.common.remote.client.ServerRequestHandler; @@ -33,8 +36,11 @@ public class NamingPushRequestHandler implements ServerRequestHandler { private final ServiceInfoHolder serviceInfoHolder; - public NamingPushRequestHandler(ServiceInfoHolder serviceInfoHolder) { + private final FuzzyWatchServiceListHolder fuzzyWatchServiceListHolder; + + public NamingPushRequestHandler(ServiceInfoHolder serviceInfoHolder, FuzzyWatchServiceListHolder fuzzyWatchServiceListHolder) { this.serviceInfoHolder = serviceInfoHolder; + this.fuzzyWatchServiceListHolder = fuzzyWatchServiceListHolder; } @Override @@ -43,6 +49,10 @@ public Response requestReply(Request request, Connection connection) { NotifySubscriberRequest notifyRequest = (NotifySubscriberRequest) request; serviceInfoHolder.processServiceInfo(notifyRequest.getServiceInfo()); return new NotifySubscriberResponse(); + } else if (request instanceof AbstractFuzzyWatchNotifyRequest) { + AbstractFuzzyWatchNotifyRequest notifyRequest = (AbstractFuzzyWatchNotifyRequest) request; + fuzzyWatchServiceListHolder.processFuzzyWatchNotify(notifyRequest); + return new NotifyFuzzyWatcherResponse(); } return null; } diff --git a/client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/redo/NamingGrpcRedoService.java b/client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/redo/NamingGrpcRedoService.java index aad883ddabe..9627684aed5 100644 --- a/client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/redo/NamingGrpcRedoService.java +++ b/client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/redo/NamingGrpcRedoService.java @@ -26,6 +26,7 @@ import com.alibaba.nacos.client.naming.remote.gprc.redo.data.BatchInstanceRedoData; import com.alibaba.nacos.client.naming.remote.gprc.redo.data.InstanceRedoData; import com.alibaba.nacos.client.naming.remote.gprc.redo.data.SubscriberRedoData; +import com.alibaba.nacos.client.naming.remote.gprc.redo.data.FuzzyWatcherRedoData; import com.alibaba.nacos.client.utils.LogUtils; import com.alibaba.nacos.common.executor.NameThreadFactory; import com.alibaba.nacos.common.remote.client.Connection; @@ -59,6 +60,8 @@ public class NamingGrpcRedoService implements ConnectionEventListener { private final ConcurrentMap subscribes = new ConcurrentHashMap<>(); + private final ConcurrentMap fuzzyWatcher = new ConcurrentHashMap<>(); + private final ScheduledExecutorService redoExecutor; private volatile boolean connected = false; @@ -100,6 +103,9 @@ public void onDisConnect(Connection connection) { synchronized (subscribes) { subscribes.values().forEach(subscriberRedoData -> subscriberRedoData.setRegistered(false)); } + synchronized (fuzzyWatcher) { + fuzzyWatcher.values().forEach(fuzzyWatcherRedoData -> fuzzyWatcherRedoData.setRegistered(false)); + } LogUtils.NAMING_LOGGER.warn("mark to redo completed"); } @@ -315,6 +321,101 @@ public Set findSubscriberRedoData() { return result; } + /** + * Cache fuzzy watcher for redo. + * + * @param serviceNamePattern service name pattern + * @param groupNamePattern group name pattern + */ + public void cacheFuzzyWatcherForRedo(String serviceNamePattern, String groupNamePattern) { + String key = NamingUtils.getGroupedName(serviceNamePattern, groupNamePattern); + FuzzyWatcherRedoData redoData = FuzzyWatcherRedoData.build(serviceNamePattern, groupNamePattern); + synchronized (fuzzyWatcher) { + fuzzyWatcher.put(key, redoData); + } + } + + /** + * Fuzzy watcher register successfully, mark registered status as {@code true}. + * + * @param serviceNamePattern service name pattern + * @param groupNamePattern group name pattern + */ + public void fuzzyWatcherRegistered(String serviceNamePattern, String groupNamePattern) { + String key = NamingUtils.getGroupedName(serviceNamePattern, groupNamePattern); + synchronized (fuzzyWatcher) { + FuzzyWatcherRedoData redoData = fuzzyWatcher.get(key); + if (null != redoData) { + redoData.setRegistered(true); + } + } + } + + /** + * Fuzzy watcher deregister, mark unregistering status as {@code true}. + * + * @param serviceNamePattern service name pattern + * @param groupNamePattern group name pattern + */ + public void fuzzyWatcherDeregister(String serviceNamePattern, String groupNamePattern) { + String key = NamingUtils.getGroupedName(serviceNamePattern, groupNamePattern); + synchronized (fuzzyWatcher) { + FuzzyWatcherRedoData redoData = fuzzyWatcher.get(key); + if (null != redoData) { + redoData.setUnregistering(true); + redoData.setExpectedRegistered(false); + } + } + } + + /** + * Remove fuzzy watcher for redo. + * + * @param serviceNamePattern service name pattern + * @param groupNamePattern group name pattern + */ + public void removeFuzzyWatcherForRedo(String serviceNamePattern, String groupNamePattern) { + String key = NamingUtils.getGroupedName(serviceNamePattern, groupNamePattern); + synchronized (fuzzyWatcher) { + FuzzyWatcherRedoData redoData = fuzzyWatcher.get(key); + if (null != redoData && !redoData.isExpectedRegistered()) { + fuzzyWatcher.remove(key); + } + } + } + + /** + * Judge fuzzy watcher has registered to server. + * + * @param serviceNamePattern service name pattern + * @param groupNamePattern group name pattern + * @return {@code true} if watched, otherwise {@code false} + */ + public boolean isFuzzyWatcherRegistered(String serviceNamePattern, String groupNamePattern) { + String key = NamingUtils.getGroupedName(serviceNamePattern, groupNamePattern); + synchronized (fuzzyWatcher) { + FuzzyWatcherRedoData redoData = fuzzyWatcher.get(key); + return null != redoData && redoData.isRegistered(); + } + } + + /** + * Find all fuzzy watcher redo data which need to redo. + * + * @return set of {@code WatcherRedoData} need to redo. + */ + public Set findFuzzyWatcherRedoData() { + Set result = new HashSet<>(); + synchronized (fuzzyWatcher) { + for (FuzzyWatcherRedoData each : fuzzyWatcher.values()) { + if (each.isNeedRedo()) { + result.add(each); + } + } + } + return result; + } + /** * get Cache service. * diff --git a/client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/redo/RedoScheduledTask.java b/client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/redo/RedoScheduledTask.java index 99d890b3e98..64908b1a0eb 100644 --- a/client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/redo/RedoScheduledTask.java +++ b/client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/redo/RedoScheduledTask.java @@ -22,6 +22,7 @@ import com.alibaba.nacos.client.naming.remote.gprc.redo.data.InstanceRedoData; import com.alibaba.nacos.client.naming.remote.gprc.redo.data.RedoData; import com.alibaba.nacos.client.naming.remote.gprc.redo.data.SubscriberRedoData; +import com.alibaba.nacos.client.naming.remote.gprc.redo.data.FuzzyWatcherRedoData; import com.alibaba.nacos.client.utils.LogUtils; import com.alibaba.nacos.common.task.AbstractExecuteTask; @@ -50,6 +51,7 @@ public void run() { try { redoForInstances(); redoForSubscribes(); + redoForFuzzyWatchers(); } catch (Exception e) { LogUtils.NAMING_LOGGER.warn("Redo task run with unexpected exception: ", e); } @@ -139,6 +141,43 @@ private void redoForSubscribe(SubscriberRedoData redoData) throws NacosException } } + private void redoForFuzzyWatchers() { + for (FuzzyWatcherRedoData each : redoService.findFuzzyWatcherRedoData()) { + try { + redoForFuzzyWatcher(each); + } catch (NacosException e) { + LogUtils.NAMING_LOGGER.error("Redo fuzzy watcher operation {} for pattern {}@@{} in namespace {} failed. ", + each.getRedoType(), each.getGroupName(), each.getServiceName(), clientProxy.getNamespaceId(), e); + } + } + } + + private void redoForFuzzyWatcher(FuzzyWatcherRedoData redoData) throws NacosException { + RedoData.RedoType redoType = redoData.getRedoType(); + String serviceNamePattern = redoData.getServiceName(); + String groupNamePattern = redoData.getGroupName(); + LogUtils.NAMING_LOGGER.info("Redo fuzzy watcher operation {} for pattern {}@@{} in namespace {}", redoType, + groupNamePattern, serviceNamePattern, clientProxy.getNamespaceId()); + switch (redoType) { + case REGISTER: + if (isClientDisabled()) { + return; + } + clientProxy.doFuzzyWatch(serviceNamePattern, groupNamePattern); + break; + case UNREGISTER: + if (isClientDisabled()) { + return; + } + clientProxy.doCancelFuzzyWatch(serviceNamePattern, groupNamePattern); + break; + case REMOVE: + redoService.removeFuzzyWatcherForRedo(serviceNamePattern, groupNamePattern); + break; + default: + } + } + private boolean isClientDisabled() { return !clientProxy.isEnable(); } diff --git a/client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/redo/data/FuzzyWatcherRedoData.java b/client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/redo/data/FuzzyWatcherRedoData.java new file mode 100644 index 00000000000..f0e820910dd --- /dev/null +++ b/client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/redo/data/FuzzyWatcherRedoData.java @@ -0,0 +1,33 @@ +/* + * Copyright 1999-2023 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.nacos.client.naming.remote.gprc.redo.data; + +/** + * Redo data for fuzzy watcher. + * + * @author tanyongquan + */ +public class FuzzyWatcherRedoData extends RedoData { + + private FuzzyWatcherRedoData(String serviceNamePattern, String groupNamePattern) { + super(serviceNamePattern, groupNamePattern); + } + + public static FuzzyWatcherRedoData build(String serviceNamePattern, String groupNamePattern) { + return new FuzzyWatcherRedoData(serviceNamePattern, groupNamePattern); + } +} diff --git a/client/src/main/java/com/alibaba/nacos/client/naming/remote/http/NamingHttpClientProxy.java b/client/src/main/java/com/alibaba/nacos/client/naming/remote/http/NamingHttpClientProxy.java index a6568fc4010..d0411ebb832 100644 --- a/client/src/main/java/com/alibaba/nacos/client/naming/remote/http/NamingHttpClientProxy.java +++ b/client/src/main/java/com/alibaba/nacos/client/naming/remote/http/NamingHttpClientProxy.java @@ -328,6 +328,21 @@ public boolean isSubscribed(String serviceName, String groupName, String cluster return true; } + @Override + public void fuzzyWatch(String serviceNamePattern, String groupNamePattern, String uuid) throws NacosException { + throw new UnsupportedOperationException("Do not support fuzzy watch services by UDP, please use gRPC replaced."); + } + + @Override + public void cancelFuzzyWatch(String serviceNamePattern, String groupNamePattern) throws NacosException { + throw new UnsupportedOperationException("Do not support fuzzy watch service by UDP, please use gRPC replaced."); + } + + @Override + public boolean isFuzzyWatched(String serviceNamePattern, String groupNamePattern) { + throw new UnsupportedOperationException("Do not support fuzzy watch service by UDP, please use gRPC replaced."); + } + public String reqApi(String api, Map params, String method) throws NacosException { return reqApi(api, params, Collections.EMPTY_MAP, method); } diff --git a/client/src/main/resources/META-INF/native-image/com.alibaba.nacos/nacos-client/reflect-config.json b/client/src/main/resources/META-INF/native-image/com.alibaba.nacos/nacos-client/reflect-config.json index 3734fbb2843..da2ba912161 100644 --- a/client/src/main/resources/META-INF/native-image/com.alibaba.nacos/nacos-client/reflect-config.json +++ b/client/src/main/resources/META-INF/native-image/com.alibaba.nacos/nacos-client/reflect-config.json @@ -535,14 +535,23 @@ {"name":"getServiceName","parameterTypes":[] } ] }, +{ + "name":"com.alibaba.nacos.api.naming.remote.request.AbstractFuzzyWatchNotifyRequest", + "allDeclaredFields":true, + "queryAllDeclaredMethods":true, + "queryAllDeclaredConstructors":true, + "methods":[ {"name":"getInstances","parameterTypes":[] }, + {"name":"getType","parameterTypes":[] }] +}, { "name":"com.alibaba.nacos.api.naming.remote.request.BatchInstanceRequest", "allDeclaredFields":true, "queryAllDeclaredMethods":true, "queryAllDeclaredConstructors":true, "methods":[ - {"name":"getInstances","parameterTypes":[] }, - {"name":"getType","parameterTypes":[] } + {"name":"","parameterTypes":[] }, + {"name":"getModule","parameterTypes":[] }, + {"name":"getServiceChangedType","parameterTypes":[] } ] }, { @@ -603,6 +612,28 @@ {"name":"isHealthyOnly","parameterTypes":[] } ] }, +{ + "name":"com.alibaba.nacos.api.naming.remote.request.FuzzyWatchNotifyChangeRequest", + "allDeclaredFields":true, + "queryAllDeclaredMethods":true, + "queryAllDeclaredConstructors":true, + "methods":[ + {"name":"","parameterTypes":[] }, + {"name":"getServiceName","parameterTypes":[] }, + {"name":"getGroupName","parameterTypes":[] } + ] +}, +{ + "name":"com.alibaba.nacos.api.naming.remote.request.FuzzyWatchNotifyInitRequest", + "allDeclaredFields":true, + "queryAllDeclaredMethods":true, + "queryAllDeclaredConstructors":true, + "methods":[ + {"name":"","parameterTypes":[] }, + {"name":"getServicesName","parameterTypes":[] }, + {"name":"getPattern","parameterTypes":[] } + ] +}, { "name":"com.alibaba.nacos.api.naming.remote.request.SubscribeServiceRequest", "allDeclaredFields":true, diff --git a/client/src/test/java/com/alibaba/nacos/client/naming/NacosNamingServiceTest.java b/client/src/test/java/com/alibaba/nacos/client/naming/NacosNamingServiceTest.java index b1c98ee7195..2232aa9f04f 100644 --- a/client/src/test/java/com/alibaba/nacos/client/naming/NacosNamingServiceTest.java +++ b/client/src/test/java/com/alibaba/nacos/client/naming/NacosNamingServiceTest.java @@ -1089,7 +1089,6 @@ public void testUnSubscribeWithServiceAndCustomSelector() throws NacosException verify(proxy, times(1)).unsubscribe(serviceName, Constants.DEFAULT_GROUP, Constants.NULL); } - @Test public void testUnSubscribeWithFullNameAndCustomSelector() throws NacosException { //given String serviceName = "service1"; diff --git a/client/src/test/java/com/alibaba/nacos/client/naming/remote/AbstractNamingClientProxyTest.java b/client/src/test/java/com/alibaba/nacos/client/naming/remote/AbstractNamingClientProxyTest.java index 7f570c2780b..43e01a350bf 100644 --- a/client/src/test/java/com/alibaba/nacos/client/naming/remote/AbstractNamingClientProxyTest.java +++ b/client/src/test/java/com/alibaba/nacos/client/naming/remote/AbstractNamingClientProxyTest.java @@ -174,6 +174,21 @@ public boolean isSubscribed(String serviceName, String groupName, String cluster return false; } + @Override + public void fuzzyWatch(String serviceNamePattern, String groupNamePattern, String uuid) throws NacosException { + + } + + @Override + public void cancelFuzzyWatch(String serviceNamePattern, String groupNamePattern) throws NacosException { + + } + + @Override + public boolean isFuzzyWatched(String serviceNamePattern, String groupNamePattern) { + return false; + } + @Override public boolean serverHealthy() { return false; diff --git a/client/src/test/java/com/alibaba/nacos/client/naming/remote/NamingClientProxyDelegateTest.java b/client/src/test/java/com/alibaba/nacos/client/naming/remote/NamingClientProxyDelegateTest.java index bdc3ac6cec1..7f201b14eb9 100644 --- a/client/src/test/java/com/alibaba/nacos/client/naming/remote/NamingClientProxyDelegateTest.java +++ b/client/src/test/java/com/alibaba/nacos/client/naming/remote/NamingClientProxyDelegateTest.java @@ -28,6 +28,7 @@ import com.alibaba.nacos.api.selector.NoneSelector; import com.alibaba.nacos.client.env.NacosClientProperties; import com.alibaba.nacos.client.naming.cache.ServiceInfoHolder; +import com.alibaba.nacos.client.naming.cache.FuzzyWatchServiceListHolder; import com.alibaba.nacos.client.naming.event.InstancesChangeNotifier; import com.alibaba.nacos.client.naming.remote.gprc.NamingGrpcClientProxy; import com.alibaba.nacos.client.naming.remote.http.NamingHttpClientProxy; @@ -65,6 +66,9 @@ class NamingClientProxyDelegateTest { @Mock NamingGrpcClientProxy mockGrpcClient; + @Mock + FuzzyWatchServiceListHolder fuzzyWatchServiceListHolder; + NamingClientProxyDelegate delegate; InstancesChangeNotifier notifier; @@ -77,7 +81,7 @@ void setUp() throws NacosException, NoSuchFieldException, IllegalAccessException props.setProperty("serverAddr", "localhost"); nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(props); notifier = new InstancesChangeNotifier(); - delegate = new NamingClientProxyDelegate(TEST_NAMESPACE, holder, nacosClientProperties, notifier); + delegate = new NamingClientProxyDelegate(TEST_NAMESPACE, holder, nacosClientProperties,fuzzyWatchServiceListHolder, notifier); Field grpcClientProxyField = NamingClientProxyDelegate.class.getDeclaredField("grpcClientProxy"); grpcClientProxyField.setAccessible(true); grpcClientProxyField.set(delegate, mockGrpcClient); @@ -102,7 +106,7 @@ void testRegisterEphemeralServiceByGrpc() throws NacosException { verify(mockGrpcClient, times(1)).registerService(serviceName, groupName, instance); } - @Test + void testBatchRegisterServiceByGrpc() throws NacosException { String serviceName = "service1"; String groupName = "group1"; @@ -169,7 +173,7 @@ void testRegisterPersistentServiceByHttp() throws NacosException, NoSuchFieldExc verify(mockHttpClient, times(1)).registerService(serviceName, groupName, instance); } - @Test + void testDeregisterEphemeralServiceGrpc() throws NacosException { String serviceName = "service1"; String groupName = "group1"; @@ -221,8 +225,7 @@ void testDeregisterPersistentServiceHttp() throws NacosException, NoSuchFieldExc delegate.deregisterService(serviceName, groupName, instance); verify(mockHttpClient, times(1)).deregisterService(serviceName, groupName, instance); } - - @Test + void testUpdateInstance() { String serviceName = "service1"; String groupName = "group1"; @@ -231,8 +234,7 @@ void testUpdateInstance() { delegate.updateInstance(serviceName, groupName, instance); }); } - - @Test + void testQueryInstancesOfService() throws NacosException { String serviceName = "service1"; String groupName = "group1"; @@ -241,13 +243,13 @@ void testQueryInstancesOfService() throws NacosException { verify(mockGrpcClient, times(1)).queryInstancesOfService(serviceName, groupName, clusters, false); } - @Test + void testQueryService() throws NacosException { Service service = delegate.queryService("a", "b"); assertNull(service); } - @Test + void testCreateService() { Service service = new Service(); Assertions.assertDoesNotThrow(() -> { @@ -255,7 +257,7 @@ void testCreateService() { }); } - @Test + void testDeleteService() throws NacosException { assertFalse(delegate.deleteService("service", "group1")); } @@ -268,7 +270,7 @@ void testUpdateService() { }); } - @Test + void testGetServiceList() throws NacosException { AbstractSelector selector = new ExpressionSelector(); int pageNo = 1; diff --git a/client/src/test/java/com/alibaba/nacos/client/naming/remote/gprc/NamingGrpcClientProxyTest.java b/client/src/test/java/com/alibaba/nacos/client/naming/remote/gprc/NamingGrpcClientProxyTest.java index bdb9472cfbc..910dbdab4c0 100644 --- a/client/src/test/java/com/alibaba/nacos/client/naming/remote/gprc/NamingGrpcClientProxyTest.java +++ b/client/src/test/java/com/alibaba/nacos/client/naming/remote/gprc/NamingGrpcClientProxyTest.java @@ -47,6 +47,8 @@ import com.alibaba.nacos.api.selector.NoneSelector; import com.alibaba.nacos.client.env.NacosClientProperties; import com.alibaba.nacos.client.naming.cache.ServiceInfoHolder; +import com.alibaba.nacos.client.naming.cache.FuzzyWatchServiceListHolder; +import com.alibaba.nacos.client.naming.event.ServerListChangedEvent; import com.alibaba.nacos.client.address.ServerListChangeEvent; import com.alibaba.nacos.client.naming.remote.gprc.redo.NamingGrpcRedoService; import com.alibaba.nacos.client.security.SecurityProxy; @@ -120,6 +122,9 @@ class NamingGrpcClientProxyTest { @Mock private ServiceInfoHolder holder; + @Mock + private FuzzyWatchServiceListHolder fuzzyWatchServiceListHolder; + @Mock private RpcClient rpcClient; @@ -145,7 +150,8 @@ void setUp() throws NacosException, NoSuchFieldException, IllegalAccessException prop = new Properties(); final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop); - client = new NamingGrpcClientProxy(NAMESPACE_ID, proxy, factory, nacosClientProperties, holder); + client = new NamingGrpcClientProxy(NAMESPACE_ID, proxy, factory, nacosClientProperties, holder, + fuzzyWatchServiceListHolder); Field uuidField = NamingGrpcClientProxy.class.getDeclaredField("uuid"); uuidField.setAccessible(true); @@ -664,7 +670,7 @@ public void close() { rpcClient.set(client, rpc); rpc.serverListFactory(factory); - rpc.registerServerRequestHandler(new NamingPushRequestHandler(holder)); + rpc.registerServerRequestHandler(new NamingPushRequestHandler(holder, fuzzyWatchServiceListHolder)); Field listenerField = NamingGrpcClientProxy.class.getDeclaredField("redoService"); listenerField.setAccessible(true); NamingGrpcRedoService listener = (NamingGrpcRedoService) listenerField.get(client); @@ -699,7 +705,8 @@ public void close() { @Test void testConfigAppNameLabels() throws Exception { final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop); - client = new NamingGrpcClientProxy(NAMESPACE_ID, proxy, factory, nacosClientProperties, holder); + client = new NamingGrpcClientProxy(NAMESPACE_ID, proxy, factory, nacosClientProperties, holder, + fuzzyWatchServiceListHolder); Field rpcClientField = NamingGrpcClientProxy.class.getDeclaredField("rpcClient"); rpcClientField.setAccessible(true); RpcClient rpcClient = (RpcClient) rpcClientField.get(client); diff --git a/client/src/test/java/com/alibaba/nacos/client/naming/remote/gprc/NamingPushRequestHandlerTest.java b/client/src/test/java/com/alibaba/nacos/client/naming/remote/gprc/NamingPushRequestHandlerTest.java index 341c26e1dcd..786a4d7b5e9 100644 --- a/client/src/test/java/com/alibaba/nacos/client/naming/remote/gprc/NamingPushRequestHandlerTest.java +++ b/client/src/test/java/com/alibaba/nacos/client/naming/remote/gprc/NamingPushRequestHandlerTest.java @@ -25,6 +25,9 @@ import com.alibaba.nacos.api.remote.request.Request; import com.alibaba.nacos.api.remote.response.Response; import com.alibaba.nacos.client.naming.cache.ServiceInfoHolder; +import com.alibaba.nacos.client.naming.cache.FuzzyWatchServiceListHolder; +import org.junit.Assert; +import org.junit.Test; import com.alibaba.nacos.client.naming.remote.TestConnection; import com.alibaba.nacos.common.remote.client.RpcClient; import org.junit.jupiter.api.Test; @@ -41,7 +44,8 @@ class NamingPushRequestHandlerTest { void testRequestReply() { //given ServiceInfoHolder holder = mock(ServiceInfoHolder.class); - NamingPushRequestHandler handler = new NamingPushRequestHandler(holder); + FuzzyWatchServiceListHolder fuzzyWatchServiceListHolder = mock(FuzzyWatchServiceListHolder.class); + NamingPushRequestHandler handler = new NamingPushRequestHandler(holder, fuzzyWatchServiceListHolder); ServiceInfo info = new ServiceInfo("name", "cluster1"); Request req = NotifySubscriberRequest.buildNotifySubscriberRequest(info); //when diff --git a/example/src/main/java/com/alibaba/nacos/example/FuzzyWatchExample.java b/example/src/main/java/com/alibaba/nacos/example/FuzzyWatchExample.java new file mode 100644 index 00000000000..20f0829eb94 --- /dev/null +++ b/example/src/main/java/com/alibaba/nacos/example/FuzzyWatchExample.java @@ -0,0 +1,115 @@ +/* + * Copyright 1999-2023 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.nacos.example; + +import com.alibaba.nacos.api.exception.NacosException; +import com.alibaba.nacos.api.naming.NamingFactory; +import com.alibaba.nacos.api.naming.NamingService; +import com.alibaba.nacos.api.naming.listener.AbstractFuzzyWatchEventListener; +import com.alibaba.nacos.api.naming.listener.FuzzyWatchNotifyEvent; + +import java.util.Properties; +import java.util.concurrent.Executor; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +import static com.alibaba.nacos.api.common.Constants.DEFAULT_GROUP; + +/** + * Nacos naming fuzzy watch example. + *

Add the JVM parameter to run the NamingExample:

+ * {@code -DserverAddr=${nacos.server.ip}:${nacos.server.port} -Dnamespace=${namespaceId}} + * + * @author tanyongquan + */ +public class FuzzyWatchExample { + + public static void main(String[] args) throws NacosException, InterruptedException { + + Properties properties = new Properties(); + properties.setProperty("serverAddr", System.getProperty("serverAddr", "localhost")); + properties.setProperty("namespace", System.getProperty("namespace", "public")); + + NamingService naming = NamingFactory.createNamingService(properties); + + int num = 5; + for (int i = 1; i <= num; i++) { + String s = "nacos.test." + i; + naming.registerInstance(s, "11.11.11.11", 8888); + } + + System.out.println(num + " instance have been registered"); + + Executor executor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(), + runnable -> { + Thread thread = new Thread(runnable); + thread.setName("test-thread"); + return thread; + }); + + naming.fuzzyWatch(DEFAULT_GROUP, new AbstractFuzzyWatchEventListener() { + + //EventListener onEvent is sync to handle, If process too low in onEvent, maybe block other onEvent callback. + //So you can override getExecutor() to async handle event. + @Override + public Executor getExecutor() { + return executor; + } + + @Override + public void onEvent(FuzzyWatchNotifyEvent event) { + System.out.println("[Fuzzy-Watch-GROUP]changed service name: " + event.getService().getGroupedServiceName()); + System.out.println("[Fuzzy-Watch-GROUP]change type: " + event.getChangeType()); + } + }); + + naming.fuzzyWatch("nacos.test.*", DEFAULT_GROUP, new AbstractFuzzyWatchEventListener() { + + @Override + public Executor getExecutor() { + return executor; + } + + @Override + public void onEvent(FuzzyWatchNotifyEvent event) { + System.out.println("[Prefix-Fuzzy-Watch]changed service name: " + event.getService().getGroupedServiceName()); + System.out.println("[Prefix-Fuzzy-Watch]change type: " + event.getChangeType()); + } + }); + + naming.registerInstance("nacos.test.-1", "11.11.11.11", 8888); + + Thread.sleep(1000); + + naming.registerInstance("nacos.OTHER-PREFIX", "11.11.11.11", 8888); + + Thread.sleep(1000); + + naming.registerInstance("nacos.OTHER-GROUP", "OTHER-GROUP", "11.11.11.11", 8888); + + Thread.sleep(1000); + + for (int i = 1; i <= num; i++) { + String s = "nacos.test." + i; + naming.deregisterInstance(s, "11.11.11.11", 8888); + } + + Thread.sleep(1000); + + } +} diff --git a/naming/src/main/java/com/alibaba/nacos/naming/core/v2/client/AbstractClient.java b/naming/src/main/java/com/alibaba/nacos/naming/core/v2/client/AbstractClient.java index 5662a5ae257..4c967a4b1c7 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/core/v2/client/AbstractClient.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/core/v2/client/AbstractClient.java @@ -17,6 +17,7 @@ package com.alibaba.nacos.naming.core.v2.client; import com.alibaba.nacos.common.notify.NotifyCenter; +import com.alibaba.nacos.common.utils.ConcurrentHashSet; import com.alibaba.nacos.naming.core.v2.event.client.ClientEvent; import com.alibaba.nacos.naming.core.v2.pojo.BatchInstanceData; import com.alibaba.nacos.naming.core.v2.pojo.BatchInstancePublishInfo; @@ -47,6 +48,8 @@ public abstract class AbstractClient implements Client { protected final ConcurrentHashMap subscribers = new ConcurrentHashMap<>(16, 0.75f, 1); + protected final ConcurrentHashSet watchedPattern = new ConcurrentHashSet<>(); + protected volatile long lastUpdatedTime; protected final AtomicLong revision; @@ -134,6 +137,34 @@ public Collection getAllSubscribeService() { return subscribers.keySet(); } + @Override + public boolean addWatchedPattern(String watchPattern) { + if (watchedPattern.add(watchPattern)) { + // TODO:Watch MetricsMonitor + return true; + } + return true; + } + + @Override + public boolean removeWatchedPattern(String watchPattern) { + if (watchedPattern.remove(watchPattern)) { + // TODO:Watch MetricsMonitor + return true; + } + return true; + } + + @Override + public boolean isWatchedPattern(String watchPattern) { + return watchedPattern.contains(watchPattern); + } + + @Override + public Collection getAllFuzzyWatchPattern() { + return watchedPattern; + } + @Override public ClientSyncData generateSyncData() { List namespaces = new LinkedList<>(); diff --git a/naming/src/main/java/com/alibaba/nacos/naming/core/v2/client/Client.java b/naming/src/main/java/com/alibaba/nacos/naming/core/v2/client/Client.java index 6cbe0112dc8..94bfb33f7a9 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/core/v2/client/Client.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/core/v2/client/Client.java @@ -122,6 +122,37 @@ public interface Client { */ Collection getAllSubscribeService(); + /** + * Add a watched pattern for this client. + * + * @param pattern watch pattern. + * @return true if add successfully, otherwise false + */ + boolean addWatchedPattern(String pattern); + + /** + * Remove a watched pattern for this client. + * + * @param pattern watch pattern. + * @return true if remove successfully, otherwise false + */ + boolean removeWatchedPattern(String pattern); + + /** + * Judge whether watch this pattern of this client. + * + * @param watchPattern watch patten + * @return true if client watch the given pattern, otherwise false + */ + boolean isWatchedPattern(String watchPattern); + + /** + * Get all watched pattern of current client. + * + * @return watch patterns + */ + Collection getAllFuzzyWatchPattern(); + /** * Generate sync data. * diff --git a/naming/src/main/java/com/alibaba/nacos/naming/core/v2/event/client/ClientOperationEvent.java b/naming/src/main/java/com/alibaba/nacos/naming/core/v2/event/client/ClientOperationEvent.java index 93ecd25cdea..4274cfef9ea 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/core/v2/event/client/ClientOperationEvent.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/core/v2/event/client/ClientOperationEvent.java @@ -94,6 +94,45 @@ public ClientUnsubscribeServiceEvent(Service service, String clientId) { } } + /** + * Client fuzzy watch service event. + */ + public static class ClientFuzzyWatchEvent extends ClientOperationEvent { + + private static final long serialVersionUID = -4518919987813223119L; + + private final String pattern; + + public ClientFuzzyWatchEvent(String pattern, String clientId) { + super(clientId, null); + this.pattern = pattern; + } + + public String getPattern() { + return pattern; + } + + } + + /** + * Client cancel fuzzy watch service event. + */ + public static class ClientCancelFuzzyWatchEvent extends ClientOperationEvent { + + private static final long serialVersionUID = -4518919987813223118L; + + private final String pattern; + + public ClientCancelFuzzyWatchEvent(String pattern, String clientId) { + super(clientId, null); + this.pattern = pattern; + } + + public String getPattern() { + return pattern; + } + } + public static class ClientReleaseEvent extends ClientOperationEvent { private static final long serialVersionUID = -281486927726245701L; diff --git a/naming/src/main/java/com/alibaba/nacos/naming/core/v2/event/service/ServiceEvent.java b/naming/src/main/java/com/alibaba/nacos/naming/core/v2/event/service/ServiceEvent.java index 6d559529880..76ee2a7b2b9 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/core/v2/event/service/ServiceEvent.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/core/v2/event/service/ServiceEvent.java @@ -19,6 +19,8 @@ import com.alibaba.nacos.common.notify.Event; import com.alibaba.nacos.naming.core.v2.pojo.Service; +import java.util.Collection; + /** * Service event. * @@ -28,7 +30,10 @@ public class ServiceEvent extends Event { private static final long serialVersionUID = -9173247502346692418L; - private final Service service; + private Service service; + + public ServiceEvent() { + } public ServiceEvent(Service service) { this.service = service; @@ -45,17 +50,25 @@ public static class ServiceChangedEvent extends ServiceEvent { private static final long serialVersionUID = 2123694271992630822L; - public ServiceChangedEvent(Service service) { - this(service, false); + private final String changedType; + + public ServiceChangedEvent(Service service, String changedType) { + this(service, changedType, false); } - public ServiceChangedEvent(Service service, boolean incrementRevision) { + public ServiceChangedEvent(Service service, String changedType, boolean incrementRevision) { super(service); + this.changedType = changedType; service.renewUpdateTime(); if (incrementRevision) { service.incrementRevision(); } } + + public String getChangedType() { + return changedType; + } + } /** @@ -77,4 +90,36 @@ public String getClientId() { } } + /** + * A client initiates a fuzzy watch request. + */ + public static class ServiceFuzzyWatchInitEvent extends ServiceEvent { + + private static final long serialVersionUID = -2645441445867337345L; + + private final String clientId; + + private final String pattern; + + private final Collection matchedService; + + public ServiceFuzzyWatchInitEvent(String clientId, String pattern, Collection matchedService) { + super(); + this.clientId = clientId; + this.pattern = pattern; + this.matchedService = matchedService; + } + + public String getClientId() { + return clientId; + } + + public String getPattern() { + return pattern; + } + + public Collection getMatchedService() { + return matchedService; + } + } } diff --git a/naming/src/main/java/com/alibaba/nacos/naming/core/v2/index/ClientServiceIndexesManager.java b/naming/src/main/java/com/alibaba/nacos/naming/core/v2/index/ClientServiceIndexesManager.java index 48ea5611ae6..c5506e2dc4f 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/core/v2/index/ClientServiceIndexesManager.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/core/v2/index/ClientServiceIndexesManager.java @@ -16,21 +16,27 @@ package com.alibaba.nacos.naming.core.v2.index; +import com.alibaba.nacos.api.common.Constants; +import com.alibaba.nacos.api.naming.utils.NamingUtils; import com.alibaba.nacos.common.notify.Event; import com.alibaba.nacos.common.notify.NotifyCenter; import com.alibaba.nacos.common.notify.listener.SmartSubscriber; import com.alibaba.nacos.common.trace.DeregisterInstanceReason; import com.alibaba.nacos.common.trace.event.naming.DeregisterInstanceTraceEvent; +import com.alibaba.nacos.common.utils.CollectionUtils; import com.alibaba.nacos.common.utils.ConcurrentHashSet; +import com.alibaba.nacos.naming.core.v2.ServiceManager; import com.alibaba.nacos.naming.core.v2.client.Client; import com.alibaba.nacos.naming.core.v2.event.client.ClientOperationEvent; import com.alibaba.nacos.naming.core.v2.event.publisher.NamingEventPublisherFactory; import com.alibaba.nacos.naming.core.v2.event.service.ServiceEvent; import com.alibaba.nacos.naming.core.v2.pojo.InstancePublishInfo; import com.alibaba.nacos.naming.core.v2.pojo.Service; +import com.alibaba.nacos.naming.misc.Loggers; import org.springframework.stereotype.Component; import java.util.Collection; +import java.util.HashSet; import java.util.LinkedList; import java.util.List; import java.util.Set; @@ -49,6 +55,16 @@ public class ClientServiceIndexesManager extends SmartSubscriber { private final ConcurrentMap> subscriberIndexes = new ConcurrentHashMap<>(); + /** + * The content of map is {fuzzy watch pattern -> Set[watcher clientID]}. + */ + private final ConcurrentMap> fuzzyWatcherIndexes = new ConcurrentHashMap<>(); + + /** + * The content of map is {service -> Set[matched fuzzy watch patterns]}. + */ + private final ConcurrentMap> fuzzyWatchPatternMatchIndexes = new ConcurrentHashMap<>(); + public ClientServiceIndexesManager() { NotifyCenter.registerSubscriber(this, NamingEventPublisherFactory.getInstance()); } @@ -65,6 +81,15 @@ public Collection getSubscribedService() { return subscriberIndexes.keySet(); } + public Collection getServiceMatchedPatterns(Service service) { + return fuzzyWatchPatternMatchIndexes.containsKey(service) + ? fuzzyWatchPatternMatchIndexes.get(service) : new ConcurrentHashSet<>(); + } + + public Collection getAllClientFuzzyWatchedPattern(String pattern) { + return fuzzyWatcherIndexes.containsKey(pattern) ? fuzzyWatcherIndexes.get(pattern) : new ConcurrentHashSet<>(); + } + /** * Clear the service index without instances. * @@ -73,6 +98,7 @@ public Collection getSubscribedService() { public void removePublisherIndexesByEmptyService(Service service) { if (publisherIndexes.containsKey(service) && publisherIndexes.get(service).isEmpty()) { publisherIndexes.remove(service); + fuzzyWatchPatternMatchIndexes.remove(service); } } @@ -83,6 +109,8 @@ public List> subscribeTypes() { result.add(ClientOperationEvent.ClientDeregisterServiceEvent.class); result.add(ClientOperationEvent.ClientSubscribeServiceEvent.class); result.add(ClientOperationEvent.ClientUnsubscribeServiceEvent.class); + result.add(ClientOperationEvent.ClientFuzzyWatchEvent.class); + result.add(ClientOperationEvent.ClientCancelFuzzyWatchEvent.class); result.add(ClientOperationEvent.ClientReleaseEvent.class); return result; } @@ -101,6 +129,9 @@ private void handleClientDisconnect(ClientOperationEvent.ClientReleaseEvent even for (Service each : client.getAllSubscribeService()) { removeSubscriberIndexes(each, client.getClientId()); } + for (String eachPattern : client.getAllFuzzyWatchPattern()) { + removeFuzzyWatcherIndexes(eachPattern, client.getClientId()); + } DeregisterInstanceReason reason = event.isNative() ? DeregisterInstanceReason.NATIVE_DISCONNECTED : DeregisterInstanceReason.SYNCED_DISCONNECTED; long currentTimeMillis = System.currentTimeMillis(); @@ -124,18 +155,32 @@ private void handleClientOperation(ClientOperationEvent event) { addSubscriberIndexes(service, clientId); } else if (event instanceof ClientOperationEvent.ClientUnsubscribeServiceEvent) { removeSubscriberIndexes(service, clientId); + } else if (event instanceof ClientOperationEvent.ClientFuzzyWatchEvent) { + String completedPattern = ((ClientOperationEvent.ClientFuzzyWatchEvent) event).getPattern(); + addFuzzyWatcherIndexes(completedPattern, clientId); + } else if (event instanceof ClientOperationEvent.ClientCancelFuzzyWatchEvent) { + String completedPattern = ((ClientOperationEvent.ClientCancelFuzzyWatchEvent) event).getPattern(); + removeFuzzyWatcherIndexes(completedPattern, clientId); } } private void addPublisherIndexes(Service service, String clientId) { + String serviceChangedType = Constants.ServiceChangedType.INSTANCE_CHANGED; + if (!publisherIndexes.containsKey(service)) { + // The only time the index needs to be updated is when the service is first created + updateWatchMatchIndex(service); + serviceChangedType = Constants.ServiceChangedType.ADD_SERVICE; + } + NotifyCenter.publishEvent(new ServiceEvent.ServiceChangedEvent(service, serviceChangedType, true)); publisherIndexes.computeIfAbsent(service, key -> new ConcurrentHashSet<>()).add(clientId); - NotifyCenter.publishEvent(new ServiceEvent.ServiceChangedEvent(service, true)); } private void removePublisherIndexes(Service service, String clientId) { publisherIndexes.computeIfPresent(service, (s, ids) -> { ids.remove(clientId); - NotifyCenter.publishEvent(new ServiceEvent.ServiceChangedEvent(service, true)); + String serviceChangedType = ids.isEmpty() ? Constants.ServiceChangedType.DELETE_SERVICE : + Constants.ServiceChangedType.INSTANCE_CHANGED; + NotifyCenter.publishEvent(new ServiceEvent.ServiceChangedEvent(service, serviceChangedType, true)); return ids.isEmpty() ? null : ids; }); } @@ -158,4 +203,86 @@ private void removeSubscriberIndexes(Service service, String clientId) { subscriberIndexes.remove(service); } } + + private void addFuzzyWatcherIndexes(String completedPattern, String clientId) { + fuzzyWatcherIndexes.computeIfAbsent(completedPattern, key -> new ConcurrentHashSet<>()); + fuzzyWatcherIndexes.get(completedPattern).add(clientId); + Collection matchedService = updateWatchMatchIndex(completedPattern); + NotifyCenter.publishEvent(new ServiceEvent.ServiceFuzzyWatchInitEvent(clientId, completedPattern, matchedService)); + } + + private void removeFuzzyWatcherIndexes(String completedPattern, String clientId) { + if (!fuzzyWatcherIndexes.containsKey(completedPattern)) { + return; + } + fuzzyWatcherIndexes.get(completedPattern).remove(clientId); + if (fuzzyWatcherIndexes.get(completedPattern).isEmpty()) { + fuzzyWatcherIndexes.remove(completedPattern); + } + } + + /** + * This method will build/update the fuzzy watch match index of all patterns. + * + * @param service The service of the Nacos. + */ + public void updateWatchMatchIndex(Service service) { + long matchBeginTime = System.currentTimeMillis(); + Set filteredPattern = NamingUtils.filterPatternWithNamespace(service.getNamespace(), fuzzyWatcherIndexes.keySet()); + Set matchedPattern = NamingUtils.getServiceMatchedPatterns(service.getName(), service.getGroup(), + filteredPattern); + if (CollectionUtils.isNotEmpty(matchedPattern)) { + fuzzyWatchPatternMatchIndexes.computeIfAbsent(service, key -> new ConcurrentHashSet<>()); + for (String each : matchedPattern) { + fuzzyWatchPatternMatchIndexes.get(service).add(NamingUtils.getPatternWithNamespace(service.getNamespace(), each)); + } + Loggers.PERFORMANCE_LOG.info("WATCH: new service {} match {} pattern, {}ms", service.getGroupedServiceName(), + matchedPattern.size(), System.currentTimeMillis() - matchBeginTime); + } + } + + /** + * This method will build/update the fuzzy watch match index for given patterns. + * + * @param completedPattern the completed pattern of watch (with namespace id). + * @return Updated set of services in Nacos server that can match this pattern. + */ + public Collection updateWatchMatchIndex(String completedPattern) { + long matchBeginTime = System.currentTimeMillis(); + String namespaceId = NamingUtils.getNamespaceFromPattern(completedPattern); + Collection serviceSet = ServiceManager.getInstance().getSingletons(namespaceId); + String pattern = NamingUtils.getPatternRemovedNamespace(completedPattern); + + Set matchedService = new HashSet<>(); + for (Service service : serviceSet) { + String serviceName = service.getName(); + String groupName = service.getGroup(); + String serviceNamePattern = NamingUtils.getServiceName(pattern); + String groupNamePattern = NamingUtils.getGroupName(pattern); + if (NamingUtils.isMatchPattern(serviceName, groupName, serviceNamePattern, groupNamePattern)) { + fuzzyWatchPatternMatchIndexes.computeIfAbsent(service, key -> new ConcurrentHashSet<>()); + fuzzyWatchPatternMatchIndexes.get(service).add(completedPattern); + matchedService.add(service); + } + } + Loggers.PERFORMANCE_LOG.info("WATCH: pattern {} match {} services, {}ms", completedPattern, + matchedService.size(), System.currentTimeMillis() - matchBeginTime); + return matchedService; + } + + /** + * This method will remove the match index of fuzzy watch pattern. + * + * @param service The service of the Nacos. + * @param matchedPattern the pattern to remove + */ + public void removeWatchPatternMatchIndex(Service service, String matchedPattern) { + if (!fuzzyWatchPatternMatchIndexes.containsKey(service)) { + return; + } + fuzzyWatchPatternMatchIndexes.get(service).remove(matchedPattern); + if (fuzzyWatchPatternMatchIndexes.get(service).isEmpty()) { + fuzzyWatchPatternMatchIndexes.remove(service); + } + } } diff --git a/naming/src/main/java/com/alibaba/nacos/naming/core/v2/metadata/InstanceMetadataProcessor.java b/naming/src/main/java/com/alibaba/nacos/naming/core/v2/metadata/InstanceMetadataProcessor.java index 92cb1ee4f51..d7e6514f287 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/core/v2/metadata/InstanceMetadataProcessor.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/core/v2/metadata/InstanceMetadataProcessor.java @@ -108,7 +108,8 @@ private void updateInstanceMetadata(MetadataOperation op) { Service service = Service.newService(op.getNamespace(), op.getGroup(), op.getServiceName()); service = ServiceManager.getInstance().getSingleton(service); namingMetadataManager.updateInstanceMetadata(service, op.getTag(), op.getMetadata()); - NotifyCenter.publishEvent(new ServiceEvent.ServiceChangedEvent(service, true)); + NotifyCenter.publishEvent(new ServiceEvent.ServiceChangedEvent(service, + com.alibaba.nacos.api.common.Constants.ServiceChangedType.INSTANCE_CHANGED, true)); } private void deleteInstanceMetadata(MetadataOperation op) { diff --git a/naming/src/main/java/com/alibaba/nacos/naming/core/v2/service/ClientOperationService.java b/naming/src/main/java/com/alibaba/nacos/naming/core/v2/service/ClientOperationService.java index 3addf41e62f..b2696dbc68a 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/core/v2/service/ClientOperationService.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/core/v2/service/ClientOperationService.java @@ -85,6 +85,30 @@ default void unsubscribeService(Service service, Subscriber subscriber, String c } + /** + * Fuzzy watch a pattern. + * + * @param namespaceId namespace id of this pattern + * @param serviceNamePattern watch service name pattern rule + * @param groupNamePattern watch service name pattern rule + * @param clientId id of client + */ + default void fuzzyWatch(String namespaceId, String serviceNamePattern, String groupNamePattern, String clientId) { + + } + + /** + * Cancel fuzzy watch a pattern. + * + * @param namespaceId namespace id of this pattern + * @param serviceNamePattern watch service name pattern + * @param groupNamePattern watch service name pattern + * @param clientId id of client + */ + default void cancelFuzzyWatch(String namespaceId, String serviceNamePattern, String groupNamePattern, String clientId) { + + } + /** * get publish info. * diff --git a/naming/src/main/java/com/alibaba/nacos/naming/core/v2/service/ClientOperationServiceProxy.java b/naming/src/main/java/com/alibaba/nacos/naming/core/v2/service/ClientOperationServiceProxy.java index 2efa7c6dd9f..7284af587e1 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/core/v2/service/ClientOperationServiceProxy.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/core/v2/service/ClientOperationServiceProxy.java @@ -85,6 +85,16 @@ public void unsubscribeService(Service service, Subscriber subscriber, String cl ephemeralClientOperationService.unsubscribeService(service, subscriber, clientId); } + @Override + public void fuzzyWatch(String namespaceId, String serviceNamePattern, String groupNamePattern, String clientId) { + ephemeralClientOperationService.fuzzyWatch(namespaceId, serviceNamePattern, groupNamePattern, clientId); + } + + @Override + public void cancelFuzzyWatch(String namespaceId, String serviceNamePattern, String groupNamePattern, String clientId) { + ephemeralClientOperationService.cancelFuzzyWatch(namespaceId, serviceNamePattern, groupNamePattern, clientId); + } + private ClientOperationService chooseClientOperationService(final Instance instance) { return instance.isEphemeral() ? ephemeralClientOperationService : persistentClientOperationService; } diff --git a/naming/src/main/java/com/alibaba/nacos/naming/core/v2/service/impl/EphemeralClientOperationServiceImpl.java b/naming/src/main/java/com/alibaba/nacos/naming/core/v2/service/impl/EphemeralClientOperationServiceImpl.java index 5bcb460fc9d..1ef746435a0 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/core/v2/service/impl/EphemeralClientOperationServiceImpl.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/core/v2/service/impl/EphemeralClientOperationServiceImpl.java @@ -136,6 +136,35 @@ public void unsubscribeService(Service service, Subscriber subscriber, String cl client.setLastUpdatedTime(); NotifyCenter.publishEvent(new ClientOperationEvent.ClientUnsubscribeServiceEvent(singleton, clientId)); } + + @Override + public void fuzzyWatch(String namespaceId, String serviceNamePattern, String groupNamePattern, String clientId) { + String patternWithoutNamespace = NamingUtils.getGroupedName(serviceNamePattern, groupNamePattern); + // need store namespace id in server side + String completedPattern = NamingUtils.getPatternWithNamespace(namespaceId, patternWithoutNamespace); + Client client = clientManager.getClient(clientId); + if (!clientIsLegal(client, clientId)) { + return; + } + client.addWatchedPattern(completedPattern); + client.setLastUpdatedTime(); + NotifyCenter.publishEvent(new ClientOperationEvent.ClientFuzzyWatchEvent(completedPattern, clientId)); + } + + @Override + public void cancelFuzzyWatch(String namespaceId, String serviceNamePattern, String groupNamePattern, String clientId) { + String patternWithoutNamespace = NamingUtils.getGroupedName(serviceNamePattern, groupNamePattern); + String completedPattern = NamingUtils.getPatternWithNamespace(namespaceId, patternWithoutNamespace); + Client client = clientManager.getClient(clientId); + if (!clientIsLegal(client, clientId)) { + return; + } + client.removeWatchedPattern(completedPattern); + client.setLastUpdatedTime(); + NotifyCenter.publishEvent(new ClientOperationEvent.ClientCancelFuzzyWatchEvent(completedPattern, clientId)); + } + + private boolean clientIsLegal(Client client, String clientId) { private void checkClientIsLegal(Client client, String clientId) { if (client == null) { diff --git a/naming/src/main/java/com/alibaba/nacos/naming/core/v2/service/impl/PersistentClientOperationServiceImpl.java b/naming/src/main/java/com/alibaba/nacos/naming/core/v2/service/impl/PersistentClientOperationServiceImpl.java index 7cb0d32cc9e..d52d35d46d0 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/core/v2/service/impl/PersistentClientOperationServiceImpl.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/core/v2/service/impl/PersistentClientOperationServiceImpl.java @@ -182,6 +182,16 @@ public void unsubscribeService(Service service, Subscriber subscriber, String cl throw new UnsupportedOperationException("No persistent subscribers"); } + @Override + public void fuzzyWatch(String namespaceId, String serviceNamePattern, String groupNamePattern, String clientId) { + throw new UnsupportedOperationException("No persistent fuzzy watcher"); + } + + @Override + public void cancelFuzzyWatch(String namespaceId, String serviceNamePattern, String groupNamePattern, String clientId) { + throw new UnsupportedOperationException("No persistent fuzzy watcher"); + } + @Override public Response onRequest(ReadRequest request) { throw new UnsupportedOperationException("Temporary does not support"); diff --git a/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/heartbeat/ClientBeatProcessorV2.java b/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/heartbeat/ClientBeatProcessorV2.java index dee58b8c67f..d1125aac39a 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/heartbeat/ClientBeatProcessorV2.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/heartbeat/ClientBeatProcessorV2.java @@ -16,6 +16,7 @@ package com.alibaba.nacos.naming.healthcheck.heartbeat; +import com.alibaba.nacos.api.common.Constants; import com.alibaba.nacos.api.naming.utils.NamingUtils; import com.alibaba.nacos.common.notify.NotifyCenter; import com.alibaba.nacos.common.trace.event.naming.HealthStateChangeTraceEvent; @@ -67,7 +68,7 @@ public void run() { instance.setHealthy(true); Loggers.EVT_LOG.info("service: {} {POS} {IP-ENABLED} valid: {}:{}@{}, region: {}, msg: client beat ok", rsInfo.getServiceName(), ip, port, rsInfo.getCluster(), UtilsAndCommons.LOCALHOST_SITE); - NotifyCenter.publishEvent(new ServiceEvent.ServiceChangedEvent(service)); + NotifyCenter.publishEvent(new ServiceEvent.ServiceChangedEvent(service, Constants.ServiceChangedType.HEART_BEAT)); NotifyCenter.publishEvent(new ClientEvent.ClientChangedEvent(client)); NotifyCenter.publishEvent(new HealthStateChangeTraceEvent(System.currentTimeMillis(), service.getNamespace(), service.getGroup(), service.getName(), instance.getIp(), diff --git a/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/heartbeat/UnhealthyInstanceChecker.java b/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/heartbeat/UnhealthyInstanceChecker.java index 1614a5cf7aa..b3d65847d56 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/heartbeat/UnhealthyInstanceChecker.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/heartbeat/UnhealthyInstanceChecker.java @@ -76,7 +76,7 @@ private void changeHealthyStatus(Client client, Service service, HealthCheckInst .info("{POS} {IP-DISABLED} valid: {}:{}@{}@{}, region: {}, msg: client last beat: {}", instance.getIp(), instance.getPort(), instance.getCluster(), service.getName(), UtilsAndCommons.LOCALHOST_SITE, instance.getLastHeartBeatTime()); - NotifyCenter.publishEvent(new ServiceEvent.ServiceChangedEvent(service)); + NotifyCenter.publishEvent(new ServiceEvent.ServiceChangedEvent(service, Constants.ServiceChangedType.HEART_BEAT)); NotifyCenter.publishEvent(new ClientEvent.ClientChangedEvent(client)); NotifyCenter.publishEvent(new HealthStateChangeTraceEvent(System.currentTimeMillis(), service.getNamespace(), service.getGroup(), service.getName(), instance.getIp(), instance.getPort(), diff --git a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/NamingSubscriberServiceV2Impl.java b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/NamingSubscriberServiceV2Impl.java index 9dd25664f54..95d5fe481f9 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/NamingSubscriberServiceV2Impl.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/NamingSubscriberServiceV2Impl.java @@ -16,7 +16,9 @@ package com.alibaba.nacos.naming.push.v2; +import com.alibaba.nacos.api.common.Constants; import com.alibaba.nacos.api.naming.utils.NamingUtils; +import com.alibaba.nacos.api.utils.StringUtils; import com.alibaba.nacos.common.notify.Event; import com.alibaba.nacos.common.notify.NotifyCenter; import com.alibaba.nacos.common.notify.listener.SmartSubscriber; @@ -33,6 +35,9 @@ import com.alibaba.nacos.naming.pojo.Subscriber; import com.alibaba.nacos.naming.push.NamingSubscriberService; import com.alibaba.nacos.naming.push.v2.executor.PushExecutorDelegate; +import com.alibaba.nacos.naming.push.v2.task.FuzzyWatchInitDelayTask; +import com.alibaba.nacos.naming.push.v2.task.FuzzyWatchNotifyChangeDelayTask; +import com.alibaba.nacos.naming.push.v2.task.FuzzyWatchPushDelayTaskEngine; import com.alibaba.nacos.naming.push.v2.task.PushDelayTask; import com.alibaba.nacos.naming.push.v2.task.PushDelayTaskExecuteEngine; @@ -40,6 +45,7 @@ import java.util.HashSet; import java.util.LinkedList; import java.util.List; +import java.util.stream.Collectors; import java.util.stream.Stream; /** @@ -58,6 +64,8 @@ public class NamingSubscriberServiceV2Impl extends SmartSubscriber implements Na private final PushDelayTaskExecuteEngine delayTaskEngine; + private final FuzzyWatchPushDelayTaskEngine fuzzyWatchPushDelayTaskEngine; + public NamingSubscriberServiceV2Impl(ClientManagerDelegate clientManager, ClientServiceIndexesManager indexesManager, ServiceStorage serviceStorage, NamingMetadataManager metadataManager, PushExecutorDelegate pushExecutor, SwitchDomain switchDomain) { @@ -65,6 +73,8 @@ public NamingSubscriberServiceV2Impl(ClientManagerDelegate clientManager, this.indexesManager = indexesManager; this.delayTaskEngine = new PushDelayTaskExecuteEngine(clientManager, indexesManager, serviceStorage, metadataManager, pushExecutor, switchDomain); + this.fuzzyWatchPushDelayTaskEngine = new FuzzyWatchPushDelayTaskEngine(clientManager, indexesManager, + serviceStorage, metadataManager, pushExecutor, switchDomain); NotifyCenter.registerSubscriber(this, NamingEventPublisherFactory.getInstance()); } @@ -108,6 +118,7 @@ public List> subscribeTypes() { List> result = new LinkedList<>(); result.add(ServiceEvent.ServiceChangedEvent.class); result.add(ServiceEvent.ServiceSubscribedEvent.class); + result.add(ServiceEvent.ServiceFuzzyWatchInitEvent.class); return result; } @@ -118,6 +129,9 @@ public void onEvent(Event event) { ServiceEvent.ServiceChangedEvent serviceChangedEvent = (ServiceEvent.ServiceChangedEvent) event; Service service = serviceChangedEvent.getService(); delayTaskEngine.addTask(service, new PushDelayTask(service, PushConfig.getInstance().getPushTaskDelay())); + // watch notify push task specify by service + fuzzyWatchPushDelayTaskEngine.addTask(service, new FuzzyWatchNotifyChangeDelayTask(service, + serviceChangedEvent.getChangedType(), PushConfig.getInstance().getPushTaskDelay())); MetricsMonitor.incrementServiceChangeCount(service); } else if (event instanceof ServiceEvent.ServiceSubscribedEvent) { // If service is subscribed by one client, only push this client. @@ -125,7 +139,31 @@ public void onEvent(Event event) { Service service = subscribedEvent.getService(); delayTaskEngine.addTask(service, new PushDelayTask(service, PushConfig.getInstance().getPushTaskDelay(), subscribedEvent.getClientId())); + } else if (event instanceof ServiceEvent.ServiceFuzzyWatchInitEvent) { + ServiceEvent.ServiceFuzzyWatchInitEvent serviceFuzzyWatchInitEvent = (ServiceEvent.ServiceFuzzyWatchInitEvent) event; + + String completedPattern = serviceFuzzyWatchInitEvent.getPattern(); + String clientId = serviceFuzzyWatchInitEvent.getClientId(); + String taskKey = getTaskKey(clientId, completedPattern); + Collection matchedService = serviceFuzzyWatchInitEvent.getMatchedService().stream() + .map(Service::getGroupedServiceName) + .collect(Collectors.toSet()); + int originSize = matchedService.size(); + // watch init push task is specify by client id with pattern + // The key is just used to differentiate between different initialization tasks and merge them if needed. + fuzzyWatchPushDelayTaskEngine.addTask(taskKey, new FuzzyWatchInitDelayTask(taskKey, clientId, completedPattern, matchedService, + originSize, PushConfig.getInstance().getPushTaskDelay(), false)); + } + } + + private String getTaskKey(String clientId, String pattern) { + if (StringUtils.isBlank(clientId)) { + throw new IllegalArgumentException("Param 'clientId' is illegal, clientId is blank"); + } + if (StringUtils.isBlank(pattern)) { + throw new IllegalArgumentException("Param 'pattern' is illegal, pattern is blank"); } + return clientId + Constants.SERVICE_INFO_SPLITER + pattern; } private Stream getServiceStream() { diff --git a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/executor/PushExecutor.java b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/executor/PushExecutor.java index 21bcfab0d4b..553ac555a5f 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/executor/PushExecutor.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/executor/PushExecutor.java @@ -16,6 +16,8 @@ package com.alibaba.nacos.naming.push.v2.executor; +import com.alibaba.nacos.api.naming.remote.request.AbstractFuzzyWatchNotifyRequest; +import com.alibaba.nacos.api.remote.PushCallBack; import com.alibaba.nacos.naming.pojo.Subscriber; import com.alibaba.nacos.naming.push.v2.PushDataWrapper; import com.alibaba.nacos.naming.push.v2.task.NamingPushCallback; @@ -45,4 +47,23 @@ public interface PushExecutor { * @param callBack callback */ void doPushWithCallback(String clientId, Subscriber subscriber, PushDataWrapper data, NamingPushCallback callBack); + + + /** + * Do push to notify fuzzy watcher. + * + * @param clientId client id + * @param fuzzyWatchNotifyRequest request for fuzzy watch notification + */ + void doWatcherNotifyPush(String clientId, AbstractFuzzyWatchNotifyRequest fuzzyWatchNotifyRequest); + + /** + * Do push to notify fuzzy watcher with call back. + * + * @param clientId client id + * @param fuzzyWatchNotifyRequest request for fuzzy watch notification + * @param callBack callback + */ + void doFuzzyWatchNotifyPushWithCallBack(String clientId, AbstractFuzzyWatchNotifyRequest fuzzyWatchNotifyRequest, PushCallBack callBack); + } diff --git a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/executor/PushExecutorDelegate.java b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/executor/PushExecutorDelegate.java index f1fa7288e07..4acad457d27 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/executor/PushExecutorDelegate.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/executor/PushExecutorDelegate.java @@ -16,6 +16,8 @@ package com.alibaba.nacos.naming.push.v2.executor; +import com.alibaba.nacos.api.naming.remote.request.AbstractFuzzyWatchNotifyRequest; +import com.alibaba.nacos.api.remote.PushCallBack; import com.alibaba.nacos.naming.core.v2.client.impl.IpPortBasedClient; import com.alibaba.nacos.naming.pojo.Subscriber; import com.alibaba.nacos.naming.push.v2.PushDataWrapper; @@ -53,6 +55,19 @@ public void doPushWithCallback(String clientId, Subscriber subscriber, PushDataW getPushExecuteService(clientId, subscriber).doPushWithCallback(clientId, subscriber, data, callBack); } + @Override + public void doWatcherNotifyPush(String clientId, AbstractFuzzyWatchNotifyRequest watchNotifyRequest) { + // only support fuzzy watch by rpc + rpcPushExecuteService.doWatcherNotifyPush(clientId, watchNotifyRequest); + } + + @Override + public void doFuzzyWatchNotifyPushWithCallBack(String clientId, AbstractFuzzyWatchNotifyRequest watchNotifyRequest, + PushCallBack callBack) { + // only support fuzzy watch by rpc + rpcPushExecuteService.doFuzzyWatchNotifyPushWithCallBack(clientId, watchNotifyRequest, callBack); + } + private PushExecutor getPushExecuteService(String clientId, Subscriber subscriber) { Optional result = SpiImplPushExecutorHolder.getInstance() .findPushExecutorSpiImpl(clientId, subscriber); diff --git a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/executor/PushExecutorRpcImpl.java b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/executor/PushExecutorRpcImpl.java index 143d4715bdf..d540ca54223 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/executor/PushExecutorRpcImpl.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/executor/PushExecutorRpcImpl.java @@ -17,7 +17,9 @@ package com.alibaba.nacos.naming.push.v2.executor; import com.alibaba.nacos.api.naming.pojo.ServiceInfo; +import com.alibaba.nacos.api.naming.remote.request.AbstractFuzzyWatchNotifyRequest; import com.alibaba.nacos.api.naming.remote.request.NotifySubscriberRequest; +import com.alibaba.nacos.api.remote.PushCallBack; import com.alibaba.nacos.core.remote.RpcPushService; import com.alibaba.nacos.naming.misc.GlobalExecutor; import com.alibaba.nacos.naming.pojo.Subscriber; @@ -60,4 +62,15 @@ private ServiceInfo getServiceInfo(PushDataWrapper data, Subscriber subscriber) .selectInstancesWithHealthyProtection(data.getOriginalData(), data.getServiceMetadata(), false, true, subscriber); } + + @Override + public void doWatcherNotifyPush(String clientId, AbstractFuzzyWatchNotifyRequest watchNotifyRequest) { + pushService.pushWithoutAck(clientId, watchNotifyRequest); + } + + @Override + public void doFuzzyWatchNotifyPushWithCallBack(String clientId, AbstractFuzzyWatchNotifyRequest watchNotifyRequest, PushCallBack callBack) { + pushService.pushWithCallback(clientId, watchNotifyRequest, callBack, GlobalExecutor.getCallbackExecutor()); + } + } diff --git a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/executor/PushExecutorUdpImpl.java b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/executor/PushExecutorUdpImpl.java index c60478166d9..3324a3345d2 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/executor/PushExecutorUdpImpl.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/executor/PushExecutorUdpImpl.java @@ -17,7 +17,9 @@ package com.alibaba.nacos.naming.push.v2.executor; import com.alibaba.nacos.api.naming.pojo.ServiceInfo; +import com.alibaba.nacos.api.naming.remote.request.AbstractFuzzyWatchNotifyRequest; import com.alibaba.nacos.api.naming.utils.NamingUtils; +import com.alibaba.nacos.api.remote.PushCallBack; import com.alibaba.nacos.common.utils.StringUtils; import com.alibaba.nacos.naming.pojo.Subscriber; import com.alibaba.nacos.naming.push.UdpPushService; @@ -93,4 +95,14 @@ private ServiceInfo handleClusterData(ServiceInfo data, Subscriber subscriber) { return StringUtils.isBlank(subscriber.getCluster()) ? data : ServiceUtil.selectInstances(data, subscriber.getCluster()); } + + @Override + public void doWatcherNotifyPush(String clientId, AbstractFuzzyWatchNotifyRequest watchNotifyRequest) { + + } + + @Override + public void doFuzzyWatchNotifyPushWithCallBack(String clientId, AbstractFuzzyWatchNotifyRequest watchNotifyRequest, PushCallBack callBack) { + + } } diff --git a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/FuzzyWatchInitDelayTask.java b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/FuzzyWatchInitDelayTask.java new file mode 100644 index 00000000000..3e5f9c66c28 --- /dev/null +++ b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/FuzzyWatchInitDelayTask.java @@ -0,0 +1,91 @@ +/* + * Copyright 1999-2023 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.nacos.naming.push.v2.task; + +import com.alibaba.nacos.common.task.AbstractDelayTask; +import com.alibaba.nacos.naming.misc.Loggers; + +import java.util.Collection; + +/** + * Nacos naming fuzzy watch initial push delay task. + * + * @author tanyongquan + */ +public class FuzzyWatchInitDelayTask extends AbstractDelayTask { + + private final String taskKey; + + private final String clientId; + + private final String pattern; + + private final Collection matchedService; + + private final int originSize; + + private final boolean isFinishInit; + + public FuzzyWatchInitDelayTask(String taskKey, String clientId, String pattern, Collection matchedService, + int originSize, long delay, boolean isFinishInit) { + this.taskKey = taskKey; + this.clientId = clientId; + this.pattern = pattern; + this.matchedService = matchedService; + this.originSize = originSize; + this.isFinishInit = isFinishInit; + setTaskInterval(delay); + setLastProcessTime(System.currentTimeMillis()); + } + + @Override + public void merge(AbstractDelayTask task) { + if (!(task instanceof FuzzyWatchInitDelayTask)) { + return; + } + FuzzyWatchInitDelayTask oldTask = (FuzzyWatchInitDelayTask) task; + if (!isFinishInit) { + matchedService.addAll(oldTask.getMatchedService()); + } + setLastProcessTime(Math.min(getLastProcessTime(), task.getLastProcessTime())); + Loggers.PUSH.info("[FUZZY-WATCH-INIT-PUSH] Task merge for pattern {}", pattern); + } + + public String getTaskKey() { + return taskKey; + } + + public String getPattern() { + return pattern; + } + + public Collection getMatchedService() { + return matchedService; + } + + public boolean isFinishInit() { + return isFinishInit; + } + + public int getOriginSize() { + return originSize; + } + + public String getClientId() { + return clientId; + } +} diff --git a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/FuzzyWatchInitExecuteTask.java b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/FuzzyWatchInitExecuteTask.java new file mode 100644 index 00000000000..b4cb47469ea --- /dev/null +++ b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/FuzzyWatchInitExecuteTask.java @@ -0,0 +1,209 @@ +/* + * Copyright 1999-2023 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.nacos.naming.push.v2.task; + +import com.alibaba.nacos.api.naming.remote.request.FuzzyWatchNotifyInitRequest; +import com.alibaba.nacos.api.naming.utils.NamingUtils; +import com.alibaba.nacos.api.remote.PushCallBack; +import com.alibaba.nacos.common.task.AbstractExecuteTask; +import com.alibaba.nacos.naming.core.v2.client.Client; +import com.alibaba.nacos.naming.core.v2.client.manager.ClientManager; +import com.alibaba.nacos.naming.misc.Loggers; +import com.alibaba.nacos.naming.push.v2.NoRequiredRetryException; +import com.alibaba.nacos.naming.push.v2.PushConfig; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashSet; +import java.util.Set; + +/** + * Nacos naming fuzzy watch initial push execute task. + * + * @author tanyongquan + */ +public class FuzzyWatchInitExecuteTask extends AbstractExecuteTask { + + private final String taskKey; + + private final String clientId; + + private final String pattern; + + private final FuzzyWatchPushDelayTaskEngine delayTaskEngine; + + private final FuzzyWatchInitDelayTask delayTask; + + /** + * Fuzzy watch origin push matched service size, if there is no failure while executing push, {@code originSize == latch}. + * just use to record log after finish all push. + */ + private final int originSize; + + private final int latch; + + /** + * TODO set batch size from config. + */ + private final int batchSize = 10; + + private int sendCount; + + private boolean isFinishInitTask; + + private boolean haveFailPush; + + public FuzzyWatchInitExecuteTask(String taskKey, String clientId, String pattern, int originSize, + FuzzyWatchPushDelayTaskEngine delayTaskEngine, FuzzyWatchInitDelayTask delayTask, boolean isFinishInitTask) { + this.taskKey = taskKey; + this.clientId = clientId; + this.pattern = pattern; + this.delayTaskEngine = delayTaskEngine; + this.delayTask = delayTask; + this.originSize = originSize; + this.latch = delayTask.getMatchedService().size(); + this.sendCount = 0; + this.isFinishInitTask = isFinishInitTask; + this.haveFailPush = false; + } + + @Override + public void run() { + ClientManager clientManager = delayTaskEngine.getClientManager(); + Collection> dividedServices = divideServiceByBatch(delayTask.getMatchedService()); + Client client = clientManager.getClient(clientId); + String patternWithoutNameSpace = NamingUtils.getPatternRemovedNamespace(pattern); + String nameSpaceId = NamingUtils.getNamespaceFromPattern(pattern); + if (null == client) { + return; + } + if (!client.isWatchedPattern(pattern)) { + return; + } + if (isFinishInitTask || delayTask.getMatchedService().isEmpty()) { + // do not match any exist service, just finish init + delayTaskEngine.getPushExecutor().doFuzzyWatchNotifyPushWithCallBack(clientId, + FuzzyWatchNotifyInitRequest.buildInitFinishRequest(nameSpaceId, patternWithoutNameSpace), + new FuzzyWatchInitPushCallback(clientId, null, originSize, true, haveFailPush)); + } else { + for (Collection batchData : dividedServices) { + delayTaskEngine.getPushExecutor().doFuzzyWatchNotifyPushWithCallBack(clientId, FuzzyWatchNotifyInitRequest.buildInitRequest( + nameSpaceId, patternWithoutNameSpace, batchData), + new FuzzyWatchInitPushCallback(clientId, batchData, originSize, false, haveFailPush)); + } + } + + } + + private Collection> divideServiceByBatch(Collection matchedService) { + Collection> result = new ArrayList<>(); + if (matchedService.isEmpty()) { + return result; + } + Set currentBatch = new HashSet<>(); + for (String groupedServiceName : matchedService) { + currentBatch.add(groupedServiceName); + if (currentBatch.size() >= this.batchSize) { + result.add(currentBatch); + currentBatch = new HashSet<>(); + } + } + if (!currentBatch.isEmpty()) { + result.add(currentBatch); + } + return result; + } + + private class FuzzyWatchInitPushCallback implements PushCallBack { + + private final String clientId; + + private final Collection groupedServiceName; + + private final int originSize; + + /** + * Record the push task execute start time. + */ + private final long executeStartTime; + + private boolean isFinishInitTask; + + private boolean haveFailPush; + + private FuzzyWatchInitPushCallback(String clientId, Collection groupedServiceName, int originSize, + boolean isFinishInitTask, boolean haveFailPush) { + this.clientId = clientId; + this.groupedServiceName = groupedServiceName; + this.originSize = originSize; + this.executeStartTime = System.currentTimeMillis(); + this.isFinishInitTask = isFinishInitTask; + this.haveFailPush = haveFailPush; + } + + @Override + public long getTimeout() { + return PushConfig.getInstance().getPushTaskTimeout(); + } + + @Override + public void onSuccess() { + long pushFinishTime = System.currentTimeMillis(); + long pushCostTimeForNetWork = pushFinishTime - executeStartTime; + long pushCostTimeForAll = pushFinishTime - delayTask.getLastProcessTime(); + + if (isFinishInitTask) { + Loggers.PUSH.info("[FUZZY-WATCH-INIT-COMPLETE] {}ms, all delay time {}ms for client {} watch init push finish," + + " pattern {}, all push service size {}", + pushCostTimeForNetWork, pushCostTimeForAll, clientId, pattern, originSize); + } else { + Loggers.PUSH.info("[FUZZY-WATCH-PUSH-SUCC] {}ms, all delay time {}ms for client {}, pattern {}, push size {} : {}", + pushCostTimeForNetWork, pushCostTimeForAll, clientId, pattern, groupedServiceName.size(), + groupedServiceName); + sendCount += groupedServiceName.size(); + // this task is an init push task(not finish notify), and with no failure in this task when executing push batched services + if (!haveFailPush && sendCount >= latch) { + delayTaskEngine.addTask(taskKey, new FuzzyWatchInitDelayTask(taskKey, clientId, pattern, null, + originSize, PushConfig.getInstance().getPushTaskDelay(), true)); + } + } + } + + @Override + public void onFail(Throwable e) { + long pushCostTime = System.currentTimeMillis() - executeStartTime; + Loggers.PUSH.error("[FUZZY-WATCH-PUSH-FAIL] {}ms, pattern {} match {} service: {}, reason={}, client={}", pushCostTime, pattern, + groupedServiceName.size(), groupedServiceName, e.getMessage(), clientId); + setHaveFailPush(true); + if (!(e instanceof NoRequiredRetryException)) { + Loggers.PUSH.error("Reason detail: ", e); + if (isFinishInitTask) { + delayTaskEngine.addTask(taskKey, new FuzzyWatchInitDelayTask(taskKey, clientId, pattern, null, + originSize, PushConfig.getInstance().getPushTaskRetryDelay(), true)); + } else { + delayTaskEngine.addTask(taskKey, new FuzzyWatchInitDelayTask(taskKey, clientId, pattern, groupedServiceName, + originSize, PushConfig.getInstance().getPushTaskRetryDelay(), false)); + } + + } + } + } + + private void setHaveFailPush(boolean haveFailPush) { + this.haveFailPush = haveFailPush; + } +} \ No newline at end of file diff --git a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/FuzzyWatchNotifyChangeDelayTask.java b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/FuzzyWatchNotifyChangeDelayTask.java new file mode 100644 index 00000000000..e6e8a1893b7 --- /dev/null +++ b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/FuzzyWatchNotifyChangeDelayTask.java @@ -0,0 +1,90 @@ +/* + * Copyright 1999-2023 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.nacos.naming.push.v2.task; + +import com.alibaba.nacos.common.task.AbstractDelayTask; +import com.alibaba.nacos.naming.core.v2.pojo.Service; +import com.alibaba.nacos.naming.misc.Loggers; + +import java.util.HashSet; +import java.util.Set; + +/** + * Nacos naming fuzzy watch notify service change push delay task. + * + * @author tanyongquan + */ +public class FuzzyWatchNotifyChangeDelayTask extends AbstractDelayTask { + private final Service service; + + private final String serviceChangedType; + + private boolean pushToAll; + + private Set targetClients; + + public FuzzyWatchNotifyChangeDelayTask(Service service, String serviceChangedType, long delay) { + this.service = service; + this.serviceChangedType = serviceChangedType; + pushToAll = true; + targetClients = null; + setTaskInterval(delay); + setLastProcessTime(System.currentTimeMillis()); + } + + public FuzzyWatchNotifyChangeDelayTask(Service service, String serviceChangedType, long delay, String targetClient) { + this.service = service; + this.serviceChangedType = serviceChangedType; + this.pushToAll = false; + this.targetClients = new HashSet<>(1); + this.targetClients.add(targetClient); + setTaskInterval(delay); + setLastProcessTime(System.currentTimeMillis()); + } + + @Override + public void merge(AbstractDelayTask task) { + if (!(task instanceof FuzzyWatchNotifyChangeDelayTask)) { + return; + } + FuzzyWatchNotifyChangeDelayTask oldTask = (FuzzyWatchNotifyChangeDelayTask) task; + if (isPushToAll() || oldTask.isPushToAll()) { + pushToAll = true; + targetClients = null; + } else { + targetClients.addAll(oldTask.getTargetClients()); + } + setLastProcessTime(Math.min(getLastProcessTime(), task.getLastProcessTime())); + Loggers.PUSH.info("[FUZZY-WATCH-PUSH] Task merge for {}", service); + } + + public Service getService() { + return service; + } + + public boolean isPushToAll() { + return pushToAll; + } + + public String getServiceChangedType() { + return serviceChangedType; + } + + public Set getTargetClients() { + return targetClients; + } +} diff --git a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/FuzzyWatchNotifyChangeExecuteTask.java b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/FuzzyWatchNotifyChangeExecuteTask.java new file mode 100644 index 00000000000..49e1d7a07d2 --- /dev/null +++ b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/FuzzyWatchNotifyChangeExecuteTask.java @@ -0,0 +1,145 @@ +/* + * Copyright 1999-2023 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.nacos.naming.push.v2.task; + +import com.alibaba.nacos.api.naming.remote.request.FuzzyWatchNotifyChangeRequest; +import com.alibaba.nacos.api.remote.PushCallBack; +import com.alibaba.nacos.common.task.AbstractExecuteTask; +import com.alibaba.nacos.naming.core.v2.client.Client; +import com.alibaba.nacos.naming.core.v2.client.manager.ClientManager; +import com.alibaba.nacos.naming.core.v2.index.ClientServiceIndexesManager; +import com.alibaba.nacos.naming.core.v2.pojo.Service; +import com.alibaba.nacos.naming.misc.Loggers; +import com.alibaba.nacos.naming.push.v2.NoRequiredRetryException; +import com.alibaba.nacos.naming.push.v2.PushConfig; + +import java.util.Collection; +import java.util.HashSet; +import java.util.Set; + +/** + * Nacos naming fuzzy watch notify service change push execute task. + * + * @author tanyongquan + */ +public class FuzzyWatchNotifyChangeExecuteTask extends AbstractExecuteTask { + + private final Service service; + + private final FuzzyWatchPushDelayTaskEngine delayTaskEngine; + + private final FuzzyWatchNotifyChangeDelayTask delayTask; + + public FuzzyWatchNotifyChangeExecuteTask(Service service, FuzzyWatchPushDelayTaskEngine delayTaskEngine, + FuzzyWatchNotifyChangeDelayTask delayTask) { + this.service = service; + this.delayTaskEngine = delayTaskEngine; + this.delayTask = delayTask; + } + + @Override + public void run() { + try { + ClientManager clientManager = delayTaskEngine.getClientManager(); + String serviceChangedType = delayTask.getServiceChangedType(); + for (String clientId : getWatchTargetClientIds()) { + Client client = clientManager.getClient(clientId); + if (null == client) { + continue; + } + delayTaskEngine.getPushExecutor().doFuzzyWatchNotifyPushWithCallBack(clientId, new FuzzyWatchNotifyChangeRequest( + service.getNamespace(), service.getName(), service.getGroup(), serviceChangedType), + new WatchNotifyPushCallback(clientId, serviceChangedType)); + } + } catch (Exception e) { + Loggers.PUSH.error("Fuzzy watch notify task for service" + service.getGroupedServiceName() + " execute failed ", e); + delayTaskEngine.addTask(service, new FuzzyWatchNotifyChangeDelayTask(service, delayTask.getServiceChangedType(), 1000L)); + } + } + + /** + * get watch notify client id. + * + * @return A set of ClientID need to be notified + */ + private Set getWatchTargetClientIds() { + if (!delayTask.isPushToAll()) { + return delayTask.getTargetClients(); + } + Set watchNotifyClientIds = new HashSet<>(16); + ClientServiceIndexesManager indexesManager = delayTaskEngine.getIndexesManager(); + // get match result from index + Collection matchedPatterns = indexesManager.getServiceMatchedPatterns(service); + + for (String eachPattern : matchedPatterns) { + // for every matched pattern, get client id which watching this pattern + Collection clientIDs = indexesManager.getAllClientFuzzyWatchedPattern(eachPattern); + if (clientIDs == null || clientIDs.isEmpty()) { + // find there is nobody watch this pattern anymore (lazy remove) + indexesManager.removeWatchPatternMatchIndex(service, eachPattern); + continue; + } + watchNotifyClientIds.addAll(clientIDs); + } + return watchNotifyClientIds; + } + + private class WatchNotifyPushCallback implements PushCallBack { + + private final String clientId; + + private final String serviceChangedType; + + /** + * Record the push task execute start time. + */ + private final long executeStartTime; + + private WatchNotifyPushCallback(String clientId, String serviceChangedType) { + this.clientId = clientId; + this.serviceChangedType = serviceChangedType; + this.executeStartTime = System.currentTimeMillis(); + } + + @Override + public long getTimeout() { + return PushConfig.getInstance().getPushTaskTimeout(); + } + + @Override + public void onSuccess() { + long pushFinishTime = System.currentTimeMillis(); + long pushCostTimeForNetWork = pushFinishTime - executeStartTime; + long pushCostTimeForAll = pushFinishTime - delayTask.getLastProcessTime(); + + Loggers.PUSH.info("[WATCH-PUSH-SUCC] {}ms, all delay time {}ms for client {}, service {}, changed type {} ", + pushCostTimeForNetWork, pushCostTimeForAll, clientId, service.getGroupedServiceName(), serviceChangedType); + } + + @Override + public void onFail(Throwable e) { + long pushCostTime = System.currentTimeMillis() - executeStartTime; + Loggers.PUSH.error("[WATCH-PUSH-FAIL] {}ms, service {}, changed type {}, reason={}, client={}", pushCostTime, + service.getGroupedServiceName(), serviceChangedType, e.getMessage(), clientId); + if (!(e instanceof NoRequiredRetryException)) { + Loggers.PUSH.error("Reason detail: ", e); + delayTaskEngine.addTask(service, new FuzzyWatchNotifyChangeDelayTask(service, + serviceChangedType, PushConfig.getInstance().getPushTaskDelay(), clientId)); + } + } + } +} diff --git a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/FuzzyWatchPushDelayTaskEngine.java b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/FuzzyWatchPushDelayTaskEngine.java new file mode 100644 index 00000000000..dba23e8bab5 --- /dev/null +++ b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/FuzzyWatchPushDelayTaskEngine.java @@ -0,0 +1,120 @@ +/* + * Copyright 1999-2023 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.nacos.naming.push.v2.task; + +import com.alibaba.nacos.common.task.NacosTask; +import com.alibaba.nacos.common.task.NacosTaskProcessor; +import com.alibaba.nacos.common.task.engine.NacosDelayTaskExecuteEngine; +import com.alibaba.nacos.naming.core.v2.client.manager.ClientManager; +import com.alibaba.nacos.naming.core.v2.index.ClientServiceIndexesManager; +import com.alibaba.nacos.naming.core.v2.index.ServiceStorage; +import com.alibaba.nacos.naming.core.v2.metadata.NamingMetadataManager; +import com.alibaba.nacos.naming.core.v2.pojo.Service; +import com.alibaba.nacos.naming.misc.Loggers; +import com.alibaba.nacos.naming.misc.NamingExecuteTaskDispatcher; +import com.alibaba.nacos.naming.misc.SwitchDomain; +import com.alibaba.nacos.naming.push.v2.executor.PushExecutor; + +/** + * Nacos naming fuzzy watch notify service change push delay task execute engine. + * + * @author tanyongquan + */ +public class FuzzyWatchPushDelayTaskEngine extends NacosDelayTaskExecuteEngine { + + private final ClientManager clientManager; + + private final ClientServiceIndexesManager indexesManager; + + private final ServiceStorage serviceStorage; + + private final NamingMetadataManager metadataManager; + + private final PushExecutor pushExecutor; + + private final SwitchDomain switchDomain; + + public FuzzyWatchPushDelayTaskEngine(ClientManager clientManager, ClientServiceIndexesManager indexesManager, + ServiceStorage serviceStorage, NamingMetadataManager metadataManager, + PushExecutor pushExecutor, SwitchDomain switchDomain) { + super(FuzzyWatchPushDelayTaskEngine.class.getSimpleName(), Loggers.PUSH); + this.clientManager = clientManager; + this.indexesManager = indexesManager; + this.serviceStorage = serviceStorage; + this.metadataManager = metadataManager; + this.pushExecutor = pushExecutor; + this.switchDomain = switchDomain; + setDefaultTaskProcessor(new WatchPushDelayTaskProcessor(this)); + } + + public ClientManager getClientManager() { + return clientManager; + } + + public ClientServiceIndexesManager getIndexesManager() { + return indexesManager; + } + + public ServiceStorage getServiceStorage() { + return serviceStorage; + } + + public NamingMetadataManager getMetadataManager() { + return metadataManager; + } + + public PushExecutor getPushExecutor() { + return pushExecutor; + } + + @Override + protected void processTasks() { + if (!switchDomain.isPushEnabled()) { + return; + } + super.processTasks(); + } + + private static class WatchPushDelayTaskProcessor implements NacosTaskProcessor { + + private final FuzzyWatchPushDelayTaskEngine executeEngine; + + public WatchPushDelayTaskProcessor(FuzzyWatchPushDelayTaskEngine executeEngine) { + this.executeEngine = executeEngine; + } + + @Override + public boolean process(NacosTask task) { + if (task instanceof FuzzyWatchNotifyChangeDelayTask) { + FuzzyWatchNotifyChangeDelayTask notifyDelayTask = (FuzzyWatchNotifyChangeDelayTask) task; + Service service = notifyDelayTask.getService(); + NamingExecuteTaskDispatcher.getInstance() + .dispatchAndExecuteTask(service, new FuzzyWatchNotifyChangeExecuteTask(service, executeEngine, notifyDelayTask)); + } else if (task instanceof FuzzyWatchInitDelayTask) { + FuzzyWatchInitDelayTask fuzzyWatchInitDelayTask = (FuzzyWatchInitDelayTask) task; + String pattern = fuzzyWatchInitDelayTask.getPattern(); + String clientId = fuzzyWatchInitDelayTask.getClientId(); + String taskKey = fuzzyWatchInitDelayTask.getTaskKey(); + NamingExecuteTaskDispatcher.getInstance() + .dispatchAndExecuteTask(taskKey, new FuzzyWatchInitExecuteTask(taskKey, clientId, pattern, + fuzzyWatchInitDelayTask.getOriginSize(), executeEngine, fuzzyWatchInitDelayTask, + fuzzyWatchInitDelayTask.isFinishInit())); + } + return true; + } + } +} diff --git a/naming/src/main/java/com/alibaba/nacos/naming/remote/rpc/handler/FuzzyWatchRequestHandler.java b/naming/src/main/java/com/alibaba/nacos/naming/remote/rpc/handler/FuzzyWatchRequestHandler.java new file mode 100644 index 00000000000..bd7f07f0147 --- /dev/null +++ b/naming/src/main/java/com/alibaba/nacos/naming/remote/rpc/handler/FuzzyWatchRequestHandler.java @@ -0,0 +1,63 @@ +/* + * Copyright 1999-2023 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.nacos.naming.remote.rpc.handler; + +import com.alibaba.nacos.api.exception.NacosException; +import com.alibaba.nacos.api.naming.remote.NamingRemoteConstants; +import com.alibaba.nacos.api.naming.remote.request.FuzzyWatchRequest; +import com.alibaba.nacos.api.naming.remote.response.FuzzyWatchResponse; +import com.alibaba.nacos.api.remote.request.RequestMeta; +import com.alibaba.nacos.auth.annotation.Secured; +import com.alibaba.nacos.core.remote.RequestHandler; +import com.alibaba.nacos.naming.core.v2.service.impl.EphemeralClientOperationServiceImpl; +import com.alibaba.nacos.plugin.auth.constant.ActionTypes; +import org.springframework.stereotype.Component; + +/** + * Fuzzy watch service request handler. + * + * @author tanyongquan + */ +@Component("fuzzyWatchRequestHandler") +public class FuzzyWatchRequestHandler extends RequestHandler { + + private final EphemeralClientOperationServiceImpl clientOperationService; + + public FuzzyWatchRequestHandler(EphemeralClientOperationServiceImpl clientOperationService) { + this.clientOperationService = clientOperationService; + } + + @Override + @Secured(action = ActionTypes.READ) + public FuzzyWatchResponse handle(FuzzyWatchRequest request, RequestMeta meta) throws NacosException { + String serviceNamePattern = request.getServiceName(); + String groupNamePattern = request.getGroupName(); + String namespaceId = request.getNamespace(); + + switch (request.getType()) { + case NamingRemoteConstants.FUZZY_WATCH_SERVICE: + clientOperationService.fuzzyWatch(namespaceId, serviceNamePattern, groupNamePattern, meta.getConnectionId()); + return FuzzyWatchResponse.buildSuccessResponse(NamingRemoteConstants.FUZZY_WATCH_SERVICE); + case NamingRemoteConstants.CANCEL_FUZZY_WATCH_SERVICE: + clientOperationService.cancelFuzzyWatch(namespaceId, serviceNamePattern, groupNamePattern, meta.getConnectionId()); + return FuzzyWatchResponse.buildSuccessResponse(NamingRemoteConstants.CANCEL_FUZZY_WATCH_SERVICE); + default: + throw new NacosException(NacosException.INVALID_PARAM, + String.format("Unsupported request type %s", request.getType())); + } + } +} diff --git a/naming/src/test/java/com/alibaba/nacos/naming/core/v2/index/ClientServiceIndexesManagerTest.java b/naming/src/test/java/com/alibaba/nacos/naming/core/v2/index/ClientServiceIndexesManagerTest.java index 0c9bc8676a4..b83a9ea6604 100644 --- a/naming/src/test/java/com/alibaba/nacos/naming/core/v2/index/ClientServiceIndexesManagerTest.java +++ b/naming/src/test/java/com/alibaba/nacos/naming/core/v2/index/ClientServiceIndexesManagerTest.java @@ -119,8 +119,9 @@ void testRemovePublisherIndexesByEmptyService() throws NoSuchFieldException, Ill void testSubscribeTypes() { List> classes = clientServiceIndexesManager.subscribeTypes(); + assertNotNull(classes); - assertEquals(5, classes.size()); + assertEquals(7, classes.size()); } @Test diff --git a/naming/src/test/java/com/alibaba/nacos/naming/push/v2/NamingSubscriberServiceV2ImplTest.java b/naming/src/test/java/com/alibaba/nacos/naming/push/v2/NamingSubscriberServiceV2ImplTest.java index 84db805b5e5..2ee1f4c0386 100644 --- a/naming/src/test/java/com/alibaba/nacos/naming/push/v2/NamingSubscriberServiceV2ImplTest.java +++ b/naming/src/test/java/com/alibaba/nacos/naming/push/v2/NamingSubscriberServiceV2ImplTest.java @@ -16,6 +16,7 @@ package com.alibaba.nacos.naming.push.v2; +import com.alibaba.nacos.api.common.Constants; import com.alibaba.nacos.naming.core.v2.client.Client; import com.alibaba.nacos.naming.core.v2.client.manager.ClientManagerDelegate; import com.alibaba.nacos.naming.core.v2.event.service.ServiceEvent; @@ -116,8 +117,8 @@ void testGetFuzzySubscribersByService() { } @Test - void onEvent() { - subscriberService.onEvent(new ServiceEvent.ServiceChangedEvent(service)); + public void onEvent() { + subscriberService.onEvent(new ServiceEvent.ServiceChangedEvent(service, Constants.ServiceChangedType.ADD_SERVICE)); verify(delayTaskEngine).addTask(eq(service), any(PushDelayTask.class)); } } diff --git a/naming/src/test/java/com/alibaba/nacos/naming/push/v2/task/FixturePushExecutor.java b/naming/src/test/java/com/alibaba/nacos/naming/push/v2/task/FixturePushExecutor.java index f26fb516a50..2a28f0196ad 100644 --- a/naming/src/test/java/com/alibaba/nacos/naming/push/v2/task/FixturePushExecutor.java +++ b/naming/src/test/java/com/alibaba/nacos/naming/push/v2/task/FixturePushExecutor.java @@ -16,6 +16,8 @@ package com.alibaba.nacos.naming.push.v2.task; +import com.alibaba.nacos.api.naming.remote.request.AbstractFuzzyWatchNotifyRequest; +import com.alibaba.nacos.api.remote.PushCallBack; import com.alibaba.nacos.naming.pojo.Subscriber; import com.alibaba.nacos.naming.push.v2.PushDataWrapper; import com.alibaba.nacos.naming.push.v2.executor.PushExecutor; @@ -39,6 +41,21 @@ public void doPushWithCallback(String clientId, Subscriber subscriber, PushDataW } } + @Override + public void doWatcherNotifyPush(String clientId, AbstractFuzzyWatchNotifyRequest watchNotifyRequest) { + + } + + @Override + public void doFuzzyWatchNotifyPushWithCallBack(String clientId, AbstractFuzzyWatchNotifyRequest watchNotifyRequest, + PushCallBack callBack) { + if (shouldSuccess) { + callBack.onSuccess(); + } else { + callBack.onFail(failedException); + } + } + public void setShouldSuccess(boolean shouldSuccess) { this.shouldSuccess = shouldSuccess; }