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
10 * http://www.apache.org/licenses/LICENSE-2.0
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
16 * ============LICENSE_END==========================================================================
17 ******************************************************************************/
18 package org.onap.ccsdk.features.sdnr.wt.devicemanager.base.database;
20 import static org.elasticsearch.node.NodeBuilder.nodeBuilder;
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 javax.annotation.Nullable;
31 import org.apache.lucene.util.Version;
32 import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
33 import org.elasticsearch.client.Client;
34 import org.elasticsearch.common.settings.Settings;
35 import org.elasticsearch.common.unit.TimeValue;
36 import org.elasticsearch.node.Node;
37 import org.onap.ccsdk.features.sdnr.wt.devicemanager.base.internalTypes.Resources;
38 import org.onap.ccsdk.features.sdnr.wt.devicemanager.config.impl.AkkaConfig;
39 import org.onap.ccsdk.features.sdnr.wt.devicemanager.config.impl.EsConfig;
40 import org.onap.ccsdk.features.sdnr.wt.devicemanager.config.impl.GeoConfig;
41 import org.onap.ccsdk.features.sdnr.wt.devicemanager.config.util.ClusterNodeInfo;
42 import org.slf4j.Logger;
43 import org.slf4j.LoggerFactory;
45 public class HtDatabaseNode implements AutoCloseable {
47 private static final Logger LOGGER = LoggerFactory.getLogger(HtDatabaseNode.class);
48 private static final String DBCONFIGFILENAME = "etc/elasticsearch.yml";
49 private static final String RESFOLDER_PLUGIN = "elasticsearch/plugins";
50 private static final String RESFOLDER_PLUGINHEAD = RESFOLDER_PLUGIN + "/head";
51 private static final String RESFOLDER_PLUGINDELETE = RESFOLDER_PLUGIN + "/delete-by-query";
52 private static int MIN_PORT_NUMBER = 1024;
53 private static int MAX_PORT_NUMBER = 65535;
54 private static int ES_PORT = 9200;
55 private static int DELAYSECONDS = 120;
56 private static String PLUGINFOLDER = "etc/elasticsearch-plugins";
58 private static HtDatabaseNode oneNode = null;
59 private static Object initializationLock = new Object();
60 private static Integer initializedTarget = 0;
61 private static Integer initializedReached = 0;
63 private final Node node;
65 private HtDatabaseNode() {
66 LOGGER.debug("Start elasticsearch service");
68 LOGGER.debug("Lucine version: " + Version.LATEST);
70 node = nodeBuilder().settings(Settings.builder().put("path.home", "etc").put("path.conf", "etc")).node();
71 LOGGER.info("Starting Database service. Wait {} s", DELAYSECONDS);
72 // Wait for orange status for single node without redundancy
73 ClusterHealthResponse nodeStatus = node.client().admin().cluster().prepareHealth().setWaitForYellowStatus()
74 .setTimeout(TimeValue.timeValueSeconds(DELAYSECONDS)).get();
76 LOGGER.debug("Elasticsearch service started with status {}", nodeStatus.toString());
85 oneNode = null; // Release the one instance that was started !
89 * Provide indication if all Index initializations are done.
91 * @return true if all index initializations are ready, false if not
93 public Boolean getInitialized() {
94 synchronized (initializationLock) {
95 return initializedTarget != 0 && initializedReached == initializedTarget;
99 public void setInitializedReached() {
100 synchronized (initializationLock) {
101 HtDatabaseNode.initializedReached++;
105 public void setInitializedTarget() {
106 synchronized (initializationLock) {
107 HtDatabaseNode.initializedTarget++;
111 public Client getClient() {
112 return node.client();
117 * --------------------------------------- Static functions below
121 // Visibility package for test purpose
122 static void checkorcreateplugins(String pluginFolder) {
123 File f = new File(pluginFolder);
127 Resources.copyFolderInto(RESFOLDER_PLUGINHEAD, PLUGINFOLDER, RESFOLDER_PLUGIN);
128 Resources.copyFolderInto(RESFOLDER_PLUGINDELETE, PLUGINFOLDER, RESFOLDER_PLUGIN);
132 * Checks to see if a specific port is available.
134 * @param port the port to check for availability
136 private static boolean isPortAvailable(int port) {
137 if (port < MIN_PORT_NUMBER || port > MAX_PORT_NUMBER) {
138 throw new IllegalArgumentException("Invalid start port: " + port);
141 ServerSocket ss = null;
142 DatagramSocket ds = null;
144 ss = new ServerSocket(port);
145 ss.setReuseAddress(true);
146 ds = new DatagramSocket(port);
147 ds.setReuseAddress(true);
149 } catch (IOException e) {
158 } catch (IOException e) {
159 /* should not be thrown */
167 private static void checkorcreateConfigFile(EsConfig config, AkkaConfig akkaConfig, GeoConfig geoConfig) {
168 File f = new File(DBCONFIGFILENAME);
170 LOGGER.debug("no " + DBCONFIGFILENAME + " found - extracting from resources");
171 if (Resources.extractFileTo("elasticsearch/elasticsearch.yml", f)) {
172 // replace template values
173 LOGGER.debug("replace template values with config:" + config);
174 Charset charset = StandardCharsets.UTF_8;
177 String hostName = "0.0.0.0"; // Default as initialisation value
178 if (akkaConfig != null && akkaConfig.isCluster()) {
179 LOGGER.debug("cluster configuration found");
180 hostName = akkaConfig.getClusterConfig().getHostName(hostName);
181 String clusterDBName = akkaConfig.getClusterConfig().getDBClusterName(null);
182 String nodeName = String.format("node%d.%s", akkaConfig.getClusterConfig().getRoleMemberIndex(),
184 if (clusterDBName != null) {
185 config.setCluster(clusterDBName);
186 config.setNode(nodeName);
188 LOGGER.info("set db name to " + clusterDBName + " nodename=" + nodeName);
190 LOGGER.warn("unable to set correct db clustername");
193 String content = new String(Files.readAllBytes(p), charset);
194 content = content.replaceAll("\\$clustername", config.getCluster())
195 .replaceAll("\\$nodename", config.getNode()).replaceAll("\\$hostname", hostName);
197 // add cluster configuration
198 if (akkaConfig != null && akkaConfig.isCluster()) {
199 List<ClusterNodeInfo> seedNodes = akkaConfig.getClusterConfig().getSeedNodes();
200 String nodesJSONString = "[\"" + seedNodes.get(0).getRemoteAddress() + "\"";
201 for (int i = 1; i < seedNodes.size(); i++) {
202 nodesJSONString += ",\"" + seedNodes.get(i).getRemoteAddress() + "\"";
204 nodesJSONString += "]";
205 content += System.lineSeparator()
206 + String.format("discovery.zen.ping.unicast.hosts: %s", nodesJSONString);
208 if (geoConfig != null) {
209 LOGGER.debug("adding zone configuration");
210 content += System.lineSeparator() + String
211 .format("cluster.routing.allocation.awareness.force.zone.values: zone1,zone2");
212 content += System.lineSeparator()
213 + String.format("cluster.routing.allocation.awareness.attributes: zone");
214 if (geoConfig.isPrimary(akkaConfig.getClusterConfig().getRoleMember())) {
215 content += System.lineSeparator() + String.format("node.zone: zone1");
216 LOGGER.debug("setting zone to zone1");
218 content += System.lineSeparator() + String.format("node.zone: zone2");
219 LOGGER.debug("setting zone to zone2");
223 Files.write(p, content.getBytes(charset));
224 } catch (IOException e) {
225 LOGGER.warn("problem replacing values in file: " + e.getMessage());
229 LOGGER.warn("problem writing database.yml to etc folder from res");
237 * @return the node or null if external node used
239 public static @Nullable HtDatabaseNode start(EsConfig config) throws IllegalStateException {
240 return start(config, null, null);
243 public static HtDatabaseNode start(EsConfig config, AkkaConfig akkaConfig, GeoConfig geoConfig) {
244 if (isPortAvailable(ES_PORT)) {
245 LOGGER.info("ES Port not in use. Start internal ES.");
246 if (oneNode == null) {
247 checkorcreateplugins(PLUGINFOLDER);
248 checkorcreateConfigFile(config, akkaConfig, geoConfig);
249 oneNode = new HtDatabaseNode();
251 throw new IllegalStateException(
252 "Database is already started, but can only be started once. Stop here.");
255 LOGGER.info("ES Port in use. External ES used.");