v3election: Election RPC service

Fixes #7589
This commit is contained in:
Anthony Romano 2017-03-30 13:16:39 -07:00
parent 9ba69ff317
commit dc8115a534
4 changed files with 2291 additions and 0 deletions

View File

@ -0,0 +1,16 @@
// Copyright 2017 The etcd Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// Package v3election provides a v3 election service from an etcdserver.
package v3election

View File

@ -0,0 +1,123 @@
// Copyright 2017 The etcd Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package v3election
import (
"golang.org/x/net/context"
"github.com/coreos/etcd/clientv3"
"github.com/coreos/etcd/clientv3/concurrency"
epb "github.com/coreos/etcd/etcdserver/api/v3election/v3electionpb"
)
type electionServer struct {
c *clientv3.Client
}
func NewElectionServer(c *clientv3.Client) epb.ElectionServer {
return &electionServer{c}
}
func (es *electionServer) Campaign(ctx context.Context, req *epb.CampaignRequest) (*epb.CampaignResponse, error) {
s, err := es.session(ctx, req.Lease)
if err != nil {
return nil, err
}
e := concurrency.NewElection(s, string(req.Name))
if err = e.Campaign(ctx, string(req.Value)); err != nil {
return nil, err
}
return &epb.CampaignResponse{
Header: e.Header(),
Leader: &epb.LeaderKey{
Name: req.Name,
Key: []byte(e.Key()),
Rev: e.Rev(),
Lease: int64(s.Lease()),
},
}, nil
}
func (es *electionServer) Proclaim(ctx context.Context, req *epb.ProclaimRequest) (*epb.ProclaimResponse, error) {
s, err := es.session(ctx, req.Leader.Lease)
if err != nil {
return nil, err
}
e := concurrency.ResumeElection(s, string(req.Leader.Name), string(req.Leader.Key), req.Leader.Rev)
if err := e.Proclaim(ctx, string(req.Value)); err != nil {
return nil, err
}
return &epb.ProclaimResponse{Header: e.Header()}, nil
}
func (es *electionServer) Observe(req *epb.LeaderRequest, stream epb.Election_ObserveServer) error {
s, err := es.session(stream.Context(), -1)
if err != nil {
return err
}
e := concurrency.NewElection(s, string(req.Name))
ch := e.Observe(stream.Context())
for stream.Context().Err() == nil {
select {
case <-stream.Context().Done():
case resp, ok := <-ch:
if !ok {
return nil
}
lresp := &epb.LeaderResponse{Header: resp.Header, Kv: resp.Kvs[0]}
if err := stream.Send(lresp); err != nil {
return err
}
}
}
return stream.Context().Err()
}
func (es *electionServer) Leader(ctx context.Context, req *epb.LeaderRequest) (*epb.LeaderResponse, error) {
s, err := es.session(ctx, -1)
if err != nil {
return nil, err
}
l, lerr := concurrency.NewElection(s, string(req.Name)).Leader(ctx)
if lerr != nil {
return nil, lerr
}
return &epb.LeaderResponse{Header: l.Header, Kv: l.Kvs[0]}, nil
}
func (es *electionServer) Resign(ctx context.Context, req *epb.ResignRequest) (*epb.ResignResponse, error) {
s, err := es.session(ctx, req.Leader.Lease)
if err != nil {
return nil, err
}
e := concurrency.ResumeElection(s, string(req.Leader.Name), string(req.Leader.Key), req.Leader.Rev)
if err := e.Resign(ctx); err != nil {
return nil, err
}
return &epb.ResignResponse{Header: e.Header()}, nil
}
func (es *electionServer) session(ctx context.Context, lease int64) (*concurrency.Session, error) {
s, err := concurrency.NewSession(
es.c,
concurrency.WithLease(clientv3.LeaseID(lease)),
concurrency.WithContext(ctx),
)
if err != nil {
return nil, err
}
s.Orphan()
return s, nil
}

File diff suppressed because it is too large Load Diff

View File

@ -0,0 +1,119 @@
syntax = "proto3";
package v3electionpb;
import "gogoproto/gogo.proto";
import "etcd/etcdserver/etcdserverpb/rpc.proto";
import "etcd/mvcc/mvccpb/kv.proto";
// for grpc-gateway
import "google/api/annotations.proto";
option (gogoproto.marshaler_all) = true;
option (gogoproto.unmarshaler_all) = true;
// The election service exposes client-side election facilities as a gRPC interface.
service Election {
// Campaign waits to acquire leadership in an election, returning a LeaderKey
// representing the leadership if successful. The LeaderKey can then be used
// to issue new values on the election, transactionally guard API requests on
// leadership still being held, and resign from the election.
rpc Campaign(CampaignRequest) returns (CampaignResponse) {
option (google.api.http) = {
post: "/v3alpha/election/campaign"
body: "*"
};
}
// Proclaim updates the leader's posted value with a new value.
rpc Proclaim(ProclaimRequest) returns (ProclaimResponse) {
option (google.api.http) = {
post: "/v3alpha/election/proclaim"
body: "*"
};
}
// Leader returns the current election proclamation, if any.
rpc Leader(LeaderRequest) returns (LeaderResponse) {
option (google.api.http) = {
post: "/v3alpha/election/leader"
body: "*"
};
}
// Observe streams election proclamations in-order as made by the election's
// elected leaders.
rpc Observe(LeaderRequest) returns (stream LeaderResponse) {
option (google.api.http) = {
post: "/v3alpha/election/observe"
body: "*"
};
}
// Resign releases election leadership so other campaigners may acquire
// leadership on the election.
rpc Resign(ResignRequest) returns (ResignResponse) {
option (google.api.http) = {
post: "/v3alpha/election/resign"
body: "*"
};
}
}
message CampaignRequest {
// name is the election's identifier for the campaign.
bytes name = 1;
// lease is the ID of the lease attached to leadership of the election. If the
// lease expires or is revoked before resigning leadership, then the
// leadership is transferred to the next campaigner, if any.
int64 lease = 2;
// value is the initial proclaimed value set when the campaigner wins the
// election.
bytes value = 3;
}
message CampaignResponse {
etcdserverpb.ResponseHeader header = 1;
// leader describes the resources used for holding leadereship of the election.
LeaderKey leader = 2;
}
message LeaderKey {
// name is the election identifier that correponds to the leadership key.
bytes name = 1;
// key is an opaque key representing the ownership of the election. If the key
// is deleted, then leadership is lost.
bytes key = 2;
// rev is the creation revision of the key. It can be used to test for ownership
// of an election during transactions by testing the key's creation revision
// matches rev.
int64 rev = 3;
// lease is the lease ID of the election leader.
int64 lease = 4;
}
message LeaderRequest {
// name is the election identifier for the leadership information.
bytes name = 1;
}
message LeaderResponse {
etcdserverpb.ResponseHeader header = 1;
// kv is the key-value pair representing the latest leader update.
mvccpb.KeyValue kv = 2;
}
message ResignRequest {
// leader is the leadership to relinquish by resignation.
LeaderKey leader = 1;
}
message ResignResponse {
etcdserverpb.ResponseHeader header = 1;
}
message ProclaimRequest {
// leader is the leadership hold on the election.
LeaderKey leader = 1;
// value is an update meant to overwrite the leader's current value.
bytes value = 2;
}
message ProclaimResponse {
etcdserverpb.ResponseHeader header = 1;
}