apache-ignite

1
# Licensed to the Apache Software Foundation (ASF) under one or more
2
# contributor license agreements.  See the NOTICE file distributed with
3
# this work for additional information regarding copyright ownership.
4
# The ASF licenses this file to You under the Apache License, Version 2.0
5
# (the "License"); you may not use this file except in compliance with
6
# the License.  You may obtain a copy of the License at
7
#
8
#    http://www.apache.org/licenses/LICENSE-2.0
9
#
10
# Unless required by applicable law or agreed to in writing, software
11
# distributed under the License is distributed on an "AS IS" BASIS,
12
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13
# See the License for the specific language governing permissions and
14
# limitations under the License.
15

16
"""
17
This module contains Spec classes that describes config and command line to start Ignite services
18
"""
19

20
import base64
21
import importlib
22
import json
23
import os
24
import subprocess
25
from abc import ABCMeta, abstractmethod
26
import re
27
from copy import deepcopy
28
from itertools import chain
29

30
from ignitetest.services.utils import IgniteServiceType
31
from ignitetest.services.utils.config_template import IgniteClientConfigTemplate, IgniteServerConfigTemplate, \
32
    IgniteLoggerConfigTemplate, IgniteThinClientConfigTemplate
33
from ignitetest.services.utils.jvm_utils import create_jvm_settings, merge_jvm_settings
34
from ignitetest.services.utils.path import get_home_dir, IgnitePathAware
35
from ignitetest.services.utils.ssl.ssl_params import is_ssl_enabled
36
from ignitetest.services.utils.metrics.metrics import is_opencensus_metrics_enabled, configure_opencensus_metrics,\
37
    is_jmx_metrics_enabled, configure_jmx_metrics
38
from ignitetest.services.utils.jmx_remote.jmx_remote_params import get_jmx_remote_params
39
from ignitetest.utils.ignite_test import JFR_ENABLED
40
from ignitetest.utils.version import DEV_BRANCH
41

42
SHARED_PREPARED_FILE = ".ignite_prepared"
43

44

45
def resolve_spec(service, **kwargs):
46
    """
47
    Resolve Spec classes for IgniteService and IgniteApplicationService
48
    """
49

50
    def _resolve_spec(name, default):
51
        if name in service.context.globals:
52
            fqdn = service.context.globals[name]
53
            (module, clazz) = fqdn.rsplit('.', 1)
54
            module = importlib.import_module(module)
55
            return getattr(module, clazz)
56
        return default
57

58
    def is_impl(impl):
59
        classes = map(lambda s: s.__name__, service.__class__.mro())
60
        impl_filter = list(filter(lambda c: c == impl, classes))
61
        return len(impl_filter) > 0
62

63
    if is_impl("IgniteService"):
64
        return _resolve_spec("NodeSpec", IgniteNodeSpec)(service=service, **kwargs)
65

66
    if is_impl("IgniteApplicationService"):
67
        return _resolve_spec("AppSpec", IgniteApplicationSpec)(service=service, **kwargs)
68

69
    raise Exception("There is no specification for class %s" % type(service))
70

71

72
def envs_to_exports(envs):
73
    """
74
    :return: line with exports env variables: export A=B; export C=D;
75
    """
76
    exports = ["export %s=%s" % (key, envs[key]) for key in envs]
77
    return "; ".join(exports) + ";"
78

79

80
class IgniteSpec(metaclass=ABCMeta):
81
    """
82
    This class is a basic Spec
83
    """
84

85
    def __init__(self, service, jvm_opts=None, merge_with_default=True):
86
        """
87
        :param service: Service
88
        :param jvm_opts: If passed will be added with higher priority to or overwrite completely the default options
89
                         depending on the merge_with_default. Either string or list of strings is allowed.
90
        :param merge_with_default: If False jvm_opts will overide the default options completely. None of the
91
                         default options will be applied.
92
        """
93
        self.service = service
94
        self.jvm_opts = merge_jvm_settings(self.__get_default_jvm_opts() if merge_with_default else [],
95
                                           jvm_opts if jvm_opts else [])
96

97
    def __get_default_jvm_opts(self):
98
        """
99
        Return a set of default JVM options.
100
        """
101
        default_jvm_opts = create_jvm_settings(gc_dump_path=os.path.join(self.service.log_dir, "gc.log"),
102
                                               oom_path=os.path.join(self.service.log_dir, "out_of_mem.hprof"),
103
                                               vm_error_path=os.path.join(self.service.log_dir, "hs_err_pid%p.log"))
104

105
        default_jvm_opts = merge_jvm_settings(
106
            default_jvm_opts, ["-DIGNITE_SUCCESS_FILE=" + os.path.join(self.service.persistent_root, "success_file"),
107
                               "-Dlog4j.configDebug=true"])
108

