Merge pull request #232 from den818/OPENTELEMETRY

support open telemetry

Author: Enmk

clickhouse/CMakeLists.txt

@@ -92,12 +92,14 @@ INSTALL(FILES query.h DESTINATION include/clickhouse/)
 INSTALL(FILES base/buffer.h DESTINATION include/clickhouse/base/)
 INSTALL(FILES base/compressed.h DESTINATION include/clickhouse/base/)
 INSTALL(FILES base/input.h DESTINATION include/clickhouse/base/)
+INSTALL(FILES base/open_telemetry.h DESTINATION include/clickhouse/base/)
 INSTALL(FILES base/output.h DESTINATION include/clickhouse/base/)
 INSTALL(FILES base/platform.h DESTINATION include/clickhouse/base/)
 INSTALL(FILES base/singleton.h DESTINATION include/clickhouse/base/)
 INSTALL(FILES base/socket.h DESTINATION include/clickhouse/base/)
 INSTALL(FILES base/string_utils.h DESTINATION include/clickhouse/base/)
 INSTALL(FILES base/string_view.h DESTINATION include/clickhouse/base/)
+INSTALL(FILES base/uuid.h DESTINATION include/clickhouse/base/)
 INSTALL(FILES base/wire_format.h DESTINATION include/clickhouse/base/)
 
 # columns

clickhouse/base/open_telemetry.h

@@ -0,0 +1,23 @@
+#pragma once
+
+#include "uuid.h"
+
+#include <string>
+
+namespace clickhouse::open_telemetry {
+
+/// See https://www.w3.org/TR/trace-context/ for trace_flags definition
+enum TraceFlags : uint8_t {
+    TRACE_FLAG_NONE    = 0,
+    TRACE_FLAG_SAMPLED = 1,
+};
+
+/// The runtime info we need to create new OpenTelemetry spans.
+struct TracingContext {
+    UUID trace_id{};
+    uint64_t span_id = 0;
+    std::string tracestate;
+    uint8_t trace_flags = TRACE_FLAG_NONE;
+};
+
+}  // namespace clickhouse::open_telemetry

clickhouse/base/uuid.h

@@ -0,0 +1,12 @@
+#pragma once
+
+#include <cstdint>
+#include <utility>
+
+namespace clickhouse {
+
+using UInt128 = std::pair<uint64_t, uint64_t>;
+
+using UUID = UInt128;
+
+}

clickhouse/client.cpp

@@ -37,8 +37,10 @@
 #define DBMS_MIN_REVISION_WITH_COLUMN_DEFAULTS_METADATA 54410
 #define DBMS_MIN_REVISION_WITH_CLIENT_WRITE_INFO        54420
 #define DBMS_MIN_REVISION_WITH_SETTINGS_SERIALIZED_AS_STRINGS 54429
+#define DBMS_MIN_REVISION_WITH_INTERSERVER_SECRET       54441
+#define DBMS_MIN_REVISION_WITH_OPENTELEMETRY            54442
 
-#define REVISION  DBMS_MIN_REVISION_WITH_SETTINGS_SERIALIZED_AS_STRINGS
+#define REVISION  DBMS_MIN_REVISION_WITH_OPENTELEMETRY
 
 namespace clickhouse {
 
@@ -661,6 +663,27 @@ void Client::Impl::SendQuery(const Query& query) {
         if (server_info_.revision >= DBMS_MIN_REVISION_WITH_VERSION_PATCH) {
             WireFormat::WriteUInt64(*output_, info.client_version_patch);
         }
+
+        if (server_info_.revision >= DBMS_MIN_REVISION_WITH_OPENTELEMETRY) {
+            if (const auto& tracing_context = query.GetTracingContext()) {
+                // Have OpenTelemetry header.
+                WireFormat::WriteFixed(*output_, uint8_t(1));
+                // No point writing these numbers with variable length, because they
+                // are random and will probably require the full length anyway.
+                WireFormat::WriteFixed(*output_, tracing_context->trace_id);
+                WireFormat::WriteFixed(*output_, tracing_context->span_id);
+                WireFormat::WriteString(*output_, tracing_context->tracestate);
+                WireFormat::WriteFixed(*output_, tracing_context->trace_flags);
+            } else {
+                // Don't have OpenTelemetry header.
+                WireFormat::WriteFixed(*output_, uint8_t(0));
+            }
+        } else {
+            if (query.GetTracingContext()) {
+                // Current implementation works only for server version >= v20.11.2.1-stable
+                throw UnimplementedError(std::string("Can't send open telemetry tracing context to a server, server version is too old"));
+            }
+        }
     }
 
     /// Per query settings
@@ -678,6 +701,10 @@ void Client::Impl::SendQuery(const Query& query) {
     // Empty string signals end of serialized settings
     WireFormat::WriteString(*output_, std::string());
 
+    if (server_info_.revision >= DBMS_MIN_REVISION_WITH_INTERSERVER_SECRET) {
+        WireFormat::WriteString(*output_, "");
+    }
+
     WireFormat::WriteUInt64(*output_, Stages::Complete);
     WireFormat::WriteUInt64(*output_, compression_);
     WireFormat::WriteString(*output_, query.GetText());

clickhouse/columns/uuid.cpp

@@ -21,7 +21,7 @@ ColumnUUID::ColumnUUID(ColumnRef data)
     }
 }
 
