vbatts/ntfy
1
0
Fork 0
Code Pull requests Activity

Allow /metrics on default port; reduce memory if not enabled

Browse source
This commit is contained in:
binwiederhier 2023-03-15 22:34:06 -04:00
parent bb3fe4f830
commit 358b344916
9 changed files with 184 additions and 125 deletions
Download patch file Download diff file Expand all files Collapse all files

8
server/config.go
View file

@ -61,7 +61,7 @@ var (
// DefaultDisallowedTopics defines the topics that are forbidden, because they are used elsewhere. This array can be
// extended using the server.yml config. If updated, also update in Android and web app.
DefaultDisallowedTopics = []string{"docs", "static", "file", "app", "account", "settings", "signup", "login", "v1"}
DefaultDisallowedTopics = []string{"docs", "static", "file", "app", "metrics", "account", "settings", "signup", "login", "v1"}
)
// Config is the main config struct for the application. Use New to instantiate a default config struct.
@ -72,7 +72,6 @@ type Config struct {
ListenHTTPS string
ListenUnix string
ListenUnixMode fs.FileMode
ListenMetricsHTTP string
KeyFile string
CertFile string
FirebaseKeyFile string
@ -106,6 +105,8 @@ type Config struct {
SMTPServerListen string
SMTPServerDomain string
SMTPServerAddrPrefix string
MetricsEnable bool
MetricsListenHTTP string
MessageLimit int
MinDelay time.Duration
MaxDelay time.Duration
@ -134,7 +135,8 @@ type Config struct {
EnableWeb bool
EnableSignup bool // Enable creation of accounts via API and UI
EnableLogin bool
EnableReservations bool // Allow users with role "user" to own/reserve topics
EnableReservations bool // Allow users with role "user" to own/reserve topics
EnableMetrics bool
AccessControlAllowOrigin string // CORS header field to restrict access from web clients
Version string // injected by App
}

4
server/file_cache.go
View file

@ -67,7 +67,7 @@ func (c *fileCache) Write(id string, in io.Reader, limiters ...util.Limiter) (in
}
c.mu.Lock()
c.totalSizeCurrent += size
metrics.attachmentsTotalSize.Set(float64(c.totalSizeCurrent))
mset(metricAttachmentsTotalSize, c.totalSizeCurrent)
c.mu.Unlock()
return size, nil
}
@ -90,7 +90,7 @@ func (c *fileCache) Remove(ids ...string) error {
c.mu.Lock()
c.totalSizeCurrent = size
c.mu.Unlock()
metrics.attachmentsTotalSize.Set(float64(size))
mset(metricAttachmentsTotalSize, size)
return nil
}

52
server/server.go
View file

@ -52,6 +52,7 @@ type Server struct {
fileCache *fileCache // File system based cache that stores attachments
stripe stripeAPI // Stripe API, can be replaced with a mock
priceCache *util.LookupCache[map[string]int64] // Stripe price ID -> price as cents (USD implied!)
metricsHandler http.Handler // Handles /metrics if enable-metrics set, and listen-metrics-http not set
closeChan chan bool
mu sync.Mutex
}
@ -74,6 +75,7 @@ var (
webConfigPath = "/config.js"
accountPath = "/account"
matrixPushPath = "/_matrix/push/v1/notify"
metricsPath = "/metrics"
apiHealthPath = "/v1/health"
apiTiers = "/v1/tiers"
apiAccountPath = "/v1/account"
@ -212,6 +214,9 @@ func (s *Server) Run() error {
if s.config.SMTPServerListen != "" {
listenStr += fmt.Sprintf(" %s[smtp]", s.config.SMTPServerListen)
}
if s.config.MetricsListenHTTP != "" {
listenStr += fmt.Sprintf(" %s[http/metrics]", s.config.MetricsListenHTTP)
}
log.Tag(tagStartup).Info("Listening on%s, ntfy %s, log level is %s", listenStr, s.config.Version, log.CurrentLevel().String())
if log.IsFile() {
fmt.Fprintf(os.Stderr, "Listening on%s, ntfy %s\n", listenStr, s.config.Version)
@ -258,11 +263,15 @@ func (s *Server) Run() error {
errChan <- httpServer.Serve(s.unixListener)
}()
}
if s.config.ListenMetricsHTTP != "" {
s.httpMetricsServer = &http.Server{Addr: s.config.ListenMetricsHTTP, Handler: promhttp.Handler()}
if s.config.MetricsListenHTTP != "" {
initMetrics()
s.httpMetricsServer = &http.Server{Addr: s.config.MetricsListenHTTP, Handler: promhttp.Handler()}
go func() {
errChan <- s.httpMetricsServer.ListenAndServe()
}()
} else if s.config.EnableMetrics {
initMetrics()
s.metricsHandler = promhttp.Handler()
}
if s.config.SMTPServerListen != "" {
go func() {
@ -324,7 +333,9 @@ func (s *Server) handle(w http.ResponseWriter, r *http.Request) {
s.handleError(w, r, v, err)
return
}
metrics.httpRequests.WithLabelValues("200", "20000", r.Method).Inc()
if metricHTTPRequests != nil {
metricHTTPRequests.WithLabelValues("200", "20000", r.Method).Inc()
}
}).
Debug("HTTP request finished")
}
@ -334,7 +345,9 @@ func (s *Server) handleError(w http.ResponseWriter, r *http.Request, v *visitor,
if !ok {
httpErr = errHTTPInternalError
}
metrics.httpRequests.WithLabelValues(fmt.Sprintf("%d", httpErr.HTTPCode), fmt.Sprintf("%d", httpErr.Code), r.Method).Inc()
if metricHTTPRequests != nil {
metricHTTPRequests.WithLabelValues(fmt.Sprintf("%d", httpErr.HTTPCode), fmt.Sprintf("%d", httpErr.Code), r.Method).Inc()
}
isRateLimiting := util.Contains(rateLimitingErrorCodes, httpErr.HTTPCode)
isNormalError := strings.Contains(err.Error(), "i/o timeout") || util.Contains(normalErrorCodes, httpErr.HTTPCode)
ev := logvr(v, r).Err(err)
@ -415,6 +428,8 @@ func (s *Server) handleInternal(w http.ResponseWriter, r *http.Request, v *visit
return s.ensurePaymentsEnabled(s.handleBillingTiersGet)(w, r, v)
} else if r.Method == http.MethodGet && r.URL.Path == matrixPushPath {
return s.handleMatrixDiscovery(w)
} else if r.Method == http.MethodGet && r.URL.Path == metricsPath && s.metricsHandler != nil {
return s.handleMetrics(w, r, v)
} else if r.Method == http.MethodGet && staticRegex.MatchString(r.URL.Path) {
return s.ensureWebEnabled(s.handleStatic)(w, r, v)
} else if r.Method == http.MethodGet && docsRegex.MatchString(r.URL.Path) {
@ -507,6 +522,13 @@ func (s *Server) handleWebConfig(w http.ResponseWriter, _ *http.Request, _ *visi
return err
}
// handleMetrics returns Prometheus metrics. This endpoint is only called if enable-metrics is set,
// and listen-metrics-http is not set.
func (s *Server) handleMetrics(w http.ResponseWriter, r *http.Request, _ *visitor) error {
s.metricsHandler.ServeHTTP(w, r)
return nil
}
func (s *Server) handleStatic(w http.ResponseWriter, r *http.Request, _ *visitor) error {
r.URL.Path = webSiteDir + r.URL.Path
util.Gzip(http.FileServer(http.FS(webFsCached))).ServeHTTP(w, r)
@ -683,7 +705,7 @@ func (s *Server) handlePublishInternal(r *http.Request, v *visitor) (*message, e
s.messages++
s.mu.Unlock()
if unifiedpush {
metrics.unifiedPushPublishedSuccess.Inc()
minc(metricUnifiedPushPublishedSuccess)
}
return m, nil
}
@ -691,18 +713,18 @@ func (s *Server) handlePublishInternal(r *http.Request, v *visitor) (*message, e
func (s *Server) handlePublish(w http.ResponseWriter, r *http.Request, v *visitor) error {
m, err := s.handlePublishInternal(r, v)
if err != nil {
metrics.messagesPublishedFailure.Inc()
minc(metricMessagesPublishedFailure)
return err
}
metrics.messagesPublishedSuccess.Inc()
minc(metricMessagesPublishedSuccess)
return s.writeJSON(w, m)
}
func (s *Server) handlePublishMatrix(w http.ResponseWriter, r *http.Request, v *visitor) error {
_, err := s.handlePublishInternal(r, v)
if err != nil {
metrics.messagesPublishedFailure.Inc()
metrics.matrixPublishedFailure.Inc()
minc(metricMessagesPublishedFailure)
minc(metricMatrixPublishedFailure)
if e, ok := err.(*errHTTP); ok && e.HTTPCode == errHTTPInsufficientStorageUnifiedPush.HTTPCode {
topic, err := fromContext[*topic](r, contextTopic)
if err != nil {
@ -718,15 +740,15 @@ func (s *Server) handlePublishMatrix(w http.ResponseWriter, r *http.Request, v *
}
return err
}
metrics.messagesPublishedSuccess.Inc()
metrics.matrixPublishedSuccess.Inc()
minc(metricMessagesPublishedSuccess)
minc(metricMatrixPublishedSuccess)
return writeMatrixSuccess(w)
}
func (s *Server) sendToFirebase(v *visitor, m *message) {
logvm(v, m).Tag(tagFirebase).Debug("Publishing to Firebase")
if err := s.firebaseClient.Send(v, m); err != nil {
metrics.firebasePublishedFailure.Inc()
minc(metricFirebasePublishedFailure)
if err == errFirebaseTemporarilyBanned {
logvm(v, m).Tag(tagFirebase).Err(err).Debug("Unable to publish to Firebase: %v", err.Error())
} else {
@ -734,17 +756,17 @@ func (s *Server) sendToFirebase(v *visitor, m *message) {
}
return
}
metrics.firebasePublishedSuccess.Inc()
minc(metricFirebasePublishedSuccess)
}
func (s *Server) sendEmail(v *visitor, m *message, email string) {
logvm(v, m).Tag(tagEmail).Field("email", email).Debug("Sending email to %s", email)
if err := s.smtpSender.Send(v, m, email); err != nil {
logvm(v, m).Tag(tagEmail).Field("email", email).Err(err).Warn("Unable to send email to %s: %v", email, err.Error())
metrics.emailsPublishedFailure.Inc()
minc(metricEmailsPublishedFailure)
return
}
metrics.emailsPublishedSuccess.Inc()
minc(metricEmailsPublishedSuccess)
}
func (s *Server) forwardPollRequest(v *visitor, m *message) {

13
server/server.yml
View file

@ -263,6 +263,19 @@
# stripe-webhook-key:
# billing-contact:
# Metrics
#
# ntfy can expose Prometheus-style metrics via a /metrics endpoint, or on a dedicated listen IP/port.
# Metrics may be considered sensitive information, so before you enable them, be sure you know what you are
# doing, and/or secure access to the endpoint in your reverse proxy.
#
# - enable-metrics enables the /metrics endpoint for the default ntfy server (i.e. HTTP, HTTPS and/or Unix socket)
# - metrics-listen-http exposes the metrics endpoint via a dedicated [IP]:port. If set, this option implicitly
# enables metrics as well, e.g. "10.0.1.1:9090" or ":9090"
#
# enable-metrics: false
# metrics-listen-http:
# Logging options
#
# By default, ntfy logs to the console (stderr), with an "info" log level, and in a human-readable text format.

10
server/server_manager.go
View file

@ -83,12 +83,10 @@ func (s *Server) execManager() {
"emails_sent_failure": sentMailFailure,
}).
Info("Server stats")
if s.httpMetricsServer != nil {
metrics.messagesCached.Set(float64(messagesCached))
metrics.visitors.Set(float64(visitorsCount))
metrics.subscribers.Set(float64(subscribers))
metrics.topics.Set(float64(topicsCount))
}
mset(metricMessagesCached, messagesCached)
mset(metricVisitors, visitorsCount)
mset(metricSubscribers, subscribers)
mset(metricTopics, topicsCount)
}
func (s *Server) pruneVisitors() {

191
server/server_metrics.go
View file

@ -5,101 +5,108 @@ import (
)
var (
metrics = newMetrics()
metricMessagesPublishedSuccess prometheus.Counter
metricMessagesPublishedFailure prometheus.Counter
metricMessagesCached prometheus.Gauge
metricFirebasePublishedSuccess prometheus.Counter
metricFirebasePublishedFailure prometheus.Counter
metricEmailsPublishedSuccess prometheus.Counter
metricEmailsPublishedFailure prometheus.Counter
metricEmailsReceivedSuccess prometheus.Counter
metricEmailsReceivedFailure prometheus.Counter
metricUnifiedPushPublishedSuccess prometheus.Counter
metricMatrixPublishedSuccess prometheus.Counter
metricMatrixPublishedFailure prometheus.Counter
metricAttachmentsTotalSize prometheus.Gauge
metricVisitors prometheus.Gauge
metricSubscribers prometheus.Gauge
metricTopics prometheus.Gauge
metricHTTPRequests *prometheus.CounterVec
)
type serverMetrics struct {
messagesPublishedSuccess prometheus.Counter
messagesPublishedFailure prometheus.Counter
messagesCached prometheus.Gauge
firebasePublishedSuccess prometheus.Counter
firebasePublishedFailure prometheus.Counter
emailsPublishedSuccess prometheus.Counter
emailsPublishedFailure prometheus.Counter
emailsReceivedSuccess prometheus.Counter
emailsReceivedFailure prometheus.Counter
unifiedPushPublishedSuccess prometheus.Counter
matrixPublishedSuccess prometheus.Counter
matrixPublishedFailure prometheus.Counter
attachmentsTotalSize prometheus.Gauge
visitors prometheus.Gauge
subscribers prometheus.Gauge
topics prometheus.Gauge
httpRequests *prometheus.CounterVec
func initMetrics() {
metricMessagesPublishedSuccess = prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_messages_published_success",
})
metricMessagesPublishedFailure = prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_messages_published_failure",
})
metricMessagesCached = prometheus.NewGauge(prometheus.GaugeOpts{
Name: "ntfy_messages_cached_total",
})
metricFirebasePublishedSuccess = prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_firebase_published_success",
})
metricFirebasePublishedFailure = prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_firebase_published_failure",
})
metricEmailsPublishedSuccess = prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_emails_sent_success",
})
metricEmailsPublishedFailure = prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_emails_sent_failure",
})
metricEmailsReceivedSuccess = prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_emails_received_success",
})
metricEmailsReceivedFailure = prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_emails_received_failure",
})
metricUnifiedPushPublishedSuccess = prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_unifiedpush_published_success",
})
metricMatrixPublishedSuccess = prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_matrix_published_success",
})
metricMatrixPublishedFailure = prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_matrix_published_failure",
})
metricAttachmentsTotalSize = prometheus.NewGauge(prometheus.GaugeOpts{
Name: "ntfy_attachments_total_size",
})
metricVisitors = prometheus.NewGauge(prometheus.GaugeOpts{
Name: "ntfy_visitors_total",
})
metricSubscribers = prometheus.NewGauge(prometheus.GaugeOpts{
Name: "ntfy_subscribers_total",
})
metricTopics = prometheus.NewGauge(prometheus.GaugeOpts{
Name: "ntfy_topics_total",
})
metricHTTPRequests = prometheus.NewCounterVec(prometheus.CounterOpts{
Name: "ntfy_http_requests_total",
}, []string{"http_code", "ntfy_code", "http_method"})
prometheus.MustRegister(
metricMessagesPublishedSuccess,
metricMessagesPublishedFailure,
metricMessagesCached,
metricFirebasePublishedSuccess,
metricFirebasePublishedFailure,
metricEmailsPublishedSuccess,
metricEmailsPublishedFailure,
metricEmailsReceivedSuccess,
metricEmailsReceivedFailure,
metricUnifiedPushPublishedSuccess,
metricMatrixPublishedSuccess,
metricMatrixPublishedFailure,
metricAttachmentsTotalSize,
metricVisitors,
metricSubscribers,
metricTopics,
metricHTTPRequests,
)
}
func newMetrics() *serverMetrics {
m := &serverMetrics{
messagesPublishedSuccess: prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_messages_published_success",
}),
messagesPublishedFailure: prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_messages_published_failure",
}),
messagesCached: prometheus.NewGauge(prometheus.GaugeOpts{
Name: "ntfy_messages_cached_total",
}),
firebasePublishedSuccess: prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_firebase_published_success",
}),
firebasePublishedFailure: prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_firebase_published_failure",
}),
emailsPublishedSuccess: prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_emails_sent_success",
}),
emailsPublishedFailure: prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_emails_sent_failure",
}),
emailsReceivedSuccess: prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_emails_received_success",
}),
emailsReceivedFailure: prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_emails_received_failure",
}),
unifiedPushPublishedSuccess: prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_unifiedpush_published_success",
}),
matrixPublishedSuccess: prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_matrix_published_success",
}),
matrixPublishedFailure: prometheus.NewCounter(prometheus.CounterOpts{
Name: "ntfy_matrix_published_failure",
}),
attachmentsTotalSize: prometheus.NewGauge(prometheus.GaugeOpts{
Name: "ntfy_attachments_total_size",
}),
visitors: prometheus.NewGauge(prometheus.GaugeOpts{
Name: "ntfy_visitors_total",
}),
subscribers: prometheus.NewGauge(prometheus.GaugeOpts{
Name: "ntfy_subscribers_total",
}),
topics: prometheus.NewGauge(prometheus.GaugeOpts{
Name: "ntfy_topics_total",
}),
httpRequests: prometheus.NewCounterVec(prometheus.CounterOpts{
Name: "ntfy_http_requests_total",
}, []string{"http_code", "ntfy_code", "http_method"}),
// minc increments a prometheus.Counter if it is non-nil
func minc(counter prometheus.Counter) {
if counter != nil {
counter.Inc()
}
}
// mset sets a prometheus.Gauge if it is non-nil
func mset[T int | int64 | float64](gauge prometheus.Gauge, value T) {
if gauge != nil {
gauge.Set(float64(value))
}
prometheus.MustRegister(
m.messagesPublishedSuccess,
m.messagesPublishedFailure,
m.messagesCached,
m.firebasePublishedSuccess,
m.firebasePublishedFailure,
m.emailsPublishedSuccess,
m.emailsPublishedFailure,
m.emailsReceivedSuccess,
m.emailsReceivedFailure,
m.unifiedPushPublishedSuccess,
m.matrixPublishedSuccess,
m.matrixPublishedFailure,
m.attachmentsTotalSize,
m.visitors,
m.subscribers,
m.topics,
m.httpRequests,
)
return m
}

4
server/smtp_server.go
View file

@ -165,7 +165,7 @@ func (s *smtpSession) Data(r io.Reader) error {
s.backend.mu.Lock()
s.backend.success++
s.backend.mu.Unlock()
metrics.emailsReceivedSuccess.Inc()
minc(metricEmailsReceivedSuccess)
return nil
})
}
@ -218,7 +218,7 @@ func (s *smtpSession) withFailCount(fn func() error) error {
// We do not want to spam the log with WARN messages.
logem(s.conn).Err(err).Debug("Incoming mail error")
s.backend.failure++
metrics.emailsReceivedFailure.Inc()
minc(metricEmailsReceivedFailure)
}
return err
}