##### Coverage Report - com.ctrip.framework.apollo.internals.RemoteConfigRepository

Classes in this File | Line Coverage | Branch Coverage | Complexity
RemoteConfigRepository |
90% | 181/200
75% | 36/48
| 3.4375;3.438
RemoteConfigRepository$1 |
100% | 6/6
N/A | N/A
| 3.4375;3.438
RemoteConfigRepository$2 |
100% | 3/3
N/A | N/A
| 3.4375;3.438
RemoteConfigRepository$3 |
100% | 3/3
N/A | N/A
| 3.4375;3.438

1 |  |
     package com.ctrip.framework.apollo.internals;
2 |  |
3 |  |
     import com.google.common.base.Joiner;
4 |  |
     import com.google.common.base.Strings;
5 |  |
     import com.google.common.collect.Lists;
6 |  |
     import com.google.common.collect.Maps;
7 |  |
     import com.google.common.escape.Escaper;
8 |  |
     import com.google.common.net.UrlEscapers;
9 |  |
     import com.google.common.util.concurrent.RateLimiter;
10 |  |
11 |  |
     import com.ctrip.framework.apollo.Apollo;
12 |  |
     import com.ctrip.framework.apollo.core.ConfigConsts;
13 |  |
     import com.ctrip.framework.apollo.core.dto.ApolloConfig;
14 |  |
     import com.ctrip.framework.apollo.core.dto.ApolloConfigNotification;
15 |  |
     import com.ctrip.framework.apollo.core.dto.ServiceDTO;
16 |  |
     import com.ctrip.framework.apollo.core.schedule.ExponentialSchedulePolicy;
17 |  |
     import com.ctrip.framework.apollo.core.schedule.SchedulePolicy;
18 |  |
     import com.ctrip.framework.apollo.core.utils.ApolloThreadFactory;
19 |  |
     import com.ctrip.framework.apollo.exceptions.ApolloConfigException;
20 |  |
     import com.ctrip.framework.apollo.util.ConfigUtil;
21 |  |
     import com.ctrip.framework.apollo.util.ExceptionUtil;
22 |  |
     import com.ctrip.framework.apollo.util.http.HttpRequest;
23 |  |
     import com.ctrip.framework.apollo.util.http.HttpResponse;
24 |  |
     import com.ctrip.framework.apollo.util.http.HttpUtil;
25 |  |
     import com.dianping.cat.Cat;
26 |  |
     import com.dianping.cat.message.Message;
27 |  |
     import com.dianping.cat.message.Transaction;
28 |  |
29 |  |
     import org.codehaus.plexus.PlexusContainer;
30 |  |
     import org.codehaus.plexus.component.repository.exception.ComponentLookupException;
31 |  |
     import org.slf4j.Logger;
32 |  |
     import org.slf4j.LoggerFactory;
33 |  |
     import org.unidal.lookup.ContainerLoader;
34 |  |
35 |  |
     import java.util.Collections;
36 |  |
     import java.util.List;
37 |  |
     import java.util.Map;
38 |  |
     import java.util.Properties;
39 |  |
     import java.util.Random;
40 |  |
     import java.util.concurrent.ExecutorService;
41 |  |
     import java.util.concurrent.Executors;
42 |  |
     import java.util.concurrent.ScheduledExecutorService;
43 |  |
     import java.util.concurrent.TimeUnit;
44 |  |
     import java.util.concurrent.atomic.AtomicBoolean;
45 |  |
     import java.util.concurrent.atomic.AtomicReference;
46 |  |
47 |  |
     /**
48 |  |
      * @author Jason Song(song_s@ctrip.com)
49 |  |
      */
50 | 59 |
     public class RemoteConfigRepository extends AbstractConfigRepository {
51 | 1 |
       private static final Logger logger = LoggerFactory.getLogger(RemoteConfigRepository.class);
52 | 1 |
       private static final Joiner STRING_JOINER = Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR);
