go/go-clickhouse
1
0
Fork 0
mirror of https://github.com/uptrace/go-clickhouse.git synced 2025-06-14 23:44:59 +02:00
Code Issues Releases Activity

feat: enable opentelemetry support in protocol

Browse Source
This commit is contained in:
Vladimir Mihailenco 2022-05-18 16:23:57 +03:00
parent c84dc481ac
commit fb79ac4b75
8 changed files with 183 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
ch/chpool/conn.go
View File

@ -45,6 +45,10 @@ func (cn *Conn) SetUsedAt(tm time.Time) {
atomic.StoreInt64(&cn.usedAt, tm.Unix())
}
func (cn *Conn) LocalAddr() net.Addr {
return cn.netConn.LocalAddr()
}
func (cn *Conn) RemoteAddr() net.Addr {
return cn.netConn.RemoteAddr()
}

22
ch/chproto/proto.go
View File

@ -23,6 +23,10 @@ const (
ServerTablesStatus = 9
ServerLog = 10
ServerTableColumns = 11
ServerPartUUIDs = 12
ServerReadTaskRequest = 13
ServerProfileEvents = 14
ServerTreeReadTaskRequest = 15
)
const (
@ -30,3 +34,21 @@ const (
QueryInitial = 1
QuerySecondary = 2
)
// see https://github.com/ClickHouse/ClickHouse/blob/master/src/Core/Protocol.h
const (
DBMS_MIN_REVISION_WITH_CLIENT_INFO = 54032
DBMS_MIN_REVISION_WITH_SERVER_TIMEZONE = 54058
DBMS_MIN_REVISION_WITH_QUOTA_KEY_IN_CLIENT_INFO = 54060
DBMS_MIN_REVISION_WITH_SERVER_DISPLAY_NAME = 54372
DBMS_MIN_REVISION_WITH_VERSION_PATCH = 54401
DBMS_MIN_REVISION_WITH_CLIENT_WRITE_INFO = 54420
DBMS_MIN_REVISION_WITH_SETTINGS_SERIALIZED_AS_STRINGS = 54429
DBMS_MIN_REVISION_WITH_INTERSERVER_SECRET = 54441
DBMS_MIN_REVISION_WITH_OPENTELEMETRY = 54442
DBMS_MIN_PROTOCOL_VERSION_WITH_DISTRIBUTED_DEPTH = 54448
DBMS_MIN_PROTOCOL_VERSION_WITH_INITIAL_QUERY_START_TIME = 54449
DBMS_MIN_PROTOCOL_VERSION_WITH_INCREMENTAL_PROFILE_EVENTS = 54451
DBMS_MIN_REVISION_WITH_PARALLEL_REPLICAS = 54453
DBMS_TCP_PROTOCOL_VERSION = DBMS_MIN_REVISION_WITH_PARALLEL_REPLICAS
)

6
ch/chproto/server_info.go
View File

@ -21,15 +21,21 @@ func (srv *ServerInfo) ReadFrom(rd *Reader) (err error) {
return err
}
if srv.Revision >= DBMS_MIN_REVISION_WITH_SERVER_TIMEZONE {
if _, err := rd.String(); err != nil { // timezone
return err
}
}
if srv.Revision >= DBMS_MIN_REVISION_WITH_SERVER_DISPLAY_NAME {
if _, err := rd.String(); err != nil { // display name
return err
}
}
if srv.Revision >= DBMS_MIN_REVISION_WITH_VERSION_PATCH {
if _, err := rd.Uvarint(); err != nil { // server version patch
return err
}
}
return nil
}

4
ch/chproto/writer.go
View File

