fix: renamed storage endpoint into dsn
Changes: - Renamed storage endpoint into dsn (data source name). - Add additional dsn fallback property. This dsn will be used in futes to store informations, if the main dsn backend does not work correctly. For example, if no connection can be established over the network to a database.
This commit is contained in:
parent
43e9d00dcb
commit
10069568f9
@ -19,7 +19,7 @@ func InitCmd(cmd *cobra.Command) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
daemonCmd.Flags().Bool("compression", true, "Compress measured values")
|
daemonCmd.Flags().Bool("compression", true, "Compress measured values")
|
||||||
daemonCmd.Flags().Uint("cached-values", 500, "Number of cached values before saveing into the storage endpoint")
|
daemonCmd.Flags().Uint("cached-values", 500, "Number of cached values before saveing into the backend")
|
||||||
daemonCmd.Flags().Float64("round", 0.5, "Round values. The value 0 deactivates the function")
|
daemonCmd.Flags().Float64("round", 0.5, "Round values. The value 0 deactivates the function")
|
||||||
cmd.AddCommand(daemonCmd)
|
cmd.AddCommand(daemonCmd)
|
||||||
|
|
||||||
|
@ -83,7 +83,7 @@ func preRunError(cmd *cobra.Command, args []string) error {
|
|||||||
Name: hostname,
|
Name: hostname,
|
||||||
CreationDate: postgresTimeStamp,
|
CreationDate: postgresTimeStamp,
|
||||||
},
|
},
|
||||||
StorageEndpoint: "sqlite3:///var/log/flucky/sqlite.db?cache=shared&mode=memory&FKSupport=True",
|
DSN: "sqlite3:///var/log/flucky/sqlite.db?cache=shared&mode=memory&FKSupport=True",
|
||||||
}
|
}
|
||||||
|
|
||||||
err = config.Write(&cnf, configFile)
|
err = config.Write(&cnf, configFile)
|
||||||
|
@ -164,7 +164,7 @@ func addSensor(cmd *cobra.Command, args []string) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
storageEndpoint, err := url.Parse(cnf.StorageEndpoint)
|
dsnURL, err := url.Parse(cnf.DSN)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -176,7 +176,7 @@ func addSensor(cmd *cobra.Command, args []string) error {
|
|||||||
|
|
||||||
flogger := logger.NewDefaultLogger(logger.LogLevelDebug)
|
flogger := logger.NewDefaultLogger(logger.LogLevelDebug)
|
||||||
|
|
||||||
repo, err := repository.New(storageEndpoint, flogger)
|
repo, err := repository.New(dsnURL, flogger)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -208,7 +208,7 @@ func disableSensor(cmd *cobra.Command, args []string) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
storageEndpoint, err := url.Parse(cnf.StorageEndpoint)
|
dsnURL, err := url.Parse(cnf.DSN)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -220,7 +220,7 @@ func disableSensor(cmd *cobra.Command, args []string) error {
|
|||||||
|
|
||||||
flogger := logger.NewDefaultLogger(logger.LogLevelDebug)
|
flogger := logger.NewDefaultLogger(logger.LogLevelDebug)
|
||||||
|
|
||||||
repo, err := repository.New(storageEndpoint, flogger)
|
repo, err := repository.New(dsnURL, flogger)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -240,7 +240,7 @@ func enableSensor(cmd *cobra.Command, args []string) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
storageEndpoint, err := url.Parse(cnf.StorageEndpoint)
|
dsnURL, err := url.Parse(cnf.DSN)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -252,7 +252,7 @@ func enableSensor(cmd *cobra.Command, args []string) error {
|
|||||||
|
|
||||||
flogger := logger.NewDefaultLogger(logger.LogLevelDebug)
|
flogger := logger.NewDefaultLogger(logger.LogLevelDebug)
|
||||||
|
|
||||||
repo, err := repository.New(storageEndpoint, flogger)
|
repo, err := repository.New(dsnURL, flogger)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -272,7 +272,7 @@ func listSensors(cmd *cobra.Command, args []string) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
storageEndpoint, err := url.Parse(cnf.StorageEndpoint)
|
dsnURL, err := url.Parse(cnf.DSN)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -284,7 +284,7 @@ func listSensors(cmd *cobra.Command, args []string) error {
|
|||||||
|
|
||||||
flogger := logger.NewDefaultLogger(logger.LogLevelDebug)
|
flogger := logger.NewDefaultLogger(logger.LogLevelDebug)
|
||||||
|
|
||||||
repo, err := repository.New(storageEndpoint, flogger)
|
repo, err := repository.New(dsnURL, flogger)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -315,7 +315,7 @@ func removeSensor(cmd *cobra.Command, args []string) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
storageEndpoint, err := url.Parse(cnf.StorageEndpoint)
|
dsnURL, err := url.Parse(cnf.DSN)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -327,7 +327,7 @@ func removeSensor(cmd *cobra.Command, args []string) error {
|
|||||||
|
|
||||||
flogger := logger.NewDefaultLogger(logger.LogLevelDebug)
|
flogger := logger.NewDefaultLogger(logger.LogLevelDebug)
|
||||||
|
|
||||||
repo, err := repository.New(storageEndpoint, flogger)
|
repo, err := repository.New(dsnURL, flogger)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -347,7 +347,7 @@ func renameSensor(cmd *cobra.Command, args []string) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
storageEndpoint, err := url.Parse(cnf.StorageEndpoint)
|
dsnURL, err := url.Parse(cnf.DSN)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -359,7 +359,7 @@ func renameSensor(cmd *cobra.Command, args []string) error {
|
|||||||
|
|
||||||
flogger := logger.NewDefaultLogger(logger.LogLevelDebug)
|
flogger := logger.NewDefaultLogger(logger.LogLevelDebug)
|
||||||
|
|
||||||
repo, err := repository.New(storageEndpoint, flogger)
|
repo, err := repository.New(dsnURL, flogger)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
@ -20,9 +20,10 @@ var (
|
|||||||
|
|
||||||
// Config represent the configuration
|
// Config represent the configuration
|
||||||
type Config struct {
|
type Config struct {
|
||||||
Device *types.Device `json:"device"`
|
Device *types.Device `json:"device"`
|
||||||
Sensors []*types.Sensor `json:"sensors"`
|
Sensors []*types.Sensor `json:"sensors"`
|
||||||
StorageEndpoint string `json:"storage_endpoint"`
|
DSN string `json:"dsn"`
|
||||||
|
DSNFallback string `json:"dsn_fallback"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// AddSensor add a new sensor
|
// AddSensor add a new sensor
|
||||||
|
@ -18,13 +18,13 @@ func Start(cnf *config.Config, flogger logger.Logger) error {
|
|||||||
|
|
||||||
measuredValueChannel := make(chan *types.MeasuredValue, 0)
|
measuredValueChannel := make(chan *types.MeasuredValue, 0)
|
||||||
|
|
||||||
// load storage endpoint
|
// load data source name (dsn)
|
||||||
storageEndpointURL, err := url.Parse(cnf.StorageEndpoint)
|
dsnURL, err := url.Parse(cnf.DSN)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
repo, err := repository.New(storageEndpointURL, flogger)
|
repo, err := repository.New(dsnURL, flogger)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
@ -13,7 +13,8 @@ import (
|
|||||||
"github.com/volker-raschek/go-logger/pkg/logger"
|
"github.com/volker-raschek/go-logger/pkg/logger"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Database is a general interface for a storage endpoint
|
// Database is a general interface for a database backend like postgres, oracle
|
||||||
|
// or sqlite
|
||||||
type Database interface {
|
type Database interface {
|
||||||
DeleteDevices(ctx context.Context, deviceIDs ...string) error
|
DeleteDevices(ctx context.Context, deviceIDs ...string) error
|
||||||
DeleteSensors(ctx context.Context, sensorIDs ...string) error
|
DeleteSensors(ctx context.Context, sensorIDs ...string) error
|
||||||
@ -35,12 +36,12 @@ type Database interface {
|
|||||||
UpdateSensors(ctx context.Context, sensors ...*types.Sensor) error
|
UpdateSensors(ctx context.Context, sensors ...*types.Sensor) error
|
||||||
}
|
}
|
||||||
|
|
||||||
// New returns a new storage provider
|
// New returns a new database backend interface
|
||||||
func New(storageEndpoint *url.URL, flogger logger.Logger) (Database, error) {
|
func New(dsnURL *url.URL, flogger logger.Logger) (Database, error) {
|
||||||
|
|
||||||
// Check of nil pointer
|
// Check of nil pointer
|
||||||
for _, parameter := range []interface{}{
|
for _, parameter := range []interface{}{
|
||||||
storageEndpoint,
|
dsnURL,
|
||||||
flogger,
|
flogger,
|
||||||
} {
|
} {
|
||||||
if parameter == nil {
|
if parameter == nil {
|
||||||
@ -51,7 +52,7 @@ func New(storageEndpoint *url.URL, flogger logger.Logger) (Database, error) {
|
|||||||
// Load Queryfiles
|
// Load Queryfiles
|
||||||
queries := make(map[string]string, 0)
|
queries := make(map[string]string, 0)
|
||||||
for _, asset := range AssetNames() {
|
for _, asset := range AssetNames() {
|
||||||
if !strings.Contains(asset, storageEndpoint.Scheme) {
|
if !strings.Contains(asset, dsnURL.Scheme) {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -70,7 +71,7 @@ func New(storageEndpoint *url.URL, flogger logger.Logger) (Database, error) {
|
|||||||
err error
|
err error
|
||||||
)
|
)
|
||||||
|
|
||||||
switch storageEndpoint.Scheme {
|
switch dsnURL.Scheme {
|
||||||
// case "postgres":
|
// case "postgres":
|
||||||
// // postgres://[user]:[password]@[host]:[port]/[path]?[query]
|
// // postgres://[user]:[password]@[host]:[port]/[path]?[query]
|
||||||
// newDBO, err := sql.Open(storageEndpoint.Scheme, storageEndpoint.String())
|
// newDBO, err := sql.Open(storageEndpoint.Scheme, storageEndpoint.String())
|
||||||
@ -86,8 +87,8 @@ func New(storageEndpoint *url.URL, flogger logger.Logger) (Database, error) {
|
|||||||
case "sqlite3":
|
case "sqlite3":
|
||||||
|
|
||||||
// Create directory where the db file will be created if not exists.
|
// Create directory where the db file will be created if not exists.
|
||||||
if _, err := os.Stat(filepath.Dir(storageEndpoint.Path)); os.IsNotExist(err) {
|
if _, err := os.Stat(filepath.Dir(dsnURL.Path)); os.IsNotExist(err) {
|
||||||
err := os.MkdirAll(filepath.Dir(storageEndpoint.Path), 0755)
|
err := os.MkdirAll(filepath.Dir(dsnURL.Path), 0755)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@ -95,7 +96,7 @@ func New(storageEndpoint *url.URL, flogger logger.Logger) (Database, error) {
|
|||||||
|
|
||||||
// sqlite3:///[path]?[query] flucky dsn
|
// sqlite3:///[path]?[query] flucky dsn
|
||||||
// file:///[path]?[query] sql-lib dsn
|
// file:///[path]?[query] sql-lib dsn
|
||||||
newDBO, err := sql.Open(storageEndpoint.Scheme, fmt.Sprintf("file://%v?%v", storageEndpoint.Path, storageEndpoint.RawQuery))
|
newDBO, err := sql.Open(dsnURL.Scheme, fmt.Sprintf("file://%v?%v", dsnURL.Path, dsnURL.RawQuery))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@ -106,7 +107,7 @@ func New(storageEndpoint *url.URL, flogger logger.Logger) (Database, error) {
|
|||||||
queries: queries,
|
queries: queries,
|
||||||
}
|
}
|
||||||
default:
|
default:
|
||||||
return nil, fmt.Errorf("Unsupported database scheme: %v", storageEndpoint.Scheme)
|
return nil, fmt.Errorf("Unsupported database scheme: %v", dsnURL.Scheme)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Initialize database scheme if not exists
|
// Initialize database scheme if not exists
|
||||||
|
@ -10,10 +10,6 @@ import (
|
|||||||
"github.com/volker-raschek/go-logger/pkg/logger"
|
"github.com/volker-raschek/go-logger/pkg/logger"
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
|
||||||
sqliteAssetPath = "pkg/storage/sqlite"
|
|
||||||
)
|
|
||||||
|
|
||||||
// SQLite implementation
|
// SQLite implementation
|
||||||
type SQLite struct {
|
type SQLite struct {
|
||||||
dbo *sql.DB
|
dbo *sql.DB
|
||||||
|
@ -176,9 +176,9 @@ func (repo *Repository) UpdateSensors(sensors ...*types.Sensor) error {
|
|||||||
return repo.database.UpdateSensors(context.Background(), sensors...)
|
return repo.database.UpdateSensors(context.Background(), sensors...)
|
||||||
}
|
}
|
||||||
|
|
||||||
// New returns a new repository based on the storageEndpoint
|
// New returns a new repository based on the data source name (dsn)
|
||||||
func New(storageEndpoint *url.URL, flogger logger.Logger) (*Repository, error) {
|
func New(dsnURL *url.URL, flogger logger.Logger) (*Repository, error) {
|
||||||
database, err := db.New(storageEndpoint, flogger)
|
database, err := db.New(dsnURL, flogger)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
@ -42,10 +42,10 @@ func TestPostgresBackend(t *testing.T) {
|
|||||||
require.NoError(err)
|
require.NoError(err)
|
||||||
|
|
||||||
// postgres://[user]:[password]@[host]:[port]/[path]?[query]
|
// postgres://[user]:[password]@[host]:[port]/[path]?[query]
|
||||||
storageEndpoint, err := url.Parse(fmt.Sprintf("postgres://postgres:%v@localhost:%v", postgresDBPasswort, postgresHostPort))
|
dsnURL, err := url.Parse(fmt.Sprintf("postgres://postgres:%v@localhost:%v", postgresDBPasswort, postgresHostPort))
|
||||||
require.NoError(err)
|
require.NoError(err)
|
||||||
|
|
||||||
repo, err := repository.New(storageEndpoint, logger.NewDefaultLogger(logger.LogLevelDebug))
|
repo, err := repository.New(dsnURL, logger.NewDefaultLogger(logger.LogLevelDebug))
|
||||||
require.NoError(err)
|
require.NoError(err)
|
||||||
|
|
||||||
testBackend(t, repo)
|
testBackend(t, repo)
|
||||||
@ -58,10 +58,10 @@ func TestSQLiteBackend(t *testing.T) {
|
|||||||
require.NoError(err)
|
require.NoError(err)
|
||||||
defer os.RemoveAll(workspace)
|
defer os.RemoveAll(workspace)
|
||||||
|
|
||||||
storageEndpoint, err := url.Parse(fmt.Sprintf("sqlite3://%v/test.db", workspace))
|
dsnURL, err := url.Parse(fmt.Sprintf("sqlite3://%v/test.db", workspace))
|
||||||
require.NoError(err)
|
require.NoError(err)
|
||||||
|
|
||||||
repo, err := repository.New(storageEndpoint, logger.NewDefaultLogger(logger.LogLevelDebug))
|
repo, err := repository.New(dsnURL, logger.NewDefaultLogger(logger.LogLevelDebug))
|
||||||
require.NoError(err)
|
require.NoError(err)
|
||||||
|
|
||||||
testBackend(t, repo)
|
testBackend(t, repo)
|
||||||
|
Loading…
Reference in New Issue
Block a user