Sync upstream/integration (f850d9f) -> main 2026-06-23

README.md

@@ -33,6 +33,21 @@ It provides deep visibility into infrastructure costs, automated optimization re
 
 </div>
 
+<br>
+<br>
+
+<div>
+<br>  
+  <img src="documentation/images/OptScale-AI-description-ReadMe-GitHub.png" width="92" align="left" style="border-radius: 50%; margin-right: 15px">
+  <b>
+    🔖 NEW 🔖 <br>Cloud costs under control? Time to go beyond – into AI.
+  </b>
+  <br> OptScale AI covers it all – cost control, security and guardrails, and full visibility over every AI prompt, model, and agent.
+
+  <b>Try free at [optscale.ai](https://optscale.ai) </b>
+</div>
+
+<br>
 <br>
 
 <div>

insider/insider_worker/http_client/client.py

@@ -35,12 +35,13 @@ def request(self, url, method):
         response_body = None
         # pylint: disable=E1101
         if response.status_code != requests.codes.no_content:
-            if 'application/json' in response.headers['Content-Type']:
+            content_type = response.headers.get('Content-Type', '')
+            if 'application/json' in content_type:
                 response_body = json.loads(
                     response.content.decode('utf-8'))
-            if 'text/plain' in response.headers['Content-Type']:
+            elif 'text/plain' in content_type:
                 response_body = response.content.decode()
-            if 'application/octet-stream' in response.headers['Content-Type']:
+            else:
                 response_body = response.content
         return response.status_code, response_body
 

insider/insider_worker/main.py

@@ -21,6 +21,7 @@
 LOG = get_logger(__name__)
 TASK_EXCHANGE = Exchange(EXCHANGE_NAME, type='direct')
 TASK_QUEUE = Queue(QUEUE_NAME, TASK_EXCHANGE, routing_key=QUEUE_NAME)
+DISCOVERIES_THRESHOLD = 43200  # 12 hours in seconds
 
 
 class InsiderWorker(ConsumerMixin):
@@ -42,21 +43,36 @@ def discoveries(self):
 
     def get_consumers(self, consumer, channel):
         return [consumer(queues=[TASK_QUEUE], accept=['json'],
-                         callbacks=[self.process_task], prefetch_count=10)]
+                         callbacks=[self.process_task], prefetch_count=1)]
+
+    def get_last_discovery_ts(self, cloud_type):
+        discoveries = self.discoveries.find(
+            {'cloud_type': cloud_type, 'completed_at': {'$ne': 0}}
+        ).sort(
+            [('completed_at', -1)]).limit(1)
+        try:
+            discovery = next(discoveries)
+            return discovery.get('started_at', 0)
+        except StopIteration:
+            return 0
 
     def _process_task(self, task):
         start_process_time = int(datetime.now(tz=timezone.utc).timestamp())
         cloud_type = task.get('cloud_type')
         if not cloud_type:
             raise Exception('Invalid task received: {}'.format(task))
