support open telemetry

den818 · den818 · commit 56408ad8a385 · 2022-10-25T12:50:17.000+04:00
diff --git a/clickhouse/CMakeLists.txt b/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
diff --git a/clickhouse/base/open_telemetry.h b/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
diff --git a/clickhouse/base/string_utils.h b/clickhouse/base/string_utils.h
@@ -23,4 +23,4 @@ inline T FromString(const StringView& s) {
    return result;
 }
 
-}
+}  // namespace clickhouse
diff --git a/clickhouse/base/uuid.h b/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;
+
+}
diff --git a/clickhouse/client.cpp b/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());
diff --git a/clickhouse/columns/uuid.cpp b/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 {
 }
 
 }
-
diff --git a/clickhouse/columns/uuid.h b/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;
 
diff --git a/clickhouse/query.h b/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_;
diff --git a/clickhouse/types/type_parser.cpp b/clickhouse/types/type_parser.cpp
@@ -1,5 +1,4 @@
 #include "type_parser.h"
-#include "../base/string_utils.h"
 
 #include <algorithm>
 #include <map>
diff --git a/ut/client_ut.cpp b/ut/client_ut.cpp
@@ -1,3 +1,4 @@
+#include <clickhouse/base/string_utils.h>
 #include <clickhouse/client.h>
 
 #include "readonly_client_test.h"
@@ -43,6 +44,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. Usualy 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 +875,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 automaticaly. Usualy 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 +1128,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"))
diff --git a/ut/columns_ut.cpp b/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;
     }
 }
-
diff --git a/ut/utils.cpp b/ut/utils.cpp
@@ -1,5 +1,6 @@
 #include "utils.h"
 
+#include <clickhouse/base/string_utils.h>
 #include <clickhouse/block.h>
 #include <clickhouse/client.h>
 #include <clickhouse/columns/column.h>
@@ -12,6 +13,7 @@
 #include <clickhouse/columns/numeric.h>
 #include <clickhouse/columns/string.h>
 #include <clickhouse/columns/tuple.h>
+#include <clickhouse/columns/uuid.h>
 
 #include <clickhouse/base/socket.h> // for ipv4-ipv6 platform-specific stuff
 
@@ -114,6 +116,15 @@ bool doPrintValue<ColumnTuple, void>(const ColumnRef & c, const size_t row, std:
     return false;
 }
 
+template <>
+bool doPrintValue<ColumnUUID, void>(const ColumnRef & c, const size_t row, std::ostream & ostr) {
+    if (const auto & uuid_col = c->As<ColumnUUID>()) {
+        ostr << ToString(uuid_col->At(row));
+        return true;
+    }
+    return false;
+}
+
 std::ostream & printColumnValue(const ColumnRef& c, const size_t row, std::ostream & ostr) {
 
     const auto r = false
@@ -138,7 +149,8 @@ std::ostream & printColumnValue(const ColumnRef& c, const size_t row, std::ostre
         || doPrintValue<ColumnIPv4>(c, row, ostr)
         || doPrintValue<ColumnIPv6>(c, row, ostr)
         || doPrintValue<ColumnArray, void>(c, row, ostr)
-        || doPrintValue<ColumnTuple, void>(c, row, ostr);
+        || doPrintValue<ColumnTuple, void>(c, row, ostr)
+        || doPrintValue<ColumnUUID, void>(c, row, ostr);
     if (!r)
         ostr << "Unable to print value of type " << c->GetType().GetName();
 
@@ -271,3 +283,14 @@ std::ostream & operator<<(std::ostream & ostr, const ServerInfo & server_info) {
 uint64_t versionNumber(const ServerInfo & server_info) {
     return versionNumber(server_info.version_major, server_info.version_minor, server_info.version_patch, server_info.revision);
 }
+
+std::string ToString(const clickhouse::UUID& v) {
+    std::string result(36, 0);
+    // ffff ff ff ss ssssss
+    const int count = std::snprintf(result.data(), result.size() + 1, "%.8" PRIx64 "-%.4" PRIx64 "-%.4" PRIx64 "-%.4" PRIx64 "-%.12" PRIx64,
+                                    v.first >> 32, (v.first >> 16) & 0xffff, v.first & 0xffff, v.second >> 48, v.second & 0xffffffffffff);
+    if (count != 36) {
+        throw std::runtime_error("Error while converting UUID to string");
+    }
+    return result;
+}
diff --git a/ut/utils.h b/ut/utils.h
diff --git a/ut/utils_ut.cpp b/ut/utils_ut.cpp
diff --git a/ut/value_generators.cpp b/ut/value_generators.cpp
diff --git a/ut/value_generators.h b/ut/value_generators.h

Original file line number	Diff line number	Diff line change
`@@ -23,4 +23,4 @@ inline T FromString(const StringView& s) {`
`23`	`23`	`return result;`
`24`	`24`	`}`
`25`	`25`
`26`		`-}`
	`26`	`+} // namespace clickhouse`
Original file line number	Diff line number	Diff line change
`@@ -21,7 +21,7 @@ ColumnUUID::ColumnUUID(ColumnRef data)`
`21`	`21`	`}`
`22`	`22`	`}`
`23`	`23`
`24`		`-void ColumnUUID::Append(const UInt128& value) {`
	`24`	`+void ColumnUUID::Append(const UUID& value) {`
`25`	`25`	`data_->Append(value.first);`
`26`	`26`	`data_->Append(value.second);`
`27`	`27`	`}`
`@@ -30,12 +30,12 @@ void ColumnUUID::Clear() {`
`30`	`30`	`data_->Clear();`
`31`	`31`	`}`
`32`	`32`
`33`		`-const UInt128 ColumnUUID::At(size_t n) const {`
`34`		`- return UInt128(data_->At(n * 2), data_->At(n * 2 + 1));`
	`33`	`+const UUID ColumnUUID::At(size_t n) const {`
	`34`	`+ return UUID(data_->At(n * 2), data_->At(n * 2 + 1));`
`35`	`35`	`}`
`36`	`36`
`37`		`-const UInt128 ColumnUUID::operator [] (size_t n) const {`
`38`		`- return UInt128((data_)[n 2], (data_)[n 2 + 1]);`
	`37`	`+const UUID ColumnUUID::operator [] (size_t n) const {`
	`38`	`+ return UUID((data_)[n 2], (data_)[n 2 + 1]);`
`39`	`39`	`}`
`40`	`40`
`41`	`41`	`void ColumnUUID::Append(ColumnRef column) {`
`@@ -78,4 +78,3 @@ ItemView ColumnUUID::GetItem(size_t index) const {`
`78`	`78`	`}`
`79`	`79`
`80`	`80`	`}`
`81`		`-`
Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,4 @@`
`1`	`1`	`#include "type_parser.h"`
`2`		`-#include "../base/string_utils.h"`
`3`	`2`
`4`	`3`	`#include <algorithm>`
`5`	`4`	`#include <map>`