Nacos Client Configuration Loading Mechanism
The externalized configuration implementation in Nacos is primarily handled by the NacosPropertySourceLocator. This component is responsible for loading configuraton data from Nacos and making it available to the Spring application context.
The configuration loading process involves several key steps:
- Configuration Retrieval: The client attempts to fetch configuration data based on a specified
dataId,group, and optionaltenant. - Local Cache Check: Before making a remote call, the client checks its local cache for the requested configuration.
- Remote Server Call: If the configuration is not found in the local cache, the client makes a request to the Nacos server.
- Snapshot Fallback: In case of server errors, the client falls back to a local snapshot of the configuration.
The core method for retrieving configuration is configService.getConfig, which is part of the Nacos SDK.
String getConfig(String dataId, String group, long timeoutMs) throws NacosException;
For more details on the Nacos SDK, refer to the official documentation: https://nacos.io/zh-cn/docs/sdk.html
NacosConfigService Configuration Retrieval
The NacosConfigService.getConfig method initiates the configuration retrieval process. It delegates the actual work to an internal method, getConfigInner.
@Override
public String getConfig(String dataId, String group, long timeoutMs) throws NacosException {
return getConfigInner(namespace, dataId, group, timeoutMs);
}
The getConfigInner method orchestrates the entire loading process:
private String getConfigInner(String tenant, String dataId, String group, long timeoutMs) throws NacosException {
// Ensure a default group if none is provided
group = blank2defaultGroup(group);
// Validate input parameters
ParamUtils.checkKeyParam(dataId, group);
// Prepare the response object
ConfigResponse cr = new ConfigResponse();
cr.setDataId(dataId);
cr.setTenant(tenant);
cr.setGroup(group);
// Attempt to load from local failover cache
String content = LocalConfigInfoProcessor.getFailover(worker.getAgentName(), dataId, group, tenant);
if (content != null) {
LOGGER.warn("[{}] [get-config] Retrieved from failover cache, dataId={}, group={}, tenant={}, config={}",
worker.getAgentName(), dataId, group, tenant, ContentUtils.truncateContent(content));
cr.setContent(content);
String encryptedDataKey = LocalEncryptedDataKeyProcessor
.getEncryptDataKeyFailover(agent.getName(), dataId, group, tenant);
cr.setEncryptedDataKey(encryptedDataKey);
configFilterChainManager.doFilter(null, cr);
return cr.getContent();
}
// If not in failover, attempt to fetch from the server
try {
ConfigResponse response = worker.getServerConfig(dataId, group, tenant, timeoutMs, false);
cr.setContent(response.getContent());
cr.setEncryptedDataKey(response.getEncryptedDataKey());
configFilterChainManager.doFilter(null, cr);
return cr.getContent();
} catch (NacosException ioe) {
if (NacosException.NO_RIGHT == ioe.getErrCode()) {
throw ioe;
}
LOGGER.warn("[{}] [get-config] Server retrieval failed, dataId={}, group={}, tenant={}, msg={}",
worker.getAgentName(), dataId, group, tenant, ioe.toString());
}
// Fallback to local snapshot if server call fails
content = LocalConfigInfoProcessor.getSnapshot(worker.getAgentName(), dataId, group, tenant);
if (content != null) {
LOGGER.warn("[{}] [get-config] Retrieved from local snapshot, dataId={}, group={}, tenant={}, config={}",
worker.getAgentName(), dataId, group, tenant, ContentUtils.truncateContent(content));
}
cr.setContent(content);
String encryptedDataKey = LocalEncryptedDataKeyProcessor
.getEncryptDataKeySnapshot(agent.getName(), dataId, group, tenant);
cr.setEncryptedDataKey(encryptedDataKey);
configFilterChainManager.doFilter(null, cr);
return cr.getContent();
}
Local Cache Retrieval
By default, Nacos first checks the local cache for configurasion data. The cache is typically located in a directory structure like:
C:\Users\snail\nacos\config\fixed-121.5.102.117_8848_nacos
If the configuration exists in the local cache, it is returned; otherwise, an empty result is returned.
public static String getFailover(String serverName, String dataId, String group, String tenant) {
File localPath = getFailoverFile(serverName, dataId, group, tenant);
if (!localPath.exists() || !localPath.isFile()) {
return null;
}
try {
return readFile(localPath);
} catch (IOException ioe) {
LOGGER.error("[" + serverName + "] Failed to read failover file, " + localPath, ioe);
return null;
}
}
The method constructs the file path based on the server name, data ID, group, and tenant.
static File getFailoverFile(String serverName, String dataId, String group, String tenant) {
File tmp = new File(LOCAL_SNAPSHOT_PATH, serverName + SUFFIX);
tmp = new File(tmp, FAILOVER_FILE_CHILD_1);
if (StringUtils.isBlank(tenant)) {
tmp = new File(tmp, FAILOVER_FILE_CHILD_2);
} else {
tmp = new File(tmp, FAILOVER_FILE_CHILD_3);
tmp = new File(tmp, tenant);
}
return new File(new File(tmp, group), dataId);
}
Remote Server Interaction
The ClientWorker class is responsible for interacting with the Nacos server. It uses the getServerConfig method to fetch configuration data.
public ConfigResponse getServerConfig(String dataId, String group, String tenant, long readTimeout, boolean notify)
throws NacosException {
if (StringUtils.isBlank(group)) {
group = Constants.DEFAULT_GROUP;
}
return this.agent.queryConfig(dataId, group, tenant, readTimeout, notify);
}
The agent.queryConfig method handles the actual remote call. In Nacos 2.x, this interaction is performed using gRPC, a significant change from the HTTP-based approach in Nacos 1.x.
@Override
public ConfigResponse queryConfig(String dataId, String group, String tenant, long readTimeouts, boolean notify)
throws NacosException {
ConfigQueryRequest request = ConfigQueryRequest.build(dataId, group, tenant);
request.putHeader(NOTIFY_HEADER, String.valueOf(notify));
RpcClient rpcClient = getOneRunningClient();
if (notify) {
CacheData cacheData = cacheMap.get().get(GroupKey.getKeyTenant(dataId, group, tenant));
if (cacheData != null) {
rpcClient = ensureRpcClient(String.valueOf(cacheData.getTaskId()));
}
}
ConfigQueryResponse response = (ConfigQueryResponse) requestProxy(rpcClient, request, readTimeouts);
ConfigResponse configResponse = new ConfigResponse();
if (response.isSuccess()) {
LocalConfigInfoProcessor.saveSnapshot(this.getName(), dataId, group, tenant, response.getContent());
configResponse.setContent(response.getContent());
String configType = StringUtils.isNotBlank(response.getContentType()) ? response.getContentType() : ConfigType.TEXT.getType();
configResponse.setConfigType(configType);
String encryptedDataKey = response.getEncryptedDataKey();
LocalEncryptedDataKeyProcessor.saveEncryptDataKeySnapshot(agent.getName(), dataId, group, tenant, encryptedDataKey);
configResponse.setEncryptedDataKey(encryptedDataKey);
return configResponse;
} else if (response.getErrorCode() == ConfigQueryResponse.CONFIG_NOT_FOUND) {
LocalConfigInfoProcessor.saveSnapshot(this.getName(), dataId, group, tenant, null);
LocalEncryptedDataKeyProcessor.saveEncryptDataKeySnapshot(agent.getName(), dataId, group, tenant, null);
return configResponse;
} else if (response.getErrorCode() == ConfigQueryResponse.CONFIG_QUERY_CONFLICT) {
LOGGER.error(
"[{}] [sub-server-error] Configuration being modified concurrently, dataId={}, group={}, tenant={}",
this.getName(), dataId, group, tenant);
throw new NacosException(NacosException.CONFLICT,
"Data being modified, dataId=" + dataId + ",group=" + group + ",tenant=" + tenant);
} else {
LOGGER.error("[{}] [sub-server-error] dataId={}, group={}, tenant={}, code={}", this.getName(), dataId,
group, tenant, response);
throw new NacosException(response.getErrorCode(),
"Server error, code=" + response.getErrorCode() + ",msg=" + response.getMessage() + ",dataId=" + dataId + ",group=" + group
+ ",tenant=" + tenant);
}
}
The requestProxy method in Nacos 2.x handles the gRPC communication:
private Response requestProxy(RpcClient rpcClientInner, Request request, long timeoutMills) throws NacosException {
try {
request.putAllHeader(super.getSecurityHeaders(resourceBuild(request)));
request.putAllHeader(super.getCommonHeader());
} catch (Exception e) {
throw new NacosException(NacosException.CLIENT_INVALID_PARAM, e);
}
JsonObject asJsonObjectTemp = new Gson().toJsonTree(request).getAsJsonObject();
asJsonObjectTemp.remove("headers");
asJsonObjectTemp.remove("requestId");
boolean limit = Limiter.isLimit(request.getClass() + asJsonObjectTemp.toString());
if (limit) {
throw new NacosException(NacosException.CLIENT_OVER_THRESHOLD,
"Exceeded client-side limit threshold");
}
return rpcClientInner.request(request, timeoutMills);
}
public Response request(Request request, long timeoutMills) throws NacosException {
int retryTimes = 0;
Response response;
Exception exceptionThrow = null;
long start = System.currentTimeMillis();
while (retryTimes < RETRY_TIMES && System.currentTimeMillis() < timeoutMills + start) {
boolean waitReconnect = false;
try {
if (this.currentConnection == null || !isRunning()) {
waitReconnect = true;
throw new NacosException(NacosException.CLIENT_DISCONNECT,
"Client not connected, current status:" + rpcClientStatus.get());
}
response = this.currentConnection.request(request, timeoutMills);
if (response == null) {
throw new NacosException(SERVER_ERROR, "Unknown Exception.");
}
if (response instanceof ErrorResponse) {
if (response.getErrorCode() == NacosException.UN_REGISTER) {
synchronized (this) {
waitReconnect = true;
if (rpcClientStatus.compareAndSet(RpcClientStatus.RUNNING, RpcClientStatus.UNHEALTHY)) {
LoggerUtils.printIfErrorEnabled(LOGGER,
"Connection unregistered, switching server, connectionId = {}, request = {}",
currentConnection.getConnectionId(), request.getClass().getSimpleName());
switchServerAsync();
}
}
}
throw new NacosException(response.getErrorCode(), response.getMessage());
}
lastActiveTimeStamp = System.currentTimeMillis();
return response;
} catch (Exception e) {
if (waitReconnect) {
try {
Thread.sleep(Math.min(100, timeoutMills / 3));
} catch (Exception exception) {
// Do nothing.
}
}
LoggerUtils.printIfErrorEnabled(LOGGER, "Request failed, request = {}, retryTimes = {}, errorMessage = {}", request, retryTimes, e.getMessage());
exceptionThrow = e;
}
retryTimes++;
}
if (rpcClientStatus.compareAndSet(RpcClientStatus.RUNNING, RpcClientStatus.UNHEALTHY)) {
switchServerAsyncOnRequestFail();
}
if (exceptionThrow != null) {
throw (exceptionThrow instanceof NacosException) ? (NacosException) exceptionThrow
: new NacosException(SERVER_ERROR, exceptionThrow);
} else {
throw new NacosException(SERVER_ERROR, "Request failed, unknown error");
}
}
Server-Side Configuration Handling
The client communicates with the server via the /nacos/v1/cs/configs endpoint. The server-side handling is found in the ConfigController.getConfig method.
@GetMapping
@Secured(action = ActionTypes.READ, parser = ConfigResourceParser.class)
public void getConfig(HttpServletRequest request, HttpServletResponse response,
@RequestParam("dataId") String dataId, @RequestParam("group") String group,
@RequestParam(value = "tenant", required = false, defaultValue = StringUtils.EMPTY) String tenant,
@RequestParam(value = "tag", required = false) String tag)
throws IOException, ServletException, NacosException {
ParamUtils.checkTenant(tenant);
tenant = NamespaceUtil.processNamespaceParameter(tenant);
ParamUtils.checkParam(dataId, group, "datumId", "content");
ParamUtils.checkParam(tag);
final String clientIp = RequestUtil.getRemoteIp(request);
inner.doGetConfig(request, response, dataId, group, tenant, tag, clientIp);
}
The inner.doGetConfig method processes the request and retrieves the configuration data.
public String doGetConfig(HttpServletRequest request, HttpServletResponse response, String dataId, String group, String tenant, String tag, String clientIp) throws IOException, ServletException {
final String groupKey = GroupKey2.getKey(dataId, group, tenant);
String autoTag = request.getHeader("Vipserver-Tag");
String requestIpApp = RequestUtil.getAppName(request);
int lockResult = tryConfigReadLock(groupKey);
final String requestIp = RequestUtil.getRemoteIp(request);
boolean isBeta = false;
if (lockResult > 0) {
FileInputStream fis = null;
try {
String md5 = Constants.NULL;
long lastModified = 0L;
CacheItem cacheItem = ConfigCacheService.getContentCache(groupKey);
if (cacheItem.isBeta() && cacheItem.getIps4Beta().contains(clientIp)) {
isBeta = true;
}
final String configType = (null != cacheItem.getType()) ? cacheItem.getType() : FileTypeEnum.TEXT.getFileType();
response.setHeader("Config-Type", configType);
FileTypeEnum fileTypeEnum = FileTypeEnum.getFileTypeEnumByFileExtensionOrFileType(configType);
String contentTypeHeader = fileTypeEnum.getContentType();
response.setHeader(HttpHeaderConsts.CONTENT_TYPE, contentTypeHeader);
File file = null;
ConfigInfoBase configInfoBase = null;
PrintWriter out = null;
if (isBeta) {
md5 = cacheItem.getMd54Beta();
lastModified = cacheItem.getLastModifiedTs4Beta();
if (PropertyUtil.isDirectRead()) {
configInfoBase = persistService.findConfigInfo4Beta(dataId, group, tenant);
} else {
file = DiskUtil.targetBetaFile(dataId, group, tenant);
}
response.setHeader("isBeta", "true");
} else {
if (StringUtils.isBlank(tag)) {
if (isUseTag(cacheItem, autoTag)) {
if (cacheItem.tagMd5 != null) {
md5 = cacheItem.tagMd5.get(autoTag);
}
if (cacheItem.tagLastModifiedTs != null) {
lastModified = cacheItem.tagLastModifiedTs.get(autoTag);
}
if (PropertyUtil.isDirectRead()) {
configInfoBase = persistService.findConfigInfo4Tag(dataId, group, tenant, autoTag);
} else {
file = DiskUtil.targetTagFile(dataId, group, tenant, autoTag);
}
response.setHeader("Vipserver-Tag",
URLEncoder.encode(autoTag, StandardCharsets.UTF_8.displayName()));
} else {
md5 = cacheItem.getMd5();
lastModified = cacheItem.getLastModifiedTs();
if (PropertyUtil.isDirectRead()) {
configInfoBase = persistService.findConfigInfo(dataId, group, tenant);
} else {
file = DiskUtil.targetFile(dataId, group, tenant);
}
if (configInfoBase == null && fileNotExist(file)) {
ConfigTraceService.logPullEvent(dataId, group, tenant, requestIpApp, -1,
ConfigTraceService.PULL_EVENT_NOTFOUND, -1, requestIp);
response.setStatus(HttpServletResponse.SC_NOT_FOUND);
response.getWriter().println("config data not exist");
return HttpServletResponse.SC_NOT_FOUND + "";
}
}
} else {
if (cacheItem.tagMd5 != null) {
md5 = cacheItem.tagMd5.get(tag);
}
if (cacheItem.tagLastModifiedTs != null) {
Long lm = cacheItem.tagLastModifiedTs.get(tag);
if (lm != null) {
lastModified = lm;
}
}
if (PropertyUtil.isDirectRead()) {
configInfoBase = persistService.findConfigInfo4Tag(dataId, group, tenant, tag);
} else {
file = DiskUtil.targetTagFile(dataId, group, tenant, tag);
}
if (configInfoBase == null && fileNotExist(file)) {
ConfigTraceService.logPullEvent(dataId, group, tenant, requestIpApp, -1,
ConfigTraceService.PULL_EVENT_NOTFOUND, -1, requestIp);
response.setStatus(HttpServletResponse.SC_NOT_FOUND);
response.getWriter().println("config data not exist");
return HttpServletResponse.SC_NOT_FOUND + "";
}
}
}
response.setHeader(Constants.CONTENT_MD5, md5);
response.setHeader("Pragma", "no-cache");
response.setDateHeader("Expires", 0);
response.setHeader("Cache-Control", "no-cache,no-store");
if (PropertyUtil.isDirectRead()) {
response.setDateHeader("Last-Modified", lastModified);
} else {
fis = new FileInputStream(file);
response.setDateHeader("Last-Modified", file.lastModified());
}
if (PropertyUtil.isDirectRead()) {
out = response.getWriter();
out.print(configInfoBase.getContent());
out.flush();
out.close();
} else {
fis.getChannel()
.transferTo(0L, fis.getChannel().size(), Channels.newChannel(response.getOutputStream()));
}
LogUtil.PULL_CHECK_LOG.warn("{}|{}|{}|{}", groupKey, requestIp, md5, TimeUtils.getCurrentTimeStr());
final long delayed = System.currentTimeMillis() - lastModified;
ConfigTraceService.logPullEvent(dataId, group, tenant, requestIpApp, lastModified,
ConfigTraceService.PULL_EVENT_OK, delayed, requestIp);
} finally {
releaseConfigReadLock(groupKey);
IoUtils.closeQuietly(fis);
}
} else if (lockResult == 0) {
ConfigTraceService
.logPullEvent(dataId, group, tenant, requestIpApp, -1, ConfigTraceService.PULL_EVENT_NOTFOUND, -1,
requestIp);
response.setStatus(HttpServletResponse.SC_NOT_FOUND);
response.getWriter().println("config data not exist");
return HttpServletResponse.SC_NOT_FOUND + "";
} else {
PULL_LOG.info("[client-get] clientIp={}, {}, get data during dump", clientIp, groupKey);
response.setStatus(HttpServletResponse.SC_CONFLICT);
response.getWriter().println("requested file is being modified, please try later.");
return HttpServletResponse.SC_CONFLICT + "";
}
return HttpServletResponse.SC_OK + "";
}
Database and Disk Access
The server can retrieve configuration data from a database (e.g., Derby) or from disk, depending on the configuration mode.
@Override
public ConfigInfo findConfigInfo(final String dataId, final String group, final String tenant) {
final String tenantTmp = StringUtils.isBlank(tenant) ? StringUtils.EMPTY : tenant;
final String sql = "SELECT ID,data_id,group_id,tenant_id,app_name,content,md5,type FROM config_info "
+ " WHERE data_id=? AND group_id=? AND tenant_id=?";
final Object[] args = new Object[] {dataId, group, tenantTmp};
return databaseOperate.queryOne(sql, args, CONFIG_INFO_ROW_MAPPER);
}
Disk access is handled by DiskUtil.targetFile, which constructs the file path based on the data ID, group, and tenant.
public static File targetFile(String dataId, String group, String tenant) {
File file = null;
if (StringUtils.isBlank(tenant)) {
file = new File(EnvUtil.getNacosHome(), BASE_DIR);
} else {
file = new File(EnvUtil.getNacosHome(), TENANT_BASE_DIR);
file = new File(file, tenant);
}
file = new File(file, group);
file = new File(file, dataId);
return file;
}
Dynamic Configuration Updates
To ensure that the client's configuration stays in sync with the server, Nacos employs a long polling mechanism. The process is as follows:
- The client initiates a long polling request.
- The server compares the client's requested data with its own cached data.
- If the data differs, the server immediately returns the updated configuration.
- If the data is the same, the server schedules a check after a delay (e.g., 29.5 seconds).
- During this delay, the server listens for data change events. If a change occurs, it triggers a
DataChangeTaskto push the updated data to the client. - To prevent conflicts, the long polling task cancels any pending scheduled checks before processing a data change event.
This mechanism ensures that clients receive configuration updates in a timely manner without excessive polling.