forked from timescale/tsbs
-
Notifications
You must be signed in to change notification settings - Fork 6
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merged in rrk/dbcreator-interface (pull request timescale#73)
Unify the database setup in loaders to use DBCreator interface Approved-by: Lee Hampton <leejhampton@gmail.com>
- Loading branch information
Showing
18 changed files
with
741 additions
and
571 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,86 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "log" | ||
| "strings" | ||
| "time" | ||
|
|
||
| "github.com/gocql/gocql" | ||
| ) | ||
|
|
||
| type dbCreator struct { | ||
| globalSession *gocql.Session | ||
| clientSession *gocql.Session | ||
| } | ||
|
|
||
| func (d *dbCreator) Init() { | ||
| cluster := gocql.NewCluster(strings.Split(hosts, ",")...) | ||
| cluster.Consistency = consistencyMapping[consistencyLevel] | ||
| cluster.ProtoVersion = 4 | ||
| cluster.Timeout = 10 * time.Second | ||
| session, err := cluster.CreateSession() | ||
| if err != nil { | ||
| log.Fatal(err) | ||
| } | ||
| d.globalSession = session | ||
| } | ||
|
|
||
| func (d *dbCreator) DBExists(dbName string) bool { | ||
| iter := d.globalSession.Query(fmt.Sprintf("SELECT keyspace_name FROM system_schema.keyspaces;")).Iter() | ||
| defer iter.Close() | ||
| row := "" | ||
| for iter.Scan(&row) { | ||
| if row == dbName { | ||
| return true | ||
| } | ||
| } | ||
| return false | ||
| } | ||
|
|
||
| func (d *dbCreator) RemoveOldDB(dbName string) error { | ||
| if err := d.globalSession.Query(fmt.Sprintf("drop keyspace if exists %s;", dbName)).Exec(); err != nil { | ||
| return err | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (d *dbCreator) CreateDB(dbName string) error { | ||
| defer d.globalSession.Close() | ||
| replicationConfiguration := fmt.Sprintf("{ 'class': 'SimpleStrategy', 'replication_factor': %d }", replicationFactor) | ||
| if err := d.globalSession.Query(fmt.Sprintf("create keyspace %s with replication = %s;", dbName, replicationConfiguration)).Exec(); err != nil { | ||
| return err | ||
| } | ||
| for _, cassandraTypename := range []string{"bigint", "float", "double", "boolean", "blob"} { | ||
| q := fmt.Sprintf(`CREATE TABLE %s.series_%s ( | ||
| series_id text, | ||
| timestamp_ns bigint, | ||
| value %s, | ||
| PRIMARY KEY (series_id, timestamp_ns) | ||
| ) | ||
| WITH COMPACT STORAGE;`, | ||
| dbName, cassandraTypename, cassandraTypename) | ||
| if err := d.globalSession.Query(q).Exec(); err != nil { | ||
| return err | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (d *dbCreator) PostCreateDB(dbName string) error { | ||
| cluster := gocql.NewCluster(strings.Split(hosts, ",")...) | ||
| cluster.Keyspace = dbName | ||
| cluster.Timeout = writeTimeout | ||
| cluster.Consistency = consistencyMapping[consistencyLevel] | ||
| cluster.ProtoVersion = 4 | ||
| session, err := cluster.CreateSession() | ||
| if err != nil { | ||
| return err | ||
| } | ||
| d.clientSession = session | ||
| return nil | ||
| } | ||
|
|
||
| func (d *dbCreator) Close() { | ||
| d.clientSession.Close() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,120 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "encoding/json" | ||
| "fmt" | ||
| "io/ioutil" | ||
| "log" | ||
| "net/http" | ||
| "net/url" | ||
| "time" | ||
| ) | ||
|
|
||
| type dbCreator struct { | ||
| daemonURL string | ||
| } | ||
|
|
||
| func (d *dbCreator) Init() { | ||
| d.daemonURL = daemonURLs[0] // pick first one since it always exists | ||
| } | ||
|
|
||
| func (d *dbCreator) DBExists(dbName string) bool { | ||
| dbs, err := d.listDatabases() | ||
| if err != nil { | ||
| log.Fatal(err) | ||
| } | ||
|
|
||
| for _, db := range dbs { | ||
| if db == loader.DatabaseName() { | ||
| return true | ||
| } | ||
| } | ||
| return false | ||
| } | ||
|
|
||
| func (d *dbCreator) listDatabases() ([]string, error) { | ||
| u := fmt.Sprintf("%s/query?q=show%%20databases", d.daemonURL) | ||
| resp, err := http.Get(u) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("listDatabases error: %s", err.Error()) | ||
| } | ||
| defer resp.Body.Close() | ||
|
|
||
| body, err := ioutil.ReadAll(resp.Body) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| // Do ad-hoc parsing to find existing database names: | ||
| // {"results":[{"series":[{"name":"databases","columns":["name"],"values":[["_internal"],["benchmark_db"]]}]}]}% | ||
| type listingType struct { | ||
| Results []struct { | ||
| Series []struct { | ||
| Values [][]string | ||
| } | ||
| } | ||
| } | ||
| var listing listingType | ||
| err = json.Unmarshal(body, &listing) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| ret := []string{} | ||
| for _, nestedName := range listing.Results[0].Series[0].Values { | ||
| name := nestedName[0] | ||
| // the _internal database is skipped: | ||
| if name == "_internal" { | ||
| continue | ||
| } | ||
| ret = append(ret, name) | ||
| } | ||
| return ret, nil | ||
| } | ||
|
|
||
| func (d *dbCreator) RemoveOldDB(dbName string) error { | ||
| u := fmt.Sprintf("%s/query?q=drop+database+%s", d.daemonURL, dbName) | ||
| resp, err := http.Post(u, "text/plain", nil) | ||
| if err != nil { | ||
| return fmt.Errorf("drop db error: %s", err.Error()) | ||
| } | ||
| if resp.StatusCode != 200 { | ||
| return fmt.Errorf("drop db returned non-200 code: %d", resp.StatusCode) | ||
| } | ||
| time.Sleep(time.Second) | ||
| return nil | ||
| } | ||
|
|
||
| func (d *dbCreator) CreateDB(dbName string) error { | ||
| u, err := url.Parse(d.daemonURL) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| // serialize params the right way: | ||
| u.Path = "query" | ||
| v := u.Query() | ||
| v.Set("consistency", "all") | ||
| v.Set("q", fmt.Sprintf("CREATE DATABASE %s WITH REPLICATION %d", dbName, replicationFactor)) | ||
| u.RawQuery = v.Encode() | ||
|
|
||
| req, err := http.NewRequest("GET", u.String(), nil) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| client := &http.Client{} | ||
| resp, err := client.Do(req) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| defer resp.Body.Close() | ||
| // does the body need to be read into the void? | ||
|
|
||
| if resp.StatusCode != 200 { | ||
| return fmt.Errorf("bad db create") | ||
| } | ||
|
|
||
| time.Sleep(time.Second) | ||
| return nil | ||
| } |
Oops, something went wrong.