+        last_discovery_ts = self.get_last_discovery_ts(cloud_type)
+        if last_discovery_ts + DISCOVERIES_THRESHOLD >= start_process_time:
+            LOG.info('Skipping task for %s by threshold', cloud_type)
+            return
         discovery_id = self.discoveries.insert_one({
             'cloud_type': cloud_type,
             'started_at': start_process_time,
             'completed_at': 0
         }).inserted_id
 
         get_processor_class(cloud_type)(
-            self.mongo_client, self.config_cl).process_prices()
+            self.mongo_client, self.config_cl).process_prices(last_discovery_ts)
 
         end_process_time = int(datetime.now(tz=timezone.utc).timestamp())
         self.discoveries.update_one(

insider/insider_worker/migrations/20260618150755_last_seen_currency_index.py

@@ -0,0 +1,53 @@
+import logging
+from insider.insider_worker.migrations.base import BaseMigration
+
+NEW_INDEXES = {
+    'CurrencyLastSeen': ['currencyCode', 'last_seen']
+}
+OLD_INDEXES = {
+    'LastSeen': ['last_seen']
+}
+LOG = logging.getLogger(__name__)
+
+
+class Migration(BaseMigration):
+    def get_indexes(self):
+        return [x['name'] for x in self.azure_prices.list_indexes()]
+
+    def upgrade(self):
+        existing_indexes = self.get_indexes()
+        for index_name, index_fields in NEW_INDEXES.items():
+            if index_name in existing_indexes:
+                LOG.info(f'Index {index_name} already exists')
+                continue
+            LOG.info(f'Creating index {index_name}')
+            self.azure_prices.create_index(
+                [(f, 1) for f in index_fields],
+                name=index_name,
+                background=True
+            )
+        for index_name, index_fields in OLD_INDEXES.items():
+            if index_name in existing_indexes:
+                LOG.info(f'Dropping index {index_name}')
+                self.azure_prices.drop_index(index_name)
+            else:
+                LOG.info(f'Index {index_name} doesn\'t exist')
+
+    def downgrade(self):
+        existing_indexes = self.get_indexes()
+        for index_name, index_fields in OLD_INDEXES.items():
+            if index_name in existing_indexes:
+                LOG.info(f'Index {index_name} already exists')
+                continue
+            LOG.info(f'Creating index {index_name}')
+            self.azure_prices.create_index(
+                [(f, 1) for f in index_fields],
+                name=index_name,
+                background=True
+            )
+        for index_name, index_fields in NEW_INDEXES.items():
+            if index_name in existing_indexes:
+                LOG.info(f'Dropping index {index_name}')
+                self.azure_prices.drop_index(index_name)
+            else:
+                LOG.info(f'Index {index_name} doesn\'t exist')

insider/insider_worker/processors/azure.py

@@ -15,8 +15,8 @@
 ACTIVITIES_EXCHANGE_NAME = 'activities-tasks'
 ACTIVITIES_EXCHANGE = Exchange(ACTIVITIES_EXCHANGE_NAME, type='topic')
 LOG = get_logger(__name__)
-PRICES_PER_REQUEST = 100
 PRICES_COUNT_TO_LOG = 1000
+CHINA_CURRENCY_CODE = 'CNY'
 
 
 class AzurePriceProcessor(BasePriceProcessor):
@@ -35,16 +35,6 @@ def discoveries(self):
     def prices(self):
         return self.mongo_client.insider.azure_prices
 
-    def get_last_discovery(self):
-        discoveries = self.discoveries.find(
-            {'cloud_type': self.CLOUD_TYPE, 'completed_at': {'$ne': 0}}
-        ).sort(
-            [('completed_at', -1)]).limit(1)
-        try:
-            return next(discoveries)
-        except StopIteration:
-            return {}
-
     @staticmethod
     def unique_values(price):
         return tuple(price.get(p) for p in AzurePriceProcessor.UNIQUE_FIELDS)
@@ -82,42 +72,39 @@ def _get_currencies_list(self):
         currencies = set(map(lambda x: x['currency'], orgs['organizations']))
         return list(currencies)
 
-    def _process_global_prices(self, http_client, old_prices_map):
-        LOG.info('Start processing Azure Global prices')
-        for currency in self._get_currencies_list():
-            LOG.info('Processing Azure prices for currency: %s', currency)
-            processed_keys = {}
-            prices_counter = 0
-
-            next_page = 'https://prices.azure.com/api/retail/prices'
-            next_page += '?currencyCode=%s' % currency
-            while True:
-                if prices_counter % PRICES_COUNT_TO_LOG == 0:
-                    LOG.info('Total number of prices got from '
-                             'cloud: %s', prices_counter)
-                try:
-                    code, response = http_client.get(next_page)
-                except SSLError:
-                    LOG.error('Getting Azure prices failed with SSL '
-                              'verification error. Will try to get prices'
-                              'without SSL verification')
-                    self.send_sslerror_service_email()
-                    http_client = Client(verify=False)
-                    code, response = http_client.get(next_page)
-                items = response.get('Items', [])
-                new_prices_map = {self.unique_values(p): p for p in items}
-                self.update_price_records(new_prices_map, old_prices_map,
-                                          processed_keys)
-                new_url = response.get('NextPageLink')
-                if not new_url or new_url == next_page:
-                    LOG.info('Total number of prices got from '
-                             'cloud: %s', prices_counter)
-                    break
-                next_page = new_url
-                prices_counter += response.get('Count', 0)
-
-    def _process_china_prices(self, http_client, old_prices_map):
-        LOG.info('Start processing Azure China prices')
+    def _process_global_prices(self, http_client, old_prices_map, currency):
+        LOG.info('Processing Azure prices for currency: %s', currency)
+        processed_keys = {}
+        prices_counter = 0
+        next_page = 'https://prices.azure.com/api/retail/prices'
+        next_page += '?currencyCode=%s' % currency
+        while True:
+            if prices_counter % PRICES_COUNT_TO_LOG == 0:
+                LOG.info('Total number of prices got from '
+                         'cloud: %s', prices_counter)
+            try:
+                code, response = http_client.get(next_page)
+            except SSLError:
+                LOG.error('Getting Azure prices failed with SSL '
+                          'verification error. Will try to get prices'
+                          'without SSL verification')
+                self.send_sslerror_service_email()
+                http_client = Client(verify=False)
+                code, response = http_client.get(next_page)
+            items = response.get('Items', [])
+            new_prices_map = {self.unique_values(p): p for p in items}
+            self.update_price_records(new_prices_map, old_prices_map,
+                                      processed_keys)
+            new_url = response.get('NextPageLink')
+            if not new_url or new_url == next_page:
+                LOG.info('Total number of prices got from '
+                         'cloud: %s', prices_counter)
+                break
+            next_page = new_url
+            prices_counter += response.get('Count', 0)
+
+    def _process_china_prices(self, http_client, old_prices_map, currency):
+        LOG.info('Start processing Azure China prices (%s)', currency)
         url = 'https://prices.azure.cn/api/retail/pricesheet/download?' \
               'api-version=2023-06-01-preview'
         _, response = http_client.get(url)
@@ -130,17 +117,28 @@ def _process_china_prices(self, http_client, old_prices_map):
         LOG.info('Total number of prices got from cloud: %s',
                  len(new_prices_map))
 
-    def process_prices(self):
-        last_discovery = self.get_last_discovery()
-        old_prices = self.prices.find(
-            {'last_seen': {'$gte': last_discovery.get('started_at', 0)}},
-            {k: 1 for k in self.UNIQUE_FIELDS + self.CHANGE_FIELDS + ['last_seen']}
-        )
-        old_prices_map = {self.unique_values(p): p for p in old_prices}
-
+    def process_prices(self, last_discovery_ts):
         http_client = Client()
-        self._process_global_prices(http_client, old_prices_map)
-        self._process_china_prices(http_client, old_prices_map)
+        process_func_map = {
+            CHINA_CURRENCY_CODE: self._process_china_prices
+        }
+        for currency in self._get_currencies_list():
+            old_prices = self.prices.find(
+                {
+                    'last_seen': {
+                        '$gte': last_discovery_ts
+                    },
+                    'currencyCode': currency
+                },
+                {
+                    k: 1 for k in
+                    self.UNIQUE_FIELDS + self.CHANGE_FIELDS + ['last_seen']
+                }
+            )
+            old_prices_map = {self.unique_values(p): p for p in old_prices}
+            process_func = process_func_map.get(
+                currency, self._process_global_prices)
+            process_func(http_client, old_prices_map, currency)
 
     def update_price_records(self, new_prices_map, old_prices_map,
                              processed_keys):

insider/insider_worker/processors/base.py

@@ -11,8 +11,5 @@ def discoveries(self):
     def prices(self):
         raise NotImplementedError()
 
-    def get_last_discovery(self):
-        raise NotImplementedError()
-
-    def process_prices(self):
+    def process_prices(self, last_discovery_ts):
         raise NotImplementedError()

ngui/ui/src/components/TopAlertWrapper/TopAlert.styles.ts

@@ -33,6 +33,13 @@ const useStyles = makeStyles()((theme) => ({
       color: theme.palette.common.white,
     },
   },
+  promo: {
+    backgroundColor: "#2c67ce",
+    color: theme.palette.common.white,
+    ".close-alert-button": {
+      color: theme.palette.common.white,
+    },
+  },
 }));
 
 export default useStyles;

