-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Create integration for using a atomic broadcast. (#4)
The current implementation is backed by etcd. Using this approach we should have a primitive that is consistent and totally orders all messages. Along with this changes, added a generic interface to be possible to use different atomic broadcast protocols.
- Loading branch information
Showing
18 changed files
with
1,378 additions
and
377 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -22,14 +22,11 @@ jobs: | |
| - name: Check out code into the Go module directory | ||
| uses: actions/checkout@v2 | ||
|
|
||
| - name: Setup RabbitMQ with username and password | ||
| uses: getong/[email protected] | ||
| with: | ||
| rabbitmq version: '3.8.2-management-alpine' | ||
| host port: 5672 | ||
| rabbitmq user: 'guest' | ||
| rabbitmq password: 'guest' | ||
| rabbitmq vhost: '/' | ||
|
|
||
| - name: All | ||
| run: make ci | ||
| - name: Setup etcd server and make | ||
| run: | | ||
| ETCD_VER=$(curl --silent https://api.github.com/repos/etcd-io/etcd/releases/latest | grep "tag_name" | cut -d ' ' -f4 | awk -F'"' '$0=$2') | ||
| curl -sL https://storage.googleapis.com/etcd/${ETCD_VER}/etcd-${ETCD_VER}-linux-amd64.tar.gz -o ./etcd.tar.gz | ||
| mkdir etcd && tar xzvf etcd.tar.gz -C etcd --strip-components=1 | ||
| etcd/etcd > /dev/null & | ||
| sleep 30 | ||
| make ci | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,31 @@ | ||
| package internal | ||
|
|
||
| import "sync/atomic" | ||
|
|
||
| const ( | ||
| // Constant to represent the `active` state on the Flag. | ||
| active = 0x0 | ||
|
|
||
| // Constant to represent the `inactive` state on the Flag. | ||
| inactive = 0x1 | ||
| ) | ||
|
|
||
| // An atomic boolean implementation, to act specifically as a flag. | ||
| type Flag struct { | ||
| flag int32 | ||
| } | ||
|
|
||
| // Verify if the flag still on `active` state. | ||
| func (f *Flag) IsActive() bool { | ||
| return atomic.LoadInt32(&f.flag) == active | ||
| } | ||
|
|
||
| // Verify if the flag is on `inactive` state. | ||
| func (f *Flag) IsInactive() bool { | ||
| return atomic.LoadInt32(&f.flag) == inactive | ||
| } | ||
|
|
||
| // Transition the flag from `active` to `inactive`. | ||
| func (f *Flag) Inactivate() bool { | ||
| return atomic.CompareAndSwapInt32(&f.flag, active, inactive) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,177 @@ | ||
| package internal | ||
|
|
||
| import ( | ||
| "context" | ||
| "github.com/coreos/etcd/clientv3" | ||
| "io" | ||
| "time" | ||
| ) | ||
|
|
||
| // A single write requests to be applied to etcd. | ||
| type request struct { | ||
| // Issuer writer context. | ||
| ctx context.Context | ||
|
|
||
| // Event to be sent to etcd. | ||
| event Event | ||
|
|
||
| // Channel to send response back. | ||
| response chan error | ||
| } | ||
|
|
||
| // Configuration for the coordinator. | ||
| type CoordinatorConfiguration struct { | ||
| // Each Coordinator will handle only a single partition. | ||
| // This will avoid peers with overlapping partitions. | ||
| Partition string | ||
|
|
||
| // Address for etcd server. | ||
| Server string | ||
|
|
||
| // Parent context that the Coordinator will derive it's own context. | ||
| Ctx context.Context | ||
|
|
||
| // Handler for managing goroutines. | ||
| Handler *GoRoutineHandler | ||
| } | ||
|
|
||
| // Coordinator interface that should be implemented by the | ||
| // atomic broadcast handler. | ||
| // Commands should be issued through the coordinator to be delivered | ||
| // to other peers | ||
| type Coordinator interface { | ||
| io.Closer | ||
|
|
||
| // Watch for changes on the partition. | ||
| // After called, this method will start a new goroutine that only | ||
| // returns when the Coordinator context is done. | ||
| Watch(received chan<- Event) error | ||
|
|
||
| // Issues an Event. | ||
| Write(ctx context.Context, event Event) <-chan error | ||
| } | ||
|
|
||
| // Create a new Coordinator using the given configuration. | ||
| // The current implementation is the EtcdCoordinator, backed by etcd. | ||
| func NewCoordinator(configuration CoordinatorConfiguration) (Coordinator, error) { | ||
| cli, err := clientv3.New(clientv3.Config{ | ||
| DialTimeout: 30 * time.Second, | ||
| Endpoints: []string{configuration.Server}, | ||
| }) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| kv := clientv3.NewKV(cli) | ||
| ctx, cancel := context.WithCancel(configuration.Ctx) | ||
| coord := &EtcdCoordinator{ | ||
| configuration: configuration, | ||
| cli: cli, | ||
| kv: kv, | ||
| ctx: ctx, | ||
| cancel: cancel, | ||
| writeChan: make(chan request), | ||
| } | ||
| configuration.Handler.Spawn(coord.writer) | ||
| return coord, nil | ||
| } | ||
|
|
||
| // EtcdCoordinator will use etcd for atomic broadcast. | ||
| type EtcdCoordinator struct { | ||
| // Configuration parameters. | ||
| configuration CoordinatorConfiguration | ||
|
|
||
| // Current Coordinator context, created from the parent context. | ||
| ctx context.Context | ||
|
|
||
| // Function to cancel the current context. | ||
| cancel context.CancelFunc | ||
|
|
||
| // A client for the etcd server. | ||
| cli *clientv3.Client | ||
|
|
||
| // The key-value entry point for issuing requests. | ||
| kv clientv3.KV | ||
|
|
||
| // Channel to receive write requests. | ||
| writeChan chan request | ||
| } | ||
|
|
||
| // Listen and apply write requests. | ||
| // This will keep running while the application context is available. | ||
| // Receiving commands through the channel will ensure that they are | ||
| // applied synchronously to the etcd. | ||
| func (e *EtcdCoordinator) writer() { | ||
| for { | ||
| select { | ||
| case <-e.ctx.Done(): | ||
| return | ||
| case req := <-e.writeChan: | ||
| _, err := e.kv.Put(req.ctx, req.event.Key, string(req.event.Value)) | ||
| req.response <- err | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Starts a new coroutine for watching the Coordinator partition. | ||
| // All received information will be published back through the channel | ||
| // received as parameter. | ||
| // | ||
| // After calling a routine will run bounded to the application lifetime. | ||
| func (e *EtcdCoordinator) Watch(received chan<- Event) error { | ||
| watchChan := e.cli.Watch(e.ctx, e.configuration.Partition) | ||
| watchChanges := func() { | ||
| for response := range watchChan { | ||
| select { | ||
| case <-e.ctx.Done(): | ||
| return | ||
| default: | ||
| e.handleResponse(response, received) | ||
| } | ||
| } | ||
| } | ||
| e.configuration.Handler.Spawn(watchChanges) | ||
| return nil | ||
| } | ||
|
|
||
| // Write the given event using the KV interface. | ||
| func (e *EtcdCoordinator) Write(ctx context.Context, event Event) <-chan error { | ||
| res := make(chan error) | ||
| e.writeChan <- request{ | ||
| ctx: ctx, | ||
| event: event, | ||
| response: res, | ||
| } | ||
| return res | ||
| } | ||
|
|
||
| // Stop the etcd client connection. | ||
| func (e *EtcdCoordinator) Close() error { | ||
| e.cancel() | ||
| return e.cli.Close() | ||
| } | ||
|
|
||
| // This method is responsible for handling events from the etcd client. | ||
| // | ||
| // This method will transform each received event into Event object and | ||
| // publish it back using the given channel. A buffered channel will be created | ||
| // and a goroutine will be spawned, so we can publish the received messages | ||
| // asynchronously without blocking. This can cause the Close to hold, if there | ||
| // exists pending messages to be consumed by the channel, this method can cause a deadlock. | ||
| func (e *EtcdCoordinator) handleResponse(response clientv3.WatchResponse, received chan<- Event) { | ||
| buffered := make(chan Event, len(response.Events)) | ||
| defer close(buffered) | ||
|
|
||
| e.configuration.Handler.Spawn(func() { | ||
| for ev := range buffered { | ||
| received <- ev | ||
| } | ||
| }) | ||
|
|
||
| for _, event := range response.Events { | ||
| buffered <- Event{ | ||
| Key: string(event.Kv.Key), | ||
| Value: event.Kv.Value, | ||
| Error: nil, | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.