53 | 1 |
       private static final Joiner.MapJoiner MAP_JOINER = Joiner.on("&").withKeyValueSeparator("=");
54 |  |
       private PlexusContainer m_container;
55 |  |
       private final ConfigServiceLocator m_serviceLocator;
56 |  |
       private final HttpUtil m_httpUtil;
57 |  |
       private final ConfigUtil m_configUtil;
58 |  |
       private volatile AtomicReference<ApolloConfig> m_configCache;
59 |  |
       private final String m_namespace;
60 |  |
       private final static ScheduledExecutorService m_executorService;
61 |  |
       private final ExecutorService m_longPollingService;
62 |  |
       private final AtomicBoolean m_longPollingStopped;
63 |  |
       private SchedulePolicy m_longPollFailSchedulePolicyInSecond;
64 |  |
       private AtomicReference<ServiceDTO> m_longPollServiceDto;
65 |  |
       private AtomicReference<ApolloConfigNotification> m_longPollResult;
66 |  |
       private RateLimiter m_longPollRateLimiter;
67 |  |
       private RateLimiter m_loadConfigRateLimiter;
68 | 1 |
       private static final Escaper pathEscaper = UrlEscapers.urlPathSegmentEscaper();
69 | 1 |
       private static final Escaper queryParamEscaper = UrlEscapers.urlFormParameterEscaper();
70 |  |
71 |  |
       static {
72 | 2 |
         m_executorService = Executors.newScheduledThreadPool(1,
73 | 1 |
             ApolloThreadFactory.create("RemoteConfigRepository", true));
74 | 1 |
       }
75 |  |
76 |  |
       /**
77 |  |
        * Constructor.
78 |  |
        *
79 |  |
        * @param namespace the namespace
80 |  |
        */
81 | 14 |
       public RemoteConfigRepository(String namespace) {
82 | 14 |
         m_namespace = namespace;
83 | 14 |
         m_configCache = new AtomicReference<>();
84 | 14 |
         m_container = ContainerLoader.getDefaultContainer();
85 |  |
         try {
86 | 14 |
           m_configUtil = m_container.lookup(ConfigUtil.class);
87 | 14 |
           m_httpUtil = m_container.lookup(HttpUtil.class);
88 | 14 |
           m_serviceLocator = m_container.lookup(ConfigServiceLocator.class);
89 | 0 |
         } catch (ComponentLookupException ex) {
90 | 0 |
           Cat.logError(ex);
91 | 0 |
           throw new ApolloConfigException("Unable to load component!", ex);
92 | 14 |
         }
93 | 14 |
         m_longPollFailSchedulePolicyInSecond = new ExponentialSchedulePolicy(1, 120); //in second
94 | 14 |
         m_longPollingStopped = new AtomicBoolean(false);
95 | 28 |
         m_longPollingService = Executors.newSingleThreadExecutor(
96 | 14 |
             ApolloThreadFactory.create("RemoteConfigRepository-LongPolling", true));
97 | 14 |
         m_longPollServiceDto = new AtomicReference<>();
98 | 14 |
         m_longPollResult = new AtomicReference<>();
99 | 14 |
         m_longPollRateLimiter = RateLimiter.create(m_configUtil.getLongPollQPS());
100 | 14 |
         m_loadConfigRateLimiter = RateLimiter.create(m_configUtil.getLoadConfigQPS());
101 | 14 |
         this.trySync();
102 | 14 |
         this.schedulePeriodicRefresh();
103 | 14 |
         this.scheduleLongPollingRefresh();
104 | 14 |
       }
105 |  |
106 |  |
       @Override
107 |  |
       public Properties getConfig() {
108 | 34 |
         if (m_configCache.get() == null) {
109 | 4 |
           this.sync();
110 |  |
         }
111 | 30 |
         return transformApolloConfigToProperties(m_configCache.get());
112 |  |
       }
113 |  |
114 |  |
       @Override
