gerrit.onap Code Review - ccsdk/features.git/blob

   1 /*******************************************************************************
   2  * ============LICENSE_START========================================================================
   3  * ONAP : ccsdk feature sdnr wt
   4  * =================================================================================================
   5  * Copyright (C) 2019 highstreet technologies GmbH Intellectual Property. All rights reserved.
   6  * =================================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
   8  * in compliance with the License. You may obtain a copy of the License at
   9  *
  10  * http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software distributed under the License
  13  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  14  * or implied. See the License for the specific language governing permissions and limitations under
  15  * the License.
  16  * ============LICENSE_END==========================================================================
  17  ******************************************************************************/
  18 package org.onap.ccsdk.features.sdnr.wt.devicemanager.base.database;
  19
  20 import static org.elasticsearch.node.NodeBuilder.nodeBuilder;
  21 import java.io.File;
  22 import java.io.IOException;
  23 import java.net.DatagramSocket;
  24 import java.net.ServerSocket;
  25 import java.nio.charset.Charset;
  26 import java.nio.charset.StandardCharsets;
  27 import java.nio.file.Files;
  28 import java.nio.file.Path;
  29 import java.util.List;
  30 import org.apache.lucene.util.Version;
  31 import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
  32 import org.elasticsearch.client.Client;
  33 import org.elasticsearch.common.settings.Settings;
  34 import org.elasticsearch.common.unit.TimeValue;
  35 import org.elasticsearch.node.Node;
  36 import org.onap.ccsdk.features.sdnr.wt.devicemanager.base.internalTypes.Resources;
  37 import org.onap.ccsdk.features.sdnr.wt.devicemanager.config.impl.AkkaConfig;
  38 import org.onap.ccsdk.features.sdnr.wt.devicemanager.config.impl.EsConfig;
  39 import org.onap.ccsdk.features.sdnr.wt.devicemanager.config.impl.GeoConfig;
  40 import org.onap.ccsdk.features.sdnr.wt.devicemanager.config.util.ClusterNodeInfo;
  41 import org.slf4j.Logger;
  42 import org.slf4j.LoggerFactory;
  43
  44 public class HtDatabaseNode implements AutoCloseable {
  45
  46     private static final Logger LOGGER = LoggerFactory.getLogger(HtDatabaseNode.class);
  47     private static final String DBCONFIGFILENAME = "etc/elasticsearch.yml";
  48     private static final String RESFOLDER_PLUGIN = "elasticsearch/plugins";
  49     private static final String RESFOLDER_PLUGINHEAD = RESFOLDER_PLUGIN + "/head";
  50     private static final String RESFOLDER_PLUGINDELETE = RESFOLDER_PLUGIN + "/delete-by-query";
  51     private static int MIN_PORT_NUMBER = 1024;
  52     private static int MAX_PORT_NUMBER = 65535;
  53     private static int ES_PORT = 9200;
  54     private static int DELAYSECONDS = 120;
  55     private static String PLUGINFOLDER = "etc/elasticsearch-plugins";
  56
  57     private static HtDatabaseNode oneNode = null;
  58     private static Object initializationLock = new Object();
  59     private static Integer initializedTarget = 0;
  60     private static Integer initializedReached = 0;
  61
  62     private final Node node;
  63
  64     private HtDatabaseNode() {
  65         LOGGER.debug("Start elasticsearch service");
  66
  67         LOGGER.debug("Lucine version: " + Version.LATEST);
  68
  69         node = nodeBuilder().settings(Settings.builder().put("path.home", "etc").put("path.conf", "etc")).node();
  70         LOGGER.info("Starting Database service. Wait {} s", DELAYSECONDS);
  71         // Wait for orange status for single node without redundancy
  72         ClusterHealthResponse nodeStatus = node.client().admin().cluster().prepareHealth().setWaitForYellowStatus()
  73                 .setTimeout(TimeValue.timeValueSeconds(DELAYSECONDS)).get();
  74
  75         LOGGER.debug("Elasticsearch service started with status {}", nodeStatus.toString());
  76     }
  77
  78     /**
  79      * Close node
  80      */
  81     @Override
  82     public void close() {
  83         node.close();
  84         oneNode = null; // Release the one instance that was started !
  85     }
  86
  87     /**
  88      * Provide indication if all Index initializations are done.
  89      *
  90      * @return true if all index initializations are ready, false if not
  91      */
  92     public Boolean getInitialized() {
  93         synchronized (initializationLock) {
  94             return initializedTarget != 0 && initializedReached == initializedTarget;
  95         }
  96     }
  97
  98     public void setInitializedReached() {
  99         synchronized (initializationLock) {
 100             HtDatabaseNode.initializedReached++;
 101         }
 102     }
 103
 104     public void setInitializedTarget() {
 105         synchronized (initializationLock) {
 106             HtDatabaseNode.initializedTarget++;
 107         }
 108     }
 109
 110     public Client getClient() {
 111         return node.client();
 112     }
 113
 114
 115     /*
 116      * --------------------------------------- Static functions below
 117      */
 118
 119
 120     // Visibility package for test purpose
 121     static void checkorcreateplugins(String pluginFolder) {
 122         File f = new File(pluginFolder);
 123         if (!f.exists()) {
 124             f.mkdir();
 125         }
 126         Resources.copyFolderInto(RESFOLDER_PLUGINHEAD, PLUGINFOLDER, RESFOLDER_PLUGIN);
 127         Resources.copyFolderInto(RESFOLDER_PLUGINDELETE, PLUGINFOLDER, RESFOLDER_PLUGIN);
 128     }
 129
 130     /**
 131      * Checks to see if a specific port is available.
 132      *
 133      * @param port the port to check for availability
 134      */
 135     private static boolean isPortAvailable(int port) {
 136         if (port < MIN_PORT_NUMBER || port > MAX_PORT_NUMBER) {
 137             throw new IllegalArgumentException("Invalid start port: " + port);
 138         }
 139
 140         ServerSocket ss = null;
 141         DatagramSocket ds = null;
 142         try {
 143             ss = new ServerSocket(port);
 144             ss.setReuseAddress(true);
 145             ds = new DatagramSocket(port);
 146             ds.setReuseAddress(true);
 147             return true;
 148         } catch (IOException e) {
 149         } finally {
 150             if (ds != null) {
 151                 ds.close();
 152             }
 153
 154             if (ss != null) {
 155                 try {
 156                     ss.close();
 157                 } catch (IOException e) {
 158                     /* should not be thrown */
 159                 }
 160             }
 161         }
 162
 163         return false;
 164     }
 165
 166     private static void checkorcreateConfigFile(EsConfig config, AkkaConfig akkaConfig, GeoConfig geoConfig) {
 167         File f = new File(DBCONFIGFILENAME);
 168         if (!f.exists()) {
 169             LOGGER.debug("no " + DBCONFIGFILENAME + " found - extracting from resources");
 170             if (Resources.extractFileTo("elasticsearch/elasticsearch.yml", f)) {
 171                 // replace template values
 172                 LOGGER.debug("replace template values with config:" + config);
 173                 Charset charset = StandardCharsets.UTF_8;
 174                 try {
 175                     Path p = f.toPath();
 176                     String hostName = "0.0.0.0"; // Default as initialisation value
 177                     if (akkaConfig != null && akkaConfig.isCluster()) {
 178                         LOGGER.debug("cluster configuration found");
 179                         hostName = akkaConfig.getClusterConfig().getHostName(hostName);
 180                         String clusterDBName = akkaConfig.getClusterConfig().getDBClusterName(null);
 181                         String nodeName = String.format("node%d.%s", akkaConfig.getClusterConfig().getRoleMemberIndex(),
 182                                 clusterDBName);
 183                         if (clusterDBName != null) {
 184                             config.setCluster(clusterDBName);
 185                             config.setNode(nodeName);
 186                             config.save();
 187                             LOGGER.info("set db name to " + clusterDBName + " nodename=" + nodeName);
 188                         } else {
 189                             LOGGER.warn("unable to set correct db clustername");
 190                         }
 191                     }
 192                     String content = new String(Files.readAllBytes(p), charset);
 193                     content = content.replaceAll("\\$clustername", config.getCluster())
 194                             .replaceAll("\\$nodename", config.getNode()).replaceAll("\\$hostname", hostName);
 195
 196                     // add cluster configuration
 197                     if (akkaConfig != null && akkaConfig.isCluster()) {
 198                         List<ClusterNodeInfo> seedNodes = akkaConfig.getClusterConfig().getSeedNodes();
 199                         String nodesJSONString = "[\"" + seedNodes.get(0).getRemoteAddress() + "\"";
 200                         for (int i = 1; i < seedNodes.size(); i++) {
 201                             nodesJSONString += ",\"" + seedNodes.get(i).getRemoteAddress() + "\"";
 202                         }
 203                         nodesJSONString += "]";
 204                         content += System.lineSeparator()
 205                                 + String.format("discovery.zen.ping.unicast.hosts: %s", nodesJSONString);
 206
 207                         if (geoConfig != null) {
 208                             LOGGER.debug("adding zone configuration");
 209                             content += System.lineSeparator() + String
 210                                     .format("cluster.routing.allocation.awareness.force.zone.values: zone1,zone2");
 211                             content += System.lineSeparator()
 212                                     + String.format("cluster.routing.allocation.awareness.attributes: zone");
 213                             if (geoConfig.isPrimary(akkaConfig.getClusterConfig().getRoleMember())) {
 214                                 content += System.lineSeparator() + String.format("node.zone: zone1");
 215                                 LOGGER.debug("setting zone to zone1");
 216                             } else {
 217                                 content += System.lineSeparator() + String.format("node.zone: zone2");
 218                                 LOGGER.debug("setting zone to zone2");
 219                             }
 220                         }
 221                     }
 222                     Files.write(p, content.getBytes(charset));
 223                 } catch (IOException e) {
 224                     LOGGER.warn("problem replacing values in file: " + e.getMessage());
 225
 226                 }
 227             } else {
 228                 LOGGER.warn("problem writing database.yml to etc folder from res");
 229             }
 230         }
 231     }
 232
 233     /**
 234      * Start as singleton
 235      * @param config data
 236      * @param akkaConfig data
 237      * @param geoConfig data
 238      * @return the node or null if external node used
 239      */
 240     public static HtDatabaseNode start(EsConfig config, AkkaConfig akkaConfig, GeoConfig geoConfig) {
 241         if (isPortAvailable(ES_PORT)) {
 242             LOGGER.info("ES Port not in use. Start internal ES.");
 243             if (oneNode == null) {
 244                 checkorcreateplugins(PLUGINFOLDER);
 245                 checkorcreateConfigFile(config, akkaConfig, geoConfig);
 246                 oneNode = new HtDatabaseNode();
 247             } else {
 248                 throw new IllegalStateException(
 249                         "Database is already started, but can only be started once. Stop here.");
 250             }
 251         } else {
 252             LOGGER.info("ES Port in use. External ES used.");
 253         }
 254
 255         return oneNode;
 256     }
 257
 258 }