sdclient/discovery-service/src/main/java/org/onap/msb/sdclient/wrapper/ConsulClientApp.java

   1 /**
   2  * Copyright 2016-2017 ZTE, Inc. and others.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
   5  * in compliance with the License. You may obtain a copy of the License at
   6  *
   7  * http://www.apache.org/licenses/LICENSE-2.0
   8  *
   9  * Unless required by applicable law or agreed to in writing, software distributed under the License
  10  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  11  * or implied. See the License for the specific language governing permissions and limitations under
  12  * the License.
  13  */
  14 package org.onap.msb.sdclient.wrapper;
  15
  16 import java.net.MalformedURLException;
  17 import java.net.URL;
  18 import java.util.ArrayList;
  19 import java.util.HashSet;
  20 import java.util.List;
  21 import java.util.Map;
  22 import java.util.Set;
  23
  24 import org.onap.msb.sdclient.core.KeyVaulePair;
  25 import org.onap.msb.sdclient.core.MicroServiceFullInfo;
  26 import org.onap.msb.sdclient.core.NodeInfo;
  27 import org.onap.msb.sdclient.wrapper.consul.Consul;
  28 import org.onap.msb.sdclient.wrapper.consul.HealthClient;
  29 import org.onap.msb.sdclient.wrapper.consul.cache.HealthCache;
  30 import org.onap.msb.sdclient.wrapper.consul.model.health.Service;
  31 import org.onap.msb.sdclient.wrapper.consul.model.health.ServiceHealth;
  32 import org.onap.msb.sdclient.wrapper.util.JacksonJsonUtil;
  33 import org.slf4j.Logger;
  34 import org.slf4j.LoggerFactory;
  35
  36 public class ConsulClientApp {
  37
  38     private final Consul consul;
  39     private final HealthClient healthClient;
  40     // private AtomicReference<List<HealthCache>> cacheList = new
  41     // AtomicReference<List<HealthCache>>(
  42     // new ArrayList<HealthCache>());
  43
  44
  45     private static final Logger LOGGER = LoggerFactory.getLogger(ConsulClientApp.class);
  46
  47     public ConsulClientApp(String ip, int port) {
  48         URL url = null;
  49         try {
  50             url = new URL("http", ip, port, "");
  51         } catch (MalformedURLException e1) {
  52             // TODO Auto-generated catch block
  53             LOGGER.error("start  ConsulClientApp throw exception", e1);
  54             throw new RuntimeException(e1);
  55         }
  56         this.consul = Consul.builder().withUrl(url).build(); // connect to Consul on localhost
  57         this.healthClient = consul.healthClient();
  58     }
  59
  60     public Consul getConsul() {
  61         return consul;
  62     }
  63
  64
  65
  66     /**
  67      * @Title startHealthNodeListen
  68      * @Description TODO(开启某个服务的node变化监听,只返回健康状态服务)
  69      * @param serviceName
  70      * @return
  71      * @return HealthCache
  72      */
  73     public HealthCache startHealthNodeListen(final String serviceName) {
  74         final HealthCache healthCache = HealthCache.newCache(healthClient, serviceName, 30);
  75         healthCache.addListener(new HealthCache.Listener<String, ServiceHealth>() {
  76             @Override
  77             public void notify(Map<String, ServiceHealth> newValues) {
  78                 // do Something with updated server map
  79                 LOGGER.info(serviceName + "--new node notify--");
  80
  81
  82                 if (newValues.isEmpty()) {
  83                     LOGGER.warn(serviceName + "--nodeList is Empty--");
  84                     PublishAddressWrapper.publishApigateWayList.remove(serviceName);
  85
  86                     try {
  87                         healthCache.stop();
  88                         LOGGER.info(serviceName + " Node Listen stopped");
  89                     } catch (Exception e) {
  90                         LOGGER.error(serviceName + " Node Listen stop throw exception", e);
  91                     }
  92
  93                     return;
  94                 }
  95                 // 服务发现变化
  96                 List<MicroServiceFullInfo> nodeAddressList = new ArrayList<MicroServiceFullInfo>();
  97                 for (Map.Entry<String, ServiceHealth> entry : newValues.entrySet()) {
  98
  99                     MicroServiceFullInfo microServiceInfo = new MicroServiceFullInfo();
 100
 101                     ServiceHealth value = (ServiceHealth) entry.getValue();
 102                     Service service = value.getService();
 103
 104                     NodeInfo node = new NodeInfo();
 105                     node.setIp(service.getAddress());
 106                     node.setPort(String.valueOf(service.getPort()));
 107                     Set<NodeInfo> nodes = new HashSet<NodeInfo>();
 108                     nodes.add(node);
 109                     microServiceInfo.setNodes(nodes);
 110
 111
 112                     microServiceInfo.setServiceName(serviceName);
 113
 114                     try {
 115                         List<String> tagList = service.getTags();
 116
 117
 118                         for (String tag : tagList) {
 119
 120                             if (tag.startsWith("\"ns\"")) {
 121                                 String ms_ns_json = tag.split("\"ns\":")[1];
 122                                 Map<String, String> nsMap =
 123                                                 (Map<String, String>) JacksonJsonUtil.jsonToBean(ms_ns_json, Map.class);
 124
 125                                 if (nsMap.get("namespace") != null) {
 126                                     microServiceInfo.setNamespace(nsMap.get("namespace"));
 127                                 }
 128
 129                                 continue;
 130                             }
 131
 132                             if (tag.startsWith("\"labels\"")) {
 133                                 String ms_labels_json = "{" + tag.split("\"labels\":\\{")[1];
 134                                 Map<String, String> labelMap = (Map<String, String>) JacksonJsonUtil
 135                                                 .jsonToBean(ms_labels_json, Map.class);
 136
 137                                 List<String> nodeLabels = new ArrayList<String>();
 138                                 for (Map.Entry<String, String> labelEntry : labelMap.entrySet()) {
 139                                     if ("visualRange".equals(labelEntry.getKey())) {
 140                                         microServiceInfo.setVisualRange(labelEntry.getValue());
 141                                     } else if ("network_plane_type".equals(labelEntry.getKey())) {
 142                                         microServiceInfo.setNetwork_plane_type(labelEntry.getValue());
 143                                     } else {
 144                                         nodeLabels.add(labelEntry.getKey() + ":" + labelEntry.getValue());
 145                                     }
 146
 147                                 }
 148
 149                                 microServiceInfo.setLabels(nodeLabels);
 150                                 continue;
 151                             }
 152
 153                             if (tag.startsWith("\"metadata\"")) {
 154                                 String ms_metadata_json = "{" + tag.split("\"metadata\":\\{")[1];
 155                                 Map<String, String> metadataMap = (Map<String, String>) JacksonJsonUtil
 156                                                 .jsonToBean(ms_metadata_json, Map.class);
 157
 158                                 List<KeyVaulePair> ms_metadata = new ArrayList<KeyVaulePair>();
 159
 160
 161                                 for (Map.Entry<String, String> metadataEntry : metadataMap.entrySet()) {
 162                                     KeyVaulePair keyVaulePair = new KeyVaulePair();
 163                                     keyVaulePair.setKey(metadataEntry.getKey());
 164                                     keyVaulePair.setValue(metadataEntry.getValue());
 165                                     ms_metadata.add(keyVaulePair);
 166                                 }
 167                                 microServiceInfo.setMetadata(ms_metadata);
 168                                 continue;
 169                             }
 170
 171                         }
 172
 173
 174                     } catch (Exception e) {
 175                         LOGGER.error(serviceName + " read tag  throw exception", e);
 176                     }
 177
 178                     nodeAddressList.add(microServiceInfo);
 179                 }
 180
 181                 PublishAddressWrapper.publishApigateWayList.put(serviceName, nodeAddressList);
 182
 183             }
 184         });
 185         try {
 186             LOGGER.info(serviceName + " Node Listen start");
 187             // cacheList.get().add(healthCache);
 188             healthCache.start();
 189
 190         } catch (Exception e) {
 191             // TODO Auto-generated catch block
 192             LOGGER.error(serviceName + " Node Listen start throw exception", e);
 193         }
 194
 195         return healthCache;
 196     }
 197
 198
 199
 200     public static void main(String[] args) {
 201         ConsulClientApp consulTest = new ConsulClientApp("127.0.0.1", 8500);
 202         // 监听服务变化
 203         consulTest.startHealthNodeListen("apigateway");
 204
 205
 206     }
 207
 208
 209 }