115 |  |
       public void setUpstreamRepository(ConfigRepository upstreamConfigRepository) {
116 |  |
         //remote config doesn't need upstream
117 | 0 |
       }
118 |  |
119 |  |
       private void schedulePeriodicRefresh() {
120 | 28 |
         logger.debug("Schedule periodic refresh with interval: {} {}",
121 | 14 |
             m_configUtil.getRefreshInterval(), m_configUtil.getRefreshIntervalTimeUnit());
122 | 28 |
         this.m_executorService.scheduleAtFixedRate(
123 | 14 |
             new Runnable() {
124 |  |
               @Override
125 |  |
               public void run() {
126 | 15 |
                 Cat.logEvent("Apollo.ConfigService", String.format("periodicRefresh: %s", m_namespace));
127 | 15 |
                 logger.debug("refresh config for namespace: {}", m_namespace);
128 | 15 |
                 trySync();
129 | 14 |
                 Cat.logEvent("Apollo.Client.Version", Apollo.VERSION);
130 | 14 |
               }
131 | 14 |
             }, m_configUtil.getRefreshInterval(), m_configUtil.getRefreshInterval(),
132 | 14 |
             m_configUtil.getRefreshIntervalTimeUnit());
133 | 14 |
       }
134 |  |
135 |  |
       @Override
136 |  |
       protected synchronized void sync() {
137 | 38 |
         Transaction transaction = Cat.newTransaction("Apollo.ConfigService", "syncRemoteConfig");
138 |  |
139 |  |
         try {
140 | 38 |
           ApolloConfig previous = m_configCache.get();
141 | 38 |
           ApolloConfig current = loadApolloConfig();
142 |  |
143 |  |
           //reference equals means HTTP 304
144 | 23 |
           if (previous != current) {
145 | 23 |
             logger.debug("Remote Config refreshed!");
146 | 23 |
             m_configCache.set(current);
147 | 23 |
             this.fireRepositoryChange(m_namespace, this.getConfig());
148 |  |
           }
149 |  |
150 | 23 |
           transaction.setStatus(Message.SUCCESS);
151 | 14 |
         } catch (Throwable ex) {
152 | 14 |
           transaction.setStatus(ex);
153 | 14 |
           throw ex;
154 |  |
         } finally {
155 | 37 |
           transaction.complete();
156 | 23 |
         }
157 | 23 |
       }
158 |  |
159 |  |
       private Properties transformApolloConfigToProperties(ApolloConfig apolloConfig) {
160 | 30 |
         Properties result = new Properties();
161 | 30 |
         result.putAll(apolloConfig.getConfigurations());
162 | 30 |
         return result;
163 |  |
       }
