feat: add initial implementation of event mechanism

Author: buraksezer

cluster.go

@@ -161,9 +161,11 @@ func (db *Olric) clusterMembersCommandHandler(conn redcon.Conn, cmd redcon.Comma
 	members := db.rt.Discovery().GetMembers()
 	conn.WriteArray(len(members))
 	for _, member := range members {
-		conn.WriteArray(4)
+		conn.WriteArray(3)
 		conn.WriteBulkString(member.Name)
-		conn.WriteUint64(member.ID)
+		// go-redis/redis package cannot handle uint64. At the time of this writing,
+		// there is no solution for this, and I don't want to use a soft fork to repair it.
+		//conn.WriteUint64(member.ID)
 		conn.WriteInt64(member.Birthdate)
 		if coordinator.CompareByID(member) {
 			conn.WriteBulkString("true")

cluster_client.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/buraksezer/olric/internal/discovery"
 	"log"
 	"os"
 	"time"
@@ -491,16 +492,13 @@ func (cl *ClusterClient) Members(ctx context.Context) ([]Member, error) {
 		m := Member{}
 		item := rawItem.([]interface{})
 		m.Name = item[0].(string)
+		m.Birthdate = item[1].(int64)
 
-		switch id := item[1].(type) {
-		case uint64:
-			m.ID = id
-		case int64:
-			m.ID = uint64(id)
-		}
+		// go-redis/redis package cannot handle uint64 type. At the time of this writing,
+		// there is no solution for this, and I don't want to use a soft fork to repair it.
+		m.ID = discovery.MemberID(m.Name, m.Birthdate)
 
-		m.Birthdate = item[2].(int64)
-		if item[3] == "true" {
+		if item[2] == "true" {
 			m.Coordinator = true
 		}
 		members = append(members, m)

cmd/olricd/olricd.yaml

@@ -42,6 +42,8 @@ olricd:
   memberCountQuorum: 1
   routingTablePushInterval: 1m
 
+  enableClusterEventsChannel: true
+
 storageEngines:
   config:
     kvstore:
@@ -186,7 +188,7 @@ dmaps:
   engine:
     name: kvstore
     config:
-      tableSize: 4096
+      tableSize: 1048576
 #  checkEmptyFragmentsInterval: 1m
 #  triggerCompactionInterval: 10m
 #  numEvictionWorkers: 1

config/config.go

@@ -225,6 +225,8 @@ type Config struct {
 	// load for a server in the cluster. Keep it small.
 	LoadFactor float64
 
+	EnableClusterEventsChannel bool
+
 	// Default hasher is github.com/cespare/xxhash/v2
 	Hasher hasher.Hasher
 
@@ -370,6 +372,8 @@ func (c *Config) Sanitize() error {
 
 	if c.Logger == nil {
 		c.Logger = log.New(c.LogOutput, "", log.LstdFlags)
+	} else {
+		c.Logger.SetOutput(c.LogOutput)
 	}
 
 	if c.Hasher == nil {
@@ -443,8 +447,10 @@ func (c *Config) Sanitize() error {
 	return nil
 }
 
-// New returns a Config with sane defaults.
-// It takes an env parameter used by memberlist: local, lan and wan.
+// New returns a Config with sane defaults. If you change a configuration parameter,
+// please run Sanitize and Validate functions respectively.
+//
+// New takes an env parameter used by memberlist: local, lan and wan.
 //
 // local:
 //

config/internal/loader/loader.go

@@ -17,23 +17,24 @@ package loader
 import "gopkg.in/yaml.v2"
 
 type olricd struct {
-	Name                     string  `yaml:"name"`
-	BindAddr                 string  `yaml:"bindAddr"`
-	BindPort                 int     `yaml:"bindPort"`
-	Interface                string  `yaml:"interface"`
-	ReplicationMode          int     `yaml:"replicationMode"`
-	PartitionCount           uint64  `yaml:"partitionCount"`
-	LoadFactor               float64 `yaml:"loadFactor"`
-	KeepAlivePeriod          string  `yaml:"keepAlivePeriod"`
-	BootstrapTimeout         string  `yaml:"bootstrapTimeout"`
-	ReplicaCount             int     `yaml:"replicaCount"`
-	WriteQuorum              int     `yaml:"writeQuorum"`
-	ReadQuorum               int     `yaml:"readQuorum"`
-	ReadRepair               bool    `yaml:"readRepair"`
-	MemberCountQuorum        int32   `yaml:"memberCountQuorum"`
-	RoutingTablePushInterval string  `yaml:"routingTablePushInterval"`
-	TriggerBalancerInterval  string  `yaml:"triggerBalancerInterval"`
-	LeaveTimeout             string  `yaml:"leaveTimeout"`
+	Name                       string  `yaml:"name"`
+	BindAddr                   string  `yaml:"bindAddr"`
+	BindPort                   int     `yaml:"bindPort"`
+	Interface                  string  `yaml:"interface"`
+	ReplicationMode            int     `yaml:"replicationMode"`
+	PartitionCount             uint64  `yaml:"partitionCount"`
+	LoadFactor                 float64 `yaml:"loadFactor"`
+	KeepAlivePeriod            string  `yaml:"keepAlivePeriod"`
+	BootstrapTimeout           string  `yaml:"bootstrapTimeout"`
+	ReplicaCount               int     `yaml:"replicaCount"`
+	WriteQuorum                int     `yaml:"writeQuorum"`
+	ReadQuorum                 int     `yaml:"readQuorum"`
+	ReadRepair                 bool    `yaml:"readRepair"`
+	MemberCountQuorum          int32   `yaml:"memberCountQuorum"`
+	RoutingTablePushInterval   string  `yaml:"routingTablePushInterval"`
+	TriggerBalancerInterval    string  `yaml:"triggerBalancerInterval"`
+	LeaveTimeout               string  `yaml:"leaveTimeout"`
+	EnableClusterEventsChannel bool    `yaml:"enableClusterEventsChannel"`
 }
 
 type client struct {

config/load.go

@@ -364,35 +364,36 @@ func Load(filename string) (*Config, error) {
 	}
 
 	cfg := &Config{
-		BindAddr:                 c.Olricd.BindAddr,
-		BindPort:                 c.Olricd.BindPort,
-		Interface:                c.Olricd.Interface,
-		ServiceDiscovery:         c.ServiceDiscovery,
-		MemberlistInterface:      c.Memberlist.Interface,
-		MemberlistConfig:         memberlistConfig,
-		Client:                   &clientConfig,
-		LogLevel:                 c.Logging.Level,
-		JoinRetryInterval:        joinRetryInterval,
-		RoutingTablePushInterval: routingTablePushInterval,
-		TriggerBalancerInterval:  triggerBalancerInterval,
-		MaxJoinAttempts:          c.Memberlist.MaxJoinAttempts,
-		Peers:                    c.Memberlist.Peers,
-		PartitionCount:           c.Olricd.PartitionCount,
-		ReplicaCount:             c.Olricd.ReplicaCount,
-		WriteQuorum:              c.Olricd.WriteQuorum,
-		ReadQuorum:               c.Olricd.ReadQuorum,
-		ReplicationMode:          c.Olricd.ReplicationMode,
-		ReadRepair:               c.Olricd.ReadRepair,
-		LoadFactor:               c.Olricd.LoadFactor,
-		MemberCountQuorum:        c.Olricd.MemberCountQuorum,
-		Logger:                   log.New(logOutput, "", log.LstdFlags),
-		LogOutput:                logOutput,
-		LogVerbosity:             c.Logging.Verbosity,
-		Hasher:                   hasher.NewDefaultHasher(),
-		KeepAlivePeriod:          keepAlivePeriod,
-		BootstrapTimeout:         bootstrapTimeout,
-		LeaveTimeout:             leaveTimeout,
-		DMaps:                    dmapConfig,
+		BindAddr:                   c.Olricd.BindAddr,
+		BindPort:                   c.Olricd.BindPort,
+		Interface:                  c.Olricd.Interface,
+		ServiceDiscovery:           c.ServiceDiscovery,
+		MemberlistInterface:        c.Memberlist.Interface,
+		MemberlistConfig:           memberlistConfig,
+		Client:                     &clientConfig,
+		LogLevel:                   c.Logging.Level,
+		JoinRetryInterval:          joinRetryInterval,
+		RoutingTablePushInterval:   routingTablePushInterval,
+		TriggerBalancerInterval:    triggerBalancerInterval,
+		EnableClusterEventsChannel: c.Olricd.EnableClusterEventsChannel,
+		MaxJoinAttempts:            c.Memberlist.MaxJoinAttempts,
+		Peers:                      c.Memberlist.Peers,
+		PartitionCount:             c.Olricd.PartitionCount,
+		ReplicaCount:               c.Olricd.ReplicaCount,
+		WriteQuorum:                c.Olricd.WriteQuorum,
+		ReadQuorum:                 c.Olricd.ReadQuorum,
+		ReplicationMode:            c.Olricd.ReplicationMode,
+		ReadRepair:                 c.Olricd.ReadRepair,
+		LoadFactor:                 c.Olricd.LoadFactor,
+		MemberCountQuorum:          c.Olricd.MemberCountQuorum,
+		Logger:                     log.New(logOutput, "", log.LstdFlags),
+		LogOutput:                  logOutput,
+		LogVerbosity:               c.Logging.Verbosity,
+		Hasher:                     hasher.NewDefaultHasher(),
+		KeepAlivePeriod:            keepAlivePeriod,
+		BootstrapTimeout:           bootstrapTimeout,
+		LeaveTimeout:               leaveTimeout,
+		DMaps:                      dmapConfig,
 	}
 
 	if err := cfg.Sanitize(); err != nil {

events.go

@@ -0,0 +1,9 @@
+package olric
+
+import "time"
+
+type ClusterTopologyEvent struct {
+	Event     string
+	Name      string
+	Timestamp time.Time
+}