Merge pull request #436 from tidwall/webhook-metrics

Added geofence delivery counter

internal/server/hooks.go

@@ -139,6 +139,7 @@ func (c *Server) cmdSetHook(msg *Message, chanCmd bool) (
 		Metas:     metas,
 		channel:   chanCmd,
 		cond:      sync.NewCond(&sync.Mutex{}),
+		counter:   &c.statsTotalMsgsSent,
 	}
 	if expiresSet {
 		hook.expires =
@@ -460,6 +461,7 @@ type Hook struct {
 	query      string
 	epm        *endpoint.Manager
 	expires    time.Time
+	counter    *aint // counter that grows when a message was sent
 }
 
 // Expires returns when the hook expires. Required by the expire.Item interface.
@@ -648,6 +650,7 @@ func (h *Hook) proc() (ok bool) {
 			}
 			log.Debugf("Endpoint send ok: %v: %v: %v", idx, endpoint, err)
 			sent = true
+			h.counter.add(1)
 			break
 		}
 		if !sent {

internal/server/live.go

@@ -193,7 +193,9 @@ func (server *Server) goLive(
 					return nil // nil return is fine here
 				}
 			}
+			server.statsTotalMsgsSent.add(len(msgs))
 			lb.cond.L.Lock()
+
 		}
 		lb.cond.Wait()
 		lb.cond.L.Unlock()

internal/server/pubsub.go

@@ -280,6 +280,7 @@ func (c *Server) liveSubscription(
 				write(b)
 			}
 		}
+		c.statsTotalMsgsSent.add(1)
 	}
 
 	m := [2]map[string]bool{

internal/server/server.go

@@ -80,6 +80,7 @@ type Server struct {
 	followc            aint // counter increases when follow property changes
 	statsTotalConns    aint // counter for total connections
 	statsTotalCommands aint // counter for total commands
+	statsTotalMsgsSent aint // counter for total sent webhook messages
 	statsExpired       aint // item expiration counter
 	lastShrinkDuration aint
 	stopServer         abool

internal/server/stats.go

@@ -267,6 +267,8 @@ func (c *Server) extStats(m map[string]interface{}) {
 	m["tile38_total_connections_received"] = c.statsTotalConns.get()
 	// Number of commands processed by the server
 	m["tile38_total_commands_processed"] = c.statsTotalCommands.get()
+	// Number of webhook messages sent by server
+	m["tile38_total_messages_sent"] = c.statsTotalMsgsSent.get()
 	// Number of key expiration events
 	m["tile38_expired_keys"] = c.statsExpired.get()
 	// Number of connected slaves
@@ -350,6 +352,7 @@ func (c *Server) writeInfoPersistence(w *bytes.Buffer) {
 func (c *Server) writeInfoStats(w *bytes.Buffer) {
 	fmt.Fprintf(w, "total_connections_received:%d\r\n", c.statsTotalConns.get())  // Total number of connections accepted by the server
 	fmt.Fprintf(w, "total_commands_processed:%d\r\n", c.statsTotalCommands.get()) // Total number of commands processed by the server
+	fmt.Fprintf(w, "total_messages_sent:%d\r\n", c.statsTotalMsgsSent.get())      // Total number of commands processed by the server
 	fmt.Fprintf(w, "expired_keys:%d\r\n", c.statsExpired.get())                   // Total number of key expiration events
 }