164 |  |
165 |  |
       private ApolloConfig loadApolloConfig() {
166 | 38 |
         if (!m_loadConfigRateLimiter.tryAcquire(5, TimeUnit.SECONDS)) {
167 |  |
           //wait at most 5 seconds
168 |  |
           try {
169 | 0 |
             TimeUnit.SECONDS.sleep(5);
170 | 0 |
           } catch (InterruptedException e) {
171 | 0 |
           }
172 |  |
         }
173 | 38 |
         String appId = m_configUtil.getAppId();
174 | 38 |
         String cluster = m_configUtil.getCluster();
175 | 38 |
         String dataCenter = m_configUtil.getDataCenter();
176 | 38 |
         Cat.logEvent("Apollo.Client.ConfigInfo", STRING_JOINER.join(appId, cluster, m_namespace));
177 | 38 |
         int maxRetries = 2;
178 | 38 |
         Throwable exception = null;
179 |  |
180 | 38 |
         List<ServiceDTO> configServices = getConfigServices();
181 | 67 |
         for (int i = 0; i < maxRetries; i++) {
182 | 53 |
           List<ServiceDTO> randomConfigServices = Lists.newLinkedList(configServices);
183 | 53 |
           Collections.shuffle(randomConfigServices);
184 |  |
           //Access the server which notifies the client first
185 | 53 |
           if (m_longPollServiceDto.get() != null) {
186 | 4 |
             randomConfigServices.add(0, m_longPollServiceDto.getAndSet(null));
187 |  |
           }
188 |  |
189 | 53 |
           for (ServiceDTO configService : randomConfigServices) {
190 | 53 |
             String url =
191 | 106 |
                 assembleQueryConfigUrl(configService.getHomepageUrl(), appId, cluster, m_namespace,
192 | 53 |
                     dataCenter, m_configCache.get());
193 |  |
194 | 53 |
             logger.debug("Loading config from {}", url);
195 | 53 |
             HttpRequest request = new HttpRequest(url);
196 |  |
197 | 53 |
             Transaction transaction = Cat.newTransaction("Apollo.ConfigService", "queryConfig");
198 | 53 |
             transaction.addData("Url", url);
199 |  |
             try {
200 |  |
201 | 53 |
               HttpResponse<ApolloConfig> response = m_httpUtil.doGet(request, ApolloConfig.class);
202 |  |
203 | 31 |
               transaction.addData("StatusCode", response.getStatusCode());
204 | 31 |
               transaction.setStatus(Message.SUCCESS);
205 |  |
206 | 31 |
               if (response.getStatusCode() == 304) {
207 | 0 |
                 logger.debug("Config server responds with 304 HTTP status code.");
208 | 0 |
                 return m_configCache.get();
209 |  |
               }
210 |  |
211 | 31 |
               ApolloConfig result = response.getBody();
212 |  |
213 | 54 |
               Cat.logEvent("Apollo.Client.ConfigLoaded." + result.getNamespaceName(),
214 | 23 |
                   result.getReleaseKey());
215 | 23 |
               logger.debug("Loaded config for {}: {}", m_namespace, result);
216 |  |
217 | 23 |
               return result;
218 | 30 |
             } catch (Throwable ex) {
219 | 30 |
               Cat.logError(ex);
220 | 30 |
               transaction.setStatus(ex);
221 | 30 |
               exception = ex;
222 |  |
             } finally {
223 | 53 |
               transaction.complete();
224 | 30 |
             }
225 |  |
226 | 30 |
           }
227 |  |
228 |  |
           try {
229 | 30 |
             TimeUnit.SECONDS.sleep(1);
230 | 0 |
           } catch (InterruptedException ex) {
231 |  |
             //ignore
232 | 29 |
           }
233 |  |
         }
234 | 14 |
         String message = String.format(
235 |  |
             "Load Apollo Config failed - appId: %s, cluster: %s, namespace: %s, services: %s",
236 |  |
             appId, cluster, m_namespace, configServices);
237 | 14 |
         throw new ApolloConfigException(message, exception);
238 |  |
       }
239 |  |
240 |  |
       String assembleQueryConfigUrl(String uri, String appId, String cluster, String namespace,
241 |  |
                                             String dataCenter, ApolloConfig previousConfig) {
242 |  |
243 | 54 |
         String path = "configs/%s/%s/%s";
244 | 54 |
         List<String> pathParams =
245 | 108 |
             Lists.newArrayList(pathEscaper.escape(appId), pathEscaper.escape(cluster),
246 | 54 |
                 pathEscaper.escape(namespace));
247 | 54 |
         Map<String, String> queryParams = Maps.newHashMap();
248 |  |
249 | 54 |
         if (previousConfig != null) {
250 | 24 |
           queryParams.put("releaseKey", queryParamEscaper.escape(previousConfig.getReleaseKey()));
251 |  |
         }
252 |  |
253 | 54 |
         if (!Strings.isNullOrEmpty(dataCenter)) {
254 | 38 |
           queryParams.put("dataCenter", queryParamEscaper.escape(dataCenter));
255 |  |
         }
256 |  |
257 | 54 |
         String localIp = m_configUtil.getLocalIp();
258 | 54 |
         if (!Strings.isNullOrEmpty(localIp)) {
259 | 54 |
           queryParams.put("ip", queryParamEscaper.escape(localIp));
260 |  |
         }
261 |  |
262 | 54 |
         String pathExpanded = String.format(path, pathParams.toArray());
263 |  |
264 | 54 |
         if (!queryParams.isEmpty()) {
265 | 54 |
           pathExpanded += "?" + MAP_JOINER.join(queryParams);
266 |  |
         }
267 | 54 |
         if (!uri.endsWith("/")) {
268 | 54 |
           uri += "/";
269 |  |
         }
270 | 54 |
         return uri + pathExpanded;
271 |  |
       }