@ -79,7 +79,7 @@ func (w *Writer) Write(b []byte) {
w.err = err
}
func (w *Writer) writeByte(c byte) {
func (w *Writer) WriteByte(c byte) {
if w.err != nil {
return
}
@ -100,7 +100,7 @@ func (w *Writer) Uvarint(num uint64) {
}
func (w *Writer) UInt8(num uint8) {
w.writeByte(num)
w.WriteByte(num)
}
func (w *Writer) UInt16(num uint16) {

10
ch/db.go
View File

@ -253,14 +253,14 @@ func (db *DB) _exec(ctx context.Context, query string) (*result, error) {
var res *result
err := db.withConn(ctx, func(cn *chpool.Conn) error {
if err := cn.WithWriter(ctx, db.cfg.WriteTimeout, func(wr *chproto.Writer) {
db.writeQuery(wr, query)
db.writeQuery(ctx, cn, wr, query)
db.writeBlock(ctx, wr, nil)
}); err != nil {
return err
}
return cn.WithReader(ctx, db.cfg.ReadTimeout, func(rd *chproto.Reader) error {
var err error
res, err = db.readDataBlocks(rd)
res, err = db.readDataBlocks(cn, rd)
return err
})
})
@ -323,7 +323,7 @@ func (db *DB) _query(ctx context.Context, query string) (*blockIter, error) {
}
if err := cn.WithWriter(ctx, db.cfg.WriteTimeout, func(wr *chproto.Writer) {
db.writeQuery(wr, query)
db.writeQuery(ctx, cn, wr, query)
db.writeBlock(ctx, wr, nil)
}); err != nil {
return nil, err
@ -363,7 +363,7 @@ func (db *DB) _insert(
var res *result
err := db.withConn(ctx, func(cn *chpool.Conn) error {
if err := cn.WithWriter(ctx, db.cfg.WriteTimeout, func(wr *chproto.Writer) {
db.writeQuery(wr, query)
db.writeQuery(ctx, cn, wr, query)
db.writeBlock(ctx, wr, nil)
}); err != nil {
return err
@ -385,7 +385,7 @@ func (db *DB) _insert(
return cn.WithReader(ctx, db.cfg.ReadTimeout, func(rd *chproto.Reader) error {
var err error
res, err = readPacket(rd)
res, err = readPacket(cn, rd)
if err != nil {
return err
}

140
ch/proto.go
View File

@ -10,14 +10,19 @@ import (
"github.com/uptrace/go-clickhouse/ch/chpool"
"github.com/uptrace/go-clickhouse/ch/chproto"
"github.com/uptrace/go-clickhouse/ch/chschema"
"go.opentelemetry.io/otel/trace"
)
const (
clientName = "go-clickhouse"
chVersionMajor = 19
chVersionMinor = 17
chVersionPatch = 5
chRevision = 54428
chVersionMajor = 1
chVersionMinor = 1
chProtoVersion = chproto.DBMS_TCP_PROTOCOL_VERSION
)
var (
osUser = os.Getenv("USER")
hostname, _ = os.Hostname()
)
type blockIter struct {
@ -79,14 +84,14 @@ func (it *blockIter) read(ctx context.Context, block *chschema.Block) (bool, err
switch packet {
case chproto.ServerData:
if err := it.db.readBlock(rd, block); err != nil {
if err := it.db.readBlock(rd, block, true); err != nil {
return false, err
}
return true, nil
case chproto.ServerException:
return false, readException(rd)
case chproto.ServerProgress:
if err := readProgress(rd); err != nil {
if err := readProgress(it.cn, rd); err != nil {
return false, err
}
case chproto.ServerProfileInfo:
@ -97,6 +102,11 @@ func (it *blockIter) read(ctx context.Context, block *chschema.Block) (bool, err
if err := readServerTableColumns(rd); err != nil {
return false, err
}
case chproto.ServerProfileEvents:
block := new(chschema.Block)
if err := it.db.readBlock(rd, block, false); err != nil {
return false, err
}
case chproto.ServerEndOfStream:
return false, nil
default:
@ -107,7 +117,7 @@ func (it *blockIter) read(ctx context.Context, block *chschema.Block) (bool, err
func (db *DB) hello(ctx context.Context, cn *chpool.Conn) error {
err := cn.WithWriter(ctx, db.cfg.WriteTimeout, func(wr *chproto.Writer) {
wr.Uvarint(chproto.ClientHello)
wr.WriteByte(chproto.ClientHello)
writeClientInfo(wr)
wr.String(db.cfg.Database)
@ -138,7 +148,7 @@ func writeClientInfo(wr *chproto.Writer) {
wr.String(clientName)
wr.Uvarint(chVersionMajor)
wr.Uvarint(chVersionMinor)
wr.Uvarint(chRevision)
wr.Uvarint(chProtoVersion)
}
func readException(rd *chproto.Reader) (err error) {
@ -194,7 +204,7 @@ func readProfileInfo(rd *chproto.Reader) error {
return nil
}
func readProgress(rd *chproto.Reader) error {
func readProgress(cn *chpool.Conn, rd *chproto.Reader) error {
if _, err := rd.Uvarint(); err != nil {
return err
}
@ -204,17 +214,19 @@ func readProgress(rd *chproto.Reader) error {
if _, err := rd.Uvarint(); err != nil {
return err
}
if cn.ServerInfo.Revision >= chproto.DBMS_MIN_REVISION_WITH_CLIENT_WRITE_INFO {
if _, err := rd.Uvarint(); err != nil {
return err
}
if _, err := rd.Uvarint(); err != nil {
return err
}
}
return nil
}
func writePing(wr *chproto.Writer) {
wr.Uvarint(chproto.ClientPing)
wr.WriteByte(chproto.ClientPing)
}
func readPong(rd *chproto.Reader) error {
@ -237,38 +249,82 @@ func readPong(rd *chproto.Reader) error {
}
}
var hostname string
func (db *DB) writeQuery(wr *chproto.Writer, query string) {
if hostname == "" {
hostname, _ = os.Hostname()
}
wr.Uvarint(chproto.ClientQuery)
wr.String("")
func (db *DB) writeQuery(ctx context.Context, cn *chpool.Conn, wr *chproto.Writer, query string) {
wr.WriteByte(chproto.ClientQuery)
wr.String("") // query id
// TODO: use QuerySecondary - https://github.com/ClickHouse/ClickHouse/blob/master/dbms/src/Client/Connection.cpp#L388-L404
wr.Uvarint(chproto.QueryInitial)
wr.WriteByte(chproto.QueryInitial)
wr.String("") // initial user
wr.String("") // initial query id
wr.String("[::ffff:127.0.0.1]:0")
wr.Uvarint(1) // iface type TCP
wr.String(hostname)
wr.String(cn.LocalAddr().String())
if cn.ServerInfo.Revision >= chproto.DBMS_MIN_PROTOCOL_VERSION_WITH_INITIAL_QUERY_START_TIME {
wr.Int64(0) // initial_query_start_time_microseconds
}
wr.WriteByte(1) // interface [tcp - 1, http - 2]
wr.String(osUser)
wr.String(hostname)
writeClientInfo(wr)
if cn.ServerInfo.Revision >= chproto.DBMS_MIN_REVISION_WITH_QUOTA_KEY_IN_CLIENT_INFO {
wr.String("") // quota key
wr.Uvarint(chVersionPatch) // client version patch
}
if cn.ServerInfo.Revision >= chproto.DBMS_MIN_PROTOCOL_VERSION_WITH_DISTRIBUTED_DEPTH {
wr.Uvarint(0)
}
if cn.ServerInfo.Revision >= chproto.DBMS_MIN_REVISION_WITH_VERSION_PATCH {
wr.Uvarint(0) // client version patch
}
if cn.ServerInfo.Revision >= chproto.DBMS_MIN_REVISION_WITH_OPENTELEMETRY {
if spanCtx := trace.SpanContextFromContext(ctx); spanCtx.IsValid() {
wr.WriteByte(1)
{
v := spanCtx.TraceID()
fmt.Println(v.String())
wr.UUID(v[:])
}
{
v := spanCtx.SpanID()
wr.Write(reverseBytes(v[:]))
}
wr.String(spanCtx.TraceState().String())
wr.WriteByte(byte(spanCtx.TraceFlags()))
} else {
wr.WriteByte(0)
}
}
if cn.ServerInfo.Revision >= chproto.DBMS_MIN_REVISION_WITH_PARALLEL_REPLICAS {
wr.Uvarint(0) // collaborate_with_initiator
wr.Uvarint(0) // count_participating_replicas
wr.Uvarint(0) // number_of_current_replica
}
db.writeSettings(wr)
db.writeSettings(cn, wr)
wr.Uvarint(2)
if cn.ServerInfo.Revision >= chproto.DBMS_MIN_REVISION_WITH_INTERSERVER_SECRET {
wr.String("")
}
wr.Uvarint(2) // state complete
wr.Bool(db.cfg.Compression)
wr.String(query)
}
func (db *DB) writeSettings(wr *chproto.Writer) {
func reverseBytes(b []byte) []byte {
for i, j := 0, len(b)-1; i < j; i, j = i+1, j-1 {
b[i], b[j] = b[j], b[i]
}
return b
}
func (db *DB) writeSettings(cn *chpool.Conn, wr *chproto.Writer) {
for key, value := range db.cfg.QuerySettings {
wr.String(key)
if cn.ServerInfo.Revision > chproto.DBMS_MIN_REVISION_WITH_SETTINGS_SERIALIZED_AS_STRINGS {
wr.Bool(true) // is_important
wr.String(fmt.Sprint(value))
continue
}
switch value := value.(type) {
case string:
wr.String(value)
@ -283,9 +339,10 @@ func (db *DB) writeSettings(wr *chproto.Writer) {
default:
panic(fmt.Errorf("%s setting has unsupported type: %T", key, value))
}
}
wr.String("")
wr.String("") // end of settings
}
var emptyBlock chschema.Block
@ -294,7 +351,7 @@ func (db *DB) writeBlock(ctx context.Context, wr *chproto.Writer, block *chschem
if block == nil {
block = &emptyBlock
}
wr.Uvarint(chproto.ClientData)
wr.WriteByte(chproto.ClientData)
wr.String("")
wr.WithCompression(db.cfg.Compression, func() error {
@ -323,7 +380,7 @@ func (db *DB) readSampleBlock(rd *chproto.Reader) (*chschema.Block, error) {
switch packet {
case chproto.ServerData:
block := new(chschema.Block)
if err := db.readBlock(rd, block); err != nil {
if err := db.readBlock(rd, block, true); err != nil {
return nil, err
}
return block, nil
@ -339,7 +396,7 @@ func (db *DB) readSampleBlock(rd *chproto.Reader) (*chschema.Block, error) {
}
}
func (db *DB) readDataBlocks(rd *chproto.Reader) (*result, error) {
func (db *DB) readDataBlocks(cn *chpool.Conn, rd *chproto.Reader) (*result, error) {
var res *result
block := new(chschema.Block)
for {
@ -349,8 +406,8 @@ func (db *DB) readDataBlocks(rd *chproto.Reader) (*result, error) {
}
switch packet {
case chproto.ServerData:
if err := db.readBlock(rd, block); err != nil {
case chproto.ServerData, chproto.ServerTotals, chproto.ServerExtremes:
if err := db.readBlock(rd, block, true); err != nil {
return nil, err
}
@ -361,7 +418,7 @@ func (db *DB) readDataBlocks(rd *chproto.Reader) (*result, error) {
case chproto.ServerException:
return nil, readException(rd)
case chproto.ServerProgress:
if err := readProgress(rd); err != nil {
if err := readProgress(cn, rd); err != nil {
return nil, err
}
case chproto.ServerProfileInfo:
@ -372,6 +429,11 @@ func (db *DB) readDataBlocks(rd *chproto.Reader) (*result, error) {
if err := readServerTableColumns(rd); err != nil {
return nil, err
}
case chproto.ServerProfileEvents:
block := new(chschema.Block)
if err := db.readBlock(rd, block, false); err != nil {
return nil, err
}
case chproto.ServerEndOfStream:
return res, nil
default:
@ -380,7 +442,7 @@ func (db *DB) readDataBlocks(rd *chproto.Reader) (*result, error) {
}
}
func readPacket(rd *chproto.Reader) (*result, error) {
func readPacket(cn *chpool.Conn, rd *chproto.Reader) (*result, error) {
packet, err := rd.Uvarint()
if err != nil {
return nil, err
@ -391,7 +453,7 @@ func readPacket(rd *chproto.Reader) (*result, error) {
case chproto.ServerException:
return nil, readException(rd)
case chproto.ServerProgress:
if err := readProgress(rd); err != nil {
if err := readProgress(cn, rd); err != nil {
return nil, err
}
return res, nil
@ -412,12 +474,12 @@ func readPacket(rd *chproto.Reader) (*result, error) {
}
}
func (db *DB) readBlock(rd *chproto.Reader, block *chschema.Block) error {
func (db *DB) readBlock(rd *chproto.Reader, block *chschema.Block, compressible bool) error {
if _, err := rd.String(); err != nil {
return err
}
return rd.WithCompression(db.cfg.Compression, func() error {
return rd.WithCompression(compressible && db.cfg.Compression, func() error {
if err := readBlockInfo(rd); err != nil {
return err
}
@ -484,7 +546,7 @@ func readBlockInfo(rd *chproto.Reader) error {
}
func writeCancel(wr *chproto.Writer) {
wr.Uvarint(chproto.ClientCancel)
wr.WriteByte(chproto.ClientCancel)
}
func readServerTableColumns(rd *chproto.Reader) error {

5
go.mod
View File

@ -11,16 +11,21 @@ require (
github.com/pierrec/lz4/v4 v4.1.14
github.com/stretchr/testify v1.7.1
github.com/uptrace/go-clickhouse/chdebug v0.2.7
github.com/uptrace/go-clickhouse/chotel v0.2.7
go.opentelemetry.io/otel/trace v1.7.0
golang.org/x/exp v0.0.0-20220428152302-39d4317da171
)
require (
github.com/davecgh/go-spew v1.1.1 // indirect
github.com/fatih/color v1.13.0 // indirect
github.com/go-logr/logr v1.2.3 // indirect
github.com/go-logr/stdr v1.2.2 // indirect
github.com/kr/pretty v0.1.0 // indirect
github.com/mattn/go-colorable v0.1.12 // indirect
github.com/mattn/go-isatty v0.0.14 // indirect
github.com/pmezard/go-difflib v1.0.0 // indirect
go.opentelemetry.io/otel v1.7.0 // indirect
golang.org/x/sys v0.0.0-20220429233432-b5fbb4746d32 // indirect
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 // indirect
gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect

14
go.sum
View File

@ -7,6 +7,13 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/fatih/color v1.13.0 h1:8LOYc1KYPPmyKMuN8QV2DNRWNbLo6LZ0iLs8+mlH53w=
github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk=
github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
github.com/go-logr/logr v1.2.3 h1:2DntVwHkVopvECVRSlL5PSo9eG+cAkDCuckLubN+rq0=
github.com/go-logr/logr v1.2.3/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
github.com/google/go-cmp v0.5.7 h1:81/ik6ipDQS2aGcBfIN5dHDB36BwrStyeAQquSYCV4o=
github.com/google/go-cmp v0.5.7/go.mod h1:n+brtR0CgQNWTVd5ZUFpTBC8YFBDLK/h/bpaJ8/DtOE=
github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E=
github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc=
github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
@ -27,6 +34,12 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.7.1 h1:5TQK59W5E3v0r2duFAb7P95B6hEeOyEnHRa8MjYSMTY=
github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/uptrace/go-clickhouse/chotel v0.2.7 h1:MQ96m5c0EdEVdS0hwdGsSsaBimXOR6QbW7XlBrgvqeo=
github.com/uptrace/go-clickhouse/chotel v0.2.7/go.mod h1:MiJZdf9NrA5XiCzxHlGWGTm/6ctib6EsIihVZmz+R7E=
go.opentelemetry.io/otel v1.7.0 h1:Z2lA3Tdch0iDcrhJXDIlC94XE+bxok1F9B+4Lz/lGsM=
go.opentelemetry.io/otel v1.7.0/go.mod h1:5BdUoMIz5WEs0vt0CUEMtSSaTSHBBVwrhnz7+nrD5xk=
go.opentelemetry.io/otel/trace v1.7.0 h1:O37Iogk1lEkMRXewVtZ1BBTVn5JEp8GrJvP92bJqC6o=
go.opentelemetry.io/otel/trace v1.7.0/go.mod h1:fzLSB9nqR2eXzxPXb2JW9IKE+ScyXA48yyE4TNvoHqU=
golang.org/x/exp v0.0.0-20220428152302-39d4317da171 h1:TfdoLivD44QwvssI9Sv1xwa5DcL5XQr4au4sZ2F2NV4=
golang.org/x/exp v0.0.0-20220428152302-39d4317da171/go.mod h1:lgLbSvA5ygNOMpwM/9anMpWVlVJ7Z+cHWq/eFuinpGE=
golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@ -35,6 +48,7 @@ golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBc
golang.org/x/sys v0.0.0-20210927094055-39ccf1dd6fa6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220429233432-b5fbb4746d32 h1:Js08h5hqB5xyWR789+QqueR6sDE8mk+YvpETZ+F6X9Y=
golang.org/x/sys v0.0.0-20220429233432-b5fbb4746d32/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=