From 06ee748c1ca23bf489ae707041f0e8f3e67fbaf7 Mon Sep 17 00:00:00 2001 From: Valery Piashchynski Date: Thu, 22 Apr 2021 22:53:12 +0300 Subject: - Add redis driver Signed-off-by: Valery Piashchynski --- plugins/kv/drivers/boltdb/driver.go | 14 +-- plugins/kv/drivers/redis/config.go | 34 ++++++ plugins/kv/drivers/redis/driver.go | 238 ++++++++++++++++++++++++++++++++++++ plugins/kv/drivers/redis/plugin.go | 51 ++++++++ 4 files changed, 330 insertions(+), 7 deletions(-) create mode 100644 plugins/kv/drivers/redis/config.go create mode 100644 plugins/kv/drivers/redis/driver.go create mode 100644 plugins/kv/drivers/redis/plugin.go (limited to 'plugins/kv/drivers') diff --git a/plugins/kv/drivers/boltdb/driver.go b/plugins/kv/drivers/boltdb/driver.go index 8c3bb68f..b596d4c3 100644 --- a/plugins/kv/drivers/boltdb/driver.go +++ b/plugins/kv/drivers/boltdb/driver.go @@ -80,7 +80,7 @@ func NewBoltDBDriver(log logger.Logger, key string, cfgPlugin config.Configurer, } func (d *Driver) Has(keys ...string) (map[string]bool, error) { - const op = errors.Op("boltdb_plugin_has") + const op = errors.Op("boltdb_driver_has") d.log.Debug("boltdb HAS method called", "args", keys) if keys == nil { return nil, errors.E(op, errors.NoKeys) @@ -121,7 +121,7 @@ func (d *Driver) Has(keys ...string) (map[string]bool, error) { // Returns a nil value if the key does not exist or if the key is a nested bucket. // The returned value is only valid for the life of the transaction. func (d *Driver) Get(key string) ([]byte, error) { - const op = errors.Op("boltdb_plugin_get") + const op = errors.Op("boltdb_driver_get") // to get cases like " " keyTrimmed := strings.TrimSpace(key) if keyTrimmed == "" { @@ -161,7 +161,7 @@ func (d *Driver) Get(key string) ([]byte, error) { } func (d *Driver) MGet(keys ...string) (map[string]interface{}, error) { - const op = errors.Op("boltdb_plugin_mget") + const op = errors.Op("boltdb_driver_mget") // defense if keys == nil { return nil, errors.E(op, errors.NoKeys) @@ -213,7 +213,7 @@ func (d *Driver) MGet(keys ...string) (map[string]interface{}, error) { // Set puts the K/V to the bolt func (d *Driver) Set(items ...kv.Item) error { - const op = errors.Op("boltdb_plugin_set") + const op = errors.Op("boltdb_driver_set") if items == nil { return errors.E(op, errors.NoKeys) } @@ -276,7 +276,7 @@ func (d *Driver) Set(items ...kv.Item) error { // Delete all keys from DB func (d *Driver) Delete(keys ...string) error { - const op = errors.Op("boltdb_plugin_delete") + const op = errors.Op("boltdb_driver_delete") if keys == nil { return errors.E(op, errors.NoKeys) } @@ -323,7 +323,7 @@ func (d *Driver) Delete(keys ...string) error { // MExpire sets the expiration time to the key // If key already has the expiration time, it will be overwritten func (d *Driver) MExpire(items ...kv.Item) error { - const op = errors.Op("boltdb_plugin_mexpire") + const op = errors.Op("boltdb_driver_mexpire") for i := range items { if items[i].TTL == "" || strings.TrimSpace(items[i].Key) == "" { return errors.E(op, errors.Str("should set timeout and at least one key")) @@ -341,7 +341,7 @@ func (d *Driver) MExpire(items ...kv.Item) error { } func (d *Driver) TTL(keys ...string) (map[string]interface{}, error) { - const op = errors.Op("boltdb_plugin_ttl") + const op = errors.Op("boltdb_driver_ttl") if keys == nil { return nil, errors.E(op, errors.NoKeys) } diff --git a/plugins/kv/drivers/redis/config.go b/plugins/kv/drivers/redis/config.go new file mode 100644 index 00000000..41348236 --- /dev/null +++ b/plugins/kv/drivers/redis/config.go @@ -0,0 +1,34 @@ +package redis + +import "time" + +type Config struct { + Addrs []string `mapstructure:"addrs"` + DB int `mapstructure:"db"` + Username string `mapstructure:"username"` + Password string `mapstructure:"password"` + MasterName string `mapstructure:"master_name"` + SentinelPassword string `mapstructure:"sentinel_password"` + RouteByLatency bool `mapstructure:"route_by_latency"` + RouteRandomly bool `mapstructure:"route_randomly"` + MaxRetries int `mapstructure:"max_retries"` + DialTimeout time.Duration `mapstructure:"dial_timeout"` + MinRetryBackoff time.Duration `mapstructure:"min_retry_backoff"` + MaxRetryBackoff time.Duration `mapstructure:"max_retry_backoff"` + PoolSize int `mapstructure:"pool_size"` + MinIdleConns int `mapstructure:"min_idle_conns"` + MaxConnAge time.Duration `mapstructure:"max_conn_age"` + ReadTimeout time.Duration `mapstructure:"read_timeout"` + WriteTimeout time.Duration `mapstructure:"write_timeout"` + PoolTimeout time.Duration `mapstructure:"pool_timeout"` + IdleTimeout time.Duration `mapstructure:"idle_timeout"` + IdleCheckFreq time.Duration `mapstructure:"idle_check_freq"` + ReadOnly bool `mapstructure:"read_only"` +} + +// InitDefaults initializing fill config with default values +func (s *Config) InitDefaults() { + if s.Addrs == nil { + s.Addrs = []string{"localhost:6379"} // default addr is pointing to local storage + } +} diff --git a/plugins/kv/drivers/redis/driver.go b/plugins/kv/drivers/redis/driver.go new file mode 100644 index 00000000..3cc35c13 --- /dev/null +++ b/plugins/kv/drivers/redis/driver.go @@ -0,0 +1,238 @@ +package redis + +import ( + "context" + "strings" + "time" + + "github.com/go-redis/redis/v8" + "github.com/spiral/errors" + "github.com/spiral/roadrunner/v2/plugins/config" + "github.com/spiral/roadrunner/v2/plugins/kv" + "github.com/spiral/roadrunner/v2/plugins/logger" +) + +var EmptyItem = kv.Item{} + +type Driver struct { + universalClient redis.UniversalClient + log logger.Logger + cfg *Config +} + +func NewRedisDriver(log logger.Logger, key string, cfgPlugin config.Configurer) (kv.Storage, error) { + const op = errors.Op("new_boltdb_driver") + + d := &Driver{ + log: log, + } + + err := cfgPlugin.UnmarshalKey(key, &d.cfg) + if err != nil { + return nil, errors.E(op, err) + } + + d.cfg.InitDefaults() + d.log = log + + d.universalClient = redis.NewUniversalClient(&redis.UniversalOptions{ + Addrs: d.cfg.Addrs, + DB: d.cfg.DB, + Username: d.cfg.Username, + Password: d.cfg.Password, + SentinelPassword: d.cfg.SentinelPassword, + MaxRetries: d.cfg.MaxRetries, + MinRetryBackoff: d.cfg.MaxRetryBackoff, + MaxRetryBackoff: d.cfg.MaxRetryBackoff, + DialTimeout: d.cfg.DialTimeout, + ReadTimeout: d.cfg.ReadTimeout, + WriteTimeout: d.cfg.WriteTimeout, + PoolSize: d.cfg.PoolSize, + MinIdleConns: d.cfg.MinIdleConns, + MaxConnAge: d.cfg.MaxConnAge, + PoolTimeout: d.cfg.PoolTimeout, + IdleTimeout: d.cfg.IdleTimeout, + IdleCheckFrequency: d.cfg.IdleCheckFreq, + ReadOnly: d.cfg.ReadOnly, + RouteByLatency: d.cfg.RouteByLatency, + RouteRandomly: d.cfg.RouteRandomly, + MasterName: d.cfg.MasterName, + }) + + return d, nil +} + +// Has checks if value exists. +func (d *Driver) Has(keys ...string) (map[string]bool, error) { + const op = errors.Op("redis_driver_has") + if keys == nil { + return nil, errors.E(op, errors.NoKeys) + } + + m := make(map[string]bool, len(keys)) + for _, key := range keys { + keyTrimmed := strings.TrimSpace(key) + if keyTrimmed == "" { + return nil, errors.E(op, errors.EmptyKey) + } + + exist, err := d.universalClient.Exists(context.Background(), key).Result() + if err != nil { + return nil, err + } + if exist == 1 { + m[key] = true + } + } + return m, nil +} + +// Get loads key content into slice. +func (d *Driver) Get(key string) ([]byte, error) { + const op = errors.Op("redis_driver_get") + // to get cases like " " + keyTrimmed := strings.TrimSpace(key) + if keyTrimmed == "" { + return nil, errors.E(op, errors.EmptyKey) + } + return d.universalClient.Get(context.Background(), key).Bytes() +} + +// MGet loads content of multiple values (some values might be skipped). +// https://redis.io/commands/mget +// Returns slice with the interfaces with values +func (d *Driver) MGet(keys ...string) (map[string]interface{}, error) { + const op = errors.Op("redis_driver_mget") + if keys == nil { + return nil, errors.E(op, errors.NoKeys) + } + + // should not be empty keys + for _, key := range keys { + keyTrimmed := strings.TrimSpace(key) + if keyTrimmed == "" { + return nil, errors.E(op, errors.EmptyKey) + } + } + + m := make(map[string]interface{}, len(keys)) + + for _, k := range keys { + cmd := d.universalClient.Get(context.Background(), k) + if cmd.Err() != nil { + if cmd.Err() == redis.Nil { + continue + } + return nil, errors.E(op, cmd.Err()) + } + + m[k] = cmd.Val() + } + + return m, nil +} + +// Set sets value with the TTL in seconds +// https://redis.io/commands/set +// Redis `SET key value [expiration]` command. +// +// Use expiration for `SETEX`-like behavior. +// Zero expiration means the key has no expiration time. +func (d *Driver) Set(items ...kv.Item) error { + const op = errors.Op("redis_driver_set") + if items == nil { + return errors.E(op, errors.NoKeys) + } + now := time.Now() + for _, item := range items { + if item == EmptyItem { + return errors.E(op, errors.EmptyKey) + } + + if item.TTL == "" { + err := d.universalClient.Set(context.Background(), item.Key, item.Value, 0).Err() + if err != nil { + return err + } + } else { + t, err := time.Parse(time.RFC3339, item.TTL) + if err != nil { + return err + } + err = d.universalClient.Set(context.Background(), item.Key, item.Value, t.Sub(now)).Err() + if err != nil { + return err + } + } + } + return nil +} + +// Delete one or multiple keys. +func (d *Driver) Delete(keys ...string) error { + const op = errors.Op("redis_driver_delete") + if keys == nil { + return errors.E(op, errors.NoKeys) + } + + // should not be empty keys + for _, key := range keys { + keyTrimmed := strings.TrimSpace(key) + if keyTrimmed == "" { + return errors.E(op, errors.EmptyKey) + } + } + return d.universalClient.Del(context.Background(), keys...).Err() +} + +// MExpire https://redis.io/commands/expire +// timeout in RFC3339 +func (d *Driver) MExpire(items ...kv.Item) error { + const op = errors.Op("redis_driver_mexpire") + now := time.Now() + for _, item := range items { + if item.TTL == "" || strings.TrimSpace(item.Key) == "" { + return errors.E(op, errors.Str("should set timeout and at least one key")) + } + + t, err := time.Parse(time.RFC3339, item.TTL) + if err != nil { + return err + } + + // t guessed to be in future + // for Redis we use t.Sub, it will result in seconds, like 4.2s + d.universalClient.Expire(context.Background(), item.Key, t.Sub(now)) + } + + return nil +} + +// TTL https://redis.io/commands/ttl +// return time in seconds (float64) for a given keys +func (d *Driver) TTL(keys ...string) (map[string]interface{}, error) { + const op = errors.Op("redis_driver_ttl") + if keys == nil { + return nil, errors.E(op, errors.NoKeys) + } + + // should not be empty keys + for _, key := range keys { + keyTrimmed := strings.TrimSpace(key) + if keyTrimmed == "" { + return nil, errors.E(op, errors.EmptyKey) + } + } + + m := make(map[string]interface{}, len(keys)) + + for _, key := range keys { + duration, err := d.universalClient.TTL(context.Background(), key).Result() + if err != nil { + return nil, err + } + + m[key] = duration.Seconds() + } + return m, nil +} diff --git a/plugins/kv/drivers/redis/plugin.go b/plugins/kv/drivers/redis/plugin.go new file mode 100644 index 00000000..d2183411 --- /dev/null +++ b/plugins/kv/drivers/redis/plugin.go @@ -0,0 +1,51 @@ +package redis + +import ( + "github.com/spiral/errors" + "github.com/spiral/roadrunner/v2/plugins/config" + "github.com/spiral/roadrunner/v2/plugins/kv" + "github.com/spiral/roadrunner/v2/plugins/logger" +) + +const PluginName = "redis" + +// Plugin BoltDB K/V storage. +type Plugin struct { + cfgPlugin config.Configurer + // logger + log logger.Logger +} + +func (s *Plugin) Init(log logger.Logger, cfg config.Configurer) error { + if !cfg.Has(kv.PluginName) { + return errors.E(errors.Disabled) + } + + s.log = log + s.cfgPlugin = cfg + return nil +} + +// Serve is noop here +func (s *Plugin) Serve() chan error { + return make(chan error, 1) +} + +func (s *Plugin) Stop() error { + return nil +} + +func (s *Plugin) Provide(key string) (kv.Storage, error) { + const op = errors.Op("redis_plugin_provide") + st, err := NewRedisDriver(s.log, key, s.cfgPlugin) + if err != nil { + return nil, errors.E(op, err) + } + + return st, nil +} + +// Name returns plugin name +func (s *Plugin) Name() string { + return PluginName +} -- cgit v1.2.3