Pull request update/250303

499deb8 OSN-684. Updated README
9765df5 OSN-665. Fixed OOMKilled on getting Azure flavors
9e1610f OSN-660. Handle HttpError for Alibaba resources discovery
d128dd0 OSN-587. Limit/offset for clean_expenses and list_organizations APIs

Author: stanfra

README.md

@@ -144,7 +144,7 @@ source ~/.profile
 ```
 to add local ~/bin path to the system $PATH variable
 
-You can build local images running
+**Note:** you can build local images running
 
 ```
 cd .. && ./build.sh --use-nerdctl
@@ -159,13 +159,19 @@ Pay attention to "service_credentials" parameter, as OptScale uses it to retriev
 
 #### Cluster installation
 
-run the following command:
+Run the following command to start cluster from the required version:
 
 ```
 ./runkube.py --with-elk  -o overlay/user_template.yml -- <deployment name> <version>
 ```
 
-or if you want to use socket:
+or use `--no-pull` to start cluster from local images:
+
+```
+./runkube.py --with-elk --no-pull -o overlay/user_template.yml -- <deployment name> local
+```
+
+If you want to use socket:
 
 ```
 ./runkube.py --use-socket --with-elk  -o overlay/user_template.yml -- <deployment name> <version>
@@ -189,7 +195,7 @@ If you have insecure registry (with self-signed certificate) you can use --insec
 **please note**: if you use key authentication, you should have the required key (id_rsa) on the machine
 
 Check the state of the pods using `kubectl get pods` command.
-When all of the pods are running your OptScale is ready to use. Try to access it by `https://<ip address>`.
+When all the pods are running your OptScale is ready to use. Try to access it by `https://<ip address>`.
 
 #### Cluster update
 

docker_images/resource_discovery/Dockerfile

@@ -4,6 +4,7 @@ WORKDIR /usr/src/app/
 ENV PYTHONPATH /usr/src/app/
 
 COPY optscale_client/config_client optscale_client/config_client
+COPY optscale_client/insider_client optscale_client/insider_client
 COPY optscale_client/rest_api_client optscale_client/rest_api_client
 COPY tools/cloud_adapter tools/cloud_adapter
 COPY tools/optscale_time tools/optscale_time

docker_images/resource_discovery/requirements.txt

@@ -1,7 +1,7 @@
 kombu==5.3.4
-
 # OptScale packages
 -e optscale_client/config_client
+-e optscale_client/insider_client
 -e optscale_client/rest_api_client
 -e tools/cloud_adapter
 -e tools/optscale_time

docker_images/resource_discovery/worker.py

@@ -17,10 +17,12 @@
 from tools.cloud_adapter.exceptions import InvalidResourceTypeException
 from tools.cloud_adapter.model import ResourceTypes, RES_MODEL_MAP
 from optscale_client.config_client.client import Client as ConfigClient
+from optscale_client.insider_client.client import Client as InsiderClient
 from optscale_client.rest_api_client.client_v2 import Client as RestClient
 from tools.optscale_time import utcnow, utcnow_timestamp
 
 
+BYTES_IN_MB = 1024 * 1024
 CHUNK_SIZE = 200
 EXCHANGE_NAME = 'resource-discovery'
 QUEUE_NAME = 'discovery'
@@ -32,15 +34,19 @@
 
 
 class ResourcesSaver:
-    def __init__(self, rest_cl, limit, timeout, pause_timeout):
+    def __init__(self, rest_cl, insider_cl, limit, timeout, pause_timeout):
         queue_len = int(limit / CHUNK_SIZE) if CHUNK_SIZE else 0
         self.queue = queue.Queue(queue_len)
+        self.insider_cl = insider_cl
         self.rest_cl = rest_cl
         self.timeout = timeout
         self.pause_timeout = pause_timeout
         self.recording_available = Event()
         self.empty = Event()
         self._proc = None
+        self._cloud_account_id = None
+        self._cloud_type = None
+        self._resource_type = None
         self.start()
 
     def __del__(self):
@@ -95,9 +101,9 @@ def is_finished(self):
     def _save_chunks(self):
         while True:
             try:
-                chunk, resource_type, cloud_acc_id = self.queue.get(timeout=1)
+                chunk = self.queue.get(timeout=1)
                 self.empty.clear()
-                self.save_bulk_resources(chunk, resource_type, cloud_acc_id)
+                self.save_bulk_resources(chunk)
             except queue.Empty:
                 self.empty.set()
             except Exception as exc:
@@ -110,26 +116,72 @@ def get_resource_type_model(resource_type):
         except KeyError:
             raise Exception(f'Invalid resource type {resource_type}')
 
-    def build_payload(self, resource, resource_type):
+    def build_payload(self, resource):
         obj = {}
         for field in resource.fields(meta_fields_incl=False):
             val = getattr(resource, field)
             if val is not None and (isinstance(val, bool) or val):
                 obj[field] = val
         obj.pop('resource_id', None)
         obj.pop('organization_id', None)