-void ColumnUUID::Append(const UInt128& value) {
+void ColumnUUID::Append(const UUID& value) {
     data_->Append(value.first);
     data_->Append(value.second);
 }
@@ -30,12 +30,12 @@ void ColumnUUID::Clear() {
     data_->Clear();
 }
 
-const UInt128 ColumnUUID::At(size_t n) const {
-    return UInt128(data_->At(n * 2), data_->At(n * 2 + 1));
+const UUID ColumnUUID::At(size_t n) const {
+    return UUID(data_->At(n * 2), data_->At(n * 2 + 1));
 }
 
-const UInt128 ColumnUUID::operator [] (size_t n) const {
-    return UInt128((*data_)[n * 2], (*data_)[n * 2 + 1]);
+const UUID ColumnUUID::operator [] (size_t n) const {
+    return UUID((*data_)[n * 2], (*data_)[n * 2 + 1]);
 }
 
 void ColumnUUID::Append(ColumnRef column) {
@@ -78,4 +78,3 @@ ItemView ColumnUUID::GetItem(size_t index) const {
 }
 
 }
-

clickhouse/columns/uuid.h

@@ -1,11 +1,11 @@
 #pragma once
 
+#include "../base/uuid.h"
 #include "column.h"
 #include "numeric.h"
 
 namespace clickhouse {
 
-using UInt128 = std::pair<uint64_t, uint64_t>;
 
 /**
  * Represents a UUID column.
@@ -17,13 +17,13 @@ class ColumnUUID : public Column {
     explicit ColumnUUID(ColumnRef data);
 
     /// Appends one element to the end of column.
-    void Append(const UInt128& value);
+    void Append(const UUID& value);
 
     /// Returns element at given row number.
-    const UInt128 At(size_t n) const;
+    const UUID At(size_t n) const;
 
     /// Returns element at given row number.
-    const UInt128 operator [] (size_t n) const;
+    const UUID operator [] (size_t n) const;
 
 public:
     /// Appends content of given column to the end of current one.
@@ -34,7 +34,7 @@ class ColumnUUID : public Column {
 
     /// Saves column data to output stream.
     void SaveBody(OutputStream* output) override;
-    
+
     /// Clear column data .
     void Clear() override;
 

clickhouse/query.h

@@ -3,9 +3,12 @@
 #include "block.h"
 #include "server_exception.h"
 
+#include "base/open_telemetry.h"
+
 #include <cstdint>
 #include <functional>
 #include <memory>
+#include <optional>
 #include <string>
 #include <unordered_map>
 
@@ -106,6 +109,16 @@ class Query : public QueryEvents {
         return *this;
     }
 
+    inline const std::optional<open_telemetry::TracingContext>& GetTracingContext() const {
+        return tracing_context_;
+    }
+
+    /// Set tracing context for open telemetry signals
+    inline Query& SetTracingContext(open_telemetry::TracingContext tracing_context) {
+        tracing_context_ = std::move(tracing_context);
+        return *this;
+    }
+
     /// Set handler for receiving result data.
     inline Query& OnData(SelectCallback cb) {
         select_cb_ = std::move(cb);
@@ -180,6 +193,7 @@ class Query : public QueryEvents {
 private:
     const std::string query_;
     const std::string query_id_;
+    std::optional<open_telemetry::TracingContext> tracing_context_;
     QuerySettings query_settings_;
     ExceptionCallback exception_cb_;
     ProgressCallback progress_cb_;

clickhouse/types/type_parser.cpp

@@ -1,5 +1,4 @@
 #include "type_parser.h"
-#include "../base/string_utils.h"
 
 #include <algorithm>
 #include <map>

ut/client_ut.cpp

@@ -43,6 +43,30 @@ class ClientCase : public testing::TestWithParam<ClientOptions> {
         return "SELECT " + column_name + " FROM " + table_name;
     }
 
+    void FlushLogs() {
+        try {
+            client_->Execute("SYSTEM FLUSH LOGS");
+        } catch (const std::exception & e) {
+            std::cerr << "Got error while flushing logs: " << e.what() << std::endl;
+            const auto wait_for_flush = []() {
+                // Insufficient privileges, the only safe way is to wait long enough for system
+                // to flush the logs automaticaly. Usually it takes 7.5 seconds, so just in case,
+                // wait 3 times that to ensure that all previously executed queries are in the logs now.
+                const auto wait_duration = std::chrono::seconds(23);
+                std::cerr << "Now we wait " << wait_duration << "..." << std::endl;
+                std::this_thread::sleep_for(wait_duration);
+            };
+            // DB::Exception: clickhouse_cpp_cicd: Not enough privileges. To execute this query it's necessary to have grant SYSTEM FLUSH LOGS ON
+            if (std::string(e.what()).find("To execute this query it's necessary to have grant SYSTEM FLUSH LOGS ON") != std::string::npos) {
+                wait_for_flush();
+            }
+            // DB::Exception: clickhouse_cpp_cicd: Cannot execute query in readonly mode
+            if (std::string(e.what()).find("Cannot execute query in readonly mode") != std::string::npos) {
+                wait_for_flush();
+            }
+        }
+    }
+
     std::unique_ptr<Client> client_;
     const std::string table_name = "test_clickhouse_cpp_test_ut_table";
     const std::string column_name = "test_column";
@@ -850,19 +874,7 @@ TEST_P(ClientCase, Query_ID) {
     client_->SelectCancelable("SELECT 'b', count(*) FROM " + table_name, query_id, [](const Block &) { return true; });
     client_->Execute(Query("TRUNCATE TABLE " + table_name, query_id));
 
-    try {
-        client_->Execute("SYSTEM FLUSH LOGS");
-    } catch (const std::exception & e) {
-        // DB::Exception: clickhouse_cpp_cicd: Not enough privileges. To execute this query it's necessary to have grant SYSTEM FLUSH LOGS ON
-        if (std::string(e.what()).find("To execute this query it's necessary to have grant SYSTEM FLUSH LOGS ON") != std::string::npos) {
-            // Insufficient privileges, the only safe way is to wait long enough for system
-            // to flush the logs automatically. Usually it takes 7.5 seconds, so just in case,
-            // wait 3 times that to ensure that all previously executed queries are in the logs now.
-            const auto wait_duration = std::chrono::seconds(23);
-            std::cerr << "Got error while flushing logs, now we wait " << wait_duration << "..." << std::endl;
-            std::this_thread::sleep_for(wait_duration);
-        }
-    }
+    FlushLogs();
 
     size_t total_count = 0;
     client_->Select("SELECT type, query_kind, query_id, query "
@@ -1115,6 +1127,31 @@ TEST_P(ClientCase, ServerLogs) {
 }
 
 
+TEST_P(ClientCase, TracingContext) {
+    Block block;
+    createTableWithOneColumn<ColumnString>(block);
+
+    Query query("INSERT INTO " + table_name + " (*) VALUES (\'Foo\'), (\'Bar\')" );
+    open_telemetry::TracingContext tracing_context;
+    std::srand(std::time(0));
+    tracing_context.trace_id = {std::rand(), std::rand()};
+    query.SetTracingContext(tracing_context);
+    client_->Execute(query);
+
+    FlushLogs();
+
+    size_t received_rows = 0;
+    client_->Select("SELECT trace_id, toString(trace_id), operation_name "
+                   "FROM system.opentelemetry_span_log "
+                   "WHERE trace_id = toUUID(\'" + ToString(tracing_context.trace_id) + "\');",
+        [&](const Block& block) {
+            // std::cerr << PrettyPrintBlock{block} << std::endl;
+            received_rows += block.GetRowCount();
+    });
+
+    EXPECT_GT(received_rows, 0u);
+}
+
 const auto LocalHostEndpoint = ClientOptions()
         .SetHost(           getEnvOrDefault("CLICKHOUSE_HOST",     "localhost"))
         .SetPort(   getEnvOrDefault<size_t>("CLICKHOUSE_PORT",     "9000"))

ut/columns_ut.cpp

@@ -377,17 +377,17 @@ TEST(ColumnsCase, UUIDInit) {
     auto col = std::make_shared<ColumnUUID>(std::make_shared<ColumnUInt64>(MakeUUID_data()));
 
     ASSERT_EQ(col->Size(), 3u);
-    ASSERT_EQ(col->At(0), UInt128(0xbb6a8c699ab2414cllu, 0x86697b7fd27f0825llu));
-    ASSERT_EQ(col->At(2), UInt128(0x3507213c178649f9llu, 0x9faf035d662f60aellu));
+    ASSERT_EQ(col->At(0), UUID(0xbb6a8c699ab2414cllu, 0x86697b7fd27f0825llu));
+    ASSERT_EQ(col->At(2), UUID(0x3507213c178649f9llu, 0x9faf035d662f60aellu));
 }
 
 TEST(ColumnsCase, UUIDSlice) {
     auto col = std::make_shared<ColumnUUID>(std::make_shared<ColumnUInt64>(MakeUUID_data()));
     auto sub = col->Slice(1, 2)->As<ColumnUUID>();
 
     ASSERT_EQ(sub->Size(), 2u);
-    ASSERT_EQ(sub->At(0), UInt128(0x84b9f24bc26b49c6llu, 0xa03b4ab723341951llu));
-    ASSERT_EQ(sub->At(1), UInt128(0x3507213c178649f9llu, 0x9faf035d662f60aellu));
+    ASSERT_EQ(sub->At(0), UUID(0x84b9f24bc26b49c6llu, 0xa03b4ab723341951llu));
+    ASSERT_EQ(sub->At(1), UUID(0x3507213c178649f9llu, 0x9faf035d662f60aellu));
 }
 
 TEST(ColumnsCase, Int128) {
@@ -787,4 +787,3 @@ TEST(ColumnsCase, ColumnLowCardinalityString_WithEmptyString_3) {
         EXPECT_EQ(values[i], col.At(i)) << " at pos: " << i;
     }
 }
-