Refactor of agents: new snapshot sore client, task manager, and agent…

… api

gossip/config.go

@@ -100,14 +100,23 @@ type Config struct {
 	MemberlistConfig *memberlist.Config
 
 	// Comma-delimited list of Alert servers ([host]:port), through which an agent can post alerts
-	AlertsUrls []string
+	AlertsServiceUrls []string
+
+	// QED API Key
+	APIKey string
+
+	// QED endpoint list
+	QEDUrls []string
+
+	// SnapshotStoreUrls contains the http endpoints of the snapshot store
+	SnapshotStoreUrls []string
 
 	// Timeout enqueuing elements on a channel
 	TimeoutQueues time.Duration
 
 	// Interval to send out messages to other agents
 	ProcessInterval time.Duration
-	
+
 	// Address to bind the metrics endpoint
 	MetricsAddr string
 }

gossip/delegate.go

@@ -37,14 +37,14 @@ type eventDelegate struct {
 func (e *eventDelegate) NotifyJoin(n *memberlist.Node) {
 	peer := member.ParsePeer(n)
 	peer.Status = member.Alive
-	e.agent.Topology.Update(peer)
+	e.agent.topology.Update(peer)
 	log.Debugf("member joined: %+v ", peer)
 }
 
 // NotifyLeave is invoked when a node is detected to have left.
 func (e *eventDelegate) NotifyLeave(n *memberlist.Node) {
 	peer := member.ParsePeer(n)
-	e.agent.Topology.Delete(peer)
+	e.agent.topology.Delete(peer)
 	log.Debugf("member left:  %+v", peer)
 }
 
@@ -53,7 +53,7 @@ func (e *eventDelegate) NotifyLeave(n *memberlist.Node) {
 func (e *eventDelegate) NotifyUpdate(n *memberlist.Node) {
 	// ignore
 	peer := member.ParsePeer(n)
-	e.agent.Topology.Update(peer)
+	e.agent.topology.Update(peer)
 	log.Debugf("member updated: %+v ", peer)
 }
 
@@ -71,7 +71,7 @@ func newAgentDelegate(agent *Agent) *agentDelegate {
 // when broadcasting an alive message. It's length is limited to
 // the given byte size. This metadata is available in the Node structure.
 func (d *agentDelegate) NodeMeta(limit int) []byte {
-	meta, err := d.agent.Self.Meta.Encode()
+	meta, err := d.agent.self.Meta.Encode()
 	if err != nil {
 		log.Fatalf("Unable to encode node metadata: %v", err)
 	}
@@ -101,7 +101,7 @@ func (d *agentDelegate) NotifyMsg(msg []byte) {
 	// hashs the snaapshots to deduplicate processing inside the agent
 	hash := hashing.NewSha256Hasher().Do(*tmp["Snapshots"])
 	log.Debugf("Notifying batch %v\n", hash)
-	d.agent.In <- &hashedBatch{&batch, hash}
+	d.agent.inCh <- &hashedBatch{&batch, hash}
 }
 
 // GetBroadcasts is called when user data messages can be broadcast.

gossip/errors.go

@@ -0,0 +1,20 @@
+/*
+   Copyright 2018-2019 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 gossip
+
+import "errors"
+
+var ChTimedOut error = errors.New("Timeout sending data to channel")