From e311b94b9e8359eeadc7c683a25bd78a31d5a1d7 Mon Sep 17 00:00:00 2001
From: Fabian Seidl <fabian.b.seidl@stud.h-da.de>
Date: Fri, 14 Oct 2022 11:13:05 +0000
Subject: [PATCH] Resolve "Error handling in event publishing via entity
services could be improved"
See merge request danet/gosdn!370
Co-authored-by: Fabian Seidl <fabian.seidl@h-da.de>
---
controller/eventService/Service.go | 144 +++++++++++++++++-
controller/eventService/utils.go | 12 ++
controller/interfaces/event/service.go | 2 +
controller/mocks/Service.go | 14 ++
controller/nucleus/networkElementService.go | 48 +++++-
controller/nucleus/sbiService.go | 24 ++-
controller/rbac/rbacService.go | 73 +++++++--
controller/topology/nodes/nodeService.go | 36 ++++-
controller/topology/ports/portService.go | 36 ++++-
.../routing-tables/routingTableService.go | 36 ++++-
controller/topology/topologyService.go | 36 ++++-
go.mod | 1 +
go.sum | 6 +-
13 files changed, 408 insertions(+), 60 deletions(-)
diff --git a/controller/eventService/Service.go b/controller/eventService/Service.go
index 871065515..7c822fc8a 100644
--- a/controller/eventService/Service.go
+++ b/controller/eventService/Service.go
@@ -3,6 +3,7 @@ package eventservice
import (
"context"
"encoding/json"
+ "time"
"code.fbi.h-da.de/danet/gosdn/controller/config"
"code.fbi.h-da.de/danet/gosdn/controller/customerrs"
@@ -11,13 +12,37 @@ import (
interfaces "code.fbi.h-da.de/danet/gosdn/controller/interfaces/event"
amqp "github.com/rabbitmq/amqp091-go"
+ "github.com/sethvargo/go-retry"
log "github.com/sirupsen/logrus"
)
+const (
+ // Duration to try reconnecting until cancelation.
+ reconnectDelay = 5 * time.Second
+
+ // Duration to try re-initializing a channel.
+ reInitDelay = 2 * time.Second
+
+ // Max number of tries to reconnect to Broker.
+ reconnectAttempts = 5
+
+ // Max number of tries to re-initialize a channel. As re-init is part of the reconnect process this multiplies
+ // with number of reconnectAttempts.
+ reInitAttempts = 5
+
+ // Max number of tries to republish a failed message.
+ rePublishAttempts = 5
+)
+
// EventService is used to setup a connection to a broker and publish events to topics.
type EventService struct {
- connection *amqp.Connection
- channel *amqp.Channel
+ connectionAddr string
+ connection *amqp.Connection
+ channel *amqp.Channel
+ done chan bool
+ notifyConnClose chan *amqp.Error
+ notifyChanClose chan *amqp.Error
+ isReady bool
}
// NewEventService creates a new connection to the broker and opens a channel for later usage.
@@ -27,20 +52,108 @@ func NewEventService() (interfaces.Service, error) {
return NewMockEventService(), nil
}
- conn, err := amqp.Dial(amqpURIBuilder(config.AMQPPrefix, config.AMQPUser, config.AMQPPassword, config.AMQPHost, config.AMQPPort))
+ addr := amqpURIBuilder(config.AMQPPrefix, config.AMQPUser, config.AMQPPassword, config.AMQPHost, config.AMQPPort)
+
+ conn, err := connect(addr)
+ if err != nil {
+ return nil, err
+ }
+
+ notifyConnClose := make(chan *amqp.Error)
+ conn.NotifyClose(notifyConnClose)
+
+ ch, err := initChannel(conn)
+ if err != nil {
+ return nil, err
+ }
+ notifyChanClose := make(chan *amqp.Error)
+ ch.NotifyClose(notifyChanClose)
+
+ return &EventService{
+ connectionAddr: addr,
+ connection: conn,
+ channel: ch,
+ done: make(chan bool),
+ notifyConnClose: notifyConnClose,
+ notifyChanClose: notifyChanClose,
+ }, nil
+}
+
+func connect(addr string) (*amqp.Connection, error) {
+ conn, err := amqp.Dial(addr)
if err != nil {
return nil, &customerrs.AMQPInitFailError{Action: "failed to connect to RabbitMQ", Err: err}
}
+ return conn, nil
+}
+
+// Reconnect attempts to setup a new connection to the RabbitMQ server after an disconnect.
+func (e *EventService) Reconnect() {
+ i := 1
+ for {
+ e.isReady = false
+
+ conn, err := connect(e.connectionAddr)
+
+ if err != nil {
+ select {
+ case <-e.done:
+ return
+ case <-time.After(reconnectDelay):
+ }
+ continue
+ }
+
+ done := e.handleReInit(conn)
+ if done || i == reconnectAttempts {
+ break
+ }
+ i++
+ }
+}
+
+func (e *EventService) handleReInit(conn *amqp.Connection) bool {
+ i := 1
+ for {
+ e.isReady = false
+
+ ch, err := initChannel(conn)
+ if err != nil {
+ select {
+ case <-e.done:
+ return true
+ case <-time.After(reInitDelay):
+ i++
+ }
+ if i == reInitAttempts {
+ return false
+ }
+ continue
+ }
+
+ notifyConnClose := make(chan *amqp.Error)
+ conn.NotifyClose(notifyConnClose)
+ e.notifyConnClose = notifyConnClose
+ e.connection = conn
+
+ notifyChanClose := make(chan *amqp.Error)
+ ch.NotifyClose(notifyChanClose)
+ e.notifyChanClose = notifyChanClose
+ e.channel = ch
+ e.isReady = true
+
+ return true
+ }
+}
+
+func initChannel(conn *amqp.Connection) (*amqp.Channel, error) {
ch, err := conn.Channel()
if err != nil {
return nil, &customerrs.AMQPInitFailError{Action: "failed to open a channel", Err: err}
}
- return &EventService{
- connection: conn,
- channel: ch,
- }, nil
+ return ch, nil
}
// PublishEvent declares a queue and publishes events.
@@ -79,6 +192,23 @@ func (e *EventService) PublishEvent(topic string, event event.Event) error {
return nil
}
+// RetryPublish is used to retry publishing an event after a failed attempt.
+func (e *EventService) RetryPublish(topic string, event event.Event) error {
+ ctx := context.Background()
+ backOff := retry.NewFibonacci(time.Second)
+
+ if err := retry.Do(ctx, retry.WithMaxRetries(rePublishAttempts, backOff), func(ctx context.Context) error {
+ if pubErr := e.PublishEvent(topic, event); pubErr != nil {
+ return retry.RetryableError(&customerrs.AMQPMessageFailError{Action: "retrying to publish message failed", Err: pubErr})
+ }
+ return nil
+ }); err != nil {
+ return err
+ }
+
+ return nil
+}
+
// CloseConnection closes an exisiting connection.
func (e *EventService) CloseConnection() {
if err := e.connection.Close(); err != nil {
diff --git a/controller/eventService/utils.go b/controller/eventService/utils.go
index 891142fef..12bb8ef48 100644
--- a/controller/eventService/utils.go
+++ b/controller/eventService/utils.go
@@ -33,3 +33,15 @@ func (e *MockEventService) PublishEvent(topic string, event event.Event) error {
// CloseConnection closes an exisiting connection.
func (e *MockEventService) CloseConnection() {
}
+
+// RetryPublish declares a queue and publishes events.
+func (e *MockEventService) RetryPublish(topic string, event event.Event) error {
+ e.Queue[topic] = append(e.Queue[topic], event)
+
+ return nil
+}
+
+// Reconnect reconnects to the RabbitMQ server.
+func (e *MockEventService) Reconnect() {
+ // do nothing
+}
diff --git a/controller/interfaces/event/service.go b/controller/interfaces/event/service.go
index 73f3c97d2..34b7e26fc 100644
--- a/controller/interfaces/event/service.go
+++ b/controller/interfaces/event/service.go
@@ -6,4 +6,6 @@ import "code.fbi.h-da.de/danet/gosdn/controller/event"
type Service interface {
PublishEvent(topic string, event event.Event) error
CloseConnection()
+ RetryPublish(topic string, event event.Event) error
+ Reconnect()
}
diff --git a/controller/mocks/Service.go b/controller/mocks/Service.go
index 55386e172..de81b8be6 100644
--- a/controller/mocks/Service.go
+++ b/controller/mocks/Service.go
@@ -32,6 +32,20 @@ func (_m *Service) PublishEvent(topic string, _a1 controllerevent.Event) error {
return r0
}
+// RetryPublish provides a mock function with given fields: topic, _a1
+func (_m *Service) RetryPublish(topic string, _a1 controllerevent.Event) error {
+ ret := _m.Called(topic, _a1)
+
+ var r0 error
+ if rf, ok := ret.Get(0).(func(string, controllerevent.Event) error); ok {
+ r0 = rf(topic, _a1)
+ } else {
+ r0 = ret.Error(0)
+ }
+
+ return r0
+}
+
type mockConstructorTestingTNewService interface {
mock.TestingT
Cleanup(func())
diff --git a/controller/nucleus/networkElementService.go b/controller/nucleus/networkElementService.go
index 19e6a9b23..b2af03f9a 100644
--- a/controller/nucleus/networkElementService.go
+++ b/controller/nucleus/networkElementService.go
@@ -98,8 +98,16 @@ func (s *NetworkElementService) Add(networkElementToAdd networkelement.NetworkEl
return err
}
- if err := s.eventService.PublishEvent(NetworkElementEventTopic, event.NewAddEvent(networkElementToAdd.ID())); err != nil {
- log.Error(err)
+ pubEvent := event.NewAddEvent(networkElementToAdd.ID())
+ if err := s.eventService.PublishEvent(NetworkElementEventTopic, pubEvent); err != nil {
+ go func() {
+ s.eventService.Reconnect()
+
+ retryErr := s.eventService.RetryPublish(NetworkElementEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return nil
@@ -129,8 +137,16 @@ func (s *NetworkElementService) UpdateModel(networkElementToUpdate networkelemen
return err
}
- if err := s.eventService.PublishEvent(NetworkElementEventTopic, event.NewUpdateEvent(networkElementToUpdate.ID())); err != nil {
- log.Error(err)
+ pubEvent := event.NewUpdateEvent(networkElementToUpdate.ID())
+ if err := s.eventService.PublishEvent(NetworkElementEventTopic, pubEvent); err != nil {
+ go func() {
+ s.eventService.Reconnect()
+
+ retryErr := s.eventService.RetryPublish(NetworkElementEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return nil
@@ -143,8 +159,16 @@ func (s *NetworkElementService) Update(networkElementToUpdate networkelement.Net
return err
}
- if err := s.eventService.PublishEvent(NetworkElementEventTopic, event.NewUpdateEvent(networkElementToUpdate.ID())); err != nil {
- log.Error(err)
+ pubEvent := event.NewUpdateEvent(networkElementToUpdate.ID())
+ if err := s.eventService.PublishEvent(NetworkElementEventTopic, pubEvent); err != nil {
+ go func() {
+ s.eventService.Reconnect()
+
+ retryErr := s.eventService.RetryPublish(NetworkElementEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return nil
@@ -163,8 +187,16 @@ func (s *NetworkElementService) Delete(networkElementToDelete networkelement.Net
}
}
- if err := s.eventService.PublishEvent(NetworkElementEventTopic, event.NewDeleteEvent(networkElementToDelete.ID())); err != nil {
- log.Error(err)
+ pubEvent := event.NewDeleteEvent(networkElementToDelete.ID())
+ if err := s.eventService.PublishEvent(NetworkElementEventTopic, pubEvent); err != nil {
+ go func() {
+ s.eventService.Reconnect()
+
+ retryErr := s.eventService.RetryPublish(NetworkElementEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return nil
diff --git a/controller/nucleus/sbiService.go b/controller/nucleus/sbiService.go
index 1b68b9017..fbf9ab5c6 100644
--- a/controller/nucleus/sbiService.go
+++ b/controller/nucleus/sbiService.go
@@ -73,8 +73,16 @@ func (s *SbiService) Add(sbiToAdd southbound.SouthboundInterface) error {
return err
}
- if err := s.eventService.PublishEvent(SbiEventTopic, event.NewAddEvent(sbiToAdd.ID())); err != nil {
- log.Error(err)
+ pubEvent := event.NewAddEvent(sbiToAdd.ID())
+ if err := s.eventService.PublishEvent(SbiEventTopic, pubEvent); err != nil {
+ go func() {
+ s.eventService.Reconnect()
+
+ retryErr := s.eventService.RetryPublish(SbiEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return nil
@@ -94,8 +102,16 @@ func (s *SbiService) Delete(sbiToDelete southbound.SouthboundInterface) error {
}
}
- if err := s.eventService.PublishEvent(SbiEventTopic, event.NewDeleteEvent(sbiToDelete.ID())); err != nil {
- log.Error(err)
+ pubEvent := event.NewDeleteEvent(sbiToDelete.ID())
+ if err := s.eventService.PublishEvent(SbiEventTopic, pubEvent); err != nil {
+ go func() {
+ s.eventService.Reconnect()
+
+ retryErr := s.eventService.RetryPublish(SbiEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return nil
diff --git a/controller/rbac/rbacService.go b/controller/rbac/rbacService.go
index a2b4a288c..414ad1884 100644
--- a/controller/rbac/rbacService.go
+++ b/controller/rbac/rbacService.go
@@ -40,8 +40,16 @@ func (s *UserService) Add(userToAdd rbac.User) error {
return err
}
- if err := s.eventService.PublishEvent(UserEventTopic, event.NewAddEvent(userToAdd.ID())); err != nil {
- log.Error(err)
+ pubEvent := event.NewAddEvent(userToAdd.ID())
+ if err := s.eventService.PublishEvent(UserEventTopic, pubEvent); err != nil {
+ go func() {
+ s.eventService.Reconnect()
+
+ retryErr := s.eventService.RetryPublish(UserEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return nil
@@ -54,8 +62,16 @@ func (s *UserService) Delete(userToDelete rbac.User) error {
return err
}
- if err := s.eventService.PublishEvent(UserEventTopic, event.NewDeleteEvent(userToDelete.ID())); err != nil {
- log.Error(err)
+ pubEvent := event.NewDeleteEvent(userToDelete.ID())
+ if err := s.eventService.PublishEvent(UserEventTopic, pubEvent); err != nil {
+ go func() {
+ s.eventService.Reconnect()
+
+ retryErr := s.eventService.RetryPublish(UserEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return nil
@@ -68,8 +84,16 @@ func (s *UserService) Update(userToUpdate rbac.User) error {
return err
}
- if err := s.eventService.PublishEvent(UserEventTopic, event.NewUpdateEvent(userToUpdate.ID())); err != nil {
- log.Error(err)
+ pubEvent := event.NewUpdateEvent(userToUpdate.ID())
+ if err := s.eventService.PublishEvent(UserEventTopic, pubEvent); err != nil {
+ go func() {
+ s.eventService.Reconnect()
+
+ retryErr := s.eventService.RetryPublish(UserEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return nil
@@ -126,8 +150,16 @@ func (s *RoleService) Add(roleToAdd rbac.Role) error {
return err
}
- if err := s.eventService.PublishEvent(RoleEventTopic, event.NewAddEvent(roleToAdd.ID())); err != nil {
- log.Error(err)
+ pubEvent := event.NewAddEvent(roleToAdd.ID())
+ if err := s.eventService.PublishEvent(RoleEventTopic, pubEvent); err != nil {
+ go func() {
+ s.eventService.Reconnect()
+
+ retryErr := s.eventService.RetryPublish(RoleEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return nil
@@ -140,10 +172,17 @@ func (s *RoleService) Delete(roleToDelete rbac.Role) error {
return err
}
- if err := s.eventService.PublishEvent(RoleEventTopic, event.NewDeleteEvent(roleToDelete.ID())); err != nil {
- log.Error(err)
- }
+ pubEvent := event.NewDeleteEvent(roleToDelete.ID())
+ if err := s.eventService.PublishEvent(RoleEventTopic, pubEvent); err != nil {
+ go func() {
+ s.eventService.Reconnect()
+ retryErr := s.eventService.RetryPublish(RoleEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
+ }
return nil
}
@@ -154,8 +193,16 @@ func (s *RoleService) Update(roleToUpdate rbac.Role) error {
return err
}
- if err := s.eventService.PublishEvent(RoleEventTopic, event.NewUpdateEvent(roleToUpdate.ID())); err != nil {
- log.Error(err)
+ pubEvent := event.NewUpdateEvent(roleToUpdate.ID())
+ if err := s.eventService.PublishEvent(RoleEventTopic, pubEvent); err != nil {
+ go func() {
+ s.eventService.Reconnect()
+
+ retryErr := s.eventService.RetryPublish(RoleEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return nil
diff --git a/controller/topology/nodes/nodeService.go b/controller/topology/nodes/nodeService.go
index 76e2de209..518626c15 100644
--- a/controller/topology/nodes/nodeService.go
+++ b/controller/topology/nodes/nodeService.go
@@ -60,8 +60,16 @@ func (n *NodeService) createNode(node Node) (Node, error) {
return node, err
}
- if err := n.eventService.PublishEvent(NodeEventTopic, event.NewAddEvent(node.ID)); err != nil {
- log.Error(err)
+ pubEvent := event.NewAddEvent(node.ID)
+ if err := n.eventService.PublishEvent(NodeEventTopic, pubEvent); err != nil {
+ go func() {
+ n.eventService.Reconnect()
+
+ retryErr := n.eventService.RetryPublish(NodeEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return node, nil
@@ -74,8 +82,16 @@ func (n *NodeService) Update(node Node) error {
return err
}
- if err := n.eventService.PublishEvent(NodeEventTopic, event.NewUpdateEvent(node.ID)); err != nil {
- log.Error(err)
+ pubEvent := event.NewUpdateEvent(node.ID)
+ if err := n.eventService.PublishEvent(NodeEventTopic, pubEvent); err != nil {
+ go func() {
+ n.eventService.Reconnect()
+
+ retryErr := n.eventService.RetryPublish(NodeEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return nil
@@ -88,8 +104,16 @@ func (n *NodeService) Delete(node Node) error {
return err
}
- if err := n.eventService.PublishEvent(NodeEventTopic, event.NewDeleteEvent(node.ID)); err != nil {
- log.Error(err)
+ pubEvent := event.NewDeleteEvent(node.ID)
+ if err := n.eventService.PublishEvent(NodeEventTopic, pubEvent); err != nil {
+ go func() {
+ n.eventService.Reconnect()
+
+ retryErr := n.eventService.RetryPublish(NodeEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return nil
diff --git a/controller/topology/ports/portService.go b/controller/topology/ports/portService.go
index 1625155a0..25bfc09c6 100644
--- a/controller/topology/ports/portService.go
+++ b/controller/topology/ports/portService.go
@@ -58,8 +58,16 @@ func (p *PortService) createPort(port Port) (Port, error) {
return port, err
}
- if err := p.eventService.PublishEvent(PortEventTopic, event.NewAddEvent(port.ID)); err != nil {
- log.Error(err)
+ pubEvent := event.NewAddEvent(port.ID)
+ if err := p.eventService.PublishEvent(PortEventTopic, pubEvent); err != nil {
+ go func() {
+ p.eventService.Reconnect()
+
+ retryErr := p.eventService.RetryPublish(PortEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return port, nil
@@ -72,8 +80,16 @@ func (p *PortService) Update(port Port) error {
return err
}
- if err := p.eventService.PublishEvent(PortEventTopic, event.NewUpdateEvent(port.ID)); err != nil {
- log.Error(err)
+ pubEvent := event.NewUpdateEvent(port.ID)
+ if err := p.eventService.PublishEvent(PortEventTopic, pubEvent); err != nil {
+ go func() {
+ p.eventService.Reconnect()
+
+ retryErr := p.eventService.RetryPublish(PortEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return nil
@@ -86,8 +102,16 @@ func (p *PortService) Delete(port Port) error {
return err
}
- if err := p.eventService.PublishEvent(PortEventTopic, event.NewDeleteEvent(port.ID)); err != nil {
- log.Error(err)
+ pubEvent := event.NewDeleteEvent(port.ID)
+ if err := p.eventService.PublishEvent(PortEventTopic, pubEvent); err != nil {
+ go func() {
+ p.eventService.Reconnect()
+
+ retryErr := p.eventService.RetryPublish(PortEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return nil
diff --git a/controller/topology/routing-tables/routingTableService.go b/controller/topology/routing-tables/routingTableService.go
index eec4efb2f..2a1fc1348 100644
--- a/controller/topology/routing-tables/routingTableService.go
+++ b/controller/topology/routing-tables/routingTableService.go
@@ -68,8 +68,16 @@ func (r *RoutingTableService) createRoutingTable(routingTable RoutingTable) (Rou
return routingTable, err
}
- if err := r.eventService.PublishEvent(RoutingTableEventTopic, event.NewAddEvent(routingTable.ID)); err != nil {
- log.Error(err)
+ pubEvent := event.NewAddEvent(routingTable.ID)
+ if err := r.eventService.PublishEvent(RoutingTableEventTopic, pubEvent); err != nil {
+ go func() {
+ r.eventService.Reconnect()
+
+ retryErr := r.eventService.RetryPublish(RoutingTableEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return routingTable, nil
@@ -82,8 +90,16 @@ func (r *RoutingTableService) Update(routingTable RoutingTable) error {
return err
}
- if err := r.eventService.PublishEvent(RoutingTableEventTopic, event.NewUpdateEvent(routingTable.ID)); err != nil {
- log.Error(err)
+ pubEvent := event.NewUpdateEvent(routingTable.ID)
+ if err := r.eventService.PublishEvent(RoutingTableEventTopic, pubEvent); err != nil {
+ go func() {
+ r.eventService.Reconnect()
+
+ retryErr := r.eventService.RetryPublish(RoutingTableEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return nil
@@ -96,8 +112,16 @@ func (r *RoutingTableService) Delete(routingTable RoutingTable) error {
return err
}
- if err := r.eventService.PublishEvent(RoutingTableEventTopic, event.NewDeleteEvent(routingTable.ID)); err != nil {
- log.Error(err)
+ pubEvent := event.NewDeleteEvent(routingTable.ID)
+ if err := r.eventService.PublishEvent(RoutingTableEventTopic, pubEvent); err != nil {
+ go func() {
+ r.eventService.Reconnect()
+
+ retryErr := r.eventService.RetryPublish(RoutingTableEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return nil
diff --git a/controller/topology/topologyService.go b/controller/topology/topologyService.go
index 694910843..685302693 100644
--- a/controller/topology/topologyService.go
+++ b/controller/topology/topologyService.go
@@ -74,8 +74,16 @@ func (t *service) AddLink(link links.Link) error {
return err
}
- if err := t.eventService.PublishEvent(LinkEventTopic, event.NewAddEvent(link.ID)); err != nil {
- log.Error(err)
+ pubEvent := event.NewAddEvent(link.ID)
+ if err := t.eventService.PublishEvent(LinkEventTopic, pubEvent); err != nil {
+ go func() {
+ t.eventService.Reconnect()
+
+ retryErr := t.eventService.RetryPublish(LinkEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return nil
@@ -88,8 +96,16 @@ func (t *service) UpdateLink(link links.Link) error {
return err
}
- if err := t.eventService.PublishEvent(LinkEventTopic, event.NewUpdateEvent(link.ID)); err != nil {
- log.Error(err)
+ pubEvent := event.NewUpdateEvent(link.ID)
+ if err := t.eventService.PublishEvent(LinkEventTopic, pubEvent); err != nil {
+ go func() {
+ t.eventService.Reconnect()
+
+ retryErr := t.eventService.RetryPublish(LinkEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return nil
@@ -104,8 +120,16 @@ func (t *service) DeleteLink(link links.Link) error {
return err
}
- if err := t.eventService.PublishEvent(LinkEventTopic, event.NewDeleteEvent(link.ID)); err != nil {
- log.Error(err)
+ pubEvent := event.NewDeleteEvent(link.ID)
+ if err := t.eventService.PublishEvent(LinkEventTopic, pubEvent); err != nil {
+ go func() {
+ t.eventService.Reconnect()
+
+ retryErr := t.eventService.RetryPublish(LinkEventTopic, pubEvent)
+ if retryErr != nil {
+ log.Error(retryErr)
+ }
+ }()
}
return nil
diff --git a/go.mod b/go.mod
index 0f802294d..968323405 100644
--- a/go.mod
+++ b/go.mod
@@ -77,6 +77,7 @@ require (
github.com/prometheus/procfs v0.8.0 // indirect
github.com/rabbitmq/amqp091-go v1.4.0
github.com/rivo/uniseg v0.2.0 // indirect
+ github.com/sethvargo/go-retry v0.2.3
github.com/spf13/afero v1.9.2 // indirect
github.com/spf13/cast v1.5.0 // indirect
github.com/spf13/jwalterweatherman v1.1.0 // indirect
diff --git a/go.sum b/go.sum
index fdb18fa57..0b4b959c6 100644
--- a/go.sum
+++ b/go.sum
@@ -33,12 +33,8 @@ cloud.google.com/go/datastore v1.1.0/go.mod h1:umbIZjpQpHh4hmRpGhH4tLFup+FVzqBi1
cloud.google.com/go/firestore v1.1.0/go.mod h1:ulACoGHTpvq5r8rxGJ4ddJZBZqakUQqClKRT5SZwBmk=
cloud.google.com/go/pubsub v1.0.1/go.mod h1:R0Gpsv3s54REJCy4fxDixWD93lHJMoZTyQ2kNxGRt3I=
cloud.google.com/go/pubsub v1.1.0/go.mod h1:EwwdRX2sKPjnvnqCa270oGRyludottCI76h+R3AArQw=
-cloud.google.com/go/pubsub v1.1.0/go.mod h1:EwwdRX2sKPjnvnqCa270oGRyludottCI76h+R3AArQw=
-cloud.google.com/go/pubsub v1.2.0/go.mod h1:jhfEVHT8odbXTkndysNHCcx0awwzvfOlguIAii9o8iA=
cloud.google.com/go/pubsub v1.2.0/go.mod h1:jhfEVHT8odbXTkndysNHCcx0awwzvfOlguIAii9o8iA=
cloud.google.com/go/pubsub v1.3.1/go.mod h1:i+ucay31+CNRpDW4Lu78I4xXG+O1r/MAHgjpRVR+TSU=
-cloud.google.com/go/pubsub v1.3.1/go.mod h1:i+ucay31+CNRpDW4Lu78I4xXG+O1r/MAHgjpRVR+TSU=
-cloud.google.com/go/storage v1.0.0/go.mod h1:IhtSnM/ZTZV8YYJWCY8RULGVqBDmpoyjwiyrjsg+URw=
cloud.google.com/go/storage v1.0.0/go.mod h1:IhtSnM/ZTZV8YYJWCY8RULGVqBDmpoyjwiyrjsg+URw=
cloud.google.com/go/storage v1.5.0/go.mod h1:tpKbwo567HUNpVclU5sGELwQWBDZ8gh0ZeosJ0Rtdos=
cloud.google.com/go/storage v1.6.0/go.mod h1:N7U0C8pVQ/+NIKOBQyamJIeKQKkZ+mxpohlUTyfDhBk=
@@ -866,6 +862,8 @@ github.com/sergi/go-diff v1.2.0 h1:XU+rvMAioB0UC3q1MFrIQy4Vo5/4VsRDQQXHsEya6xQ=
github.com/sergi/go-diff v1.2.0/go.mod h1:STckp+ISIX8hZLjrqAeVduY0gWCT9IjLuqbuNXdaHfM=
github.com/sethvargo/go-password v0.2.0 h1:BTDl4CC/gjf/axHMaDQtw507ogrXLci6XRiLc7i/UHI=
github.com/sethvargo/go-password v0.2.0/go.mod h1:Ym4Mr9JXLBycr02MFuVQ/0JHidNetSgbzutTr3zsYXE=
+github.com/sethvargo/go-retry v0.2.3 h1:oYlgvIvsju3jNbottWABtbnoLC+GDtLdBHxKWxQm/iU=
+github.com/sethvargo/go-retry v0.2.3/go.mod h1:1afjQuvh7s4gflMObvjLPaWgluLLyhA1wmVZ6KLpICw=
github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
github.com/sirupsen/logrus v1.0.4-0.20170822132746-89742aefa4b2/go.mod h1:pMByvHTf9Beacp5x1UXfOR9xyW/9antXMhjMPG0dEzc=
github.com/sirupsen/logrus v1.0.6/go.mod h1:pMByvHTf9Beacp5x1UXfOR9xyW/9antXMhjMPG0dEzc=
--
GitLab