diff --git a/pkg/prom/middleware.go b/pkg/prom/middleware.go index 741fed9a..b2daa79f 100644 --- a/pkg/prom/middleware.go +++ b/pkg/prom/middleware.go @@ -2,10 +2,12 @@ package prom import ( "net/http" + "time" - "github.com/prometheus/client_golang/prometheus" + "github.com/ethereum/go-ethereum/rpc" ) +// HTTPMiddleware http connection metric reader func HTTPMiddleware(next http.Handler) http.Handler { if !metrics { return next @@ -14,12 +16,14 @@ func HTTPMiddleware(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { httpCount.Inc() - timer := prometheus.NewTimer(httpDuration) + start := time.Now() next.ServeHTTP(w, r) - timer.ObserveDuration() + duration := time.Now().Sub(start) + httpDuration.Observe(float64(duration.Seconds())) }) } +// WSMiddleware websocket connection counter func WSMiddleware(next http.Handler) http.Handler { if !metrics { return next @@ -27,9 +31,18 @@ func WSMiddleware(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { wsCount.Inc() - - timer := prometheus.NewTimer(wsDuration) next.ServeHTTP(w, r) - timer.ObserveDuration() + wsCount.Dec() }) } + +// IPCMiddleware unix-socket connection counter +func IPCMiddleware(server *rpc.Server, client rpc.Conn) { + if metrics { + ipcCount.Inc() + } + server.ServeCodec(rpc.NewCodec(client), 0) + if metrics { + ipcCount.Dec() + } +} diff --git a/pkg/prom/prom.go b/pkg/prom/prom.go index 828ed160..f17ef546 100644 --- a/pkg/prom/prom.go +++ b/pkg/prom/prom.go @@ -19,12 +19,8 @@ var ( httpCount prometheus.Counter httpDuration prometheus.Histogram - - wsCount prometheus.Counter - wsDuration prometheus.Histogram - - ipcCount prometheus.Counter - ipcDuration prometheus.Gauge + wsCount prometheus.Gauge + ipcCount prometheus.Gauge ) // Init module initialization @@ -35,39 +31,27 @@ func Init() { Namespace: namespace, Subsystem: subsystemHTTP, Name: "count", - Help: "", + Help: "http request count", }) httpDuration = promauto.NewHistogram(prometheus.HistogramOpts{ Namespace: namespace, Subsystem: subsystemHTTP, Name: "duration", - Help: "", + Help: "http request duration", }) - wsCount = promauto.NewCounter(prometheus.CounterOpts{ + wsCount = promauto.NewGauge(prometheus.GaugeOpts{ Namespace: namespace, Subsystem: subsystemWS, Name: "count", - Help: "", - }) - wsDuration = promauto.NewHistogram(prometheus.HistogramOpts{ - Namespace: namespace, - Subsystem: subsystemWS, - Name: "duration", - Help: "", + Help: "websocket conntection count", }) - ipcCount = promauto.NewCounter(prometheus.CounterOpts{ + ipcCount = promauto.NewGauge(prometheus.GaugeOpts{ Namespace: namespace, Subsystem: subsystemIPC, Name: "count", - Help: "", - }) - ipcDuration = promauto.NewGauge(prometheus.GaugeOpts{ - Namespace: namespace, - Subsystem: subsystemIPC, - Name: "duration", - Help: "", + Help: "unix socket connection count", }) } diff --git a/pkg/rpc/ipc.go b/pkg/rpc/ipc.go index 297be5ab..4e170ebe 100644 --- a/pkg/rpc/ipc.go +++ b/pkg/rpc/ipc.go @@ -1,15 +1,15 @@ package rpc import ( - "C" - "fmt" "net" "os" "path/filepath" + "github.com/ethereum/go-ethereum/p2p/netutil" "github.com/ethereum/go-ethereum/rpc" log "github.com/sirupsen/logrus" + "github.com/vulcanize/ipld-eth-server/pkg/prom" ) var ( @@ -38,6 +38,22 @@ func ipcListen(endpoint string) (net.Listener, error) { return l, nil } +func ipcServe(srv *rpc.Server, listener net.Listener) { + for { + conn, err := listener.Accept() + if netutil.IsTemporaryError(err) { + log.WithError(err).Warn("rpc accept error") + continue + } + if err != nil { + log.WithError(err).Warn("unknown error") + continue + } + log.WithField("addr", conn.RemoteAddr()).Trace("accepted ipc connection") + go prom.IPCMiddleware(srv, conn) + } +} + // StartIPCEndpoint starts an IPC endpoint. func StartIPCEndpoint(ipcEndpoint string, apis []rpc.API) (net.Listener, *rpc.Server, error) { // Register all the APIs exposed by the services. @@ -53,6 +69,7 @@ func StartIPCEndpoint(ipcEndpoint string, apis []rpc.API) (net.Listener, *rpc.Se if err != nil { return nil, nil, err } - go handler.ServeListener(listener) + + go ipcServe(handler, listener) return listener, handler, nil }