server

Documentation

Overview

This package defines a LRU cache server which supports client-side consistent hashing, TLS (and mTLS), client access via both HTTP/gRPC,

Index

• Constants
• func GetSugaredZapLogger(logFile string, errFile string, verbose bool) *zap.SugaredLogger
• func LoadTLSCredentials(client_auth bool) (credentials.TransportCredentials, error)
• func LoadTlsConfig(client_auth bool) (*tls.Config, error)
• func NewGrpcClientForNode(node *node.Node, clientAuth bool, httpsEnabled bool) pb.CacheServiceClient
• type CacheServer
  • func NewCacheServer(capacity int, configFile string, verbose bool, nodeID string, ...) (*grpc.Server, *CacheServer)
  • func (s *CacheServer) AnnounceNewLeader(winner string)
  • func (s *CacheServer) Get(ctx context.Context, req *pb.GetRequest) (*pb.GetResponse, error)
  • func (s *CacheServer) GetClusterConfig(ctx context.Context, req *pb.ClusterConfigRequest) (*pb.ClusterConfig, error)
  • func (s *CacheServer) GetHandler(c *gin.Context)
  • func (s *CacheServer) GetHeartbeat(ctx context.Context, request *pb.HeartbeatRequest) (*empty.Empty, error)
  • func (s *CacheServer) GetLeader(ctx context.Context, request *pb.LeaderRequest) (*pb.LeaderResponse, error)
  • func (s *CacheServer) GetPid(ctx context.Context, request *pb.PidRequest) (*pb.PidResponse, error)
  • func (s *CacheServer) IsLeaderAlive() bool
  • func (s *CacheServer) LogInfoLevel(msg string)
  • func (s *CacheServer) NewCacheClient(serverHost string, serverPort int) (pb.CacheServiceClient, error)
  • func (s *CacheServer) Put(ctx context.Context, req *pb.PutRequest) (*empty.Empty, error)
  • func (s *CacheServer) PutHandler(c *gin.Context)
  • func (s *CacheServer) RegisterNodeInternal()
  • func (s *CacheServer) RegisterNodeWithCluster(ctx context.Context, nodeInfo *pb.Node) (*pb.GenericResponse, error)
  • func (s *CacheServer) RequestElection(ctx context.Context, request *pb.ElectionRequest) (*pb.GenericResponse, error)
  • func (s *CacheServer) RunAndReturnHTTPServer(port int) *http.Server
  • func (s *CacheServer) RunElection()
  • func (s *CacheServer) StartLeaderHeartbeatMonitor()
  • func (s *CacheServer) UpdateClusterConfig(ctx context.Context, req *pb.ClusterConfig) (*empty.Empty, error)
  • func (s *CacheServer) UpdateLeader(ctx context.Context, request *pb.NewLeaderAnnouncement) (*pb.GenericResponse, error)
• type Pair
• type ServerComponents
  • func CreateAndRunAllFromConfig(capacity int, configFile string, verbose bool, insecureHTTP bool) []ServerComponents

Constants

const (
	ELECTION_RUNNING    = true
	NO_ELECTION_RUNNING = false
	NO_LEADER           = "NO LEADER"
)

const (
	PROD_DB = 0
	TEST_DB = 1
	SUCCESS = "OK"
)

const (
	DYNAMIC = "DYNAMIC"
)

Functions

func GetSugaredZapLogger

func GetSugaredZapLogger(logFile string, errFile string, verbose bool) *zap.SugaredLogger

Set up logger at the specified verbosity level

func LoadTLSCredentials

func LoadTLSCredentials(client_auth bool) (credentials.TransportCredentials, error)

Set up mutual TLS config and credentials

func LoadTlsConfig

func LoadTlsConfig(client_auth bool) (*tls.Config, error)

Set up mutual TLS config

func NewGrpcClientForNode

func NewGrpcClientForNode(node *node.Node, clientAuth bool, httpsEnabled bool) pb.CacheServiceClient

New gRPC client for a server node

Types

type CacheServer

type CacheServer struct {
	pb.UnimplementedCacheServiceServer
	// contains filtered or unexported fields
}

func NewCacheServer

func NewCacheServer(capacity int, configFile string, verbose bool, nodeID string, httpsEnabled bool, clientAuth bool) (*grpc.Server, *CacheServer)

Utility function for creating a new gRPC server secured with mTLS, and registering a cache server service with it. Set node_id param to DYNAMIC to dynamically discover node id. Otherwise, manually set it to a valid nodeID from the config file. Returns tuple of (gRPC server instance, registered Cache CacheServer instance).

