Pull request update/250915

5484a58 OSN-1048. Fix 500 on invalid region in insider
9f95697 OSN-1036. JSONObject scalar codegen improvement
0c07c91 OSN-1066. Added --no-cache flag to build.sh script
5ab0937 OSN-1070. Diworker thread safety

Author: stanfra

build.sh

@@ -1,6 +1,6 @@
 #!/usr/bin/env bash
 
-# ./build.sh [component] [tag] [-r registry] [-u username] [-p password] [--use-nerdctl]
+# ./build.sh [component] [tag] [-r registry] [-u username] [-p password] [--no-cache] [--use-nerdctl]
 # leave registry empty if default registry [docker.io] used
 
 # Initialize default values
@@ -10,6 +10,8 @@ LOGIN=""
 PASSWORD=""
 COMPONENT=""
 INPUT_TAG=""
+FLAGS=""
+NO_CACHE=false
 USE_NERDCTL=false
 BUILD_TOOL="docker"
 
@@ -19,6 +21,7 @@ while [[ "$#" -gt 0 ]]; do
         -r) REGISTRY="$2"; shift ;;
         -u) LOGIN="$2"; shift ;;
         -p) PASSWORD="$2"; shift ;;
+        --no-cache) NO_CACHE=true ;;
         --use-nerdctl) USE_NERDCTL=true ;;
         *)
             # Check if COMPONENT is empty
@@ -38,6 +41,11 @@ if [[ "$USE_NERDCTL" == true ]]; then
     BUILD_TOOL="nerdctl"
 fi
 
+# Set --no-cache flag
+if [[ "$NO_CACHE" == true ]]; then
+    FLAGS="--no-cache"
+fi
+
 COMMIT_ID=$(git rev-parse --verify HEAD)
 
 use_registry() {
@@ -101,7 +109,7 @@ do
       echo "component $COMPONENT re-tagged $COMMIT_ID -> $BUILD_TAG"
     else
       echo "Building image for ${COMPONENT}, build tag: ${BUILD_TAG}"
-      $BUILD_TOOL build -t ${COMPONENT}:${BUILD_TAG} -f ${DOCKERFILE} .
+      $BUILD_TOOL build $FLAGS -t ${COMPONENT}:${BUILD_TAG} -f ${DOCKERFILE} .
     fi
 
     if use_registry; then

diworker/diworker/main.py

@@ -1,16 +1,15 @@
 #!/usr/bin/env python
 import os
 import time
+import logging
 
 import urllib3
 from concurrent.futures import ThreadPoolExecutor
 from threading import Lock, Thread
 from etcd import Lock as EtcdLock
 from kombu import Exchange, Queue, Connection as QConnection
 from kombu.pools import producers
-from kombu.log import get_logger
 from kombu.mixins import ConsumerMixin
-from kombu.utils.debug import setup_logging
 from pymongo import MongoClient
 from urllib3.exceptions import InsecureRequestWarning
 import clickhouse_connect
@@ -36,68 +35,70 @@
     queue_arguments=ARGUMENTS
 )
 
-LOG = get_logger(__name__)
+LOG = logging.getLogger(__name__)
 ENVIRONMENT_CLOUD_TYPE = 'environment'
 HEARTBEAT_INTERVAL = 300
 DEFAULT_MAX_WORKERS = 4
 
 
 class DIWorker(ConsumerMixin):
-    def __init__(self, connection, config_cl):
+    def __init__(self, connection, rabbitmq_conn_str, diworker_settings,
+                 config_params):
         self.connection = connection
-        self.config_cl = config_cl
-        self._rest_cl = None
-        self._mongo_cl = None
-        self._clickhouse_cl = None
+        self.rabbitmq_conn_str = rabbitmq_conn_str
+        self.diworker_settings = diworker_settings
+        self.config_cl_params = config_params
         self.active_report_import_ids = set()
         self.active_reports_lock = Lock()
         self.running = True
