storj/cmd/statreceiver/db.go
JT Olio 362f447d9f
cmd/statreceiver: lua-scriptable stat receiver (#636)
* cmd/statreceiver: lua-scriptable stat receiver

Change-Id: I3ce0fe3f1ef4b1f4f27eed90bac0e91cfecf22d7

* some updates

Change-Id: I7c3485adcda1278fce01ae077b4761b3ddb9fb7a

* more comments

Change-Id: I0bb22993cd934c3d40fc1da80d07e49e686b80dd

* linter fixes

Change-Id: Ied014304ecb9aadcf00a6b66ad28f856a428d150

* catch errors

Change-Id: I6e1920f1fd941e66199b30bc427285c19769fc70

* review feedback

Change-Id: I9d4051851eab18970c5f5ddcf4ff265508e541d3

* errorgroup improvements

Change-Id: I4699dda3022f0485fbb50c9dafe692d3921734ff

* too tricky

the previous thing was better for memory with lots of errors at a time
but https://play.golang.org/p/RweTMRjoSCt is too much of a foot gun

Change-Id: I23f0b3d77dd4288fcc20b3756a7110359576bf44
2018-12-11 11:24:31 -07:00

63 lines
1.5 KiB
Go

// Copyright (C) 2018 Storj Labs, Inc.
// See LICENSE for copying information.
package main
import (
"database/sql"
"fmt"
"sync"
"time"
_ "github.com/lib/pq"
_ "github.com/mattn/go-sqlite3"
)
var (
sqlupsert = map[string]string{
"sqlite3": "INSERT INTO metrics (metric, instance, val, timestamp) " +
"VALUES (?, ?, ?, ?) ON CONFLICT(metric, instance) DO UPDATE SET " +
"val=excluded.val, timestamp=excluded.timestamp;",
"postgres": "INSERT INTO metrics (metric, instance, val, timestamp) " +
"VALUES ($1, $2, $3, $4) ON CONFLICT(metric, instance) DO UPDATE SET " +
"val=EXCLUDED.val, timestamp=EXCLUDED.timestamp;",
}
)
// DBDest is a database metric destination. It stores the latest value given
// a metric key and application per instance.
type DBDest struct {
mtx sync.Mutex
driver, address string
db *sql.DB
}
// NewDBDest creates a DBDest
func NewDBDest(driver, address string) *DBDest {
if _, found := sqlupsert[driver]; !found {
panic(fmt.Sprintf("driver %s not supported", driver))
}
return &DBDest{
driver: driver,
address: address,
}
}
// Metric implements the MetricDest interface
func (db *DBDest) Metric(application, instance string, key []byte, val float64,
ts time.Time) error {
db.mtx.Lock()
if db.db == nil {
conn, err := sql.Open(db.driver, db.address)
if err != nil {
db.mtx.Unlock()
return err
}
db.db = conn
}
db.mtx.Unlock()
_, err := db.db.Exec(sqlupsert[db.driver], application+"."+string(key),
instance, val, ts.Unix())
return err
}