[Monitor] Fix #32631: az monitor log-analytics query: Add AzureChinaCloud/sovereign cloud Log Analytics query endpoint mapping

src/azure-cli-core/azure/cli/core/tests/test_cloud.py

@@ -21,6 +21,8 @@
                                   update_cloud,
                                   cloud_is_registered,
                                   AZURE_PUBLIC_CLOUD,
+                                  AZURE_CHINA_CLOUD,
+                                  AZURE_US_GOV_CLOUD,
                                   KNOWN_CLOUDS,
                                   update_cloud,
                                   CloudEndpointNotSetException,
@@ -245,6 +247,26 @@ def test_metadata_url_endpoints(self):
                 v2 = metadata_url_cloud.suffixes.__dict__[k]
                 self.assertEqual(v1, v2)
 
+    def test_log_analytics_resource_id_sovereign_clouds(self):
+        """Regression test: log_analytics_resource_id must be set for all supported sovereign clouds
+        so that 'az vm monitor log show' and related Log Analytics data-plane commands work."""
+        clouds_with_log_analytics = [
+            (AZURE_PUBLIC_CLOUD, 'https://api.loganalytics.io'),
+            (AZURE_CHINA_CLOUD, 'https://api.loganalytics.azure.cn'),
+            (AZURE_US_GOV_CLOUD, 'https://api.loganalytics.us'),
+        ]
+        for cloud, expected_endpoint in clouds_with_log_analytics:
+            with self.subTest(cloud=cloud.name):
+                self.assertTrue(
+                    cloud.endpoints.has_endpoint_set('log_analytics_resource_id'),
+                    msg="log_analytics_resource_id not set for {}".format(cloud.name)
+                )
+                self.assertEqual(
+                    cloud.endpoints.log_analytics_resource_id,
+                    expected_endpoint,
+                    msg="Unexpected log_analytics_resource_id value for {}".format(cloud.name)
+                )
+
 
 if __name__ == '__main__':
     unittest.main()

src/azure-cli/azure/cli/command_modules/vm/_client_factory.py

@@ -37,8 +37,20 @@ def cf_log_analytics_data_plane(cli_ctx, _):
     """Initialize Log Analytics data client for use with CLI."""
     from azure.monitor.query import LogsQueryClient
     from azure.cli.core._profile import Profile
+    from knack.util import CLIError
+
+    if not cli_ctx.cloud.endpoints.has_endpoint_set('log_analytics_resource_id'):
+        raise CLIError(
+            "The Log Analytics query data-plane endpoint is not configured for cloud '{cloud}'. "
+            "This feature may not be available in '{cloud}'. "
+            "If you believe this is an error, configure the endpoint with: "
+            "az cloud update --endpoint-log-analytics-resource-id <endpoint-url>".format(
+                cloud=cli_ctx.cloud.name
+            )
+        )
+
     profile = Profile(cli_ctx=cli_ctx)
     cred, _, _ = profile.get_login_credentials()
     api_version = 'v1'
-    return LogsQueryClient(cred, endpoint=cli_ctx.cloud.endpoints.log_analytics_resource_id + '/' + api_version,
-                           audience=cli_ctx.cloud.endpoints.log_analytics_resource_id)
+    endpoint = cli_ctx.cloud.endpoints.log_analytics_resource_id
+    return LogsQueryClient(cred, endpoint=endpoint + '/' + api_version, audience=endpoint)

src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_actions.py

@@ -649,5 +649,75 @@ def test_process_gallery_image_version_namespace(self):
             process_gallery_image_version_namespace(cmd, np)
 
 
+class TestLogAnalyticsDataPlaneClient(unittest.TestCase):
+    """Tests for cf_log_analytics_data_plane endpoint resolution across sovereign clouds."""
+
+    def _make_mock_cli_ctx(self, has_endpoint=True, endpoint_value='https://api.loganalytics.io',
+                           cloud_name='AzureCloud'):
+        cli_ctx = mock.MagicMock()
+        cli_ctx.cloud.name = cloud_name
+        if has_endpoint:
+            cli_ctx.cloud.endpoints.has_endpoint_set.return_value = True
+            cli_ctx.cloud.endpoints.log_analytics_resource_id = endpoint_value
+        else:
+            cli_ctx.cloud.endpoints.has_endpoint_set.return_value = False
+        return cli_ctx
+
+    @mock.patch('azure.monitor.query.LogsQueryClient')
+    @mock.patch('azure.cli.core._profile.Profile')
+    def test_cf_log_analytics_data_plane_public_cloud(self, mock_profile_cls, mock_client_cls):
+        from azure.cli.command_modules.vm._client_factory import cf_log_analytics_data_plane
+        mock_profile_cls.return_value.get_login_credentials.return_value = ('cred', None, None)
+        cli_ctx = self._make_mock_cli_ctx(has_endpoint=True, endpoint_value='https://api.loganalytics.io')
+        cf_log_analytics_data_plane(cli_ctx, None)
+        mock_client_cls.assert_called_once_with(
+            'cred',
+            endpoint='https://api.loganalytics.io/v1',
+            audience='https://api.loganalytics.io'
+        )
+
+    @mock.patch('azure.monitor.query.LogsQueryClient')
+    @mock.patch('azure.cli.core._profile.Profile')
+    def test_cf_log_analytics_data_plane_china_cloud(self, mock_profile_cls, mock_client_cls):
+        from azure.cli.command_modules.vm._client_factory import cf_log_analytics_data_plane
+        mock_profile_cls.return_value.get_login_credentials.return_value = ('cred', None, None)
+        cli_ctx = self._make_mock_cli_ctx(
+            has_endpoint=True,
+            endpoint_value='https://api.loganalytics.azure.cn',
+            cloud_name='AzureChinaCloud'
+        )
+        cf_log_analytics_data_plane(cli_ctx, None)
+        mock_client_cls.assert_called_once_with(
+            'cred',
+            endpoint='https://api.loganalytics.azure.cn/v1',
+            audience='https://api.loganalytics.azure.cn'
+        )
+
+    @mock.patch('azure.monitor.query.LogsQueryClient')
+    @mock.patch('azure.cli.core._profile.Profile')
+    def test_cf_log_analytics_data_plane_usgov_cloud(self, mock_profile_cls, mock_client_cls):
+        from azure.cli.command_modules.vm._client_factory import cf_log_analytics_data_plane
+        mock_profile_cls.return_value.get_login_credentials.return_value = ('cred', None, None)
+        cli_ctx = self._make_mock_cli_ctx(
+            has_endpoint=True,
+            endpoint_value='https://api.loganalytics.us',
+            cloud_name='AzureUSGovernment'
+        )
+        cf_log_analytics_data_plane(cli_ctx, None)
+        mock_client_cls.assert_called_once_with(
+            'cred',
+            endpoint='https://api.loganalytics.us/v1',
+            audience='https://api.loganalytics.us'
+        )
+
+    def test_cf_log_analytics_data_plane_missing_endpoint_raises_clear_error(self):
+        from azure.cli.command_modules.vm._client_factory import cf_log_analytics_data_plane
+        cli_ctx = self._make_mock_cli_ctx(has_endpoint=False, cloud_name='AzureCustomCloud')
+        with self.assertRaises(CLIError) as cm:
+            cf_log_analytics_data_plane(cli_ctx, None)
+        self.assertIn('AzureCustomCloud', str(cm.exception))
+        self.assertIn('az cloud update', str(cm.exception))
+
+
 if __name__ == '__main__':
     unittest.main()