func (*CacheServer) AnnounceNewLeader

func (s *CacheServer) AnnounceNewLeader(winner string)

Announce new leader to all nodes

func (*CacheServer) Get

func (s *CacheServer) Get(ctx context.Context, req *pb.GetRequest) (*pb.GetResponse, error)

gRPC handler for getting item from cache. Any replica in the group can serve read requests.

func (*CacheServer) GetClusterConfig

func (s *CacheServer) GetClusterConfig(ctx context.Context, req *pb.ClusterConfigRequest) (*pb.ClusterConfig, error)

gRPC handler for getting cluster config

func (*CacheServer) GetHandler

func (s *CacheServer) GetHandler(c *gin.Context)

GET /get/:key REST API endpoint to get value for key from the LRU cache

func (*CacheServer) GetHeartbeat

func (s *CacheServer) GetHeartbeat(ctx context.Context, request *pb.HeartbeatRequest) (*empty.Empty, error)

Return current status of this node (leader/follower)

func (*CacheServer) GetLeader

func (s *CacheServer) GetLeader(ctx context.Context, request *pb.LeaderRequest) (*pb.LeaderResponse, error)

Returns current leader

func (*CacheServer) GetPid

func (s *CacheServer) GetPid(ctx context.Context, request *pb.PidRequest) (*pb.PidResponse, error)

gRPC handler that receives a request with the caller's PID and returns its own PID. If the PID is higher than the caller PID, we take over the election process.

func (*CacheServer) IsLeaderAlive

func (s *CacheServer) IsLeaderAlive() bool

Check if leader node is alive (3 second timeout)

func (*CacheServer) LogInfoLevel

func (s *CacheServer) LogInfoLevel(msg string)

Log function that can be called externally

func (*CacheServer) NewCacheClient

func (s *CacheServer) NewCacheClient(serverHost string, serverPort int) (pb.CacheServiceClient, error)

Utility function to get a new Cache Client which uses gRPC secured with mTLS

func (*CacheServer) Put

func (s *CacheServer) Put(ctx context.Context, req *pb.PutRequest) (*empty.Empty, error)

gRPC handler for putting item in cache.

func (*CacheServer) PutHandler

func (s *CacheServer) PutHandler(c *gin.Context)

POST /put (Body: {"key": "1", "value": "2"}) REST API endpoint to put a new key-value pair in the LRU cache

func (*CacheServer) RegisterNodeInternal

func (s *CacheServer) RegisterNodeInternal()

Register node with the cluster. This is a function to be called internally by server code (as opposed to the gRPC handler to register node, which is what receives the RPC sent by this function).

func (*CacheServer) RegisterNodeWithCluster

func (s *CacheServer) RegisterNodeWithCluster(ctx context.Context, nodeInfo *pb.Node) (*pb.GenericResponse, error)

gRPC handler for registering a new node with the cluster. New nodes call this RPC on the leader when they come online.

func (*CacheServer) RequestElection

func (s *CacheServer) RequestElection(ctx context.Context, request *pb.ElectionRequest) (*pb.GenericResponse, error)

gRPC handler which allows other nodes to ask this node to start a new election

func (*CacheServer) RunAndReturnHTTPServer

func (s *CacheServer) RunAndReturnHTTPServer(port int) *http.Server

func (*CacheServer) RunElection

func (s *CacheServer) RunElection()

Run an election using the Bully Algorithm (https://en.wikipedia.org/wiki/Bully_algorithm)

func (*CacheServer) StartLeaderHeartbeatMonitor

func (s *CacheServer) StartLeaderHeartbeatMonitor()

Checks if leader is alive every 1 second. If no response for 3 seconds, new election is held.

func (*CacheServer) UpdateClusterConfig

func (s *CacheServer) UpdateClusterConfig(ctx context.Context, req *pb.ClusterConfig) (*empty.Empty, error)

gRPC handler for updating cluster config with incoming info

func (*CacheServer) UpdateLeader

func (s *CacheServer) UpdateLeader(ctx context.Context, request *pb.NewLeaderAnnouncement) (*pb.GenericResponse, error)

gRPC handler for updating the leader after

type Pair

type Pair struct {
	Key   string `json:"key"`
	Value string `json:"value"`
}

type ServerComponents

type ServerComponents struct {
	GrpcServer *grpc.Server
	HttpServer *http.Server
}

func CreateAndRunAllFromConfig

func CreateAndRunAllFromConfig(capacity int, configFile string, verbose bool, insecureHTTP bool) []ServerComponents

Create and run all servers defined in config file and return list of server components