272 |  |
273 |  |
       private void scheduleLongPollingRefresh() {
274 |  |
         try {
275 | 14 |
           final String appId = m_configUtil.getAppId();
276 | 14 |
           final String cluster = m_configUtil.getCluster();
277 | 14 |
           final String dataCenter = m_configUtil.getDataCenter();
278 | 14 |
           m_longPollingService.submit(new Runnable() {
279 |  |
             @Override
280 |  |
             public void run() {
281 | 14 |
               doLongPollingRefresh(appId, cluster, dataCenter);
282 | 4 |
             }
283 |  |
           });
284 | 0 |
         } catch (Throwable ex) {
285 | 0 |
           ApolloConfigException exception =
286 |  |
               new ApolloConfigException("Schedule long polling refresh failed", ex);
287 | 0 |
           Cat.logError(exception);
288 | 0 |
           logger.warn(ExceptionUtil.getDetailMessage(exception));
289 | 14 |
         }
290 | 14 |
       }
291 |  |
292 |  |
       private void doLongPollingRefresh(String appId, String cluster, String dataCenter) {
293 | 14 |
         final Random random = new Random();
294 | 14 |
         ServiceDTO lastServiceDto = null;
295 | 735 |
         while (!m_longPollingStopped.get() && !Thread.currentThread().isInterrupted()) {
296 | 731 |
           if (!m_longPollRateLimiter.tryAcquire(5, TimeUnit.SECONDS)) {
297 |  |
             //wait at most 5 seconds
298 |  |
             try {
299 | 0 |
               TimeUnit.SECONDS.sleep(5);
300 | 0 |
             } catch (InterruptedException e) {
301 | 0 |
             }
302 |  |
           }
303 | 731 |
           Transaction transaction = Cat.newTransaction("Apollo.ConfigService", "pollNotification");
304 |  |
           try {
305 | 731 |
             if (lastServiceDto == null) {
306 | 33 |
               List<ServiceDTO> configServices = getConfigServices();
307 | 33 |
               lastServiceDto = configServices.get(random.nextInt(configServices.size()));
308 |  |
             }
309 |  |
310 | 731 |
             String url =
311 | 1462 |
                 assembleLongPollRefreshUrl(lastServiceDto.getHomepageUrl(), appId, cluster,
312 | 731 |
                     m_namespace, dataCenter, m_longPollResult.get());
313 |  |
314 | 731 |
             logger.debug("Long polling from {}", url);
315 | 731 |
             HttpRequest request = new HttpRequest(url);
316 |  |
             //longer timeout for read - 1 minute
317 | 731 |
             request.setReadTimeout(60000);
318 |  |
319 | 731 |
             transaction.addData("Url", url);
320 |  |
321 | 731 |
             HttpResponse<ApolloConfigNotification> response =
322 | 731 |
                 m_httpUtil.doGet(request, ApolloConfigNotification.class);
323 |  |
324 | 700 |
             logger.debug("Long polling response: {}, url: {}", response.getStatusCode(), url);
325 | 700 |
             if (response.getStatusCode() == 200) {
326 | 4 |
               m_longPollServiceDto.set(lastServiceDto);
327 | 4 |
               if (response.getBody() != null) {
328 | 1 |
                 m_longPollResult.set(response.getBody());
329 | 1 |
                 transaction.addData("Result", response.getBody().toString());
330 |  |
               }
331 | 4 |
               m_executorService.submit(new Runnable() {
332 |  |
                 @Override
333 |  |
                 public void run() {
334 | 4 |
                   trySync();
335 | 4 |
                 }
336 |  |
               });
337 |  |
             }
338 |  |
339 | 700 |
             m_longPollFailSchedulePolicyInSecond.success();
340 | 700 |
             transaction.addData("StatusCode", response.getStatusCode());
341 | 700 |
             transaction.setStatus(Message.SUCCESS);
342 | 29 |
           } catch (Throwable ex) {
343 | 29 |
             lastServiceDto = null;
344 | 29 |
             Cat.logError(ex);
345 | 29 |
             transaction.setStatus(ex);
346 | 29 |
             long sleepTimeInSecond = m_longPollFailSchedulePolicyInSecond.fail();
347 | 58 |
             logger.warn(
348 |  |
                 "Long polling failed, will retry in {} seconds. appId: {}, cluster: {}, namespace: {}, reason: {}",
349 | 29 |
                 sleepTimeInSecond, appId, cluster, m_namespace, ExceptionUtil.getDetailMessage(ex));
350 |  |
             try {
351 | 29 |
               TimeUnit.SECONDS.sleep(sleepTimeInSecond);
352 | 0 |
             } catch (InterruptedException ie) {
353 |  |
               //ignore
354 | 21 |
             }
355 |  |
           } finally {
356 | 721 |
             transaction.complete();
357 | 721 |
           }
358 | 721 |
         }
359 | 4 |
       }
