Add notification queue metrics (#16026)

This commit is contained in:
Klaus Post 2022-11-09 01:36:47 +01:00 committed by GitHub
parent 1b0dfb0f58
commit bbc312fce6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 217 additions and 50 deletions

View File

@ -58,6 +58,7 @@ func init() {
getCacheMetrics(), getCacheMetrics(),
getGoMetrics(), getGoMetrics(),
getHTTPMetrics(), getHTTPMetrics(),
getNotificationMetrics(),
getLocalStorageMetrics(), getLocalStorageMetrics(),
getMinioProcMetrics(), getMinioProcMetrics(),
getMinioVersionMetrics(), getMinioVersionMetrics(),
@ -83,6 +84,7 @@ func init() {
getNetworkMetrics(), getNetworkMetrics(),
getMinioVersionMetrics(), getMinioVersionMetrics(),
getS3TTFBMetric(), getS3TTFBMetric(),
getNotificationMetrics(),
}) })
clusterCollector = newMinioClusterCollector(allMetricsGroups) clusterCollector = newMinioClusterCollector(allMetricsGroups)
} }
@ -127,6 +129,7 @@ const (
scannerSubsystem MetricSubsystem = "scanner" scannerSubsystem MetricSubsystem = "scanner"
iamSubsystem MetricSubsystem = "iam" iamSubsystem MetricSubsystem = "iam"
kmsSubsystem MetricSubsystem = "kms" kmsSubsystem MetricSubsystem = "kms"
notifySubsystem MetricSubsystem = "notify"
) )
// MetricName are the individual names for the metric. // MetricName are the individual names for the metric.
@ -1536,6 +1539,39 @@ func getCacheMetrics() *MetricsGroup {
return mg return mg
} }
func getNotificationMetrics() *MetricsGroup {
mg := &MetricsGroup{}
mg.RegisterRead(func(ctx context.Context) []Metric {
stats := globalConfigTargetList.Stats()
metrics := make([]Metric, 0, 1+len(stats.TargetStats))
metrics = append(metrics, Metric{
Description: MetricDescription{
Namespace: minioNamespace,
Subsystem: notifySubsystem,
Name: "current_send_in_progress",
Help: "Number of concurrent async Send calls active to all targets",
Type: gaugeMetric,
},
Value: float64(stats.CurrentSendCalls),
})
for _, st := range stats.TargetStats {
metrics = append(metrics, Metric{
Description: MetricDescription{
Namespace: minioNamespace,
Subsystem: notifySubsystem,
Name: "target_queue_length",
Help: "Number of unsent notifications in queue for target",
Type: gaugeMetric,
},
VariableLabels: map[string]string{"target_id": st.ID.ID, "target_name": st.ID.Name},
Value: float64(st.CurrentQueue),
})
}
return metrics
})
return mg
}
func getHTTPMetrics() *MetricsGroup { func getHTTPMetrics() *MetricsGroup {
mg := &MetricsGroup{} mg := &MetricsGroup{}
mg.RegisterRead(func(ctx context.Context) (metrics []Metric) { mg.RegisterRead(func(ctx context.Context) (metrics []Metric) {

View File

@ -128,6 +128,11 @@ func (target *AMQPTarget) ID() event.TargetID {
return target.id return target.id
} }
// Store returns any underlying store if set.
func (target *AMQPTarget) Store() event.TargetStore {
return target.store
}
// IsActive - Return true if target is up and active // IsActive - Return true if target is up and active
func (target *AMQPTarget) IsActive() (bool, error) { func (target *AMQPTarget) IsActive() (bool, error) {
if err := target.init(); err != nil { if err := target.init(); err != nil {

View File

@ -167,6 +167,11 @@ func (target *ElasticsearchTarget) ID() event.TargetID {
return target.id return target.id
} }
// Store returns any underlying store if set.
func (target *ElasticsearchTarget) Store() event.TargetStore {
return target.store
}
// IsActive - Return true if target is up and active // IsActive - Return true if target is up and active
func (target *ElasticsearchTarget) IsActive() (bool, error) { func (target *ElasticsearchTarget) IsActive() (bool, error) {
if err := target.init(); err != nil { if err := target.init(); err != nil {

View File

@ -33,7 +33,7 @@ import (
"github.com/minio/minio/internal/logger" "github.com/minio/minio/internal/logger"
xnet "github.com/minio/pkg/net" xnet "github.com/minio/pkg/net"
sarama "github.com/Shopify/sarama" "github.com/Shopify/sarama"
saramatls "github.com/Shopify/sarama/tools/tls" saramatls "github.com/Shopify/sarama/tools/tls"
) )
@ -139,6 +139,11 @@ func (target *KafkaTarget) ID() event.TargetID {
return target.id return target.id
} }
// Store returns any underlying store if set.
func (target *KafkaTarget) Store() event.TargetStore {
return target.store
}
// IsActive - Return true if target is up and active // IsActive - Return true if target is up and active
func (target *KafkaTarget) IsActive() (bool, error) { func (target *KafkaTarget) IsActive() (bool, error) {
if err := target.init(); err != nil { if err := target.init(); err != nil {

View File

@ -121,6 +121,11 @@ func (target *MQTTTarget) ID() event.TargetID {
return target.id return target.id
} }
// Store returns any underlying store if set.
func (target *MQTTTarget) Store() event.TargetStore {
return target.store
}
// IsActive - Return true if target is up and active // IsActive - Return true if target is up and active
func (target *MQTTTarget) IsActive() (bool, error) { func (target *MQTTTarget) IsActive() (bool, error) {
if err := target.init(); err != nil { if err := target.init(); err != nil {

View File

@ -165,6 +165,11 @@ func (target *MySQLTarget) ID() event.TargetID {
return target.id return target.id
} }
// Store returns any underlying store if set.
func (target *MySQLTarget) Store() event.TargetStore {
return target.store
}
// IsActive - Return true if target is up and active // IsActive - Return true if target is up and active
func (target *MySQLTarget) IsActive() (bool, error) { func (target *MySQLTarget) IsActive() (bool, error) {
if err := target.init(); err != nil { if err := target.init(); err != nil {

View File

@ -230,6 +230,11 @@ func (target *NATSTarget) ID() event.TargetID {
return target.id return target.id
} }
// Store returns any underlying store if set.
func (target *NATSTarget) Store() event.TargetStore {
return target.store
}
// IsActive - Return true if target is up and active // IsActive - Return true if target is up and active
func (target *NATSTarget) IsActive() (bool, error) { func (target *NATSTarget) IsActive() (bool, error) {
if err := target.init(); err != nil { if err := target.init(); err != nil {

View File

@ -105,6 +105,11 @@ func (target *NSQTarget) ID() event.TargetID {
return target.id return target.id
} }
// Store returns any underlying store if set.
func (target *NSQTarget) Store() event.TargetStore {
return target.store
}
// IsActive - Return true if target is up and active // IsActive - Return true if target is up and active
func (target *NSQTarget) IsActive() (bool, error) { func (target *NSQTarget) IsActive() (bool, error) {
if err := target.init(); err != nil { if err := target.init(); err != nil {

View File

@ -157,6 +157,11 @@ func (target *PostgreSQLTarget) ID() event.TargetID {
return target.id return target.id
} }
// Store returns any underlying store if set.
func (target *PostgreSQLTarget) Store() event.TargetStore {
return target.store
}
// IsActive - Return true if target is up and active // IsActive - Return true if target is up and active
func (target *PostgreSQLTarget) IsActive() (bool, error) { func (target *PostgreSQLTarget) IsActive() (bool, error) {
if err := target.init(); err != nil { if err := target.init(); err != nil {

View File

@ -19,11 +19,12 @@ package target
import ( import (
"encoding/json" "encoding/json"
"math"
"os" "os"
"path/filepath" "path/filepath"
"sort" "sort"
"strings"
"sync" "sync"
"time"
"github.com/minio/minio/internal/event" "github.com/minio/minio/internal/event"
) )
@ -36,9 +37,10 @@ const (
// QueueStore - Filestore for persisting events. // QueueStore - Filestore for persisting events.
type QueueStore struct { type QueueStore struct {
sync.RWMutex sync.RWMutex
currentEntries uint64 entryLimit uint64
entryLimit uint64 directory string
directory string
entries map[string]int64 // key -> modtime as unix nano
} }
// NewQueueStore - Creates an instance for QueueStore. // NewQueueStore - Creates an instance for QueueStore.
@ -50,6 +52,7 @@ func NewQueueStore(directory string, limit uint64) Store {
return &QueueStore{ return &QueueStore{
directory: directory, directory: directory,
entryLimit: limit, entryLimit: limit,
entries: make(map[string]int64, limit),
} }
} }
@ -62,17 +65,24 @@ func (store *QueueStore) Open() error {
return err return err
} }
names, err := store.list() files, err := store.list()
if err != nil { if err != nil {
return err return err
} }
currentEntries := uint64(len(names)) // Truncate entries.
if currentEntries >= store.entryLimit { if uint64(len(files)) > store.entryLimit {
return errLimitExceeded files = files[:store.entryLimit]
}
for _, file := range files {
if file.IsDir() {
continue
}
key := strings.TrimSuffix(file.Name(), eventExt)
if fi, err := file.Info(); err == nil {
store.entries[key] = fi.ModTime().UnixNano()
}
} }
store.currentEntries = currentEntries
return nil return nil
} }
@ -91,7 +101,7 @@ func (store *QueueStore) write(key string, e event.Event) error {
} }
// Increment the event count. // Increment the event count.
store.currentEntries++ store.entries[key] = time.Now().UnixNano()
return nil return nil
} }
@ -100,7 +110,7 @@ func (store *QueueStore) write(key string, e event.Event) error {
func (store *QueueStore) Put(e event.Event) error { func (store *QueueStore) Put(e event.Event) error {
store.Lock() store.Lock()
defer store.Unlock() defer store.Unlock()
if store.currentEntries >= store.entryLimit { if uint64(len(store.entries)) >= store.entryLimit {
return errLimitExceeded return errLimitExceeded
} }
key, err := getNewUUID() key, err := getNewUUID()
@ -146,44 +156,51 @@ func (store *QueueStore) Del(key string) error {
return store.del(key) return store.del(key)
} }
// Len returns the entry count.
func (store *QueueStore) Len() int {
store.RLock()
l := len(store.entries)
defer store.RUnlock()
return l
}
// lockless call // lockless call
func (store *QueueStore) del(key string) error { func (store *QueueStore) del(key string) error {
if err := os.Remove(filepath.Join(store.directory, key+eventExt)); err != nil { err := os.Remove(filepath.Join(store.directory, key+eventExt))
return err
}
// Decrement the current entries count. // Delete as entry no matter the result
store.currentEntries-- delete(store.entries, key)
// Current entries can underflow, when multiple return err
// events are being pushed in parallel, this code
// is needed to ensure that we don't underflow.
//
// queueStore replayEvents is not serialized,
// this code is needed to protect us under
// such situations.
if store.currentEntries == math.MaxUint64 {
store.currentEntries = 0
}
return nil
} }
// List - lists all files from the directory. // List - lists all files registered in the store.
func (store *QueueStore) List() ([]string, error) { func (store *QueueStore) List() ([]string, error) {
store.RLock() store.RLock()
defer store.RUnlock() l := make([]string, 0, len(store.entries))
return store.list() for k := range store.entries {
} l = append(l, k)
// list lock less.
func (store *QueueStore) list() ([]string, error) {
var names []string
files, err := os.ReadDir(store.directory)
if err != nil {
return names, err
} }
// Sort the dentries. // Sort entries...
sort.Slice(l, func(i, j int) bool {
return store.entries[l[i]] < store.entries[l[j]]
})
store.RUnlock()
return l, nil
}
// list will read all entries from disk.
// Entries are returned sorted by modtime, oldest first.
// Underlying entry list in store is *not* updated.
func (store *QueueStore) list() ([]os.DirEntry, error) {
files, err := os.ReadDir(store.directory)
if err != nil {
return nil, err
}
// Sort the entries.
sort.Slice(files, func(i, j int) bool { sort.Slice(files, func(i, j int) bool {
ii, err := files[i].Info() ii, err := files[i].Info()
if err != nil { if err != nil {
@ -196,9 +213,5 @@ func (store *QueueStore) list() ([]string, error) {
return ii.ModTime().Before(ji.ModTime()) return ii.ModTime().Before(ji.ModTime())
}) })
for _, file := range files { return files, nil
names = append(names, file.Name())
}
return names, nil
} }

View File

@ -202,12 +202,33 @@ func TestQueueStoreListN(t *testing.T) {
t.Fatalf("List() Expected: 10, got %d", len(names)) t.Fatalf("List() Expected: 10, got %d", len(names))
} }
if err = os.RemoveAll(queueDir); err != nil { // re-open
store, err = setUpStore(queueDir, 10)
if err != nil {
t.Fatal("Failed to create a queue store ", err)
}
names, err = store.List()
if err != nil {
t.Fatal(err) t.Fatal(err)
} }
_, err = store.List() if len(names) != 10 {
if !os.IsNotExist(err) { t.Fatalf("List() Expected: 10, got %d", len(names))
t.Fatalf("Expected List() to fail with os.ErrNotExist, %s", err) }
if len(names) != store.Len() {
t.Fatalf("List() Expected: 10, got %d", len(names))
}
// Delete all
for _, key := range names {
err := store.Del(key)
if err != nil {
t.Fatal(err)
}
}
// Re-list
lst, err := store.List()
if len(lst) > 0 || err != nil {
t.Fatalf("Expected List() to return empty list and no error, got %v err: %v", lst, err)
} }
} }

View File

@ -133,6 +133,11 @@ func (target *RedisTarget) ID() event.TargetID {
return target.id return target.id
} }
// Store returns any underlying store if set.
func (target *RedisTarget) Store() event.TargetStore {
return target.store
}
// IsActive - Return true if target is up and active // IsActive - Return true if target is up and active
func (target *RedisTarget) IsActive() (bool, error) { func (target *RedisTarget) IsActive() (bool, error) {
if err := target.init(); err != nil { if err := target.init(); err != nil {

View File

@ -41,6 +41,7 @@ var errLimitExceeded = errors.New("the maximum store limit reached")
type Store interface { type Store interface {
Put(event event.Event) error Put(event event.Event) error
Get(key string) (event.Event, error) Get(key string) (event.Event, error)
Len() int
List() ([]string, error) List() ([]string, error)
Del(key string) error Del(key string) error
Open() error Open() error

View File

@ -115,6 +115,11 @@ func (target *WebhookTarget) IsActive() (bool, error) {
return target.isActive() return target.isActive()
} }
// Store returns any underlying store if set.
func (target *WebhookTarget) Store() event.TargetStore {
return target.store
}
func (target *WebhookTarget) isActive() (bool, error) { func (target *WebhookTarget) isActive() (bool, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel() defer cancel()

View File

@ -19,6 +19,7 @@ package event
import ( import (
"fmt" "fmt"
"strings"
"sync" "sync"
"sync/atomic" "sync/atomic"
) )
@ -35,6 +36,26 @@ type Target interface {
Save(Event) error Save(Event) error
Send(string) error Send(string) error
Close() error Close() error
Store() TargetStore
}
// TargetStore is a shallow version of a target.Store
type TargetStore interface {
Len() int
}
// TargetStats is a collection of stats for multiple targets.
type TargetStats struct {
// CurrentSendCalls is the number of concurrent async Send calls to all targets
CurrentSendCalls int64
TargetStats map[string]TargetStat
}
// TargetStat is the stats of a single target.
type TargetStat struct {
ID TargetID
CurrentQueue int // Populated if target has a store.
} }
// TargetList - holds list of targets indexed by target ID. // TargetList - holds list of targets indexed by target ID.
@ -166,6 +187,26 @@ func (list *TargetList) Send(event Event, targetIDset TargetIDSet, resCh chan<-
}() }()
} }
// Stats returns stats for targets.
func (list *TargetList) Stats() TargetStats {
t := TargetStats{}
if list == nil {
return t
}
t.CurrentSendCalls = atomic.LoadInt64(&list.currentSendCalls)
list.RLock()
defer list.RUnlock()
t.TargetStats = make(map[string]TargetStat, len(list.targets))
for id, target := range list.targets {
ts := TargetStat{ID: id}
if st := target.Store(); st != nil {
ts.CurrentQueue = st.Len()
}
t.TargetStats[strings.ReplaceAll(id.String(), ":", "_")] = ts
}
return t
}
// NewTargetList - creates TargetList. // NewTargetList - creates TargetList.
func NewTargetList() *TargetList { func NewTargetList() *TargetList {
return &TargetList{targets: make(map[TargetID]Target)} return &TargetList{targets: make(map[TargetID]Target)}

View File

@ -40,6 +40,11 @@ func (target ExampleTarget) Save(eventData Event) error {
return target.send(eventData) return target.send(eventData)
} }
// Store - Returns a nil store.
func (target ExampleTarget) Store() TargetStore {
return nil
}
func (target ExampleTarget) send(eventData Event) error { func (target ExampleTarget) send(eventData Event) error {
b := make([]byte, 1) b := make([]byte, 1)
if _, err := rand.Read(b); err != nil { if _, err := rand.Read(b); err != nil {