Develop to beta

Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.8-buster
+FROM python:3.9-buster
 
 RUN apt-get update && apt-get install -y wget git libxslt-dev iptables kmod swig3.0
 RUN ln -s /usr/bin/swig3.0 /usr/bin/swig

core/node.py

@@ -25,7 +25,6 @@
 import logging
 import platform
 import hashlib
-import distro
 
 import requests
 
@@ -314,8 +313,6 @@ def get_node_hardware_info() -> dict:
     system_release = f'{platform.system()}-{platform.release()}'
     uname_version = platform.uname().version
     attached_storage_size = get_block_device_size()
-    os_name = distro.id()
-    os_version = distro.version()
     return {
         'cpu_total_cores': psutil.cpu_count(logical=True),
         'cpu_physical_cores': psutil.cpu_count(logical=False),
@@ -325,9 +322,7 @@ def get_node_hardware_info() -> dict:
         'mem_available': psutil.virtual_memory().available,
         'system_release': system_release,
         'uname_version': uname_version,
-        'attached_storage_size': attached_storage_size,
-        'os_name': os_name,
-        'os_version': os_version
+        'attached_storage_size': attached_storage_size
     }
 
 

core/schains/checks.py

@@ -17,28 +17,37 @@
 #   You should have received a copy of the GNU Affero General Public License
 #   along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
+import filecmp
 import os
 import time
 import logging
+from abc import ABC, abstractmethod
+from typing import Any, Dict
 
 from core.schains.config.directory import (
+    upstreams_for_rotation_id_version,
+    get_schain_check_filepath,
     get_schain_config,
     schain_config_dir,
-    schain_config_filepath,
-    get_schain_check_filepath
+    schain_config_filepath
 )
 from core.schains.config.helper import (
     get_base_port_from_config,
     get_node_ips_from_config,
     get_own_ip_from_config,
     get_local_schain_http_endpoint
 )
-from core.schains.config.main import schain_config_version_match
+from core.schains.config.main import (
+    get_upstream_config_filepath,
+    get_rotation_ids_from_config_file
+)
 from core.schains.dkg.utils import get_secret_key_share_filepath
 from core.schains.firewall.types import IRuleController
 from core.schains.process_manager_helper import is_monitor_process_alive
 from core.schains.rpc import (
-    check_endpoint_alive, check_endpoint_blocks, get_endpoint_alive_check_timeout
+    check_endpoint_alive,
+    check_endpoint_blocks,
+    get_endpoint_alive_check_timeout
 )
 from core.schains.runner import get_container_name
 from core.schains.skaled_exit_codes import SkaledExitCodes
@@ -74,27 +83,37 @@ def __init__(self, status: bool, data: dict = None):
         self.status = status
         self.data = data if data else {}
 
+    def __bool__(self) -> bool:
+        return self.status
+
+    def __str__(self) -> str:
+        return f'CheckRes<{self.status}>'
+
+
+class IChecks(ABC):
+    @abstractmethod
+    def get_all(self, log=True, save=False, checks_filter=None) -> Dict:
+        pass
+
+    def is_healthy(self) -> bool:
+        checks = self.get_all()
+        return False not in checks.values()
 
-class SChainChecks:
+
+class ConfigChecks(IChecks):
     def __init__(
         self,
         schain_name: str,
         node_id: int,
         schain_record: SChainRecord,
-        rule_controller: IRuleController,
-        rotation_id: int = 0,
-        *,
-        ima_linked: bool = True,
-        dutils: DockerUtils = None
+        rotation_id: int,
+        stream_version: str
     ):
         self.name = schain_name
         self.node_id = node_id
         self.schain_record = schain_record
         self.rotation_id = rotation_id
-        self.dutils = dutils or DockerUtils()
-        self.container_name = get_container_name(SCHAIN_CONTAINER, self.name)
-        self.ima_linked = ima_linked
-        self.rc = rule_controller
+        self.stream_version = stream_version
 
     @property
     def config_dir(self) -> CheckRes:
@@ -112,14 +131,108 @@ def dkg(self) -> CheckRes:
         return CheckRes(os.path.isfile(secret_key_share_filepath))
 
     @property
