47d7570e87b7cd22f2e4556172d67790a87287c5
[ccsdk/features.git] /
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 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;
44
45 public class HtDatabaseNode implements AutoCloseable {
46
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";
57
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;
62
63     private final Node node;
64
65     private HtDatabaseNode() {
66         LOGGER.debug("Start elasticsearch service");
67
68         LOGGER.debug("Lucine version: " + Version.LATEST);
69
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();
75
76         LOGGER.debug("Elasticsearch service started with status {}", nodeStatus.toString());
77     }
78
79     /**
80      * Close node
81      */
82     @Override
83     public void close() {
84         node.close();
85         oneNode = null; // Release the one instance that was started !
86     }
87
88     /**
89      * Provide indication if all Index initializations are done.
90      *
91      * @return true if all index initializations are ready, false if not
92      */
93     public Boolean getInitialized() {
94         synchronized (initializationLock) {
95             return initializedTarget != 0 && initializedReached == initializedTarget;
96         }
97     }
98
99     public void setInitializedReached() {
100         synchronized (initializationLock) {
101             HtDatabaseNode.initializedReached++;
102         }
103     }
104
105     public void setInitializedTarget() {
106         synchronized (initializationLock) {
107             HtDatabaseNode.initializedTarget++;
108         }
109     }
110
111     public Client getClient() {
112         return node.client();
113     }
114
115
116     /*
117      * --------------------------------------- Static functions below
118      */
119
120
121     // Visibility package for test purpose
122     static void checkorcreateplugins(String pluginFolder) {
123         File f = new File(pluginFolder);
124         if (!f.exists()) {
125             f.mkdir();
126         }
127         Resources.copyFolderInto(RESFOLDER_PLUGINHEAD, PLUGINFOLDER, RESFOLDER_PLUGIN);
128         Resources.copyFolderInto(RESFOLDER_PLUGINDELETE, PLUGINFOLDER, RESFOLDER_PLUGIN);
129     }
130
131     /**
132      * Checks to see if a specific port is available.
133      *
134      * @param port the port to check for availability
135      */
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);
139         }
140
141         ServerSocket ss = null;
142         DatagramSocket ds = null;
143         try {
144             ss = new ServerSocket(port);
145             ss.setReuseAddress(true);
146             ds = new DatagramSocket(port);
147             ds.setReuseAddress(true);
148             return true;
149         } catch (IOException e) {
150         } finally {
151             if (ds != null) {
152                 ds.close();
153             }
154
155             if (ss != null) {
156                 try {
157                     ss.close();
158                 } catch (IOException e) {
159                     /* should not be thrown */
160                 }
161             }
162         }
163
164         return false;
165     }
166
167     private static void checkorcreateConfigFile(EsConfig config, AkkaConfig akkaConfig, GeoConfig geoConfig) {
168         File f = new File(DBCONFIGFILENAME);
169         if (!f.exists()) {
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;
175                 try {
176                     Path p = f.toPath();
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(),
183                                 clusterDBName);
184                         if (clusterDBName != null) {
185                             config.setCluster(clusterDBName);
186                             config.setNode(nodeName);
187                             config.save();
188                             LOGGER.info("set db name to " + clusterDBName + " nodename=" + nodeName);
189                         } else {
190                             LOGGER.warn("unable to set correct db clustername");
191                         }
192                     }
193                     String content = new String(Files.readAllBytes(p), charset);
194                     content = content.replaceAll("\\$clustername", config.getCluster())
195                             .replaceAll("\\$nodename", config.getNode()).replaceAll("\\$hostname", hostName);
196
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() + "\"";
203                         }
204                         nodesJSONString += "]";
205                         content += System.lineSeparator()
206                                 + String.format("discovery.zen.ping.unicast.hosts: %s", nodesJSONString);
207
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");
217                             } else {
218                                 content += System.lineSeparator() + String.format("node.zone: zone2");
219                                 LOGGER.debug("setting zone to zone2");
220                             }
221                         }
222                     }
223                     Files.write(p, content.getBytes(charset));
224                 } catch (IOException e) {
225                     LOGGER.warn("problem replacing values in file: " + e.getMessage());
226
227                 }
228             } else {
229                 LOGGER.warn("problem writing database.yml to etc folder from res");
230             }
231         }
232     }
233
234     /**
235      * Start as singleton
236      *
237      * @return the node or null if external node used
238      */
239     public static @Nullable HtDatabaseNode start(EsConfig config) throws IllegalStateException {
240         return start(config, null, null);
241     }
242
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();
250             } else {
251                 throw new IllegalStateException(
252                         "Database is already started, but can only be started once. Stop here.");
253             }
254         } else {
255             LOGGER.info("ES Port in use. External ES used.");
256         }
257
258         return oneNode;
259     }
260
261 }