-        self.thread = Thread(target=self.heartbeat)
+        self.thread = Thread(
+            target=self.heartbeat, args=(self.config_cl_params,))
         self.thread.start()
         self.executor = ThreadPoolExecutor(
             max_workers=int(
-                self.config_cl.diworker_settings().get(
+                self.diworker_settings.get(
                     'max_report_imports_workers',
                     DEFAULT_MAX_WORKERS
                 )
             )
         )
 
-    def heartbeat(self):
+    def heartbeat(self, config_params):
+        config_cl = self.get_config_cl(config_params)
+        rest_cl = self.get_rest_cl(config_cl)
         while self.running:
             with self.active_reports_lock:
                 report_import_ids = list(self.active_report_import_ids)
             for report_import_id in report_import_ids:
                 try:
-                    self.rest_cl.report_import_update(report_import_id, {})
+                    rest_cl.report_import_update(report_import_id, {})
                 except Exception as e:
                     LOG.warning("Heartbeat update failed for %s: %s", report_import_id, e)
             time.sleep(HEARTBEAT_INTERVAL)
+        rest_cl.close()
 
-    @property
-    def rest_cl(self):
-        if self._rest_cl is None:
-            self._rest_cl = RestClient(
-                url=self.config_cl.restapi_url(), verify=False)
-            self._rest_cl.secret = self.config_cl.cluster_secret()
-        return self._rest_cl
+    @staticmethod
+    def get_config_cl(config_params):
+        return ConfigClient(**config_params)
 
-    @property
-    def mongo_cl(self):
-        if self._mongo_cl is None:
-            mongo_params = self.config_cl.mongo_params()
-            self._mongo_cl = MongoClient(mongo_params[0])
-        return self._mongo_cl
+    @staticmethod
+    def get_rest_cl(config_cl):
+        url = config_cl.restapi_url()
+        secret = config_cl.cluster_secret()
+        return RestClient(url=url, verify=False, secret=secret)
 
-    @property
-    def clickhouse_cl(self):
-        if not self._clickhouse_cl:
-            user, password, host, db_name, port, secure = (
-                self.config_cl.clickhouse_params())
-            self._clickhouse_cl = clickhouse_connect.get_client(
-                host=host, password=password, database=db_name, user=user,
-                port=port, secure=secure)
-        return self._clickhouse_cl
+    @staticmethod
+    def get_mongo_cl(config_cl):
+        mongo_params = config_cl.mongo_params()
+        return MongoClient(mongo_params[0])
+
+    @staticmethod
+    def get_clickhouse_cl(config_cl):
+        user, password, host, db_name, port, secure = (
+            config_cl.clickhouse_params())
+        return clickhouse_connect.get_client(
+            host=host, password=password, database=db_name, user=user,
+            port=port, secure=secure)
 
     def publish_activities_task(self, organization_id, object_id, object_type,
                                 action, routing_key, meta=None):
@@ -108,8 +109,7 @@ def publish_activities_task(self, organization_id, object_id, object_type,
             'action': action,
             'meta': meta
         }
-        queue_conn = QConnection('amqp://{user}:{pass}@{host}:{port}'.format(
-            **self.config_cl.read_branch('/rabbit')))
+        queue_conn = QConnection(self.rabbitmq_conn_str)
         task_exchange = Exchange(ACTIVITIES_EXCHANGE_NAME, type='topic')
         with producers[queue_conn].acquire(block=True) as producer:
             producer.publish(
@@ -127,60 +127,59 @@ def get_consumers(self, Consumer, channel):
             accept=['json'],
             callbacks=[self.process_task],
             prefetch_count=int(
-                self.config_cl.diworker_settings().get(
+                self.diworker_settings.get(
                     'max_report_imports_workers',
                     DEFAULT_MAX_WORKERS
                 )
             ),
         )]
 
-    def report_import(self, task):
+    def report_import(self, task, config_cl, rest_cl, mongo_cl, clickhouse_cl):
         report_import_id = task.get('report_import_id')
         if not report_import_id:
             raise Exception('invalid task received: {}'.format(task))
 
         with self.active_reports_lock:
             self.active_report_import_ids.add(report_import_id)
 
-        _, import_dict = self.rest_cl.report_import_get(report_import_id)
+        _, import_dict = rest_cl.report_import_get(report_import_id)
         cloud_acc_id = import_dict.get('cloud_account_id')
-        _, resp = self.rest_cl.report_import_list(cloud_acc_id, show_active=True)
+        _, resp = rest_cl.report_import_list(cloud_acc_id, show_active=True)
         imports = list(filter(
             lambda x: x['id'] != report_import_id, resp['report_imports']))
         if imports:
             reason = 'Import cancelled due another import: %s' % imports[0]['id']
-            self.rest_cl.report_import_update(
+            rest_cl.report_import_update(
                 report_import_id,
                 {'state': 'failed', 'state_reason': reason}
             )
             return
         is_recalculation = import_dict.get('is_recalculation', False)
         LOG.info('Starting processing for task: %s, purpose %s',
                  task, 'recalculation ' if is_recalculation else 'import')
-        self.rest_cl.report_import_update(report_import_id,
-                                          {'state': 'in_progress'})
+        rest_cl.report_import_update(report_import_id, {'state': 'in_progress'})
 
         importer_params = {
             'cloud_account_id': cloud_acc_id,
-            'rest_cl': self.rest_cl,
-            'config_cl': self.config_cl,
-            'mongo_raw': self.mongo_cl.restapi['raw_expenses'],
-            'mongo_resources': self.mongo_cl.restapi['resources'],
-            'clickhouse_cl': self.clickhouse_cl,
+            'rest_cl': rest_cl,
+            'config_cl': config_cl,
+            'mongo_raw': mongo_cl.restapi['raw_expenses'],
+            'mongo_resources': mongo_cl.restapi['resources'],
+            'clickhouse_cl': clickhouse_cl,
             'import_file': import_dict.get('import_file'),
             'recalculate': is_recalculation}
         importer = None
         ca = None
         previous_attempt_ts = 0
         try:
-            _, ca = self.rest_cl.cloud_account_get(
+            _, ca = rest_cl.cloud_account_get(
                 importer_params.get('cloud_account_id'))
             organization_id = ca.get('organization_id')
-            _, org = self.rest_cl.organization_get(organization_id)
+            _, org = rest_cl.organization_get(organization_id)
             if org.get('disabled'):
                 reason = ('Import cancelled due to disabled '
                           'organization: %s') % report_import_id
-                self.rest_cl.report_import_update(
+                rest_cl.report_import_update(
                     report_import_id,
                     {'state': 'failed', 'state_reason': reason}
                 )
@@ -192,11 +191,11 @@ def report_import(self, task):
             importer = get_importer_class(cc_type, export_scheme)(
                 **importer_params)
             importer.import_report()
-            self.rest_cl.report_import_update(
+            rest_cl.report_import_update(
                 report_import_id, {'state': 'completed'})
             if start_last_import_ts == 0 and cc_type != ENVIRONMENT_CLOUD_TYPE:
                 all_reports_finished = True
-                _, resp = self.rest_cl.cloud_account_list(organization_id)
+                _, resp = rest_cl.cloud_account_list(organization_id)
                 for acc in resp['cloud_accounts']:
                     if (acc['type'] != ENVIRONMENT_CLOUD_TYPE and
                             acc['last_import_at'] == 0):
@@ -212,7 +211,7 @@ def report_import(self, task):
                 # pylint: disable=E1101
                 LOG.error('Mongo exception details: %s', exc.details)
             reason = str(exc)
-            self.rest_cl.report_import_update(
+            rest_cl.report_import_update(
                 report_import_id,
                 {'state': 'failed', 'state_reason': reason}
             )
@@ -245,24 +244,36 @@ def process_task(self, body, message):
         self.executor.submit(self._process_task, body, message)
 
     def _process_task(self, body, message):
+        config_cl = self.get_config_cl(self.config_cl_params)
+        rest_cl = self.get_rest_cl(config_cl)
+        mongo_cl = self.get_mongo_cl(config_cl)
+        clickhouse_cl = self.get_clickhouse_cl(config_cl)
         try:
-            self.report_import(body)
+            self.report_import(body, config_cl=config_cl, rest_cl=rest_cl,
+                               mongo_cl=mongo_cl, clickhouse_cl=clickhouse_cl)
         except Exception as exc:
             LOG.exception('Data import failed: %s', str(exc))
         finally:
+            mongo_cl.close()
+            clickhouse_cl.close()
+            rest_cl.close()
             with self.active_reports_lock:
                 self.active_report_import_ids.discard(body.get('report_import_id'))
         message.ack()
 
 
 if __name__ == '__main__':
     urllib3.disable_warnings(InsecureRequestWarning)
-    setup_logging(loglevel='INFO', loggers=[''])
-
-    config_cl = ConfigClient(
-        host=os.environ.get('HX_ETCD_HOST'),
-        port=int(os.environ.get('HX_ETCD_PORT')),
+    logging.basicConfig(
+        level=logging.INFO,
+        format='[%(threadName)s] %(levelname)s: %(message)s'
     )
+
+    config_cl_params = {
+        'host': os.environ.get('HX_ETCD_HOST'),
+        'port': int(os.environ.get('HX_ETCD_PORT'))
+    }
+    config_cl = ConfigClient(**config_cl_params)
     config_cl.wait_configured()
     migrator = Migrator(config_cl, 'restapi', 'diworker/diworker/migrations')
     # Use lock to avoid migration problems with several diworkers
@@ -272,9 +283,10 @@ def _process_task(self, body, message):
     LOG.info("starting worker")
     conn_str = 'amqp://{user}:{pass}@{host}:{port}'.format(
         **config_cl.read_branch('/rabbit'))
+    dw_settings = config_cl.diworker_settings()
     with QConnection(conn_str) as conn:
         try:
-            worker = DIWorker(conn, config_cl)
+            worker = DIWorker(conn, conn_str, dw_settings, config_cl_params)
             worker.run()
         except KeyboardInterrupt:
             worker.running = False

insider/insider_api/controllers/flavor_price.py

@@ -341,6 +341,8 @@ def _load_flavor_prices(self, region, flavor, os_type='linux',
         }
         price_infos = list(self.prices_collection.find(query))
         if not price_infos:
+            if region not in self.cloud_adapter.get_regions_coordinates():
+                raise WrongArgumentsException(Err.OI0012, [region])
             prices = self.cloud_adapter.get_flavor_prices(
                 [flavor], region, os_type=os_type,
                 billing_method=billing_method, quantity=quantity,

insider/insider_api/tests/unittests/test_flavor_prices.py

@@ -65,6 +65,11 @@ def setUp(self):
             'GcpProvider.cloud_adapter.get_regions_coordinates',
             return_value={'us-central1': {'name': 'US West (Oregon)'}}
         ).start()
+        patch(
+            'insider.insider_api.controllers.flavor_price.'
+            'AlibabaProvider.cloud_adapter.get_regions_coordinates',
+            return_value={'Singapore': {'name': 'US West (Oregon)'}}
+        ).start()
         super().setUp()
         patch('insider.insider_api.controllers.flavor_price.'
               'BaseProvider.mongo_client',

ngui/codegen.ts

@@ -1,5 +1,4 @@
 import type { CodegenConfig } from "@graphql-codegen/cli";
-import { JSONObjectResolver } from "graphql-scalars";
 
 const auth = {
   "./server/graphql/__generated__/types/auth.ts": {
@@ -38,7 +37,7 @@ const restapi = {
       defaultScalarType: "unknown",
       contextType: "../../../server#ContextValue",
       scalars: {
-        JSONObject: JSONObjectResolver.extensions.codegenScalarType
+        JSONObject: "Record<string, unknown>"
       }
     }
   },
@@ -53,7 +52,10 @@ const restapi = {
       withHooks: true,
       withMutationFn: true,
       withRefetchFn: true,
-      reactApolloVersion: 3
+      reactApolloVersion: 3,
+      scalars: {
+        JSONObject: "Record<string, unknown>"
+      }
     }
   }
 };

ngui/package.json

@@ -40,7 +40,6 @@
     "eslint-plugin-react-hooks": "5.1.0",
     "eslint-plugin-unused-imports": "4.1.4",
     "globals": "15.13.0",
-    "graphql-scalars": "1.24.0",
     "prettier": "3.3.3",
     "typescript": "5.7.2",
     "typescript-eslint": "8.41.0"