-    def config(self) -> CheckRes:
-        """Checks that sChain config file exists"""
-        config_filepath = schain_config_filepath(self.name)
-        if not os.path.isfile(config_filepath):
+    def upstream_config(self) -> CheckRes:
+        """Checks that config exists for rotation id and stream"""
+        upstreams = upstreams_for_rotation_id_version(
+            self.name,
+            self.rotation_id,
+            self.stream_version
+        )
+        logger.debug('Upstream configs for %s: %s', self.name, upstreams)
+        return len(upstreams) > 0
+
+    def get_all(self, log=True, save=False, checks_filter=None) -> Dict:
+        if not checks_filter:
+            checks_filter = API_ALLOWED_CHECKS
+        checks_dict = {}
+        for check in checks_filter:
+            if hasattr(self, check):
+                if check not in API_ALLOWED_CHECKS:
+                    logger.warning('Check %s is not allowed or does not exist', check)
+                else:
+                    checks_dict[check] = getattr(self, check).status
+        if log:
+            log_checks_dict(self.name, checks_dict)
+        if save:
+            save_checks_dict(self.name, checks_dict)
+        return checks_dict
+
+    def is_healthy(self) -> bool:
+        checks = self.get_all()
+        return False not in checks.values()
+
+
+class SkaledChecks(IChecks):
+    def __init__(
+        self,
+        schain_name: str,
+        schain_record: SChainRecord,
+        rule_controller: IRuleController,
+        *,
+        ima_linked: bool = True,
+        dutils: DockerUtils = None
+    ):
+        self.name = schain_name
+        self.schain_record = schain_record
+        self.dutils = dutils or DockerUtils()
+        self.container_name = get_container_name(SCHAIN_CONTAINER, self.name)
+        self.ima_linked = ima_linked
+        self.rc = rule_controller
+
+    def get_all(self, log=True, save=False, checks_filter=None) -> Dict:
+        if not checks_filter:
+            checks_filter = API_ALLOWED_CHECKS
+        checks_dict = {}
+        for check in checks_filter:
+            if check == 'ima_container' and (DISABLE_IMA or not self.ima_linked):
+                logger.info(f'Check {check} will be skipped - IMA is not linked')
+            elif check not in API_ALLOWED_CHECKS:
+                logger.warning(f'Check {check} is not allowed or does not exist')
+            else:
+                if hasattr(self, check):
+                    checks_dict[check] = getattr(self, check).status
+        if log:
+            log_checks_dict(self.name, checks_dict)
+        if save:
+            save_checks_dict(self.name, checks_dict)
+        return checks_dict
+
+    @property
+    def upstream_exists(self) -> CheckRes:
+        upstream_path = get_upstream_config_filepath(self.name)
+        return CheckRes(upstream_path is not None)
+
+    @property
+    def rotation_id_updated(self) -> int:
+        if not self.config:
             return CheckRes(False)
