Skip to content

Commit d69823e

Browse files
KomachiSionKomachiSion
authored
Refactor Instance Controller (#3864)
* Move some old logic code to InstanceOperator * Support sync ip port client
1 parent 1c935d1 commit d69823e

14 files changed

+419
-111
lines changed

‎naming/src/main/java/com/alibaba/nacos/naming/consistency/ephemeral/distro/v2/DistroClientDataProcessor.java

+1-3
Original file line numberDiff line numberDiff line change
@@ -111,9 +111,7 @@ public boolean processData(DistroData distroData) {
111111
}
112112

113113
private void handlerClientSyncData(ClientSyncData clientSyncData) {
114-
if (!clientManager.allClientId().contains(clientSyncData.getClientId())) {
115-
clientManager.clientConnected(new ConnectionBasedClient(clientSyncData.getClientId(), false));
116-
}
114+
clientManager.syncClientConnected(clientSyncData.getClientId());
117115
Client client = clientManager.getClient(clientSyncData.getClientId());
118116
upgradeClient(client, clientSyncData);
119117
}

‎naming/src/main/java/com/alibaba/nacos/naming/controllers/InstanceController.java

+14-60
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525
import com.alibaba.nacos.common.utils.JacksonUtils;
2626
import com.alibaba.nacos.core.utils.WebUtils;
2727
import com.alibaba.nacos.naming.core.Instance;
28-
import com.alibaba.nacos.naming.core.InstanceService;
28+
import com.alibaba.nacos.naming.core.InstanceOperatorClientImpl;
2929
import com.alibaba.nacos.naming.core.Service;
3030
import com.alibaba.nacos.naming.core.ServiceManager;
3131
import com.alibaba.nacos.naming.healthcheck.RsInfo;
@@ -71,7 +71,6 @@
7171
*/
7272
@RestController
7373
@RequestMapping(UtilsAndCommons.NACOS_NAMING_CONTEXT + "/instance")
74-
@SuppressWarnings("PMD.RemoveCommentedCodeRule")
7574
public class InstanceController {
7675

7776
@Autowired
@@ -84,8 +83,12 @@ public class InstanceController {
8483
private ServiceManager serviceManager;
8584

8685
@Autowired
87-
private InstanceService instanceService;
86+
private InstanceOperatorClientImpl instanceService;
8887

88+
/**
89+
* Move to {@link com.alibaba.nacos.naming.core.InstanceOperatorServiceImpl}.
90+
*/
91+
@Deprecated
8992
private DataSource pushDataSource = new DataSource() {
9093

9194
@Override
@@ -126,7 +129,6 @@ public String register(HttpServletRequest request) throws Exception {
126129

127130
final Instance instance = parseInstance(request);
128131

129-
// serviceManager.registerInstance(namespaceId, serviceName, instance);
130132
instanceService.registerInstance(namespaceId, serviceName, instance);
131133
return "ok";
132134
}
@@ -147,13 +149,6 @@ public String deregister(HttpServletRequest request) throws Exception {
147149
String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
148150
checkServiceNameFormat(serviceName);
149151

150-
// Service service = serviceManager.getService(namespaceId, serviceName);
151-
// if (service == null) {
152-
// Loggers.SRV_LOG.warn("remove instance from non-exist service: {}", serviceName);
153-
// return "ok";
154-
// }
155-
// serviceManager.removeInstance(namespaceId, serviceName, instance.isEphemeral(), instance);
156-
157152
instanceService.removeInstance(namespaceId, serviceName, instance);
158153
return "ok";
159154
}
@@ -259,20 +254,17 @@ public Object list(HttpServletRequest request) throws Exception {
259254
String clusters = WebUtils.optional(request, "clusters", StringUtils.EMPTY);
260255
String clientIP = WebUtils.optional(request, "clientIP", StringUtils.EMPTY);
261256
int udpPort = Integer.parseInt(WebUtils.optional(request, "udpPort", "0"));
262-
String env = WebUtils.optional(request, "env", StringUtils.EMPTY);
257+
boolean healthyOnly = Boolean.parseBoolean(WebUtils.optional(request, "healthyOnly", "false"));
258+
263259
boolean isCheck = Boolean.parseBoolean(WebUtils.optional(request, "isCheck", "false"));
264260

265261
String app = WebUtils.optional(request, "app", StringUtils.EMPTY);
266-
262+
String env = WebUtils.optional(request, "env", StringUtils.EMPTY);
267263
String tenant = WebUtils.optional(request, "tid", StringUtils.EMPTY);
268264

269-
boolean healthyOnly = Boolean.parseBoolean(WebUtils.optional(request, "healthyOnly", "false"));
270-
271265
Subscriber subscriber =
272-
udpPort > 0 ? new Subscriber(clientIP + ":" + udpPort, agent, app, clientIP, namespaceId, serviceName)
273-
: null;
274-
// return doSrvIpxt(namespaceId, serviceName, agent, clusters, clientIP, udpPort, env, isCheck, app, tenant,
275-
// healthyOnly);
266+
udpPort > 0 ? new Subscriber(clientIP + ":" + udpPort, agent, app, clientIP, namespaceId, serviceName,
267+
udpPort) : null;
276268
return instanceService.listInstance(namespaceId, serviceName, subscriber, clusters, healthyOnly);
277269
}
278270

@@ -365,47 +357,6 @@ public ObjectNode beat(HttpServletRequest request) throws Exception {
365357
checkServiceNameFormat(serviceName);
366358
Loggers.SRV_LOG.debug("[CLIENT-BEAT] full arguments: beat: {}, serviceName: {}", clientBeat, serviceName);
367359

368-
// Instance instance = serviceManager.getInstance(namespaceId, serviceName, clusterName, ip, port);
369-
//
370-
// if (instance == null) {
371-
// if (clientBeat == null) {
372-
// result.put(CommonParams.CODE, NamingResponseCode.RESOURCE_NOT_FOUND);
373-
// return result;
374-
// }
375-
//
376-
// Loggers.SRV_LOG.warn("[CLIENT-BEAT] The instance has been removed for health mechanism, "
377-
// + "perform data compensation operations, beat: {}, serviceName: {}", clientBeat, serviceName);
378-
//
379-
// instance = new Instance();
380-
// instance.setPort(clientBeat.getPort());
381-
// instance.setIp(clientBeat.getIp());
382-
// instance.setWeight(clientBeat.getWeight());
383-
// instance.setMetadata(clientBeat.getMetadata());
384-
// instance.setClusterName(clusterName);
385-
// instance.setServiceName(serviceName);
386-
// instance.setInstanceId(instance.getInstanceId());
387-
// instance.setEphemeral(clientBeat.isEphemeral());
388-
//
389-
// serviceManager.registerInstance(namespaceId, serviceName, instance);
390-
// }
391-
//
392-
// Service service = serviceManager.getService(namespaceId, serviceName);
393-
//
394-
// if (service == null) {
395-
// throw new NacosException(NacosException.SERVER_ERROR,
396-
// "service not found: " + serviceName + "@" + namespaceId);
397-
// }
398-
// if (clientBeat == null) {
399-
// clientBeat = new RsInfo();
400-
// clientBeat.setIp(ip);
401-
// clientBeat.setPort(port);
402-
// clientBeat.setCluster(clusterName);
403-
// }
404-
// service.processClientBeat(clientBeat);
405-
// result.put(CommonParams.CODE, NamingResponseCode.OK);
406-
// if (instance.containsMetadata(PreservedMetadataKeys.HEART_BEAT_INTERVAL)) {
407-
// result.put(SwitchEntry.CLIENT_BEAT_INTERVAL, instance.getInstanceHeartBeatInterval());
408-
// }
409360
int resultCode = instanceService.handleBeat(namespaceId, serviceName, ip, port, clusterName, clientBeat);
410361
result.put(CommonParams.CODE, resultCode);
411362
result.put(SwitchEntry.CLIENT_BEAT_INTERVAL,
@@ -550,7 +501,10 @@ private void checkIfDisabled(Service service) throws Exception {
550501
* @param healthyOnly whether only for healthy check
551502
* @return service full information with instances
552503
* @throws Exception any error during handle
504+
* @deprecated will be replace by {@link com.alibaba.nacos.naming.core.InstanceOperator#listInstance(String, String,
505+
* Subscriber, String, boolean)}
553506
*/
507+
@Deprecated
554508
public ObjectNode doSrvIpxt(String namespaceId, String serviceName, String agent, String clusters, String clientIP,
555509
int udpPort, String env, boolean isCheck, String app, String tid, boolean healthyOnly) throws Exception {
556510

‎naming/src/main/java/com/alibaba/nacos/naming/core/InstanceOperator.java

+92
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,92 @@
1+
/*
2+
* Copyright 1999-2018 Alibaba Group Holding Ltd.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.alibaba.nacos.naming.core;
18+
19+
import com.alibaba.nacos.api.exception.NacosException;
20+
import com.alibaba.nacos.api.naming.pojo.Instance;
21+
import com.alibaba.nacos.api.naming.pojo.ServiceInfo;
22+
import com.alibaba.nacos.naming.healthcheck.RsInfo;
23+
import com.alibaba.nacos.naming.pojo.Subscriber;
24+
25+
/**
26+
* Instance operator.
27+
*
28+
* @author xiweng.yy
29+
*/
30+
public interface InstanceOperator {
31+
32+
/**
33+
* Register an instance to a service in AP mode.
34+
*
35+
* @param namespaceId id of namespace
36+
* @param serviceName service name
37+
* @param instance instance to register
38+
* @throws NacosException nacos exception when register failed
39+
*/
40+
void registerInstance(String namespaceId, String serviceName, Instance instance) throws NacosException;
41+
42+
/**
43+
* Remove instance from service.
44+
*
45+
* @param namespaceId namespace
46+
* @param serviceName service name
47+
* @param instance instance
48+
* @throws NacosException nacos exception when remove failed
49+
*/
50+
void removeInstance(String namespaceId, String serviceName, Instance instance) throws NacosException;
51+
52+
/**
53+
* Get all instance of input service.
54+
*
55+
* @param namespaceId namespace
56+
* @param serviceName service name
57+
* @param subscriber subscriber info
58+
* @param cluster cluster of instances
59+
* @param healthOnly whether only return health instances
60+
* @return service info
61+
* @throws Exception exception when list instance failed
62+
*/
63+
ServiceInfo listInstance(String namespaceId, String serviceName, Subscriber subscriber, String cluster,
64+
boolean healthOnly) throws Exception;
65+
66+
/**
67+
* Handle beat request.
68+
*
69+
* @param namespaceId namespace
70+
* @param serviceName service name
71+
* @param ip ip of instance
72+
* @param port port of instance
73+
* @param cluster cluster of instance
74+
* @param clientBeat client beat info
75+
* @return result code
76+
* @throws NacosException nacos exception when service non-exist and client beat info is null
77+
*/
78+
int handleBeat(String namespaceId, String serviceName, String ip, int port, String cluster, RsInfo clientBeat)
79+
throws NacosException;
80+
81+
/**
82+
* Get heart beat interval for specified instance.
83+
*
84+
* @param namespaceId namespace
85+
* @param serviceName service name
86+
* @param ip ip of instance
87+
* @param port port of instance
88+
* @param cluster cluster of instance
89+
* @return heart beat interval
90+
*/
91+
long getHeartBeatInterval(String namespaceId, String serviceName, String ip, int port, String cluster);
92+
}

‎naming/src/main/java/com/alibaba/nacos/naming/core/InstanceService.java renamed to ‎naming/src/main/java/com/alibaba/nacos/naming/core/InstanceOperatorClientImpl.java

+9-39
Original file line numberDiff line numberDiff line change
@@ -40,30 +40,25 @@
4040
* @author xiweng.yy
4141
*/
4242
@org.springframework.stereotype.Service
43-
public class InstanceService {
43+
public class InstanceOperatorClientImpl implements InstanceOperator {
4444

4545
private final IpPortBasedClientManager ipPortBasedClientManager;
4646

4747
private final ClientOperationService clientOperationService;
4848

4949
private final ServiceStorage serviceStorage;
5050

51-
public InstanceService(IpPortBasedClientManager ipPortBasedClientManager,
51+
public InstanceOperatorClientImpl(IpPortBasedClientManager ipPortBasedClientManager,
5252
ClientOperationService clientOperationService, ServiceStorage serviceStorage) {
5353
this.ipPortBasedClientManager = ipPortBasedClientManager;
5454
this.clientOperationService = clientOperationService;
5555
this.serviceStorage = serviceStorage;
5656
}
5757

5858
/**
59-
* Register an instance to a service in AP mode.
60-
*
61-
* <p>This method creates {@code IpPortBasedClient} if it don't exist.
62-
*
63-
* @param namespaceId id of namespace
64-
* @param serviceName service name
65-
* @param instance instance to register
59+
* This method creates {@code IpPortBasedClient} if it don't exist.
6660
*/
61+
@Override
6762
public void registerInstance(String namespaceId, String serviceName, Instance instance) {
6863
String clientId = instance.toInetAddr();
6964
createIpPortClientIfAbsent(clientId, instance.isEphemeral());
@@ -73,13 +68,7 @@ public void registerInstance(String namespaceId, String serviceName, Instance in
7368
clientOperationService.registerInstance(service, instance, clientId);
7469
}
7570

76-
/**
77-
* Remove instance from service.
78-
*
79-
* @param namespaceId namespace
80-
* @param serviceName service name
81-
* @param instance instance
82-
*/
71+
@Override
8372
public void removeInstance(String namespaceId, String serviceName, Instance instance) {
8473
String clientId = instance.toInetAddr();
8574
if (!ipPortBasedClientManager.allClientId().contains(clientId)) {
@@ -92,21 +81,12 @@ public void removeInstance(String namespaceId, String serviceName, Instance inst
9281
clientOperationService.deregisterInstance(service, instance, clientId);
9382
}
9483

95-
/**
96-
* Get all instance of input service.
97-
*
98-
* @param namespaceId namespace
99-
* @param serviceName service name
100-
* @param subscriber subscriber info
101-
* @param cluster cluster of instances
102-
* @param healthOnly whether only return health instances
103-
* @return service info
104-
*/
84+
@Override
10585
public ServiceInfo listInstance(String namespaceId, String serviceName, Subscriber subscriber, String cluster,
10686
boolean healthOnly) {
10787
String groupName = NamingUtils.getGroupName(serviceName);
10888
String serviceNameNoGrouped = NamingUtils.getServiceName(serviceName);
109-
Service service = Service.newService(namespaceId, groupName, serviceNameNoGrouped, true);
89+
Service service = Service.newService(namespaceId, groupName, serviceNameNoGrouped);
11090
if (null != subscriber) {
11191
createIpPortClientIfAbsent(subscriber.getAddrStr(), true);
11292
clientOperationService.subscribeService(service, subscriber, subscriber.getAddrStr());
@@ -115,18 +95,7 @@ public ServiceInfo listInstance(String namespaceId, String serviceName, Subscrib
11595
return ServiceUtil.filterInstances(serviceInfo, cluster, healthOnly);
11696
}
11797

118-
/**
119-
* Handle beat request.
120-
*
121-
* @param namespaceId namespace
122-
* @param serviceName service name
123-
* @param ip ip of instance
124-
* @param port port of instance
125-
* @param cluster cluster of instance
126-
* @param clientBeat client beat info
127-
* @return result code
128-
* @throws NacosException nacos exception when service non-exist and client beat info is null
129-
*/
98+
@Override
13099
public int handleBeat(String namespaceId, String serviceName, String ip, int port, String cluster,
131100
RsInfo clientBeat) throws NacosException {
132101
String groupName = NamingUtils.getGroupName(serviceName);
@@ -167,6 +136,7 @@ public int handleBeat(String namespaceId, String serviceName, String ip, int por
167136
return NamingResponseCode.OK;
168137
}
169138

139+
@Override
170140
public long getHeartBeatInterval(String namespaceId, String serviceName, String ip, int port, String cluster) {
171141
// TODO Get heart beat interval from CP metadata
172142
return 5000L;

0 commit comments

Comments
 (0)