elector

README

gocron-etcd-elector

install

go get github.com/go-co-op/gocron-etcd-elector

usage

gocron v2

For a full production environment example with etcd clusters, you can see gocron-etcd-elector-example

Here is an example usage that would be deployed in multiple instances.

package main

import (
	"context"
	"fmt"
	"time"

	elector "github.com/go-co-op/gocron-etcd-elector"
	"github.com/go-co-op/gocron/v2"
)

func main() {
	// Printing some value to verify if container is running in docker logs
	fmt.Println("go-app")

	// Configuring elector with etdc cluster
	cfg := elector.Config{
		Endpoints:   []string{"http://etcd-1:2379", "http://etcd-2:2379", "http://etcd-3:2379"},
		DialTimeout: 3 * time.Second,
	}

	// Build new elector
	el, err := elector.NewElector(context.Background(), cfg, elector.WithTTL(10))
	if err != nil {
		panic(err)
	}

	// el.Start() is a blocking method
	// so running with goroutine
	go func() {
		err := el.Start("/rahul-gandhi/pappu") // specify a directory for storing key value for election
		if err == elector.ErrClosed {
			return
		}
	}()

	// New scheduler with elector
	sh, err := gocron.NewScheduler(gocron.WithDistributedElector(el))
	if err != nil {
		panic(err)
	}

	// The scheduler elected as the leader is only allowed to run, other instances don't execute
	_, err = sh.NewJob(gocron.DurationJob(1*time.Second), gocron.NewTask(func() {
		// This if statement doesn't work as intended as only the leader is running
		// So true always
		if el.IsLeader(context.Background()) == nil {
			fmt.Println("Current instance is leader")
			fmt.Println("executing job")
		} else {
			// To see this log printed remove gocron.WithDistributedElector(el) option from the scheduler
			fmt.Printf("Not leader, current leader is %s\n", el.GetLeaderID())
		}
	}))
	if err != nil {
		panic(err)
	}

	sh.Start()
	fmt.Println("exit")
}

Documentation

Index

• Variables
• type Config
• type Elector
  • func NewElector(ctx context.Context, cfg clientv3.Config, options ...concurrency.SessionOption) (*Elector, error)
  • func NewElectorWithClient(ctx context.Context, cli *clientv3.Client, ...) (*Elector, error)
  • func (e *Elector) GetID() string
  • func (e *Elector) GetLeaderID() string
  • func (e *Elector) IsLeader(_ context.Context) error
  • func (e *Elector) SetLogger(fn func(msg ...interface{}))
  • func (e *Elector) Start(electionPath string) error
  • func (e *Elector) Stop() error

Variables

var (
	ErrNonLeader = errors.New("the elector is not leader")
	ErrClosed    = errors.New("the elector is already closed")
	ErrPingEtcd  = errors.New("ping etcd server timeout")
)

var (
	// alias options
	WithTTL     = concurrency.WithTTL
	WithContext = concurrency.WithContext
	WithLease   = concurrency.WithLease
)

Types

type Config

type Config = clientv3.Config

alias clientv3.config

type Elector

type Elector struct {
	// contains filtered or unexported fields
}

func NewElector

func NewElector(ctx context.Context, cfg clientv3.Config, options ...concurrency.SessionOption) (*Elector, error)

func NewElectorWithClient

func NewElectorWithClient(ctx context.Context, cli *clientv3.Client, options ...concurrency.SessionOption) (*Elector, error)

func (*Elector) GetID

func (e *Elector) GetID() string

func (*Elector) GetLeaderID

func (e *Elector) GetLeaderID() string

func (*Elector) IsLeader

func (e *Elector) IsLeader(_ context.Context) error

func (*Elector) SetLogger

func (e *Elector) SetLogger(fn func(msg ...interface{}))

func (*Elector) Start

func (e *Elector) Start(electionPath string) error

Start Start the election. This method will keep trying the election. When the election is successful, set isleader to true. If it fails, the election directory will be monitored until the election is successful. The parameter electionPath is used to specify the etcd directory for the operation.

func (*Elector) Stop

func (e *Elector) Stop() error