-        return CheckRes(
-            schain_config_version_match(self.name, self.schain_record)
+        upstream_path = get_upstream_config_filepath(self.name)
+        config_path = schain_config_filepath(self.name)
+        upstream_rotations = get_rotation_ids_from_config_file(upstream_path)
+        config_rotations = get_rotation_ids_from_config_file(config_path)
+        logger.debug(
+            'Comparing rotation_ids between upstream %s and %s',
+            upstream_path,
+            config_path
         )
+        return CheckRes(upstream_rotations == config_rotations)
+
+    @property
+    def config_updated(self) -> CheckRes:
+        if not self.config:
+            return CheckRes(False)
+        upstream_path = get_upstream_config_filepath(self.name)
+        config_path = schain_config_filepath(self.name)
+        logger.debug('Checking if %s updated according to %s', config_path, upstream_path)
+        if not upstream_path:
+            return CheckRes(True)
+        return CheckRes(filecmp.cmp(upstream_path, config_path))
+
+    @property
+    def config(self) -> CheckRes:
+        """ Checks that sChain config file exists """
+        config_path = schain_config_filepath(self.name)
+        return CheckRes(os.path.isfile(config_path))
 
     @property
     def volume(self) -> CheckRes:
@@ -129,7 +242,7 @@ def volume(self) -> CheckRes:
     @property
     def firewall_rules(self) -> CheckRes:
         """Checks that firewall rules are set correctly"""
-        if self.config.status:
+        if self.config:
             conf = get_schain_config(self.name)
             base_port = get_base_port_from_config(conf)
             node_ips = get_node_ips_from_config(conf)
@@ -167,7 +280,7 @@ def ima_container(self) -> CheckRes:
     def rpc(self) -> CheckRes:
         """Checks that local skaled RPC is accessible"""
         res = False
-        if self.config.status:
+        if self.config:
             http_endpoint = get_local_schain_http_endpoint(self.name)
             timeout = get_endpoint_alive_check_timeout(
                 self.schain_record.failed_rpc_count
@@ -178,7 +291,7 @@ def rpc(self) -> CheckRes:
     @property
     def blocks(self) -> CheckRes:
         """Checks that local skaled is mining blocks"""
-        if self.config.status:
+        if self.config:
             http_endpoint = get_local_schain_http_endpoint(self.name)
             return CheckRes(check_endpoint_blocks(http_endpoint))
         return CheckRes(False)
@@ -188,22 +301,61 @@ def process(self) -> CheckRes:
         """Checks that sChain monitor process is running"""
         return CheckRes(is_monitor_process_alive(self.schain_record.monitor_id))
 
+
+class SChainChecks(IChecks):
+    def __init__(
+        self,
+        schain_name: str,
+        node_id: int,
+        schain_record: SChainRecord,
+        rule_controller: IRuleController,
+        stream_version: str,
+        rotation_id: int = 0,
+        *,
+        ima_linked: bool = True,
+        dutils: DockerUtils = None
+    ):
+        self._subjects = [
+            ConfigChecks(
+                schain_name=schain_name,
+                node_id=node_id,
+                schain_record=schain_record,
+                rotation_id=rotation_id,
+                stream_version=stream_version
+            ),
+            SkaledChecks(
+                schain_name=schain_name,
+                schain_record=schain_record,
+                rule_controller=rule_controller,
+                ima_linked=ima_linked,
+                dutils=dutils
+            )
+        ]
+
+    def __getattr__(self, attr: str) -> Any:
+        for subj in self._subjects:
+            if attr in dir(subj):
+                return getattr(subj, attr)
+        raise AttributeError(f'No such attribute {attr}')
+
     def get_all(self, log=True, save=False, checks_filter=None):
         if not checks_filter:
             checks_filter = API_ALLOWED_CHECKS
-        checks_dict = {}
-        for check in checks_filter:
-            if check == 'ima_container' and (DISABLE_IMA or not self.ima_linked):
-                logger.info(f'Check {check} will be skipped - IMA is not linked')
-            elif check not in API_ALLOWED_CHECKS:
-                logger.warning(f'Check {check} is not allowed or does not exist')
-            else:
-                checks_dict[check] = getattr(self, check).status
+
+        plain_checks = {}
+        for subj in self._subjects:
+            subj_checks = subj.get_all(
+                log=False,
+                save=False,
+                checks_filter=checks_filter
+            )
+            plain_checks.update(subj_checks)
+
         if log:
-            log_checks_dict(self.name, checks_dict)
+            log_checks_dict(self.name, plain_checks)
         if save:
-            save_checks_dict(self.name, checks_dict)
-        return checks_dict
+            save_checks_dict(self.name, plain_checks)
+        return plain_checks
 
     def is_healthy(self):
         checks = self.get_all()

core/schains/cleaner.py

@@ -24,6 +24,7 @@
 
 from sgx import SgxClient
 
+from core.node import get_skale_node_version
 from core.schains.checks import SChainChecks
 from core.schains.config.directory import schain_config_dir
 from core.schains.dkg.utils import get_secret_key_share_filepath
@@ -58,7 +59,7 @@
 
 
 def run_cleaner(skale, node_config):
-    process = Process(target=monitor, args=(skale, node_config))
+    process = Process(name='cleaner', target=monitor, args=(skale, node_config))
     process.start()
     logger.info('Cleaner process started')
     process.join(JOIN_TIMEOUT)
@@ -202,22 +203,33 @@ def remove_schain(skale, node_id, schain_name, msg, dutils=None) -> None:
     terminate_schain_process(schain_record)
     delete_bls_keys(skale, schain_name)
     sync_agent_ranges = get_sync_agent_ranges(skale)
-    cleanup_schain(node_id, schain_name, sync_agent_ranges, dutils=dutils)
+    rotation_data = skale.node_rotation.get_rotation(schain_name)
+    rotation_id = rotation_data['rotation_id']
+    cleanup_schain(
+        node_id,
+        schain_name,
+        sync_agent_ranges,
+        rotation_id=rotation_id,
+        dutils=dutils
+    )
 
 
-def cleanup_schain(node_id, schain_name, sync_agent_ranges, dutils=None) -> None:
+def cleanup_schain(node_id, schain_name, sync_agent_ranges, rotation_id, dutils=None) -> None:
     dutils = dutils or DockerUtils()
     schain_record = upsert_schain_record(schain_name)
 
     rc = get_default_rule_controller(
         name=schain_name,
         sync_agent_ranges=sync_agent_ranges
     )
+    stream_version = get_skale_node_version()
     checks = SChainChecks(
         schain_name,
         node_id,
         rule_controller=rc,
-        schain_record=schain_record
+        stream_version=stream_version,
+        schain_record=schain_record,
+        rotation_id=rotation_id
     )
     if checks.skaled_container.status or is_exited(
         schain_name,