Merge pull request #908 from mysteriumnetwork/feature/890-send-hole-p…

…unching-events

Send events to ELK about hole-punching

cmd/di.go

@@ -493,7 +493,9 @@ func (di *Dependencies) bootstrapNATComponents(options node.Options) {
 			config.NewConfigParser(),
 			traversal.NewNATProxy(),
 			di.PortPool,
-			mapping.StageName)
+			mapping.StageName,
+			di.EventBus,
+		)
 	} else {
 		di.NATPinger = &traversal.NoopPinger{}
 	}

nat/traversal/pinger.go

@@ -34,6 +34,7 @@ import (
 const prefix = "[NATPinger] "
 const pingInterval = 200
 const pingTimeout = 10000
+const stageName = "hole_punching"
 
 // Pinger represents NAT pinger structure
 type Pinger struct {
@@ -47,6 +48,7 @@ type Pinger struct {
 	portPool       PortSupplier
 	consumerPort   int
 	previousStage  string
+	eventPublisher Publisher
 }
 
 // NatEventWaiter is responsible for waiting for nat events
@@ -64,8 +66,13 @@ type PortSupplier interface {
 	Acquire() (port.Port, error)
 }
 
+// Publisher is responsible for publishing given events
+type Publisher interface {
+	Publish(topic string, data interface{})
+}
+
 // NewPingerFactory returns Pinger instance
-func NewPingerFactory(waiter NatEventWaiter, parser ConfigParser, proxy natProxy, portPool PortSupplier, previousStage string) *Pinger {
+func NewPingerFactory(waiter NatEventWaiter, parser ConfigParser, proxy natProxy, portPool PortSupplier, previousStage string, publisher Publisher) *Pinger {
 	target := make(chan *Params)
 	cancel := make(chan struct{})
 	stop := make(chan struct{})
@@ -78,6 +85,7 @@ func NewPingerFactory(waiter NatEventWaiter, parser ConfigParser, proxy natProxy
 		natProxy:       proxy,
 		portPool:       portPool,
 		previousStage:  previousStage,
+		eventPublisher: publisher,
 	}
 }
 
@@ -215,21 +223,29 @@ func (p *Pinger) ping(conn *net.UDPConn) error {
 				n = 128
 			}
 
-			err := ipv4.NewConn(conn).SetTTL(n)
+			err := p.sendPingRequest(conn, n)
 			if err != nil {
-				return errors.Wrap(err, "pinger setting ttl failed")
+				p.eventPublisher.Publish(EventTopic, BuildFailureEvent(stageName, err))
+				return err
 			}
 
-			n++
+			p.eventPublisher.Publish(EventTopic, BuildSuccessEvent(stageName))
 
-			_, err = conn.Write([]byte("continuously pinging to " + conn.RemoteAddr().String()))
-			if err != nil {
-				return err
-			}
+			n++
 		}
 	}
 }
 
+func (p *Pinger) sendPingRequest(conn *net.UDPConn, ttl int) error {
+	err := ipv4.NewConn(conn).SetTTL(ttl)
+	if err != nil {
+		return errors.Wrap(err, "pinger setting ttl failed")
+	}
+
+	_, err = conn.Write([]byte("continuously pinging to " + conn.RemoteAddr().String()))
+	return errors.Wrap(err, "pinging request failed")
+}
+
 func (p *Pinger) getConnection(ip string, port int, pingerPort int) (*net.UDPConn, error) {
 	udpAddr, err := net.ResolveUDPAddr("udp4", fmt.Sprintf("%s:%d", ip, port))
 	if err != nil {