109
        default_jvm_opts = merge_jvm_settings(
110
            default_jvm_opts, ["-Dlog4j.configurationFile=file:" + self.service.log_config_file,
111
                               "-DappId=ignite"])
112

113
        if self.service.context.globals.get(JFR_ENABLED, False):
114
            default_jvm_opts = merge_jvm_settings(default_jvm_opts,
115
                                                  ["-XX:+UnlockCommercialFeatures",
116
                                                   "-XX:+FlightRecorder",
117
                                                   "-XX:StartFlightRecording=dumponexit=true," +
118
                                                   f"filename={self.service.jfr_dir}/recording.jfr"])
119

120
        jmx_remote_params = get_jmx_remote_params(self.service.context.globals)
121
        if jmx_remote_params.enabled:
122
            default_jvm_opts = merge_jvm_settings(default_jvm_opts,
123
                                                  ["-Dcom.sun.management.jmxremote",
124
                                                   "-Dcom.sun.management.jmxremote.port=%d" % jmx_remote_params.port,
125
                                                   "-Dcom.sun.management.jmxremote.local.only=false",
126
                                                   "-Dcom.sun.management.jmxremote.authenticate=false",
127
                                                   "-Dcom.sun.management.jmxremote.ssl=false"])
128

129
        return default_jvm_opts
130

131
    def config_templates(self):
132
        """
133
        :return: config that service will use to start on a node
134
        """
135
        config_templates = [(IgnitePathAware.IGNITE_LOG_CONFIG_NAME, IgniteLoggerConfigTemplate())]
136

137
        if self.service.config.service_type == IgniteServiceType.NODE:
138
            config_templates.append((IgnitePathAware.IGNITE_CONFIG_NAME,
139
                                     IgniteClientConfigTemplate() if self.service.config.client_mode
140
                                     else IgniteServerConfigTemplate()))
141

142
        if self.service.config.service_type == IgniteServiceType.THIN_CLIENT:
143
            config_templates.append((IgnitePathAware.IGNITE_THIN_CLIENT_CONFIG_NAME, IgniteThinClientConfigTemplate()))
144

145
        return config_templates
146

147
    def extend_config(self, config):
148
        """
149
        Extend config with custom variables
150
        """
151
        if config.service_type == IgniteServiceType.NODE:
152
            if is_opencensus_metrics_enabled(self.service):
153
                config = configure_opencensus_metrics(config, self.service.context.globals, self)
154

155
            if is_jmx_metrics_enabled(self.service):
156
                config = configure_jmx_metrics(config)
157

158
            if (is_opencensus_metrics_enabled(self.service) or
159
                    is_jmx_metrics_enabled(self.service)):
160

161
                ignite_instance_name = self._test_id
162

163
                if config.ignite_instance_name:
164
                    ignite_instance_name = ignite_instance_name[:250 - 2 - len(config.ignite_instance_name)] +\
165
                                           "--" + config.ignite_instance_name
166

167
                config = config._replace(ignite_instance_name=ignite_instance_name)
168

169
        config = config.prepare_ssl(self.service.globals, self.service.shared_root)
170

171
        return config
172

173
    @property
174
    def _test_id(self):
175
        return re.sub("^[0-9A-Fa-f]+@ignitetest\\.tests\\.", "", self.service.context.test_name).replace("=", ".")[:250]
176

177
    def __home(self, product=None):
178
        """
179
        Get home directory for current spec.
180
        """
181
        product = product if product else self.service.product
182
        return get_home_dir(self.service.install_root, product)
183

184
    @staticmethod
185
    def __get_module_libs(project_dir, module_name, is_dev):
186
        """
187
        Get absolute paths to be added to classpath for the specified module.
188
        """
189
        if is_dev:
190
            module_libs = [
191
                os.path.join("modules", module_name, "target"),
192
                os.path.join("modules", module_name, "target", "libs")
193
            ]
194
        else:
195
            module_libs = [
196
                os.path.join("libs", "optional", "ignite-%s" % module_name)
197
            ]
198

199
        return [os.path.join(project_dir, module_path) for module_path in module_libs]
200

201
    def _module_libs(self, module_name):
202
        """
203
        Get list of paths to be added to classpath for the passed module for current spec.
204
        """
205
        if module_name == "ducktests":
206
            return self.__get_module_libs(self.__home(str(DEV_BRANCH)), module_name, is_dev=True)
207

208
        return self.__get_module_libs(self.__home(), module_name, self.service.config.version.is_dev)
209

210
    @abstractmethod
211
    def command(self, node):
212
        """
213
        :return: string that represents command to run service on a node
214
        """
215

216
    def modules(self):
217
        """
218
        :return: modules set.
219
        """
220
        if self.service.modules:
221
            modules = deepcopy(self.service.modules)
222
        else:
223
            modules = []
224

225
        modules.append("log4j2")
226
        modules.append("ducktests")
227

228
        if is_opencensus_metrics_enabled(self.service):
