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(newValues -> {
  76             // do Something with updated server map
  77             LOGGER.info(serviceName + "--new node notify--");
  78
  79
  80             if (newValues.isEmpty()) {
  81                 LOGGER.warn(serviceName + "--nodeList is Empty--");
  82                 PublishAddressWrapper.publishApigateWayList.remove(serviceName);
  83
  84                 try {
  85                     healthCache.stop();
  86                     LOGGER.info(serviceName + " Node Listen stopped");
  87                 } catch (Exception e) {
  88                     LOGGER.error(serviceName + " Node Listen stop throw exception", e);
  89                 }
  90
  91                 return;
  92             }
  93             // 服务发现变化
  94             List<MicroServiceFullInfo> nodeAddressList = new ArrayList<MicroServiceFullInfo>();
  95             for (Map.Entry<String, ServiceHealth> entry : newValues.entrySet()) {
  96
  97                 MicroServiceFullInfo microServiceInfo = new MicroServiceFullInfo();
  98
  99                 ServiceHealth value = (ServiceHealth) entry.getValue();
 100                 Service service = value.getService();
 101
 102                 NodeInfo node = new NodeInfo();
 103                 node.setIp(service.getAddress());
 104                 node.setPort(String.valueOf(service.getPort()));
 105                 Set<NodeInfo> nodes = new HashSet<NodeInfo>();
 106                 nodes.add(node);
 107                 microServiceInfo.setNodes(nodes);
 108
 109
 110                 microServiceInfo.setServiceName(serviceName);
 111
 112                 try {
 113                     List<String> tagList = service.getTags();
 114
 115
 116                     for (String tag : tagList) {
 117
 118                         if (tag.startsWith("\"ns\"")) {
 119                             String ms_ns_json = tag.split("\"ns\":")[1];
 120                             Map<String, String> nsMap =
 121                                             (Map<String, String>) JacksonJsonUtil.jsonToBean(ms_ns_json, Map.class);
 122
 123                             if (nsMap.get("namespace") != null) {
 124                                 microServiceInfo.setNamespace(nsMap.get("namespace"));
 125                             }
 126
 127                             continue;
 128                         }
 129
 130                         if (tag.startsWith("\"labels\"")) {
 131                             String ms_labels_json = "{" + tag.split("\"labels\":\\{")[1];
 132                             Map<String, String> labelMap = (Map<String, String>) JacksonJsonUtil
 133                                             .jsonToBean(ms_labels_json, Map.class);
 134
 135                             List<String> nodeLabels = new ArrayList<String>();
 136                             for (Map.Entry<String, String> labelEntry : labelMap.entrySet()) {
 137                                 if ("visualRange".equals(labelEntry.getKey())) {
 138                                     microServiceInfo.setVisualRange(labelEntry.getValue());
 139                                 } else if ("network_plane_type".equals(labelEntry.getKey())) {
 140                                     microServiceInfo.setNetwork_plane_type(labelEntry.getValue());
 141                                 } else {
 142                                     nodeLabels.add(labelEntry.getKey() + ":" + labelEntry.getValue());
 143                                 }
 144
 145                             }
 146
 147                             microServiceInfo.setLabels(nodeLabels);
 148                             continue;
 149                         }
 150
 151                         if (tag.startsWith("\"metadata\"")) {
 152                             String ms_metadata_json = "{" + tag.split("\"metadata\":\\{")[1];
 153                             Map<String, String> metadataMap = (Map<String, String>) JacksonJsonUtil
 154                                             .jsonToBean(ms_metadata_json, Map.class);
 155
 156                             List<KeyVaulePair> ms_metadata = new ArrayList<KeyVaulePair>();
 157
 158
 159                             for (Map.Entry<String, String> metadataEntry : metadataMap.entrySet()) {
 160                                 KeyVaulePair keyVaulePair = new KeyVaulePair();
 161                                 keyVaulePair.setKey(metadataEntry.getKey());
 162                                 keyVaulePair.setValue(metadataEntry.getValue());
 163                                 ms_metadata.add(keyVaulePair);
 164                             }
 165                             microServiceInfo.setMetadata(ms_metadata);
 166                             continue;
 167                         }
 168
 169                     }
 170
 171
 172                 } catch (Exception e) {
 173                     LOGGER.error(serviceName + " read tag  throw exception", e);
 174                 }
 175
 176                 nodeAddressList.add(microServiceInfo);
 177             }
 178
 179             PublishAddressWrapper.publishApigateWayList.put(serviceName, nodeAddressList);
 180
 181         });
 182         try {
 183             LOGGER.info(serviceName + " Node Listen start");
 184             // cacheList.get().add(healthCache);
 185             healthCache.start();
 186
 187         } catch (Exception e) {
 188             // TODO Auto-generated catch block
 189             LOGGER.error(serviceName + " Node Listen start throw exception", e);
 190         }
 191
 192         return healthCache;
 193     }
 194
 195
 196
 197     public static void main(String[] args) {
 198         ConsulClientApp consulTest = new ConsulClientApp("127.0.0.1", 8500);
 199         // 监听服务变化
 200         consulTest.startHealthNodeListen("apigateway");
 201
 202
 203     }
 204
 205
 206 }