From 179edecfe0befd26e57643b629b7313784b5ca36 Mon Sep 17 00:00:00 2001 From: ice-myles <96409608+ice-myles@users.noreply.github.com> Date: Fri, 1 Aug 2025 20:31:32 +0300 Subject: [PATCH] Metrics --- .../subzero_ion_connect.go | 3 + database/query/global.go | 14 +- database/query/internal/connector/storage.go | 19 +- database/query/query.go | 9 + go.mod | 3 +- go.sum | 2 + monitoring/metrics.go | 360 ++++++++++++++++++ server/http/nip96/storage_nip96.go | 34 ++ server/server.go | 4 +- server/ws/auth.go | 5 + server/ws/internal/adapters/websocket.go | 5 + server/ws/subscriptions.go | 25 ++ server/ws/ws.go | 16 + storage/global.go | 51 +++ 14 files changed, 544 insertions(+), 6 deletions(-) create mode 100644 monitoring/metrics.go diff --git a/cmd/subzero-ion-connect/subzero_ion_connect.go b/cmd/subzero-ion-connect/subzero_ion_connect.go index 05c8ca8f..328fb81e 100644 --- a/cmd/subzero-ion-connect/subzero_ion_connect.go +++ b/cmd/subzero-ion-connect/subzero_ion_connect.go @@ -23,6 +23,7 @@ import ( "github.com/ice-blockchain/subzero/dvm" hashtagssender "github.com/ice-blockchain/subzero/hashtags-sender" "github.com/ice-blockchain/subzero/model" + "github.com/ice-blockchain/subzero/monitoring" pushnotifications "github.com/ice-blockchain/subzero/push-notifications" "github.com/ice-blockchain/subzero/server" wsserver "github.com/ice-blockchain/subzero/server/ws" @@ -40,6 +41,7 @@ var ( Version: getVersion(), Run: func(cmd *cobra.Command, _ []string) { cfg.MustInit(configPath) + monitoring.MustInit() validation.MustInit() query.MustInit(cmd.Context()) command.MustInit(cmd.Context()) @@ -127,6 +129,7 @@ func init() { if err := query.AcceptEvents(ctx, events...); err != nil { return errors.Wrap(err, "query.AcceptEvent failed") } + monitoring.RecordEventStored(len(events)) if ch, err := dvm.AcceptJob(ctx, events[0]); err == nil && ch != nil { antsPool.Submit(func() { result := <-ch diff --git a/database/query/global.go b/database/query/global.go index 0347afc4..1decdac7 100644 --- a/database/query/global.go +++ b/database/query/global.go @@ -14,6 +14,7 @@ import ( "github.com/ice-blockchain/subzero/cfg" "github.com/ice-blockchain/subzero/model" + "github.com/ice-blockchain/subzero/monitoring" ) var ( @@ -22,6 +23,7 @@ var ( Once sync.Once } UsedDatabaseStorage atomic.Uint64 + DatabaseEventsCount atomic.Uint64 ) type ( @@ -121,7 +123,7 @@ func MustInit(ctx context.Context, opts ...Option) { WithRelayURL(conf.RelayURL) go globalDB.Client.StartExpiredEventsCleanup(ctx) - go globalDB.Client.StartCollectingUsedDatabaseStorage(ctx) + go globalDB.Client.StartCollectingUsedDatabaseInfo(ctx) go func() { <-ctx.Done() @@ -206,7 +208,7 @@ func (db *dbClient) StartExpiredEventsCleanup(ctx context.Context) { } } -func (db *dbClient) StartCollectingUsedDatabaseStorage(ctx context.Context) { +func (db *dbClient) StartCollectingUsedDatabaseInfo(ctx context.Context) { ticks := make(chan struct{}, 1) ticks <- struct{}{} go func() { @@ -233,7 +235,15 @@ func (db *dbClient) StartCollectingUsedDatabaseStorage(ctx context.Context) { log.Printf("failed to query database size: %v", err) } else { UsedDatabaseStorage.Store(usedDatabaseStorage) + monitoring.SetDatabaseSize("subzero", "localhost", int64(usedDatabaseStorage)) } + if eventsCount, err := db.queryEventsCount(queryCtx); err != nil { + log.Printf("failed to query events count: %v", err) + } else { + DatabaseEventsCount.Store(eventsCount) + monitoring.SetDatabaseEventsCount("subzero", "localhost", eventsCount) + } + cancel() } } diff --git a/database/query/internal/connector/storage.go b/database/query/internal/connector/storage.go index 4ff593df..58984676 100644 --- a/database/query/internal/connector/storage.go +++ b/database/query/internal/connector/storage.go @@ -7,6 +7,7 @@ import ( "fmt" "log" "math" + "net/url" "reflect" "strings" "sync" @@ -22,6 +23,7 @@ import ( "github.com/jackc/pgx/v5/tracelog" "github.com/ice-blockchain/subzero/model" + "github.com/ice-blockchain/subzero/monitoring" ) func WithWriteURLs(urls ...string) Option { @@ -244,8 +246,21 @@ func poolConnect(ctx context.Context, connectionString string, log tracelog.Logg if !strings.Contains(strings.ToLower(connectionString), "pool_min_conns") { conf.MinConns = 1 } - conf.ConnConfig.Tracer = &tracelog.TraceLog{Logger: log, LogLevel: tracelog.LogLevelDebug} - return pgxpool.NewWithConfig(ctx, conf) + host := conf.ConnConfig.Host + if host == "" { + if u, parseErr := url.Parse(connectionString); parseErr == nil { + host = u.Host + } + } + dbTracer := monitoring.NewAdvancedDBTracer(host, "main") + conf.ConnConfig.Tracer = dbTracer + pool, err := pgxpool.NewWithConfig(ctx, conf) + if err != nil { + return nil, err + } + monitoring.RegisterPoolFromConnector(pool, "subzero", host) + + return pool, nil } func poolDoAfterConnect(ctx context.Context, conn *pgx.Conn) error { diff --git a/database/query/query.go b/database/query/query.go index 81780a2f..1b8da304 100644 --- a/database/query/query.go +++ b/database/query/query.go @@ -1216,3 +1216,12 @@ func (db *dbClient) queryDatabaseSize(ctx context.Context) (uint64, error) { } return *sizePtr, nil } + +func (db *dbClient) queryEventsCount(ctx context.Context) (uint64, error) { + countPtr, err := connector.Get[uint64](ctx, db.db, `SELECT count(*) FROM events WHERE deleted = false`) + if err != nil { + return 0, errors.Wrapf(err, "failed to query events count") + } + + return *countPtr, nil +} diff --git a/go.mod b/go.mod index 23d35c99..e9645829 100644 --- a/go.mod +++ b/go.mod @@ -10,6 +10,7 @@ replace ( require ( firebase.google.com/go/v4 v4.18.0 + github.com/IBM/pgxpoolprometheus v1.1.2 github.com/alitto/pond/v2 v2.5.0 github.com/bzick/tokenizer v1.4.10 github.com/caddyserver/certmagic v0.23.0 @@ -45,6 +46,7 @@ require ( github.com/mxschmitt/golang-combinations v1.2.0 github.com/nbd-wtf/go-nostr v0.51.12 github.com/panjf2000/ants/v2 v2.11.3 + github.com/prometheus/client_golang v1.23.0 github.com/puzpuzpuz/xsync/v4 v4.1.0 github.com/quic-go/quic-go v0.54.0 github.com/quic-go/webtransport-go v0.9.0 @@ -207,7 +209,6 @@ require ( github.com/planetscale/vtprotobuf v0.6.1-0.20250313105119-ba97887b0a25 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect - github.com/prometheus/client_golang v1.23.0 // indirect github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/common v0.65.0 // indirect github.com/prometheus/procfs v0.17.0 // indirect diff --git a/go.sum b/go.sum index 1718fe1e..2a41624d 100644 --- a/go.sum +++ b/go.sum @@ -50,6 +50,8 @@ github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0 github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0.53.0/go.mod h1:jUZ5LYlw40WMd07qxcQJD5M40aUxrfwqQX1g7zxYnrQ= github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.53.0 h1:Ron4zCA/yk6U7WOBXhTJcDpsUBG9npumK6xw2auFltQ= github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.53.0/go.mod h1:cSgYe11MCNYunTnRXrKiR/tHc0eoKjICUuWpNZoVCOo= +github.com/IBM/pgxpoolprometheus v1.1.2 h1:sHJwxoL5Lw4R79Zt+H4Uj1zZ4iqXJLdk7XDE7TPs97U= +github.com/IBM/pgxpoolprometheus v1.1.2/go.mod h1:+vWzISN6S9ssgurhUNmm6AlXL9XLah3TdWJktquKTR8= github.com/MarvinJWendt/testza v0.1.0/go.mod h1:7AxNvlfeHP7Z/hDQ5JtE3OKYT3XFUeLCDE2DQninSqs= github.com/MarvinJWendt/testza v0.2.1/go.mod h1:God7bhG8n6uQxwdScay+gjm9/LnO4D3kkcZX4hv9Rp8= github.com/MarvinJWendt/testza v0.2.8/go.mod h1:nwIcjmr0Zz+Rcwfh3/4UhBp7ePKVhuBExvZqnKYWlII= diff --git a/monitoring/metrics.go b/monitoring/metrics.go new file mode 100644 index 00000000..4c64d393 --- /dev/null +++ b/monitoring/metrics.go @@ -0,0 +1,360 @@ +// SPDX-License-Identifier: ice License 1.0 + +package monitoring + +import ( + "context" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/IBM/pgxpoolprometheus" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" +) + +var ( + activeWSConnectionsCounter atomic.Int64 + authenticatedUsersCounter atomic.Int64 + activeSubscriptionsCounter atomic.Int64 + + globalPrometheusCollector *PrometheusCollector + prometheusCollectorOnce sync.Once + + // === DATABASE QUERY METRICS === + dbQueryDuration = promauto.NewHistogramVec( + prometheus.HistogramOpts{ + Name: "subzero_db_query_duration_seconds", + Help: "Database query execution duration in seconds", + Buckets: []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0}, + }, + []string{"host", "operation", "success", "query_type"}, + ) + + dbQueryTotal = promauto.NewCounterVec( + prometheus.CounterOpts{ + Name: "subzero_db_queries_total", + Help: "Total number of database queries executed", + }, + []string{"host", "operation", "query_type", "success"}, + ) + + // === CONNECTION POOL METRICS === + dbConnectionEstablishDuration = promauto.NewHistogramVec( + prometheus.HistogramOpts{ + Name: "subzero_db_connection_establish_duration_seconds", + Help: "Time spent establishing new database connection", + Buckets: []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0}, + }, + []string{"host", "success"}, + ) + + // === END-TO-END REQUEST METRICS === + requestDuration = promauto.NewHistogramVec( + prometheus.HistogramOpts{ + Name: "subzero_request_duration_seconds", + Help: "End-to-end request duration in seconds", + Buckets: []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0}, + }, + []string{"operation", "success", "protocol"}, + ) + + requestTotal = promauto.NewCounterVec( + prometheus.CounterOpts{ + Name: "subzero_requests_total", + Help: "Total number of requests processed", + }, + []string{"operation", "success", "protocol"}, + ) + + // === WEBSOCKET METRICS === + wsOperationDuration = promauto.NewHistogramVec( + prometheus.HistogramOpts{ + Name: "subzero_ws_operation_duration_seconds", + Help: "WebSocket operation duration in seconds", + Buckets: []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0}, + }, + []string{"operation", "success"}, + ) + + wsOperationTotal = promauto.NewCounterVec( + prometheus.CounterOpts{ + Name: "subzero_ws_operations_total", + Help: "Total number of WebSocket operations", + }, + []string{"operation", "success"}, + ) + + wsActiveConnections = promauto.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "subzero_ws_active_connections", + Help: "Number of active WebSocket connections", + }, + []string{"type"}, + ) + + // === NOSTR-SPECIFIC METRICS === + authenticatedUsers = promauto.NewGauge( + prometheus.GaugeOpts{ + Name: "subzero_authenticated_users", + Help: "Number of currently authenticated users", + }, + ) + + activeSubscriptions = promauto.NewGauge( + prometheus.GaugeOpts{ + Name: "subzero_subscriptions_active", + Help: "Number of active subscriptions", + }, + ) + + eventsValidation = promauto.NewCounterVec( + prometheus.CounterOpts{ + Name: "subzero_events_validation_total", + Help: "Total number of event validation results", + }, + []string{"result", "reason"}, + ) + + eventsStored = promauto.NewCounter( + prometheus.CounterOpts{ + Name: "subzero_events_stored_total", + Help: "Total number of events successfully stored", + }, + ) + + // === DATABASE SIZE METRICS === + dbSize = promauto.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "subzero_db_size_bytes", + Help: "Database size in bytes", + }, + []string{"database", "host"}, + ) + + dbEventsCount = promauto.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "subzero_db_events_count", + Help: "Total number of events in database", + }, + []string{"database", "host"}, + ) + + // === STORAGE METRICS === + storageSize = promauto.NewGauge( + prometheus.GaugeOpts{ + Name: "subzero_storage_size_bytes", + Help: "Storage size in bytes", + }, + ) + + storageRequests = promauto.NewCounterVec( + prometheus.CounterOpts{ + Name: "subzero_storage_requests_total", + Help: "Total number of storage requests", + }, + []string{"operation", "success"}, + ) +) + +type ( + PrometheusCollector struct { + registeredHosts map[string]bool + mutex sync.Mutex + } + AdvancedDBTracer struct { + host string + poolName string + } +) + +func MustInit() { + prometheusCollectorOnce.Do(func() { + globalPrometheusCollector = &PrometheusCollector{ + registeredHosts: make(map[string]bool), + } + }) +} + +func NewAdvancedDBTracer(host, poolName string) *AdvancedDBTracer { + return &AdvancedDBTracer{ + host: host, + poolName: poolName, + } +} + +func (pc *PrometheusCollector) RegisterPgxPool(pool *pgxpool.Pool, dbName, host string) { + pc.mutex.Lock() + defer pc.mutex.Unlock() + + if pc.registeredHosts[host] { + return + } + collector := pgxpoolprometheus.NewCollector(pool, map[string]string{ + "db_name": dbName, + "host": host, + }) + prometheus.MustRegister(collector) + pc.registeredHosts[host] = true +} + +func RecordDBQuery(host, operation, queryType string, duration time.Duration, success bool) { + successStr := "true" + if !success { + successStr = "false" + } + + dbQueryDuration.WithLabelValues(host, operation, successStr, queryType).Observe(duration.Seconds()) + dbQueryTotal.WithLabelValues(host, operation, queryType, successStr).Inc() +} + +func RecordConnectionEstablish(host string, duration time.Duration, success bool) { + successStr := "true" + if !success { + successStr = "false" + } + dbConnectionEstablishDuration.WithLabelValues(host, successStr).Observe(duration.Seconds()) +} + +func RecordWSOperation(operation string, duration time.Duration, success bool, count int) { + successStr := "true" + if !success { + successStr = "false" + } + + wsOperationDuration.WithLabelValues(operation, successStr).Observe(duration.Seconds()) + wsOperationTotal.WithLabelValues(operation, successStr).Add(float64(count)) +} + +func IncreaseActiveConnections(connectionType string) { + count := activeWSConnectionsCounter.Add(1) + wsActiveConnections.WithLabelValues(connectionType).Set(float64(count)) +} + +func DecreaseActiveConnections(connectionType string) { + count := activeWSConnectionsCounter.Add(-1) + wsActiveConnections.WithLabelValues(connectionType).Set(float64(count)) +} + +func IncreaseAuthenticatedUsers() { + count := authenticatedUsersCounter.Add(1) + authenticatedUsers.Set(float64(count)) +} + +func DecreaseAuthenticatedUsers() { + count := authenticatedUsersCounter.Add(-1) + authenticatedUsers.Set(float64(count)) +} + +func IncreaseActiveSubscriptions() { + count := activeSubscriptionsCounter.Add(1) + activeSubscriptions.Set(float64(count)) +} + +func DecreaseActiveSubscriptions(delta int64) { + count := activeSubscriptionsCounter.Add(-delta) + activeSubscriptions.Set(float64(count)) +} + +func RecordRequest(operation, protocol string, duration time.Duration, success bool) { + successStr := "true" + if !success { + successStr = "false" + } + + requestDuration.WithLabelValues(operation, successStr, protocol).Observe(duration.Seconds()) + requestTotal.WithLabelValues(operation, successStr, protocol).Inc() +} + +func (t *AdvancedDBTracer) TraceConnectStart(ctx context.Context, data pgx.TraceConnectStartData) context.Context { + return context.WithValue(ctx, "connect_start_time", time.Now()) +} + +func (t *AdvancedDBTracer) TraceConnectEnd(ctx context.Context, data pgx.TraceConnectEndData) { + if startTime, ok := ctx.Value("connect_start_time").(time.Time); ok { + duration := time.Since(startTime) + RecordConnectionEstablish(t.host, duration, data.Err == nil) + } +} + +func (t *AdvancedDBTracer) TraceQueryStart(ctx context.Context, conn *pgx.Conn, data pgx.TraceQueryStartData) context.Context { + ctx = context.WithValue(ctx, "query_start_time", time.Now()) + + queryType := getQueryType(data.SQL) + ctx = context.WithValue(ctx, "query_type", queryType) + operation := queryType + ctx = context.WithValue(ctx, "query_operation", operation) + + return ctx +} + +func (t *AdvancedDBTracer) TraceQueryEnd(ctx context.Context, conn *pgx.Conn, data pgx.TraceQueryEndData) { + if startTime, ok := ctx.Value("query_start_time").(time.Time); ok { + duration := time.Since(startTime) + success := data.Err == nil + + operation := "query" + if op, ok := ctx.Value("query_operation").(string); ok { + operation = op + } + + queryType := "unknown" + if qt, ok := ctx.Value("query_type").(string); ok { + queryType = qt + } + + RecordDBQuery(t.host, operation, queryType, duration, success) + } +} + +func getQueryType(sql string) string { + sql = strings.ToUpper(strings.TrimSpace(sql)) + + switch { + case strings.HasPrefix(sql, "SELECT"): + return "SELECT" + case strings.HasPrefix(sql, "INSERT"): + return "INSERT" + case strings.HasPrefix(sql, "UPDATE"): + return "UPDATE" + case strings.HasPrefix(sql, "DELETE"): + return "DELETE" + default: + return "OTHER" + } +} + +func RecordEventValidation(result, reason string, count int) { + eventsValidation.WithLabelValues(result, reason).Add(float64(count)) +} + +func RecordEventStored(count int) { + eventsStored.Add(float64(count)) +} + +func SetDatabaseSize(database, host string, sizeBytes int64) { + dbSize.WithLabelValues(database, host).Set(float64(sizeBytes)) +} + +func SetDatabaseEventsCount(database, host string, count uint64) { + dbEventsCount.WithLabelValues(database, host).Set(float64(count)) +} + +func SetStorageSize(sizeBytes int64) { + storageSize.Set(float64(sizeBytes)) +} + +func RecordStorageRequest(operation string, success bool) { + successStr := "true" + if !success { + successStr = "false" + } + storageRequests.WithLabelValues(operation, successStr).Inc() +} + +func RegisterPoolFromConnector(pool *pgxpool.Pool, dbName, host string) { + globalPrometheusCollector.RegisterPgxPool(pool, dbName, host) +} diff --git a/server/http/nip96/storage_nip96.go b/server/http/nip96/storage_nip96.go index f954734c..a54a8b4c 100644 --- a/server/http/nip96/storage_nip96.go +++ b/server/http/nip96/storage_nip96.go @@ -20,6 +20,7 @@ import ( "github.com/gin-gonic/gin/binding" "github.com/nbd-wtf/go-nostr" + "github.com/ice-blockchain/subzero/monitoring" "github.com/ice-blockchain/subzero/server/http/nip11" "github.com/ice-blockchain/subzero/server/http/nip98" "github.com/ice-blockchain/subzero/storage" @@ -86,12 +87,14 @@ func (s *storageHandler) Upload() gin.HandlerFunc { token, authErr := s.auth.VerifyToken(gCtx, authHeader, now) if authErr != nil { log.Printf("ERROR: endpoint authentification failed: %v", errors.Wrap(authErr, "endpoint authentification failed")) + monitoring.RecordStorageRequest("upload", false) gCtx.JSON(http.StatusUnauthorized, uploadErr("Unauthorized")) return } attestationValid := token.ValidateAttestation(ctx, nostr.KindFileMetadata, now) if attestationValid != nil { log.Printf("ERROR: on-behalf attestation failed: %v", errors.Wrap(attestationValid, "endpoint authentification failed")) + monitoring.RecordStorageRequest("upload", false) gCtx.JSON(http.StatusForbidden, uploadErr("Forbidden: on-behalf attestation failed")) return } @@ -100,6 +103,7 @@ func (s *storageHandler) Upload() gin.HandlerFunc { uploadingFilePath, input, hash, err := s.storageClient.SaveFile(ctx, now, token.MasterPubKey(), gCtx.Request, maxUploadSize) if err != nil { log.Printf("ERROR: failed to save temp file while processing upload %v", err) + monitoring.RecordStorageRequest("upload", false) switch { case errors.Is(err, storage.ErrValidationFailed): gCtx.JSON(http.StatusBadRequest, uploadErr("failed validate upload request")) @@ -115,6 +119,7 @@ func (s *storageHandler) Upload() gin.HandlerFunc { hashHex := hex.EncodeToString(hash) if hashHex != token.ExpectedHash() { log.Printf("ERROR: endpoint authentification failed: %v", errors.Errorf("payload hash mismatch actual>%v token>%v", hashHex, token.ExpectedHash())) + monitoring.RecordStorageRequest("upload", false) gCtx.JSON(http.StatusForbidden, uploadErr("Unauthorized")) os.Remove(uploadingFilePath) return @@ -123,6 +128,7 @@ func (s *storageHandler) Upload() gin.HandlerFunc { if err != nil { log.Printf("ERROR: failed to upload file: %v", errors.Wrap(err, "failed to upload file to ion storage")) + monitoring.RecordStorageRequest("upload", false) gCtx.JSON(http.StatusInternalServerError, uploadErr("oops, error occurred!")) os.Remove(uploadingFilePath) return @@ -131,6 +137,7 @@ func (s *storageHandler) Upload() gin.HandlerFunc { if existed { resStatus = http.StatusOK } + monitoring.RecordStorageRequest("upload", true) gCtx.JSON(resStatus, fileUploadResponse{ Status: "success", Message: "Upload successful.", @@ -160,11 +167,13 @@ func (s *storageHandler) redirectToDistributedStorageUrl() gin.HandlerFunc { token, authErr := s.auth.VerifyToken(gCtx, authHeader, now) if authErr != nil { log.Printf("ERROR: endpoint authentification failed: %v", errors.Wrap(authErr, "endpoint authentification failed")) + monitoring.RecordStorageRequest("download", false) gCtx.JSON(http.StatusUnauthorized, uploadErr("Unauthorized")) return } file := gCtx.Param("file") if strings.TrimSpace(file) == "" { + monitoring.RecordStorageRequest("download", false) gCtx.JSON(http.StatusBadRequest, uploadErr("filename is required")) return } @@ -177,13 +186,16 @@ func (s *storageHandler) redirectToDistributedStorageUrl() gin.HandlerFunc { url, err := s.storageClient.DownloadUrl(masterPubkey, file) if err != nil { if errors.Is(err, storage.ErrNotFound) { + monitoring.RecordStorageRequest("download", false) gCtx.Status(http.StatusNotFound) return } log.Printf("ERROR: failed to build download url %v", err) + monitoring.RecordStorageRequest("download", false) gCtx.JSON(http.StatusInternalServerError, uploadErr("oops, error occurred!")) return } + monitoring.RecordStorageRequest("download", true) gCtx.Redirect(http.StatusFound, url) } } @@ -191,6 +203,7 @@ func (s *storageHandler) serveFileFromStorage() gin.HandlerFunc { return func(gCtx *gin.Context) { file := gCtx.Param("file") if strings.TrimSpace(file) == "" { + monitoring.RecordStorageRequest("download", false) gCtx.JSON(http.StatusBadRequest, uploadErr("filename is required")) return } @@ -206,13 +219,16 @@ func (s *storageHandler) serveFileFromStorage() gin.HandlerFunc { filePath, err := s.storageClient.FilePath(masterPubkey, file) if err != nil { if errors.Is(err, storage.ErrNotFound) { + monitoring.RecordStorageRequest("download", false) gCtx.Status(http.StatusNotFound) return } log.Printf("ERROR: failed to build download url %v", err) + monitoring.RecordStorageRequest("download", false) gCtx.JSON(http.StatusInternalServerError, uploadErr("oops, error occurred!")) return } + monitoring.RecordStorageRequest("download", true) gCtx.File(filePath) } } @@ -232,22 +248,26 @@ func (s *storageHandler) Delete() gin.HandlerFunc { token, authErr := s.auth.VerifyToken(gCtx, authHeader, now) if authErr != nil { log.Printf("ERROR: endpoint authentification failed: %v", errors.Wrap(authErr, "endpoint authentification failed")) + monitoring.RecordStorageRequest("delete", false) gCtx.JSON(http.StatusUnauthorized, uploadErr("Unauthorized")) return } attestationValid := token.ValidateAttestation(ctx, nostr.KindFileMetadata, now) if attestationValid != nil { log.Printf("ERROR: on-behalf attestation failed: %v", errors.Wrap(attestationValid, "endpoint authentification failed")) + monitoring.RecordStorageRequest("delete", false) gCtx.JSON(http.StatusForbidden, uploadErr("Forbidden: on-behalf attestation failed")) return } file := gCtx.Param("file") if strings.TrimSpace(file) == "" { + monitoring.RecordStorageRequest("delete", false) gCtx.JSON(http.StatusBadRequest, uploadErr("filehash is required")) return } if err := s.storageClient.Delete(ctx, token.PubKey(), token.MasterPubKey(), file); err != nil { log.Printf("ERROR: failed to delete file %v %v", file, err) + monitoring.RecordStorageRequest("delete", false) if errors.Is(err, storage.ErrNotFound) || errors.Is(err, storage.ErrForbidden) { gCtx.JSON(http.StatusForbidden, uploadErr("user do not own file")) return @@ -255,6 +275,7 @@ func (s *storageHandler) Delete() gin.HandlerFunc { gCtx.JSON(http.StatusInternalServerError, uploadErr("oops, error occurred!")) return } + monitoring.RecordStorageRequest("delete", true) gCtx.JSON(http.StatusOK, map[string]any{"status": "success", "message": "deleted"}) } } @@ -266,6 +287,7 @@ func (s *storageHandler) ListFiles() gin.HandlerFunc { token, authErr := s.auth.VerifyToken(gCtx, authHeader, now) if authErr != nil { log.Printf("ERROR: endpoint authentification failed: %v", errors.Wrap(authErr, "endpoint authentification failed")) + monitoring.RecordStorageRequest("list_files", false) gCtx.JSON(http.StatusUnauthorized, uploadErr("Unauthorized")) return } @@ -275,6 +297,7 @@ func (s *storageHandler) ListFiles() gin.HandlerFunc { } if err := gCtx.ShouldBindWith(¶ms, binding.Query); err != nil { log.Printf("ERROR: failed to bind data : %v", errors.Wrap(err, "failed to bind data")) + monitoring.RecordStorageRequest("list_files", false) gCtx.JSON(http.StatusBadRequest, uploadErr("invalid data")) return } @@ -284,6 +307,7 @@ func (s *storageHandler) ListFiles() gin.HandlerFunc { total, filesList, err := s.storageClient.ListFiles(token.MasterPubKey(), params.Page, params.Count) if err != nil { log.Printf("ERROR: failed to list files for user %v %v", token.MasterPubKey(), err) + monitoring.RecordStorageRequest("list_files", false) gCtx.JSON(http.StatusInternalServerError, uploadErr("oops, error occurred!")) return } @@ -303,6 +327,7 @@ func (s *storageHandler) ListFiles() gin.HandlerFunc { CreatedAt uint64 `json:"created_at"` }{Tags: f.ToTags(), Content: f.Content, CreatedAt: f.CreatedAt}) } + monitoring.RecordStorageRequest("list_files", true) gCtx.JSON(http.StatusOK, res) } } @@ -320,22 +345,26 @@ func (s *storageHandler) CrossRelayDownload() gin.HandlerFunc { token, authErr := s.auth.VerifyToken(gCtx, authHeader, now) if authErr != nil { log.Printf("ERROR: endpoint authentification failed: %v", errors.Wrap(authErr, "endpoint authentification failed")) + monitoring.RecordStorageRequest("cross_relay_download", false) gCtx.JSON(http.StatusUnauthorized, uploadErr("Unauthorized")) return } senderUrl := gCtx.GetHeader("Referer") if senderUrl == "" { + monitoring.RecordStorageRequest("cross_relay_download", false) gCtx.JSON(http.StatusBadRequest, uploadErr("unknown sender")) return } senderNIP11, err := s.nip11Fetcher.Fetch(ctx, senderUrl) if err != nil { log.Printf("ERROR: endpoint authentification failed: %v", errors.Wrap(err, "failed to fetch sender NIP11")) + monitoring.RecordStorageRequest("cross_relay_download", false) gCtx.JSON(http.StatusInternalServerError, uploadErr("oops, error occurred")) return } if token.PubKey() != senderNIP11.PubKey { log.Printf("ERROR: endpoint authentification failed: sender pubkey mismatch nip11>%v, token>%v", senderNIP11.PubKey, token.PubKey()) + monitoring.RecordStorageRequest("cross_relay_download", false) gCtx.JSON(http.StatusUnauthorized, uploadErr("sender pubkey mismatch")) return } @@ -348,6 +377,7 @@ func (s *storageHandler) CrossRelayDownload() gin.HandlerFunc { } if err = storage.VerifyFileOwnershipAndAttestationForFileReplication(ctx, now, file, masterPubkey, senderUrl); err != nil { log.Printf("ERROR: not owning the file: %v %v user %v req from %v", err, file, masterPubkey, senderUrl) + monitoring.RecordStorageRequest("cross_relay_download", false) gCtx.JSON(http.StatusConflict, uploadErr("relay does not own the file")) return } @@ -356,18 +386,22 @@ func (s *storageHandler) CrossRelayDownload() gin.HandlerFunc { } if err := gCtx.ShouldBindWith(¶ms, binding.Query); err != nil { log.Printf("ERROR: failed to bind data : %v", errors.Wrap(err, "failed to bind data")) + monitoring.RecordStorageRequest("cross_relay_download", false) gCtx.JSON(http.StatusBadRequest, uploadErr("invalid data")) return } if params.I == "" { + monitoring.RecordStorageRequest("cross_relay_download", false) gCtx.JSON(http.StatusBadRequest, uploadErr("invalid data: i tag not passed")) return } if err := s.storageClient.StartDownloadNewBag(ctx, file, masterPubkey, params.I); err != nil { log.Printf("ERROR: failed to accept new info hash %v for %v: %v", params.I, masterPubkey, err) + monitoring.RecordStorageRequest("cross_relay_download", false) gCtx.JSON(http.StatusInternalServerError, uploadErr("oops, error occurred!")) return } + monitoring.RecordStorageRequest("cross_relay_download", true) gCtx.Status(http.StatusAccepted) } } diff --git a/server/server.go b/server/server.go index 22466503..8b904a6d 100644 --- a/server/server.go +++ b/server/server.go @@ -20,6 +20,7 @@ import ( "github.com/ice-blockchain/subzero/server/http/nip11" "github.com/ice-blockchain/subzero/server/http/nip96" wsserver "github.com/ice-blockchain/subzero/server/ws" + "github.com/prometheus/client_golang/prometheus/promhttp" ) type ( @@ -180,5 +181,6 @@ func (r *router) RegisterRoutes(ctx context.Context, wsroutes wsserver.Router) { GET("/.well-known/nostr/nip96.json", uploader.NIP96Info()). GET("/health-check", func(c *gin.Context) { c.JSON(http.StatusOK, map[string]any{}) - }) + }). + GET("/metrics", gin.WrapH(promhttp.Handler())) } diff --git a/server/ws/auth.go b/server/ws/auth.go index be6bd8c6..f2545451 100644 --- a/server/ws/auth.go +++ b/server/ws/auth.go @@ -11,6 +11,7 @@ import ( "github.com/ice-blockchain/subzero/database/query" "github.com/ice-blockchain/subzero/model" + "github.com/ice-blockchain/subzero/monitoring" "github.com/ice-blockchain/subzero/validation" ) @@ -160,6 +161,10 @@ func (h *handler) handleAuth(ctx context.Context, respWriter Writer, e *model.Ev h.ConnAuth.Store(respWriter, userdata) + if !state.Authenticated { + monitoring.IncreaseAuthenticatedUsers() + } + resp.OK = true return &resp diff --git a/server/ws/internal/adapters/websocket.go b/server/ws/internal/adapters/websocket.go index 2c15dd40..11058d56 100644 --- a/server/ws/internal/adapters/websocket.go +++ b/server/ws/internal/adapters/websocket.go @@ -16,6 +16,7 @@ import ( "github.com/gobwas/ws/wsutil" h2ec "github.com/ice-blockchain/go/src/net/http" + "github.com/ice-blockchain/subzero/monitoring" ) func NewWebSocketAdapter(ctx context.Context, conn net.Conn, readTimeout, writeTimeout time.Duration, shutdownChannel <-chan struct{}) (WSWithWriter, context.Context) { @@ -27,6 +28,8 @@ func NewWebSocketAdapter(ctx context.Context, conn net.Conn, readTimeout, writeT writeTimeout: writeTimeout, } + monitoring.IncreaseActiveConnections("websocket") + return wt, NewCustomCancelContext(ctx, wt.closeChannel, shutdownChannel) } @@ -145,6 +148,8 @@ func (w *WebsocketAdapter) Close() error { return nil } + monitoring.DecreaseActiveConnections("websocket") + close(w.closeChannel) var wErr error diff --git a/server/ws/subscriptions.go b/server/ws/subscriptions.go index d1e81518..5f806b27 100644 --- a/server/ws/subscriptions.go +++ b/server/ws/subscriptions.go @@ -19,6 +19,7 @@ import ( "github.com/ice-blockchain/subzero/database/query" "github.com/ice-blockchain/subzero/model" + "github.com/ice-blockchain/subzero/monitoring" "github.com/ice-blockchain/subzero/validation" ) @@ -131,22 +132,37 @@ func (h *handler) linkSubscription(respWriter Writer, sub *model.Subscription) { }) if loaded { log.Printf("WARN: subscription %s already exists, overwriting it", sub.ID) + } else { + monitoring.IncreaseActiveSubscriptions() } } func (h *handler) unlinkSubscription(respWriter Writer, ID *string) bool { if ID == nil { // Connection is closing, remove all subscriptions. + if authData, ok := h.ConnAuth.Load(respWriter); ok && authData.Authenticated { + monitoring.DecreaseAuthenticatedUsers() + } + + subsCount := 0 h.Subscriptions.Range(func(_ string, sub subscription) bool { if sub.Writer == respWriter { h.Subscriptions.Delete(sub.Source.ID) + subsCount++ } return true }) + if subsCount > 0 { + monitoring.DecreaseActiveSubscriptions(int64(subsCount)) + } + return false } _, ok := h.Subscriptions.LoadAndDelete(*ID) + if ok { + monitoring.DecreaseActiveSubscriptions(1) + } return ok } @@ -402,6 +418,7 @@ func (h *handler) handleReq(ctx context.Context, respWriter Writer, sub *model.S func (h *handler) handleEvents(ctx context.Context, respWriter Writer, events []*model.Event) error { if err := validation.Validate(ctx, events...); err != nil { + monitoring.RecordEventValidation("rejected", "validation_failed", len(events)) return errors.Wrapf(err, "events %v: invalid", events) } @@ -417,6 +434,7 @@ func (h *handler) handleEvents(ctx context.Context, respWriter Writer, events [] }, false }) if !status.Authenticated { + monitoring.RecordEventValidation("rejected", "auth_required", len(events)) err := h.writeResponse(ctx, respWriter, &nostr.AuthEnvelope{ Challenge: &status.Challenge, }) @@ -425,6 +443,8 @@ func (h *handler) handleEvents(ctx context.Context, respWriter Writer, events [] } return errAuthRequired } else if !status.IsEventAllowed(events...) { + monitoring.RecordEventValidation("rejected", "not_allowed", len(events)) + return errors.New("error: not allowed to publish given events") } } @@ -432,10 +452,15 @@ func (h *handler) handleEvents(ctx context.Context, respWriter Writer, events [] if err := wsEventListener(ctx, events...); err != nil { if errors.Is(err, query.ErrReadOnly) { + monitoring.RecordEventValidation("rejected", "read_only", len(events)) + return errRelayReadOnly } + monitoring.RecordEventValidation("rejected", "processing_error", len(events)) + return errors.Wrapf(err, "failed to handle events: %s", model.Events(events).String()) } + monitoring.RecordEventValidation("accepted", "valid", len(events)) return nil } diff --git a/server/ws/ws.go b/server/ws/ws.go index d318bd46..5cd81593 100644 --- a/server/ws/ws.go +++ b/server/ws/ws.go @@ -18,6 +18,7 @@ import ( "github.com/ice-blockchain/subzero/database/query" "github.com/ice-blockchain/subzero/model" + "github.com/ice-blockchain/subzero/monitoring" "github.com/ice-blockchain/subzero/server/ws/internal" "github.com/ice-blockchain/subzero/server/ws/internal/adapters" "github.com/ice-blockchain/subzero/validation" @@ -143,6 +144,8 @@ func (h *handler) Handle(ctx context.Context, respWriter adapters.WSWriter, msgB log.Printf("ERROR: cannot process events: %s: %v", model.Events(events).String(), err) } h.logOperation(respWriter, time.Since(start), "events: handle [%d] events: %v", len(events), string(msgBytes)) + + monitoring.RecordWSOperation("event_handle", time.Since(start), err == nil, len(events)) sendStart := time.Now() for i := range e.Events { resp := &nostr.OKEnvelope{ @@ -162,16 +165,23 @@ func (h *handler) Handle(ctx context.Context, respWriter adapters.WSWriter, msgB } } h.logOperation(respWriter, time.Since(sendStart), "events: send [%d] responses", len(events)) + + monitoring.RecordWSOperation("event_send", time.Since(sendStart), true, len(events)) + return case *nostr.AuthEnvelope: ev := &model.Event{Event: e.Event} err = h.writeResponse(ctx, respWriter, h.handleAuth(ctx, respWriter, ev)) h.logOperation(respWriter, time.Since(start), "auth") + + monitoring.RecordWSOperation("auth", time.Since(start), err == nil, 1) case *nostr.ReqEnvelope: err = h.handleReq(h.populateContext(ctx, respWriter), respWriter, model.NewSubscription(e.SubscriptionID, e.Filters)) h.logOperation(respWriter, time.Since(start), "req: %s: handle [%d] filters: %v: [%v]", e.SubscriptionID, len(e.Filters), string(msgBytes), err) + + monitoring.RecordWSOperation("subscription", time.Since(start), err == nil, len(e.Filters)) case *nostr.CountEnvelope: err = h.handleCount(h.populateContext(ctx, respWriter), e) if err != nil { @@ -188,9 +198,13 @@ func (h *handler) Handle(ctx context.Context, respWriter adapters.WSWriter, msgB case *nostr.CloseEnvelope: h.unlinkSubscription(respWriter, (*string)(e)) h.logOperation(respWriter, time.Since(start), "req: close: %s", (*string)(e)) + + monitoring.RecordWSOperation("close", time.Since(start), true, 1) case *model.BroadcastEnvelope: h.handleBroadcast(h.populateContext(ctx, respWriter), e) h.logOperation(respWriter, time.Since(start), "broadcast") + + monitoring.RecordWSOperation("broadcast", time.Since(start), true, 1) default: err = errors.Errorf("unknown message type %v", input.Label()) } @@ -200,6 +214,8 @@ func (h *handler) Handle(ctx context.Context, respWriter adapters.WSWriter, msgB notice := nostr.NoticeEnvelope(err.Error()) log.Printf("ERROR:%v", errors.Join(err, h.writeResponse(ctx, respWriter, ¬ice))) } + + monitoring.RecordRequest(strings.ToLower(input.Label()), "websocket", time.Since(start), err == nil) } func (h *handler) handleBroadcast(ctx context.Context, e *model.BroadcastEnvelope) { diff --git a/storage/global.go b/storage/global.go index 8ad5c423..4b8e4a4e 100644 --- a/storage/global.go +++ b/storage/global.go @@ -32,6 +32,7 @@ import ( "github.com/ice-blockchain/subzero/cfg" "github.com/ice-blockchain/subzero/database/query" "github.com/ice-blockchain/subzero/model" + "github.com/ice-blockchain/subzero/monitoring" "github.com/ice-blockchain/subzero/storage/statistics" ) @@ -305,6 +306,9 @@ func mustInit(ctx context.Context) *client { if globalConfig.Debug { go cl.report(ctx) } + + go cl.startStorageMetricsCollection(ctx) + loadMonitoringCh := make(chan *db.Event, 1000000) go func() { for ev := range loadMonitoringCh { @@ -452,3 +456,50 @@ func VerifyFileOwnershipAndAttestationForFileReplication(ctx context.Context, no return nil } + +func calculateDirectorySize(dirPath string) (int64, error) { + var totalSize int64 + + err := filepath.Walk(dirPath, func(path string, info os.FileInfo, err error) error { + if err != nil { + return nil + } + if !info.IsDir() { + totalSize += info.Size() + } + return nil + }) + + return totalSize, err +} + +func (c *client) startStorageMetricsCollection(ctx context.Context) { + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + + c.updateStorageMetrics() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + c.updateStorageMetrics() + } + } +} + +func (c *client) updateStorageMetrics() { + if c.rootStoragePath == "" { + return + } + + size, err := calculateDirectorySize(c.rootStoragePath) + if err != nil { + log.Printf("failed to calculate storage size: %v", err) + + return + } + + monitoring.SetStorageSize(size) +}