change metric

Author: Nathan903

components/core/CMakeLists.txt

@@ -22,7 +22,7 @@ endif()
 
 if (CMAKE_BUILD_TYPE MATCHES "Release")
     include(CheckIPOSupported)
-    check_ipo_supported(RESULT IPO_SUPPORTED)
+    check_ipo_supported(RESULT IPO_SUPPORTED OUTPUT)
     if(IPO_SUPPORTED)
         message(STATUS "Link-time optimization enabled.")
         set(CMAKE_INTERPROCEDURAL_OPTIMIZATION ON)

components/core/src/clp/clo/clo.cpp

@@ -82,8 +82,7 @@ static SearchFilesResult search_file(
         Query& query,
         Archive& archive,
         MetadataDB::FileIterator& file_metadata_ix,
-        std::unique_ptr<OutputHandler>& output_handler,
-        size_t& total_bytes_output
+        std::unique_ptr<OutputHandler>& output_handler
 );
 /**
  * Searches all files referenced by a given database cursor
@@ -98,9 +97,7 @@ static void search_files(
         Archive& archive,
         MetadataDB::FileIterator& file_metadata_ix,
         std::unique_ptr<OutputHandler>& output_handler,
-        std::set<clp::segment_id_t> const& segments_to_search,
-        size_t& total_bytes_scanned,
-        size_t& total_bytes_output
+        std::set<clp::segment_id_t> const& segments_to_search
 );
 /**
  * Searches an archive with the given path
@@ -390,8 +387,7 @@ static SearchFilesResult search_file(
         Query& query,
         Archive& archive,
         MetadataDB::FileIterator& file_metadata_ix,
-        std::unique_ptr<OutputHandler>& output_handler,
-        size_t& total_bytes_output
+        std::unique_ptr<OutputHandler>& output_handler
 ) {
     File compressed_file;
     Message encoded_message;
@@ -419,7 +415,6 @@ static SearchFilesResult search_file(
             decompressed_message
     ))
     {
-        total_bytes_output += decompressed_message.length();
         if (ErrorCode_Success
             != output_handler->add_result(
                     compressed_file.get_orig_path(),
@@ -442,9 +437,7 @@ void search_files(
         Archive& archive,
         MetadataDB::FileIterator& file_metadata_ix,
         std::unique_ptr<OutputHandler>& output_handler,
-        std::set<clp::segment_id_t> const& segments_to_search,
-        size_t& total_bytes_scanned,
-        size_t& total_bytes_output
+        std::set<clp::segment_id_t> const& segments_to_search
 ) {
     if (query.contains_sub_queries()) {
         for (; file_metadata_ix.has_next(); file_metadata_ix.next()) {
@@ -456,15 +449,7 @@ void search_files(
                 continue;
             }
 
-            total_bytes_scanned += file_metadata_ix.get_num_uncompressed_bytes();
-
-            auto result = search_file(
-                    query,
-                    archive,
-                    file_metadata_ix,
-                    output_handler,
-                    total_bytes_output
-            );
+            auto result = search_file(query, archive, file_metadata_ix, output_handler);
             if (SearchFilesResult::OpenFailure == result) {
                 continue;
             }
@@ -478,15 +463,7 @@ void search_files(
                 continue;
             }
 
-            total_bytes_scanned += file_metadata_ix.get_num_uncompressed_bytes();
-
-            auto result = search_file(
-                    query,
-                    archive,
-                    file_metadata_ix,
-                    output_handler,
-                    total_bytes_output
-            );
+            auto result = search_file(query, archive, file_metadata_ix, output_handler);
             if (SearchFilesResult::OpenFailure == result) {
                 continue;
             }
@@ -572,16 +549,12 @@ static bool search_archive(
             true
     );
     auto& file_metadata_ix = *file_metadata_ix_ptr;
-    size_t total_bytes_scanned = 0;
-    size_t total_bytes_output = 0;
     search_files(
             query,
             archive_reader,
             file_metadata_ix,
             output_handler,
-            ids_of_segments_to_search,
-            total_bytes_scanned,
-            total_bytes_output
+            ids_of_segments_to_search
     );
     file_metadata_ix_ptr.reset(nullptr);
 
@@ -595,12 +568,6 @@ static bool search_archive(
         );
         return false;
     }
-
-    nlohmann::json json_msg;
-    json_msg["stats"]["bytes_scanned"] = total_bytes_scanned;
-    json_msg["stats"]["bytes_output"] = total_bytes_output;
-    std::cout << json_msg.dump(-1, ' ', true, nlohmann::json::error_handler_t::ignore) << std::endl;
-
     return true;
 }
 

components/core/src/clp_s/search/Output.cpp

@@ -1,10 +1,8 @@
 #include "Output.hpp"
 
-#include <iostream>
 #include <memory>
 #include <vector>
 
-#include <nlohmann/json.hpp>
 #include <spdlog/spdlog.h>
 
 #include "../../clp/type_utils.hpp"
@@ -91,7 +89,6 @@ bool Output::filter() {
     m_archive_reader->open_packed_streams();
 
     std::string message;
-    size_t total_bytes_output = 0;
     auto const archive_id = m_archive_reader->get_archive_id();
     for (int32_t schema_id : matched_schemas) {
         if (EvaluatedValue::False == m_query_runner.schema_init(schema_id)) {
@@ -115,12 +112,10 @@ bool Output::filter() {
                     m_query_runner
             ))
             {
-                total_bytes_output += message.length();
                 m_output_handler->write(message, timestamp, archive_id, log_event_idx);
             }
         } else {
             while (reader.get_next_message(message, m_query_runner)) {
-                total_bytes_output += message.length();
                 m_output_handler->write(message);
             }
         }
@@ -141,12 +136,6 @@ bool Output::filter() {
         );
         return false;
     }
-
-    nlohmann::json json_msg;
-    json_msg["stats"] = nlohmann::json::object();
-    json_msg["stats"]["bytes_output"] = total_bytes_output;
-    std::cout << json_msg.dump(-1, ' ', true, nlohmann::json::error_handler_t::ignore) << std::endl;
-
     return true;
 }
 }  // namespace clp_s::search

components/job-orchestration/job_orchestration/executor/query/celery.py

@@ -1,22 +1,9 @@
 from celery import Celery
-from celery.signals import worker_process_init, worker_process_shutdown
-from clp_py_utils.telemetry import init_telemetry, shutdown_telemetry
 
 from job_orchestration.executor.query import celeryconfig
 
 app = Celery("query")
 app.config_from_object(celeryconfig)
 
-
-@worker_process_init.connect
-def setup_telemetry(**kwargs) -> None:
-    init_telemetry()
-
-
-@worker_process_shutdown.connect
-def teardown_telemetry(**kwargs) -> None:
-    shutdown_telemetry()
-
-
 if "__main__" == __name__:
     app.start()

components/job-orchestration/job_orchestration/executor/query/fs_search_task.py

@@ -1,5 +1,4 @@
 import datetime
-import json
 import os
 from pathlib import Path
 from typing import Any
@@ -8,19 +7,6 @@
 from celery.app.task import Task
 from celery.exceptions import SoftTimeLimitExceeded
 from celery.utils.log import get_task_logger
-from opentelemetry import metrics
-
-meter = metrics.get_meter("clp_py_utils")
-bytes_scanned_counter = meter.create_counter(
-    "clp.query.bytes_scanned_total",
-    unit="By",
-    description="Total uncompressed bytes scanned by the query",
-)
-bytes_output_counter = meter.create_counter(
-    "clp.query.bytes_output_total", unit="By", description="Total bytes output by the query"
-)
-
-
 from clp_py_utils.clp_config import (
     Database,
     StorageEngine,
@@ -275,7 +261,7 @@ def search_entry_point(
             start_time=start_time,
         )
 
-    task_results, stdout_data = run_query_task(
+    task_results, _ = run_query_task(
         sql_adapter=sql_adapter,
         logger=logger,
         clp_logs_dir=clp_logs_dir,
@@ -298,48 +284,6 @@ def search_entry_point(
         dest_path = f"{job_id}/{archive_id}"
         upload_results_to_s3(task_results, s3_config, src_file, dest_path)
 
-    # Telemetry
-    bytes_scanned = None
-    bytes_output = None
-
-    if stdout_data:
-        for line in reversed(stdout_data.rsplit("\n", 10)):
-            if '"stats"' not in line:
-                continue
-            try:
-                data = json.loads(line)
-                if isinstance(data, dict) and "stats" in data:
-                    stats = data["stats"]
-                    bytes_scanned = stats.get("bytes_scanned", None)
-                    bytes_output = stats.get("bytes_output", None)
-                    break
-            except json.JSONDecodeError:
-                pass
-
-    status_str = "success" if QueryTaskStatus.SUCCEEDED == task_results.status else "failure"
-
-    storage_engine_str = (
-        "clp_s" if worker_config.package.storage_engine == StorageEngine.CLP_S else "clp"
-    )
-    if search_config.aggregation_config is not None:
-        output_type = "reducer"
-    elif search_config.network_address is not None:
-        output_type = "network"
-    elif search_config.write_to_file:
-        output_type = "file"
-    else:
-        output_type = "results_cache"
-
-    attributes = {
-        "status": status_str,
-        "clp.storage.engine": storage_engine_str,
-        "clp.query.output_type": output_type,
-    }
-    if bytes_scanned is not None:
-        bytes_scanned_counter.add(bytes_scanned, attributes)
-    if bytes_output is not None:
-        bytes_output_counter.add(bytes_output, attributes)
-
     return task_results.model_dump()
 
 

components/job-orchestration/job_orchestration/scheduler/query/query_scheduler.py

@@ -44,6 +44,9 @@
 from clp_py_utils.core import read_yaml_config_file
 from clp_py_utils.decorators import exception_default_value
 from clp_py_utils.sql_adapter import ConnectionPoolWrapper, SqlAdapter
+from clp_py_utils.telemetry import init_telemetry, shutdown_telemetry
+from opentelemetry import metrics
+from opentelemetry.metrics import Observation
 from pydantic import ValidationError
 
 from job_orchestration.executor.query.celery import app
@@ -94,6 +97,37 @@
 
 reducer_connection_queue: asyncio.Queue | None = None
 
+# OpenTelemetry metrics
+meter = metrics.get_meter("clp_py_utils")
+
+
+def _observe_active_jobs(options) -> list[Observation]:
+    return [Observation(len(active_jobs), {})]
+
+
+def _observe_outstanding_tasks(options) -> list[Observation]:
+    total = 0
+    for job in active_jobs.values():
+        if isinstance(job, SearchJob):
+            total += job.num_archives_to_search - job.num_archives_searched
+        else:
+            total += 1
+    return [Observation(total, {})]
+
+
+active_jobs_gauge = meter.create_observable_gauge(
+    "clp.query.active_jobs",
+    callbacks=[_observe_active_jobs],
+    unit="1",
+    description="Number of active query jobs",
+)
+outstanding_tasks_gauge = meter.create_observable_gauge(
+    "clp.query.outstanding_tasks",
+    callbacks=[_observe_outstanding_tasks],
+    unit="1",
+    description="Total number of outstanding tasks across all active query jobs",
+)
+
 
 class DispatchExecutor:
     # Globals for dispatch executor pool
@@ -1137,6 +1171,8 @@ async def handle_jobs(
 async def main(argv: list[str]) -> int:
     global reducer_connection_queue
 
+    init_telemetry()
+
     args_parser = argparse.ArgumentParser(description="Wait for and run query jobs.")
     args_parser.add_argument("--config", "-c", required=True, help="CLP configuration file.")
 
@@ -1229,6 +1265,7 @@ async def main(argv: list[str]) -> int:
     except Exception:
         logger.exception("Uncaught exception in job handling loop.")
 
+    shutdown_telemetry()
     return 0
 
 

integration-tests/tests/binary_tests/test_log_converter.py

@@ -89,8 +89,6 @@ def _convert_and_compress(
     search_result = search_action.verify_returncode()
     assert search_result, search_result.failure_message
     lines = search_action.completed_proc.stdout.splitlines()
-    # clp-s now outputs a stats object at the end of its stdout for telemetry
-    lines = [line for line in lines if not line.strip().startswith('{"stats":')]
     if len(lines) != test_paths.num_log_events:
         pytest.fail(
             f"Expected {test_paths.num_log_events} log events after conversion, "

integration-tests/tests/utils/fs_validation.py

@@ -1,6 +1,5 @@
 """File structure validators."""
 
-import json
 from pathlib import Path
 from tempfile import NamedTemporaryFile
 from typing import IO
@@ -41,20 +40,15 @@ def _sort_json_keys_and_rows(json_fp: Path) -> IO[str]:
 
     :param json_fp:
     :return: A named temporary file (delete on close) that contains the sorted JSON content.
+    :raise: RuntimeError if jq is missing or fails due to execution errors.
     """
-    lines = []
-    with json_fp.open("r") as f:
-        for line in f:
-            if not line.strip():
-                continue
-            try:
-                obj = json.loads(line)
-                lines.append(json.dumps(obj, separators=(",", ":"), sort_keys=True))
-            except json.JSONDecodeError:
-                lines.append(line.strip())
+    jq_action = NonClpAction(
+        cmd=[get_binary_path("jq"), "--sort-keys", "--compact-output", ".", str(json_fp)],
+    )
+    jq_action.check_returncode()
 
     sorted_fp = NamedTemporaryFile(mode="w+")  # noqa: SIM115
-    sorted_lines = sorted(lines)
+    sorted_lines = sorted(jq_action.completed_proc.stdout.splitlines())
     for line in sorted_lines:
         sorted_fp.write(f"{line}\n")
     sorted_fp.flush()