Solomon external data source bugfixes (#26319)

Co-authored-by: Copilot <[email protected]>

Author: evanevanevanevannnn

ydb/library/yql/providers/solomon/actors/dq_solomon_read_actor.cpp

@@ -348,7 +348,7 @@ class TDqSolomonReadActor : public NActors::TActorBootstrapped<TDqSolomonReadAct
 
             for (size_t i = 0; i < timestamps.size(); ++i){
                 TInstant timestamp = TInstant::MilliSeconds(timestamps[i]);
-                if (timestamp < from || timestamp > to) {
+                if (timestamp < from || timestamp >= to) {
                     continue;
                 }
 

ydb/library/yql/providers/solomon/provider/yql_solomon_dq_integration.cpp

@@ -137,7 +137,7 @@ class TSolomonDqIntegration: public TDqIntegrationBase {
                         return {};
                     }
                     if (!TInstant::TryParseIso8601(value, from)) {
-                        ctx.AddError(TIssue(ctx.GetPosition(settingsRef.Child(i)->Head().Pos()), "couldn't parse `from`, use Iso8601 format, e.g. 2025-03-12T14:40:39Z"));
+                        ctx.AddError(TIssue(ctx.GetPosition(settingsRef.Child(i)->Head().Pos()), "couldn't parse `from`, use ISO8601 format, e.g. 2025-03-12T14:40:39Z"));
                         return {};
                     }
                     continue;
@@ -148,7 +148,7 @@ class TSolomonDqIntegration: public TDqIntegrationBase {
                         return {};
                     }
                     if (!TInstant::TryParseIso8601(value, to)) {
-                        ctx.AddError(TIssue(ctx.GetPosition(settingsRef.Child(i)->Head().Pos()), "couldn't parse `to`, use Iso8601 format, e.g. 2025-03-12T14:40:39Z"));
+                        ctx.AddError(TIssue(ctx.GetPosition(settingsRef.Child(i)->Head().Pos()), "couldn't parse `to`, use ISO8601 format, e.g. 2025-03-12T14:40:39Z"));
                         return {};
                     }
                     continue;

ydb/library/yql/providers/solomon/provider/yql_solomon_load_meta.cpp

@@ -69,14 +69,20 @@ class TSolomonLoadTableMetadataTransformer : public TGraphTransformerBase {
 
                 TInstant from;
                 if (auto time = ExtractSetting(settings, "from")) {
-                    from = TInstant::ParseIso8601(*time);
+                    if (!TInstant::TryParseIso8601(*time, from)) {
+                        ctx.AddError(TIssue(ctx.GetPosition(n->Pos()), "couldn't parse `from`, use ISO8601 format, e.g. 2025-03-12T14:40:39Z"));
+                        return TStatus::Error;
+                    }
                 } else {
-                    from = TInstant::Now() - TDuration::Days(7);
+                    from = TInstant::Zero();
                 }
 
                 TInstant to;
                 if (auto time = ExtractSetting(settings, "to")) {
-                    to = TInstant::ParseIso8601(*time);
+                    if (!TInstant::TryParseIso8601(*time, to)) {
+                        ctx.AddError(TIssue(ctx.GetPosition(n->Pos()), "couldn't parse `to`, use ISO8601 format, e.g. 2025-03-12T14:40:39Z"));
+                        return TStatus::Error;
+                    }
                 } else {
                     to = TInstant::Now();
                 }

ydb/library/yql/providers/solomon/solomon_accessor/client/solomon_accessor_client.cpp

@@ -74,32 +74,33 @@ TGetLabelsResponse ProcessGetLabelsResponse(NYql::IHTTPGateway::TResult&& respon
     TGetLabelsResult result;
 
     if (response.CurlResponseCode != CURLE_OK) {
-        return TGetLabelsResponse(TStringBuilder{} << "Error while sending list metric names request to monitoring api: " << response.Issues.ToOneLineString() << " (curl status: " << curl_easy_strerror(response.CurlResponseCode) << ")");
+        return TGetLabelsResponse(TStringBuilder{} << "Monitoring api get labels response: " << response.Issues.ToOneLineString() <<
+            ", internal code: " << static_cast<int>(response.CurlResponseCode));
     }
 
     if (response.Content.HttpResponseCode < 200 || response.Content.HttpResponseCode >= 300) {
-        return TGetLabelsResponse(TStringBuilder{} << "Error while sending list metric names request to monitoring api: " << response.Content.data() << " (http status: " << response.Content.HttpResponseCode << ")");
+        return TGetLabelsResponse(TStringBuilder{} << "Monitoring api get labels response: " << response.Content.data() <<
+            ", internal code: " << response.Content.HttpResponseCode);
     }
 
     NJson::TJsonValue json;
     try {
         NJson::ReadJsonTree(response.Content.data(), &json, /*throwOnError*/ true);
     } catch (const std::exception& e) {
-        return TGetLabelsResponse(TStringBuilder{} << "Failed to parse response from monitoring api: " << e.what());
+        return TGetLabelsResponse("Monitoring api get labels response is not a valid json");
     }
 
     if (!json.IsMap() || !json.Has("names") || !json["names"].IsArray()) {
-        return TGetLabelsResponse("Invalid result from monitoring api");
+        return TGetLabelsResponse("Monitoring api get labels response doesn't contain requested info");
     }
 
     const auto names = json["names"].GetArray();
 
     for (const auto& name : names) {
         if (!name.IsString()) {
-            return TGetLabelsResponse("Invalid label names from monitoring api");
-        } else {
-            result.Labels.push_back(name.GetString());
+            return TGetLabelsResponse("Monitoring api get labels response contains invalid label names");
         }
+        result.Labels.push_back(name.GetString());
     }
     for (const auto& [key, selector] : knownSelectors) {
         result.Labels.push_back(key);
@@ -112,37 +113,39 @@ TListMetricsResponse ProcessListMetricsResponse(NYql::IHTTPGateway::TResult&& re
     TListMetricsResult result;
 
     if (response.CurlResponseCode != CURLE_OK) {
-        return TListMetricsResponse(TStringBuilder{} << "Error while sending list metrics request to monitoring api: " << response.Issues.ToOneLineString());
+        return TListMetricsResponse(TStringBuilder{} << "Monitoring api list metrics response: " << response.Issues.ToOneLineString() <<
+            ", internal code: " << static_cast<int>(response.CurlResponseCode));
     }
 
     if (response.Content.HttpResponseCode < 200 || response.Content.HttpResponseCode >= 300) {
-        return TListMetricsResponse(TStringBuilder{} << "Error while sending list metrics request to monitoring api: " << response.Content.data());
+        return TListMetricsResponse(TStringBuilder{} << "Monitoring api list metrics response: " << response.Content.data() <<
+            ", internal code: " << response.Content.HttpResponseCode);
     }
 
     NJson::TJsonValue json;
     try {
         NJson::ReadJsonTree(response.Content.data(), &json, /*throwOnError*/ true);
     } catch (const std::exception& e) {
-        return TListMetricsResponse(TStringBuilder{} << "Failed to parse response from monitoring api: " << e.what());
+        return TListMetricsResponse("Monitoring api list metrics response is not a valid json" );
     }
 
     if (!json.IsMap() || !json.Has("result") || !json.Has("page")) {
-        return TListMetricsResponse("Invalid list metrics result from monitoring api");
+        return TListMetricsResponse("Monitoring api list metrics response doesn't contain requested info");
     }
 
     const auto pagesInfo = json["page"];
     if (!pagesInfo.IsMap() || 
         !pagesInfo.Has("pagesCount") || !pagesInfo["pagesCount"].IsInteger() || 
         !pagesInfo.Has("totalCount") || !pagesInfo["totalCount"].IsInteger()) {
-        return TListMetricsResponse("Invalid paging info from monitoring api");
+        return TListMetricsResponse("Monitoring api list metrics response doesn't contain paging info");
     }
 
     result.PagesCount = pagesInfo["pagesCount"].GetInteger();
     result.TotalCount = pagesInfo["totalCount"].GetInteger();
 
     for (const auto& metricObj : json["result"].GetArray()) {
         if (!metricObj.IsMap() || !metricObj.Has("labels") || !metricObj["labels"].IsMap() || !metricObj.Has("type") || !metricObj["type"].IsString()) {
-            return TListMetricsResponse("Invalid list metrics result from monitoring api");
+            return TListMetricsResponse("Monitoring api list metrics response contains invalid metrics");
         }
 
         TSelectors selectors;
@@ -173,22 +176,24 @@ TGetPointsCountResponse ProcessGetPointsCountResponse(NYql::IHTTPGateway::TResul
             }
         }
 
-        return TGetPointsCountResponse(TStringBuilder() << "Error while sending points count request to monitoring api: " << issues);
+        return TGetPointsCountResponse(TStringBuilder() << "Monitoring api points count response: " << issues <<
+            ", internal code: " << static_cast<int>(response.CurlResponseCode));
     }
 
     if (response.Content.HttpResponseCode < 200 || response.Content.HttpResponseCode >= 300) {
-        return TGetPointsCountResponse(TStringBuilder{} << "Error while sending points count request to monitoring api: " << response.Content.data());
+        return TGetPointsCountResponse(TStringBuilder{} << "Monitoring api points count response: " << response.Content.data() <<
+            ", internal code: " << response.Content.HttpResponseCode);
     }
 
     NJson::TJsonValue json;
     try {
         NJson::ReadJsonTree(response.Content.data(), &json, /*throwOnError*/ true);
     } catch (const std::exception& e) {
-        return TGetPointsCountResponse(TStringBuilder{} << "Failed to parse points count response from monitoring api: " << e.what());
+        return TGetPointsCountResponse("Monitoring api points count response is not a valid json");
     }
 
     if (!json.IsMap() || !json.Has("scalar") || !json["scalar"].IsInteger()) {
-        return TGetPointsCountResponse("Invalid points count result from monitoring api");
+        return TGetPointsCountResponse("Monitoring api points count response doesn't contain requested info");
     }
 
     result.PointsCount = json["scalar"].GetInteger() + downsampledPointsCount;
@@ -200,15 +205,15 @@ TGetDataResponse ProcessGetDataResponse(NYdbGrpc::TGrpcStatus&& status, ReadResp
     TGetDataResult result;
 
     if (!status.Ok()) {
-        TString error = TStringBuilder{} << "Error while sending data request to monitoring api: " << status.Msg;
+        TString error = TStringBuilder{} << "Monitoring api get data response: " << status.Msg;
         if (status.GRpcStatusCode == grpc::StatusCode::RESOURCE_EXHAUSTED || status.GRpcStatusCode == grpc::StatusCode::UNAVAILABLE) {
             return TGetDataResponse(error, EStatus::STATUS_RETRIABLE_ERROR);
         }
         return TGetDataResponse(error);
     }
 
     if (response.response_per_query_size() != 1) {
-        return TGetDataResponse("Invalid get data repsonse size from monitoring api");
+        return TGetDataResponse("Monitoring api get data response is invalid");
     }
 
     const auto& responseValue = response.response_per_query()[0];
@@ -579,7 +584,7 @@ class TSolomonAccessorClient : public ISolomonAccessorClient, public std::enable
                 fullSelectors[labelName] = {"=", "-"};
             }
         }
-        return selectors;
+        return fullSelectors;
     }
 
     TString BuildSelectorsProgram(const TSelectors& selectors, bool useNewFormat = false) const {

ydb/library/yql/tests/sql/solomon/canondata/test.test_solomon-InvalidProject-_/extracted

@@ -6,4 +6,4 @@
     <tmp_path>/program.sql:<main>: Error: Fatal Error
 
 
-        <tmp_path>/program.sql:<main>: Error: Error while sending data request to monitoring api: Project invalid does not exist (appeared 1 time at ISOTIME)
+        <tmp_path>/program.sql:<main>: Error: Monitoring api get data response: Project invalid does not exist (appeared 1 time at ISOTIME)

ydb/tests/fq/solomon/canondata/test.test_solomon-InvalidProject-_/extracted

@@ -1,7 +1,7 @@
 Failed to execute query, reason:
 Request finished with status: EXTERNAL_ERROR
 Issues:
-<main>: Error: Error while sending data request to monitoring api: Project invalid does not exist
+<main>: Error: Monitoring api get data response: Project invalid does not exist
 
 
 

ydb/tests/solomon/reading/base.py

@@ -13,8 +13,8 @@
 class SolomonReadingTestBase(object):
     @classmethod
     def setup_class(cls):
-        cls.basic_reading_timestamps = [0, 5, 10, 15, 20, 25, 30, 35, 40, 45, 50, 55, 60]
-        cls.basic_reading_values = [0, 1,  2,  3,  4,  5,  6,  7,  8,  9, 10, 11, 12]
+        cls.basic_reading_timestamps = [0, 5, 10, 15, 20, 25, 30, 35, 40, 45, 50, 55]
+        cls.basic_reading_values = [0, 1,  2,  3,  4,  5,  6,  7,  8,  9, 10, 11]
 
         cls.listing_paging_metrics_size = 1000
 
@@ -96,7 +96,7 @@ def execute_query(cls, query):
 
     @staticmethod
     def _generate_listing_paging_test_metrics(size):
-        return [
+        listing_paging_metrics = [
             {
                 "labels"        : {"test_type": "listing_paging_test", "test_label": str(i)},
                 "type"          : "DGAUGE",
@@ -106,6 +106,15 @@ def _generate_listing_paging_test_metrics(size):
             for i in range(size)
         ]
 
+        listing_paging_metrics.append({
+            "labels"        : {"test_type": "listing_paging_test"},
+            "type"          : "DGAUGE",
+            "timestamps"    : [0],
+            "values"        : [0]
+        })
+
+        return listing_paging_metrics
+
     @staticmethod
     def _generate_data_paging_timeseries(size):
         timestamps = [i * 5 for i in range(size)]

ydb/tests/solomon/reading/basic_reading.py

@@ -27,6 +27,17 @@ def check_query_result(self, result, error, downsampling_disabled):
             return False, "values differ from canonical, have {}, should be {}".format(values, canon_values)
         return True, None
 
+    def check_query_result_size(self, result, error):
+        if error is not None:
+            return False, error
+
+        result_size = len(result[0].rows)
+
+        if (result_size != 1):
+            return False, "should only have a single return row, have: {}".format(result_size)
+
+        return True, None
+
     @link_test_case("#16398")
     def test_basic_reading_solomon(self):
         data_source_query = f"""
@@ -144,6 +155,18 @@ def test_basic_reading_solomon(self):
         assert error is None, error
         assert any(column.name == "tt" for column in result[0].columns)
 
+        # query with a single second interval
+        query = """
+            SELECT * FROM local_solomon.basic_reading WITH (
+                selectors = @@{cluster="basic_reading", service="my_service", test_type="basic_reading_test"}@@,
+
+                from = "1970-01-01T00:00:00Z",
+                to = "1970-01-01T00:00:01Z"
+            )
+        """
+        succes, error = self.check_query_result_size(*self.execute_query(query))
+        assert succes, error
+
     @link_test_case("#23192")
     def test_basic_reading_monitoring(self):
         data_source_query = f"""
@@ -262,3 +285,15 @@ def test_basic_reading_monitoring(self):
         result, error = self.execute_query(query)
         assert error is None, error
         assert any(column.name == "tt" for column in result[0].columns)
+
+        # query with a single second interval
+        query = """
+            SELECT * FROM local_monitoring.my_service WITH (
+                selectors = @@{test_type="basic_reading_test"}@@,
+
+                from = "1970-01-01T00:00:00Z",
+                to = "1970-01-01T00:00:01Z"
+            )
+        """
+        succes, error = self.check_query_result_size(*self.execute_query(query))
+        assert succes, error

ydb/tests/solomon/reading/listing_paging.py

@@ -7,7 +7,7 @@
 
 
 class TestListingPaging(SolomonReadingTestBase):
-    def check_listing_paging_result(self, result_set, error):
+    def check_full_listing_result(self, result_set, error):
         if error is not None:
             return False, error
 
@@ -29,6 +29,19 @@ def check_listing_paging_result(self, result_set, error):
 
         return True, None
 
+    def check_listing_size(self, result_set, error):
+        if error is not None:
+            return False, error
+
+        rows = []
+        for result in result_set:
+            rows.extend(result.rows)
+
+        if (len(rows) != self.listing_paging_metrics_size + 1):
+            return False, "Result size differs from expected: have {}, should be {}".format(len(rows), self.listing_paging_metrics_size)
+
+        return True, None
+
     @link_test_case("#16395")
     def test_listing_paging_solomon(self):
         data_source_query = f"""
@@ -42,7 +55,6 @@ def test_listing_paging_solomon(self):
         result, error = self.execute_query(data_source_query)
         assert error is None
 
-        # simplest query with default downsampling settings
         query = """
             SELECT test_label FROM local_solomon.listing_paging WITH (
                 selectors = @@{cluster="listing_paging", service="my_service", test_type="listing_paging_test", test_label="*"}@@,
@@ -53,7 +65,18 @@ def test_listing_paging_solomon(self):
                 to = "1970-01-01T00:01:00Z"
             )
         """
-        success, error = self.check_listing_paging_result(*self.execute_query(query))
+        success, error = self.check_full_listing_result(*self.execute_query(query))
+        assert success, error
+
+        query = """
+            SELECT * FROM local_solomon.listing_paging WITH (
+                selectors = @@{cluster="listing_paging", service="my_service", test_type="listing_paging_test"}@@,
+
+                from = "1970-01-01T00:00:00Z",
+                to = "1970-01-01T00:01:00Z"
+            )
+        """
+        success, error = self.check_listing_size(*self.execute_query(query))
         assert success, error
 
     @link_test_case("#23190")
@@ -71,7 +94,6 @@ def test_listing_paging_monitoring(self):
         result, error = self.execute_query(data_source_query)
         assert error is None
 
-        # simplest query with default downsampling settings
         query = """
             SELECT test_label FROM local_monitoring.my_service WITH (
                 selectors = @@{test_type="listing_paging_test", test_label="*"}@@,
@@ -82,5 +104,16 @@ def test_listing_paging_monitoring(self):
                 to = "1970-01-01T00:01:00Z"
             )
         """
-        success, error = self.check_listing_paging_result(*self.execute_query(query))
+        success, error = self.check_full_listing_result(*self.execute_query(query))
+        assert success, error
+
+        query = """
+            SELECT * FROM local_monitoring.my_service WITH (
+                selectors = @@{test_type="listing_paging_test"}@@,
+
+                from = "1970-01-01T00:00:00Z",
+                to = "1970-01-01T00:01:00Z"
+            )
+        """
+        success, error = self.check_listing_size(*self.execute_query(query))
         assert success, error