ngui/ui/src/components/TopAlertWrapper/TopAlertWrapper.tsx

@@ -1,13 +1,13 @@
 import { useCallback, useEffect, useMemo } from "react";
-import { Box } from "@mui/material";
+import { Box, Link } from "@mui/material";
 import { render as renderGithubButton } from "github-buttons";
 import { FormattedMessage, useIntl } from "react-intl";
 import { useDispatch } from "react-redux";
 import { useAllDataSources } from "hooks/coreData/useAllDataSources";
 import { useGetToken } from "hooks/useGetToken";
 import { useOrganizationInfo } from "hooks/useOrganizationInfo";
 import { useRootData } from "hooks/useRootData";
-import { GITHUB_HYSTAX_OPTSCALE_REPO } from "urls";
+import { GITHUB_HYSTAX_OPTSCALE_REPO, OPTSCALE_AI } from "urls";
 import { AZURE_TENANT, ENVIRONMENT } from "utils/constants";
 import { SPACING_1 } from "utils/layouts";
 import { updateOrganizationTopAlert as updateOrganizationTopAlertActionCreator } from "./actionCreators";
@@ -132,12 +132,14 @@ const TopAlertWrapper = ({ blacklistIds = [] }: TopAlertWrapperProps) => {
       },
       {
         id: ALERT_TYPES.OPEN_SOURCE_ANNOUNCEMENT,
+        // Temporarily disabled — replaced by OPTSCALE_AI_PROMO_ANNOUNCEMENT below.
+        // To restore, revert condition back to: !isExistingUser && (!userId || organizationId)
         // isExistingUser — true only if user was logged in/visited optscale before. Set in migrations.
         // organizationId — wont be presented on initial load (so storedAlerts will be empty, so even if banner was closed, we would not know that,
         //                  so we need to wait for organizationId. But if user is not logged in — there also wont be organizationId, so we use next flag)
         // userId — presented after login
         // this check means "condition: not logged in new user (!isExistingUser && !userId) OR new user and we know organization id (!isExistingUser && organizationId)"
-        condition: !isExistingUser && (!userId || organizationId),
+        condition: false,
         getContent: () => (
           <Box sx={{ textAlign: "center" }}>
             <FormattedMessage
@@ -163,6 +165,44 @@ const TopAlertWrapper = ({ blacklistIds = [] }: TopAlertWrapperProps) => {
         },
         dataTestId: "top_alert_open_source_announcement",
       },
+      {
+        id: ALERT_TYPES.OPTSCALE_AI_PROMO_ANNOUNCEMENT,
+        // isExistingUser — true only if user was logged in/visited optscale before. Set in migrations.
+        // organizationId — wont be presented on initial load (so storedAlerts will be empty, so even if banner was closed, we would not know that,
+        //                  so we need to wait for organizationId. But if user is not logged in — there also wont be organizationId, so we use next flag)
+        // userId — presented after login
+        // this check means "condition: not logged in new user (!isExistingUser && !userId) OR new user and we know organization id (!isExistingUser && organizationId)"
+        condition: !isExistingUser && (!userId || organizationId),
+        getContent: () => (
+          <Box sx={{ textAlign: "center" }}>
+            <FormattedMessage
+              id="optScaleAiPromoAnnouncement"
+              values={{
+                strong: (chunks) => <strong>{chunks}</strong>,
+                br: () => <br />,
+                link: (chunks) => (
+                  <Link
+                    href={OPTSCALE_AI}
+                    target="_blank"
+                    rel="noopener"
+                    color="inherit"
+                    underline="always"
+                    sx={{ fontWeight: "bold" }}
+                  >
+                    {chunks}
+                  </Link>
+                ),
+              }}
+            />
+          </Box>
+        ),
+        type: "promo",
+        triggered: isTriggered(ALERT_TYPES.OPTSCALE_AI_PROMO_ANNOUNCEMENT),
+        onClose: () => {
+          updateOrganizationTopAlert({ id: ALERT_TYPES.OPTSCALE_AI_PROMO_ANNOUNCEMENT, closed: true });
+        },
+        dataTestId: "top_alert_optscale_ai_promo_announcement",
+      },
     ];
   }, [
     storedAlerts,

ngui/ui/src/components/TopAlertWrapper/constants.ts

@@ -3,6 +3,7 @@ export const ALERT_TYPES = Object.freeze({
   DATA_SOURCES_PROCEEDED: 3,
   OPEN_SOURCE_ANNOUNCEMENT: 4,
   INACTIVE_ORGANIZATION: 5,
+  OPTSCALE_AI_PROMO_ANNOUNCEMENT: 7,
 });
 
 export const IS_EXISTING_USER = "isExistingUser";