-        obj['resource_type'] = getattr(ResourceTypes, resource_type).value
+        obj['resource_type'] = getattr(ResourceTypes, self.resource_type).value
         obj['last_seen'] = utcnow_timestamp()
         obj['active'] = True
         return obj
 
-    def save_bulk_resources(self, resources, resource_type, cloud_acc_id):
+    @property
+    def resource_type(self):
+        return self._resource_type
+
+    @resource_type.setter
+    def resource_type(self, value):
+        self._resource_type = value
+
+    @property
+    def cloud_account_id(self):
+        return self._cloud_account_id
+
+    @cloud_account_id.setter
+    def cloud_account_id(self, value):
+        self._cloud_account_id = value
+
+    @property
+    def cloud_type(self):
+        return self._cloud_type
+
+    @cloud_type.setter
+    def cloud_type(self, value):
+        self._cloud_type = value
+
+    def process_resource_obj(self, resources):
+        if self.resource_type != 'instance' or self.cloud_type != 'azure_cnr':
+            return resources
+        flavors = {}
+        for resource in resources:
+            flavor = flavors.get(resource.flavor)
+            if not flavor:
+                try:
+                    _, flavor = self.insider_cl.find_flavor(
+                        self.cloud_type, self.resource_type, resource.region,
+                        {'source_flavor_id': resource.flavor}, 'current',
+                        cloud_account_id=self.cloud_account_id)
+                except Exception as exc:
+                    LOG.exception(exc)
+                    continue
+            if flavor:
+                flavors[resource.flavor] = flavor
+                resource.cpu_count = flavor['cpu']
+                resource.ram = flavor['ram'] * BYTES_IN_MB
+        return resources
+
+    def save_bulk_resources(self, resources):
         payload = []
+        resources = self.process_resource_obj(resources)
         for rss in resources:
-            payload.append(self.build_payload(rss, resource_type))
+            payload.append(self.build_payload(rss))
         if payload:
             _, response = self.rest_cl.cloud_resource_create_bulk(
-                cloud_acc_id, {'resources': payload},
+                self.cloud_account_id, {'resources': payload},
                 behavior='update_existing', return_resources=True)
         for resource in resources:
             try:
@@ -143,6 +195,7 @@ def __init__(self, connection, config_cl):
         self.connection = connection
         self.config_cl = config_cl
         self.set_discover_settings()
+        self._insider_cl = None
         self._rest_cl = None
         self._res_saving = None
         self.running = True
@@ -153,6 +206,14 @@ def __del__(self):
         if self._res_saving:
             self.res_saving.shutdown()
 
+    @property
+    def insider_cl(self):
+        if not self._insider_cl:
+            self._insider_cl = InsiderClient(
+                url=self.config_cl.insider_url(),
+                secret=self.config_cl.cluster_secret())
+        return self._insider_cl
+
     @property
     def rest_cl(self):
         if self._rest_cl is None:
@@ -166,6 +227,7 @@ def rest_cl(self):
     def res_saving(self):
         if self._res_saving is None:
             self._res_saving = ResourcesSaver(
+                insider_cl=self.insider_cl,
                 rest_cl=self.rest_cl,
                 limit=self.discover_size,
                 timeout=self.timeout,
@@ -256,7 +318,10 @@ def _discover_resources(self, cloud_acc_id, resource_type):
             return
         config = self.get_config(cloud_acc_id)
         gen_list = self.discover(config, resource_type)
-        discovered_resources = []
+        self.res_saving.cloud_account_id = cloud_acc_id
+        self.res_saving.cloud_type = config['type']
+        self.res_saving.resource_type = resource_type
+        discovered_resources = set()
         resources_count = 0
         max_parallel_requests = self.max_parallel_requests(config)
         errors = set()
@@ -279,18 +344,16 @@ def _discover_resources(self, cloud_acc_id, resource_type):
                         gen_list_chunk.remove(gen)
                         errors.add(str(res))
                     elif res:
-                        discovered_resources.append(res)
+                        discovered_resources.add(res)
                     else:
                         gen_list_chunk.remove(gen)
                     if len(discovered_resources) >= CHUNK_SIZE:
                         resources_count += len(discovered_resources)
-                        self.res_saving.send((discovered_resources.copy(),
-                                              resource_type, cloud_acc_id))
+                        self.res_saving.send(discovered_resources)
                         discovered_resources.clear()
         if len(discovered_resources):
             resources_count += len(discovered_resources)
-            self.res_saving.send((
-                discovered_resources.copy(), resource_type, cloud_acc_id))
+            self.res_saving.send(discovered_resources)
         LOG.info("%s %s resources have been discovered for cloud %s",
                  resources_count, resource_type, cloud_acc_id)
         self.res_saving.pause()

rest_api/rest_api_server/controllers/expense.py

@@ -934,12 +934,50 @@ def _process_clustered_expenses(self, expenses, clustered_resources_map):
         return list(result.values())
 
     def get_expenses(self, cloud_account_ids, resource_ids, start_date,
-                     end_date, limit=None) -> tuple:
+                     end_date, limit=0, offset=0) -> tuple:
         return self._get_expenses_clickhouse(
-            cloud_account_ids, resource_ids, start_date, end_date, limit)
+            cloud_account_ids, resource_ids, start_date, end_date, limit, offset)
+
+    def _get_offset_resource_ids(self, cloud_account_ids, resource_ids,
+                                 start_date, end_date, offset):
+        if not offset:
+            return []
+        query = """
+            SELECT resource_id, SUM(cost * sign) AS total_cost
+            FROM expenses
+            WHERE cloud_account_id IN cloud_account_ids
+                AND resource_id IN resource_ids
+                AND date >= %(start_date)s
+                AND date <= %(end_date)s
+            GROUP BY resource_id
+            HAVING SUM(sign) > 0
+            ORDER BY total_cost DESC
+            LIMIT %(limit)s
+        """
+        result = self.execute_clickhouse(
+            query=query,
+            params={
+                'start_date': start_date,
+                'end_date': end_date,
+                'limit': offset,
+            },
+            external_tables=[
+                {
+                    'name': 'resource_ids',
+                    'structure': [('_id', 'String')],
+                    'data': [{'_id': r_id} for r_id in resource_ids]
+                },
+                {
+                    'name': 'cloud_account_ids',
+                    'structure': [('_id', 'String')],
+                    'data': [{'_id': r_id} for r_id in cloud_account_ids]
+                }
+            ],
+        )
+        return [x[0] for x in result]
 
     def _get_expenses_clickhouse(self, cloud_account_ids, resource_ids,
-                                 start_date, end_date, limit) -> tuple:
+                                 start_date, end_date, limit, offset) -> tuple:
         query = """
             SELECT
                 cloud_account_id,
@@ -957,12 +995,15 @@ def _get_expenses_clickhouse(self, cloud_account_ids, resource_ids,
         """
         if limit:
             query += 'LIMIT %(limit)s'
+            if offset:
+                query += ' OFFSET %(offset)s'
         result = self.execute_clickhouse(
             query=query,
             params={
                 'start_date': start_date,
                 'end_date': end_date,
-                'limit': limit
+                'limit': limit,
+                'offset': offset
             },
             external_tables=[
                 {
@@ -1046,7 +1087,8 @@ def _split_params(self, organization_id, params):
     def split_params(self, organization_id, params):
         query_filters, data_filters, extra_filters = self._split_params(
             organization_id, params)
-        extra_filters['limit'] = query_filters.pop('limit', None)
+        extra_filters['limit'] = query_filters.pop('limit', 0)
+        extra_filters['offset'] = query_filters.pop('offset', 0)
         return query_filters, data_filters, extra_filters
 
     @staticmethod
@@ -1384,12 +1426,13 @@ def process_data(self, resources_data, organization_id, filters, **kwargs):
         total_cost = 0
         cloud_account_ids = kwargs['cloud_account_id']
         limit = kwargs['limit']
+        offset = kwargs['offset']
         _, organization_cloud_accs = self.get_organization_and_cloud_accs(
             organization_id)
         if not_clustered_resources:
             not_clustered_expenses, cost = self.get_expenses(
                 cloud_account_ids, not_clustered_resources, self.start_date,
-                self.end_date, limit)
+                self.end_date, limit=limit, offset=offset)
             total_cost += cost
         if clustered_resources_map:
             all_account_ids = list(map(
@@ -1406,6 +1449,13 @@ def process_data(self, resources_data, organization_id, filters, **kwargs):
             resource_ids.update(
                 list(map(lambda x: x.get('resource_id'), expenses))
             )
+            if offset and len(resource_ids) < limit:
+                offset_ids = self._get_offset_resource_ids(
+                    cloud_account_ids, not_clustered_resources, self.start_date,
+                    self.end_date, offset)
+                joined_ids = list(filter(
+                    lambda x: x not in offset_ids, joined_ids
+                ))[offset - len(offset_ids):]
             for r_id in joined_ids:
                 if len(resource_ids) == limit:
                     break
@@ -1430,6 +1480,8 @@ def process_data(self, resources_data, organization_id, filters, **kwargs):
         }
         if limit:
             res['limit'] = limit
+        if offset:
+            res['offset'] = offset
         return res
 
     def handle_filters(self, params, filters, organization_id):
@@ -1640,7 +1692,7 @@ def process_data(self, resources_data, organization_id, filters, **kwargs):
         return res
 
     def get_expenses(self, cloud_account_ids, resource_ids, start_date,
-                     end_date, limit=None) -> tuple:
+                     end_date, limit=0, offset=0) -> tuple:
         start = datetime.fromtimestamp(start_date)
         end = datetime.fromtimestamp(end_date)
         (