Ci listener (#4912)

* start elasticsearch container in CI

* setup listener in CI

* adjust

Co-authored-by: Sophie <84560950+Sophie-Xie@users.noreply.github.com>

Author: cangfengzhs

.github/workflows/nightly.yml

@@ -220,6 +220,18 @@ jobs:
       OSS_DIR: nebula-graph/package/nightly
     container:
       image: vesoft/nebula-dev:${{ matrix.os }}
+    services:
+      elasticsearch:
+        image: elasticsearch:7.17.7
+        ports:
+          - 9200:9200
+        env:
+          discovery.type: single-node
+        options: >-
+          --health-cmd "curl elasticsearch:9200"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
     steps:
       - uses: webiny/action-post-run@2.0.1
         with:

.github/workflows/pull_request.yml

@@ -72,6 +72,18 @@ jobs:
       volumes:
         - /tmp/ccache/nebula/${{ matrix.os }}-${{ matrix.compiler }}:/tmp/ccache/nebula/${{ matrix.os }}-${{ matrix.compiler }}
       options: --cap-add=SYS_PTRACE
+    services:
+      elasticsearch:
+        image: elasticsearch:7.17.7
+        ports:
+          - 9200:9200
+        env:
+          discovery.type: single-node
+        options: >-
+          --health-cmd "curl elasticsearch:9200"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
     steps:
       - uses: webiny/action-post-run@2.0.1
         with:

conf/nebula-storaged-listener.conf.default

@@ -0,0 +1,58 @@
+########## nebula-storaged-listener ###########
+########## basics ##########
+# Whether to run as a daemon process
+--daemonize=true
+# The file to host the process id
+--pid_file=pids_listener/nebula-storaged.pid
+# Whether to use the configuration obtained from the configuration file
+--local_config=true
+
+########## logging ##########
+# The directory to host logging files
+--log_dir=logs_listener
+# Log level, 0, 1, 2, 3 for INFO, WARNING, ERROR, FATAL respectively
+--minloglevel=0
+# Verbose log level, 1, 2, 3, 4, the higher of the level, the more verbose of the logging
+--v=0
+# Maximum seconds to buffer the log messages
+--logbufsecs=0
+# Whether to redirect stdout and stderr to separate output files
+--redirect_stdout=true
+# Destination filename of stdout and stderr, which will also reside in log_dir.
+--stdout_log_file=storaged-stdout.log
+--stderr_log_file=storaged-stderr.log
+# Copy log messages at or above this level to stderr in addition to logfiles. The numbers of severity levels INFO, WARNING, ERROR, and FATAL are 0, 1, 2, and 3, respectively.
+--stderrthreshold=2
+# Wether logging files' name contain timestamp.
+--timestamp_in_logfile_name=true
+
+########## networking ##########
+# Meta server address
+--meta_server_addrs=127.0.0.1:9559
+# Local ip
+--local_ip=127.0.0.1
+# Storage daemon listening port
+--port=9789
+# HTTP service ip
+--ws_ip=127.0.0.1
+# HTTP service port
+--ws_http_port=19789
+# heartbeat with meta service
+--heartbeat_interval_secs=10
+
+########## storage ##########
+# Listener wal directory. only one path is allowed.
+--listener_path=data/listener
+# This parameter can be ignored for compatibility. let's fill A default value of "data"
+--data_path=data
+# The type of part manager, [memory | meta]
+--part_man_type=memory
+# The default reserved bytes for one batch operation
+--rocksdb_batch_size=4096
+# The default block cache size used in BlockBasedTable.
+# The unit is MB.
+--rocksdb_block_cache=4
+# The type of storage engine, `rocksdb', `memory', etc.
+--engine_type=rocksdb
+# The type of part, `simple', `consensus'...
+--part_type=simple

tests/README.md

@@ -175,15 +175,15 @@ e.g.
 ```gherkin
 Feature: Nebula service termination test
   Scenario: Basic termination test
-    Given a nebulacluster with 1 graphd and 1 metad and 1 storaged
+    Given a nebulacluster with 1 graphd and 1 metad and 1 storaged and 0 listener
     When the cluster was terminated
     Then no service should still running after 4s
 ```
 
 ```gherkin
 Feature: Example
   Scenario: test with disable authorize
-    Given a nebulacluster with 1 graphd and 1 metad and 1 storaged:
+    Given a nebulacluster with 1 graphd and 1 metad and 1 storaged and 0 listener:
       """
       graphd:enable_authorize=false
       """
@@ -201,7 +201,7 @@ Feature: Example
     Then the execution should be successful
 
   Scenario: test with enable authorize
-    Given a nebulacluster with 1 graphd and 1 metad and 1 storaged:
+    Given a nebulacluster with 1 graphd and 1 metad and 1 storaged and 0 listener:
       """
       graphd:enable_authorize=true
       """

tests/common/nebula_service.py

@@ -41,6 +41,12 @@ def __init__(self, name, ports, suffix_index=0, params=None, is_standalone=False
             self.tcp_port, self.tcp_internal_port, self.http_port, self.https_port = ports[0:4]
             self.meta_port, self.meta_tcp_internal_port, self.meta_http_port, self.meta_https_port = ports[4:8]
             self.storage_port, self.storage_tcp_internal_port, self.storage_http_port, self.storage_https_port = ports[8:12]
+        if name == "listener":
+            self.binary_name = "storaged"
+            self.conf_name = "storaged-listener"
+        else:
+            self.binary_name = name
+            self.conf_name = name
         self.suffix_index = suffix_index
         self.params = params
         self.host = '127.0.0.1'
@@ -56,14 +62,14 @@ def _format_nebula_command(self):
         if self.is_sa == False:
             process_params = {
                 'log_dir': 'logs{}'.format(self.suffix_index),
-                'pid_file': 'pids{}/nebula-{}.pid'.format(self.suffix_index, self.name),
+                'pid_file': 'pids{}/nebula-{}.pid'.format(self.suffix_index, self.binary_name),
                 'port': self.tcp_port,
                 'ws_http_port': self.http_port,
             }
         else:
             process_params = {
                 'log_dir': 'logs{}'.format(self.suffix_index),
-                'pid_file': 'pids{}/nebula-{}.pid'.format(self.suffix_index, self.name),
+                'pid_file': 'pids{}/nebula-{}.pid'.format(self.suffix_index, self.binary_name),
                 'port': self.tcp_port,
                 'ws_http_port': self.http_port,
                 'meta_port': self.meta_port,
@@ -72,16 +78,16 @@ def _format_nebula_command(self):
                 'ws_storage_http_port': self.storage_http_port,
             }
         # data path
-        if self.name.upper() != 'GRAPHD':
+        if self.binary_name.upper() != 'GRAPHD':
             process_params['data_path'] = 'data{}/{}'.format(
-                self.suffix_index, self.name
+                self.suffix_index, self.binary_name
             )
 
         process_params.update(self.params)
         cmd = [
-            'bin/nebula-{}'.format(self.name),
+            'bin/nebula-{}'.format(self.binary_name),
             '--flagfile',
-            'conf/nebula-{}.conf'.format(self.name),
+            'conf/nebula-{}.conf'.format(self.conf_name),
         ] + ['--{}={}'.format(key, value) for key, value in process_params.items()]
 
         return " ".join(cmd)
@@ -126,35 +132,39 @@ def __init__(
         metad_num=1,
         storaged_num=1,
         graphd_num=1,
+        listener_num=1,
         ca_signed=False,
         debug_log=True,
         use_standalone=False,
         query_concurrently=False,
         **kwargs,
     ):
-        assert graphd_num > 0 and metad_num > 0 and storaged_num > 0
+        assert graphd_num > 0 and metad_num > 0 and storaged_num > 0 and listener_num >= 0
         self.build_dir = str(build_dir)
         self.src_dir = str(src_dir)
         self.work_dir = os.path.join(
             self.build_dir,
             'server_' + time.strftime('%Y-%m-%dT%H-%M-%S', time.localtime()),
         )
         self.pids = {}
-        self.metad_num, self.storaged_num, self.graphd_num = (
+        self.metad_num, self.storaged_num, self.graphd_num, self.listener_num = (
             metad_num,
             storaged_num,
             graphd_num,
+            listener_num,
         )
-        self.metad_processes, self.storaged_processes, self.graphd_processes = (
+        self.metad_processes, self.storaged_processes, self.graphd_processes, self.listener_processes = (
+            [],
             [],
             [],
             [],
         )
         self.all_processes = []
         self.all_ports = []
-        self.metad_param, self.storaged_param, self.graphd_param = {}, {}, {}
+        self.metad_param, self.storaged_param, self.graphd_param, self.listener_param = {}, {}, {}, {}
         self.storaged_port = 0
         self.graphd_port = 0
+        self.listener_port = 0
         self.ca_signed = ca_signed
         self.is_graph_ssl = (
             kwargs.get("enable_graph_ssl", "false").upper() == "TRUE"
@@ -175,7 +185,6 @@ def __init__(
             self._make_sa_params(**kwargs)
             self.init_standalone()
 
-
     def init_standalone(self):
         process_count = self.metad_num + self.storaged_num + self.graphd_num
         ports_count = process_count * self.ports_per_process
@@ -184,7 +193,7 @@ def init_standalone(self):
         index = 0
         standalone = NebulaProcess(
             "standalone",
-            self.all_ports[index : index + ports_count ],
+            self.all_ports[index: index + ports_count],
             index,
             self.graphd_param,
             is_standalone=True
@@ -205,15 +214,15 @@ def init_standalone(self):
             p.update_meta_server_addrs(meta_server_addrs)
 
     def init_process(self):
-        process_count = self.metad_num + self.storaged_num + self.graphd_num
+        process_count = self.metad_num + self.storaged_num + self.graphd_num + self.listener_num
         ports_count = process_count * self.ports_per_process
         self.all_ports = self._find_free_port(ports_count)
         index = 0
 
         for suffix_index in range(self.metad_num):
             metad = NebulaProcess(
                 "metad",
-                self.all_ports[index : index + self.ports_per_process],
+                self.all_ports[index: index + self.ports_per_process],
                 suffix_index,
                 self.metad_param,
             )
@@ -223,7 +232,7 @@ def init_process(self):
         for suffix_index in range(self.storaged_num):
             storaged = NebulaProcess(
                 "storaged",
-                self.all_ports[index : index + self.ports_per_process],
+                self.all_ports[index: index + self.ports_per_process],
                 suffix_index,
                 self.storaged_param,
             )
@@ -235,7 +244,7 @@ def init_process(self):
         for suffix_index in range(self.graphd_num):
             graphd = NebulaProcess(
                 "graphd",
-                self.all_ports[index : index + self.ports_per_process],
+                self.all_ports[index: index + self.ports_per_process],
                 suffix_index,
                 self.graphd_param,
             )
@@ -244,8 +253,20 @@ def init_process(self):
             if suffix_index == 0:
                 self.graphd_port = self.all_ports[0]
 
+        for suffix_index in range(self.storaged_num, self.storaged_num+self.listener_num):
+            listener = NebulaProcess(
+                "listener",
+                self.all_ports[index: index + self.ports_per_process],
+                suffix_index,
+                self.listener_param
+            )
+            self.listener_processes.append(listener)
+            index += self.ports_per_process
+            if suffix_index == 0:
+                self.listener_port = self.all_ports[0]
+
         self.all_processes = (
-            self.metad_processes + self.storaged_processes + self.graphd_processes
+            self.metad_processes + self.storaged_processes + self.graphd_processes + self.listener_processes
         )
         # update meta address
         meta_server_addrs = ','.join(
@@ -301,11 +322,14 @@ def _make_params(self, **kwargs):
         self.storaged_param['raft_heartbeat_interval_secs'] = '30'
         self.storaged_param['skip_wait_in_rate_limiter'] = 'true'
 
+        # params for listener only
+        self.listener_param = copy.copy(self.storaged_param)
+
         # params for meta only
         self.metad_param = copy.copy(_params)
         self.metad_param["default_parts_num"] = 1
 
-        for p in [self.metad_param, self.storaged_param, self.graphd_param]:
+        for p in [self.metad_param, self.storaged_param, self.graphd_param, self.listener_param]:
             p.update(kwargs)
 
     def _make_sa_params(self, **kwargs):
@@ -358,6 +382,7 @@ def _copy_nebula_conf(self):
                 conf_path + '{}.conf.default'.format(item),
                 self.work_dir + '/conf/{}.conf'.format(item),
             )
+        shutil.copy(conf_path+'nebula-storaged-listener.conf.default', self.work_dir+'/conf/nebula-storaged-listener.conf')
 
         resources_dir = self.work_dir + '/share/resources/'
         os.makedirs(resources_dir)

tests/nebula-test-run.py

@@ -241,6 +241,7 @@ def stop_nebula(nb, configs=None):
             NEBULA_HOME,
             graphd_num=graphd_inst,
             storaged_num=1,
+            listener_num=1,
             debug_log=opt_is(configs.debug, "true"),
             ca_signed=opt_is(configs.ca_signed, "true"),
             enable_ssl=configs.enable_ssl,

tests/tck/cluster/Example.feature

@@ -5,7 +5,7 @@
 Feature: Example
 
   Scenario: test with disable authorize
-    Given a nebulacluster with 1 graphd and 1 metad and 1 storaged:
+    Given a nebulacluster with 1 graphd and 1 metad and 1 storaged and 0 listener:
       """
       graphd:enable_authorize=false
       """
@@ -23,7 +23,7 @@ Feature: Example
     Then an PermissionError should be raised at runtime: No permission to grant/revoke god user.
 
   Scenario: test with enable authorize
-    Given a nebulacluster with 1 graphd and 1 metad and 1 storaged:
+    Given a nebulacluster with 1 graphd and 1 metad and 1 storaged and 0 listener:
       """
       graphd:enable_authorize=true
       """
@@ -41,7 +41,7 @@ Feature: Example
     Then an PermissionError should be raised at runtime: No permission to grant/revoke god user.
 
   Scenario: test with auth type is cloud
-    Given a nebulacluster with 1 graphd and 1 metad and 1 storaged:
+    Given a nebulacluster with 1 graphd and 1 metad and 1 storaged and 0 listener:
       """
       graphd:auth_type=cloud
       """

tests/tck/cluster/terminate.feature

@@ -5,6 +5,6 @@ Feature: Nebula service termination test
 
   # All nebula services should exit as expected after termination
   Scenario: Basic termination test
-    Given a nebulacluster with 1 graphd and 1 metad and 1 storaged
+    Given a nebulacluster with 1 graphd and 1 metad and 1 storaged and 0 listener
     When the cluster was terminated
     Then no service should still running after 4s