229
            modules.append("opencensus")
230

231
        return modules
232

233
    def libs(self):
234
        """
235
        :return: list of paths for all modules to be added to classpath
236
        """
237
        flat_libs_list = chain(*[self._module_libs(module) for module in self.modules()])
238

239
        return list(map(lambda lib: os.path.join(lib, "*"), flat_libs_list))
240

241
    def envs(self):
242
        """
243
        :return: environment set.
244
        """
245
        return {
246
            'EXCLUDE_TEST_CLASSES': 'true',
247
            'IGNITE_LOG_DIR': self.service.log_dir,
248
            'USER_LIBS': ":".join(self.libs())
249
        }
250

251
    def config_file_path(self):
252
        """
253
        :return: path to project configuration file
254
        """
255
        return self.service.config_file
256

257
    def is_prepare_shared_files(self, local_dir):
258
        """
259
        :return True if we have something to prepare.
260
        """
261
        if not is_ssl_enabled(self.service.context.globals) and \
262
                not (self.service.config.service_type == IgniteServiceType.NODE and self.service.config.ssl_params):
263
            self.service.logger.debug("Ssl disabled. Nothing to generate.")
264
            return False
265

266
        if os.path.isfile(os.path.join(local_dir, SHARED_PREPARED_FILE)):
267
            self.service.logger.debug("Local shared dir already prepared. Exiting. " + local_dir)
268
            return False
269

270
        return True
271

272
    def prepare_shared_files(self, local_dir):
273
        """
274
        Prepare files that should be copied on all nodes.
275
        """
276
        self.service.logger.debug("Local shared dir not exists. Creating. " + local_dir)
277
        try:
278
            os.mkdir(local_dir)
279
        except FileExistsError:
280
            self.service.logger.debug("Shared dir already exists, ignoring and continue." + local_dir)
281

282
        script_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "..", "certs")
283

284
        self._runcmd(f"cp {script_dir}/* {local_dir}")
285
        self._runcmd(f"chmod a+x {local_dir}/*.sh")
286
        self._runcmd(f"{local_dir}/mkcerts.sh")
287

288
    def _jvm_opts(self):
289
        """
290
        :return: line with extra JVM params for ignite.sh script: -J-Dparam=value -J-ea
291
        """
292
        opts = ["-J%s" % o for o in self.jvm_opts]
293
        return " ".join(opts)
294

295
    def _runcmd(self, cmd):
296
        self.service.logger.debug(cmd)
297
        proc = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
298
        stdout, _ = proc.communicate()
299

300
        if proc.returncode != 0:
301
            raise RuntimeError("Command '%s' returned non-zero exit status %d: %s" % (cmd, proc.returncode, stdout))
302

303

304
class IgniteNodeSpec(IgniteSpec):
305
    """
306
    Spec to run ignite node
307
    """
308

309
    def command(self, node):
310
        cmd = "%s %s %s %s 2>&1 | tee -a %s &" % \
311
              (envs_to_exports(self.envs()),
312
               self.service.script("ignite.sh"),
313
               self._jvm_opts(),
314
               self.config_file_path(),
315
               os.path.join(self.service.log_dir, "console.log"))
316

317
        return cmd
318

319

320
class IgniteApplicationSpec(IgniteSpec):
321
    """
322
    Spec to run ignite application
323
    """
324
    def __init__(self, service, jvm_opts=None, merge_with_default=True):
325
        super().__init__(
326
            service,
327
            merge_jvm_settings(self.__get_default_jvm_opts() if merge_with_default else [],
328
                               jvm_opts if jvm_opts else []),
329
            merge_with_default)
330

331
    def __get_default_jvm_opts(self):
332
        return [
333
            "-DIGNITE_NO_SHUTDOWN_HOOK=true",  # allows performing operations on app termination.
334
            "-Xmx1G",
335
            "-ea",
336
            "-DIGNITE_ALLOW_ATOMIC_OPS_IN_TX=false"
337
        ]
338

339
    def command(self, node):
340
        args = [
341
            str(self.service.config.service_type.name),
342
            self.service.java_class_name,
343
            self.config_file_path(),
344
            str(base64.b64encode(json.dumps(self.service.params).encode('utf-8')), 'utf-8')
345
        ]
346

347
        cmd = "%s %s %s %s 2>&1 | tee -a %s &" % \
348
              (envs_to_exports(self.envs()),
349
               self.service.script("ignite.sh"),
350
               self._jvm_opts(),
351
               ",".join(args),
352
               os.path.join(self.service.log_dir, "console.log"))
353

354
        return cmd
355

356
    def config_file_path(self):
357
        return self.service.config_file if self.service.config.service_type == IgniteServiceType.NODE \
358
            else self.service.thin_client_config_file
359

360
    def envs(self):
361
        return {**super().envs(), **{"MAIN_CLASS": self.service.main_java_class}}
362