360 |  |
361 |  |
       String assembleLongPollRefreshUrl(String uri, String appId, String cluster,
362 |  |
                                                 String namespace, String dataCenter,
363 |  |
                                                 ApolloConfigNotification previousResult) {
364 | 732 |
         Map<String, String> queryParams = Maps.newHashMap();
365 | 732 |
         queryParams.put("appId", queryParamEscaper.escape(appId));
366 | 732 |
         queryParams.put("cluster", queryParamEscaper.escape(cluster));
367 |  |
368 | 732 |
         if (!Strings.isNullOrEmpty(namespace)) {
369 | 732 |
           queryParams.put("namespace", queryParamEscaper.escape(namespace));
370 |  |
         }
371 | 732 |
         if (!Strings.isNullOrEmpty(dataCenter)) {
372 | 30 |
           queryParams.put("dataCenter", queryParamEscaper.escape(dataCenter));
373 |  |
         }
374 | 732 |
         String localIp = m_configUtil.getLocalIp();
375 | 732 |
         if (!Strings.isNullOrEmpty(localIp)) {
376 | 732 |
           queryParams.put("ip", queryParamEscaper.escape(localIp));
377 |  |
         }
378 |  |
379 | 732 |
         if (previousResult != null) {
380 |  |
           //number doesn't need encode
381 | 4 |
           queryParams.put("notificationId", String.valueOf(previousResult.getNotificationId()));
382 |  |
         }
383 |  |
384 | 732 |
         String params = MAP_JOINER.join(queryParams);
385 | 732 |
         if (!uri.endsWith("/")) {
386 | 732 |
           uri += "/";
387 |  |
         }
388 |  |
389 | 732 |
         return uri + "notifications?" + params;
390 |  |
       }
391 |  |
392 |  |
       void stopLongPollingRefresh() {
393 | 4 |
         this.m_longPollingStopped.compareAndSet(false, true);
394 | 4 |
       }
395 |  |
396 |  |
       private List<ServiceDTO> getConfigServices() {
397 | 71 |
         List<ServiceDTO> services = m_serviceLocator.getConfigServices();
398 | 71 |
         if (services.size() == 0) {
399 | 0 |
           throw new ApolloConfigException("No available config service");
400 |  |
         }
401 |  |
402 | 71 |
         return services;
403 |  |
       }
404 |  |
     }