-
Notifications
You must be signed in to change notification settings - Fork 19
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
WIP: Integrate gossip agent into QED. Hungs in raftwal tests
Co-authored-by: Gabriel Díaz <g.diaz.lopezllave@bbva.com>
- Loading branch information
Showing
10 changed files
with
315 additions
and
20 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
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,114 @@ | ||
| /* | ||
| Copyright 2018 Banco Bilbao Vizcaya Argentaria, S.A. | ||
| 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 sender | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "fmt" | ||
| "time" | ||
|
|
||
| "github.com/bbva/qed/gossip" | ||
| "github.com/bbva/qed/log" | ||
| "github.com/bbva/qed/protocol" | ||
| "github.com/bbva/qed/sign" | ||
| "github.com/hashicorp/go-msgpack/codec" | ||
| ) | ||
|
|
||
| type Config struct { | ||
| BatchSize uint | ||
| BatchInterval time.Duration | ||
| TTL int | ||
| } | ||
|
|
||
| func DefaultConfig() *Config { | ||
| return &Config{ | ||
| 100, | ||
| 1 * time.Second, | ||
| 2, | ||
| } | ||
| } | ||
|
|
||
| func Start(n *gossip.Node, ch chan *protocol.Snapshot) { | ||
| ticker := time.NewTicker(1 * time.Second) | ||
|
|
||
| for { | ||
| select { | ||
| case <-ticker.C: | ||
| msg, _ := encode(getBatch(ch)) | ||
|
|
||
| peers := n.GetPeers(1, gossip.AuditorType) | ||
| peers = append(peers, n.GetPeers(1, gossip.MonitorType)...) | ||
| peers = append(peers, n.GetPeers(1, gossip.PublisherType)...) | ||
|
|
||
| for _, peer := range peers { | ||
| err := n.Memberlist().SendReliable(peer.Node, msg) | ||
| if err != nil { | ||
| log.Errorf("Failed send message: %v", err) | ||
| } | ||
| } | ||
| // TODO: Implement graceful shutdown. | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func encode(msg protocol.BatchSnapshots) ([]byte, error) { | ||
| var buf bytes.Buffer | ||
| encoder := codec.NewEncoder(&buf, &codec.MsgpackHandle{}) | ||
| if err := encoder.Encode(msg); err != nil { | ||
| log.Errorf("Failed to encode message: %v", err) | ||
| return nil, err | ||
| } | ||
| return buf.Bytes(), nil | ||
| } | ||
|
|
||
| func getBatch(ch chan *protocol.Snapshot) protocol.BatchSnapshots { | ||
|
|
||
| var snapshot *protocol.Snapshot | ||
| var batch protocol.BatchSnapshots | ||
| var batchSize int = 100 | ||
| var counter int = 0 | ||
| batch.Snapshots = make([]*protocol.SignedSnapshot, 0) | ||
| batch.TTL = 3 | ||
|
|
||
| for { | ||
| select { | ||
| case snapshot = <-ch: | ||
| counter++ | ||
| default: | ||
| return batch | ||
| } | ||
|
|
||
| ss, err := doSign(sign.NewEd25519Signer(), snapshot) | ||
| if err != nil { | ||
| log.Errorf("Failed signing message: %v", err) | ||
| } | ||
| batch.Snapshots = append(batch.Snapshots, ss) | ||
|
|
||
| if counter == batchSize { | ||
| return batch | ||
| } | ||
|
|
||
| } | ||
|
|
||
| } | ||
|
|
||
| func doSign(signer sign.Signer, snapshot *protocol.Snapshot) (*protocol.SignedSnapshot, error) { | ||
|
|
||
| signature, err := signer.Sign([]byte(fmt.Sprintf("%v", snapshot))) | ||
| if err != nil { | ||
| fmt.Println("Publisher: error signing commitment") | ||
| return nil, err | ||
| } | ||
| return &protocol.SignedSnapshot{snapshot, signature}, nil | ||
| } |
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,117 @@ | ||
| /* | ||
| Copyright 2018 Banco Bilbao Vizcaya Argentaria, S.A. | ||
| 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 protocol | ||
|
|
||
| import ( | ||
| "github.com/bbva/qed/balloon" | ||
| "github.com/bbva/qed/balloon/history" | ||
| "github.com/bbva/qed/balloon/hyper" | ||
| "github.com/bbva/qed/balloon/visitor" | ||
| "github.com/bbva/qed/hashing" | ||
| "github.com/bbva/qed/util" | ||
| ) | ||
|
|
||
| // Event is the public struct that Add handler function uses to | ||
| // parse the post params. | ||
| type Event struct { | ||
| Event []byte | ||
| } | ||
|
|
||
| // MembershipQuery is the public struct that apihttp.Membership | ||
| // Handler uses to parse the post params. | ||
| type MembershipQuery struct { | ||
| Key []byte | ||
| Version uint64 | ||
| } | ||
|
|
||
| // Snapshot is the public struct that apihttp.Add Handler call returns. | ||
| type Snapshot struct { | ||
| HistoryDigest hashing.Digest | ||
| HyperDigest hashing.Digest | ||
| Version uint64 | ||
| EventDigest hashing.Digest | ||
| } | ||
|
|
||
| type SignedSnapshot struct { | ||
| Snapshot *Snapshot | ||
| Signature []byte | ||
| } | ||
|
|
||
| type BatchSnapshots struct { | ||
| Snapshots []*SignedSnapshot | ||
| TTL int | ||
| } | ||
|
|
||
| type MembershipResult struct { | ||
| Exists bool | ||
| Hyper visitor.AuditPath | ||
| History visitor.AuditPath | ||
| CurrentVersion uint64 | ||
| QueryVersion uint64 | ||
| ActualVersion uint64 | ||
| KeyDigest hashing.Digest | ||
| Key []byte | ||
| } | ||
|
|
||
| type IncrementalRequest struct { | ||
| Start uint64 | ||
| End uint64 | ||
| } | ||
|
|
||
| type IncrementalResponse struct { | ||
| Start uint64 | ||
| End uint64 | ||
| AuditPath visitor.AuditPath | ||
| } | ||
|
|
||
| // ToMembershipProof translates internal api balloon.MembershipProof to the | ||
| // public struct protocol.MembershipResult. | ||
| func ToMembershipResult(key []byte, mp *balloon.MembershipProof) *MembershipResult { | ||
| return &MembershipResult{ | ||
| mp.Exists, | ||
| mp.HyperProof.AuditPath(), | ||
| mp.HistoryProof.AuditPath(), | ||
| mp.CurrentVersion, | ||
| mp.QueryVersion, | ||
| mp.ActualVersion, | ||
| mp.KeyDigest, | ||
| key, | ||
| } | ||
| } | ||
|
|
||
| // ToBaloonProof translate public protocol.MembershipResult:w to internal | ||
| // balloon.Proof. | ||
| func ToBalloonProof(id []byte, mr *MembershipResult, hasherF func() hashing.Hasher) *balloon.MembershipProof { | ||
|
|
||
| historyProof := history.NewMembershipProof(mr.ActualVersion, mr.QueryVersion, mr.History, hasherF()) | ||
| hyperProof := hyper.NewQueryProof(mr.KeyDigest, util.Uint64AsBytes(mr.ActualVersion), mr.Hyper, hasherF()) | ||
|
|
||
| return balloon.NewMembershipProof(mr.Exists, hyperProof, historyProof, mr.CurrentVersion, mr.ActualVersion, mr.QueryVersion, mr.KeyDigest, hasherF()) | ||
|
|
||
| } | ||
|
|
||
| func ToIncrementalResponse(proof *balloon.IncrementalProof) *IncrementalResponse { | ||
| return &IncrementalResponse{ | ||
| proof.Start, | ||
| proof.End, | ||
| proof.AuditPath, | ||
| } | ||
| } | ||
|
|
||
| func ToIncrementalProof(ir *IncrementalResponse, hasher hashing.Hasher) *balloon.IncrementalProof { | ||
| return balloon.NewIncrementalProof(ir.Start, ir.End, ir.AuditPath, hasher) | ||
| } |
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
Oops, something went wrong.