CK server Code:1000 error
#61
Comments
|
How you do create a connection? can you please share options? |
|
|
type Bulk interface {
Append(row map[string]interface{})
Length() int
Full() bool
LowWaterMarkFull() bool
}
// this channel is shared globally.
var bulkChan = make(chan Bulk, 0)
// this struct implement `Bulk` interface and hold all data
// once bulk is full in collector, it will be pushed into `bulkChan`, then a `BulkWrite` will be fired.
// two rules:
// 1. Length() > config.BULK_SIZE (default 50,000)
// 2. last commit time is before now() - 10s(default) &&
// Length() > config.LOW_WATER_MARK(default 5,000)
// one collector goroutine will only have one bulk struct at the same time, no reuse. no share.
type ClickHouseBulk struct {
Rows []map[string]interface{}
DB string
Table string
}
func NewClickHouseBulk(db, table string) *ClickHouseBulk {
return &ClickHouseBulk{
DB: db,
Table: table,
Rows: make([]map[string]interface{}, 0),
}
}
type CkServer struct {
conn *sql.DB
...
}
func NewCkServer(...) (*CkServer, error){
// http://username:password@ip:port/database?enable_http_comression=1"
// http://username:password@ip:port/database"
// no more options to form a url
db, err := sql.Open("clickhouse", url)
if err != nil {
return nil, err
}
...
// ~~no goroutine will be started.~~
// return &CkServer{
// conn: db,
// }, nil
// forget about ping goroutine.
ck := &CkServer{
conn: db,
}
go ck.ping()
return ck, nil
}
func (ck *CkServer) ping() {
// ping this ckserver every one second with an increment backoff once failed.
// if failed 5 times or more, we will stop write to this ckserver.
}
func (ck *CkServer) BulkWrite(bulk *ClickHouseBulk) error {
// ...
key := bulk.DB + ":" + bulk.Table
if table, ok := ck.tables[key]; ok == true {
return table.bulkWrite(ck.conn, bulk)
} else {
// we only write to a single table with in one process. this branch
// will only be called once in the beginning. and will fetch table
// desc blockingly(no goroutine).
table, err := newCkTable(ck.conn, bulk.DB, bulk.Table, ck.logger)
if err != nil {
return err
}
ck.tables[key] = table
return table.bulkWrite(ck.conn, bulk)
}
}
func newCkTable(conn *sql.DB, db, table string, logger zerolog.Logger) (*ckTable, error) {
lg := logger.With().Str("db", db).Str("table", table).Logger()
sampled := util.GetBrustSampledLogger(lg)
cktable := &ckTable{db: db, table: table, logger: lg, sampledLogger: sampled}
//TODO: what if refresh failed the first time ?
err := cktable.refreshDescAndDDLOnce(conn)
if err != nil {
return nil, err
}
return cktable, nil
}
func (t *ckTable) bulkWrite(conn *sql.DB, bulk *ClickHouseBulk) error {
defer logx.RecoverPrintStack()
t.logger.Debug().Msg("bulkWrite start")
desc := t.getDesc()
if desc == nil {
t.logger.Debug().Msg("table desc is nil")
return errors.New(fmt.Sprintf("table desc is nil, db: %s, table: %s", t.db, t.table))
}
tx, err := conn.Begin()
if err != nil {
return err
}
// odd... this has to be added while using `github.com/kshvakov/clickhouse`, otherwise OOM
// with this driver. even comment this out, `Code:1000` remains
//defer tx.Rollback()
stmt, err := tx.Prepare(desc.insertDDL)
if err != nil {
return err
}
// comment this out, `Code:1000` remains
defer stmt.Close()
for _, msg := range bulk.Rows {
var row []interface{}
// .... process row here.
for _, key := range desc.keys {
row_desc, ok := desc.dMap[key]
if ok == false {
//TODO: should not happend, replace it with orderedMap.
t.sampledLogger.Error().Msg("desc key not in desc.dMap")
break
}
if value, ok := msg[key]; ok == true && value != nil {
row = append(row, value)
} else {
row = append(row, row_desc.Default)
}
}
if _, err := stmt.Exec(row...); err != nil {
t.sampledLogger.Error().Err(err).Msg("stmt.Exec failed.")
}
}
// commit time
start := time.Now()
err = tx.Commit()
ometrics.BulkCommitTimer.UpdateSince(start)
return err
}
// start many worker to consume `bulkCh`
for {
select {
case bulk := <-bulkCh:
err := ckSrv.BulkWrite(bulk)
if err != nil {
ckSrv.logger.Error().Err(err).Msg("bulk write failed")
}
}
}
// so the path is clear
// bulk := <- bulkCh
// ckServer.BulkWrite(bulk)
// ckTable.bulkWrite(bulk) |
|
commet |
|
|
|
same happens with me:
I use conn string like that: |
|
Hi @chobostar can you check if the problem stays on a newer version 1.5.0 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
initially, we use
github.com/kshvakov/clickhousefor a little while. but we found it write data much slower than java http client (the only one listed in ck document). (snmpdump shows 10x smaller package size compare to java http client)so we replaced it with this driver and outcome is good (similar performance as java http client). there are only minor flaws: once we start the program ck server pop out many
Code: 1000error. which never happens before. even withgithub.com/kshvakov/clickhouseor java client.since
db/sqlprovider high level abstraction. so we don't touch the code to write clickhouse while replacing the driver. we only modifed some args to setup the connection.main code looks like this
no where to handle connection anywhere in our code.... and the worker goroutine never exit
The text was updated successfully, but these errors were encountered: